import sys import json import threading import logging from time import time, sleep from datetime import datetime from collections import deque from copy import copy from uuid import uuid4 from .task import Task from .task_manager import TaskManager from .instruction import Instruction from biopro.device.manager import DeviceManager from biopro.text import * from biopro.db.base import Session from biopro.db.collection import Collection key_list = { 'deviceList': 'device', } class Project(threading.Thread): def __init__(self, project, device_manager: DeviceManager, mqttThread = None, log_verbose = None, name="project"): super(Project, self).__init__(name = name) self._project = project self._device_manager = device_manager self._mqtt_thread = mqttThread self._time_interval = 0.1 self._start_time = None self._end_time = None self._id = None self._uuid = str(uuid4()) self._name = None self._desc = None self._device = None self._complete_device = {} self._status = 0 self._instruction_set = Instruction() self._stop_flag = False self._task_manager = None self._cycle = [] self._count = 1 #流水號 self.log_verbose = log_verbose self._logger = logging.getLogger('project') self._logger.setLevel('DEBUG') self._formatter = logging.Formatter('[%(asctime)s.%(msecs)03d] %(message)s', datefmt='%Y-%m-%d %H:%M:%S') self._formatter_with_nothing = logging.Formatter('%(message)s') self.setup_project(project) self.setup_device(self._device) # create log file handler fh = logging.FileHandler(f'/home/pi/logger/project/{self.uuid}.log', mode="w") fh.setFormatter(self._formatter) self._logger.addHandler(fh) default_name = 'admin' default_parent = {"folder": [1]} collection = Collection.find_collection(default_name, default_parent) parent = {"folder": [collection.id]} # create project folder collection = Collection.create_collection(self.name, parent) self.setup_collection(collection) def setup_project(self, project): for (key, value) in project.items(): if key in key_list.keys(): key = key_list[key] if key == 'task': self._task_manager = TaskManager(project['task'], project['cycle']) elif key == 'uuid': pass else: setattr(self, key, value) def setup_device(self, device_list): for device in device_list: mac_address = device_list[device]['pair'] complete_device = self._device_manager.get_device(mac_address) complete_device.occupied_by_project = self._uuid self._complete_device[device] = complete_device def setup_collection(self, collection): self._task_manager.create_collection(collection) @property def id(self) -> int: 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 name(self) -> str: return self._name @name.setter def name(self, new_name): self._name = new_name @property def cycle(self) -> list: return self._cycle @cycle.setter def cycle(self, new_cycle): self._cycle = new_cycle @property def desc(self) -> str: return self._desc @desc.setter def desc(self, new_desc): self._desc = new_desc @property def status(self) -> str: return self._status @status.setter def status(self, new_status): self._status = new_status @property def device(self) -> list: return self._device @device.setter def device(self, new_device): self._device = new_device @property def task_list(self): return self._task_manager.export_task_list @property def mqtt_thread(self): return self._mqtt_thread @property def running_task(self): return self._task_manager.running_task def get_device_parameter_set(self, device_address): parameter_set = self._task_manager.running_task.parameter_set return next((x for x in parameter_set if parameter_set[x]['target'] == device_address), None) def get_cycle(self, task): for index, _cycle in enumerate(self.cycle): if task.uuid in _cycle['range']: return _cycle def run(self): # project status change running (1) self._status = 1 # save start time self._start_time = time() # saving & broadcast message self._logger.info('Project ' + self.name + ' start') self.log_verbose('Project ' + self.name + ' start') self.mqtt_thread.broadcast_command('project:' + self._name + ' start at ' + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]) while not self._stop_flag : # check running task first delay_time = 0 check_list = copy(self._task_manager.check_list) for task in check_list: if task != None: now = time() match_condition_list = task.check_condition( project_start_time = self._start_time, delay_time = delay_time, running_task= self._task_manager.running_task, previous_task = self._task_manager.prev_task, self_task = task ) for condition in match_condition_list: match_action_list = task.get_match_action(condition.id) print('match_action_list', match_action_list, match_action_list) for action in match_action_list: if action.type == 'cycle' and condition.type == 'previous_task_done': self.mqtt_thread.broadcast_command('project:task ' + task.name + ' start at ' + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]) self._task_manager.set_running_task(task) cycle_info = self.get_cycle(task) if cycle_info['useID'] == False: self._logger.info(f'---------- Cycle {task.name} Round {cycle_info["count"]} -----------') if action.type == 'cycle' and condition.type == 'until_button_trigger': self._task_manager.running_task.stop() self._logger.info(f'---------- Cycle {cycle_info["id"]} Round {cycle_info["count"]} -----------') if self._task_manager.running_task != None: if self._task_manager.running_task.uuid == task.uuid and action.type == 'cycle' and condition.type == 'after_task_run': self._task_manager.running_task.stop() if action.type == 'start': if task.status != 1: self._task_manager.set_running_task(task) self._logger.info(f'Task {str(self._task_manager.running_task.name)} start') self.mqtt_thread.broadcast_command('project:task ' + task.name + ' start at ' + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]) # elif action.type == 'stop': # self.mqtt_thread.broadcast_command('project:task ' + str(self._task_manager.running_task.name) + ' stop at ' + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]) # self._count += 1 # if self.check_running_task_not_run() == True: # self._task_manager.running_task.stop() elif action.type == 'idle': self._task_manager.running_task.stop() # elif action.type == 'stop': # self._task_manager.running_task.stop() instruction_set = None if action.target is not None: device = self._complete_device[action.target] if ((action.type == 'start' and device.status == 0) or (action.type == 'stop' and device.status == 1)): self._logger.info(f'Device {device.mac_address_in_str} {action.type}') self.log_verbose('Project' + self.name + 'Task ' + task.name + ' match_condition ' + condition.type + ' trigger_action ' + action.type + ' ' + str(action.target)) task_info = task.get_task_info(action) instruction_set = getattr(self._instruction_set, action.type, None) # 檔名修正 # device task_info['file_name'] += '#' + device.mac_address_in_str[12:].upper() # 流水號 task_info['file_name'] += '#' + str(self._count) # 添加cycle資訊 if len(self.cycle) > 0: index = self._task_manager.get_index_by_task(task) # get list of cycle determine by index of task filtered = list(filter(lambda info: self._task_manager.get_index_by_uuid(info['range'][0]) < index and self._task_manager.get_index_by_uuid(info['range'][1]) > index, self.cycle)) if len(filtered) == 0: pass elif len(filtered) == 1: if filtered[0]['useID'] == True: task_info['file_name'] += '#' + str(filtered[0]['id']) else: task_info['file_name'] += '#' + str(filtered[0]['name']) + '-' + str(filtered[0]['count']) else: # if multi cycle, then sort by range from task index to cycle start index, then loop concat filename filtered.sort(key=lambda info: abs(self._task_manager.get_index_by_uuid(info['range'][0]) - index)) for cycle_info in filtered: if cycle_info['useID'] == True: task_info['file_name'] += '#' + str(cycle_info['id']) else: task_info['file_name'] += '#' + str(cycle_info['name']) + '-' + str(cycle_info['count']) self.log_verbose('file name ' + task_info['file_name']) if instruction_set != None and self.get_device_parameter_set(action.target) != None: for instruction in instruction_set: print('instruction 1', device, instruction, datetime.now()) args = list(map(lambda arg: task_info[arg], instruction['arguments'])) target=getattr(device, instruction['method'])(*args) print('instruction 2', device, instruction, datetime.now()) if action.type == 'start': self._count += 1 delay_time += (time() - now) # check task not running then stop if self.check_running_task_not_run() == True: # self._logger.info(f'taskType {self._task_manager.running_task.name} {self._task_manager.running_task.type}') if self._task_manager.running_task.type == '': self._logger.info(f'Task {self._task_manager.running_task.name} stop') self.mqtt_thread.broadcast_command('project:task ' + str(self._task_manager.running_task.name) + ' stop at ' + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]) self._task_manager.running_task.stop() # check project done then close project if self.check_project_done() == True: print('project stop at', datetime.now()) self.stop() if self._time_interval - delay_time > 0: sleep(self._time_interval - delay_time) def pause(self): # TODO pass def stop(self): for device in self._complete_device: self._complete_device[device].occupied_by_project = None self._task_manager.running_task.stop() self.mqtt_thread.broadcast_command('project:task ' + str(self._task_manager.running_task.name) + ' stop at ' + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]) self._status = 2 self._end_time = time() self._stop_flag = True self._logger.info('Project ' + self.name + ' stop') self.mqtt_thread.broadcast_command('project:project ' + str(self._name) + ' stop at ' + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]) def close(self): for device in self._complete_device: self._complete_device[device].occupied_by_project = None self._status = 2 self._end_time = time() self._stop_flag = True self.mqtt_thread.broadcast_command('project:project ' + str(self._name) + ' stop at ' + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]) def check_project_done(self): for task in self._task_manager.task_list: # if task never start or still task is running then reject if task.status == 1 or task.status == -1: return False return True def set_content(self, content): self.running_task.button_trigger = True def set_cycle(self, index, content): # change cycle name for key in content: self.cycle[index][key] = content[key] # change task name # self._task_manager.get_task_by_uuid(self.cycle[index]['range'][0]).name = content['name'] # self._task_manager.get_task_by_uuid(self.cycle[index]['range'][1]).name = content['name'] def check_running_task_not_run(self): # if no running task if self._task_manager.running_task == None: return False # if running is cycle then no need to stop by device if self._task_manager.running_task.type == 'cycle': return False for key in self._task_manager.running_task.action: # print('action key', key, self._task_manager.running_task, self._task_manager.running_task.action[key]) if len(self._task_manager.running_task.parameter_set) == 0 and self._task_manager.running_task.action[key]['type'] == 'stop': return False if len(self._task_manager.running_task.parameter_set) == 0 and self._task_manager.running_task.action[key]['type'] == 'idle': return False for device in self._task_manager.running_task.device: if self._complete_device[device].status == 1: return False return True def as_json(self): running_task = None if self._task_manager.running_task is not None: running_task = self._task_manager.running_task.as_json() data = { 'id': self._id, 'name': self._name, 'uuid': self._uuid, 'desc': self._desc, 'status': self._status, 'device': self._device, 'task': self.task_list, 'running_task': running_task, 'cycle': self._cycle, 'count': self._count, } return data def info_pass_data_server(self): if self._task_manager.running_task is not None: running_task = self._task_manager.running_task.as_json() data = { 'project': self._uuid, 'task': running_task, 'cycle': self._cycle, 'serial_number': self._count, } return data