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.workspace = "/tmp"
  11                 self.current_task = 0
  12                 self.callback = None
  13                 self.name = name
  14                 self.finished = False
  15                 self.end = 100
  16                 self.__progress = 0
  17                 self.weightScale = 1
  18
  19                 self.state_changed = CList()
  20
  21                 self.status = self.NOT_STARTED
  22
  23         # description is a dict
  24         def fromDescription(self, description):
  25                 pass
  26
  27         def createDescription(self):
  28                 return None
  29
  30         def getProgress(self):
  31                 if self.current_task == len(self.tasks):
  32                         return self.end
  33                 t = self.tasks[self.current_task]
  34                 jobprogress = t.weighting * t.progress / float(t.end) + sum([task.weighting for task in self.tasks[:self.current_task]])
  35                 return int(jobprogress*self.weightScale)
  36
  37         progress = property(getProgress)
  38
  39         def task_progress_changed_CB(self):
  40                 self.state_changed()
  41
  42         def addTask(self, task):
  43                 task.job = self
  44                 self.tasks.append(task)
  45
  46         def start(self, callback):
  47                 assert self.callback is None
  48                 self.callback = callback
  49                 self.status = self.IN_PROGRESS
  50                 self.state_changed()
  51                 self.runNext()
  52                 sumTaskWeightings = sum([t.weighting for t in self.tasks])
  53                 self.weightScale = (self.end+1) / float(sumTaskWeightings)
  54
  55         def runNext(self):
  56                 if self.current_task == len(self.tasks):
  57                         self.callback(self, [])
  58                         self.status = self.FINISHED
  59                         self.state_changed()
  60                 else:
  61                         self.tasks[self.current_task].run(self.taskCallback,self.task_progress_changed_CB)
  62                         self.state_changed()
  63
  64         def taskCallback(self, res):
  65                 if len(res):
  66                         print ">>> Error:", res
  67                         self.status = self.FAILED
  68                         self.state_changed()
  69                         self.callback(self, res)
  70                 else:
  71                         self.state_changed();
  72                         self.current_task += 1
  73                         self.runNext()
  74
  75         def abort(self):
  76                 if self.current_task < len(self.tasks):
  77                         self.tasks[self.current_task].abort()
  78
  79         def cancel(self):
  80                 # some Jobs might have a better idea of how to cancel a job
  81                 self.abort()
  82
  83 class Task(object)      :
  84         def __init__(self, job, name):
  85                 self.name = name
  86                 self.immediate_preconditions = [ ]
  87                 self.global_preconditions = [ ]
  88                 self.postconditions = [ ]
  89                 self.returncode = None
  90                 self.initial_input = None
  91                 self.job = None
  92
  93                 self.end = 100
  94                 self.weighting = 100
  95                 self.__progress = 0
  96                 self.cmd = None
  97                 self.cwd = "/tmp"
  98                 self.args = [ ]
  99                 self.task_progress_changed = None
 100                 self.output_line = ""
 101                 job.addTask(self)
 102
 103         def setCommandline(self, cmd, args):
 104                 self.cmd = cmd
 105                 self.args = args
 106
 107         def setTool(self, tool):
 108                 self.cmd = tool
 109                 self.args = [tool]
 110                 self.global_preconditions.append(ToolExistsPrecondition())
 111                 self.postconditions.append(ReturncodePostcondition())
 112
 113         def checkPreconditions(self, immediate = False):
 114                 not_met = [ ]
 115                 if immediate:
 116                         preconditions = self.immediate_preconditions
 117                 else:
 118                         preconditions = self.global_preconditions
 119                 for precondition in preconditions:
 120                         if not precondition.check(self):
 121                                 not_met.append(precondition)
 122                 return not_met
 123
 124         def run(self, callback, task_progress_changed):
 125                 failed_preconditions = self.checkPreconditions(True) + self.checkPreconditions(False)
 126                 if len(failed_preconditions):
 127                         callback(failed_preconditions)
 128                         return
 129                 self.prepare()
 130
 131                 self.callback = callback
 132                 self.task_progress_changed = task_progress_changed
 133                 from enigma import eConsoleAppContainer
 134                 self.container = eConsoleAppContainer()
 135                 self.container.appClosed.get().append(self.processFinished)
 136                 self.container.dataAvail.get().append(self.processOutput)
 137
 138                 assert self.cmd is not None
 139                 assert len(self.args) >= 1
 140
 141                 if self.cwd is not None:
 142                         self.container.setCWD(self.cwd)
 143
 144                 print "execute:", self.container.execute(self.cmd, self.args), self.cmd, self.args
 145                 if self.initial_input:
 146                         self.writeInput(self.initial_input)
 147
 148         def prepare(self):
 149                 pass
 150
 151         def cleanup(self, failed):
 152                 pass
 153
 154         def processOutput(self, data):
 155                 self.output_line += data
 156                 while True:
 157                         i = self.output_line.find('\n')
 158                         if i == -1:
 159                                 break
 160                         self.processOutputLine(self.output_line[:i+1])
 161                         self.output_line = self.output_line[i+1:]
 162
 163         def processOutputLine(self, line):
 164                 pass
 165
 166         def processFinished(self, returncode):
 167                 self.returncode = returncode
 168                 self.finish()
 169
 170         def abort(self):
 171                 self.container.kill()
 172                 self.finish(aborted = True)
 173
 174         def finish(self, aborted = False):
 175                 self.afterRun()
 176                 not_met = [ ]
 177                 if aborted:
 178                         not_met.append(AbortedPostcondition())
 179                 else:
 180                         for postcondition in self.postconditions:
 181                                 if not postcondition.check(self):
 182                                         not_met.append(postcondition)
 183
 184                 self.cleanup(not_met)
 185                 self.callback(not_met)
 186
 187         def afterRun(self):
 188                 pass
 189
 190         def writeInput(self, input):
 191                 self.container.write(input)
 192
 193         def getProgress(self):
 194                 return self.__progress
 195
 196         def setProgress(self, progress):
 197                 if progress > self.end:
 198                         progress = self.end
 199                 if progress < 0:
 200                         progress = 0
 201                 print "progress now", progress
 202                 self.__progress = progress
 203                 self.task_progress_changed()
 204
 205         progress = property(getProgress, setProgress)
 206
 207 class JobManager:
 208         def __init__(self):
 209                 self.active_jobs = [ ]
 210                 self.failed_jobs = [ ]
 211                 self.job_classes = [ ]
 212                 self.active_job = None
 213
 214         def AddJob(self, job):
 215                 self.active_jobs.append(job)
 216                 self.kick()
 217
 218         def kick(self):
 219                 if self.active_job is None:
 220                         if len(self.active_jobs):
 221                                 self.active_job = self.active_jobs.pop(0)
 222                                 self.active_job.start(self.jobDone)
 223
 224         def jobDone(self, job, problems):
 225                 print "job", job, "completed with", problems
 226                 if problems:
 227                         self.failed_jobs.append(self.active_job)
 228
 229                 self.active_job = None
 230                 self.kick()
 231
 232 # some examples:
 233 #class PartitionExistsPostcondition:
 234 #       def __init__(self, device):
 235 #               self.device = device
 236 #
 237 #       def check(self, task):
 238 #               import os
 239 #               return os.access(self.device + "part1", os.F_OK)
 240 #
 241 #class CreatePartitionTask(Task):
 242 #       def __init__(self, device):
 243 #               Task.__init__(self, _("Create Partition"))
 244 #               self.device = device
 245 #               self.setTool("/sbin/sfdisk")
 246 #               self.args += ["-f", self.device + "disc"]
 247 #               self.initial_input = "0,\n;\n;\n;\ny\n"
 248 #               self.postconditions.append(PartitionExistsPostcondition(self.device))
 249 #
 250 #class CreateFilesystemTask(Task):
 251 #       def __init__(self, device, partition = 1, largefile = True):
 252 #               Task.__init__(self, _("Create Filesystem"))
 253 #               self.setTool("/sbin/mkfs.ext")
 254 #               if largefile:
 255 #                       self.args += ["-T", "largefile"]
 256 #               self.args.append("-m0")
 257 #               self.args.append(device + "part%d" % partition)
 258 #
 259 #class FilesystemMountTask(Task):
 260 #       def __init__(self, device, partition = 1, filesystem = "ext3"):
 261 #               Task.__init__(self, _("Mounting Filesystem"))
 262 #               self.setTool("/bin/mount")
 263 #               if filesystem is not None:
 264 #                       self.args += ["-t", filesystem]
 265 #               self.args.append(device + "part%d" % partition)
 266
 267 class WorkspaceExistsPrecondition:
 268         def check(self, task):
 269                 return os.access(task.job.workspace, os.W_OK)
 270
 271 class DiskspacePrecondition:
 272         def __init__(self, diskspace_required):
 273                 self.diskspace_required = diskspace_required
 274
 275         def check(self, task):
 276                 import os
 277                 try:
 278                         s = os.statvfs(task.job.workspace)
 279                         return s.f_bsize * s.f_bavail >= self.diskspace_required
 280                 except OSError:
 281                         return False
 282
 283 class ToolExistsPrecondition:
 284         def check(self, task):
 285                 import os
 286                 if task.cmd[0]=='/':
 287                         realpath = task.cmd
 288                 else:
 289                         realpath = self.cwd + '/' + self.cmd
 290                 return os.access(realpath, os.X_OK)
 291
 292 class AbortedPostcondition:
 293         pass
 294
 295 class ReturncodePostcondition:
 296         def check(self, task):
 297                 return task.returncode == 0
 298
 299 #class HDDInitJob(Job):
 300 #       def __init__(self, device):
 301 #               Job.__init__(self, _("Initialize Harddisk"))
 302 #               self.device = device
 303 #               self.fromDescription(self.createDescription())
 304 #
 305 #       def fromDescription(self, description):
 306 #               self.device = description["device"]
 307 #               self.addTask(CreatePartitionTask(self.device))
 308 #               self.addTask(CreateFilesystemTask(self.device))
 309 #               self.addTask(FilesystemMountTask(self.device))
 310 #
 311 #       def createDescription(self):
 312 #               return {"device": self.device}
 313
 314 job_manager = JobManager()