contact tracing beacon scanner

scanner.py

import signal
import os
import time
from beacontools import BeaconScanner
from beacontools.scanner import Monitor
from beacontools.utils import is_packet_type, is_one_of, to_int, bin_to_int, get_mode, bt_addr_to_string, data_to_hexstring
from beacontools.const import *

class MyMonitor(Monitor):
    def run(self):
        """Continously scan for BLE advertisements."""
        self.socket = self.bluez.hci_open_dev(self.bt_device_id)
        self.seen = {}

        filtr = self.bluez.hci_filter_new()
        self.bluez.hci_filter_all_events(filtr)
        self.bluez.hci_filter_set_ptype(filtr, self.bluez.HCI_EVENT_PKT)
        self.socket.setsockopt(self.bluez.SOL_HCI, self.bluez.HCI_FILTER, filtr)

        self.set_scan_parameters()
        self.toggle_scan(True)

        while self.keep_going:
            pkt = self.socket.recv(255)
            event = to_int(pkt[1])
            subevent = to_int(pkt[3])
            if event == LE_META_EVENT and subevent == EVT_LE_ADVERTISING_REPORT:
                if pkt[19:21] == b'\x16o':
                    # Covid-19 beacon
                    addr = bt_addr_to_string(pkt[7:13])
                    if not addr in self.seen:
                        self.seen[addr] = (time.asctime(), [to_int(pkt[-1])])
                        print("%s <%s> %s %s %d" % (self.seen[addr][0], addr, data_to_hexstring(pkt[22:38]), data_to_hexstring(pkt[38:42]), to_int(pkt[-1])))
                    else:
                        self.seen[addr][1].append(to_int(pkt[-1]))

        self.socket.close()

    def dump(self):
        print(self.seen)

    def stop(self):
        self.keep_going = False

def callback(bt_addr, rssi, packet, additional_info):
    print("<%s, %d> %s %s" % (bt_addr, rssi, packet, additional_info))

if __name__ == "__main__":
    scanner = MyMonitor(callback, 0, None, None)
    signal.signal(signal.SIGUSR1, lambda x, y: scanner.dump())
    signal.signal(signal.SIGINT, lambda x, y: scanner.stop())
    # scan for all TLM frames of beacons in the namespace "12345678901234678901"
    scanner.start()
    scanner.join()