075324b01109864754640379fcd60680f03ede44
[enigma2.git] / 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                 job.addTask(self)
101
102         def setCommandline(self, cmd, args):
103                 self.cmd = cmd
104                 self.args = args
105
106         def setTool(self, tool):
107                 self.cmd = tool
108                 self.args = [tool]
109                 self.global_preconditions.append(ToolExistsPrecondition())
110                 self.postconditions.append(ReturncodePostcondition())
111
112         def checkPreconditions(self, immediate = False):
113                 not_met = [ ]
114                 if immediate:
115                         preconditions = self.immediate_preconditions
116                 else:
117                         preconditions = self.global_preconditions
118                 for precondition in preconditions:
119                         if not precondition.check(self):
120                                 not_met.append(precondition)
121                 return not_met
122
123         def run(self, callback, task_progress_changed):
124                 failed_preconditions = self.checkPreconditions(True) + self.checkPreconditions(False)
125                 if len(failed_preconditions):
126                         callback(failed_preconditions)
127                         return
128                 self.prepare()
129
130                 self.callback = callback
131                 self.task_progress_changed = task_progress_changed
132                 from enigma import eConsoleAppContainer
133                 self.container = eConsoleAppContainer()
134                 self.container.appClosed.get().append(self.processFinished)
135                 self.container.dataAvail.get().append(self.processOutput)
136
137                 assert self.cmd is not None
138                 assert len(self.args) >= 1
139
140                 if self.cwd is not None:
141                         self.container.setCWD(self.cwd)
142
143                 print "execute:", self.container.execute(self.cmd, self.args), self.cmd, self.args
144                 if self.initial_input:
145                         self.writeInput(self.initial_input)
146
147         def prepare(self):
148                 pass
149
150         def cleanup(self, failed):
151                 pass
152
153         def processOutput(self, data):
154                 pass
155
156         def processFinished(self, returncode):
157                 self.returncode = returncode
158                 self.finish()
159
160         def abort(self):
161                 self.container.kill()
162                 self.finish(aborted = True)
163
164         def finish(self, aborted = False):
165                 self.afterRun()
166                 not_met = [ ]
167                 if aborted:
168                         not_met.append(AbortedPostcondition())
169                 else:
170                         for postcondition in self.postconditions:
171                                 if not postcondition.check(self):
172                                         not_met.append(postcondition)
173
174                 self.cleanup(not_met)
175                 self.callback(not_met)
176
177         def afterRun(self):
178                 pass
179
180         def writeInput(self, input):
181                 self.container.write(input)
182
183         def getProgress(self):
184                 return self.__progress
185
186         def setProgress(self, progress):
187                 print "progress now", progress
188                 self.__progress = progress
189                 self.task_progress_changed()
190
191         progress = property(getProgress, setProgress)
192
193 class JobManager:
194         def __init__(self):
195                 self.active_jobs = [ ]
196                 self.failed_jobs = [ ]
197                 self.job_classes = [ ]
198                 self.active_job = None
199
200         def AddJob(self, job):
201                 self.active_jobs.append(job)
202                 self.kick()
203
204         def kick(self):
205                 if self.active_job is None:
206                         if len(self.active_jobs):
207                                 self.active_job = self.active_jobs.pop(0)
208                                 self.active_job.start(self.jobDone)
209
210         def jobDone(self, job, problems):
211                 print "job", job, "completed with", problems
212                 if problems:
213                         self.failed_jobs.append(self.active_job)
214
215                 self.active_job = None
216                 self.kick()
217
218 # some examples:
219 #class PartitionExistsPostcondition:
220 #       def __init__(self, device):
221 #               self.device = device
222 #
223 #       def check(self, task):
224 #               import os
225 #               return os.access(self.device + "part1", os.F_OK)
226 #
227 #class CreatePartitionTask(Task):
228 #       def __init__(self, device):
229 #               Task.__init__(self, _("Create Partition"))
230 #               self.device = device
231 #               self.setTool("/sbin/sfdisk")
232 #               self.args += ["-f", self.device + "disc"]
233 #               self.initial_input = "0,\n;\n;\n;\ny\n"
234 #               self.postconditions.append(PartitionExistsPostcondition(self.device))
235 #
236 #class CreateFilesystemTask(Task):
237 #       def __init__(self, device, partition = 1, largefile = True):
238 #               Task.__init__(self, _("Create Filesystem"))
239 #               self.setTool("/sbin/mkfs.ext")
240 #               if largefile:
241 #                       self.args += ["-T", "largefile"]
242 #               self.args.append("-m0")
243 #               self.args.append(device + "part%d" % partition)
244 #
245 #class FilesystemMountTask(Task):
246 #       def __init__(self, device, partition = 1, filesystem = "ext3"):
247 #               Task.__init__(self, _("Mounting Filesystem"))
248 #               self.setTool("/bin/mount")
249 #               if filesystem is not None:
250 #                       self.args += ["-t", filesystem]
251 #               self.args.append(device + "part%d" % partition)
252
253 class WorkspaceExistsPrecondition:
254         def check(self, task):
255                 return os.access(task.job.workspace, os.W_OK)
256
257 class DiskspacePrecondition:
258         def __init__(self, diskspace_required):
259                 self.diskspace_required = diskspace_required
260
261         def check(self, task):
262                 import os
263                 try:
264                         s = os.statvfs(task.job.workspace)
265                         return s.f_bsize * s.f_bavail >= self.diskspace_required
266                 except OSError:
267                         return False
268
269 class ToolExistsPrecondition:
270         def check(self, task):
271                 import os
272                 if task.cmd[0]=='/':
273                         realpath = task.cmd
274                 else:
275                         realpath = self.cwd + '/' + self.cmd
276                 return os.access(realpath, os.X_OK)
277
278 class AbortedPostcondition:
279         pass
280
281 class ReturncodePostcondition:
282         def check(self, task):
283                 return task.returncode == 0
284
285 #class HDDInitJob(Job):
286 #       def __init__(self, device):
287 #               Job.__init__(self, _("Initialize Harddisk"))
288 #               self.device = device
289 #               self.fromDescription(self.createDescription())
290 #
291 #       def fromDescription(self, description):
292 #               self.device = description["device"]
293 #               self.addTask(CreatePartitionTask(self.device))
294 #               self.addTask(CreateFilesystemTask(self.device))
295 #               self.addTask(FilesystemMountTask(self.device))
296 #
297 #       def createDescription(self):
298 #               return {"device": self.device}
299
300 job_manager = JobManager()