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"
19 self.afterEvent = None
21 self.state_changed = CList()
23 self.status = self.NOT_STARTED
25 # description is a dict
26 def fromDescription(self, description):
29 def createDescription(self):
32 def getProgress(self):
33 if self.current_task == len(self.tasks):
35 t = self.tasks[self.current_task]
36 jobprogress = t.weighting * t.progress / float(t.end) + sum([task.weighting for task in self.tasks[:self.current_task]])
37 return int(jobprogress*self.weightScale)
39 progress = property(getProgress)
41 def getStatustext(self):
42 return { self.NOT_STARTED: _("Waiting"), self.IN_PROGRESS: _("In Progress"), self.FINISHED: _("Finished"), self.FAILED: _("Failed") }[self.status]
44 def task_progress_changed_CB(self):
47 def addTask(self, task):
49 task.task_progress_changed = self.task_progress_changed_CB
50 self.tasks.append(task)
52 def start(self, callback):
53 assert self.callback is None
54 self.callback = callback
58 self.status = self.IN_PROGRESS
61 sumTaskWeightings = sum([t.weighting for t in self.tasks]) or 1
62 self.weightScale = self.end / float(sumTaskWeightings)
65 if self.current_task == len(self.tasks):
66 if len(self.resident_tasks) == 0:
67 self.status = self.FINISHED
69 self.callback(self, None, [])
72 print "still waiting for %d resident task(s) %s to finish" % (len(self.resident_tasks), str(self.resident_tasks))
74 self.tasks[self.current_task].run(self.taskCallback)
77 def taskCallback(self, task, res, stay_resident = False):
78 cb_idx = self.tasks.index(task)
80 if cb_idx not in self.resident_tasks:
81 self.resident_tasks.append(self.current_task)
82 print "task going resident:", task
84 print "task keeps staying resident:", task
87 print ">>> Error:", res
88 self.status = self.FAILED
90 self.callback(self, task, res)
91 if cb_idx != self.current_task:
92 if cb_idx in self.resident_tasks:
93 print "resident task finished:", task
94 self.resident_tasks.remove(cb_idx)
97 self.current_task += 1
101 assert self.status == self.FAILED
105 if self.current_task < len(self.tasks):
106 self.tasks[self.current_task].abort()
107 for i in self.resident_tasks:
108 self.tasks[i].abort()
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
130 self.task_progress_changed = None
131 self.output_line = ""
133 self.container = None
135 def setCommandline(self, cmd, args):
139 def setTool(self, tool):
142 self.global_preconditions.append(ToolExistsPrecondition())
143 self.postconditions.append(ReturncodePostcondition())
145 def setCmdline(self, cmdline):
146 self.cmdline = cmdline
148 def checkPreconditions(self, immediate = False):
151 preconditions = self.immediate_preconditions
153 preconditions = self.global_preconditions
154 for precondition in preconditions:
155 if not precondition.check(self):
156 not_met.append(precondition)
159 def run(self, callback):
160 failed_preconditions = self.checkPreconditions(True) + self.checkPreconditions(False)
161 if len(failed_preconditions):
162 callback(self, failed_preconditions)
166 self.callback = callback
167 from enigma import eConsoleAppContainer
168 self.container = eConsoleAppContainer()
169 self.container.appClosed.append(self.processFinished)
170 self.container.stdoutAvail.append(self.processStdout)
171 self.container.stderrAvail.append(self.processStderr)
173 if self.cwd is not None:
174 self.container.setCWD(self.cwd)
176 if not self.cmd and self.cmdline:
177 print "execute:", self.container.execute(self.cmdline), self.cmdline
179 assert self.cmd is not None
180 assert len(self.args) >= 1
181 print "execute:", self.container.execute(self.cmd, *self.args), ' '.join(self.args)
182 if self.initial_input:
183 self.writeInput(self.initial_input)
188 def cleanup(self, failed):
191 def processStdout(self, data):
192 self.processOutput(data)
194 def processStderr(self, data):
195 self.processOutput(data)
197 def processOutput(self, data):
198 self.output_line += data
200 i = self.output_line.find('\n')
203 self.processOutputLine(self.output_line[:i+1])
204 self.output_line = self.output_line[i+1:]
206 def processOutputLine(self, line):
209 def processFinished(self, returncode):
210 self.returncode = returncode
215 self.container.kill()
216 self.finish(aborted = True)
218 def finish(self, aborted = False):
222 not_met.append(AbortedPostcondition())
224 for postcondition in self.postconditions:
225 if not postcondition.check(self):
226 not_met.append(postcondition)
227 self.cleanup(not_met)
228 self.callback(self, not_met)
233 def writeInput(self, input):
234 self.container.write(input)
236 def getProgress(self):
237 return self.__progress
239 def setProgress(self, progress):
240 if progress > self.end:
244 self.__progress = progress
245 if self.task_progress_changed:
246 self.task_progress_changed()
248 progress = property(getProgress, setProgress)
250 # The jobmanager will execute multiple jobs, each after another.
251 # later, it will also support suspending jobs (and continuing them after reboot etc)
252 # It also supports a notification when some error occured, and possibly a retry.
255 self.active_jobs = [ ]
256 self.failed_jobs = [ ]
257 self.job_classes = [ ]
258 self.in_background = False
259 self.active_job = None
261 def AddJob(self, job):
262 self.active_jobs.append(job)
266 if self.active_job is None:
267 if len(self.active_jobs):
268 self.active_job = self.active_jobs.pop(0)
269 self.active_job.start(self.jobDone)
271 def jobDone(self, job, task, problems):
272 print "job", job, "completed with", problems, "in", task
273 from Tools import Notifications
274 if self.in_background:
275 from Screens.TaskView import JobView
276 self.in_background = False
277 Notifications.AddNotification(JobView, self.active_job)
279 from Screens.MessageBox import MessageBox
280 if problems[0].RECOVERABLE:
281 Notifications.AddNotificationWithCallback(self.errorCB, MessageBox, _("Error: %s\nRetry?") % (problems[0].getErrorMessage(task)))
283 Notifications.AddNotification(MessageBox, _("Error") + (': %s') % (problems[0].getErrorMessage(task)), type = MessageBox.TYPE_ERROR )
286 #self.failed_jobs.append(self.active_job)
288 self.active_job = None
291 def errorCB(self, answer):
294 self.active_job.retry()
296 print "not retrying job."
297 self.failed_jobs.append(self.active_job)
298 self.active_job = None
301 def getPendingJobs(self):
304 list.append(self.active_job)
305 list += self.active_jobs
309 #class PartitionExistsPostcondition:
310 # def __init__(self, device):
311 # self.device = device
313 # def check(self, task):
315 # return os.access(self.device + "part1", os.F_OK)
317 #class CreatePartitionTask(Task):
318 # def __init__(self, device):
319 # Task.__init__(self, _("Create Partition"))
320 # self.device = device
321 # self.setTool("/sbin/sfdisk")
322 # self.args += ["-f", self.device + "disc"]
323 # self.initial_input = "0,\n;\n;\n;\ny\n"
324 # self.postconditions.append(PartitionExistsPostcondition(self.device))
326 #class CreateFilesystemTask(Task):
327 # def __init__(self, device, partition = 1, largefile = True):
328 # Task.__init__(self, _("Create Filesystem"))
329 # self.setTool("/sbin/mkfs.ext")
331 # self.args += ["-T", "largefile"]
332 # self.args.append("-m0")
333 # self.args.append(device + "part%d" % partition)
335 #class FilesystemMountTask(Task):
336 # def __init__(self, device, partition = 1, filesystem = "ext3"):
337 # Task.__init__(self, _("Mounting Filesystem"))
338 # self.setTool("/bin/mount")
339 # if filesystem is not None:
340 # self.args += ["-t", filesystem]
341 # self.args.append(device + "part%d" % partition)
346 def getErrorMessage(self, task):
347 return _("An unknown error occured!") + " (%s @ task %s)" % (self.__class__.__name__, task.__class__.__name__)
349 class WorkspaceExistsPrecondition(Condition):
350 def check(self, task):
351 return os.access(task.job.workspace, os.W_OK)
353 class DiskspacePrecondition(Condition):
354 def __init__(self, diskspace_required):
355 self.diskspace_required = diskspace_required
356 self.diskspace_available = 0
358 def check(self, task):
361 s = os.statvfs(task.job.workspace)
362 self.diskspace_available = s.f_bsize * s.f_bavail
363 return self.diskspace_available >= self.diskspace_required
367 def getErrorMessage(self, task):
368 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)
370 class ToolExistsPrecondition(Condition):
371 def check(self, task):
375 self.realpath = task.cmd
376 print "[Task.py][ToolExistsPrecondition] WARNING: usage of absolute paths for tasks should be avoided!"
377 return os.access(self.realpath, os.X_OK)
379 self.realpath = task.cmd
380 path = os.environ.get('PATH', '').split(os.pathsep)
381 path.append(task.cwd + '/')
382 absolutes = filter(lambda file: os.access(file, os.X_OK), map(lambda directory, file = task.cmd: os.path.join(directory, file), path))
383 if len(absolutes) > 0:
384 self.realpath = task.cmd[0]
388 def getErrorMessage(self, task):
389 return _("A required tool (%s) was not found.") % (self.realpath)
391 class AbortedPostcondition(Condition):
392 def getErrorMessage(self, task):
393 return "Cancelled upon user request"
395 class ReturncodePostcondition(Condition):
396 def check(self, task):
397 return task.returncode == 0
399 #class HDDInitJob(Job):
400 # def __init__(self, device):
401 # Job.__init__(self, _("Initialize Harddisk"))
402 # self.device = device
403 # self.fromDescription(self.createDescription())
405 # def fromDescription(self, description):
406 # self.device = description["device"]
407 # self.addTask(CreatePartitionTask(self.device))
408 # self.addTask(CreateFilesystemTask(self.device))
409 # self.addTask(FilesystemMountTask(self.device))
411 # def createDescription(self):
412 # return {"device": self.device}
414 job_manager = JobManager()