import json as _json from pathlib import Path from typing import Any, Dict, Union, TypeVar from biopro.util.encoder import * from biopro.util.json import JsonSerialize from .library import DeviceConfiguration, SimpleDeviceConfiguration E = TypeVar('E', bound=AbstractEncoder) D = TypeVar('D', bound=AbstractDecoder) class DeviceConfigurationEncoder: """ **DeviceParameter format** :: struct { u8 parameter_count; struct { string_value parameter_name; u8 collection_type; u8 parameter_type; union { N parameter_value; collection_value; string_value; }; } parameter[parameter_count]; } struct string_value { u8 string_size; u8 string_char[string_size]; }; struct collection_value { u8 collection_size; union { N parameter_value[collection_size]; string_value; } }; **collection_type** fields = =========== T description = =========== L list S set 0 value = =========== **parameter_type** fields === ======================= T description === ======================= \\0 null, empty collection ? bool B 1 bytes uint b 1 bytes int H 2 bytes uint h 2 bytes int I 4 bytes uint i 4 bytes int d 8 bytes double floating p string === ======================= """ __slots__ = ('_all_parameter',) def __init__(self, all_parameter=False): self._all_parameter = all_parameter def encode(self, d: DeviceConfiguration) -> bytes: return self.write(d, BytesEncoder()).get() def dump(self, d: DeviceConfiguration, f: Union[str, Path]): if isinstance(f, str): f = Path(f) with f.open('wb') as _f: self._encode(d, FileEncoder(f)) def write(self, d: DeviceConfiguration, e: E) -> E: self._encode(d, e) return e def _encode(self, d: DeviceConfiguration, e: AbstractEncoder): if self._all_parameter: pi = d.keys(list_hide=True) else: pi = d.keys() e.u8(len(pi)) for name in pi: self._encode_parameter(name, d[name], e) @classmethod def _encode_parameter(cls, name: str, value: Optional[Any], e: AbstractEncoder): e.encode_string(name) if value is None: e.write(b'0\0') elif isinstance(value, list) or isinstance(value, tuple): sz = len(value) if sz == 0: e.write(b'L\0') else: parameter_type = cls._value_type(max(value)) for v in value: if isinstance(v, int) and v < 0: parameter_type = parameter_type.lower() e.write(b'L') e.write(parameter_type.encode()) e.u8(sz) for v in value: cls._encode_value(parameter_type, v, e) elif isinstance(value, set): sz = len(value) if sz == 0: e.write(b'S\0') else: parameter_type = cls._value_type(iter(value).__next__()) e.write(b'S') e.write(parameter_type.encode()) e.u8(sz) for v in value: cls._encode_value(parameter_type, v, e) else: parameter_type = cls._value_type(value) e.write(b'0') e.write(parameter_type.encode()) cls._encode_value(parameter_type, value, e) @classmethod def _encode_value(cls, parameter_type: str, value: Any, e: AbstractEncoder): if isinstance(value, bool): e.u8(1 if value else 0) elif isinstance(value, str) or isinstance(value, list): value = str(value) e.encode_string(value) elif isinstance(value, int): e.write(struct.pack('>' + parameter_type, value)) elif isinstance(value, float): e.write(struct.pack('>' + parameter_type, value)) else: raise RuntimeError('unknown parameter value : ' + value.__class__.__name__) @classmethod def _value_type(cls, value: Any) -> str: if isinstance(value, bool): return '?' elif isinstance(value, str) or isinstance(value, list): return 'p' elif isinstance(value, int): return cls._number_type(value, value) elif isinstance(value, float): return 'f' else: raise RuntimeError('unknown parameter value : ' + value.__class__.__name__) @classmethod def _number_type(cls, min_value: int, max_value: int) -> str: value = max(abs(min_value), abs(max_value)) if value < 0xFF: return 'b' if min_value < 0 else 'B' elif value < 0xFFFF: return 'h' if min_value < 0 else 'H' else: return 'i' if min_value < 0 else 'I' class DeviceConfigurationDecoder(JsonSerialize): __slots__ = ('parameter',) def __init__(self): self.parameter: Dict[str, Any] = {} @classmethod def load(cls, f: Union[str, Path]) -> DeviceConfiguration: """load device configuration from file. it could be json file or data file. :param f: file path :return: decoder :raises IOError: io error """ if isinstance(f, str): if f.endswith('.json'): with open(f) as _f: return SimpleDeviceConfiguration(_json.load(_f)) f = Path(f) with f.open('rb') as _f: return cls._decode(FileDecoder(f))._as_configuration() @classmethod def read(cls, f: Union[IO, bytes]) -> DeviceConfiguration: """decode device configuration :param f: file stream :return: decoder :raises IOError: io error """ if isinstance(f, bytes): return cls._decode(BytesDecoder(f))._as_configuration() else: return cls._decode(FileDecoder(f))._as_configuration() @classmethod def decode(cls, d: AbstractDecoder) -> DeviceConfiguration: """decode device configuration :param d: data content :return: decoder :raises IOError: io error """ return cls._decode(d)._as_configuration() def _as_configuration(self) -> DeviceConfiguration: return SimpleDeviceConfiguration(self.parameter) @classmethod def _decode(cls, d: AbstractDecoder) -> 'DeviceConfigurationDecoder': """decode device configuration :return: decoder :raises IOError: io error """ ret = DeviceConfigurationDecoder() parameter_count = d.u8() for p in range(parameter_count): parameter_name = d.decode_string() collection_type = d.read(1) parameter_type = d.read(1) if collection_type == b'0': parameter_value = cls._decode_value(parameter_type, d) elif collection_type in b'LS': parameter_value = [] if parameter_type != b'\0': collection_size = d.u8() for _ in range(collection_size): parameter_value.append(cls._decode_value(parameter_type, d)) if collection_type == b'S': parameter_value = set(parameter_value) else: raise IOError('illegal collection type for ' + str(p) + ' : ' + collection_type.decode()) ret.parameter[parameter_name] = parameter_value return ret @classmethod def _decode_value(cls, parameter_type: bytes, d: AbstractDecoder) -> Any: if parameter_type == 0: return None elif parameter_type == b'?': return d.u8() > 0 elif parameter_type == b'p': return d.decode_string() else: parameter_type = parameter_type.decode() parameter_format = '>' + parameter_type if parameter_type in 'Bb': parameter_size = 1 elif parameter_type in 'Hh': parameter_size = 2 elif parameter_type in 'Iif': parameter_size = 4 elif parameter_type == 'd': parameter_size = 8 else: raise ValueError('illegal parameter type : ' + parameter_type) return struct.unpack(parameter_format, d.read(parameter_size))[0] def as_json(self): return self.parameter