import re from random import randint from time import sleep from typing import Sequence, Tuple from biopro.util.bit import append_buffer, parse_buffer from biopro.util.console import hex_line from biopro.util.text import part_prefix, part_suffix from .device import Device, DeviceInstruction, DeviceCommonInstruction, MasterDevice from .expression import * class InstructionNotFound(RuntimeError): """error for instruction not found/support""" __slots__ = ('_instruction',) def __init__(self, instruction: str): super().__init__('instruction ' + instruction + ' not found') self._instruction = instruction @property def instruction(self) -> str: return self._instruction class DeviceInstructionError(RuntimeError): """device relate instruction runtime error""" __slots__ = ('_instruction', '_message') def __init__(self, instruction: Optional[str], message: str = None): if instruction is None: m = 'instruction unknown' elif len(instruction) == 0: m = 'instruction' else: m = 'instruction "' + instruction + '"' if message is not None: m = m + ' : ' + message else: m = m + ' : fail' super().__init__(m) self._instruction = instruction self._message = message @property def instruction(self) -> Optional[str]: return self._instruction @property def message(self) -> Optional[str]: return self._message class SendInstructionTimeoutError(DeviceInstructionError): """error when instruction sending/receiving timeout""" __slots__ = () def __init__(self, instruction: Optional[str], message: str = None): super().__init__(instruction, message) class SendInstructionNotConnectedError(DeviceInstructionError): """error when instruction sending/receiving""" __slots__ = () def __init__(self, instruction: Optional[str]): super().__init__(instruction) class BadInstructionContentError(DeviceInstructionError): """error when building instruction""" __slots__ = () def __init__(self, instruction: str, message: str = None): super().__init__(instruction, message) class UnknownInstructionError(DeviceInstructionError): """unknown instruction""" __slots__ = () def __init__(self, ins_type: int, ins_oper: int, instruction: Sequence[int]): super().__init__(None, 'header[%d]op[%d]value[%s]' % (ins_type, ins_oper, hex_line(instruction))) class Instruction(JsonSerialize, metaclass=abc.ABCMeta): """top level abstract instruction **json format** :: Instruction = ListInstruction # type: list | SendInstruction # type: dict Instruction = { "export": bool # does this instruction can be seen by public } **python class graph** :: Instruction |_ ListInstruction |_ SingleInstruction |_ SendInstruction |_ ResolvedSendInstruction |_ BuiltinInstruction |_ InternalInstruction """ __slots__ = ('_name', 'export') def __init__(self, name: str, export: bool = False): """ :param name: instruction name :param export: instruction public for outside """ self._name = name self.export = export @property def name(self) -> str: """instruction name""" return self._name @abc.abstractmethod def eval_instruction(self, context: Scope) -> Iterable['SingleInstruction']: pass def __str__(self): return self._name def __repr__(self): return self._name class SingleInstruction(Instruction, metaclass=abc.ABCMeta): __slots__ = ('_ins_type', '_ins_oper') def __init__(self, name: str, ins_type: int, ins_oper: int, export: bool = False): """ :param name: instruction name :param ins_type: instruction type :param ins_oper: instruction operator :param export: instruction public for outside """ DeviceInstruction.valid_ins_type(ins_type) super().__init__(name, export) self._ins_type = ins_type self._ins_oper = ins_oper @property def instruction_type(self) -> int: return self._ins_type @property def instruction_oper(self) -> int: return self._ins_oper @property @abc.abstractmethod def instruction_content(self) -> List[int]: pass def eval_instruction(self, context: Scope) -> Iterable['SingleInstruction']: yield self class BuiltinInstruction(SingleInstruction): """built-in instruction""" __slots__ = ('_ins_type',) def __init__(self, name: str): try: ins_oper = getattr(DeviceInstruction, name) except AttributeError as e: raise InstructionNotFound(name) from e if name.startswith('RIS_'): ins_type = DeviceInstruction.TYP_RIS elif name.startswith('VIS_'): ins_type = DeviceInstruction.TYP_VIS elif name.startswith('CIS_'): ins_type = DeviceInstruction.TYP_CIS else: raise InstructionNotFound(name) super().__init__(name, ins_type, ins_oper, True) @property def instruction_content(self) -> List[int]: return [] def as_json(self) -> JSON: return self._name class InternalInstruction(SingleInstruction): """Interface of internal Instruction scope. all internal instruction are starts with '_' **predefined** ================= ===================== ================================================ function name parameter type description ================= ===================== ================================================ _sleep (int) -> None sleep millisecond _notify (bool) -> None enable/disable notify _data_format (str) -> None set data format _data_format_cali (str, bytes?) -> None set data format with cali_coeff table. _sync (bool) -> None start/stop sync _log (str) -> None log message _set (str, int) -> None set parameter value _cdr (str)-> None get cdr content _disable_cache (bool=True) -> None disable cache manager ================= ===================== ================================================ _sleep(millisecond) sleep certain millisecond _notify(enable) enable/disable device notify (characteristic 4) _data_format(format) set notify data format. it defined in :class:`biopro.devlib.data.DataDecodeFormat` _data_format_cali(format, cali=None) set notify data format with cali_coeff table. it defined in :class:`biopro.devlib.data.DataDecodeFormat` _sync(start) start/stop data server to fetch data _log(message) log message _set(parameter_name, p_value) set parameter P value _cdr(parsing_format) get cdr content and fit into context scope according parsing_format _disable_cache(disable=True) disable cache manager. All living data stream are load from file. """ PREDEFINED_SLEEP = '_sleep' PREDEFINED_NOTIFY = '_notify' PREDEFINED_DATA_FORMAT = '_data_format' PREDEFINED_DATA_FORMAT_CALI = '_data_format_cali' PREDEFINED_SYNC = '_sync' PREDEFINED_LOG = '_log' PREDEFINED_SET = '_set' PREDEFINED_CDR = '_cdr' PREDEFINED_DISABLE_CACHE = '_disable_cache' PREDEFINED = ( PREDEFINED_SLEEP, PREDEFINED_NOTIFY, PREDEFINED_DATA_FORMAT, PREDEFINED_DATA_FORMAT_CALI, PREDEFINED_SYNC, PREDEFINED_LOG, PREDEFINED_SET, PREDEFINED_CDR, PREDEFINED_DISABLE_CACHE, ) __slots__ = ('_expr', '_para') def __init__(self, expr: str): if not expr.startswith('_'): raise InstructionNotFound(expr) i = expr.find('(') if i < 0 or not expr.endswith(')'): raise InstructionNotFound(expr) super().__init__(expr[:i], DeviceInstruction.TYP_IIS, 0) self._expr = expr self._para = eval(expr[i:]) @property def instruction_content(self) -> List[int]: return [] @property def instruction_parameter(self) -> Any: return self._para def as_json(self) -> JSON: return self._expr def handle_internal_instruction(self, master: MasterDevice, device: Device, context: Scope) -> bool: consume = master.handle_internal_instruction(device, self.name, self._para) if not consume: if self.name == self.PREDEFINED_SLEEP: self.sleep(self._para) return True elif self.name == self.PREDEFINED_CDR: self.cdr(master, device, context, self._para) return True return consume @classmethod def sleep(cls, ms: int): sleep(ms / 1000) @classmethod def cdr(cls, master: MasterDevice, device: Device, context: Scope, data_format: str): master.log_verbose('device', device.device_id, cls.PREDEFINED_CDR, data_format) data = device.read_command_return_data() # print('cdr!!!!~') # print(data) cls.cdr_parse(context, data_format, data) @classmethod def cdr_parse(cls, context: Scope, data_format: str, data: List[int] = None): parser = InstructionContent.parse(data_format) if data is None: class ListLike: def __getitem__(self, item): return 0 data = ListLike() else: data = [v for v in data] parser.parse_instruction(context, data) class ListInstruction(Instruction, ImmutableListNode[Union[Instruction, Expression[str]]]): """Instruction group, allow the name of the instruction or a expression it. If instruction name is ``None``, ignore it. **json format** :: ListInstruction = [list_instruction_element] | { "export": bool "instructions" : [list_instruction_element] } list_instruction_element = Instruction | Expression[str] """ __slots__ = () def __init__(self, name: str, ins: Iterable[Union[str, Expression[str]]], export: bool = False): Instruction.__init__(self, name, export) ImmutableListNode.__init__(self, list(ins)) def eval_instruction(self, context: Scope) -> Iterable['SingleInstruction']: for ins in self: # foreach instructions # ins type could be Union[str, Expression] if isinstance(ins, Expression): ins = ins.value(context) # ins type could be Union[None, str, Instruction] if ins is None: continue elif isinstance(ins, str): if ins.startswith('_') and '(' in ins and ins.endswith(')'): # internal command yield InternalInstruction(ins) continue elif hasattr(DeviceInstruction, ins): yield BuiltinInstruction(ins) continue # instruction try: ins = context[ins] except KeyError: pass if isinstance(ins, Instruction): yield from ins.eval_instruction(context) else: raise InstructionNotFound(str(ins)) def as_json(self): instruction = [] for i in self: if isinstance(i, str): instruction.append(i) elif isinstance(i, ComplexExpression): instruction.append(i.as_json()) else: raise RuntimeError('unknown instruction : ' + str(i)) return { "export": self.export, "instructions": instruction } class InstructionContentWidth: """instruction width. **json format** :: InstructionContentWidth :str = SIZE TYPE? ENDIAN? SIGN? base = BASE ")" SIGN = "+" **SIZE** bit width or byte length, default 1 '*' for array **TYPE** ==== ================================== TYPE description ==== ================================== b bits B bytes 2-base value o, O bytes 8-base value d, D bytes 10-base value x, X bytes 16-base value ==== ================================== **ENDIAN** ====== ================================== ENDIAN description ====== ================================== > big endian (default) < little endian ====== ================================== **BASE** value base, 2 for bit default, 16 for byte default. """ __slots__ = ('_size', '_bytes_type', '_signed', '_little_endian') BASE = { 'b': 2, 'B': 2, 'o': 8, 'O': 8, 'd': 10, 'D': 10, 'x': 16, 'X': 16, } def __init__(self, size: Optional[int], bytes_type: str, signed_value=False, little_endian=False): self._size = size self._bytes_type = bytes_type self._signed = signed_value self._little_endian = little_endian @property def size(self) -> Optional[int]: return self._size @property def little_endian(self) -> bool: return self._little_endian @property def signed(self) -> bool: # XXX InstructionContentWidth.signed not used return self._signed @property def bytes_unit(self) -> bool: return self._bytes_type in 'BODX' @property def is_array(self) -> bool: # XXX InstructionContentWidth.is_array not used return self._size is None @property def width(self) -> int: if self._bytes_type in 'BODX': return self._size * 8 else: return self._size @property def value_base(self) -> int: return self.BASE[self._bytes_type] def __str__(self): return '%s%s%s%s' % (str(self._size) if self._size is not None else '*', self._bytes_type, '<' if self._little_endian else '>', '+' if self._signed else '') PATTERN = re.compile(r'(?P(\*|\d+))?(?P[XxDdOoBb])(?P[<>])?(?P\+)?(?P.*)') @classmethod def parse(cls, expr: str) -> Tuple['InstructionContentWidth', str]: m = cls.PATTERN.match(expr) if m is None: raise RuntimeError('illegal instruction content width : ' + expr) z = m.group('SIZE') t = m.group('TYPE') e = m.group('ENDIAN') s = m.group('SIGN') x = m.group('EXPR') if z is None: z = 1 elif z == '*': z = None else: z = int(z) return InstructionContentWidth(z, t, signed_value=s is not None, little_endian=e == '<'), x class InstructionContent(JsonSerialize): """ **json format** :: InstructionContent = "#" ... # comment | InstructionContent "#" ... # instruction with comment | InstructionDataContent | InstructionConditionContent | CompactInstructionContent """ __slots__ = ('_comment',) def __init__(self, comment: Optional[str] = None): self._comment: Optional[str] = comment def build_instruction(self, context: Scope, buffer: List[int], shift: int = 0) -> int: """ build instruction :param context: general case root :param buffer: instruction buffer :param shift: data offset :return: next data offset """ return shift def parse_instruction(self, context: Scope, buffer: List[int], offset: int = 0, shift: int = 0) -> Tuple[int, int]: """parse instruction, extra parameter from instruction. :param context: :param buffer: instruction buffer :param offset: buffer byte read :param shift: buffer but read """ return offset, shift def as_json(self): return str(self) def __str__(self): if self._comment is None: return '#' else: return '#' + self._comment @classmethod def parse(cls, expr: str) -> 'InstructionContent': origin_expr = expr if expr == '#': return InstructionContent() elif expr.startswith('#'): return InstructionContent(comment=expr[1:]) elif ';' in expr: return CompactInstructionContent([cls.parse(v) for v in expr.split(';') if len(v)]) else: ins_argv = {} expr, comment = part_suffix(expr, '#') expr, condition = part_suffix(expr, '?') if condition is not None: ins_type = InstructionConditionContent if len(condition) == 0: ins_argv['if_true'] = None ins_argv['if_false'] = None else: t, f = part_prefix(condition, ':') if t is None: raise RuntimeError('illegal condition instruction : ' + origin_expr) elif len(t) == 0: t = None ins_argv['if_true'] = t ins_argv['if_false'] = f else: ins_type = InstructionDataContent width, expr = InstructionContentWidth.parse(expr) return ins_type(width, expr, **ins_argv, comment=comment) class InstructionDataContent(InstructionContent): """ **json format** :: InstructionDataContent = InstructionContentWidth instruction_content instruction_content = int # binary number | parameter_name :str | parameter_name[index:int] # array parameter access """ __slots__ = ('_width', '_value', '_index') def __init__(self, width: InstructionContentWidth, value: Union[int, str], comment: Optional[str] = None): super().__init__(comment) self._width: InstructionContentWidth = width if isinstance(value, str) and '[' in value and value.endswith(']'): value, index = part_suffix(value[:-1], '[') self._value: Union[int, str] = value self._index: str = index else: self._value: Union[int, str] = value self._index: str = None def value(self, context: Scope) -> Union[int, List[int]]: value = self._value if isinstance(value, int): return value try: return int(value, self._width.value_base) except ValueError: ret = context[value] index = self._index if index is None: return ret else: try: index = int(index) except ValueError: index = int(context[index]) return ret[index] def build_instruction(self, context: Scope, buffer: List[int], shift: int = 0) -> int: if self._width.is_array: raise NotImplementedError() if self.value(context) == 'true': value = 1 elif self.value(context) == 'false': value = 0 else: value = int(self.value(context)) if self._width.bytes_unit: if self._width.size == 1: buffer.append(value) else: tmp = [] for _ in range(self._width.size): tmp.append(value & 0xFF) value >>= 8 if self._width.little_endian: buffer.extend(tmp) else: buffer.extend(tmp[::-1]) return 8 else: return append_buffer(buffer, shift, self._width.width, value, little_endian=self._width.little_endian) def parse_instruction(self, context: Scope, buffer: List[int], offset: int = 0, shift: int = 0) -> Tuple[int, int]: if self._width.is_array: raise NotImplementedError() try: self.value(ConstantScope.EMPTY) except KeyError: # is parameter if self._width.bytes_unit: if shift != 0: offset += 1 value = 0 if self._width.size == 1: value = buffer[offset] offset += 1 else: for i in range(self._width.size): if self._width.little_endian: value |= (buffer[offset] << (1 * i)) else: value = (value << 8) | buffer[offset] offset += 1 if len(self._value) > 0: context[self._value] = value return offset, 0 else: value, offset, shift = parse_buffer(buffer, offset, shift, self._width.width, little_endian=self._width.little_endian) if len(self._value) > 0: context[self._value] = value return offset, shift else: # is constant value if self._width.bytes_unit: if shift != 0: offset += 1 return offset + self._width.size, 0 else: shift += self._width.width return offset + (shift // 8), shift % 8 def __str__(self): if isinstance(self._value, int): value = bin(self._value)[2:] else: if self._index is not None: value = '%s[%s]' % (self._value, str(self._index)) else: value = self._value if self._comment is None: return '%s%s' % (self._width, value) else: return '%s%s#%s' % (self._width, value, self._comment) class InstructionConditionContent(InstructionDataContent): """ **json format** :: InstructionConditionContent = InstructionContentWidth parameter_name "?" | InstructionContentWidth parameter_name "?:" if_false | InstructionContentWidth parameter_name "?" if_true ":" if_false if_true, if_false = int | parameter_name :str """ __slots__ = ('_true', '_false') def __init__(self, width: InstructionContentWidth, value: str, if_true: Union[None, int, str] = None, if_false: Union[None, int, str] = None, comment: Optional[str] = None): if if_true is not None and if_false is None: raise RuntimeError('lost if_true') super().__init__(width, value, comment) self._true = if_true self._false = if_false def value(self, context: Scope) -> Union[int, List[int]]: value = context[self._value] test = value is not None if self._true is None and self._false is None: return 1 if test else 0 elif self._true is None: assert self._false is not None if test: return value else: branch = self._false else: assert self._true is not None assert self._false is not None branch = self._true if test else self._false if isinstance(branch, int): return branch try: return int(branch, self._width.value_base) except ValueError: return context[branch] def __str__(self): if self._comment is None: c = '' else: c = '#' + self._comment if self._true is None and self._false is None: return "%s%s?%s" % (self._width, self._value, c) elif self._true is None: return "%s%s?:%s%s" % (self._width, self._value, self._false, c) else: return "%s%s?%s:%s%s" % (self._width, self._value, self._true, self._false, c) class CompactInstructionContent(InstructionContent): """ **json format** :: CompactInstructionContent = InstructionContent ";" InstructionContent ... """ __slots__ = ('_instruction',) def __init__(self, instruction: List[InstructionContent]): super().__init__() self._instruction = instruction def build_instruction(self, context: Scope, buffer: List[int], shift: int = 0) -> int: for instruction in self._instruction: shift = instruction.build_instruction(context, buffer, shift) return shift def parse_instruction(self, context: Scope, buffer: List[int], offset: int = 0, shift: int = 0) -> Tuple[int, int]: for instruction in self._instruction: o, s = instruction.parse_instruction(context, buffer, offset, shift) offset += o shift += s return offset, shift def __str__(self) -> str: return ';'.join(map(str, self._instruction)) class SendInstructionForeachParameter(JsonSerialize): """ **json format** :: SendInstructionPositionParameter = { # either parameter or expression "parameter" = parameter_name "expression" :Expression[Iterator[int]] "variable" : [str] # variable name "default :int? = None # default value } """ __slots__ = ('_parameter', '_expression', '_variable', '_default') def __init__(self, variable: List[str], parameter: Optional[str] = None, expression: Optional[Expression] = None, default: Optional[int] = None): if (parameter is None and expression is None) or (parameter is not None and expression is not None): raise RuntimeError('foreach parameter required either parameter or expression') self._parameter: Optional[str] = parameter self._expression: Optional[Expression] = expression self._variable = variable self._default = default def for_scope(self, context: Scope) -> Iterable[Scope]: if self._parameter is not None: it = list(sorted(context[self._parameter])) elif self._expression is not None: it = list(self._expression.value(context)) else: assert False, "unreachable" for i in range(0, len(it), len(self._variable)): local = {} for j in range(len(self._variable)): try: local[self._variable[j]] = it[i + j] except IndexError: local[self._variable[j]] = self._default yield context.child(**local) def as_json(self): return { "parameter": self._parameter, "variable": self._variable, "default": self._default } class SendInstructionKeywordParameter(JsonSerialize): """ **json format** :: SendInstructionKeywordParameter = { parameter_name : Expression[int] } """ __slots__ = ('_parameter',) def __init__(self, parameter: Dict[str, Expression[int]]): self._parameter = parameter def set_scope(self, context: Scope) -> Scope: scope = context.child() for k, v in self._parameter.items(): value = v.value(context) if isinstance(value, str): value = context[value] scope.set_local(k, value) return scope def as_json(self): return {k: v.as_json() for k, v in self._parameter.items()} class SendInstruction(SingleInstruction): """ **json format** :: SendInstruction = { "type" :str = instruction_type "operator"? :int "export": bool "guard" :ListGuardExpression "foreach-parameter"? :SendInstructionForeachParameter "parameter"? :{variable_name = Expression} "data" = [InstructionContent] } instruction_type = "RIS" | "VIS" | "CIS" """ __slots__ = ('_table', '_parameter', '_foreach_parameter', '_guard') def __init__(self, name: str, ins_type: int, ins_oper: int, table: List[InstructionContent], parameter: Optional[SendInstructionKeywordParameter] = None, foreach_parameter: Optional[SendInstructionForeachParameter] = None, guard: Optional[ListGuardExpression] = None, export: bool = False): super().__init__(name, ins_type, ins_oper, export=export) self._table: List[InstructionContent] = table self._parameter = parameter self._foreach_parameter = foreach_parameter self._guard = guard @property def instruction_content(self) -> List[int]: raise RuntimeError() def eval_instruction(self, context: Scope) -> Iterable['SingleInstruction']: if self._guard is not None: self._guard.value(context) context = context.chain(SendInstructionScope(self)) if self._parameter is not None: context = self._parameter.set_scope(context) if self._foreach_parameter is None: yield ResolvedSendInstruction(self, self._build_instruction(context)) else: for scope in self._foreach_parameter.for_scope(context): yield ResolvedSendInstruction(self, self._build_instruction(scope)) def _build_instruction(self, context: Scope) -> List[int]: buffer = [] shift = 0 try: for instruction in self._table: shift = instruction.build_instruction(context, buffer, shift) except BaseException as e: raise BadInstructionContentError(self.name, str(e)) from e else: return buffer def as_json(self): ret = { 'type': { DeviceInstruction.TYP_RIS: 'RIS', DeviceInstruction.TYP_VIS: 'VIS', DeviceInstruction.TYP_CIS: 'CIS', }[self._ins_type], 'operator': self._ins_oper, 'data': [JsonSerialize.to_json(v) for v in self._table] } if self._parameter is not None: ret['parameter'] = self._parameter.as_json() if self._foreach_parameter is not None: ret['foreach-parameter'] = self._foreach_parameter.as_json() if self._guard is not None: ret['guard'] = self._guard.as_json() return ret class ResolvedSendInstruction(SingleInstruction): """resolve the `SendInstruction`_ instruction content """ __slots__ = '_content', def __init__(self, ins_send: SendInstruction, ins_content: List[int]): super().__init__(ins_send._name, ins_send._ins_type, ins_send._ins_oper) self._content = ins_content @property def instruction_content(self) -> List[int]: return self._content def as_json(self) -> JSON: return None class SendInstructionScope(ReadonlyScope): """Instruction readonly property scope. ============== ==== ======================================================= parameter type description ============== ==== ======================================================= RANDOM int generate a random number, range from 0 to 32bit max ============== ==== ======================================================= """ VARIABLE = ('RANDOM',) __slots__ = ('_instruction',) def __init__(self, instruction: SendInstruction): super().__init__(self.VARIABLE) self._instruction = instruction def get_local(self, key: str) -> Any: if key == 'RANDOM': return randint(0, 0xFFFF_FFFF) else: raise KeyError(key) def __str__(self): return 'SendInstructionScope' class InstructionTable(DictNode['Instruction']): """ **json format** :: InstructionTable = { instruction_name = Instruction } **common instruction** ========= ========================================================= name description ========= ========================================================= reset device reset start start recording/stimulation stop stop recording/stimulation interrupt pause recording/stimulation, sometime same as stop flush flush. often used to indicate the device still connected close shutdown device ========= ========================================================= **variable scope search chain** * used when instruction guarding 1. general case root (:class:`biopro.devlib.library.DeviceLibrary`) * used when instruction building 1. general case root (:class:`biopro.devlib.library.DeviceLibrary`) #. internal instruction (:class:`biopro.devlib.instruction.InternalInstructionScope`) #. instruction. (:class:`biopro.devlib.instruction.InstructionTableScope`) #. send instruction read-only property. defined in :class:`biopro.devlib.instruction.SendInstructionScope` #. "parameter" defined in :class:`biopro.devlib.instruction.SendInstruction` #. "foreach-parameter" defined in :class:`biopro.devlib.instruction.SendInstruction` """ __slots__ = () def __init__(self, instructions: Optional[Dict[str, Instruction]] = None): super().__init__(instructions) def __missing__(self, key: str) -> Instruction: """Try found instruction in :class:`DeviceInstruction` with instruction name match to its member, and then return :class:`BuiltinInstruction`. If fail, try found instruction in *COMMON_INSTRUCTION* and then return :class:`ListInstruction`. """ if hasattr(DeviceInstruction, key): return BuiltinInstruction(key) ins = DeviceCommonInstruction(key) if ins is not None: return ListInstruction(key, ins) raise KeyError(key) def export_keys(self) -> List[str]: return list(map(lambda ins: ins.name, filter(lambda ins: ins.export, self.values()))) def call_instruction(self, master: MasterDevice, device: Device, context: Scope, instruction: str): """call instruction. chain context with :class:`DeviceInternalInstructionScope` :param master: master device :param device: device :param context: scope :param instruction: instruction name :raises InstructionNotFound: If instruction not found. """ try: ins = self[instruction] except KeyError as e: raise InstructionNotFound(instruction) from e else: context = context.chain(InstructionTableScope(self)) for single in ins.eval_instruction(context): if isinstance(single, InternalInstruction): if not single.handle_internal_instruction(master, device, context): raise InstructionNotFound(single.name) else: device.send_instruction(single.instruction_type, single.instruction_oper, *single.instruction_content) # print('call_instruction~~~~~~~~~~') # print('type',single.instruction_type) # print('oper',single.instruction_oper) # print('content',*single.instruction_content) def invoke_instruction(self, receiver: Callable[[SingleInstruction], None], context: Scope, instruction: str): """invoke instruction. chain context with :class:`InstructionTableScope` :param receiver: type (SingleInstruction) -> None :param context: scope :param instruction: instruction name :raises InstructionNotFound: If instruction not found. """ try: ins = self[instruction] except KeyError as e: raise InstructionNotFound(instruction) from e else: context = context.chain(InstructionTableScope(self)) for single in ins.eval_instruction(context): receiver(single) class InstructionTableScope(ReadonlyScope): """the scope for loop reference back to instruction table""" __slots__ = ('_table',) def __init__(self, table: InstructionTable): # all instruction variable = list(table.keys()) # all VIS instruction variable.extend(filter(lambda it: it.startswith('VIS_'), dir(DeviceInstruction))) # command instruction variable.extend(map(lambda it: getattr(DeviceCommonInstruction, it), filter(lambda it: not it.startswith('__'), dir(DeviceCommonInstruction)))) super().__init__(variable) self._table = table def get_local(self, key: str) -> Any: return self._table[key] def __str__(self): return 'InstructionTableScope'