Mercurial > lcfOS
view python/ppci/tasks.py @ 336:d1ecc493384e
Added spiffy armtoken class for bit fiddeling. Added cool test that checks for build repeatability
author | Windel Bouwman |
---|---|
date | Wed, 19 Feb 2014 22:32:15 +0100 |
parents | 6f4753202b9a |
children | 86b02c98a717 |
line wrap: on
line source
import logging class TaskError(Exception): pass class Task: """ Task that can run, and depend on other tasks """ def __init__(self, name): self.name = name self.subtasks = [] self.completed = False self.dependencies = [] self.duration = 1 def run(self): raise NotImplementedError("Implement this abstract method!") def fire(self): assert all(t.completed for t in self.dependencies) self.run() self.completed = True def addSubTask(self, tsk): self.subtasks.append(tsk) return tsk def addDependency(self, task): self.dependencies.append(task) return task def __repr__(self): return 'Task "{}"'.format(self.name) class TaskRunner: """ Basic task runner that can run some tasks in sequence """ def __init__(self): self.logger = logging.getLogger('taskrunner') self.task_list = [] def add_task(self, task): self.task_list.append(task) @property def total_duration(self): return sum(t.duration for t in self.task_list) def run_tasks(self): passed_time = 0.0 total_time = self.total_duration try: for t in self.task_list: self.report_progress(passed_time / total_time, t.name) t.fire() passed_time += t.duration except TaskError as e: print('Error: {}'.format(e)) return 1 self.report_progress(1, 'OK') return 0 def display(self): """ Display task how they would be run """ for task in self.task_list: print(task) def report_progress(self, percentage, text): self.logger.info('[{:3.1%}] {}'.format(percentage, text))