1 # A Job consists of many "Tasks".
2 # A task is the run of an external tool, with proper methods for failure handling
4 from Tools.CList import CList
7 NOT_STARTED, IN_PROGRESS, FINISHED, FAILED = range(4)
8 def __init__(self, name):
10 self.resident_tasks = [ ]
11 self.workspace = "/tmp"
20 self.state_changed = CList()
22 self.status = self.NOT_STARTED
24 # description is a dict
25 def fromDescription(self, description):
28 def createDescription(self):
31 def getProgress(self):
32 if self.current_task == len(self.tasks):
34 t = self.tasks[self.current_task]
35 jobprogress = t.weighting * t.progress / float(t.end) + sum([task.weighting for task in self.tasks[:self.current_task]])
36 return int(jobprogress*self.weightScale)
38 progress = property(getProgress)
40 def getStatustext(self):
41 return { self.NOT_STARTED: _("Waiting"), self.IN_PROGRESS: _("In Progress"), self.FINISHED: _("Finished"), self.FAILED: _("Failed") }[self.status]
43 def task_progress_changed_CB(self):
46 def addTask(self, task):
48 self.tasks.append(task)
50 def start(self, callback):
51 assert self.callback is None
52 self.callback = callback
56 self.status = self.IN_PROGRESS
59 sumTaskWeightings = sum([t.weighting for t in self.tasks]) or 1
60 self.weightScale = self.end / float(sumTaskWeightings)
63 if self.current_task == len(self.tasks):
64 if len(self.resident_tasks) == 0:
67 self.status = self.FINISHED
71 print "still waiting for %d resident task(s) %s to finish" % (len(self.resident_tasks), str(self.resident_tasks))
73 self.tasks[self.current_task].run(self.taskCallback, self.task_progress_changed_CB)
76 def taskCallback(self, task, res, stay_resident = False):
77 cb_idx = self.tasks.index(task)
79 if cb_idx not in self.resident_tasks:
80 self.resident_tasks.append(self.current_task)
81 print "task going resident:", task
83 print "task keeps staying resident:", task
86 print ">>> Error:", res
87 self.status = self.FAILED
89 self.callback(self, task, res)
90 if cb_idx != self.current_task:
91 if cb_idx in self.resident_tasks:
92 print "resident task finished:", task
93 self.resident_tasks.remove(cb_idx)
96 self.current_task += 1
100 assert self.status == self.FAILED
104 if self.current_task < len(self.tasks):
105 self.tasks[self.current_task].abort()
106 for i in self.resident_tasks:
107 self.tasks[i].abort()
110 # some Jobs might have a better idea of how to cancel a job
114 def __init__(self, job, name):
116 self.immediate_preconditions = [ ]
117 self.global_preconditions = [ ]
118 self.postconditions = [ ]
119 self.returncode = None
120 self.initial_input = None
129 self.task_progress_changed = None
130 self.output_line = ""
133 def setCommandline(self, cmd, args):
137 def setTool(self, tool):
140 self.global_preconditions.append(ToolExistsPrecondition())
141 self.postconditions.append(ReturncodePostcondition())
143 def checkPreconditions(self, immediate = False):
146 preconditions = self.immediate_preconditions
148 preconditions = self.global_preconditions
149 for precondition in preconditions:
150 if not precondition.check(self):
151 not_met.append(precondition)
154 def run(self, callback, task_progress_changed):
155 failed_preconditions = self.checkPreconditions(True) + self.checkPreconditions(False)
156 if len(failed_preconditions):
157 callback(self, failed_preconditions)
161 self.callback = callback
162 self.task_progress_changed = task_progress_changed
163 from enigma import eConsoleAppContainer
164 self.container = eConsoleAppContainer()
165 self.container.appClosed.append(self.processFinished)
166 self.container.stdoutAvail.append(self.processStdout)
167 self.container.stderrAvail.append(self.processStderr)
169 assert self.cmd is not None
170 assert len(self.args) >= 1
172 if self.cwd is not None:
173 self.container.setCWD(self.cwd)
175 execstr = self.cmd + " ".join(self.args)
176 print "execute:", self.container.execute(execstr), execstr
177 if self.initial_input:
178 self.writeInput(self.initial_input)
183 def cleanup(self, failed):
186 def processStdout(self, data):
187 self.processOutput(data)
189 def processStderr(self, data):
190 self.processOutput(data)
192 def processOutput(self, data):
193 self.output_line += data
195 i = self.output_line.find('\n')
198 self.processOutputLine(self.output_line[:i+1])
199 self.output_line = self.output_line[i+1:]
201 def processOutputLine(self, line):
204 def processFinished(self, returncode):
205 self.returncode = returncode
209 self.container.kill()
210 self.finish(aborted = True)
212 def finish(self, aborted = False):
216 not_met.append(AbortedPostcondition())
218 for postcondition in self.postconditions:
219 if not postcondition.check(self):
220 not_met.append(postcondition)
221 self.cleanup(not_met)
222 self.callback(self, not_met)
227 def writeInput(self, input):
228 self.container.write(input)
230 def getProgress(self):
231 return self.__progress
233 def setProgress(self, progress):
234 if progress > self.end:
238 self.__progress = progress
239 self.task_progress_changed()
241 progress = property(getProgress, setProgress)
243 # The jobmanager will execute multiple jobs, each after another.
244 # later, it will also support suspending jobs (and continuing them after reboot etc)
245 # It also supports a notification when some error occured, and possibly a retry.
248 self.active_jobs = [ ]
249 self.failed_jobs = [ ]
250 self.job_classes = [ ]
251 self.in_background = False
252 self.active_job = None
254 def AddJob(self, job):
255 self.active_jobs.append(job)
259 if self.active_job is None:
260 if len(self.active_jobs):
261 self.active_job = self.active_jobs.pop(0)
262 self.active_job.start(self.jobDone)
264 def jobDone(self, job, task, problems):
265 print "job", job, "completed with", problems, "in", task
266 from Tools import Notifications
267 if self.in_background:
268 from Screens.TaskView import JobView
269 Notifications.AddNotification(JobView, self.active_job)
271 from Screens.MessageBox import MessageBox
272 if problems[0].RECOVERABLE:
273 Notifications.AddNotificationWithCallback(self.errorCB, MessageBox, _("Error: %s\nRetry?") % (problems[0].getErrorMessage(task)))
275 Notifications.AddNotification(MessageBox, _("Error") + (': %s') % (problems[0].getErrorMessage(task)), type = MessageBox.TYPE_ERROR )
278 #self.failed_jobs.append(self.active_job)
280 self.active_job = None
283 def errorCB(self, answer):
286 self.active_job.retry()
288 print "not retrying job."
289 self.failed_jobs.append(self.active_job)
290 self.active_job = None
293 def getPendingJobs(self):
296 list.append(self.active_job)
297 list += self.active_jobs
300 #class PartitionExistsPostcondition:
301 # def __init__(self, device):
302 # self.device = device
304 # def check(self, task):
306 # return os.access(self.device + "part1", os.F_OK)
308 #class CreatePartitionTask(Task):
309 # def __init__(self, device):
310 # Task.__init__(self, _("Create Partition"))
311 # self.device = device
312 # self.setTool("/sbin/sfdisk")
313 # self.args += ["-f", self.device + "disc"]
314 # self.initial_input = "0,\n;\n;\n;\ny\n"
315 # self.postconditions.append(PartitionExistsPostcondition(self.device))
317 #class CreateFilesystemTask(Task):
318 # def __init__(self, device, partition = 1, largefile = True):
319 # Task.__init__(self, _("Create Filesystem"))
320 # self.setTool("/sbin/mkfs.ext")
322 # self.args += ["-T", "largefile"]
323 # self.args.append("-m0")
324 # self.args.append(device + "part%d" % partition)
326 #class FilesystemMountTask(Task):
327 # def __init__(self, device, partition = 1, filesystem = "ext3"):
328 # Task.__init__(self, _("Mounting Filesystem"))
329 # self.setTool("/bin/mount")
330 # if filesystem is not None:
331 # self.args += ["-t", filesystem]
332 # self.args.append(device + "part%d" % partition)
337 def getErrorMessage(self, task):
338 return _("An unknown error occured!") + " (%s @ task %s)" % (self.__class__.__name__, task.__class__.__name__)
340 class WorkspaceExistsPrecondition(Condition):
341 def check(self, task):
342 return os.access(task.job.workspace, os.W_OK)
344 class DiskspacePrecondition(Condition):
345 def __init__(self, diskspace_required):
346 self.diskspace_required = diskspace_required
347 self.diskspace_available = 0
349 def check(self, task):
352 s = os.statvfs(task.job.workspace)
353 self.diskspace_available = s.f_bsize * s.f_bavail
354 return self.diskspace_available >= self.diskspace_required
358 def getErrorMessage(self, task):
359 return _("Not enough diskspace. Please free up some diskspace and try again. (%d MB required, %d MB available)") % (self.diskspace_required / 1024 / 1024, self.diskspace_available / 1024 / 1024)
361 class ToolExistsPrecondition(Condition):
362 def check(self, task):
367 realpath = task.cwd + '/' + task.cmd
368 self.realpath = realpath
369 return os.access(realpath, os.X_OK)
371 def getErrorMessage(self, task):
372 return _("A required tool (%s) was not found.") % (self.realpath)
374 class AbortedPostcondition(Condition):
377 class ReturncodePostcondition(Condition):
378 def check(self, task):
379 return task.returncode == 0
381 #class HDDInitJob(Job):
382 # def __init__(self, device):
383 # Job.__init__(self, _("Initialize Harddisk"))
384 # self.device = device
385 # self.fromDescription(self.createDescription())
387 # def fromDescription(self, description):
388 # self.device = description["device"]
389 # self.addTask(CreatePartitionTask(self.device))
390 # self.addTask(CreateFilesystemTask(self.device))
391 # self.addTask(FilesystemMountTask(self.device))
393 # def createDescription(self):
394 # return {"device": self.device}
396 job_manager = JobManager()