CJ_MANUFACTURER_ID = b"\x72\x04" EDDYSTONE_UUID = b"\xaa\xfe" ESTIMOTE_MANUFACTURER_ID = b"\x5d\x01" ESTIMOTE_UUID = b"\x9a\xfe" EVT_LE_ADVERTISING_REPORT = 0x02 EXPOSURE_NOTIFICATION_UUID = b"\x6f\xfd" IBEACON_MANUFACTURER_ID = b"\x4c\x00" IBEACON_PROXIMITY_TYPE = b"\x02\x15" LE_META_EVENT = 0x3e MANUFACTURER_SPECIFIC_DATA_TYPE = 0xFF MS_FRACTION_DIVIDER = 0.625 OCF_LE_SET_SCAN_ENABLE = 0x000C OCF_LE_SET_SCAN_PARAMETERS = 0x000B OGF_LE_CTL = 0x08 class BluetoothAddressType(IntEnum):
    """Determines the scanner MAC-address"""
    PUBLIC = 0x00  # with device MAC-address
    RANDOM = 0x01  # with a random MAC-address class ScanFilter(IntEnum):
    """Determines if only white-listed MAC addresses will be filtered or not"""
    ALL = 0x00
    WHITELIST_ONLY = 0x01 class ScannerMode(IntEnum):
    """Used to determine which packets should be parsed by the scanner."""
    MODE_NONE = 0
    MODE_IBEACON = 1
    MODE_EDDYSTONE = 2
    MODE_ESTIMOTE = 4
    MODE_CJMONITOR = 8
    MODE_EXPOSURE_NOTIFICATION = 16
    MODE_ALL = MODE_IBEACON | MODE_EDDYSTONE | MODE_ESTIMOTE | MODE_CJMONITOR  | MODE_EXPOSURE_NOTIFICATION class ScanType(IntEnum):
    """Determines which type of scan should be executed."""
    PASSIVE = 0x00
    ACTIVE = 0x01 OCF_LE_SET_EXT_SCAN_PARAMETERS = 0x0041 OCF_LE_SET_EXT_SCAN_ENABLE = 0x0042 EVT_LE_EXT_ADVERTISING_REPORT = 0x0D OGF_INFO_PARAM = 0x04 OCF_READ_LOCAL_VERSION = 0x01 EVT_CMD_COMPLETE = 0x0E class BtAddrFilter(DeviceFilter):
    """Filter by bluetooth address."""

    def __init__(self, bt_addr):
        """Initialize filter."""
        super().__init__()
        try:
            bt_addr = bt_addr.lower()
        except AttributeError as exc:
            raise ValueError("bt_addr({}) wasn't a string".format(bt_addr)) from exc
        if not is_valid_mac(bt_addr):
            raise ValueError("Invalid bluetooth MAC address given,"
                             " format should match aa:bb:cc:dd:ee:ff")
        self.properties['bt_addr'] = bt_addr class DeviceFilter(object):
    """Base class for all device filters. Should not be used by itself."""

    def __init__(self):
        """Initialize filter."""
        self.properties = {}

    def matches(self, filter_props):
        """Check if the filter matches the supplied properties."""
        if filter_props is None:
            return False

        found_one = False
        for key, value in filter_props.items():
            if key in self.properties and value != self.properties[key]:
                return False
            elif key in self.properties and value == self.properties[key]:
                found_one = True

        return found_one

    def __repr__(self):
        return "{}({})".format(
            self.__class__.__name__,
            ", ".join(["=".join((k, str(v),)) for k, v in self.properties.items()])) class EddystoneUIDFrame(object):
    """Eddystone UID frame."""

    def __init__(self, data):
        self._tx_power = data['tx_power']
        self._namespace = data_to_hexstring(data['namespace'])
        self._instance = data_to_hexstring(data['instance'])

    @property
    def tx_power(self):
        """Calibrated Tx power at 0 m."""
        return self._tx_power

    @property
    def namespace(self):
        """10-byte namespace identifier."""
        return self._namespace

    @property
    def instance(self):
        """6-byte instance identifier."""
        return self._instance

    @property
    def properties(self):
        """Get beacon properties."""
        return {'namespace': self.namespace, 'instance': self.instance}

    def __str__(self):
        return "EddystoneUIDFrame<tx_power: %d, namespace: %s, instance: %s>" \
               % (self.tx_power, self.namespace, self.instance) class EddystoneURLFrame(object):
    """Eddystone URL frame."""

    def __init__(self, data):
        self._tx_power = data['tx_power']
        url_scheme = EDDYSTONE_URL_SCHEMES[data['url_scheme']]
        url = data['url']

        # Replace url encodings with their expanded version
        for enc, tld in EDDYSTONE_TLD_ENCODINGS.items():
            url = url.replace(chr(enc), tld)

        self._url = url_scheme + url

    @property
    def tx_power(self):
        """Calibrated Tx power at 0 m."""
        return self._tx_power

    @property
    def url(self):
        """Transmitted URL."""
        return self._url

    def __str__(self):
        return "EddystoneURLFrame<tx_power: %d, url: %s>" \
               % (self.tx_power, self.url) class EddystoneEncryptedTLMFrame(object):
    """Eddystone encrypted TLM frame."""

    def __init__(self, data):
        self._encrypted_data = data_to_binstring(data['encrypted_data'])
        self._salt = data['salt']
        self._mic = data['mic']

    @property
    def encrypted_data(self):
        """Encrypted TLM data."""
        return self._encrypted_data

    @property
    def salt(self):
        """16-bit salt."""
        return self._salt

    @property
    def mic(self):
        """16-bit message integrity check."""
        return self._mic

    def __str__(self):
        return "EddystoneEncryptedTLMFrame<encrypted_data: %s, salt: %d, mic: %d>" \
               % (hexlify(self.encrypted_data), self.salt, self.mic) class EddystoneTLMFrame(object):
    """Eddystone TLM frame."""

    def __init__(self, data):
        self._voltage = data['voltage']
        self._temperature = data['temperature'] / float(256)
        self._advertising_count = data['advertising_count']
        self._seconds_since_boot = data['seconds_since_boot']

    @property
    def voltage(self):
        """Battery voltage measured in mV."""
        return self._voltage

    @property
    def temperature(self):
        """Temperature in degree Celsius."""
        return self._temperature

    @property
    def advertising_count(self):
        """Advertising PDU count."""
        return self._advertising_count

    @property
    def seconds_since_boot(self):
        """Time since power-on or reboot."""
        return self._seconds_since_boot

    def __str__(self):
        return "EddystoneTLMFrame<voltage: %d mV, temperature: %d Celsius, advertising count: %d,"\
               " seconds since boot: %d>" % (self.voltage, self.temperature, \
                self.advertising_count, self.seconds_since_boot) class EddystoneEIDFrame(object):
    """Eddystone EID frame."""

    def __init__(self, data):
        self._tx_power = data['tx_power']
        self._eid = data_to_binstring(data['eid'])

    @property
    def tx_power(self):
        """Calibrated Tx power at 0 m."""
        return self._tx_power

    @property
    def eid(self):
        """8-byte Ephemeral Identifier."""
        return self._eid

    def __str__(self):
        return "EddystoneEIDFrame<tx_power: %d, eid: %s>" \
               % (self.tx_power, hexlify(self.eid)) def parse_packet(packet):
    """Parse a beacon advertisement packet."""
    return parse_ltv_packet(packet) def bin_to_int(string):
    """Convert a one element byte string to signed int for python 2 support."""
    if isinstance(string, str):
        return struct.unpack("b", string)[0]
    else:
        return struct.unpack("b", bytes([string]))[0] def bt_addr_to_string(addr):
    """Convert a binary string to the hex representation."""
    addr_str = array.array('B', addr)
    addr_str.reverse()
    hex_str = hexlify(addr_str.tobytes()).decode('ascii')
    # insert ":" seperator between the bytes
    return ':'.join(a+b for a, b in zip(hex_str[::2], hex_str[1::2])) def get_mode(device_filter):
    """Determine which beacons the scanner should look for."""
    from .device_filters import IBeaconFilter, EddystoneFilter, BtAddrFilter, EstimoteFilter, \
                                CJMonitorFilter, ExposureNotificationFilter  # pylint: disable=import-outside-toplevel

    if device_filter is None or len(device_filter) == 0:
        return ScannerMode.MODE_ALL

    mode = ScannerMode.MODE_NONE
    for filtr in device_filter:
        if isinstance(filtr, IBeaconFilter):
            mode |= ScannerMode.MODE_IBEACON
        elif isinstance(filtr, EddystoneFilter):
            mode |= ScannerMode.MODE_EDDYSTONE
        elif isinstance(filtr, EstimoteFilter):
            mode |= ScannerMode.MODE_ESTIMOTE
        elif isinstance(filtr, CJMonitorFilter):
            mode |= ScannerMode.MODE_CJMONITOR
        elif isinstance(filtr, ExposureNotificationFilter):
            mode |= ScannerMode.MODE_EXPOSURE_NOTIFICATION
        elif isinstance(filtr, BtAddrFilter):
            mode |= ScannerMode.MODE_ALL
            break

    return mode def is_one_of(obj, types):
    """Return true iff obj is an instance of one of the types."""
    for type_ in types:
        if isinstance(obj, type_):
            return True
    return False def is_packet_type(cls):
    """Check if class is one the packet types."""
    # pylint: disable=import-outside-toplevel
    from .packet_types import EddystoneUIDFrame, EddystoneURLFrame, \
                              EddystoneEncryptedTLMFrame, EddystoneTLMFrame, \
                              EddystoneEIDFrame, IBeaconAdvertisement, \
                              EstimoteTelemetryFrameA, EstimoteTelemetryFrameB, \
                              ExposureNotificationFrame
    # pylint: enable=import-outside-toplevel

    return (cls in [EddystoneURLFrame, EddystoneUIDFrame, EddystoneEncryptedTLMFrame, \
                    EddystoneTLMFrame, EddystoneEIDFrame, IBeaconAdvertisement, \
                    EstimoteTelemetryFrameA, EstimoteTelemetryFrameB, \
                    ExposureNotificationFrame]) def to_int(string):
    """Convert a one element byte string to int for python 2 support."""
    if isinstance(string, str):
        return ord(string[0])
    else:
        return string
