173 lines
6.7 KiB
Python
173 lines
6.7 KiB
Python
"""
|
||
ble_refactor.py – a drop‑in replacement for your old ble.py + scan_macs / update_cache logic.
|
||
|
||
Key points:
|
||
* One global asyncio event loop that lives for the life of the process.
|
||
* `update_cache()` can be called from any thread (e.g. a Flask background job).
|
||
* Thread‑safe read‑access via `get_sensor_summary()` / `get_sensor_details()`.
|
||
* Optional HCI reset on read timeout (cheap, no USB re‑plug).
|
||
"""
|
||
|
||
import asyncio
|
||
import logging
|
||
import subprocess
|
||
import threading
|
||
from datetime import datetime, timezone
|
||
from typing import Dict, List, Tuple
|
||
|
||
from bleak import BleakClient, BleakScanner, BleakError
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# CONFIG
|
||
# --------------------------------------------------------------------------- #
|
||
SCAN_TIMEOUT: float = 5.0 # seconds per scan
|
||
READ_TIMEOUT: float = 10.0 # seconds per GATT read
|
||
MAX_RETRIES: int = 3
|
||
|
||
# Mi sensor UUIDs (replace with yours if needed)
|
||
MI_SERVICE_UUID: str = "ebe0ccb0-7a0a-4b0c-8a1a-6ff2997da3a6"
|
||
TEMP_CHAR_UUID: str = "ebe0ccc1-7a0a-4b0c-8a1a-6ff2997da3a6"
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# STATE & LOCK
|
||
# --------------------------------------------------------------------------- #
|
||
_mac_cache: set[str] = set() # all known MACs
|
||
_sensors_cache: Dict[str, Dict] = {} # mac → {temp: float, humidity: float}
|
||
|
||
_lock = threading.Lock() # protects the two dicts above
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# GLOBAL EVENT LOOP (one per process)
|
||
# --------------------------------------------------------------------------- #
|
||
_loop = asyncio.new_event_loop()
|
||
|
||
def _start_loop() -> None:
|
||
asyncio.set_event_loop(_loop)
|
||
_loop.run_forever()
|
||
|
||
# Start the loop in a background daemon thread
|
||
_loop_thread = threading.Thread(target=_start_loop, daemon=True, name="ble-loop")
|
||
_loop_thread.start()
|
||
|
||
# Helper: submit a coroutine to the background loop
|
||
def _run_coroutine(coro):
|
||
return asyncio.run_coroutine_threadsafe(coro, _loop)
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# CORE FUNCTIONS
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
async def _scan_once() -> List[str]:
|
||
"""Return a list of MAC addresses that advertise the Mi sensor name."""
|
||
devices = await BleakScanner.discover(timeout=SCAN_TIMEOUT)
|
||
return [dev.address for dev in devices if dev.name == "LYWSD03MMC"]
|
||
|
||
async def _read_sensor(mac: str) -> Tuple[float, int]:
|
||
"""Read temp (°C) and humidity (%) from one sensor."""
|
||
for attempt in range(1, MAX_RETRIES + 1):
|
||
try:
|
||
async with BleakClient(mac, timeout=READ_TIMEOUT) as client:
|
||
await client.wait_for_connected()
|
||
|
||
# Grab the Mi service – if missing, treat as failure
|
||
services = await client.get_services()
|
||
if MI_SERVICE_UUID not in [s.uuid for s in services]:
|
||
raise BleakError("MI service not found")
|
||
|
||
data = await client.read_gatt_char(TEMP_CHAR_UUID)
|
||
|
||
# Same parsing logic as your JS code
|
||
sign = data[1] & (1 << 7)
|
||
temp = ((data[1] & 0x7F) << 8 | data[0])
|
||
if sign:
|
||
temp -= 32767
|
||
temp = temp / 100
|
||
hum = data[2]
|
||
return temp, hum
|
||
|
||
except (asyncio.TimeoutError, BleakError) as exc:
|
||
logging.warning(f"[{mac}] attempt {attempt}/{MAX_RETRIES} failed: {exc}")
|
||
|
||
if attempt == MAX_RETRIES:
|
||
# Cheap HCI reset – often clears the “stuck” state
|
||
logging.info(f"Resetting HCI to recover from stall on {mac}")
|
||
subprocess.run(["sudo", "hciconfig", "hci0", "reset"], check=False)
|
||
raise
|
||
await asyncio.sleep(1) # back‑off before retry
|
||
|
||
raise RuntimeError(f"Unreachable code – read sensor {mac}") # pragma: no cover
|
||
|
||
async def _update_cache() -> None:
|
||
"""High‑level: scan → read all sensors → update in‑memory cache."""
|
||
try:
|
||
new_macs = await _scan_once()
|
||
except Exception as exc:
|
||
logging.error(f"Scan failed: {exc}")
|
||
return
|
||
|
||
sensor_results: List[Tuple[str, Tuple[float, int]]] = []
|
||
|
||
for mac in new_macs:
|
||
try:
|
||
temp, hum = await _read_sensor(mac)
|
||
sensor_results.append((mac, (temp, hum)))
|
||
except Exception as exc:
|
||
logging.error(f"Failed to read {mac}: {exc}")
|
||
|
||
with _lock:
|
||
for mac, (temp, hum) in sensor_results:
|
||
_sensors_cache[mac] = {"temp": temp, "humidity": hum}
|
||
_mac_cache.add(mac)
|
||
|
||
logging.info(f"Cache updated: {sorted(_mac_cache)}")
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# PUBLIC API (callable from Flask / background jobs)
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
def update_cache() -> None:
|
||
"""
|
||
Entry point used by the scheduler / Flask thread.
|
||
Schedules the async work on the global event loop.
|
||
"""
|
||
_run_coroutine(_update_cache())
|
||
|
||
def get_sensor_summary() -> List[Dict]:
|
||
"""Return a list of sensor dicts – minimal: name, temp, humidity."""
|
||
with _lock:
|
||
return [
|
||
{"name": mac, **data} for mac, data in _sensors_cache.items()
|
||
]
|
||
|
||
def get_sensor_details(friendly_date: bool = False) -> Dict:
|
||
"""Return the full sensor dict + timestamps (converted if requested)."""
|
||
def ts_to_pst(ts: float) -> str:
|
||
utc = datetime.fromtimestamp(ts, tz=timezone.utc)
|
||
pacific = utc.astimezone(datetime.now().astimezone().tzinfo) # use local tz
|
||
return pacific.strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
with _lock:
|
||
sensors = []
|
||
for mac, data in _sensors_cache.items():
|
||
# Here you can add `first_seen`/`last_seen` if you keep timestamps elsewhere
|
||
sensors.append({"name": mac, **data})
|
||
|
||
return {"sensors": sensors}
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# OPTIONAL: HCI RESET ON START
|
||
# --------------------------------------------------------------------------- #
|
||
def reset_hci_on_start() -> None:
|
||
"""Call this once during app init if you want to be extra safe."""
|
||
subprocess.run(["sudo", "hciconfig", "hci0", "reset"], check=False)
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# MAIN (for quick manual testing)
|
||
# --------------------------------------------------------------------------- #
|
||
if __name__ == "__main__":
|
||
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
|
||
reset_hci_on_start()
|
||
logging.info("Starting periodic update (every 30s)…")
|
||
while True:
|
||
update_cache()
|
||
time.sleep(30) |