controller-wisetopdataserver/python/biopro/impl/cc2650/data.py

import abc
import struct as _struct

from biopro.util.address import ADDRESS
from biopro.util.console import hex_line as hl
from .gap import *

EVENT_TYPE = {
    0x00: 'Connect-able undirected advertisement',
    0x01: 'Connect-able directed advertisement',
    0x02: 'Discoverable undirected advertisement',
    0x03: 'Non-connect-able undirected advertisement',
    0x04: 'Scan Response',
}


class DeviceInfo(metaclass=abc.ABCMeta):
    __slots__ = ()

    @property
    @abc.abstractmethod
    def event_type(self) -> uint8:
        pass

    @property
    @abc.abstractmethod
    def address_type(self) -> AddressType:
        pass

    @property
    @abc.abstractmethod
    def address(self) -> ADDRESS:
        """device address"""
        pass

    def dump(self, c: Optional[int], prefix: str = ''):
        print(pc('%sevent_type' % prefix, c), ':', self.event_type,
              pp(EVENT_TYPE.get(self.event_type, None)))
        print(pc('%saddress_type' % prefix, c), ':', self.address_type,
              pp(AddressType.type_str(self.address_type, None)))
        print(pc('%saddress' % prefix, c), ':', address_str(self.address))


class DiscoveryDevice(DeviceInfo):
    __slots__ = ('_event_type', '_address_type', '_address')

    def __init__(self, event_type: uint8, address_type: AddressType, address: ADDRESS):
        self._event_type = event_type
        self._address_type: AddressType = address_type
        self._address = address

    @property
    def event_type(self) -> uint8:
        return self._event_type

    @property
    def address_type(self) -> AddressType:
        return self._address_type

    @property
    def address(self) -> ADDRESS:
        return self._address


class LinkParameter:
    __slots__ = ('interval_min', 'interval_max', 'latency', 'timeout', 'info_min', 'info_max')

    def __init__(self, parameter: Optional['LinkParameter'] = None):
        # Minimum allowed connection interval.
        self.interval_min: uint16 = 0
        # Maximum allowed connection interval.
        self.interval_max: uint16 = 0
        # Number of skipped events (slave latency).
        self.latency: uint16 = 0
        # Connection supervision timeout.
        self.timeout: uint16 = 0
        # Info parameter about min length of conn.
        self.info_min: uint16 = 0
        # Info parameter about max length of conn.
        self.info_max: uint16 = 0

        if parameter is not None:
            self.update(parameter)

    def update(self, parameter: 'LinkParameter'):
        self.interval_min = parameter.interval_min
        self.interval_max = parameter.interval_max
        self.latency = parameter.latency
        self.timeout = parameter.timeout
        self.info_min = parameter.info_min
        self.info_max = parameter.info_max

    def valid(self, f: str, contain_info_length=True):
        if not 6 <= self.interval_min <= 0x0C80:
            raise RuntimeError('illegal ' + f + ' interval_min : ' + x16(self.interval_min))

        if not 6 <= self.interval_max <= 0x0C80:
            raise RuntimeError('illegal ' + f + ' interval_max : ' + x16(self.interval_max))

        if not self.interval_min <= self.interval_max:
            raise RuntimeError('illegal ' + f + ' interval : not %d <= %d' % (self.interval_min, self.interval_max))

        if not 0 <= self.latency <= 499:
            raise RuntimeError('illegal ' + f + ' latency : ' + x16(self.latency))

        if not 10 <= self.timeout <= 0x0C80:
            raise RuntimeError('illegal ' + f + ' timeout : ' + x16(self.timeout))

        if contain_info_length:
            # XXX LinkParameter.info_min/max validating
            pass

    def dump(self, c: Optional[int] = None, contain_info_length=True):
        print(pc('interval_min', c), ':', x16(self.interval_min), pp('%.1f ms' % (self.interval_min * 1.25)))
        print(pc('interval_max', c), ':', x16(self.interval_max), pp('%.1f ms' % (self.interval_max * 1.25)))
        print(pc('latency', c), ':', x16(self.latency), pp('%.1f ms' % (self.latency * 1.25)))
        print(pc('timeout', c), ':', x16(self.timeout), pp('%.1f ms' % (self.timeout * 10)))
        if contain_info_length:
            print(pc('info_min', c), ':', x16(self.info_min))
            print(pc('info_min', c), ':', x16(self.info_min))


class AdvertisingToken:
    __slots__ = ('ad_type', 'ad_data')

    def __init__(self, ad_type: GapAdvertisingDataType, ad_data: bytes):
        self.ad_type = ad_type
        self.ad_data = ad_data

    def dump(self, c: Optional[int] = None):
        ad_type = self.ad_type

        print(pc('ad_type', c), ':', x8(ad_type),
              pp(GapAdvertisingDataType.type_str(ad_type, None)))

        try:
            if ad_type == GapAdvertisingDataType.POWER_LEVEL:
                value = _struct.unpack('b', self.ad_data)[0]
            elif ad_type in (GapAdvertisingDataType.BIT16_MORE,
                             GapAdvertisingDataType.BIT16_COMPLETE,
                             GapAdvertisingDataType.BIT32_MORE,
                             GapAdvertisingDataType.BIT32_COMPLETE,
                             GapAdvertisingDataType.BIT128_MORE,
                             GapAdvertisingDataType.BIT128_COMPLETE,
                             GapAdvertisingDataType.OOB_CLASS_OF_DEVICE,
                             GapAdvertisingDataType.OOB_SIMPLE_PAIRING_HASHC,
                             GapAdvertisingDataType.OOB_SIMPLE_PAIRING_RANDR,
                             GapAdvertisingDataType.SERVICES_LIST_16BIT,
                             GapAdvertisingDataType.SERVICES_LIST_128BIT):
                value = hl(self.ad_data, ':')
            elif ad_type == GapAdvertisingDataType.SLAVE_CONN_INTERVAL_RANGE:
                value = _struct.unpack('HH', self.ad_data)
                value = '%s %s - %s %s' % (x16(value[0]),
                                           pp('%.1f ms' % (value[0] * 1.75)),
                                           x16(value[1]),
                                           pp('%.1f ms' % (value[1] * 1.75)))
            elif ad_type in (GapAdvertisingDataType.LOCAL_NAME_SHORT,
                             GapAdvertisingDataType.LOCAL_NAME_COMPLETE):
                value = self.ad_data.decode()
            elif ad_type == GapAdvertisingDataType.MANUFACTURER_SPECIFIC:
                print(pc('ad_info', c), ':', hl(self.ad_data[:2]))
                value = hl(self.ad_data[2:], truncate=True)
            else:
                value = hl(self.ad_data, truncate=True)
        except RuntimeError:
            # fail back
            value = hl(self.ad_data, truncate=True)

        print(pc('ad_data', c), ':', value)