import gzip import json as _json import os from .header import * from .instruction import * from .lock import * from .menu import * class DeviceLibrary(JsonSerialize): """ **file extension name** ``.devlib.json`` **json format** :: DeviceLibrary = { "name" :str "specification" :int "version" :DeviceLibraryVersion "match_rule" :MatchRule "constant"? : {name=value} "parameters" :ParameterTable "instruction" :InstructionTable "menu": [MenuEntry] } **variable scope search chain** general case root. 1. "constant" #. device constant. defined in :class:`DeviceScope` #. "parameters". #. each table (describe in table document) **special library** ============ ========================================================================== library name description ============ ========================================================================== Elite_Legacy legacy Elite device, which scan response do not follow the new specification, which need to use additional condition to filter it. GENERATOR dummy device who use generator module to generate artifact data, all generator options are export as the parameters ============ ========================================================================== **device menu** """ FILE_EXT = ('.devlib.json', '.devlib') FILE_HEADER = b'BPDevLib' __slots__ = ('_file', '_name', '_version', '_match_rule', '_constant', '_para', '_inst') def __init__(self, name: str, version: DeviceLibraryVersion): """ :param name: device library name :param version: device library version """ self._file: Optional[str] = None self._name: str = name self._version: DeviceLibraryVersion = version self._match_rule: MatchRule = ListMatchRule([]) self._constant: ConstantScope = ConstantScope.EMPTY self._para: ParameterTable = ParameterTable() self._inst = InstructionTable() @property def name(self) -> str: """ :return: device library name """ return self._name @property def version(self) -> DeviceLibraryVersion: """ :return: device library version """ return self._version @property def file_path(self) -> str: return self._file @property def constant(self) -> ConstantScope: """constant pool. :return: parameter scope """ return self._constant @property def match_rule(self) -> MatchRule: """ :return: device match rule """ return self._match_rule @property def parameter_table(self) -> ParameterTable: """ :return: device parameter table """ return self._para @property def instruction_table(self) -> InstructionTable: """ :return: instruction table """ return self._inst def is_device_match(self, device: DeviceResponseInfo) -> bool: return self._match_rule.is_device_match(device) is not None def as_json(self): return { 'name': self._name, 'specification': AbstractLibraryLoader.SPECIFICATION, 'version': str(self._version), 'match_rule': self._match_rule.as_json(), 'constant': self._constant.locals(), 'parameters': self._para.as_json(), 'instruction': self._inst.as_json(), } def __repr__(self): return str(self) def __str__(self): return 'DeviceLibrary:%s-%s' % (self._name, str(self._version)) @classmethod def is_filename_match(cls, file: Union[str, Path]) -> bool: if isinstance(file, str): name = file else: name = file.name return any(map(lambda it: name.endswith(it), cls.FILE_EXT)) @classmethod def load(cls, file: Union[str, Path]) -> 'DeviceLibrary': """ :param file: json file :return: """ if isinstance(file, str): file = Path(file) if not isinstance(file, Path): raise TypeError('not a path : ' + str(file)) if file.name.endswith(cls.FILE_EXT[0]): with file.open() as f: ret = cls.parse(file, _json.load(f)) elif file.name.endswith(cls.FILE_EXT[1]): with file.open('rb') as f: header = f.read(8) if header != cls.FILE_HEADER: raise RuntimeError('unknown file : ' + str(file)) with gzip.GzipFile(fileobj=f) as g: # json do not accept load(bytes) until python 3.6 ret = cls.parse(file, _json.loads(g.read().decode('utf8', 'surrogatepass'))) else: raise RuntimeError('unknown file ext : ' + str(file.name)) return ret @classmethod def parse(cls, path: Path, json: JSON_OBJECT) -> 'DeviceLibrary': loader = AbstractLibraryLoader.parse(path, json) ret = DeviceLibrary(loader.name, loader.version) ret._file = str(path) ret._match_rule = loader.match_rule ret._constant = loader.constant ret._para = loader.parameter ret._inst = loader.instruction return ret @classmethod def load_library(cls, path: List[Union[str, Path]], recursive=False, on_error: Callable[[Path, BaseException], None] = None) -> List['DeviceLibrary']: ret = [] def _load(_f: Path): if _f.is_file() and cls.is_filename_match(_f): try: lib = cls.load(_f) except BaseException as e: if on_error is not None: on_error(_f, e) else: ret.append(lib) for p in path: if isinstance(p, str): p = Path(p) elif not isinstance(p, Path): raise TypeError('not a path : ' + str(p)) if p.exists(): if recursive: for d, ds, fs in os.walk(str(p), followlinks=True): d = Path(d) for f in fs: _load(d / f) else: for f in p.iterdir(): _load(f) return ret class DeviceLibraryRepository: def __init__(self, path: List[Path] = None, no_system_default=False): self._library_path = [] if not no_system_default: self._library_path.append(Path('/home/pi/wisetopdataserver/python/res')) if path is not None: self._library_path.extend(path) self._recursive = False self._on_error: Callable[[Path, BaseException], None] = None self._cache: Optional[List[DeviceLibrary]] = None @property def library_path(self) -> List[Path]: return list(self._library_path) def add_library_path(self, path: Union[str, Path]): if isinstance(path, str): path = Path(path) self._library_path.append(path) self._cache = None @property def on_error(self) -> Callable[[Path, BaseException], None]: return self._on_error @on_error.setter def on_error(self, value: Callable[[Path, BaseException], None]): self._on_error = value @property def recursive(self) -> bool: return self._recursive @recursive.setter def recursive(self, value: bool): self._recursive = value def invalid_cache(self): self._cache = None def get_library_all(self) -> List[DeviceLibrary]: if self._cache is None: self._cache = DeviceLibrary.load_library(self._library_path, recursive=self._recursive, on_error=self._on_error) return self._cache def get_library(self, library_name: str, library_version: Union[str, DeviceLibraryVersion] = None) -> Optional[DeviceLibrary]: if isinstance(library_version, str): library_version = DeviceLibraryVersion.value_of(library_version) for lib in self.get_library_all(): if lib.name == library_name: if library_version is None or lib.version == library_version: return lib return None def match_library(self, response: DeviceResponseInfo) -> Optional[DeviceLibrary]: for lib in self.get_library_all(): if lib.is_device_match(response): return lib return None class DeviceParameter(JsonSerialize): """ **json format** :: DeviceParameter = { "library_name" :str # device library name "library_version" :str # device library version "parameter" = {parameter_name = parameter_value} } parameter_value = null | bool | int | float | str # current not supported | [int] # list | set # set set = { "set": [int] } **required parameter** =========== ======= ========================== parameter type description =========== ======= ========================== SAMPLE_RATE int sampling rate of the data CHANNEL [int] used channel ID list AMP_GAIN int amp gain value =========== ======= ========================== """ __slots__ = '_library', '_parameter', '_context', '_flag_disable_guarding', '_last_modified_timestamp' def __init__(self, library: DeviceLibrary): self._library = library self._parameter: Dict[str, Any] = {} self._context = library.constant.chain(DeviceParameterScope(self)) # initial parameter table for para_info in library.parameter_table.values(): try: para_value = para_info.init_para() except (PropertyDomainError, ActionDomainError): pass else: if isinstance(para_value, list): if para_info.name == ParameterInfo.CHANNEL: para_value = set(para_value) else: para_value = list(para_value) elif isinstance(para_value, set): para_value = set(para_value) elif isinstance(para_value, (bool, int, float)): # raw parameter value pass else: raise RuntimeError('unsupported value type for parameter %s : %s' % (para_info.name, type(para_value))) self._parameter[para_info.name] = para_value # check required and its type # sample_rate sample_rate = self[ParameterInfo.SAMPLE_RATE] if not isinstance(sample_rate, int): raise RuntimeError('illegal sample rate type : ' + str(type(sample_rate))) # amp_gain amp_gain = self[ParameterInfo.AMP_GAIN] if not isinstance(amp_gain, int): raise RuntimeError('illegal amp gain type : ' + str(type(sample_rate))) # channel channel = self[ParameterInfo.CHANNEL] if isinstance(channel, (list, set)): if len(channel) > 0 and not all(map(lambda v: isinstance(v, int), channel)): raise RuntimeError('illegal channel contain type + ' + str(channel)) else: raise RuntimeError('illegal channel type : ' + channel.__class__.__name__) # last modified time stamp self._last_modified_timestamp = time() # debug use flag self._flag_disable_guarding = False @property def library(self) -> DeviceLibrary: return self._library @property def context(self) -> Scope: return self._context @property def last_modified_timestamp(self) -> int: return int(self._last_modified_timestamp) def __len__(self) -> int: return len(self._library.parameter_table) def __contains__(self, item) -> bool: return item in self._library.parameter_table def __getitem__(self, item: str) -> Any: return self.get_parameter(item) def __setitem__(self, item: str, value: int): self.set_parameter(item, value) def get_parameter(self, para: str, raw_data: bool = False) -> Any: """get parameter P value :param para: parameter name :param raw_data: :return: parameter P value """ table = self._library.parameter_table info = table[para] domain = info.domain if domain == ParameterPropertyDomain: return table.get_value(info, None, self._context) if domain == ParameterActionDomain: return None ret = self._parameter[info.name] return ret ### ##print('111', domain.get_para(ret, self._context)) #print('111', ret) # if not raw_data and isinstance(domain, ParameterArrayDomain): # return domain.get_para(ret, self._context) # else: # return ret def set_parameter(self, para: str, value: int): """set parameter with P value :param para: parameter name :param value: parameter P value """ table = self._library.parameter_table info = table[para] domain = info.domain if domain == ParameterConstantDomain: raise ConstantDomainError(para) if isinstance(domain, ParameterCollectionDomain): raise RuntimeError('collection parameter ' + para + ' do not support this operation') if domain == ParameterPropertyDomain: return if domain == ParameterActionDomain: return if not domain.valid_para(value): raise RuntimeError('invalid parameter ' + para + ' : ' + str(value)) old = self._parameter[para] self._parameter[para] = value if not self._flag_disable_guarding: try: table.ensure_guard(info, self._context) except RuntimeError as e: self._parameter[para] = old raise IllegalParameterValueError(para, 'illegal value : ' + str(value)) from e self._last_modified_timestamp = time() def oper_parameter(self, para: str, oper: Union[str, ParameterCollectionValueOperator]): """do operator on parameter :param para: parameter name :param oper: operator expression """ if not isinstance(oper, ParameterCollectionValueOperator): oper = ParameterCollectionValueOperator.parse(oper) table = self._library.parameter_table info = table[para] domain = info.domain if not isinstance(domain, ParameterCollectionDomain): raise RuntimeError('not a collection parameter ' + para) target = self._parameter[para] if isinstance(target, set): old = set(target) else: old = list(target) try: domain.oper_para(target, oper, self._context) except RuntimeError as e: raise IllegalParameterValueError(para, str(oper)) from e if not self._flag_disable_guarding: try: table.ensure_guard(info, self._context) except RuntimeError as e: self._parameter[para] = old raise IllegalParameterValueError(para, str(oper)) from e self._last_modified_timestamp = time() def keys(self) -> Iterable[str]: return self._library.parameter_table.keys() def keys_all(self) -> Iterable[str]: return self._library.parameter_table.keys_all(show_private=True) def overwrite(self, other: Union[Dict[str, Any], 'DeviceParameter', 'DeviceConfiguration']): """overwrite self parameter from *other* :param other: source """ if isinstance(other, DeviceConfiguration): other = self._overwrite_for_configuration(other) elif isinstance(other, DeviceParameter): other = other._parameter elif isinstance(other, dict): pass else: raise TypeError('unsupported overwrite type : ' + other.__class__.__name__) for k, v in other.items(): self._parameter[k] = v self._last_modified_timestamp = time() def _overwrite_for_configuration(self, conf: 'DeviceConfiguration') -> Dict[str, Any]: table = self._library.parameter_table tmp = {} dep: List[ParameterInfo] = [] for k in conf.keys(list_hide=True): if k in table: para = table[k] if len(para.cast_value_dependency()) == 0: try: tmp[k] = table.cast_value(k, conf[k], self._context) except (PropertyDomainError, ActionDomainError): pass else: dep.append(para) # handle dependency parameter while len(dep) > 0: dep_next = [] dep_size = len(dep) context = self._context.child(**tmp) for para in dep: if all(map(lambda it: it in tmp, para.cast_value_dependency())): k = para.name try: # FIXME extend self._context with tmp tmp[k] = table.cast_value(k, conf[k], context) except (PropertyDomainError, ActionDomainError): pass else: dep_next.append(para) if len(dep_next) == dep_size: raise RuntimeError('recursive dependency : ' + ' '.join(map(lambda it: it.name, dep_next))) dep = dep_next return tmp def as_json(self): return { 'library_name': self._library.name, 'library_version': str(self._library.version), 'parameter': {k: JsonSerialize.to_json(v) for k, v in self._parameter.items()} } class DeviceParameterScope(ReadonlyScope): __slots__ = ('_parameter',) def __init__(self, parameter: DeviceParameter): super().__init__(list(parameter.keys_all())) self._parameter = parameter def get_local(self, key: str) -> Any: ### return self._parameter.get_parameter(key, raw_data=False) def __str__(self): return 'DeviceParameterScope' class DeviceConfiguration(JsonSerialize, metaclass=abc.ABCMeta): """device configuration. **json format** :: DeviceConfiguration = { "_LIBRARY_": str "_VERSION_": str "CHANNEL" :[int] "SAMPLE_RATE" :int "AMP_GAIN" :int # optional information "_DEVICE_NAME_": str "_DEVICE_ADDR_": [int] # above are required parameter parameter_name = parameter_value } **required parameter** ============= ===== ========================== parameter type description ============= ===== ========================== _LIBRARY_ str used library name _VERSION_ str used library version _DEVICE_NAME_ str device name _DEVICE_ADDR_ [int] device mac address CHANNEL [int] SAMPLE_RATE int AMP_GAIN int ============= ===== ========================== XXX _DEVICE_ XXX _ADDRESS_ other required parameters are defined in :class:`DeviceParameter`. **special LIBRARY** =============== ================================================================ _LIBRARY_ description =============== ================================================================ GENERATOR use generator module to generate artifact data, all generator options are export as the parameters =============== ================================================================ """ LIBRARY = '_LIBRARY_' VERSION = '_VERSION_' DEVICE_NAME = '_DEVICE_NAME_' DEVICE_ADDR = '_DEVICE_ADDR_' SAMPLE_RATE = ParameterInfo.SAMPLE_RATE CHANNEL = ParameterInfo.CHANNEL AMP_GAIN = ParameterInfo.AMP_GAIN REQUIRED = (LIBRARY, VERSION, SAMPLE_RATE, CHANNEL, AMP_GAIN) __slots__ = () @abc.abstractmethod def keys(self, list_hide=False, list_private=False) -> List[str]: """list parameter in configuration. :param list_hide: list non meta recorded parameter :param list_private: list private parameter :return: parameter name list """ pass def __contains__(self, item: str) -> bool: return self.contain_parameter(item) def __getitem__(self, item: str): return self.get_parameter(item) def __setitem__(self, item: str, value: Any): self.set_parameter(item, value) @property @abc.abstractmethod def library(self) -> str: """ :return: library name """ pass @property @abc.abstractmethod def library_version(self) -> str: """ :return: library version """ pass @property def device_name(self) -> Optional[str]: """ :return: device name """ return None @property def device_address(self) -> Optional[ADDRESS]: """ :return: device address """ return None @property def sample_rate(self) -> int: return int(self.get_parameter(ParameterInfo.SAMPLE_RATE)) @property def channel(self) -> List[int]: return list(self.get_parameter(ParameterInfo.CHANNEL)) @property def amp_gain(self) -> int: return int(self.get_parameter(ParameterInfo.AMP_GAIN)) def get(self, name: str, default=None): """ :param name: parameter name :param default: value if parameter not found :return: parameter P value or default """ try: return self.get_parameter(name) except KeyError: return default @abc.abstractmethod def contain_parameter(self, name: str) -> bool: """Does parameter in this configuration? :param name: parameter name. :return: """ pass @abc.abstractmethod def get_parameter(self, name: str, raw_data: bool = False) -> Any: """get parameter V value :param name: parameter name. :param raw_data: raw V value, especially for array parameter. :return: parameter V value """ pass @abc.abstractmethod def set_parameter(self, name: str, value: Union[int, str]): """set parameter V value :param name: parameter name :param value: new V value """ pass def as_json(self, list_hide=False, list_private=False, list_disable=False): ret = { self.LIBRARY: self.library, self.VERSION: self.library_version, } if list_disable is False: ret = { self.LIBRARY: self.library, self.VERSION: self.library_version, ParameterInfo.SAMPLE_RATE: self.sample_rate, ParameterInfo.CHANNEL: self.channel, ParameterInfo.AMP_GAIN: self.amp_gain } for k in self.keys(list_hide=list_hide, list_private=list_private): if k not in ParameterInfo.REQUIRED: ### not raw_data v = self.get_parameter(k, raw_data=False) if isinstance(v, set): v = list(v) ret[k] = v return ret class SimpleDeviceConfiguration(DeviceConfiguration): __slots__ = ('_parameter',) def __init__(self, parameter: Union[Dict[str, Any], DeviceConfiguration]): if isinstance(parameter, DeviceConfiguration): self._parameter = {k: parameter[k] for k in parameter.keys()} else: for req in ParameterInfo.REQUIRED: if req not in parameter: raise RuntimeError('lost required configuration key : ' + req) self._parameter = parameter def keys(self, list_hide=False, list_private=False) -> List[str]: return list(self._parameter.keys()) @property def library(self) -> str: return str(self.get_parameter(self.LIBRARY)) @property def library_version(self) -> str: return str(self.get_parameter(self.VERSION)) @property def device_name(self) -> Optional[str]: try: return str(self.get_parameter(self.DEVICE_NAME)) except (KeyError, ValueError): return None @property def device_address(self) -> Optional[ADDRESS]: try: return tuple(self.get_parameter(self.DEVICE_ADDR)) except KeyError: return None def contain_parameter(self, name: str) -> bool: return name in self._parameter def get_parameter(self, name: str, raw_data: bool = False) -> Any: return self._parameter[name] def set_parameter(self, name: str, value: Any): self._parameter[name] = value class DeviceParameterConfiguration(DeviceConfiguration): __slots__ = ('_library', '_parameter') def __init__(self, library: DeviceLibrary, parameter: DeviceParameter): self._library = library self._parameter = parameter @property def library(self) -> str: return self._library.name @property def library_version(self) -> str: return str(self._library.version) @property def device_name(self) -> Optional[str]: return None @property def device_address(self) -> Optional[ADDRESS]: return None @property def context(self) -> Scope: return self._parameter.context def contain_parameter(self, name: str) -> bool: return name in self._parameter or name == self.LIBRARY or name == self.VERSION def get_parameter(self, name: str, raw_data: bool = False) -> Any: if name == self.LIBRARY: return self._library.name elif name == self.VERSION: return str(self._library.version) table = self._library.parameter_table return self._parameter.get_parameter(name, raw_data=False) ### # print('222', value) # return value ## return table.get_value(name, value, self.context) def set_parameter(self, name: str, value: Union[int, str]): if name == self.LIBRARY: return elif name == self.VERSION: return try: para = self._library.parameter_table.cast_value(name, value, self.context) except PropertyDomainError: pass else: self._parameter[name] = para def keys(self, list_hide=False, list_private=False) -> List[str]: ret = [self.LIBRARY, self.VERSION] ret.extend(map(lambda p: p.name, filter(lambda p: (list_hide or p.record_meta) and (list_private or not p.private), self._library.parameter_table.values()))) return ret class CompletedDevice(Device): """ Implement Note -------------- the subclass DebugDevice._device reference to it self, so all the the function implements which call the _device function should be override in DebugDevice. Otherwise, StackOverflow. """ __slots__ = ('_master', '_device_id', '_device', '_library', '_context', '_parameter', '_configuration', '_lock', '_feature_mask', '_cache_battery', '_cache_battery_timestamp', '_coeff') def __init__(self, master: MasterDevice, library: DeviceLibrary, device_id: int, device: Device): """ :param master: owner master device :param library: device library :param device_id: device id :param device: slave device implementation """ self._master = master self._device = device self._device_id = device_id self._library = library # context s1 = library.constant s2 = DeviceScope(self) s3 = CompleteDeviceParameterScope(self, library.parameter_table) self._context: CompleteDeviceParameterScope = s1.chain(s2).chain(s3) # parameter init self._parameter = DeviceParameter(library) self._configuration = CompletedDeviceConfiguration(self) # device lock self._lock = DeviceLock() # feature support mask self._feature_mask: Optional[bytes] = None # cache battery self._cache_battery: Optional[int] = None self._cache_battery_timestamp: Optional[float] = None self._coeff = b'' # device property for information @property def device_id(self) -> int: return self._device_id @property def device_name(self) -> str: return self._device.device_name @property def serial_number(self) -> DeviceSerialNumber: return self._device.serial_number @property def mac_address(self) -> ADDRESS: return self._device.mac_address @property def device(self) -> Device: return self._device @property def library_name(self) -> str: return self._library.name @property def library_version(self) -> str: return str(self._library.version) # device data @property def parameter(self) -> DeviceParameter: return self._parameter @property def configuration(self) -> DeviceConfiguration: """device configuration """ return self._configuration @property def context(self) -> Scope: """general case root""" return self._context def configuration_reset(self): self._context.reset() @property def library(self) -> DeviceLibrary: return self._library # device lock @property def lock(self) -> DeviceLock: return self._lock # device parameter getter/setter def parameters(self) -> List[str]: return list(self._library.parameter_table.keys()) def get_parameter(self, name: str, raw_data=False): """get parameter value. :param name: parameter name :param raw_data: raw parameter P value :return: parameter P value """ return self._parameter.get_parameter(name, raw_data=False) def get_parameter_value(self, name: str): """get parameter present value. :param name: parameter name :return: parameter V value """ return self._configuration.get_parameter(name, False) def set_multi_parameters(self, parameter): for (name, value) in parameter[0].items(): if name != 'target': self.set_parameter(name, value) def set_parameter(self, name: str, value: Union[int, str]): """replace parameter value with *value* :param name: parameter name :param value: new parameter P value """ info = self._library.parameter_table[name] if isinstance(info.domain, ParameterCollectionDomain): self._parameter.oper_parameter(name, value) else: try: # value int value = int(value) except ValueError as e: # value float self._parameter.set_parameter(name, value) # raise RuntimeError('illegal parameter value : ' + value) from e else: self._parameter.set_parameter(name, value) on_change = info.on_change if on_change is not None: if on_change not in self._library.instruction_table: self._master.log_warn('on_change instruction not found : ' + on_change) else: self.call_instruction(on_change) # device instruction call def instructions(self) -> List[str]: return list(self._library.instruction_table.keys()) def reset(self): self.call_instruction(DeviceCommonInstruction.RESET) def interrupt(self): self.call_instruction(DeviceCommonInstruction.INTERRUPT) def start(self): self.call_instruction(DeviceCommonInstruction.START) def call_instruction(self, command: str): try: self._library.instruction_table.call_instruction(self._master, self, self._context, command) except DeviceInstructionError as e: raise DeviceInstructionError(command, e.message) from e def call_command(self, command: int): if not self.is_feature_support(command): raise RuntimeError('feature not support') # TODO impl call_command() # TODO decode bytes data pass def read_feature_mask(self) -> bytes: if self._feature_mask is None: self._feature_mask = self._device.read_feature_mask() return self._feature_mask def is_feature_support(self, feature: int) -> bool: mask = self.read_feature_mask() return mask[feature // 8] & (1 << (feature % 8)) != 0 def read_command_return_data(self) -> Optional[bytes]: try: return self._device.read_command_return_data() except DeviceInstructionError: raise except RuntimeError as e: raise DeviceInstructionError("CDR") from e def send_instruction(self, ins_type: int, ins_oper: int, *instruction: int): try: self._device.send_instruction(ins_type, ins_oper, *instruction) # print('----------------') # print('send_instruction') # print('type:',ins_type) # print('oper:',ins_oper) # print('ins:',*instruction) except DeviceInstructionError: raise except RuntimeError as e: ins_t = { DeviceInstruction.TYP_RIS: 'RIS', DeviceInstruction.TYP_VIS: 'VIS', DeviceInstruction.TYP_CIS: 'CIS', }.get(ins_type, None) raise DeviceInstructionError(ins_t) from e # device specific functions @property def device_version(self) -> str: return self._device.device_version @property def battery(self) -> int: if self._check_battery_expired(): self._cache_battery = self._device.battery return self._cache_battery @property def calibration(self) -> bytes: return self._coeff @property def parent(self) -> dict: return self._device.parent @property def recording_file_name(self) -> str: return self._device.recording_file_name def update_battery_info(self): self._cache_battery_timestamp = time() self._device.update_battery_info() def update_parent_info(self, _parent): self._device.update_parent_info(_parent) def update_recording_file_name_info(self, _recording_file_name): self._device.update_recording_file_name_info(_recording_file_name) def _check_battery_expired(self) -> bool: current_time_stamp = time() if self._cache_battery_timestamp is None: self._cache_battery_timestamp = time() return True else: if current_time_stamp - self._cache_battery_timestamp >= 60: self._cache_battery_timestamp = current_time_stamp return True else: return False def calibration_info(self, device_type: str) -> bytes: self._coeff = self._device.calibration_info(device_type) return self._coeff def update_calibration_info(self, device_type: str): self._device.update_calibration_info(device_type) def update_radio_frequency(self, radio_frequency): self._device.update_radio_frequency(radio_frequency) def update_led_brightness(self, led_brightness): self._device.update_led_brightness(led_brightness) def update_accelerator_sensitivity(self, accelerator_sensitivity): self._device.update_accelerator_sensitivity(accelerator_sensitivity) def central_rf_set(self, content: int) -> bool: self._device.central_rf_set(content) def central_version_get(self) -> Optional[list]: return self._device.central_version_get() # utility method def as_json(self) -> JSON_OBJECT: return { 'device_id': self.device_id, 'device_name': self.device_name, 'device_address': list(self.mac_address), 'battery': self.battery, 'parent': self.parent, 'recording_file_name': self.recording_file_name, 'device_version': self.device_version, 'serial_number': self.serial_number.as_json(), 'library_name': self._library.name, 'library_version': str(self._library.version), 'configuration': self.configuration.as_json(list_hide=True), } 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)) s2 = self.library_version + '-' + self.library_version return 'Device[%d][%s][%s][%d]' % (self._device_id, s1, s2, self.battery) def __eq__(self, other): if self is other: return True elif other is None: return False elif isinstance(other, CompletedDevice): return self.device_id == other.device_id and \ self.device_name == other.device_name and \ self.serial_number == other.serial_number and \ self.mac_address == other.mac_address and \ self.library_name == other.library_name and \ self.library_version == other.library_version return False class DebugDevice(CompletedDevice): """debug-use device which implement the CompletedDevice interface""" __slots__ = '_device_name', '_serial_number', '_mac_address' def __init__(self, master: MasterDevice, library: DeviceLibrary, device_id: int, device_name: str = None, serial_number: DeviceSerialNumber = None, mac_address: ADDRESS = None, response: DeviceResponseInfo = None): super().__init__(master, library, device_id, self) # device_name if device_name is None and response is not None: device_name = response.device_name if device_name is None: device_name = 'DebugDevice' self._device_name = device_name # serial_number if serial_number is not None: self._serial_number = serial_number elif response is not None: self._serial_number = response.serial_number else: self._serial_number = DeviceSerialNumber.UNKNOWN # mac_address if mac_address is not None: self._mac_address = mac_address elif response is not None: self._mac_address = response.mac_address else: self._mac_address = EMPTY_ADDRESS @property def device_name(self) -> 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 def read_feature_mask(self) -> bytes: # TODO return b'' def read_command_return_data(self) -> Optional[bytes]: return None def send_command(self, ins_type: int, ins_oper: int, *instruction: int): length = len(instruction) print('instruction send :', '%02X' % ((ins_type & 0xF0) | (self.device_id & 0x0F)), '%02X' % ((ins_oper & 0xF0) | (length & 0x0F)), ' '.join(map(lambda v: '%02X' % v, instruction))) def send_instruction(self, ins_type: int, ins_oper: int, *instruction: int): length = len(instruction) print('instruction send :', '%02X' % ((ins_type & 0xF0) | (self.device_id & 0x0F)), '%02X' % ((ins_oper & 0xF0) | (length & 0x0F)), ' '.join(map(lambda v: '%02X' % v, instruction))) @property def battery(self) -> int: return 100 def update_battery_info(self): pass def update_parent_info(self): pass def update_recording_file_name_info(self): pass def calibration_info(self, device_type: str) -> NoReturn: raise DeviceFeatureUnsupportedError(self.device_id, DeviceFeature.CALIBRATION) def update_calibration_info(self, device_type: str) -> NoReturn: raise DeviceFeatureUnsupportedError(self.device_id, DeviceFeature.CALIBRATION) class CompletedDeviceConfiguration(DeviceParameterConfiguration): __slots__ = ('_device',) def __init__(self, device: CompletedDevice): super().__init__(device.library, device.parameter) self._device = device @property def device_name(self) -> Optional[str]: return self._device.device_name @property def device_address(self) -> Optional[ADDRESS]: return self._device.mac_address @property def context(self) -> Scope: return self._device.context def get_parameter(self, name: str, raw_data: bool = False) -> Any: if name == self.DEVICE_NAME: return self._device.device_name elif name == self.DEVICE_ADDR: return address_str(self._device.mac_address) return super().get_parameter(name, raw_data=False) def keys(self, list_hide=False, list_private=False) -> List[str]: ret = super().keys(list_hide, list_private) ret.extend([self.DEVICE_NAME, self.DEVICE_ADDR]) return ret class DeviceScope(ReadonlyScope): """device constant property scope. ============== ============================== ======================================================= parameter type description ============== ============================== ======================================================= ID int device ID DEVICE_NAME str device name, same as name used in scan response info DEVICE_ADDRESS (int, int, int, int, int, int) device mac address ============== ============================== ======================================================= """ VARIABLE = ['ID', 'DEVICE_NAME', 'DEVICE_ADDRESS'] __slots__ = ('_device',) def __init__(self, device: CompletedDevice): super().__init__(self.VARIABLE) self._device = device def get_local(self, key: str) -> Any: if key == 'ID': return self._device.device_id elif key == 'DEVICE_NAME': return self._device.device_name elif key == 'DEVICE_ADDRESS': return self._device.mac_address raise KeyError(key) def __str__(self): return 'DeviceScope' class CompleteDeviceParameterScope(Scope): """device configuration scope.""" __slots__ = ('_device', '_table') def __init__(self, device: CompletedDevice, parameter: ParameterTable): """ :param device: device :param parameter: parameter table """ super().__init__() self._device = device self._table = parameter def locals(self) -> Iterable[str]: return self._table.keys() def contain_local(self, key: str) -> bool: """has device has parameter **key**?""" return key in self._table def get_local(self, key: str) -> Any: """get the value of the parameter **key** of this device configuration.""" return self._device.get_parameter(key, raw_data=False) def set_local(self, key: str, value: Any): """set the **value** of the parameter **key** of this device. Parameter **key** must exist in device configuration which mean that you can not new a parameter through this way. If parameter **key** not exist, raise *KeyError*. """ self._device.set_parameter(key, value) def del_local(self, key: str): """do nothing. you can not delete parameter from device configuration.""" pass def clear(self): """do nothing""" pass def reset(self): """reset all parameter in device configuration back to initial value.""" for info in self._table.values(): try: self._device.set_parameter(info.name, info.init_para()) except (PropertyDomainError, ActionDomainError): pass