X-Git-Url: https://git.cweiske.de/enigma2.git/blobdiff_plain/b5513ae969d3d8e5a0870d21cf6c8499987b1135..0c2185a46606c2ac3e41205fe92e6d5ba4ead1b9:/lib/python/Components/Task.py diff --git a/lib/python/Components/Task.py b/lib/python/Components/Task.py index 075324b0..659660e8 100644 --- a/lib/python/Components/Task.py +++ b/lib/python/Components/Task.py @@ -7,6 +7,7 @@ class Job(object): NOT_STARTED, IN_PROGRESS, FINISHED, FAILED = range(4) def __init__(self, name): self.tasks = [ ] + self.resident_tasks = [ ] self.workspace = "/tmp" self.current_task = 0 self.callback = None @@ -36,6 +37,9 @@ class Job(object): progress = property(getProgress) + def getStatustext(self): + return { self.NOT_STARTED: _("Waiting"), self.IN_PROGRESS: _("In Progress"), self.FINISHED: _("Finished"), self.FAILED: _("Failed") }[self.status] + def task_progress_changed_CB(self): self.state_changed() @@ -46,41 +50,67 @@ class Job(object): def start(self, callback): assert self.callback is None self.callback = callback + self.restart() + + def restart(self): self.status = self.IN_PROGRESS self.state_changed() self.runNext() - sumTaskWeightings = sum([t.weighting for t in self.tasks]) - self.weightScale = (self.end+1) / float(sumTaskWeightings) + sumTaskWeightings = sum([t.weighting for t in self.tasks]) or 1 + self.weightScale = self.end / float(sumTaskWeightings) def runNext(self): if self.current_task == len(self.tasks): - self.callback(self, []) - self.status = self.FINISHED - self.state_changed() + if len(self.resident_tasks) == 0: + cb = self.callback + self.callback = None + self.status = self.FINISHED + self.state_changed() + cb(self, None, []) + else: + print "still waiting for %d resident task(s) %s to finish" % (len(self.resident_tasks), str(self.resident_tasks)) else: - self.tasks[self.current_task].run(self.taskCallback,self.task_progress_changed_CB) + self.tasks[self.current_task].run(self.taskCallback, self.task_progress_changed_CB) self.state_changed() - def taskCallback(self, res): + def taskCallback(self, task, res, stay_resident = False): + cb_idx = self.tasks.index(task) + if stay_resident: + if cb_idx not in self.resident_tasks: + self.resident_tasks.append(self.current_task) + print "task going resident:", task + else: + print "task keeps staying resident:", task + return if len(res): print ">>> Error:", res self.status = self.FAILED self.state_changed() - self.callback(self, res) - else: - self.state_changed(); + self.callback(self, task, res) + if cb_idx != self.current_task: + if cb_idx in self.resident_tasks: + print "resident task finished:", task + self.resident_tasks.remove(cb_idx) + if res == []: + self.state_changed() self.current_task += 1 self.runNext() + def retry(self): + assert self.status == self.FAILED + self.restart() + def abort(self): if self.current_task < len(self.tasks): self.tasks[self.current_task].abort() + for i in self.resident_tasks: + self.tasks[i].abort() def cancel(self): # some Jobs might have a better idea of how to cancel a job self.abort() -class Task(object) : +class Task(object): def __init__(self, job, name): self.name = name self.immediate_preconditions = [ ] @@ -97,6 +127,7 @@ class Task(object) : self.cwd = "/tmp" self.args = [ ] self.task_progress_changed = None + self.output_line = "" job.addTask(self) def setCommandline(self, cmd, args): @@ -123,7 +154,7 @@ class Task(object) : def run(self, callback, task_progress_changed): failed_preconditions = self.checkPreconditions(True) + self.checkPreconditions(False) if len(failed_preconditions): - callback(failed_preconditions) + callback(self, failed_preconditions) return self.prepare() @@ -132,7 +163,8 @@ class Task(object) : from enigma import eConsoleAppContainer self.container = eConsoleAppContainer() self.container.appClosed.get().append(self.processFinished) - self.container.dataAvail.get().append(self.processOutput) + self.container.stdoutAvail.get().append(self.processStdout) + self.container.stderrAvail.get().append(self.processStderr) assert self.cmd is not None assert len(self.args) >= 1 @@ -140,7 +172,7 @@ class Task(object) : if self.cwd is not None: self.container.setCWD(self.cwd) - print "execute:", self.container.execute(self.cmd, self.args), self.cmd, self.args + print "execute:", self.container.execute(self.cmd, self.args), self.cmd, " ".join(self.args) if self.initial_input: self.writeInput(self.initial_input) @@ -149,8 +181,23 @@ class Task(object) : def cleanup(self, failed): pass + + def processStdout(self, data): + self.processOutput(data) + + def processStderr(self, data): + self.processOutput(data) def processOutput(self, data): + self.output_line += data + while True: + i = self.output_line.find('\n') + if i == -1: + break + self.processOutputLine(self.output_line[:i+1]) + self.output_line = self.output_line[i+1:] + + def processOutputLine(self, line): pass def processFinished(self, returncode): @@ -170,9 +217,8 @@ class Task(object) : for postcondition in self.postconditions: if not postcondition.check(self): not_met.append(postcondition) - self.cleanup(not_met) - self.callback(not_met) + self.callback(self, not_met) def afterRun(self): pass @@ -184,17 +230,24 @@ class Task(object) : return self.__progress def setProgress(self, progress): - print "progress now", progress + if progress > self.end: + progress = self.end + if progress < 0: + progress = 0 self.__progress = progress self.task_progress_changed() progress = property(getProgress, setProgress) +# The jobmanager will execute multiple jobs, each after another. +# later, it will also support suspending jobs (and continuing them after reboot etc) +# It also supports a notification when some error occured, and possibly a retry. class JobManager: def __init__(self): self.active_jobs = [ ] self.failed_jobs = [ ] self.job_classes = [ ] + self.in_background = False self.active_job = None def AddJob(self, job): @@ -207,14 +260,41 @@ class JobManager: self.active_job = self.active_jobs.pop(0) self.active_job.start(self.jobDone) - def jobDone(self, job, problems): - print "job", job, "completed with", problems + def jobDone(self, job, task, problems): + print "job", job, "completed with", problems, "in", task + from Tools import Notifications + if self.in_background: + from Screens.TaskView import JobView + Notifications.AddNotification(JobView, self.active_job) if problems: - self.failed_jobs.append(self.active_job) + from Screens.MessageBox import MessageBox + if problems[0].RECOVERABLE: + Notifications.AddNotificationWithCallback(self.errorCB, MessageBox, _("Error: %s\nRetry?") % (problems[0].getErrorMessage(task))) + else: + Notifications.AddNotification(MessageBox, _("Error") + (': %s') % (problems[0].getErrorMessage(task)), type = MessageBox.TYPE_ERROR ) + self.errorCB(False) + return + #self.failed_jobs.append(self.active_job) self.active_job = None self.kick() + def errorCB(self, answer): + if answer: + print "retrying job" + self.active_job.retry() + else: + print "not retrying job." + self.failed_jobs.append(self.active_job) + self.active_job = None + self.kick() + + def getPendingJobs(self): + list = [ ] + if self.active_job: + list.append(self.active_job) + list += self.active_jobs + return list # some examples: #class PartitionExistsPostcondition: # def __init__(self, device): @@ -250,35 +330,50 @@ class JobManager: # self.args += ["-t", filesystem] # self.args.append(device + "part%d" % partition) -class WorkspaceExistsPrecondition: +class Condition: + RECOVERABLE = False + + def getErrorMessage(self, task): + return _("An unknown error occured!") + " (%s @ task %s)" % (self.__class__.__name__, task.__class__.__name__) + +class WorkspaceExistsPrecondition(Condition): def check(self, task): return os.access(task.job.workspace, os.W_OK) -class DiskspacePrecondition: +class DiskspacePrecondition(Condition): def __init__(self, diskspace_required): self.diskspace_required = diskspace_required + self.diskspace_available = 0 def check(self, task): import os try: s = os.statvfs(task.job.workspace) - return s.f_bsize * s.f_bavail >= self.diskspace_required + self.diskspace_available = s.f_bsize * s.f_bavail + return self.diskspace_available >= self.diskspace_required except OSError: return False -class ToolExistsPrecondition: + def getErrorMessage(self, task): + 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) + +class ToolExistsPrecondition(Condition): def check(self, task): import os if task.cmd[0]=='/': realpath = task.cmd else: - realpath = self.cwd + '/' + self.cmd + realpath = task.cwd + '/' + task.cmd + self.realpath = realpath return os.access(realpath, os.X_OK) -class AbortedPostcondition: + def getErrorMessage(self, task): + return _("A required tool (%s) was not found.") % (self.realpath) + +class AbortedPostcondition(Condition): pass -class ReturncodePostcondition: +class ReturncodePostcondition(Condition): def check(self, task): return task.returncode == 0