from typing import AnyStr from ast import literal_eval from biopro.devlib.data import DataDecodeFormat from biopro.devlib.library import * from biopro.devlib.menu import Menu from biopro.devlib.parameter import * from biopro.impl.interface_uart import UARTInterface from biopro.impl.selector import Selector from biopro.server.socket import ServerShutdownException from biopro.server.api import ApiRequests from biopro.text import * from biopro.util.address import address_str from biopro.util.cli_base import CliOptions, cli_options from biopro.util.console import GREEN from biopro.util.lock import Synchronized, synchronized from biopro.util.logger import Logger, logging_verbose, logging_info from biopro.util.stack import print_exception from .device_cc2650 import CC2650DeviceOptions _RUNTIME_COMPILE = False T = TypeVar('T') def build_url(path: str, **options: Any) -> str: if len(options) > 0: return path + '?' + '&'.join(map(lambda it: '%s=%s' % (it[0], str(it[1])), options.items())) else: return path # noinspection PyUnusedLocal class DeviceManagerOptions(CC2650DeviceOptions, CliOptions): DEFAULT_INTERFACE = 'fallback:CC2650,Dummy' DEFAULT_USB_PORT = '/dev/ttyAMA0' def __init__(self): super().__init__() self.repository = DeviceLibraryRepository() self.interface: str = self.DEFAULT_INTERFACE # selector self.select_device: Optional[Tuple[int, ...]] = None self.device_log_level = Logger.LEVEL_INFO @cli_options('-L', '--library', value='PATH') def _library_path(self, opt: str, value: str): self.repository.add_library_path(value) @cli_options('-i', '--interface', value='INT') def _interface(self, opt: str, value: str): """hardware interface. could be: CC2650 : Bluetooth Low Energy CC2650 with multiple board CC2650_S : Bluetooth Low Energy CC2650 with single board NULL : null hardware (debug use). Dummy : Dummy hardware fallback:INT,... : try using interface (default: %DEFAULT_INTERFACE%) """ self.interface = value @cli_options('--selector', value='DEVICE,...') def _select_device(self, opt: str, value: str): """only use certain selector channel, it can reduce the controller setup times.""" self.select_device = tuple(map(int, value.split(','))) @cli_options('--device-log-level', value='LEVEL') def _device_log_level(self, opt: str, value: str): """device log level. could be (case insensitive): verbose info warn error disable, quiet """ if not _RUNTIME_COMPILE: value = value.lower() if value == 'verbose': self.device_log_level = Logger.LEVEL_VERBOSE elif value == 'info': self.device_log_level = Logger.LEVEL_INFO elif value == 'warn': self.device_log_level = Logger.LEVEL_WARN elif value == 'error': self.device_log_level = Logger.LEVEL_ERROR elif value == 'disable' or value == 'quiet': self.device_log_level = Logger.LEVEL_QUIET else: raise ValueError('illegal log level : ' + str(value)) else: print('ignore', opt, 'in product release') class DeviceHardwareInterface: INTERFACE_CC2650 = 'CC2650' INTERFACE_CC2650_S = 'CC2650_S' INTERFACE_NULL = 'NULL' INTERFACE_DUMMY = 'Dummy' INTERFACE_SHUTDOWN = 'Shutdown' __slots__ = '_master', '_interface', '_options' def __init__(self, master: MasterDevice, interface: str, options: DeviceManagerOptions): self._master = master self._interface = interface self._options = options @classmethod def fallback(cls, *interface: str): return 'fallback:' + ','.join(interface) @property def interface(self) -> str: return self._interface def new_master(self, interface: Optional[str] = None) -> MasterDevice: if interface is None: interface = self._interface if interface == self.INTERFACE_SHUTDOWN: raise ServerShutdownException() elif interface == self.INTERFACE_NULL or interface == self.INTERFACE_DUMMY: self._interface = interface return NullMasterDevice() elif interface == self.INTERFACE_CC2650_S: self._interface = interface return self.__new_cc2650_single_master() elif interface == self.INTERFACE_CC2650: self._interface = interface return self.__new_cc2650_master() elif interface.startswith('fallback:'): for fallback in interface[9:].split(','): self._master.log_verbose('fallback device', fallback) if fallback == self.INTERFACE_SHUTDOWN: raise ServerShutdownException() try: return self.new_master(fallback) except BaseException as e: self._master.log_warn(e) print_exception(self._master) continue raise RuntimeError('unsupported master device : ' + interface) def new_central_master(self, interface: Optional[str] = None) -> MasterDevice: if interface is None: interface = self._interface if interface == self.INTERFACE_SHUTDOWN: raise ServerShutdownException() elif interface == self.INTERFACE_NULL or interface == self.INTERFACE_DUMMY: self._interface = interface return NullMasterDevice() elif interface == self.INTERFACE_CC2650_S: self._interface = interface return self.__new_cc2650_single_master() elif interface == self.INTERFACE_CC2650: self._interface = interface return self.__new_cc2650_central_master() elif interface.startswith('fallback:'): for fallback in interface[9:].split(','): self._master.log_verbose('fallback device', fallback) if fallback == self.INTERFACE_SHUTDOWN: raise ServerShutdownException() try: return self.new_central_master(fallback) except BaseException as e: self._master.log_warn(e) print_exception(self._master) continue def __new_cc2650_single_master(self) -> MasterDevice: from .device_cc2650 import CC2650SingleMasterDevice uart = self.__new_uart_interface() uart.reset() uart.flush() return CC2650SingleMasterDevice(self._master, uart, self._options) def __new_cc2650_master(self) -> MasterDevice: from .device_cc2650 import CC2650MultiMasterDevice options = self._options if options.select_device is not None: Selector.set_selector(options.select_device) uart = self.__new_uart_interface() uart.reset() uart.flush() self._master.log_verbose('selector interface', Selector.UART_SELECTOR) selector = Selector.get(Selector.UART_SELECTOR) return CC2650MultiMasterDevice(self._master, uart, selector, options) def __new_cc2650_central_master(self) -> MasterDevice: from .device_cc2650 import CC2650MultiMasterCentralDevice options = self._options if options.select_device is not None: Selector.set_selector(options.select_device) uart = self.__new_uart_interface() uart.reset() uart.flush() self._master.log_verbose('selector interface', Selector.UART_SELECTOR) selector = Selector.get(Selector.UART_SELECTOR) mem_selector = Selector.get(Selector.MEM_SELECTOR) return CC2650MultiMasterCentralDevice(self._master, uart, mem_selector, selector, options) def __new_uart_interface(self) -> UARTInterface: tty = DeviceManagerOptions.DEFAULT_USB_PORT if Path(tty).exists(): self._master.log_verbose('uart interface', tty) return UARTInterface(tty) raise FileNotFoundError(tty) class DeviceInternalCommandHandler: __slots__ = () def device_internal_command(self, device: int, oper: str, value: Any) -> bool: """handle device internal command. :param device: device ID :param oper: operator :param value: operator value :return consumed or not """ return True # why inherit MasterDevice ? class DeviceManager(MasterDevice, Synchronized): """device manager. It manager the device libraries, found device, connected device and the control of the master device. """ __slots__ = ('_handler', '_options', '_library_path', '_library', '_interface', '_master', '_found_history', '_found', '_device', '_master_log_level') def __init__(self, options: DeviceManagerOptions, handler: Optional[DeviceInternalCommandHandler] = None): """ :param options: manager options. :param handler: device internal command handler """ super().__init__(log_name_color=GREEN) Synchronized.__init__(self) if handler is None: handler = DeviceInternalCommandHandler() # control server api self._handler: DeviceInternalCommandHandler = handler self._options: DeviceManagerOptions = options # library self._repository = DeviceLibraryRepository(options.repository.library_path, no_system_default=True) # hardware self._interface = options.interface self._master: Optional[MasterDevice] = None # memory board self._centralMaster: Optional[MasterDevice] = None # connected device self._device: List[CompletedDevice] = [] # demo device self._demo: Set[DeviceResponseInfo] = set() # history found device self._found: Set[DeviceResponseInfo] = set() self._master_log_level = options.device_log_level @synchronized @logging_info def setup(self, interface: Optional[str] = None): """initialize the master device. :param interface: use specific hardware interface """ if interface is None: interface = self._options.interface # dhi = DeviceHardwareInterface(self, interface, self._options) dhi_central = DeviceHardwareInterface(self, interface, self._options) # self._master = m = dhi.new_master() # m.set_log_level(self._master_log_level) self._centralMaster = central_m = dhi_central.new_central_master() # self._interface = dhi.interface self._interface = dhi_central.interface self.log_info('use interface', self._interface) @property def master_device(self) -> MasterDevice: """ :return: master device implementation """ return self._centralMaster @property def interface(self) -> str: return self._interface @property def library_path(self) -> List[Path]: return self._repository.library_path def get_device_status(self, mac_address) -> Optional[DeviceLibrary]: device = self.get_device(mac_address) return device.status @logging_info def reload_library(self): """reload device library""" # clear demo device self._demo.clear() # invalid cache self._repository.invalid_cache() def _on_error(f: Path, _: BaseException): self.log_warn('load library fail', f) print_exception() self._repository.recursive = True self._repository.on_error = _on_error for devlib in self._repository.get_library_all(): self.log_verbose('load library', devlib.name, devlib.version, devlib.file_path) def list_library(self) -> List[DeviceLibrary]: return self._repository.get_library_all() def get_library(self, name: str, version: Union[None, str, DeviceLibraryVersion] = None) -> Optional[DeviceLibrary]: return self._repository.get_library(name, version) def match_library(self, response: DeviceResponseInfo) -> Optional[DeviceLibrary]: """get the library match to the *response*""" return self._repository.match_library(response) def _ensure_initialized(self): if self._centralMaster is None: raise RuntimeError('master device not init') @logging_info def reset(self, device: Optional[List[int]] = None, software_reset=True): self._ensure_initialized() self.disconnect_all() self._reset(device, software_reset) def _reset(self, device: Optional[List[int]] = None, software_reset=True): """reset master device""" self._device.clear() self._demo.clear() self._centralMaster.reset(device, software_reset) @logging_info def shutdown(self, release_resource=True): # shutdown reset memeryboard self.reset() m: Optional[MasterDevice] = self._centralMaster if m is not None: self.disconnect_all() m.shutdown() self._master = None self._centralMaster = None self._interface = None self._device.clear() self._demo.clear() @logging_verbose def scan_callback(self, callback: Callable[[DeviceResponseInfo], None], timeout=5, all_device=False) -> bool: """perform scan. :param callback: notify scanned device. :param timeout: timeout :param all_device: all device perform scanning if not working :return: does manager perform scan, False if another thread is perform scanning. """ self._ensure_initialized() if not self.lock_acquire(): return False error = None try: result = self._scan_callback(callback, timeout=timeout, all_device=all_device) except RuntimeError as e: error = e else: if not result: raise RuntimeError(HARDWARE_SCAN_FAIL) finally: self.lock_release() if error is not None: # IncorrectError, due to a device is connected incorrectly (?), # try to disconnect all devices self.log_warn('suppressed error : ' + str(error)) raise RuntimeError(HARDWARE_SCAN_FAIL) return True def _scan_callback(self, callback: Callable[[DeviceResponseInfo], None], timeout=5, all_device=False) -> bool: self._ensure_initialized() if not self.lock_acquire(): # require lock fail, because hardware is doing something now. # Most case are another scanning is performing. return True def _callback(response: DeviceResponseInfo): library = self.match_library(response) # append library information if library is not None: response = DeviceResponseInfo(response.device_name, response.serial_number, response.mac_address, library.name, str(library.version)) self.log_verbose('found', address_str(response.mac_address), library.name, library.version) else: self.log_verbose('found', address_str(response.mac_address), 'unknown', response.serial_number) # add history self._found.add(response) callback(response) try: result = self._centralMaster.scan_callback(_callback, timeout=timeout, all_device=all_device) except RuntimeError as e: self.log_warn('suppressed error : ' + str(e)) result = False finally: self.lock_release() if not result: self._reset() return result def found(self) -> List[DeviceResponseInfo]: ret = [] for d in self._device: r = d.response_info ret.append(r) def _contain(response: DeviceResponseInfo) -> bool: for _response in ret: if response.match(_response): return True return False for r in self._centralMaster.found(): # force state if not _contain(r): r.device_status = DeviceResponseInfo.STATUS_SCANNED # print('found ', r) ret.append(r) for r in self._demo: # force state if not _contain(r): r.device_status = DeviceResponseInfo.STATUS_SCANNED # print('history', r) ret.append(r) for r in self._found: # force state if not _contain(r): r.device_status = DeviceResponseInfo.STATUS_NOT_FOUND # print('history', r) ret.append(r) return ret def found_device(self, response: DeviceInfo) -> Optional[DeviceResponseInfo]: # find from connected device for device in self._device: if response.match(device): return device.response_info # find from found device for device in self._centralMaster.found(): if response.match(device): return device # find from demo device for device in self._demo: if response.match(device): return device # find from history device for device in self._found: if response.match(device): return device # not found return None @logging_verbose def clear_found_device(self): """clear history found device""" self._found.clear() self._demo.clear() def list_device(self) -> List[CompletedDevice]: """list connected device :return: list of CompletedDevice """ return list(self._device) @logging_verbose def get_device(self, device: Union[int, DeviceInfo]) -> Optional[CompletedDevice]: """get connected device :param device: device ID or response info :return: CompletedDevice, None if not found. """ try: if isinstance(device, int): for slave in self._device: if slave.device_id == device: return slave elif isinstance(device, str): for slave in self._device: if ':'.join('{:02x}'.format(b) for b in slave.mac_address) == device.lower(): return slave elif isinstance(device, DeviceInfo): for slave in self._device: if device.match(slave): return slave except BaseException as e: self.log_error('device not found', str(device)) return None def direct_connect(self, device_mac_str: str, device_serial_number:str) -> Optional[CompletedDevice]: # neulive2.0 product number is 1, 1, 2, 0 # version is 2021/01 serial_number_list = literal_eval(device_serial_number) serial_number = DeviceSerialNumber(serial_number_list[0], serial_number_list[1], serial_number_list[2], serial_number_list[3], serial_number_list[4], serial_number_list[5]) # def mac_str_to_addr(mac_str: str) -> ADDRESS: # list_mac = list(mac_str) # device_mac_int = [] # for i in range(6): # mac_int = int(list_mac[i * 2], 16) << 4 | int(list_mac[i * 2 + 1], 16) # device_mac_int.append(mac_int) # return device_mac_int[0], device_mac_int[1], device_mac_int[2], \ # device_mac_int[3], device_mac_int[4], device_mac_int[5] # device_addr: ADDRESS = mac_str_to_addr(device_mac_str) device_addr = literal_eval(device_mac_str) # response = DeviceResponseInfo("Neulive-v3.0", serial_number, device_addr, addr_type=0) response = DeviceResponseInfo("Elite", serial_number, device_addr, addr_type=0) print(response) # get device library library = None if response.library_name is not None: library = self.get_library(response.library_name, response.library_version) if library is None: library = self.match_library(response) '''library not found''' if library is None: self.log_warn(DEVICE_CONNECT_FAIL, device_addr, LIBRARY_NOT_FOUND) return None # raise RuntimeError(LIBRARY_NOT_FOUND, address) try: device = self._centralMaster.connect(response, direct_connect=True) device = CompletedDevice(self, library, device.device_id, device) except Exception as e: print(e) return None # Clean the CC2650 found devices when it has connected a device. # self._master._cc2650[device.device_id]._found.clear() # add connected device list self._device.append(device) self.log_info(DEVICE_CONNECTED, device_addr) self.log_verbose('device reset') try: device.reset() except DeviceInstructionError as e: '''instruction reset fail. device disconnect''' self.disconnect(device.device_id, True) return None except RunTimeError as e: print('runtime', e) else: # Clean all found devices when connecting a device. for i in range(8): if self._centralMaster._cc2650[i] is not None: for device_response in self._centralMaster._cc2650[i]._found: if str(list(device_response.mac_address)) == str(device.mac_address): self._centralMaster._cc2650[i]._found.clear() self._centralMaster._cc2650[i]._found_with_id.clear() self._centralMaster._cc2650[device.device_id]._found.clear() self._centralMaster._cc2650[device.device_id]._found_with_id.clear() self._found.clear() return device @logging_info def connect(self, response: DeviceInfo, user_name: Optional[str] = None) -> Optional[CompletedDevice]: # TODO connect handle user_name, acquired lock self._ensure_initialized() device = self.get_device(response) if device is not None: # device has connected return device address = address_str(response.mac_address) self.log_info(DEVICE_CONNECTING, address) response = self.found_device(response) ''' device not found''' if response is None: self.log_warn(DEVICE_CONNECT_FAIL, address, DEVICE_NOT_FOUND) return None # raise RuntimeError(DEVICE_NOT_FOUND, address) # get device library library = None if response.library_name is not None: library = self.get_library(response.library_name, response.library_version) if library is None: library = self.match_library(response) '''library not found''' if library is None: self.log_warn(DEVICE_CONNECT_FAIL, address, LIBRARY_NOT_FOUND) return None # raise RuntimeError(LIBRARY_NOT_FOUND, address) # get device instance with self: if response.mac_address == EMPTY_ADDRESS: # demo device device = DebugDevice(self, library, device_id=10 + len(self._device), device_name=response.device_name, mac_address=EMPTY_ADDRESS, serial_number=response.serial_number) else: try: device = self._centralMaster.connect(response) device = CompletedDevice(self, library, device.device_id, device) except Exception as e: print(e) return None # Clean the CC2650 found devices when it has connected a device. # self._master._cc2650[device.device_id]._found.clear() # add connected device list self._device.append(device) self.log_info(DEVICE_CONNECTED, address) self.log_verbose('device reset') try: device.reset() except DeviceInstructionError as e: '''instruction reset fail. device disconnect''' self.disconnect(device.device_id, True) return None else: # Clean all found devices when connecting a device. self._centralMaster._cc2650[device.device_id]._found.clear() self._centralMaster._cc2650[device.device_id]._found_with_id.clear() self._found.clear() return device @logging_info def disconnect(self, device: int, force=False) -> bool: self._ensure_initialized() device = self.get_device(device) if device is None: return False try: return self._disconnect(device, force=force) finally: for i, d in enumerate(self._device): if d.device_id == device.device_id: del self._device[i] break @logging_info def disconnect_all(self, force=False): self._ensure_initialized() self.log_verbose(HARDWARE_DISCONNECTING_ALL) for device in self._device: self._disconnect(device, force=force) self.log_info(HARDWARE_DISCONNECTED_ALL) self._device.clear() def _disconnect(self, device: CompletedDevice, force=False) -> bool: address = address_str(device.mac_address) self.log_verbose(DEVICE_DISCONNECTING, device.device_name, address) if not force: try: device.interrupt() except SendInstructionTimeoutError: self.log_warn(DEVICE_NO_RESPONSE, device.device_name, address) except BaseException as e: self.log_warn(DEVICE_RESPONSE_ERROR, str(e)) try: return self._centralMaster.disconnect(device.device_id, force=force) except SendInstructionTimeoutError: self.log_warn(DEVICE_NO_RESPONSE, device.device_name, address) return False except BaseException as e: self.log_warn(DEVICE_RESPONSE_ERROR, str(e)) return False finally: self.log_info(DEVICE_DISCONNECTED, device.device_name, address) def call_device(self, device: int): self._ensure_initialized() device = self.get_device(device) if device is not None: device.call_instruction('calling') def handle_internal_instruction(self, device: Device, func: str, *para: Any) -> bool: if func == InternalInstruction.PREDEFINED_DATA_FORMAT: self._device_data_format(device, *para) elif func == InternalInstruction.PREDEFINED_DATA_FORMAT_CALI: self._device_data_format_cali(device, *para) elif func == InternalInstruction.PREDEFINED_SYNC: self._device_sync(device, *para) elif func == InternalInstruction.PREDEFINED_LOG: self._device_log(device, *para) elif func == InternalInstruction.PREDEFINED_SET: self._device_set(device, *para) elif func == InternalInstruction.PREDEFINED_DISABLE_CACHE: self._device_disable_cache(device, *para) elif func == InternalInstruction.PREDEFINED_IDLE: self._idle(device, *para) elif isinstance(device, DebugDevice): if func == InternalInstruction.PREDEFINED_NOTIFY: return True return False else: return self._centralMaster.handle_internal_instruction(device, func, *para) return True def _device_data_format(self, device: Device, expr: AnyStr): f = DataDecodeFormat.parse(expr) self._handler.device_internal_command(device.device_id, InternalInstruction.PREDEFINED_DATA_FORMAT, f) def _idle(self, device: Device, expr: AnyStr): self._handler.device_internal_command(device.device_id, InternalInstruction.PREDEFINED_IDLE, None) def _device_data_format_cali(self, device: Device, expr: str, cali: bytes = None): if cali is None: cali = device.calibration_info(expr) if cali is None: data = expr else: data = expr.encode() + b':' + cali f = DataDecodeFormat.parse(data) self._handler.device_internal_command(device.device_id, InternalInstruction.PREDEFINED_DATA_FORMAT, f) def _device_disable_cache(self, device: Device, disable: bool = True): self._handler.device_internal_command(device.device_id, InternalInstruction.PREDEFINED_DISABLE_CACHE, disable) def _device_sync(self, device: Device, start: bool): self._handler.device_internal_command(device.device_id, InternalInstruction.PREDEFINED_SYNC, start) def _device_log(self, device: Device, message: str): self.log_verbose('device', device.device_id, ':', message) def _device_set(self, device: Device, parameter: str, value: int): if not isinstance(device, CompletedDevice): device = self.get_device(device.device_id) if device is not None: device.set_parameter(parameter, value) def get_menu(self) -> Optional[Menu]: self._ensure_initialized() m1 = Menu('create-demo-device', 'create demo device') for lib in sorted(self.list_library(), key=lambda k: (k.name, k.version)): m1.append_item('%s-%s' % (lib.name, lib.version), '%s %s' % (lib.name, lib.version), command=[ build_url('create_demo_device', library=lib.name, version=lib.version) ]) ml = [m1] m2 = self._centralMaster.get_menu() if m2 is not None: ml.append(m2) return Menu('device', 'device', ml) @logging_info def create_demo_device(self, library: str, version: Optional[str] = None) -> DeviceResponseInfo: """create a dummy device with :class:`DebugDevice` :param library: library name :param version: library version """ lib = self.get_library(library, version) if lib is None: raise RuntimeError('library ' + library + ' not found') device_name = 'Demo:' + library device_index = None while True: for d in self._demo: if d.device_name == device_name: if device_index is None: device_index = 1 else: device_index += 1 device_name = 'Demo:%s (%d)' % (library, device_index) # next for loop break else: # break while loop break response = DeviceResponseInfo(device_name, lib.match_rule.baseline_serial(), EMPTY_ADDRESS, library_name=lib.name, library_version=str(lib.version), status=DeviceResponseInfo.STATUS_SCANNED) self._demo.add(response) return response @logging_verbose def handle_command(self, device: Optional[int], command: str, options: Dict[str, str]) -> DeviceCommandAction: if device is None: message = command else: message = "%d:%s" % (device, command) try: if device is None: return self._handle_command(command, options) else: return self._centralMaster.handle_command(device, command, options) except CommandNotFoundError as e: self.log_warn(DEVICE_COMMAND_NOT_FOUND, message) raise e except SendInstructionTimeoutError as e: self.log_warn(DEVICE_NO_RESPONSE, message) raise RuntimeError(DEVICE_NO_RESPONSE, message, str(e)) except BaseException as e: self.log_warn(DEVICE_COMMAND_ERROR, message) print_exception(self) raise RuntimeError(DEVICE_COMMAND_ERROR, message, str(e)) from e def _handle_command(self, command: str, options: Dict[str, str]) -> DeviceCommandAction: if command == 'create_demo_device': try: self.create_demo_device(**options) return DeviceCommandAction.SCAN except TypeError as e: raise RuntimeError('create_demo_device lost parameter : library or version') from e elif command == 'scan': self.scan() return DeviceCommandAction.SCAN elif DeviceCommonInstruction(command) is not None: # command for all device ret = DeviceCommandAction.SUCCESS for device in self._device: try: device.call_instruction(command) except SendInstructionTimeoutError: self.log_warn(DEVICE_NO_RESPONSE, '%d:%s' % (device.device_id, command)) self.disconnect(device.device_id, force=True) ret = DeviceCommandAction.DISCONNECTED return ret elif self._centralMaster is not None: # master specific command return self._centralMaster.handle_command(None, command, options) else: raise CommandNotFoundError(command) def list_device_instruction(self, device_id: int) -> List[str]: device = self.get_device(device_id) if device is None: raise RuntimeError(DEVICE_NOT_FOUND, device) ret = set() ret.update(DeviceCommonInstruction.ALL) ret.update(device.library.instruction_table.export_keys()) return list(ret) def device_instruction(self, device_id: int, instruction: str): device = self.get_device(device_id) if device is None: raise RuntimeError(DEVICE_NOT_FOUND, device) if DeviceCommonInstruction(instruction) is not None: device.call_instruction(instruction) elif instruction in device.library.instruction_table: device.call_instruction(instruction) else: raise RuntimeError(INSTRUCTION_UNKNOWN, instruction) def handle_device_parameter_value(self, device_id: int, parameter: str) -> JSON: device = self.get_device(device_id) if device is None: raise RuntimeError(CHIP_NOT_FOUND) if ',' in parameter: ret = {} for p in parameter.split(','): try: value = device.get_parameter_value(p) except (ActionDomainError, KeyError): pass else: ret[p] = JsonSerialize.to_json(value) return ret else: ret = {} try: ret[parameter] = JsonSerialize.to_json(device.get_parameter(parameter)) except KeyError as e: raise RuntimeError(PARAMETER_NOT_FOUND + parameter) from e except ActionDomainError: return None else: return ret def handle_device_parameter_command(self, device_id: int, parameter: str, value: Optional[str] = None) -> JSON: device = self.get_device(device_id) if device is None: raise RuntimeError(CHIP_NOT_FOUND) if ',' in parameter and value is None: table = device.library.parameter_table ret = {} for p in parameter.split(','): if p in table: try: ret[p] = JsonSerialize.to_json(device.get_parameter(p)) except ActionDomainError: pass return ret elif '|' in parameter and '|' in value: table = device.library.parameter_table ret = {} params = parameter.split('|') values = value.split('|') try: for i, v in enumerate(params): try: if value is not None: device.set_parameter(v, values[i]) ret[v] = JsonSerialize.to_json(device.get_parameter(v)) except: pass except ActionDomainError: pass return ret else: ret = {} try: device.library.parameter_table[parameter] except KeyError as e: raise RuntimeError(PARAMETER_NOT_FOUND + parameter) from e if value is not None: device.set_parameter(parameter, value) try: ret[parameter] = JsonSerialize.to_json(device.get_parameter(parameter)) except ActionDomainError: pass return ret # try: # return JsonSerialize.to_json(device.get_parameter(parameter)) # except ActionDomainError: # return None def hardware_test(self) -> Dict[str, Any]: return { 'hardware': 'manager', 'path': list(map(lambda path: str(path), self.library_path)), 'library': [{ 'name': lib.name, 'file': lib.file_path, 'version': str(lib.version), } for lib in self.list_library()], 'interface': self.interface, 'master': self._master.hardware_test(), 'device': [device.as_json() for device in self.list_device()], 'found': [device.as_json() for device in self.found()] }