CRAB/python/GetOutputServer.py

from Actor import *
from crab_util import *
import common
from ApmonIf import ApmonIf
import time
from ProgressBar import ProgressBar
from TerminalController import TerminalController

import xml.dom.minidom

class GetOutputServer(Actor):
 
    def __init__(self, cfg_params,):
        self.cfg_params = cfg_params
        try:  
            self.server_name = self.cfg_params['CRAB.server_name'] # gsiftp://pcpg01.cern.ch/data/SEDir/
            if not self.server_name.endswith("/"):
                self.server_name = self.server_name + "/"
        except KeyError:
            msg = 'No server selected ...' 
            msg = msg + 'Please specify a server in the crab cfg file' 
            raise CrabException(msg) 
        return
    
    def run(self):
        """
        The main method of the class: retrieve the output from server
        """
        common.logger.debug(5, "GetOutput server::run() called")

        start = time.time()

        common.scheduler.checkProxy()

        common.taskDB.load()
        WorkDirName =os.path.basename(os.path.split(common.work_space.topDir())[0])
        projectUniqName = 'crab_'+str(WorkDirName)+'_'+common.taskDB.dict("TasKUUID")     
        #Need to add a check on the treashold level 
        # and on the task readness  TODO  
        try:
            ### retrieving project from the server
            common.logger.message("Retrieving the project from the server...\n")

            copyHere = common.work_space.resDir() # MATT
            if "USER.outputdir" in self.cfg_params.keys() and os.path.isdir(self.cfg_params["USER.outputdir"]):
                  copyHere = self.cfg_params["USER.outputdir"] + "/" # MATT
             
            cmd = 'lcg-cp --vo cms --verbose gsiftp://' + str(self.server_name) + str(projectUniqName)+'/res/done.tar.gz file://'+copyHere+'done.tar.gz'# MATT
            common.logger.debug(5, cmd)
            copyOut = os.system(cmd +' >& /dev/null')
        except:
            msg = ("Output not yet available")
            raise CrabException(msg)

        zipOut = "done.tar.gz"
        if os.path.exists( copyHere + zipOut ): 
            cwd = os.getcwd()
            os.chdir( copyHere )
            common.logger.debug( 5, 'tar -zxvf ' + zipOut )
            cmd = 'tar -zxvf ' + zipOut 
            cmd += '; mv .tmpDone/* .; rm -drf .tmpDone/'
            cmd_out = runCommand(cmd)
            os.chdir(cwd)
            common.logger.debug( 5, 'rm -f '+copyHere+zipOut )
            cmd = 'rm -f '+copyHere+zipOut
            cmd_out = runCommand(cmd)

            try:
                # file = open(common.work_space.resDir()+"xmlReportFile.xml", "r")
                doc = xml.dom.minidom.parse(common.work_space.resDir()+ "xmlReportFile.xml" )

                task = doc.childNodes[0].childNodes[1].getAttribute("taskName")
                self.countToTjob = int(doc.childNodes[0].childNodes[1].getAttribute("totJob") )

                ended = doc.childNodes[0].childNodes[1].getAttribute("ended")

                addTree = 3
                if doc.childNodes[0].childNodes[3].getAttribute("id") != "all":
                    common.jobDB.load()
                    for job in range( self.countToTjob ):
                        idJob = doc.childNodes[0].childNodes[job+addTree].getAttribute("id")
                        status = doc.childNodes[0].childNodes[job+addTree].getAttribute("status")
                        cleared = doc.childNodes[0].childNodes[job+addTree].getAttribute("cleared")
                        if int(cleared) == 1 and (status == "Done" or status  == "Done (Failed)"):
                            common.jobDB.setStatus( str(int(idJob)-1), "Y" )
                        addTree += 1
                    common.jobDB.save()
            except Exception, ex:
                msg = ("Problems accessing the report file: " + str(ex))
                raise CrabException(msg)

            common.logger.message('Task Completed at '+str(ended)+' %\n')
            msg='Data for project '+str(WorkDirName)+' successfully retrieved from server \n'      
            msg+='and copied in '+copyHere+' \n'
            common.logger.message(msg)
        else:
            common.logger.message("Problems have been encoutered during project transfer. Please check with [status] option if jobs have finished .\n")

        return

