""" Command line ------------ Usage :: env PYTHONPATH=RaspBerryPi3/python python -m biopro.devlib.main -L RaspBerryPi3/python/res/devlib/json/ ... """ import json as _json from pprint import pprint from shutil import copyfile from biopro.recording import RecordingMetaFile, RecordingData from biopro.util.cli import * from biopro.util.console import hex_table, hex_value from biopro.util.text import Table from .data import DataDecodeFormat from .encoder import * from .library import * # noinspection PyUnusedLocal class Main(CliMain): """command line for testing device library. """ def __init__(self): super().__init__() self._argument: List[str] = [] self.repository = DeviceLibraryRepository() self._library: Optional[DeviceLibrary] = None @cli_flags('-h', '--help', force_return=True) def _help(self, opt: str): """print help document""" self.print_help_usage('[OPTIONS]', 'COMMAND') self.print_help_usage('[OPTIONS]', 'LIBRARY[:VERSION]') self.print_help_usage('[OPTIONS]', 'LIBRARY[:VERSION]', 'COMMAND', 'ARGS...') self.print_help(print_usage=False) @cli_flags('-r') def _recursive(self, opt: str): """""" self.recursive = True @cli_options('-L', '--library', value='PATH') def _library_path(self, opt: str, value: str): """library search path""" self.repository.add_library_path(value) @cli_command('help') def _help_command(self, act, argv: List[str]): """print help""" if len(argv) == 0: self.print_help() else: # use wrong type of library # we expect sub-command do nothing but '-h', # so library would not be used. self._library = '' self.parsing([argv[0], '-h']) @cli_command('list') def _list_command(self, act, argv: List[str]): """list library in path""" return _MainListCommand(act, self) @cli_command('match') def _find_command(self, act, argv: List[str]): """find library in path by device response info""" return _MainMatchCommand(act, self) @cli_command('load') def _load_command(self, act, argv: List[str]): """load library with library file name""" return _MainLoadCommand @cli_command('match-rule') def _match_command(self, act, argv: List[str]): """list library matching rule""" if self._library is None: raise RuntimeError('cannot call this command without LIBRARY') return _MainSimpleLibraryCommand(act, self._library) @cli_command('parameter') def _para_command(self, act, argv: List[str]): """dump library parameter""" if self._library is None: raise RuntimeError('cannot call this command without LIBRARY') return _MainParameterCommand(act, self._library) @cli_command('encode') def _encode_command(self, act, argv: List[str]): """encode parameter""" if self._library is None: raise RuntimeError('cannot call this command without LIBRARY') return _MainEncodeCommand(act, self._library) @cli_command('decode') def _decode_command(self, act, argv: List[str]): """decode parameter""" return _MainDecodeCommand(act, self) @cli_command('instruction') def _instruction_command(self, act, argv: List[str]): """call instruction""" if self._library is None: raise RuntimeError('cannot call this command without LIBRARY') return _MainInstructionCommand(act, self._library) @cli_command('ins-eval') def _instruction_eval_command(self, act, argv: List[str]): """eval instruction""" return _MainInstructionEvalCommand @cli_command('compress') def _compress_command(self, act, argv: List[str]): """compress devlib""" return _MainCompressCommand @cli_command('data') def _data_command(self, act, argv: List[str]): """data decoder""" return _MainDataCommand @cli_arguments('*', left_all=True) def _argument_handle(self, pos: int, value: str): """""" self._argument.append(value) def run(self): if len(self._argument) == 0: self._help('') else: library_expr = self._argument[0] library_name, library_version = part_suffix(library_expr, ':') self._library = library = self.repository.get_library(library_name, library_version) if library is None: print('library ' + library_expr + ' not found') elif len(self._argument) == 1: # dump library content pprint(library.as_json(), width=120) else: # redirect command command = self._argument[1] args = self._argument[2:] self.invoke_subcommand(command, args) # noinspection PyUnusedLocal class _MainLoadCommand(CliSubCommandMain): def __init__(self, command: str): super().__init__(command) self.file_name = None self.print_file_name = False self.dump_library = False @cli_flags('-h', '--help', force_return=True) def _help(self, opt: str): """print help document""" self.print_help() @cli_arguments(0, value='NAME') def _file_name(self, pos: int, value: str): """library file name""" self.file_name = value @cli_flags('--dump') def _dump_library(self, flag: str): """dump library content after loading successfully""" self.dump_library = True @cli_flags('--path') def _print_file_name(self, flag: str): """print the file path of loaded library""" self.print_file_name = True def run(self): if self.file_name is None: raise RuntimeError('lost NAME') library = self._load_library() if self.dump_library: pprint(library.as_json(), width=120) def _load_library(self) -> DeviceLibrary: p = Path(self.file_name) if p.exists(): if self.print_file_name: print(str(p)) return DeviceLibrary.load(p) else: raise FileNotFoundError(str(p)) # noinspection PyUnusedLocal class _MainListCommand(CliSubCommandMain): def __init__(self, command: str, main: Main): super().__init__(command) self._main = main self.show_library_path = False self.show_all_result = False self._check_options = [] @cli_flags('-h', '--help', force_return=True) def _help(self, opt: str): """print help document""" self.print_help() @cli_flags('--show-path') def _show_library_path(self, opt: str): """print library file path""" self.show_library_path = True @cli_options('--check', value='ITEM,...') def _check_options(self, opt: str, value: str): """""" if ',' in value: self._check_options.extend(value.split(',')) else: self._check_options.append(value) @cli_flags('--list-all') def _list_all(self, opt: str): """""" self.show_all_result = True @cli_flags('--list-check-item') def _list_check_options(self, opt: str): """""" t = Table('ITEM', 'description') t.append('match', 'check match rule') t.set_format(1, align_right=False) t.print() self.force_return() def run(self): repository = self._main.repository repository.on_error = self._on_error repository.invalid_cache() all_lib = repository.get_library_all() if len(self._check_options) == 0: for lib in all_lib: if self.show_library_path: print(lib.name, lib.version, lib.file_path) else: print(lib.name, lib.version) else: for it in self._check_options: if it == 'match': self._check_match(all_lib) else: print('unknown check item : ' + it) def _on_error(self, lib_path: Path, th: BaseException): print('error', lib_path, ':', th) def _check_match(self, all_lib: List[DeviceLibrary]): d1: Dict[DeviceMatchRule, DeviceLibrary] = {} for lib in all_lib: match_rule = lib.match_rule if isinstance(match_rule, DeviceMatchRule): match_rule = [match_rule] for m in match_rule: d1[m] = lib x = [0, 0, 0, 0, 0, 0] for m in d1.keys(): e = m.major_product_number if e is not None: v = e.to_number if v is not None: x[0] = max(x[0], v) e = m.minor_product_number if e is not None: v = e.to_number if v is not None: x[1] = max(x[1], v) e = m.major_version_number if e is not None: v = e.to_number if v is not None: x[2] = max(x[2], v) e = m.minor_version_number if e is not None: v = e.to_number if v is not None: x[3] = max(x[3], v) e = m.serial_year_number if e is not None: v = e.to_number if v is not None: x[4] = max(x[4], v) e = m.serial_month_number if e is not None: v = e.to_number if v is not None: x[5] = max(x[5], v) for p1 in range(x[0] + 1): for p2 in range(x[1] + 1): for v1 in range(x[2] + 1): for v2 in range(x[3] + 1): for s1 in range(x[4] + 1): for s2 in range(x[5] + 1): serial = DeviceSerialNumber(p1, p2, v1, v2, s1, s2) response = DeviceResponseInfo(None, serial, EMPTY_ADDRESS) ds = list(filter(lambda it: it.is_device_match(response), d1.values())) if len(ds) > 1 or self.show_all_result: header = str(serial) for i, d in enumerate(ds): if self.show_library_path: p = d.file_path else: p = '' print(header if i == 0 else ' ' * len(header), d.name, d.version, p) # noinspection PyUnusedLocal class _MainMatchCommand(_MainListCommand): def __init__(self, command: str, main: Main): super().__init__(command, main) self.device_name = None self.major_product_number = None self.minor_product_number = None self.major_version_number = None self.minor_version_number = None self.serial_year_number = 0 self.serial_month_number = 0 self.show_response_info = False @cli_flags('-h', '--help', force_return=True) def _help(self, opt: str): """print help document""" self.print_help() @cli_flags('--show-resp') def _show_response_info(self, opt: str): """print response information""" self.show_response_info = True @cli_arguments(0, value='NAME') def _device_name(self, pos: int, value: str): """device name""" self.device_name = value @cli_arguments('*', value='VALUE') def _args(self, pos: int, value: str): """device serial numbers. required at least 4, at most 6 numbers""" if pos == 1: self.major_product_number = int(value) elif pos == 2: self.minor_product_number = int(value) elif pos == 3: self.major_version_number = int(value) elif pos == 4: self.minor_version_number = int(value) elif pos == 5: self.serial_year_number = int(value) elif pos == 6: self.serial_month_number = int(value) else: raise RuntimeError('unknown arguments : ' + value) def run(self): if self.device_name is None: raise RuntimeError('lost NAME') if self.major_product_number is None: raise RuntimeError('lost major product number') if self.minor_product_number is None: raise RuntimeError('lost minor product number') if self.major_version_number is None: raise RuntimeError('lost major version number') if self.minor_version_number is None: raise RuntimeError('lost minor version number') response = DeviceResponseInfo(self.device_name, DeviceSerialNumber(self.major_product_number, self.minor_product_number, self.major_version_number, self.minor_version_number, self.serial_year_number, self.serial_month_number), EMPTY_ADDRESS) if self.show_response_info: pprint(response.as_json(), indent=1) repository = self._main.repository for lib in repository.get_library_all(): if lib.is_device_match(response): if self.show_library_path: print(lib.name, lib.version, lib.file_path) else: print(lib.name, lib.version) break else: print('no library match') # noinspection PyUnusedLocal class _MainInstructionEvalCommand(CliSubCommandMain): def __init__(self, command: str): super().__init__(command) self.parameter: Dict[str, Any] = {} self.instruction: List[str] = [] @cli_flags('-h', '--help', force_return=True) def _help(self, opt: str): """print help document""" self.print_help() @cli_options('-P', value='KEY=VALUE') def _parameter(self, opt: str, value: str): """set parameter P value""" p, v = part_suffix(value, '=') self.parameter[p] = int(v) @cli_options('-f', value='FILE') def _file(self, opt: str, value: str): """use file""" with open(value) as f: for line in f: if line.startswith('#'): continue line = line.strip() if len(line) == 0: continue if '=' in line: self._parameter('', line) else: self.instruction.append(line) @cli_arguments('*', value='EXPR') def _instruction(self, pos: int, value: str): """instruction expression""" if pos == 0: # clean content loaded from file self.instruction.clear() self.instruction.append(value) def run(self): scope = ContextScope(self.parameter) for expression in self.instruction: try: instruction = InstructionContent.parse(expression) except RuntimeError as e: print(e) else: buffer = [] try: instruction.build_instruction(scope, buffer) except KeyError as e: print(e, 'not defined') else: print(hex_line(buffer)) # noinspection PyUnusedLocal class _MainSimpleLibraryCommand(CliSubCommandMain): def __init__(self, command: str, library: DeviceLibrary): super().__init__(command) self.library = library self.output_file = None @cli_flags('-h', '--help', force_return=True) def _help(self, opt: str): """print help document""" self.print_help() @cli_options('-o', '--output', value='FILE') def _output_file(self, opt: str, value: str): """output file path""" self.output_file = value def run(self): if self.command == 'match-rule': self._dump_match_rule() else: raise RuntimeError('unknown command : ' + self.command) def _dump_match_rule(self): lib = self.library for rule in lib.match_rule: np = rule.local_name_pattern p1 = rule.major_product_number p2 = rule.minor_product_number v1 = rule.major_version_number v2 = rule.minor_version_number s1 = rule.serial_year_number s2 = rule.serial_month_number np = np if np is not None else '.*' p1 = p1.as_json() p2 = p2.as_json() v1 = v1.as_json() v2 = v2.as_json() s1 = s1.as_json() if s1 is not None else '*' s2 = s2.as_json() if s2 is not None else '*' print(repr(np), ':', p1, '.', p2, '/', v1, '.', v2, '/', s1, '.', s2, sep='') # noinspection PyUnusedLocal class _MainEncodeCommand(_MainSimpleLibraryCommand): def __init__(self, command: str, library: DeviceLibrary): super().__init__(command, library) self.parameter_file = None self.encode_all_parameter = False @cli_arguments(0, value='[FILE]') def _parameter_file(self, pos: int, value: str): """load parameter file""" self.parameter_file = value @cli_flags('-A', '--all') def _all_parameter(self, opt: str): """encode all parameter""" self.encode_all_parameter = True def run(self): if self.parameter_file is None: raise RuntimeError('lost parameter file') # load configuration try: configuration = DeviceConfigurationDecoder.load(self.parameter_file) except IOError as e: raise RuntimeError('load parameter file : ' + self.parameter_file) from e # load library library = self.library # init parameter parameter = DeviceParameter(library) # overwrite configuration parameter.overwrite(configuration) # transform to configuration configuration = DeviceParameterConfiguration(library, parameter) # encode encoder = DeviceConfigurationEncoder(all_parameter=self.encode_all_parameter) data = encoder.encode(configuration) # output if self.output_file is None: hex_table(data) else: with open(self.output_file, 'bw') as f: f.write(data) # noinspection PyUnusedLocal class _MainDecodeCommand(CliSubCommandMain): def __init__(self, command: str, main: Main): super().__init__(command) self._main = main self.parameter_file = None self.use_library: Tuple[str, Optional[str]] = None @cli_flags('-h', '--help', force_return=True) def _help(self, opt: str): """print help document""" self.print_help() @cli_arguments(0, value='FILE') def _parameter_file(self, pos: int, value: str): """parameter json file or data file""" self.parameter_file = value @cli_options('-L', '--library', value='LIBRARY[:VERSION]') def _use_library(self, flag: str, value: str): """use library to fit configuration""" self.use_library = part_suffix(value, ':') def run(self): if self.parameter_file is None: raise RuntimeError('lost parameter file') # load parameter try: configuration = DeviceConfigurationDecoder.load(self.parameter_file) except IOError as e: raise RuntimeError('load parameter file : ' + self.parameter_file) from e if self.use_library is None: pprint(configuration.as_json(), width=120) else: # load library library_name, library_version = self.use_library library = self._main.repository.get_library(library_name, library_version) if library is None: raise RuntimeError('library ' + library_name + ' not found') # init configuration parameter = DeviceParameter(library) # overwrite configuration parameter.overwrite(configuration) # output configuration = DeviceParameterConfiguration(library, parameter) pprint(configuration.as_json(), width=120) # noinspection PyUnusedLocal class _MainDeviceCommand(_MainSimpleLibraryCommand): def __init__(self, command: str, library: DeviceLibrary): super().__init__(command, library) self.parameter = DeviceParameter(library) self._cache_device: CompletedDevice = None self._cache_context: Scope = None self._extend_parameter_options() def _extend_parameter_options(self): p = self.parameter m: Dict[str, str] = {} h: List[CliHandleOption] = [] for k in p.keys(): info = self.library.parameter_table[k] o = '--' + k.lower().replace('_', '-') m[o] = k h.append(CliHandleOption(o, help_doc=info.description)) if len(h) > 0: # noinspection PyShadowingNames def _callback(flag: str, value: Optional[str]): k = m[flag] info = self.library.parameter_table[k] if isinstance(info.domain, ParameterCollectionDomain): p.oper_parameter(k, value) else: p.set_parameter(k, int(value)) self.extend_options_callback(_callback, *h, help_section='options for ' + self.library.name) @cli_options('-p', '--parameter', value='FILE') def _parameter_file(self, opt: str, value: str): """load parameter json file or data file""" # load configuration try: configuration = DeviceConfigurationDecoder.load(value) except IOError as e: raise RuntimeError('load parameter file : ' + value) from e # overwrite configuration self.parameter.overwrite(configuration) def run(self): raise RuntimeError() def _new_device(self) -> CompletedDevice: if self._cache_device is None: # init device self._cache_device = DebugDevice(None, self.library, 0) self._cache_device.parameter.overwrite(self.parameter) return self._cache_device def _new_context(self) -> Scope: if self._cache_context is None: library = self.library device = self._new_device() # init context self._cache_context = library.constant.chain(DeviceParameterScope(device.parameter)) return self._cache_context # noinspection PyUnusedLocal class _MainParameterCommand(_MainDeviceCommand): def __init__(self, command: str, library: DeviceLibrary): super().__init__(command, library) self.only_recording_parameter = False self.raw_parameter_value = False self.generate_meta_file = False self.parameter_name = None self.parameter_value = None @cli_flags('-r') def _only_recording_parameter(self, flag: str): """only recording parameter""" self.only_recording_parameter = True if self.raw_parameter_value or self.generate_meta_file: raise RuntimeError('only one of -r, -R or -m') @cli_flags('-R') def _raw_parameter_value(self, flag: str): """use raw parameter value""" self.raw_parameter_value = True if self.only_recording_parameter or self.generate_meta_file: raise RuntimeError('only one of -r, -R or -m') @cli_flags('-m') def _generate_meta_file(self, flag: str): """generate meta file""" self.generate_meta_file = True if self.raw_parameter_value or self.only_recording_parameter: raise RuntimeError('only one of -r, -R or -m') @cli_arguments(0, '[PARAMETER]') def _parameter_name(self, pos: int, value: str): """get/set parameter""" self.parameter_name = value @cli_arguments(1, '[VALUE]') def _parameter_value(self, pos: int, value: str): """set parameter""" self.parameter_value = value def run(self): if self.parameter_name is None: self._run_no_arg() else: self._run_with_arg() def _run_no_arg(self): lib = self.library parameter = self.parameter if self.generate_meta_file: configuration = DeviceParameterConfiguration(lib, parameter) if self.output_file is None: output_file = self.library.name else: output_file = self.output_file meta = RecordingMetaFile(output_file) meta.configuration = configuration meta.write() if self.output_file is None: print('generate', meta.filepath) return elif self.raw_parameter_value: j = parameter.as_json() else: configuration = DeviceParameterConfiguration(lib, parameter) if self.only_recording_parameter: j = configuration.as_json() else: j = configuration.as_json(list_hide=True) if self.output_file is None: pprint(j, width=120) else: with open(self.output_file, 'w') as f: _json.dump(j, f, indent=2) def _run_with_arg(self): lib = self.library parameter = self.parameter device = self._new_device() parameter_value = device.get_parameter(self.parameter_name) if self.parameter_value is not None: device.set_parameter(self.parameter_name, self.parameter_value) parameter_value = device.get_parameter(self.parameter_name) # TODO option flag for overwrite parameter file print(parameter_value) # noinspection PyUnusedLocal class _MainInstructionCommand(_MainDeviceCommand): def __init__(self, command: str, library: DeviceLibrary): super().__init__(command, library) self.instruction = None self._check_options = [] @cli_arguments(0, value='INSTRUCTION') def _instruction(self, pos: int, value: str): """instruction call""" self.instruction = value @cli_options('--check', value='ITEM,...') def _check_options(self, opt: str, value: str): """""" if ',' in value: self._check_options.extend(value.split(',')) else: self._check_options.append(value) @cli_flags('--list-check-item') def _list_check_options(self, opt: str): """""" t = Table('ITEM', 'description') t.append('all', 'eval all instruction') t.set_format(1, align_right=False) t.print() self.force_return() def run(self): # load library library = self.library if len(self._check_options) == 0: if self.instruction is None: for instruction in library.instruction_table: print(instruction) else: library.instruction_table.invoke_instruction(self._receive, self._new_context(), self.instruction) else: for it in self._check_options: if it == 'all': self._check_eval_all() else: print('unknown check item : ' + it) def _check_eval_all(self): table = self.library.instruction_table context = self._new_context() for instruction in table: print() print(instruction) table.invoke_instruction(self._receive, context, instruction) @staticmethod def _receive(ins: SingleInstruction): if isinstance(ins, BuiltinInstruction): buf = [ ins.instruction_type & 0xF0, (ins.instruction_oper & 0xF0), ] print('%-32s%s' % (ins.name, hex_line(buf))) elif isinstance(ins, InternalInstruction): print('%-32s%s' % (ins.name, str(ins.instruction_parameter))) if ins.name == '_cdr': data_format: str = ins.instruction_parameter context = ContextScope() InternalInstruction.cdr_parse(context, data_format) for k in context.locals(): print(' %-30s%s' % (k, context[k])) else: cnt = ins.instruction_content buf = [ ins.instruction_type & 0xF0, (ins.instruction_oper & 0xF0) | (len(cnt) & 0x0F), *cnt ] print('%-32s%s' % (ins.name, hex_line(buf))) # noinspection PyUnusedLocal class _MainCompressCommand(CliSubCommandMain): def __init__(self, command: str): super().__init__(command) self.source_file_list: List[str] = [] self.make_directory_parent = False self.flag_target_first = False self.flag_overwrite = False @cli_flags('-h', '--help', force_return=True) def _help(self, opt: str): """print help document""" self.print_help() @cli_flags('-p', '--parent') def _parent(self, opt: str): """make parent of target directory if not exists""" self.make_directory_parent = True @cli_flags('-f', '--force') def _force(self, opt: str): """overwrite if target file exist""" self.flag_overwrite = True @cli_flags('-t') def _target_first(self, opt: str): """output directory""" self.flag_target_first = True @cli_arguments('+', value='FILE') def _file(self, pos: int, value: str): """""" self.source_file_list.append(value) def run(self): if len(self.source_file_list) == 0: if self._target_first: raise RuntimeError('lost source file') else: raise RuntimeError('lost target directory') if self.flag_target_first: target_directory = self.source_file_list[0] source_file = self.source_file_list[1:] else: target_directory = self.source_file_list[-1] source_file = self.source_file_list[:-1] target_directory = Path(target_directory) if not target_directory.exists(): target_directory.mkdir(parents=True) elif not target_directory.is_dir(): raise NotADirectoryError(str(target_directory)) for p in source_file: if p.endswith(DeviceLibrary.FILE_EXT[0]): p = Path(p) t = target_directory / p.name.replace(DeviceLibrary.FILE_EXT[0], DeviceLibrary.FILE_EXT[1]) if t.exists() and self.flag_overwrite: t.unlink() self.compress(p, t) elif p.endswith(DeviceLibrary.FILE_EXT[1]): # just move p = Path(p) t = (target_directory / p.name).with_suffix(DeviceLibrary.FILE_EXT[1]) if t.exists() and self.flag_overwrite: t.unlink() copyfile(p, t) else: raise RuntimeError('unknown file type : ' + p) @classmethod def compress(cls, source: Path, target: Path): if not source.exists(): raise FileNotFoundError(str(source)) if target.exists(): raise FileExistsError(str(target)) with source.open('rb') as _s, target.open('wb') as _t: _t.write(DeviceLibrary.FILE_HEADER) with gzip.GzipFile(fileobj=_t) as _g: r = _s.read(1024) while len(r): _g.write(r) r = _s.read(1024) # noinspection PyUnusedLocal class _MainDataCommand(CliSubCommandMain): def __init__(self, command: str): super().__init__(command) self.data_format: str = None self.data_content: Union[List[str], bytes] = [] @cli_flags('-h', '--help', force_return=True) def _help(self, opt: str): """print help document""" self.print_help() @cli_arguments(0, value='FORMAT') def _data_format(self, pos: int, value: str): """data decoder name""" self.data_format = value @cli_arguments('*', value='CONTENT') def _data_content(self, pos: int, value: str): """data content in hex""" self.data_content.append(value) def run(self): data_format = DataDecodeFormat.parse(self.data_format) if isinstance(self.data_content, list): data_content = bytes(map(hex_value, self.data_content)) else: data_content = self.data_content data = data_format.decode(data_content) if data is None: print('None') elif isinstance(data, bytes): print(hex_line(data)) else: data: RecordingData = data t = Table(column=3) t.append('time stamp', data.time_stamp, 'ms') t.append('sample rate', data.sample_rate, '1/s') t.append('time', 'channel', 'value') for s, c, v in data: t.append('%.2f' % s, str(c), str(v)) t.print() if __name__ == '__main__': Main().main()