ansible/roles/bluetooth-monitor/templates/my_btmonitor.py

#!/usr/bin/env python3
import asyncio
from bleak import BleakScanner
from bleak.assigned_numbers import AdvertisementDataType
from bleak.backends.bluezdbus.advertisement_monitor import OrPattern
from bleak.backends.bluezdbus.scanner import BlueZScannerArgs 
from functools import partial
import asyncio_mqtt
import json
from datetime import datetime
import os
import time

# ------------------- Config ----------------------------------------------------------------

config = {
  "mqtt": {
    "hostname": "homeassistant.fritz.box",
    "username": "{{my_btmonitor_mqtt_username}}",
    "password": "{{my_btmonitor_mqtt_password}}",
    "room": "{{sensor_room_name_ascii}}"
  },
  "watchdog_seconds": {{my_bt_monitor_watchdog_seconds | default(None)}},
  "restart_ble_interface": {{my_btmonitor_restart_ble_interface | default(None)}},
}

stop_event = asyncio.Event()
time_last_package_received = datetime.now()


async def on_device_found_callback(mqtt_client, room, device, advertising_data):
  global time_last_package_received
  time_last_package_received = datetime.now()
  rssi = advertising_data.rssi
  tx_power = advertising_data.tx_power
  if  tx_power is not None and rssi is not None:
    topic = f"my_btmonitor/raw_measurements/{room}"
    data = {"address": device.address,
            "rssi": rssi, 
            "tx_power": tx_power}
    try:
      await mqtt_client.publish(topic, json.dumps(data).encode())
    except Exception:
      print("Probably mqtt isn't running - exit whole script and let systemd restart it")
      exit(1)


async def watchdog():
  global time_last_package_received
  timeout = config["watchdog_seconds"]
  if not timeout or timeout <= 0:
    return
  while True:
    restart = (datetime.now() - time_last_package_received).seconds > timeout
    if restart:
      stop_event.set()
    await asyncio.sleep(60) 


async def ble_scan():
    mqtt_conf = config['mqtt']
    while True:
      try:
        async with asyncio_mqtt.Client(hostname=mqtt_conf["hostname"], 
                                       username=mqtt_conf["username"], 
                                       password=mqtt_conf['password']) as mqtt_client:
          cb = partial(on_device_found_callback, mqtt_client, mqtt_conf['room'])
          active_scan = True
          if active_scan:
            async with BleakScanner(cb) as scanner:
              await stop_event.wait()
          else:
            # Doesn't work, because of the strange or_patters
            args = BlueZScannerArgs(
                or_patterns=[OrPattern(0, AdvertisementDataType.MANUFACTURER_SPECIFIC_DATA, b"\x00\x4c")]
            )
            async with BleakScanner(cb, bluez=args, scanning_mode="passive") as scanner:
              await stop_event.wait()
      except Exception as e:
        print("Error", e)              
        print("Starting again")


async def main():
  await asyncio.gather(ble_scan(), watchdog())


if __name__ == "__main__":
  restart_interface = config["restart_ble_interface"]
  if restart_interface:
    print(f"Restarting {restart_interface}")
    os.system(f"hciconfig {restart_interface} down")
    time.sleep(3)
    os.system(f"hciconfig {restart_interface} up")
    time.sleep(3)
    print("Done")
  asyncio.run(main())