Revision:	1.12
Committed:	Fri Jan 4 17:30:56 2008 UTC (17 years, 4 months ago) by slacapra
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_1_0_pre3, CRAB_2_1_0_pre2, CRAB_2_1_0_pre1
Changes since 1.11:	+0 -1 lines
Log Message:	Add support for LSF/CAF direct submission Re-establish a correct inheritance pattern for Scheruled* classes Start to remove some unneeded try: catch: statement and replace them with appropriate if:then: Erase any use of cfg_param as a common block (expecially in conjuction with apmon) and replace it with the user of task DB Several minor cleanup
#	User	Rev	Content
1	spiga	1.1	from Actor import *
2			from crab_util import *
3			import common
4			from ApmonIf import ApmonIf
5			import time
6			from ProgressBar import ProgressBar
7			from TerminalController import TerminalController
8
9	mcinquil	1.2	import xml.dom.minidom
10
11	spiga	1.1	class GetOutputServer(Actor):
12
13			def __init__(self, cfg_params,):
14			self.cfg_params = cfg_params
15	spiga	1.3	try:
16			self.server_name = self.cfg_params['CRAB.server_name'] # gsiftp://pcpg01.cern.ch/data/SEDir/
17	mcinquil	1.11	if not self.server_name.endswith("/"):
18			self.server_name = self.server_name + "/"
19	spiga	1.3	except KeyError:
20			msg = 'No server selected ...'
21			msg = msg + 'Please specify a server in the crab cfg file'
22			raise CrabException(msg)
23	spiga	1.1	return
24
25			def run(self):
26			"""
27			The main method of the class: retrieve the output from server
28			"""
29			common.logger.debug(5, "GetOutput server::run() called")
30
31			start = time.time()
32
33			common.scheduler.checkProxy()
34
35			common.taskDB.load()
36			WorkDirName =os.path.basename(os.path.split(common.work_space.topDir())[0])
37			projectUniqName = 'crab_'+str(WorkDirName)+'_'+common.taskDB.dict("TasKUUID")
38			#Need to add a check on the treashold level
39			# and on the task readness TODO
40			try:
41	mcinquil	1.7	### retrieving project from the server
42			common.logger.message("Retrieving the project from the server...\n")
43	mcinquil	1.2
44			copyHere = common.work_space.resDir() # MATT
45	farinafa	1.8	if "USER.outputdir" in self.cfg_params.keys() and os.path.isdir(self.cfg_params["USER.outputdir"]):
46			copyHere = self.cfg_params["USER.outputdir"] + "/" # MATT
47	mcinquil	1.2
48	mcinquil	1.6	cmd = 'lcg-cp --vo cms --verbose gsiftp://' + str(self.server_name) + str(projectUniqName)+'/res/done.tar.gz file://'+copyHere+'done.tar.gz'# MATT
49	mcinquil	1.2	common.logger.debug(5, cmd)
50	spiga	1.1	copyOut = os.system(cmd +' >& /dev/null')
51			except:
52			msg = ("Output not yet available")
53			raise CrabException(msg)
54
55	mcinquil	1.6	zipOut = "done.tar.gz"
56	mcinquil	1.7	if os.path.exists( copyHere + zipOut ):
57	spiga	1.1	cwd = os.getcwd()
58	mcinquil	1.7	os.chdir( copyHere )
59	spiga	1.1	common.logger.debug( 5, 'tar -zxvf ' + zipOut )
60	spiga	1.3	cmd = 'tar -zxvf ' + zipOut
61	spiga	1.5	cmd += '; mv .tmpDone/* .; rm -drf .tmpDone/'
62	spiga	1.1	cmd_out = runCommand(cmd)
63			os.chdir(cwd)
64	mcinquil	1.7	common.logger.debug( 5, 'rm -f '+copyHere+zipOut )
65			cmd = 'rm -f '+copyHere+zipOut
66	spiga	1.1	cmd_out = runCommand(cmd)
67
68	mcinquil	1.2	try:
69	farinafa	1.10	# file = open(common.work_space.resDir()+"xmlReportFile.xml", "r")
70	mcinquil	1.2	doc = xml.dom.minidom.parse(common.work_space.resDir()+ "xmlReportFile.xml" )
71
72	farinafa	1.10	task = doc.childNodes[0].childNodes[1].getAttribute("taskName")
73	mcinquil	1.2	self.countToTjob = int(doc.childNodes[0].childNodes[1].getAttribute("totJob") )
74
75	mcinquil	1.7	ended = doc.childNodes[0].childNodes[1].getAttribute("ended")
76
77	mcinquil	1.2	addTree = 3
78			if doc.childNodes[0].childNodes[3].getAttribute("id") != "all":
79			common.jobDB.load()
80			for job in range( self.countToTjob ):
81			idJob = doc.childNodes[0].childNodes[job+addTree].getAttribute("id")
82			status = doc.childNodes[0].childNodes[job+addTree].getAttribute("status")
83			cleared = doc.childNodes[0].childNodes[job+addTree].getAttribute("cleared")
84	mcinquil	1.4	if int(cleared) == 1 and (status == "Done" or status == "Done (Failed)"):
85	mcinquil	1.2	common.jobDB.setStatus( str(int(idJob)-1), "Y" )
86			addTree += 1
87			common.jobDB.save()
88			except Exception, ex:
89	farinafa	1.9	msg = ("Problems accessing the report file: " + str(ex))
90	mcinquil	1.2	raise CrabException(msg)
91
92	mcinquil	1.7	common.logger.message('Task Completed at '+str(ended)+' %\n')
93	farinafa	1.9	msg='Data for project '+str(WorkDirName)+' successfully retrieved from server \n'
94	mcinquil	1.7	msg+='and copied in '+copyHere+' \n'
95			common.logger.message(msg)
96	spiga	1.1	else:
97	farinafa	1.9	common.logger.message("Problems have been encoutered during project transfer. Please check with [status] option if jobs have finished .\n")
98	spiga	1.1
99			return
100