lib/python/Components/Task.py

   1 # A Job consists of many "Tasks".
   2 # A task is the run of an external tool, with proper methods for failure handling
   3
   4 from Tools.CList import CList
   5
   6 class Job(object):
   7         NOT_STARTED, IN_PROGRESS, FINISHED, FAILED = range(4)
   8         def __init__(self, name):
   9                 self.tasks = [ ]
  10                 self.resident_tasks = [ ]
  11                 self.workspace = "/tmp"
  12                 self.current_task = 0
  13                 self.callback = None
  14                 self.name = name
  15                 self.finished = False
  16                 self.end = 100
  17                 self.__progress = 0
  18                 self.weightScale = 1
  19
  20                 self.state_changed = CList()
  21
  22                 self.status = self.NOT_STARTED
  23
  24         # description is a dict
  25         def fromDescription(self, description):
  26                 pass
  27
  28         def createDescription(self):
  29                 return None
  30
  31         def getProgress(self):
  32                 if self.current_task == len(self.tasks):
  33                         return self.end
  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)
  37
  38         progress = property(getProgress)
  39
  40         def task_progress_changed_CB(self):
  41                 self.state_changed()
  42
  43         def addTask(self, task):
  44                 task.job = self
  45                 self.tasks.append(task)
  46
  47         def start(self, callback):
  48                 assert self.callback is None
  49                 self.callback = callback
  50                 self.restart()
  51
  52         def restart(self):
  53                 self.status = self.IN_PROGRESS
  54                 self.state_changed()
  55                 self.runNext()
  56                 sumTaskWeightings = sum([t.weighting for t in self.tasks])
  57                 self.weightScale = self.end / float(sumTaskWeightings)
  58
  59         def runNext(self):
  60                 if self.current_task == len(self.tasks):
  61                         if len(self.resident_tasks) == 0:
  62                                 cb = self.callback
  63                                 self.callback = None
  64                                 self.status = self.FINISHED
  65                                 self.state_changed()
  66                                 cb(self, None, [])
  67                         else:
  68                                 print "still waiting for %d resident task(s) %s to finish" % (len(self.resident_tasks), str(self.resident_tasks))
  69                 else:
  70                         self.tasks[self.current_task].run(self.taskCallback, self.task_progress_changed_CB)
  71                         self.state_changed()
  72
  73         def taskCallback(self, task, res, stay_resident = False):
  74                 cb_idx = self.tasks.index(task)
  75                 if stay_resident:
  76                         if cb_idx not in self.resident_tasks:
  77                                 self.resident_tasks.append(self.current_task)
  78                                 print "task going resident:", task
  79                         else:
  80                                 print "task keeps staying resident:", task
  81                                 return
  82                 if len(res):
  83                         print ">>> Error:", res
  84                         self.status = self.FAILED
  85                         self.state_changed()
  86                         self.callback(self, task, res)
  87                 if cb_idx != self.current_task:
  88                         if cb_idx in self.resident_tasks:
  89                                 print "resident task finished:", task
  90                                 self.resident_tasks.remove(cb_idx)
  91                 if res == []:
  92                         self.state_changed()
  93                         self.current_task += 1
  94                         self.runNext()
  95
  96         def retry(self):
  97                 assert self.status == self.FAILED
  98                 self.restart()
  99
 100         def abort(self):
 101                 if self.current_task < len(self.tasks):
 102                         self.tasks[self.current_task].abort()
 103                 for i in self.resident_tasks:
 104                         self.tasks[i].abort()
 105
 106         def cancel(self):
 107                 # some Jobs might have a better idea of how to cancel a job
 108                 self.abort()
 109
 110 class Task(object):
 111         def __init__(self, job, name):
 112                 self.name = name
 113                 self.immediate_preconditions = [ ]
 114                 self.global_preconditions = [ ]
 115                 self.postconditions = [ ]
 116                 self.returncode = None
 117                 self.initial_input = None
 118                 self.job = None
 119
 120                 self.end = 100
 121                 self.weighting = 100
 122                 self.__progress = 0
 123                 self.cmd = None
 124                 self.cwd = "/tmp"
 125                 self.args = [ ]
 126                 self.task_progress_changed = None
 127                 self.output_line = ""
 128                 job.addTask(self)
 129
 130         def setCommandline(self, cmd, args):
 131                 self.cmd = cmd
 132                 self.args = args
 133
 134         def setTool(self, tool):
 135                 self.cmd = tool
 136                 self.args = [tool]
 137                 self.global_preconditions.append(ToolExistsPrecondition())
 138                 self.postconditions.append(ReturncodePostcondition())
 139
 140         def checkPreconditions(self, immediate = False):
 141                 not_met = [ ]
 142                 if immediate:
 143                         preconditions = self.immediate_preconditions
 144                 else:
 145                         preconditions = self.global_preconditions
 146                 for precondition in preconditions:
 147                         if not precondition.check(self):
 148                                 not_met.append(precondition)
 149                 return not_met
 150
 151         def run(self, callback, task_progress_changed):
 152                 failed_preconditions = self.checkPreconditions(True) + self.checkPreconditions(False)
 153                 if len(failed_preconditions):
 154                         callback(self, failed_preconditions)
 155                         return
 156                 self.prepare()
 157
 158                 self.callback = callback
 159                 self.task_progress_changed = task_progress_changed
 160                 from enigma import eConsoleAppContainer
 161                 self.container = eConsoleAppContainer()
 162                 self.container.appClosed.get().append(self.processFinished)
 163                 self.container.dataAvail.get().append(self.processOutput)
 164
 165                 assert self.cmd is not None
 166                 assert len(self.args) >= 1
 167
 168                 if self.cwd is not None:
 169                         self.container.setCWD(self.cwd)
 170
 171                 print "execute:", self.container.execute(self.cmd, self.args), self.cmd, " ".join(self.args)
 172                 if self.initial_input:
 173                         self.writeInput(self.initial_input)
 174
 175         def prepare(self):
 176                 pass
 177
 178         def cleanup(self, failed):
 179                 pass
 180
 181         def processOutput(self, data):
 182                 self.output_line += data
 183                 while True:
 184                         i = self.output_line.find('\n')
 185                         if i == -1:
 186                                 break
 187                         self.processOutputLine(self.output_line[:i+1])
 188                         self.output_line = self.output_line[i+1:]
 189
 190         def processOutputLine(self, line):
 191                 pass
 192
 193         def processFinished(self, returncode):
 194                 self.returncode = returncode
 195                 self.finish()
 196
 197         def abort(self):
 198                 self.container.kill()
 199                 self.finish(aborted = True)
 200
 201         def finish(self, aborted = False):
 202                 self.afterRun()
 203                 not_met = [ ]
 204                 if aborted:
 205                         not_met.append(AbortedPostcondition())
 206                 else:
 207                         for postcondition in self.postconditions:
 208                                 if not postcondition.check(self):
 209                                         not_met.append(postcondition)
 210                 self.cleanup(not_met)
 211                 self.callback(self, not_met)
 212
 213         def afterRun(self):
 214                 pass
 215
 216         def writeInput(self, input):
 217                 self.container.write(input)
 218
 219         def getProgress(self):
 220                 return self.__progress
 221
 222         def setProgress(self, progress):
 223                 if progress > self.end:
 224                         progress = self.end
 225                 if progress < 0:
 226                         progress = 0
 227                 self.__progress = progress
 228                 self.task_progress_changed()
 229
 230         progress = property(getProgress, setProgress)
 231
 232 # The jobmanager will execute multiple jobs, each after another.
 233 # later, it will also support suspending jobs (and continuing them after reboot etc)
 234 # It also supports a notification when some error occured, and possibly a retry.
 235 class JobManager:
 236         def __init__(self):
 237                 self.active_jobs = [ ]
 238                 self.failed_jobs = [ ]
 239                 self.job_classes = [ ]
 240                 self.active_job = None
 241
 242         def AddJob(self, job):
 243                 self.active_jobs.append(job)
 244                 self.kick()
 245
 246         def kick(self):
 247                 if self.active_job is None:
 248                         if len(self.active_jobs):
 249                                 self.active_job = self.active_jobs.pop(0)
 250                                 self.active_job.start(self.jobDone)
 251
 252         def jobDone(self, job, task, problems):
 253                 print "job", job, "completed with", problems, "in", task
 254                 if problems:
 255                         from Tools import Notifications
 256                         from Screens.MessageBox import MessageBox
 257                         Notifications.AddNotificationWithCallback(self.errorCB, MessageBox, _("Error: %s\nRetry?") % (problems[0].getErrorMessage(task)))
 258                         return
 259                         #self.failed_jobs.append(self.active_job)
 260
 261                 self.active_job = None
 262                 self.kick()
 263
 264         def errorCB(self, answer):
 265                 if answer:
 266                         print "retrying job"
 267                         self.active_job.retry()
 268                 else:
 269                         print "not retrying job."
 270                         self.failed_jobs.append(self.active_job)
 271                         self.active_job = None
 272                         self.kick()
 273
 274 # some examples:
 275 #class PartitionExistsPostcondition:
 276 #       def __init__(self, device):
 277 #               self.device = device
 278 #
 279 #       def check(self, task):
 280 #               import os
 281 #               return os.access(self.device + "part1", os.F_OK)
 282 #
 283 #class CreatePartitionTask(Task):
 284 #       def __init__(self, device):
 285 #               Task.__init__(self, _("Create Partition"))
 286 #               self.device = device
 287 #               self.setTool("/sbin/sfdisk")
 288 #               self.args += ["-f", self.device + "disc"]
 289 #               self.initial_input = "0,\n;\n;\n;\ny\n"
 290 #               self.postconditions.append(PartitionExistsPostcondition(self.device))
 291 #
 292 #class CreateFilesystemTask(Task):
 293 #       def __init__(self, device, partition = 1, largefile = True):
 294 #               Task.__init__(self, _("Create Filesystem"))
 295 #               self.setTool("/sbin/mkfs.ext")
 296 #               if largefile:
 297 #                       self.args += ["-T", "largefile"]
 298 #               self.args.append("-m0")
 299 #               self.args.append(device + "part%d" % partition)
 300 #
 301 #class FilesystemMountTask(Task):
 302 #       def __init__(self, device, partition = 1, filesystem = "ext3"):
 303 #               Task.__init__(self, _("Mounting Filesystem"))
 304 #               self.setTool("/bin/mount")
 305 #               if filesystem is not None:
 306 #                       self.args += ["-t", filesystem]
 307 #               self.args.append(device + "part%d" % partition)
 308
 309 class Condition:
 310         RECOVERABLE = False
 311
 312         def getErrorMessage(self, task):
 313                 return _("An error has occured. (%s)") % (self.__class__.__name__)
 314
 315 class WorkspaceExistsPrecondition(Condition):
 316         def check(self, task):
 317                 return os.access(task.job.workspace, os.W_OK)
 318
 319 class DiskspacePrecondition(Condition):
 320         def __init__(self, diskspace_required):
 321                 self.diskspace_required = diskspace_required
 322                 self.diskspace_available = 0
 323
 324         def check(self, task):
 325                 import os
 326                 try:
 327                         s = os.statvfs(task.job.workspace)
 328                         self.diskspace_available = s.f_bsize * s.f_bavail
 329                         return self.diskspace_available >= self.diskspace_required
 330                 except OSError:
 331                         return False
 332
 333         def getErrorMessage(self, task):
 334                 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)
 335
 336 class ToolExistsPrecondition(Condition):
 337         def check(self, task):
 338                 import os
 339                 if task.cmd[0]=='/':
 340                         realpath = task.cmd
 341                 else:
 342                         realpath = task.cwd + '/' + task.cmd
 343                 self.realpath = realpath
 344                 return os.access(realpath, os.X_OK)
 345
 346         def getErrorMessage(self, task):
 347                 return _("A required tool (%s) was not found.") % (self.realpath)
 348
 349 class AbortedPostcondition(Condition):
 350         pass
 351
 352 class ReturncodePostcondition(Condition):
 353         def check(self, task):
 354                 return task.returncode == 0
 355
 356 #class HDDInitJob(Job):
 357 #       def __init__(self, device):
 358 #               Job.__init__(self, _("Initialize Harddisk"))
 359 #               self.device = device
 360 #               self.fromDescription(self.createDescription())
 361 #
 362 #       def fromDescription(self, description):
 363 #               self.device = description["device"]
 364 #               self.addTask(CreatePartitionTask(self.device))
 365 #               self.addTask(CreateFilesystemTask(self.device))
 366 #               self.addTask(FilesystemMountTask(self.device))
 367 #
 368 #       def createDescription(self):
 369 #               return {"device": self.device}
 370
 371 job_manager = JobManager()