import logging
import struct
import threading
from importlib import import_module
from enum import IntEnum
from construct import Struct, Byte, Bytes, GreedyRange, ConstructError
from ahocorapy.keywordtree import KeywordTree
from .const import (CJ_MANUFACTURER_ID, EDDYSTONE_UUID,
                    ESTIMOTE_MANUFACTURER_ID, ESTIMOTE_UUID,
                    EVT_LE_ADVERTISING_REPORT, EXPOSURE_NOTIFICATION_UUID,
                    IBEACON_MANUFACTURER_ID, IBEACON_PROXIMITY_TYPE,
                    LE_META_EVENT, MANUFACTURER_SPECIFIC_DATA_TYPE,
                    MS_FRACTION_DIVIDER, OCF_LE_SET_SCAN_ENABLE,
                    OCF_LE_SET_SCAN_PARAMETERS, OGF_LE_CTL,
                    BluetoothAddressType, ScanFilter, ScannerMode, ScanType,
                    OCF_LE_SET_EXT_SCAN_PARAMETERS, OCF_LE_SET_EXT_SCAN_ENABLE,
                    EVT_LE_EXT_ADVERTISING_REPORT, OGF_INFO_PARAM,
                    OCF_READ_LOCAL_VERSION, EVT_CMD_COMPLETE)
