controller-wisetopdataserver/python/biopro/message/encoder.py

import struct
from typing import List, Optional

from biopro.util.console import hex_str
from .data import DataPresentMode
from .message import DataMessage


class DataMessageDecodeEntry:
    __slots__ = 'device', 'channel', 'time_stamp', 'data_width', 'present_mode', 'sample_rate', 'data_entry'

    def __init__(self,
                 device: int = 0,
                 channel: int = 0,
                 time_stamp: int = 0,
                 data_width: int = 2,
                 present_mode: DataPresentMode = DataPresentMode.FIRST,
                 sample_rate: int = 1,
                 data_entry: List[Optional[int]] = None):
        self.device = int(device)
        self.channel = int(channel)
        self.time_stamp = int(time_stamp)
        self.data_width = int(data_width)
        self.present_mode: DataPresentMode = present_mode
        self.sample_rate = int(sample_rate)
        self.data_entry: List[Optional[int]] = list(data_entry) if data_entry is not None else []

    def __hash__(self) -> int:
        ret = 23
        ret = ret * 17 + hash(self.device)
        ret = ret * 17 + hash(self.channel)
        ret = ret * 17 + hash(self.data_width)
        ret = ret * 17 + hash(self.time_stamp)
        ret = ret * 17 + hash(self.sample_rate)
        ret = ret * 17 + hash(self.data_entry)
        return ret

    def __eq__(self, o: object) -> bool:
        if o is self:
            return True

        if o is None:
            return False

        if not isinstance(o, DataMessageDecodeEntry):
            return False

        return (
            self.device == o.device and
            self.channel == o.channel and
            self.time_stamp == o.time_stamp and
            self.data_width == o.data_width and
            self.sample_rate == o.sample_rate and
            self.data_entry == o.data_entry
        )

    def __repr__(self) -> str:
        return "%dc%dt%d+%d:%s" % (self.device, self.channel, self.time_stamp, self.sample_rate, repr(self.data_entry))


class DataMessageDecoder:
    """message decoder.

    for format detail please see :class:`DataMessage`

    """

    __slots__ = 'message_size', 'header', 'message', 'data_entry'

    def __init__(self):
        self.message_size = None
        '''byte use in this message'''

        self.header = None
        '''message header.'''

        self.message = None
        '''message content for header in (0, 1, 2)'''

        self.data_entry: List[DataMessageDecodeEntry] = []
        '''data message content for header 3'''

    @classmethod
    def decode(cls, data: bytes) -> 'DataMessageDecoder':
        """decode the message

        :param data: message content
        """
        ret = DataMessageDecoder()
        ret.message_size = len(data)
        ret.header = header = data[0]

        if header == DataMessage.HEADER_SYSTEM:
            ret._decode_system(data)

        else:
            ret._decode_data(data)

        return ret

    def _decode_system(self, data: bytes):
        """decode message for header 0xFF (system message)"""
        length = data[1]
        self.message = hex_str(data[2:2 + length])

    def _decode_data(self, data: bytes):
        """decode message

        :param data: data message
        :return: read offset
        """

        offset = 0

        while offset < len(data):
            header = data[offset]
            offset += 1

            if header == 0xFF:
                raise ValueError('not data header')

            device = data[offset]
            offset += 1

            sample_rate = struct.unpack('<H', data[offset:offset + 2])[0]
            offset += 2

            channel_count = data[offset]
            offset += 1

            try:
                for _ in range(channel_count):
                    offset = self._decode_data_entry(data, offset, device, sample_rate)
            except (IndexError, struct.error) as e:
                raise RuntimeError('offset 0x%02X' % offset + ' : ' + ' '.join(e.args)) from e

    def _decode_data_entry(self, data: bytes, offset: int, device: int, sample_rate: int) -> int:
        entry = DataMessageDecodeEntry()
        self.data_entry.append(entry)

        entry.device = device

        entry.sample_rate = sample_rate

        entry.channel = data[offset]
        offset += 1

        value = data[offset]
        data_width = (value >> 4) & 0x0F
        present_mode = value & 0x0F
        entry.data_width = data_width
        entry.present_mode = DataPresentMode(present_mode)
        offset += 1

        length = struct.unpack('<H', data[offset: offset + 2])[0]
        offset += 2

        entry.time_stamp = struct.unpack('<I', data[offset:offset + 4])[0]
        offset += 4

        i = 0
        prev = 0

        while i < length:
            i += 1

            index = data[offset]
            offset += 1

            if data_width == 4:
                value = struct.unpack('<I', data[offset:offset + 4])[0]
                offset += 4
            else:
                value = struct.unpack('<H', data[offset:offset + 2])[0]
                offset += 2

            if index < prev:
                prev -= 256

            if index - prev > 1:
                for _ in range(index - prev - 1):
                    entry.data_entry.append(None)

            prev = index

            if present_mode == 0 or present_mode == 1:
                if data_width == 4:
                    entry.data_entry.append(value - 0x8000_0000)
                else:
                    entry.data_entry.append(value - 0x8000)
            else:
                raise RuntimeError('unsupported data present mode : ' + entry.present_mode.name)

        return offset