from typing import List from .condition import Condition from .action import Action from time import time from datetime import datetime key_list = { 'deviceList': 'device', 'parameterSet': 'parameter_set' } class Task: def __init__(self, task): self._id = None self._uuid = None self._type = None self._name = None self._parent = None self._cycle = None self._cycle_count = 0 self._cycle_next = None self._cycle_next_uuid = None self._device = None self._event = None self._trigger = None self._parameter_set = None self._condition = None self._action = None # -1: init 0: start(idle), 1: run, 2: close(idle) self._status = 0 self._next = None # action & condition info self._condition_list: List[Condition] = [] self._action_list: List[Action] = [] self._instruction_list = [] self._record_list = [] # time info self._start_time = [] self._idle_time = [] self._end_time = [] self._period = None # trigger info self._button_trigger = False self.load_task(task) def load_task(self, task) -> None: # update task attribute for key, value in task.items(): if key in key_list.keys(): key = key_list[key] if key == 'condition': self.update_condition_list(value) if key == 'action': self.update_action_list(value) # print(key, value) setattr(self, key, value) def update_condition_list(self, condition_dict): self._condition_list.clear() for key, condition in condition_dict.items(): self._condition_list.append(Condition(key, condition)) def update_action_list(self, action_dict): self._action_list.clear() for key, action in action_dict.items(): self._action_list.append(Action(key, action)) @property def id(self) -> str: return self._id @id.setter def id(self, new_id): self._id = new_id @property def uuid(self) -> str: return self._uuid @uuid.setter def uuid(self, new_uuid): self._uuid = new_uuid @property def type(self): return self._type @type.setter def type(self, new_type): self._type = new_type @property def name(self): return self._name @name.setter def name(self, new_name): self._name = new_name @property def parent(self): if self._parent == None: return {"folder": [2]} return self._parent @parent.setter def parent(self, new_parent): self._parent = new_parent @property def cycle(self) -> int: return self._cycle @cycle.setter def cycle(self, new_cycle): self._cycle = new_cycle # TODO: not sure needed? @property def cycle_count(self) -> int: return self._cycle_count @cycle_count.setter def cycle_count(self, new_cycle_count): self._cycle_count = new_cycle_count @property def cycle_next(self) -> str: return self._cycle_next @cycle_next.setter def cycle_next(self, new_cycle_next): self._cycle_next = new_cycle_next @property def cycle_next_uuid(self) -> str: return self._cycle_next_uuid @cycle_next_uuid.setter def cycle_next_uuid(self, new_cycle_next_uuid): self._cycle_next_uuid = new_cycle_next_uuid @property def device(self) -> dict: return self._device @device.setter def device(self, new_device): self._device = new_device @property def event(self) -> list: return self._event @event.setter def event(self, new_event): self._event = new_event @property def trigger(self) -> list: return self._trigger @trigger.setter def trigger(self, new_trigger): self._trigger = new_trigger @property def parameter_set(self) -> list: return self._parameter_set @parameter_set.setter def parameter_set(self, new_parameter_set): self._parameter_set = new_parameter_set @property def condition(self) -> list: return self._condition @condition.setter def condition(self, new_condition): self._condition = new_condition @property def action(self) -> list: return self._action @action.setter def action(self, new_action): self._action = new_action @property def status(self) -> str: return self._status @status.setter def status(self, new_status): self._status = new_status @property def next(self) -> str: return self._next @next.setter def next(self, new_next): self._next = new_next @property def instruction_list(self) -> List: return self._instruction_list @property def start_time(self) -> List: return self._start_time @property def end_time(self) -> List: return self._end_time @property def button_trigger(self) -> List: return self._button_trigger @button_trigger.setter def button_trigger(self, button_trigger): self._button_trigger = button_trigger def new_start_time(self): self._start_time.append(time()) def new_end_time(self): self._end_time.append(time()) def get_parameter_set_by_device(self, device): return [parameter_set for parameter_set in self.parameter_set.values() if device == parameter_set['target']] def get_parameter(self, name_or_value, device): name_list = [] value_list = [] for parameter in self.get_parameter_set_by_device(device): for key, value in parameter.items(): if key != 'target': name_list.append(key) value_list.append(str(value)) return '|'.join(name_list) if name_or_value == 'name' else '|'.join(value_list) def run(self): print(self._name, 'run at', datetime.now()) self._status = 1 self._start_time.append(time()) def start(self): self.status = 1 def stop(self): self.status = 2 def reset(self): self.status = -1 # self.status = 0 def get_match_action_list(self, match_condition_list): return map(lambda condition: [x for x in self._action_list if condition.id in x.get_condition_list()], match_condition_list) def get_match_action(self, condition_id): return [action for action in self._action_list if condition_id in action.condition] def do_action(self, action): for instruction in action.get_instruction_list(): self._instruction_list.append(instruction) def check_condition(self, **kwargs): kwargs['task_start_time'] = self._start_time return [condition for condition in self._condition_list if condition.match_or_not(**kwargs)] def get_task_info(self, action): _type = action.type if _type == 'stop': _type = 'interrupt' return { 'device': action.target, 'parameter': self.get_parameter_set_by_device(action.target), 'file_name': self._name, 'parent': self.parent, 'instruction': _type } def as_json(self): return { 'id': self.id, 'name': self.name, 'parent': self.parent, 'type': self.type, 'uuid': self.uuid, 'cycle': self.cycle, 'cycle_count': self.cycle_count, 'cycle_next': self.cycle_next, 'cycle_next_uuid': self.cycle_next_uuid, 'device': self.device, 'event': self.event, 'trigger': self.trigger, 'parameter_set': self.parameter_set, 'condition': self.condition, 'action': self.action, 'status': self.status, 'next': self.next, }