from .device_filters import BtAddrFilter, DeviceFilter
from .packet_types import (EddystoneEIDFrame, EddystoneEncryptedTLMFrame,
                           EddystoneTLMFrame, EddystoneUIDFrame,
                           EddystoneURLFrame)
from .parser import parse_packet
from .utils import (bin_to_int, bt_addr_to_string, get_mode, is_one_of,
                    is_packet_type, to_int)
    """HCI version enumeration

    https://www.bluetooth.com/specifications/assigned-numbers/host-controller-interface/
    """
    BT_CORE_SPEC_1_0 = 0
    BT_CODE_SPEC_1_1 = 1
    BT_CODE_SPEC_1_2 = 2
    BT_CORE_SPEC_2_0 = 3
    BT_CORE_SPEC_2_1 = 4
    BT_CORE_SPEC_3_0 = 5
    BT_CORE_SPEC_4_0 = 6
    BT_CORE_SPEC_4_1 = 7
    BT_CORE_SPEC_4_2 = 8
    BT_CORE_SPEC_5_0 = 9
    BT_CORE_SPEC_5_1 = 10
    BT_CORE_SPEC_5_2 = 11


_LOGGER = logging.getLogger(__name__)
_LOGGER.setLevel(logging.DEBUG)

# pylint: disable=no-member


