Roy
1556 lines
46 KiB
Python
1556 lines
46 KiB
Python
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
|
|
"""
|
|
# parse operator return CollectValueOperator instance
|
|
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:
|
|
# replace target with new oper
|
|
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', '_status', '_occupied_by_project')
|
|
|
|
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
|
|
"""
|
|
# DeviceManager
|
|
self._master = master
|
|
self._device = device
|
|
self._device_id = device_id
|
|
self._library = library
|
|
|
|
# device status ( 0: idle, 1: working, 2: error, 3: power off)
|
|
self._status = 0
|
|
# device be occupied by project
|
|
self._occupied_by_project = None
|
|
|
|
# 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 memory_board(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 mac_address_in_str(self) -> str:
|
|
return ':'.join('{:02x}'.format(b) for b in self._device.mac_address).upper()
|
|
|
|
@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
|
|
|
|
@property
|
|
def status(self) -> int:
|
|
return self._status
|
|
|
|
@status.setter
|
|
def status(self, new_status):
|
|
self._status = new_status
|
|
|
|
@property
|
|
def occupied_by_project(self) -> str:
|
|
return self._occupied_by_project
|
|
|
|
@occupied_by_project.setter
|
|
def occupied_by_project(self, new_occupied_by_project):
|
|
self._occupied_by_project = new_occupied_by_project
|
|
|
|
# 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):
|
|
if len(parameter) == 0:
|
|
return
|
|
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:
|
|
if name == 'MODE':
|
|
if self.library_name.startswith('Elite_EDC') or 'EIS' in self.library_name:
|
|
self.set_parameter('CTRL_HIGH_Z_15', self.get_parameter('HIGHZ_TABLE')[value])
|
|
self.set_parameter('TIME_DURATION', 0)
|
|
|
|
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:
|
|
if self._status == 1:
|
|
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 {
|
|
'id': self.device_id,
|
|
'name': self.device_name,
|
|
'status': self._status,
|
|
'device_id': self.device_id,
|
|
'device_name': self.device_name,
|
|
'device_address': list(self.mac_address),
|
|
'device_status': self._status,
|
|
'mac_address': self.mac_address_in_str,
|
|
'memory_board': self.memory_board,
|
|
'serial_number': self.serial_number.as_json(),
|
|
'version': self.device_version,
|
|
'battery': self.battery,
|
|
'parent': self.parent,
|
|
'recording_file_name': self.recording_file_name,
|
|
'library_name': self._library.name,
|
|
'library_version': str(self._library.version),
|
|
'occupied_by_project': self._occupied_by_project,
|
|
'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
|
|
|
|
@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
|