#!/usr/bin/env python3 """ ButtonTask BLE Collector — GATT peripheral + advertiser on BlueZ 5.55. Accepts connections from sensors, receives 4-byte reports, keeps addr→slot registry, mirrors latest values in advertising. Pool size N=4. No scanning. ANTI-SCOPE (do not implement / do not "improve"): - encryption / pairing / bonding - provisioning of collector address - RPA resolving - web/CLI UI inside this process - multiple collectors - BLE scanning """ from __future__ import annotations import asyncio import logging import os import signal import socket import sys import threading import time from concurrent.futures import ThreadPoolExecutor from logging.handlers import RotatingFileHandler from pathlib import Path from typing import Any, Dict, List, Optional # Allow `python3 main.py` from collector/ directory. _HERE = Path(__file__).resolve().parent if str(_HERE) not in sys.path: sys.path.insert(0, str(_HERE)) from adv import BlessAdvertiser, build_manufacturer_data # noqa: E402 from bluez_adv import ( # noqa: E402 bluez_owner, connected_devices, disconnect_device, system_bus, ) from config import ( # noqa: E402 BLUEZ_CALL_TIMEOUT_S, BLUEZ_HANG_S, BLUEZ_OWNER_CHECK_S, BLUEZ_START_TIMEOUT_S, BT_SERVICE_HELPER, LOCAL_NAME, N, TICK_INTERVAL_S, collector_log_path, registry_path, run_dir, ) from gatt import CollectorGatt, build_bless_server # noqa: E402 from history import History # noqa: E402 from protocol import ( # noqa: E402 CFG_JOIN_OPEN, CFG_POOL_FULL, DIST_NO_TARGET, pack_interval_min, pack_slot_config, unpack_report, ) from registry import Registry # noqa: E402 from settings import ( # noqa: E402 load_names, load_report_interval_min, save_report_interval_min, ) from ui_io import append_ui_log, read_cmd, read_history, write_history, write_status # noqa: E402 log = logging.getLogger("collector") def setup_logging() -> None: run_dir() root = logging.getLogger("collector") root.setLevel(logging.INFO) # python3-dbus calls logging.basicConfig() on its first error; without this # every collector line would be printed twice from then on. root.propagate = False fmt = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s") sh = logging.StreamHandler(sys.stderr) sh.setFormatter(fmt) root.addHandler(sh) try: fh = RotatingFileHandler( collector_log_path(), maxBytes=512_000, backupCount=2, encoding="utf-8" ) fh.setFormatter(fmt) root.addHandler(fh) except OSError as e: root.warning("file log unavailable: %s", e) def sd_notify(state: str) -> None: """Minimal sd_notify(3): READY=1 / WATCHDOG=1 for Type=notify units.""" addr = os.environ.get("NOTIFY_SOCKET") if not addr: return if addr.startswith("@"): addr = "\0" + addr[1:] try: with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as s: s.connect(addr) s.sendall(state.encode()) except OSError as e: log.debug("sd_notify: %s", e) class BluezGone(Exception): """bluetoothd restarted or hung — our GATT/adv registrations are lost.""" def __init__(self, msg: str, restart_bluez: bool = False): super().__init__(msg) self.restart_bluez = restart_bluez def _err(e: BaseException) -> str: return f"{type(e).__name__}: {e}" if str(e) else type(e).__name__ def preflight() -> None: try: import bless # noqa: F401 except Exception as e: log.error("bless import failed: %s", e) raise SystemExit(2) from e try: bus = system_bus() bus.get_object("org.bluez", "/") except Exception as e: log.error("D-Bus / bluetoothd unavailable: %s", e) raise SystemExit(2) from e log.info("preflight ok (bless + SystemBus + bluez)") class CollectorApp: def __init__(self): self.registry = Registry(n=N, path=registry_path()) self.gatt = CollectorGatt(self.registry) self.bus = None self.advertiser: BlessAdvertiser | None = None self.server = None self._known_connected: set[str] = set() self._stop = threading.Event() self._names: Dict[str, str] = {} self._dirty_registry = False self._dirty_status = False self._dirty_adv = False self._pending_logs: List[Dict[str, Any]] = [] self._lock = threading.Lock() self._started_ts = time.time() self._history = History() self._history.load_dict(read_history()) self._dirty_history = False self.exit_code = 0 # All blocking python3-dbus calls go here, never on the asyncio thread: # a hung bluetoothd must not freeze bless (same event loop). self._executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="bluez") self._bluez_owner: Optional[str] = None self._bluez_fail_since: Optional[float] = None self._last_owner_check = 0.0 self.gatt.on_write = self._on_write self.gatt.on_read_slot = self._on_read_slot self.gatt.on_read_interval = self._on_read_interval self.gatt.on_connect = self._on_connect self.gatt.on_disconnect = self._on_disconnect def _reload_settings(self, *, force_log: bool = False) -> None: interval = load_report_interval_min() names = load_names() changed = ( interval != self.gatt.report_interval_min or names != self._names ) self.gatt.report_interval_min = interval self._names = names if force_log or changed: log.info( "settings: report_interval_min=%s names=%d", self.gatt.report_interval_min, len(self._names), ) def _queue_log(self, level: str, message: str, **extra: Any) -> None: with self._lock: self._pending_logs.append({"level": level, "msg": message, **extra}) def _mark_dirty(self, registry: bool = False, status: bool = True, adv: bool = False) -> None: with self._lock: if registry: self._dirty_registry = True if status: self._dirty_status = True if adv: self._dirty_adv = True def _connected(self) -> bool: return bool(self._known_connected) def _sync_status(self) -> None: status = self.registry.status_dict( names=self._names, report_interval_min=self.gatt.report_interval_min, ) status["started_ts"] = self._started_ts write_status(status) def _remember_report(self, addr: str, report, rssi: Optional[int]) -> None: raw = None if report.dist == DIST_NO_TARGET else int(report.dist) with self._lock: self._history.add(addr, time.time(), raw, int(report.battery), rssi) self._dirty_history = True def _mirror_bytes(self) -> bytes: bitmap, reports = self.registry.mirror_payload_parts() return build_manufacturer_data(bitmap, reports) def _refresh_mirror(self, force: bool = False) -> None: if self.advertiser is None: return if not force and self._connected(): # Do not tear down advertising while a central is connected. return self.advertiser.set_mirror(self._mirror_bytes(), force=force) def _on_connect(self, addr: str, rssi: Optional[int] = None) -> bool: addr = self.registry.normalize_addr(addr) sensor = self.registry.get(addr) if sensor is None: sensor = self.registry.assign(addr) if sensor is None: log.warning("pool-full, rejecting %s", addr) self._queue_log("warn", "pool-full", addr=addr) self._mark_dirty(registry=True, status=True) return False self._queue_log("info", "assigned", addr=addr, slot=sensor.slot) log.info("assigned slot=%s addr=%s", sensor.slot, addr) else: self.registry.touch(addr) if rssi is not None: sensor.rssi = rssi self.gatt.set_current_addr(addr) self._mark_dirty(registry=True, status=True) return True def _on_disconnect(self, addr: str) -> None: addr = self.registry.normalize_addr(addr) sensor = self.registry.get(addr) if sensor is not None: # Keep last_seen from last report; only bump state. if sensor.state != "stale": sensor.state = "live" if self.gatt.current_addr() == addr: self.gatt.set_current_addr(None) self._queue_log("info", "disconnect", addr=addr) self._mark_dirty(registry=True, status=True, adv=True) log.info("disconnect %s", addr) def _on_write(self, addr: str, data: bytes) -> None: """GATT hot-path: RAM only — no disk / D-Bus / adv / UI I/O.""" addr = self.registry.normalize_addr(addr) or (self.gatt.current_addr() or "") if not addr: if len(self._known_connected) == 1: addr = next(iter(self._known_connected)) else: log.warning("write without address — discarded") return report = unpack_report(data) if report is None: log.warning("bad report from %s — discarded", addr) return sensor = self.registry.get(addr) if sensor is None: sensor = self.registry.assign(addr) if sensor is None: log.warning("pool-full on write from %s", addr) self._queue_log("warn", "pool-full", addr=addr) self._mark_dirty(registry=True, status=True) return self.gatt.set_current_addr(addr) self._queue_log("info", "assigned", addr=addr, slot=sensor.slot) self.registry.update_report( addr, report.battery, report.flags, report.dist, rssi=None ) self._remember_report(addr, report, sensor.rssi if sensor else None) self._queue_log( "info", "report", addr=addr, battery=report.battery, flags=report.flags, dist=report.dist, ) self._mark_dirty(registry=True, status=True, adv=True) log.debug( "report slot=%s addr=%s batt=%s dist=%s", sensor.slot if sensor else "?", addr, report.battery, report.dist, ) def _on_read_slot(self, addr: str) -> bytes: addr = self.registry.normalize_addr(addr) or (self.gatt.current_addr() or "") sensor = self.registry.get(addr) if addr else None slot = sensor.slot if sensor else 0xFF flags = CFG_JOIN_OPEN if not self.registry.pool_full() else CFG_POOL_FULL return pack_slot_config(slot, flags, self.gatt.period_s()) def _on_read_interval(self) -> bytes: return pack_interval_min(self.gatt.report_interval_min) # --- BlueZ health ------------------------------------------------------- async def _bluez(self, fn, *args): """Run a blocking bluez_adv helper in the worker thread with a timeout.""" loop = asyncio.get_running_loop() return await asyncio.wait_for( loop.run_in_executor(self._executor, fn, self.bus, *args), BLUEZ_CALL_TIMEOUT_S, ) def _bluez_ok(self) -> None: if self._bluez_fail_since is not None: log.info( "bluez responsive again after %.1f s", time.monotonic() - self._bluez_fail_since, ) self._bluez_fail_since = None def _bluez_failed(self, what: str, e: BaseException) -> None: now = time.monotonic() if self._bluez_fail_since is None: self._bluez_fail_since = now log.warning("bluez %s failed: %s", what, _err(e)) return down_s = now - self._bluez_fail_since if down_s >= BLUEZ_HANG_S: raise BluezGone( f"bluetoothd unresponsive for {down_s:.0f} s ({what}: {_err(e)})", restart_bluez=True, ) async def _check_bluez_owner(self) -> None: now = time.monotonic() if now - self._last_owner_check < BLUEZ_OWNER_CHECK_S: return self._last_owner_check = now try: owner = await self._bluez(bluez_owner) except Exception as e: self._bluez_failed("owner check", e) return if owner != self._bluez_owner: raise BluezGone(f"bluetoothd owner changed {self._bluez_owner} -> {owner}") async def _restart_bluetoothd(self) -> None: log.error("restarting bluetooth.service via %s", BT_SERVICE_HELPER) try: proc = await asyncio.create_subprocess_exec( "sudo", "-n", BT_SERVICE_HELPER, "bluez-restart", stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.PIPE, ) _, err = await asyncio.wait_for(proc.communicate(), 30) if proc.returncode != 0: log.error( "bluez-restart exit=%s: %s", proc.returncode, (err or b"").decode(errors="replace").strip(), ) except Exception as e: log.error("bluez-restart failed: %s", _err(e)) async def _advertise(self) -> None: if self.advertiser is None: return try: await asyncio.wait_for(self.advertiser.tick(), BLUEZ_CALL_TIMEOUT_S) except asyncio.TimeoutError as e: log.error("adv update timed out") self.advertiser.force_reregister() self._bluez_failed("adv update", e) # --- connections --------------------------------------------------------- async def _poll_connections(self) -> None: if self.bus is None: return try: devices = await self._bluez(connected_devices) except Exception as e: self._bluez_failed("conn poll", e) return self._bluez_ok() connected = set(devices) for addr in connected - self._known_connected: ok = self._on_connect(addr, devices.get(addr)) if not ok: try: await self._bluez(disconnect_device, addr) except Exception as e: log.warning("disconnect %s: %s", addr, _err(e)) for addr in self._known_connected - connected: self._on_disconnect(addr) self._known_connected = connected if connected and self.gatt.current_addr() is None: # Prefer the sole connected peer as current writer context. if len(connected) == 1: self.gatt.set_current_addr(next(iter(connected))) def _handle_cmd(self) -> None: cmd = read_cmd() if not cmd: return op = str(cmd.get("op", "")).lower() if op == "advertise": self._queue_log("info", "manual-advertise") log.info("manual advertise requested") self._refresh_mirror(force=True) elif op == "reload_settings": self._reload_settings() self._mark_dirty(status=True) if "report_interval_min" in cmd: try: self.gatt.report_interval_min = save_report_interval_min( int(cmd["report_interval_min"]) ) except (TypeError, ValueError): pass log.info("reload_settings done interval=%s", self.gatt.report_interval_min) elif op == "set_interval": try: minutes = int(cmd.get("report_interval_min", self.gatt.report_interval_min)) except (TypeError, ValueError): return self.gatt.report_interval_min = save_report_interval_min(minutes) self._mark_dirty(status=True) self._queue_log("info", "set-interval", minutes=self.gatt.report_interval_min) log.info("set_interval %s min", self.gatt.report_interval_min) def _flush_dirty(self) -> None: with self._lock: do_reg = self._dirty_registry do_st = self._dirty_status do_adv = self._dirty_adv logs = list(self._pending_logs) do_hist = self._dirty_history hist_snapshot = self._history.to_dict() if do_hist else None self._dirty_registry = False self._dirty_status = False self._dirty_adv = False self._dirty_history = False self._pending_logs.clear() for entry in logs: level = entry.pop("level", "info") msg = entry.pop("msg", "") append_ui_log(level, msg, **entry) if do_reg: self.registry.save() if do_st: self._sync_status() if do_hist and hist_snapshot is not None: write_history(hist_snapshot) if do_adv: # Only when idle (no connected centrals), unless already forced earlier. self._refresh_mirror(force=False) async def run(self) -> None: self._reload_settings(force_log=True) self.registry.load() self._sync_status() append_ui_log("info", "startup", sensors=len(self.registry.by_addr)) self.bus = system_bus() loop = asyncio.get_running_loop() # Owner before registration: if bluetoothd restarts later, the owner # changes and we know our GATT app / advertisement are gone. self._bluez_owner = await self._bluez(bluez_owner) if self._bluez_owner is None: raise RuntimeError("bluetoothd is not running") self.server = await asyncio.wait_for( build_bless_server(self.gatt, loop, LOCAL_NAME), BLUEZ_START_TIMEOUT_S ) await asyncio.wait_for(self.server.start(), BLUEZ_START_TIMEOUT_S) log.info("GATT server started (%s, bluez=%s)", LOCAL_NAME, self._bluez_owner) self.advertiser = BlessAdvertiser(self.server) self._refresh_mirror(force=True) await self._advertise() sd_notify("READY=1") while not self._stop.is_set(): sd_notify("WATCHDOG=1") try: await self._check_bluez_owner() # Pick up web/Qt edits to names & interval without a separate watcher. self._reload_settings() await self._poll_connections() if self.registry.reevaluate(period_s=self.gatt.period_s()): self._mark_dirty(registry=True, status=True, adv=True) self._handle_cmd() self._flush_dirty() if not self._connected(): await self._advertise() # Refresh age_s even when nothing else changed. self._sync_status() except BluezGone as e: log.error("%s — exiting for a clean restart", e) if e.restart_bluez: await self._restart_bluetoothd() self.exit_code = 1 break except Exception as e: log.error("tick error: %s", _err(e)) try: await asyncio.sleep(TICK_INTERVAL_S) except asyncio.CancelledError: break await self._shutdown() async def _shutdown(self) -> None: log.info("shutdown") append_ui_log("info", "shutdown") try: if self.server is not None: await asyncio.wait_for(self.server.stop(), 5) except Exception as e: log.warning("server stop: %s", _err(e)) self._executor.shutdown(wait=False) self._flush_dirty() self.registry.save() self._sync_status() def request_stop(self) -> None: self._stop.set() def main() -> None: setup_logging() preflight() loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) app = CollectorApp() def _sig(*_a): log.info("signal → stop") app.request_stop() for task in asyncio.all_tasks(loop): task.cancel() for sig in (signal.SIGINT, signal.SIGTERM): try: loop.add_signal_handler(sig, _sig) except NotImplementedError: signal.signal(sig, lambda *_: _sig()) try: loop.run_until_complete(app.run()) except asyncio.CancelledError: pass finally: loop.close() # Non-zero → systemd (Restart=always) starts a fresh collector. sys.exit(app.exit_code) if __name__ == "__main__": main()