class BeaconScanner(object):
    """Scan for Beacon advertisements."""

    def __init__(self, callback, bt_device_id=0, device_filter=None, packet_filter=None, scan_parameters=None):
        """Initialize scanner."""
        # check if device filters are valid
        if device_filter is not None:
            if not isinstance(device_filter, list):
                device_filter = [device_filter]
            if len(device_filter) > 0:
                for filtr in device_filter:
                    if not isinstance(filtr, DeviceFilter):
                        raise ValueError("Device filters must be instances of DeviceFilter")
            else:
                device_filter = None

        # check if packet filters are valid
        if packet_filter is not None:
            if not isinstance(packet_filter, list):
                packet_filter = [packet_filter]
            if len(packet_filter) > 0:
                for filtr in packet_filter:
                    if not is_packet_type(filtr):
                        raise ValueError("Packet filters must be one of the packet types")
            else:
                packet_filter = None

        if scan_parameters is None:
            scan_parameters = {}

        self._mon = Monitor(callback, bt_device_id, device_filter, packet_filter, scan_parameters)

    def start(self):
        """Start beacon scanning."""
        self._mon.start()

    def stop(self):
        """Stop beacon scanning."""
        self._mon.terminate()


class Monitor(threading.Thread):
    """Continously scan for BLE advertisements."""

    def __init__(self, callback, bt_device_id, device_filter, packet_filter, scan_parameters):
        """Construct interface object."""
        # do import here so that the package can be used in parsing-only mode (no bluez required)
        self.backend = import_module('beacontools.backend')

        threading.Thread.__init__(self)
        self.daemon = False
        self.keep_going = True
        self.callback = callback

        # number of the bt device (hciX)
        self.bt_device_id = bt_device_id
        # list of beacons to monitor
        self.device_filter = device_filter
        self.mode = get_mode(device_filter)
        # list of packet types to monitor
        self.packet_filter = packet_filter
        # bluetooth socket
        self.socket = None
        # keep track of Eddystone Beacon <-> bt addr mapping
        self.eddystone_mappings = []
        # parameters to pass to bt device
        self.scan_parameters = scan_parameters
        # hci version
        self.hci_version = HCIVersion.BT_CORE_SPEC_1_0

        # construct an aho-corasick search tree for efficient prefiltering
        service_uuid_prefix = b"\x03\x03"
        self.kwtree = KeywordTree()
        if self.mode & ScannerMode.MODE_IBEACON:
            self.kwtree.add(bytes([MANUFACTURER_SPECIFIC_DATA_TYPE]) + IBEACON_MANUFACTURER_ID + IBEACON_PROXIMITY_TYPE)
        if self.mode & ScannerMode.MODE_EDDYSTONE:
            self.kwtree.add(service_uuid_prefix + EDDYSTONE_UUID)
        if self.mode & ScannerMode.MODE_ESTIMOTE:
            self.kwtree.add(service_uuid_prefix + ESTIMOTE_UUID)
            self.kwtree.add(bytes([MANUFACTURER_SPECIFIC_DATA_TYPE]) + ESTIMOTE_MANUFACTURER_ID)
        if self.mode & ScannerMode.MODE_CJMONITOR:
            self.kwtree.add(bytes([MANUFACTURER_SPECIFIC_DATA_TYPE]) + CJ_MANUFACTURER_ID)
        if self.mode & ScannerMode.MODE_EXPOSURE_NOTIFICATION:
            self.kwtree.add(service_uuid_prefix + EXPOSURE_NOTIFICATION_UUID)
        self.kwtree.finalize()

    def run(self):
        """Continously scan for BLE advertisements."""
        self.socket = self.backend.open_dev(self.bt_device_id)

        self.hci_version = self.get_hci_version()
        self.set_scan_parameters(**self.scan_parameters)
        self.toggle_scan(True)

        while self.keep_going:
            pkt = self.socket.recv(255)

            event = to_int(pkt[1])