"""Device interface. """ import abc import re from enum import IntEnum from functools import total_ordering from time import time from typing import Any, Optional, Tuple, Union, List, Dict, Callable, NoReturn from biopro.text import * from biopro.util.address import ADDRESS, EMPTY_ADDRESS, address_str from biopro.util.console import WHITE from biopro.util.logger import LoggerFlag from . import * from .menu import Menu class DeviceCommandAction(IntEnum): NOTHING = 0 SUCCESS = 1 CONNECTED = 2 DISCONNECTED = 3 SCAN = 4 MENU = 5 class DeviceBusyError(RuntimeError): """""" __slots__ = () class CommandNotFoundError(RuntimeError): """hardware command not found.""" __slots__ = ('_command',) def __init__(self, command: str): super().__init__(COMMAND_UNKNOWN + ' : ' + command) self._command = command @property def command(self) -> Optional[str]: return self._command class MasterDeviceError(RuntimeError): __slots__ = ('_master',) def __init__(self, master: 'MasterDevice'): super().__init__('master device error') self._master = master @property def device(self) -> 'MasterDevice': return self._master @total_ordering class DeviceLibraryVersion: r""" **version expression pattern** :: DeviceLibraryVersion = version_expression | version_instance version_instance = { "major" :int "minor" :int "patch"? :int "mark"? :str } **version_expression** :: (?P\d+)\.(?P\d+)(\.(?P\d+))?(-(?P.+))? """ __slots__ = ('_major', '_minor', '_patch', '_mark') def __init__(self, major: int, minor: int, patch: int = 0, mark: Optional[str] = None): """ :param major: major version number :param minor: minor version number :param patch: patch number :param mark: version mark """ self._major = major self._minor = minor self._patch = patch self._mark = mark @property def major(self) -> int: return self._major @property def minor(self) -> int: return self._minor @property def patch(self) -> int: return self._patch @property def mark(self) -> Optional[str]: return self._mark @classmethod def value_of(cls, json: Union[str, JSON]) -> 'DeviceLibraryVersion': if isinstance(json, str): return cls._value_of_str(json) elif isinstance(json, dict): return cls._value_of_json(json) else: raise RuntimeError('unknown version expression : ' + _truncate(json)) @staticmethod def _value_of_str(expr: str) -> 'DeviceLibraryVersion': m = re.match(r'(?P\d+)\.(?P\d+)(\.(?P\d+))?(-(?P.+))?', expr) if m is None: raise RuntimeError('illegal device library version expression : ' + expr) try: major = int(m.group('major')) minor = int(m.group('minor')) patch = m.group('patch') if patch is not None: patch = int(patch) else: patch = 0 mark = m.group('mark') except ValueError as e: raise RuntimeError('illegal device library version expression : ' + expr) from e else: return DeviceLibraryVersion(major, minor, patch, mark) @staticmethod def _value_of_json(json: JSON_OBJECT) -> 'DeviceLibraryVersion': major = _get(json, 'major', int, 'version') minor = _get(json, 'minor', int, 'version') patch = _get(json, 'patch', int, 'version', nullable=True, default=0) mark = _get(json, 'mark', str, 'version', nullable=True) return DeviceLibraryVersion(major, minor, patch, mark) def __hash__(self): return 7 * self._major + 13 * self._minor + 17 * self._patch def __eq__(self, other): if not isinstance(other, DeviceLibraryVersion): return False return ( self._major == other._major and self._minor == other._minor and self._patch == other._patch ) def __lt__(self, other: 'DeviceLibraryVersion'): if self._major != other._major: return self._major < other._major if self._minor != other._minor: return self._minor < other._minor if self._patch != other._patch: return self._patch < other._patch return False def __str__(self): return '%d.%d.%d%s' % (self._major, self._minor, self._patch, ('-%s' % self._mark) if self._mark is not None else '') def __repr__(self): return str(self) class DeviceSerialNumber(JsonSerialize): """device serial number which contain six number. **json format** :: DeviceSerialNumber : [int, int, int, int, int, int] DeviceSerialNumber[0] # major product number DeviceSerialNumber[1] # minor product number DeviceSerialNumber[2] # major version number DeviceSerialNumber[3] # minor version number DeviceSerialNumber[4] # serial year number DeviceSerialNumber[5] # serial month number """ UNKNOWN: 'DeviceSerialNumber' = None __slots__ = ('_p1', '_p2', '_v1', '_v2', '_s1', '_s2') def __init__(self, major_product_number: int, minor_product_number: int, major_version_number: int, minor_version_number: int, serial_year_number: int, serial_month_number: int): self._p1 = major_product_number self._p2 = minor_product_number self._v1 = major_version_number self._v2 = minor_version_number self._s1 = serial_year_number self._s2 = serial_month_number @property def major_product_number(self) -> int: """ :return: major product number """ return self._p1 @property def minor_product_number(self) -> int: """ :return: minor product number """ return self._p2 @property def major_version_number(self) -> int: """ :return: major version number """ return self._v1 @property def minor_version_number(self) -> int: """ :return: minor version number """ return self._v2 @property def serial_year_number(self) -> int: """ :return: serial year number. number start from 2000 """ return self._s1 @property def serial_month_number(self) -> int: """ :return: serial month number. number from 1. """ return self._s2 def as_json(self) -> List[int]: return [self._p1, self._p2, self._v1, self._v2, self._s1, self._s2] def __str__(self): return '%d.%d/%d.%d/2%03dy%02d' % (self._p1, self._p2, self._v1, self._v2, self._s1, self._s2) def __repr__(self): return str(self) def __hash__(self): ret = 7 ret = ret * 31 + self.major_product_number ret = ret * 31 + self.minor_product_number ret = ret * 31 + self.major_version_number ret = ret * 31 + self.minor_version_number ret = ret * 31 + self.serial_year_number ret = ret * 31 + self.serial_month_number return ret def __eq__(self, other): if self is other: return True if self is self.UNKNOWN or other is self.UNKNOWN: return True if not isinstance(other, DeviceSerialNumber): raise False return ( self.major_product_number == other.major_product_number and self.minor_product_number == other.minor_product_number and self.major_version_number == other.major_version_number and self.minor_version_number == other.minor_version_number and self.serial_year_number == other.serial_year_number and self.serial_month_number == other.serial_month_number ) DeviceSerialNumber.UNKNOWN = DeviceSerialNumber(0, 0, 0, 0, 0, 0) class DeviceInfo(JsonSerialize, metaclass=abc.ABCMeta): __slots__ = () @property @abc.abstractmethod def device_id(self) -> Optional[int]: """device id. which given from master device""" pass @property @abc.abstractmethod def device_name(self) -> str: """device name.""" pass @property @abc.abstractmethod def serial_number(self) -> DeviceSerialNumber: """device serial number""" pass @property @abc.abstractmethod def mac_address(self) -> ADDRESS: """device max address""" pass @property @abc.abstractmethod def library_name(self) -> str: """device library name. *NOTE* this property has overwrote by the CompletedDevice. the subclass do not need to overwrite this. """ pass @property @abc.abstractmethod def library_version(self) -> str: """device library version. *NOTE* this property has overwrote by the CompletedDevice. the subclass do not need to overwrite this. """ pass @property @abc.abstractmethod def battery(self) -> Optional[int]: """device battery voltage percentage :return: percentage, value from 0 to 100, None if unknown. """ pass @property @abc.abstractmethod def parent(self) -> Optional[dict]: """device save file parent :return: parent id {folder: id, ...} """ pass @property @abc.abstractmethod def recording_file_name(self) -> Optional[dict]: """device save file recording_file_name :return: recording_file_name """ pass def as_json(self) -> JSON_OBJECT: try: library_name = self.library_name except: library_name = 'unknown' try: library_version = self.library_version except: library_version = 'unknown' return { 'device_id': self.device_id, 'device_name': self.device_name, 'device_address': list(self.mac_address), 'serial_number': self.serial_number.as_json(), 'library_name': library_name, 'library_version': library_version, 'battery': self.battery, 'parent': self.parent, 'recording_file_name': self.recording_file_name, } def match(self, device: 'DeviceInfo'): return (self.device_id is None or self.device_id == device.device_id) and \ (self.device_name is None or self.device_name == device.device_name) and \ self.serial_number == device.serial_number and \ self.mac_address == device.mac_address class DeviceResponseInfo(DeviceInfo): """ **json format** :: DeviceResponseInfo = { "device_name" :str "device_address" :ADDRESS "serial_number" :DeviceSerialNumber "status" = "unknown" | "not_found" | "scanned" | "connected" "library_name": str? "library_version": str?, "device_id" :int? "battery" :int? # from 0 to 100 } """ STATUS_UNKNOWN = "unknown" STATUS_NOT_FOUND = "not_found" STATUS_SCANNED = "scanned" STATUS_CONNECTED = "connected" STATUS = (STATUS_UNKNOWN, STATUS_NOT_FOUND, STATUS_SCANNED, STATUS_CONNECTED) UNKNOWN: 'DeviceResponseInfo' = None __slots__ = ( '_device_id', '_device_name', '_serial_number', '_mac_address', '_library_name', '_library_version', '_status', '_battery', '_parent', '_recording_file_name', '_addr_type' ) def __init__(self, device_name: Optional[str], serial_number: DeviceSerialNumber, mac_address: ADDRESS, library_name: Optional[str] = None, library_version: Optional[str] = None, status: str = STATUS_UNKNOWN, device_id: Optional[int] = None, battery: Optional[int] = None, parent:Optional[dict] = None, recording_file_name:Optional[str] = None, addr_type: Optional[int] = None): if status is None: status = self.STATUS_UNKNOWN elif status not in self.STATUS: raise ValueError('illegal status : ' + status) if serial_number is None: raise RuntimeError() if mac_address is None: raise RuntimeError() self._device_id = device_id self._device_name = device_name self._serial_number = serial_number self._mac_address = mac_address self._library_name = library_name self._library_version = library_version self._status = status self._battery = battery self._parent = parent self._recording_file_name = recording_file_name self._addr_type = addr_type @property def device_id(self) -> Optional[int]: return self._device_id @property def device_name(self) -> Optional[str]: return self._device_name @property def serial_number(self) -> DeviceSerialNumber: return self._serial_number @property def mac_address(self) -> ADDRESS: return self._mac_address @property def library_name(self) -> Optional[str]: return self._library_name @property def library_version(self) -> Optional[str]: return self._library_version @property def battery(self) -> Optional[int]: return self._battery @property def parent(self) -> Optional[dict]: return self._parent @property def recording_file_name(self) -> Optional[str]: return self._recording_file_name @property def addr_type(self) -> Optional[int]: return self._addr_type @property def device_status(self) -> str: return self._status @device_status.setter def device_status(self, value: str): self._status = value def as_json(self): ret = super().as_json() ret['status'] = self._status return ret @classmethod def address_of(cls, address: ADDRESS) -> 'DeviceResponseInfo': return DeviceResponseInfo(None, DeviceSerialNumber.UNKNOWN, address) def __repr__(self): return str(self) def __str__(self): s1 = '%s,%s,%s' % (self._device_name, address_str(self._mac_address), str(self._serial_number)) if self._library_name is not None: s2 = self.library_version + ('-' + self._library_version) if self._library_version is not None else '' s2 = '[' + s2 + ']' else: s2 = '' if self._battery is None: s3 = '???' else: s3 = str(self._battery) return 'DeviceResponseInfo[%s]%s:%s[%s]' % (s1, s2, self._status, s3) def __hash__(self): ret = hash(self._device_name) ret = ret * 31 + hash(self._serial_number) ret = ret * 31 + hash(self._mac_address) ret = ret * 31 + hash(self._battery) return ret def __eq__(self, other): if self is other: return True if not isinstance(other, DeviceResponseInfo): raise False return ( self.device_name == other.device_name and self.serial_number == other.serial_number and self.mac_address == other.mac_address ) DeviceResponseInfo.UNKNOWN = DeviceResponseInfo(None, DeviceSerialNumber.UNKNOWN, EMPTY_ADDRESS) class DeviceInstruction: """low level device instruction type. **Instruction Send** raw byte data format. :: | 0 | 1 | 2 | 3 | 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 ----------------------------------------------------------------- | RIS | ID | | LEN | ............................. real instruction send | VIS | ID | OPER | LEN | ............................. virtual instruction send | CIS | ID | OPER | LEN | ............................. command instruction send ID chip ID RIS header for real instruction type. (header=3) VIS header for virtual instruction type. (header=C) CIS header for command instruction type. (header=7) OPER operator/command LEN, length command/data length in bytes """ TYP_RIS = 0x30 """header for real instruction type. It is real instruction send to the chip which contain the chip configuration parameter. """ TYP_VIS = 0xC0 """header for virtual instruction type. It is virtual instruction send to the controller chip to control the chip. """ TYP_CIS = 0x70 """header for command instruction type. It is command send to the controller chip for the special purpose. """ VIS_RST = 0xF0 """reset virtual instruction""" VIS_DEVICE_DETECT = 0x10 """identify device""" VIS_DEVICE_DONE = 0x20 """identify device done""" VIS_CC_ZERO = 0x40 VIS_ASK = 0x30 """ask in virtual instruction""" VIS_STI = 0xC0 """stimulation on virtual instruction""" VIS_FUH = 0x90 """flush virtual instruction""" VIS_INT = 0x60 """interrupt virtual instruction""" VIS_CAL = 0xA0 """device calling""" CIS_VOLT = 0x10 """get device battery voltage level""" CIS_LED = 0x20 """set device led level""" CIS_CALI = 0x30 """get calibration information table""" CIS_VERSION = 0x40 """get device verion""" CIS_RF = 0x50 """set device RF level""" CIS_AS = 0x60 """set device AS level""" VIS_STARTR = 0x10 VIS_STARTS = 0x20 VIS_STOPR = 0x30 VIS_STOPS = 0x40 TYP_IIS = -1 """internal instruction""" __slots__ = () @classmethod def valid_ins_type(cls, ins_type: int): if ins_type not in (cls.TYP_RIS, cls.TYP_VIS, cls.TYP_CIS, cls.TYP_IIS): raise ValueError('unknown instruction type : ' + str(ins_type)) @classmethod def get_instruction_type(cls, ins_type: str) -> int: if ins_type == 'RIS': return cls.TYP_RIS elif ins_type == 'VIS': return cls.TYP_VIS elif ins_type == 'CIS': return cls.TYP_CIS else: raise RuntimeError('unknown instruction type : ' + ins_type) class DeviceCommonInstruction: """high level device instruction.""" RESET = 'reset' START = 'start' INTERRUPT = 'interrupt' CLOSE = 'close' FLUSH = 'flush' CALL = 'call' RECORD = "record" RECORD_ALL = "record_all" STIMULATE = "stimulate" STOP_RECORD = "stop_record" STOP_STIMULATE = "stop_stimulate" VIS_DEVICE_DETECT = 'VIS_DEVICE_DETECT' VIS_DEVICE_DONE = 'VIS_DEVICE_DONE' VIS_CC_ZERO = 'VIS_CC_ZERO' CIS_VOLT = 'CIS_VOLT' CIS_VERSION = 'CIS_VERSION' ALL = [RESET, START, INTERRUPT, CLOSE, FLUSH, CALL] __slots__ = () # don't create DeviceCommonInstruction instance, just return tuple def __new__(cls, instruction: str) -> Optional[Tuple[str, ...]]: if instruction == cls.RESET: return '_sync(False)', '_notify(False)', 'VIS_RST' elif instruction == cls.START: return '_data_format("TDC4VAF2")', '_notify(True)', 'VIS_STI', '_sync(True)' elif instruction == cls.INTERRUPT: return '_sync(False)', '_notify(False)', 'VIS_INT' elif instruction == cls.CLOSE: return '_sync(False)', '_notify(False)', 'VIS_INT' elif instruction == cls.FLUSH: return 'VIS_FUH', elif instruction == cls.CALL: return 'VIS_CAL', elif instruction == cls.VIS_DEVICE_DETECT: return 'VIS_DEVICE_DETECT', elif instruction == cls.VIS_DEVICE_DONE: return 'VIS_DEVICE_DONE', elif instruction == cls.VIS_CC_ZERO: return 'VIS_CC_ZERO', elif instruction == cls.RECORD: return 'VIS_STARTR', elif instruction == cls.RECORD_ALL: return 'VIS_STARTR', elif instruction == cls.STIMULATE: return 'VIS_STARTS', elif instruction == cls.STOP_RECORD: return 'VIS_STOPR', elif instruction == cls.STOP_STIMULATE: return 'VIS_STOPS', elif instruction == cls.CIS_VOLT: return 'CIS_VOLT', elif instruction == cls.CIS_VERSION: return 'CIS_VERSION', else: return None class DeviceFeature: """TODO add doc""" BATTERY = 0x0001 CALIBRATION = 0x0002 class DeviceFeatureUnsupportedError(RuntimeError): def __init__(self, device_id: int, feature: int): super().__init__('device [%d] do not support feature code : 0x%04X' % (device_id, feature)) class Device(DeviceInfo, metaclass=abc.ABCMeta): """device with partial function which only provide necessary information. **json response format** :: Device = { "device_id" :int "device_name" :str "device_address" :mac_address "serial_number" :DeviceSerialNumber "library_name" :str "library_version" :str "battery" :int # from 0 to 100 # for completed device "configuration" :DeviceConfiguration } mac_address : [int, int, int, int, int, int] """ __slots__ = () # library information @property def library_name(self) -> str: """device library name. *NOTE* this property has overwrote by the CompletedDevice. the subclass do not need to overwrite this. """ raise RuntimeError('implement by CompletedDevice') @property def library_version(self) -> str: """device library version. *NOTE* this property has overwrote by the CompletedDevice. the subclass do not need to overwrite this. """ raise RuntimeError('implement by CompletedDevice') # device information @property def device_status(self) -> str: return DeviceResponseInfo.STATUS_CONNECTED @property def response_info(self) -> DeviceResponseInfo: """response information""" return DeviceResponseInfo(self.device_name, self.serial_number, self.mac_address, self.library_name, self.library_version, DeviceResponseInfo.STATUS_CONNECTED, self.device_id, self.battery, self.parent, self.recording_file_name) # device feature @abc.abstractmethod def read_feature_mask(self) -> bytes: pass def is_feature_support(self, feature: int) -> bool: raise RuntimeError('implement by CompletedDevice') def ensure_feature_support(self, feature: int): if not self.is_feature_support(feature): raise DeviceFeatureUnsupportedError(self.device_id, feature) # device instruction @abc.abstractmethod def read_command_return_data(self) -> Optional[bytes]: pass @abc.abstractmethod def send_instruction(self, ins_type: int, ins_oper: int, *instruction: int): """send instruction to device :param ins_type: instruction type :param ins_oper: instruction :param instruction: content :raises: DeviceInstructionError """ pass # device specific functions @property def device_version(self) -> str: return '' @property def battery(self) -> int: return 100 def update_battery_info(self): """update device battery information """ pass def update_parent_info(self, _parent: dict): """update device parent information """ pass def update_recording_file_name_info(self, _recording_file_name: str): """update device recording_file_name information """ pass def update_recording_file_name_info(self, _recording_file_name: str): """update device parrecording_file_nameent information """ pass def calibration_info(self, device_type: str) -> bytes: raise DeviceFeatureUnsupportedError(self.device_id, DeviceFeature.CALIBRATION) def update_calibration_info(self, device_type: str): raise DeviceFeatureUnsupportedError(self.device_id, DeviceFeature.CALIBRATION) def update_radio_frequency(self, radio_frequency): """update device radio frequency""" pass def update_led_brightness(self, led_brightness): """update device led brightness""" pass def update_accelerator_sensitivity(self, accelerator_sensitivity): """update device accelerator sensitivity""" pass class MasterDevice(LoggerFlag, metaclass=abc.ABCMeta): """master device, the actual hardware device to connect/control the slave device.""" __slots__ = () def __init__(self, name: Optional[str] = None, log_name_color: int = WHITE) -> None: name = name if name is not None else self.__class__.__name__ LoggerFlag.__init__(self, name, log_name_color) @abc.abstractmethod def reset(self, device: Optional[List[int]] = None, software_reset=True): """reset master device :param device: reset certain device. None for all :param software_reset: """ pass @abc.abstractmethod def shutdown(self, release_resource=True): """shutdown the master device. *release_resource* used when more than one devices share the same resource. close the resource when other device is using, may cause hardware error. This parameter is internal used because only work functionally in certain device class, user should not care it. :param release_resource: close the resource. """ pass def scan(self, timeout=5, all_device=False) -> List[DeviceResponseInfo]: """scan the nearby device. make sure the result should be equal to found(). :param timeout: max scanning time. :param all_device: all device perform scanning if not working. this options only work on some implements. :return: found device """ error = None try: result = self.scan_callback((lambda info: None), timeout=timeout, all_device=all_device) except RuntimeError as e: error = e else: if not result: raise RuntimeError(HARDWARE_SCAN_FAIL) 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 MasterDeviceError(self) return self.found() @abc.abstractmethod def scan_callback(self, callback: Callable[[DeviceResponseInfo], None], timeout=5, all_device=False) -> bool: """scan the nearby device. perform device scanning :param callback: :param timeout: max scanning time :param all_device: all device perform scanning if not working. this options only work on some implements. :return: success or not """ pass def scan_device(self, response: DeviceInfo, timeout=5) -> Optional[DeviceResponseInfo]: """scan until found the device *response*, or *timeout* reached. :param response: :param timeout: :return: """ t = time() while time() - t < timeout: found = self.scan(all_device=True) for f in found: if response.match(f): return f return None @abc.abstractmethod def found(self) -> List[DeviceResponseInfo]: """found nearby device""" pass @abc.abstractmethod def connect(self, response: DeviceResponseInfo) -> Device: """connect to the device""" pass @abc.abstractmethod def disconnect(self, device: int, force=False) -> bool: """disconnect the device :param device: device ID :param force: :return: success of not """ pass @abc.abstractmethod def get_device(self, device: int) -> Optional[Device]: """get the device by device ID""" pass def get_menu(self) -> Optional[Menu]: """get master device menu""" return None def handle_command(self, device: Optional[int], command: str, options: Dict[str, str]) -> DeviceCommandAction: """device command :param device: device ID :param command: command :param options: command options :return: action """ self._raise_command_not_found(device, command) def handle_internal_instruction(self, device: Device, func: str, *para: Any) -> bool: return False @staticmethod def _raise_command_not_found(device: Optional[int], command: str) -> NoReturn: if device is None: message = command else: message = "%d:%s" % (device, command) raise CommandNotFoundError(message) def hardware_test(self) -> Dict[str, Any]: return {} class NullMasterDevice(MasterDevice): """no function master device.""" __slots__ = () def __init__(self, name: Optional[str] = None): super().__init__(name if name is not None else 'NullDev') def reset(self, device: Optional[List[int]] = None, software_reset=True): pass def shutdown(self, release_resource=True): pass def scan(self, timeout=5, all_device=False) -> List[DeviceResponseInfo]: return [] def scan_callback(self, callback: Callable[[DeviceResponseInfo], None], timeout=5, all_device=False) -> bool: return False def scan_device(self, response: DeviceInfo, timeout=5) -> Optional[DeviceResponseInfo]: return None def found(self) -> List[DeviceResponseInfo]: return [] def connect(self, response: DeviceResponseInfo) -> Device: raise NotImplementedError('connect') def disconnect(self, device: int, force=False) -> bool: raise NotImplementedError('disconnect') def get_device(self, device: int) -> Optional[Device]: return None def hardware_test(self) -> Dict[str, Any]: return {'hardware': 'null'} def get_device_mac_in_address_format(raw_mac: list) -> ADDRESS: if len(raw_mac) is 6: return raw_mac[5], raw_mac[4], raw_mac[3], raw_mac[2], raw_mac[1], raw_mac[0] else: return EMPTY_ADDRESS def get_device_name_in_string_format(raw_name: list) -> Optional[str]: # initialization of string to "" ret = "" # traverse in the string for x in raw_name: if x is not 0: try: ret += chr(x) except UnicodeDecodeError: print("UnicodeDecodeError: ", raw_name) ret = "" # return string # ret.rstrip("\x00") if ret is "": return None else: return ret def get_device_company_code(raw_code: list) -> Optional[tuple]: if len(raw_code) is 4: return chr(raw_code[0]), chr(raw_code[1]), chr(raw_code[2]), chr(raw_code[3]) else: return None # used in scan response battery information def get_device_battery_info(raw_code: list) -> Optional[tuple]: if len(raw_code) is 5: return chr(raw_code[0]), chr(raw_code[1]), chr(raw_code[2]), raw_code[3], raw_code[4] else: return None