peterlu14
125 lines
2.9 KiB
Python
125 lines
2.9 KiB
Python
class Task:
|
|
def __init__(self, task_id, task):
|
|
self._id = task_id
|
|
self._task = task
|
|
self._status = self._task['status']
|
|
self._cycle = self._task['cycle']
|
|
# store the reference (next item)
|
|
self.next = []
|
|
|
|
@property
|
|
def task_id(self) -> str:
|
|
return self._id
|
|
|
|
@property
|
|
def cycle(self) -> int:
|
|
return self._task['cycle']
|
|
|
|
@property
|
|
def order(self) -> int:
|
|
return self._task['order']
|
|
|
|
@property
|
|
def max_cycle(self) -> int:
|
|
return self._task['max_cycle']
|
|
|
|
@property
|
|
def devices(self) -> dict:
|
|
return self._task['devices']
|
|
|
|
@property
|
|
def devices_id(self) -> list:
|
|
return [device_id for device_id in self.devices]
|
|
|
|
@property
|
|
def events(self) -> list:
|
|
return self._task['events']
|
|
|
|
@property
|
|
def triggers(self) -> list:
|
|
return self._task['triggers']
|
|
|
|
@property
|
|
def parameter_sets(self) -> list:
|
|
return self._task['parameter_sets']
|
|
|
|
@property
|
|
def instructions(self) -> list:
|
|
return self._task['instructions']
|
|
|
|
@property
|
|
def conditions(self) -> list:
|
|
return self._task['conditions']
|
|
|
|
@property
|
|
def actions(self) -> list:
|
|
return self._task['actions']
|
|
|
|
@property
|
|
def status(self) -> str:
|
|
return self._status
|
|
|
|
def set_cycle(self, cycle):
|
|
self._cycle = cycle
|
|
|
|
def match_condition_actions(self, condition_id) -> list:
|
|
return [action_id for action_id in self.actions if condition_id in self.actions[action_id]['condition']]
|
|
|
|
|
|
def conditions_meet(self, specify_type:str = None):
|
|
match_conditions = []
|
|
|
|
if specify_type != None:
|
|
search_list = []
|
|
else:
|
|
search_list = self.conditions.items()
|
|
|
|
for condition_id, condition in search_list:
|
|
pass_all_sub_condition = True
|
|
for main_type in condition:
|
|
for sub_type in condition[main_type]:
|
|
ret = self.method_mapping(main_type + '_' + sub_type)(condition_id)
|
|
pass_all_sub_condition = pass_all_sub_condition and ret
|
|
if pass_all_sub_condition:
|
|
match_conditions.append((self.task_id, condition_id))
|
|
|
|
return match_conditions
|
|
|
|
def compareWith(self, operator:str, x, y) -> bool:
|
|
cases = {
|
|
"equal": lambda a, b: a == b,
|
|
"bigger": lambda a, b: a > b,
|
|
"smaller": lambda a, b: a < b,
|
|
}
|
|
|
|
print('x','y',x, y,type(x),type(y))
|
|
return cases[operator](x, y)
|
|
|
|
def set_status(self, status):
|
|
self._status = status
|
|
|
|
def start(self):
|
|
self.set_status('running')
|
|
|
|
def done(self):
|
|
self.set_status('done')
|
|
|
|
def end(self):
|
|
self.set_status('end')
|
|
|
|
def method_mapping(self, method_name):
|
|
methods = {
|
|
"time_relative": self.time_relative,
|
|
"time_absolute": self.time_absolute,
|
|
"device_connected": self.device_connected,
|
|
"task_status": self.task_status,
|
|
}
|
|
return methods[method_name]
|
|
|
|
def status_mapping(self, status):
|
|
statuses = {
|
|
"start": self.task_start,
|
|
"done": self.task_done,
|
|
}
|
|
return statuses[status]
|