Source Code for Module arcjobtool.ArcGuiThreads

  1  # -*- coding: iso-8859-15 -*- 
  2  # generated by wxGlade HG on Tue May 19 21:48:13 2009 
  3   
  4  # 
  5  # ArcGui main window threads 
  6  # 
  7   
  8  """ 
  9  Implements the arcgui user interface threads. 
 10  """ 
 11   
 12  import os, sys, threading, types, wx, shutil 
 13   
 14  from datetime import datetime 
 15   
 16  try: 
 17      import arc 
 18  except: 
 19      print "ARC1 Python binding not found. Please check search paths." 
 20      sys.exit(-1) 
 21   
 22  # New events defined by ARCGUI 
 23   
 24  EVT_PROGRESS_UPDATE_TYPE = wx.NewEventType() 
 25  EVT_PROGRESS_UPDATE = wx.PyEventBinder(EVT_PROGRESS_UPDATE_TYPE, 1) 
 26  EVT_LOG_UPDATE_TYPE = wx.NewEventType() 
 27  EVT_LOG_UPDATE = wx.PyEventBinder(EVT_LOG_UPDATE_TYPE, 1) 
 28  EVT_UPDATE_STATUS_DONE_TYPE = wx.NewEventType() 
 29  EVT_UPDATE_STATUS_DONE = wx.PyEventBinder(EVT_UPDATE_STATUS_DONE_TYPE, 1) 
 30  EVT_RETRIEVE_JOBS_DONE_TYPE = wx.NewEventType() 
 31  EVT_RETRIEVE_JOBS_DONE = wx.PyEventBinder(EVT_RETRIEVE_JOBS_DONE_TYPE, 1) 
 32  EVT_CLEAN_JOBS_DONE_TYPE = wx.NewEventType() 
 33  EVT_CLEAN_JOBS_DONE = wx.PyEventBinder(EVT_CLEAN_JOBS_DONE_TYPE, 1) 
 34  EVT_KILL_JOBS_DONE_TYPE = wx.NewEventType() 
 35  EVT_KILL_JOBS_DONE = wx.PyEventBinder(EVT_KILL_JOBS_DONE_TYPE, 1) 
 36  EVT_SUBMIT_JOB_DONE_TYPE = wx.NewEventType() 
 37  EVT_SUBMIT_JOB_DONE = wx.PyEventBinder(EVT_SUBMIT_JOB_DONE_TYPE, 1) 
 38   
 39 -class UpdateProgressEvent(wx.PyCommandEvent): 
 40      """ 
 41      Event sent by worker threads to update progress. 
 42      """ 
 43 -    def __init__(self, etype, eid, info="update"): 
 44          wx.PyCommandEvent.__init__(self, etype, eid) 
 45          self.__info = info 
 46           
 47 -    def getInfo(self): 
 48          return self.__info 
 49       
 50      info = property(getInfo) 
 51   
 52 -class UpdateLogEvent(wx.PyCommandEvent): 
 53      """ 
 54      Event sent by worker threads to update log. 
 55      """ 
 56 -    def __init__(self, etype, eid, output="update"): 
 57          wx.PyCommandEvent.__init__(self, etype, eid) 
 58          self.__output = output 
 59           
 60 -    def getOutput(self): 
 61          return self.__output 
 62       
 63      output = property(getOutput) 
 64   
 65 -class UpdateStatusDoneEvent(wx.PyCommandEvent): 
 66      """ 
 67      Event sent when the worker thread UpdateStatusThread has finished 
 68      processing. 
 69      """ 
 70 -    def __init__(self, etype, eid): 
 71          wx.PyCommandEvent.__init__(self, etype, eid) 
 72   
 73 -class RetrieveJobsDoneEvent(wx.PyCommandEvent): 
 74      """ 
 75      Event sent when the worker thread RetrieveJobsThread has finished 
 76      processing. 
 77      """ 
 78 -    def __init__(self, etype, eid): 
 79          wx.PyCommandEvent.__init__(self, etype, eid) 
 80           
 81 -class CleanJobsDoneEvent(wx.PyCommandEvent): 
 82      """ 
 83      Event sent when the worker thread CleanJobsThread has finished 
 84      processing. 
 85      """ 
 86 -    def __init__(self, etype, eid): 
 87          wx.PyCommandEvent.__init__(self, etype, eid) 
 88           
 89 -class KillJobsDoneEvent(wx.PyCommandEvent): 
 90      """ 
 91      Event sent when the worker thread KillJobsThread has finished 
 92      processing. 
 93      """ 
 94 -    def __init__(self, etype, eid): 
 95          wx.PyCommandEvent.__init__(self, etype, eid) 
 96           
 97 -class SubmitJobDoneEvent(wx.PyCommandEvent): 
 98      """ 
 99      Event sent when the worker thread SubmitJobThread has finished 
100      processing. 
101      """ 
102 -    def __init__(self, etype, eid, resultList=[]): 
103          wx.PyCommandEvent.__init__(self, etype, eid) 
104          self.__resultList = resultList 
105           
106 -    def getResultList(self): 
107          return self.__resultList 
108       
109      resultList = property(getResultList) 
110               
111 -class ArcGuiThread(threading.Thread): 
112 -    def __init__(self, parent, arcClient): 
113          """ 
114          Base class for all ArcGui threads. 
115          """ 
116          threading.Thread.__init__(self) 
117          self.__stopEvent = threading.Event() 
118          self.__parent = parent 
119          self.__logger = None 
120          self.__arcClient = arcClient 
121          self.__arcClient.updateProgress = self.doProgress 
122          self.__debugLevel = self.__arcClient.debugLevel 
123           
124 -    def stop(self): 
125          """ 
126          Set the stop event. To signal that the thread should stop. 
127          """ 
128          if not self.__stopEvent.isSet(): 
129              self.__stopEvent.set() 
130           
131 -    def isStopped(self): 
132          """ 
133          Return thread stopped status. 
134          """ 
135          return self.__stopEvent.isSet() 
136           
137 -    def logMsg(self, level, msg): 
138          """ 
139          Log message to logger. 
140          """ 
141          if self.__logger!=None: 
142              self.__logger.msg(level, msg) 
143           
144 -    def logInfoMsg(self, msg): 
145          """ 
146          Log informational message to logger. 
147          """ 
148          if self.__logger!=None: 
149              self.logMsg(arc.INFO, msg) 
150       
151 -    def logDebugMsg(self, msg): 
152          """ 
153          Log debug message to logger. 
154          """ 
155          if self.__logger!=None: 
156              self.logMsg(arc.DEBUG, msg) 
157   
158 -    def logErrorMsg(self, msg): 
159          """ 
160          Log error message to logger. 
161          """ 
162          if self.__logger!=None: 
163              self.logMsg(arc.ERROR, msg) 
164           
165      #def write(self, string): 
166      #    evt = UpdateLogEvent(EVT_LOG_UPDATE_TYPE, -1, string) 
167      #    wx.PostEvent(self.__parent, evt) 
168   
169 -    def doProgress(self, message): 
170          """ 
171          Send message (EVT_PROGRESS_UPDATE_TYPE) to 
172          parent window for updating progress messages. 
173          """ 
174          evt = UpdateProgressEvent(EVT_PROGRESS_UPDATE_TYPE, -1, message) 
175          wx.PostEvent(self.__parent, evt) 
176   
177 -    def getParent(self): 
178          """ 
179          Return parent instance. 
180          """ 
181          return self.__parent 
182       
183 -    def getArcClient(self): 
184          """ 
185          Return ArcClient instance. 
186          """ 
187          return self.__arcClient 
188       
189 -    def setDebugLevel(self, debugLevel): 
190          """ 
191          Set debug level (deprecated). 
192          """ 
193          self.__debugLevel = debugLevel 
194           
195 -    def getDebugLevel(self): 
196          """ 
197          Return debug level (deprecated). 
198          """ 
199          return self.__debugLevel 
200       
201 -    def setLogger(self, logger): 
202          """ 
203          Set logger to be used by thread. 
204          """ 
205          self.__logger = logger 
206           
207 -    def getLogger(self): 
208          """ 
209          Return assigned logger. 
210          """ 
211          return self.__logger 
212       
213      parent = property(getParent) 
214      arcClient = property(getArcClient) 
215      debugLevel = property(getDebugLevel, setDebugLevel) 
216      logger = property(getLogger, setLogger) 
217   
218 -class UpdateStatusThread(ArcGuiThread): 
219      """ 
220      Worker thread for updating job status. 
221      """ 
222 -    def run(self): 
223          """ 
224          Main thread method for getting job status information. 
225          """ 
226          self.logInfoMsg("UpdateStatusThread Starting.") 
227          self.arcClient.updateStatus() 
228          evt = UpdateStatusDoneEvent(EVT_UPDATE_STATUS_DONE_TYPE, -1) 
229          wx.PostEvent(self.parent, evt) 
230          self.logInfoMsg("UpdateStatusThread Stopping.") 
231           
232 -class RetrieveJobsThread(ArcGuiThread): 
233      """ 
234      Worker thread for retrieving (downloading) jobs. 
235      """ 
236 -    def __init__(self, parent, arcClient, jobIds, downloadDir, downloadHistoryFile): 
237          """ 
238          Class constructor 
239          """ 
240          ArcGuiThread.__init__(self, parent, arcClient) 
241          self.__jobIds = jobIds 
242          self.__downloadDir = downloadDir 
243          self.__removeJobDir = True 
244          self.__downloadHistoryFile = downloadHistoryFile 
245           
246 -    def run(self): 
247          """ 
248          Main thread method for retrieving jobs from the grid. 
249          """ 
250          self.logInfoMsg("RetrieveJobsThread starting.") 
251          if type(self.__downloadDir) is types.StringType: 
252              self.logInfoMsg("Download to single dir.") 
253              self.arcClient.downloadDir = self.__downloadDir 
254              self.arcClient.get(self.__jobIds) 
255          elif type(self.__downloadDir) is types.ListType: 
256              self.logInfoMsg("Download to separate dirs.") 
257              for jobId, downloadDir in zip(self.__jobIds, self.__downloadDir): 
258                  if self.isStopped(): 
259                      break 
260                  self.logInfoMsg("downloading %s to %s." % (jobId, downloadDir)) 
261                  self.arcClient.downloadDir = downloadDir 
262                  self.arcClient.get([jobId]) 
263                   
264                  if self.__removeJobDir: 
265                      sessionId = jobId.split("/")[-1] 
266                      jobDir = os.path.join(self.arcClient.downloadDir, sessionId) 
267                      if os.path.isdir(jobDir): 
268                          print "Moving from ",jobDir,"to",self.arcClient.downloadDir 
269                          os.system("mv %s/* %s" % (jobDir, self.arcClient.downloadDir)) 
270                          shutil.rmtree(jobDir) 
271                          if self.__downloadHistoryFile!="": 
272                              historyFile = open(self.__downloadHistoryFile, "a") 
273                              historyFile.write("%s;%s;%s\n" % (jobId, str(datetime.now()), self.arcClient.downloadDir)) 
274                              historyFile.close() 
275                          #os.remove(jobDir) 
276                          #shutil.move(jobDir, self.arcClient.downloadDir) 
277                   
278          evt = RetrieveJobsDoneEvent(EVT_RETRIEVE_JOBS_DONE_TYPE, -1) 
279          wx.PostEvent(self.parent, evt) 
280          self.logInfoMsg("RetrieveJobsThread stopping.") 
281                   
282 -class CleanJobsThread(ArcGuiThread): 
283      """ 
284      Worker thread for cleaning jobs. 
285      """ 
286 -    def __init__(self, parent, arcClient, jobIds, force=False): 
287          """ 
288          Class constructor 
289          """ 
290          ArcGuiThread.__init__(self, parent, arcClient) 
291          self.__jobIds = jobIds 
292          self.__force = force 
293           
294 -    def run(self): 
295          """ 
296          Main thread method for cleaning a job(s) 
297          """ 
298          self.logInfoMsg("CleanJobsThread starting.") 
299          self.arcClient.clean(self.__jobIds, force=self.__force) 
300          evt = CleanJobsDoneEvent(EVT_CLEAN_JOBS_DONE_TYPE, -1) 
301          wx.PostEvent(self.parent, evt) 
302          self.logInfoMsg("CleanJobsThread stopping.") 
303               
304 -class KillJobsThread(ArcGuiThread): 
305      """ 
306      Worker thread for killing jobs. 
307      """ 
308 -    def __init__(self, parent, arcClient, jobIds, force=False): 
309          """ 
310          Class constructor 
311          """ 
312          ArcGuiThread.__init__(self, parent, arcClient) 
313          self.__jobIds = jobIds 
314          self.__force = force 
315           
316 -    def run(self): 
317          """ 
318          Main thread mehtod for killing jobs. 
319          """ 
320          self.logInfoMsg("KillJobsThread starting.") 
321          self.arcClient.kill(self.__jobIds, force=self.__force) 
322          evt = KillJobsDoneEvent(EVT_KILL_JOBS_DONE_TYPE, -1) 
323          wx.PostEvent(self.parent, evt) 
324          self.logInfoMsg("KillJobsThread stopping.") 
325           
326 -class SubmitJobThread(ArcGuiThread): 
327      """ 
328      Worker thread for submitting a job. 
329      """ 
330 -    def __init__(self, parent, arcClient, jobList): 
331          """ 
332          Class constructor 
333          """ 
334          ArcGuiThread.__init__(self, parent, arcClient) 
335          self.__jobList = jobList 
336           
337 -    def run(self): 
338          """ 
339          Main thread method for submitting a job. 
340          """ 
341          self.logInfoMsg("SubmitJobThread starting.") 
342          self.arcClient.findTargets() 
343          #self.arcClient.filterTargets(self.__jobList[0])    
344          resultList = self.arcClient.submitJobList(self.__jobList) 
345          evt = SubmitJobDoneEvent(EVT_SUBMIT_JOB_DONE_TYPE, -1, resultList) 
346          wx.PostEvent(self.parent, evt) 
347          self.logInfoMsg("SubmitJobThread stopping.") 
348