From 6564d4941facd165ec69067349b5e0364b572d66 Mon Sep 17 00:00:00 2001 From: matt Date: Sun, 18 Oct 2026 06:38:14 -0700 Subject: [PATCH] ble.py - lots of comment maintenance --- api/ble.py | 293 +++++++++++++++++++++++++++++------------------------ 1 file changed, 161 insertions(+), 132 deletions(-) diff --git a/api/ble.py b/api/ble.py index 5c30251..bc4d21e 100644 --- a/api/ble.py +++ b/api/ble.py @@ -1,21 +1,34 @@ #!/usr/bin/env python3 + +#################################################### """ -mi_ble_reader -============== -Async BLE reader for Xiaomi Mi temperature / humidity sensors. + BLE Temp Sensor Driver + First part of this file is the driver itself + Since it's intended to be used with a service that could run for months or more, + it also has a robust test system included -This module keeps the original helper names (`sensor_data_query`, `test`, etc.) so -you can keep your existing code unchanged. In addition it exposes a minimal -public API that other apps can import: +The two primary functions for interacting with the driver are: +def scan_macs() -> List[str]: +def sensor_data_query(mac_list: List[str]) -> List[dict]: -* `read_data` - async coroutine, returns SensorReading or None -* `read_multiple` - async coroutine, reads a list of MACs concurrently -* `read_data_sync` - sync wrapper around `read_data` (creates its own loop) + A lot of this came from LLM assistance, including the next blurb + + Async BLE reader for Xiaomi Mi temperature / humidity sensors. + + This module keeps the original helper names (`sensor_data_query`, `test`, etc.) so + you can keep your existing code unchanged. In addition it exposes a minimal + public API that other apps can import: + + * `read_data` - async coroutine, returns SensorReading or None + * `read_multiple` - async coroutine, reads a list of MACs concurrently + * `read_data_sync` - sync wrapper around `read_data` (creates its own loop) + + All heavy lifting is done in the async functions; the sync wrapper is only + provided for convenience. -All heavy lifting is done in the async functions; the sync wrapper is only -provided for convenience. """ +#################################################### import asyncio import subprocess @@ -28,9 +41,9 @@ from typing import List, Optional from config import * from bleak import BleakClient, BleakScanner -# --------------------------------------------------------------------------- # +#################################################### # Configuration -# --------------------------------------------------------------------------- # +#################################################### MI_SERVICE_UUID = "ebe0ccb0-7a0a-4b0c-8a1a-6ff2997da3a6" TEMP_CHAR_UUID = "ebe0ccc1-7a0a-4b0c-8a1a-6ff2997da3a6" @@ -41,9 +54,9 @@ SERVICE_DISCOVERY_TIMEOUT = 10.0 DISCOVER_TIMEOUT = SCAN_TIMEOUT + 5.0 # Timeout for discovering device current_usb_id = None -# --------------------------------------------------------------------------- # +#################################################### # Logging -# --------------------------------------------------------------------------- # +#################################################### _logger = logging.getLogger(__name__) _logger.setLevel(logging.INFO) @@ -60,9 +73,9 @@ _logger.addHandler(_stdout_handler) # finally create the log log = _logger -# --------------------------------------------------------------------------- # +#################################################### # Data model -# --------------------------------------------------------------------------- # +#################################################### @dataclass class SensorReading: """Single Mi sensor reading.""" @@ -71,11 +84,13 @@ class SensorReading: humidity: int -# --------------------------------------------------------------------------- # -# Helpers - kept identical to your original code -# --------------------------------------------------------------------------- # +#################################################### +# BLE Helpers mined from the javascript +# This magic is necessary and mysterious +#################################################### + +# Scan once for the given MAC and return its address if found async def _discover_device(mac: str) -> Optional[str]: - """Scan once for the given MAC and return its address if found.""" try: devices = await asyncio.wait_for( BleakScanner.discover(timeout=SCAN_TIMEOUT), @@ -93,8 +108,9 @@ async def _discover_device(mac: str) -> Optional[str]: reset_adapter() return None +# Don't question the ancient magic +# This evidently will "Parse the 3-byte Mi packet into temperature & humidity." def _parse_sensor_data(raw: bytes) -> (float, int): - """Parse the 3-byte Mi packet into temperature & humidity.""" sign = raw[1] & (1 << 7) temp = ((raw[1] & 0x7F) << 8 | raw[0]) if sign: @@ -103,26 +119,42 @@ def _parse_sensor_data(raw: bytes) -> (float, int): hum = raw[2] return temp, hum -# --------------------------------------------------------------------------- # -# Optional device discovery helper -# --------------------------------------------------------------------------- # +#################################################### +# BLE device discovery function and helpers +# Returns a list of MACs of sensors detected +#################################################### + +# This is the primary function for interacting with the sensor data through the driver +# Provide it with a List of MACs and it will return a list of Sensor data +def sensor_data_query(mac_list: List[str]) -> List[dict]: + result = [] + for mac in mac_list: + try: + # The main function used here is read_data + this_result = asyncio.run(read_data(mac)) + except Exception as e: + log.error(f"Error reading BLE data - {e}") + this_result = None + if "org.freedesktop.DBus.Error.AccessDenied" in e: + reset_adapter() + return result + if this_result: + result.append( + {"name": mac, "temp": this_result.temperature, "humidity": this_result.humidity} + ) + log.info(f"Data query complete - returning result") + return result + +# This non-async helper function scans for MACs +# Primary function to interact with this driver function +def scan_macs() -> List[str]: + log.info("Scanning for MACs:") + new_macs = asyncio.run(_scan_once()) + log.info(f"Found {len(new_macs)} new MACs.") + return new_macs + +# Scans nearby BT devices, return a list of detected MACs that pass the filter async def _scan_once() -> List[str]: - """ - Scan once for nearby BLE devices and return the MAC addresses of those - that satisfy *filter_fn*. - - Parameters - ---------- - filter_fn : Callable[[address, name], bool] | None - If supplied, a predicate that receives the device address and name - and returns ``True`` when the device should be kept. - When ``None`` all discovered addresses are returned. - - Returns - ------- - List[str] - A list of MAC addresses (upper-case as returned by Bleak). - """ log.info("Starting one-shot BLE scan (timeout = %.1fs)", SCAN_TIMEOUT) try: devices = await BleakScanner.discover(timeout=SCAN_TIMEOUT) @@ -131,26 +163,30 @@ async def _scan_once() -> List[str]: this_result = None if "org.freedesktop.DBus.Error.AccessDenied" in e: return [] - + # only return MACs for devices that pass the _is_match return [dev.address for dev in devices if _is_match(dev.address, dev.name)] -# this function filters MACs +# this function filters MACs for octets and model def _is_match(mac: str, name: str) -> bool: - """Return True if the MAC starts with the configured prefix and is correct model.""" if name == "LYWSD03MMC": return mac.upper().startswith(MAC_PREFIX.upper()) -# this helper function scans for MACs -def scan_macs(): - log.info("Scanning for MACs:") - new_macs = asyncio.run(_scan_once()) - - log.info(f"Found {len(new_macs)} new MACs.") - return new_macs +# This resets the BT adapter when things go caca +def reset_adapter(): + log.info("Restting BT Adapter...") + if len(current_usb_id) > 0: + log.warning(f"Doing big reset - USB ID {current_usb_id}") + log.info(subprocess.run(["usbreset", current_usb_id], capture_output=True, text=True, check=True).stdout.strip()) + time.sleep(2) + else: + log.warning("Hmm, this shouldn't happen, the current_usb_id is blank 0_o") + log.info("Reset complete. ") -# --------------------------------------------------------------------------- # -# Core async read functions (unchanged from the refactored version) -# --------------------------------------------------------------------------- # +#################################################### +# Core async read functions +# Mined directly from the javascript +# This is the actual function that does the magic +#################################################### async def read_data(mac: str) -> Optional[SensorReading]: try: @@ -165,13 +201,9 @@ async def read_data(mac: str) -> Optional[SensorReading]: if not address: return None - #address = await _discover_device(mac) - #if not address: - # return None - try: async with BleakClient(address, timeout=CONNECT_TIMEOUT) as client: - # <--- NEW BLOCK ---------------------------------------------- + # NEW BLOCK ---------------------------------------------- try: await asyncio.wait_for(client.get_services(), timeout=SERVICE_DISCOVERY_TIMEOUT) @@ -181,9 +213,10 @@ async def read_data(mac: str) -> Optional[SensorReading]: except Exception as e: log.error(f"Discovery failed - {e}") - # --------------------------------------------------------------- + #################################################### # If we get here, services are available. + #################################################### if MI_SERVICE_UUID.lower() not in ( s.uuid.lower() for s in client.services ): @@ -205,11 +238,12 @@ async def read_data(mac: str) -> Optional[SensorReading]: log.exception("Error reading from %s: %s", mac, exc) return None - +#################################################### +# Read all MACs concurrently. Skips any that fail. +# This was LLM written and is NOT TESTED!!! +# It will probably work as advertised though +#################################################### async def read_multiple(mac_list: List[str]) -> List[SensorReading]: - """ - Read all MACs concurrently. Skips any that fail. - """ tasks = [asyncio.create_task(read_data(mac)) for mac in mac_list] results = await asyncio.gather(*tasks, return_exceptions=False) readings = [r for r in results if r is not None] @@ -218,89 +252,71 @@ async def read_multiple(mac_list: List[str]) -> List[SensorReading]: log.warning("Failed to read: %s", ", ".join(failed)) return readings - -# --------------------------------------------------------------------------- # +#################################################### # Sync wrapper (kept for legacy code that cannot run an event loop) -# --------------------------------------------------------------------------- # +# Also LLM Written and not used +#################################################### def read_data_sync(mac: str) -> Optional[SensorReading]: - """ - Convenience wrapper that runs ``read_data`` in a fresh event-loop. - Use this from non-async code. - """ return asyncio.run(read_data(mac)) -# --------------------------------------------------------------------------- # -# Other helper functions -# --------------------------------------------------------------------------- # -def reset_adapter(): - log.info("Restting BT Adapter...") - if len(current_usb_id) > 0: - log.warning(f"Doing big reset - USB ID {current_usb_id}") - log.info(subprocess.run(["usbreset", current_usb_id], capture_output=True, text=True, check=True).stdout.strip()) - time.sleep(2) - else: - log.warning("Hmm, this shouldn't happen, the current_usb_id is blank 0_o") - #big_reset = False - #else: - # log.info(subprocess.run(["sudo", "rfkill", "block", "bluetooth"], capture_output=True, text=True, check=True).stdout.strip()) - # time.sleep(2) - # log.info(subprocess.run(["sudo", "rfkill", "unblock", "bluetooth"], capture_output=True, text=True, check=True).stdout.strip()) - # time.sleep(5) - log.info("Reset complete. ") - -def sensor_data_query(mac_list: List[str]) -> List[dict]: - result = [] - for mac in mac_list: - try: - this_result = asyncio.run(read_data(mac)) - except Exception as e: - log.error(f"Error reading BLE data - {e}") - this_result = None - if "org.freedesktop.DBus.Error.AccessDenied" in e: - reset_adapter() - return result - if this_result: - result.append( - {"name": mac, "temp": this_result.temperature, "humidity": this_result.humidity} - ) - log.info(f"Data query complete - returning result") - return result - #################################################### -# Test Function +# Test Function Section # This runs a single full test of the driver +# This two-step process first runs a single scan +# Then, it scans the list until all devices have been scanned #################################################### -def test(test_count = 0, bt_usb_id = None) -> None: - # first test scanner + +def test_driver(test_count = 0, bt_usb_id = None): + + # if this happens then this needs to fail if bt_usb_id == None: log.error(f"Error - no USB ID Passed, this is needed for the test") return -1 - log.info(f"Ensuring BLE active...") + + #################################################### + # first test - scanner function + # the scanner function scans for compatible devices in range + # this is pretty simple + #################################################### + test_loop_count = 0 found_macs = scan_macs() log.info("Found these MACs on a scan:") log.info(found_macs) - test_result = False + + #################################################### + # second test - scan devices for data + # the primary function to perform this task is sensor_data_query + # that runs a single scan and returns a list of sensor data dicts + # this test runs that task as many times as it takes + # to get all the MACs in the mac_list + #################################################### + + test_complete = False sensors = [] mac_list = [] + # this gets MACs from the comment dict in config for mac in comment_lookup: mac_list.append(mac['mac']) - #[ - # "A4:C1:38:96:53:30", - # "A4:C1:38:11:F4:2A", - # "A4:C1:38:24:15:B0", - # "A4:C1:38:D7:8D:26", - #] log.info("Starting BLE test...") log.info(f" Scanning MACs: {mac_list}") + # test_list shrinks with successful tests test_list = mac_list + # for the history logic passed_list = [] - # History buffer - keep the last N snapshots (N=5) mac_history = [] prev_list_snapshot = None - while not test_result: - # Check if the failed list has been the same for 5 loops + + #################################################### + # here is the big main test loop + # only for second function + #################################################### + + while not test_complete: + + # Check if the failed list has been the same for reset_retries loops + # If so, reset adapter and history if prev_list_snapshot is not None and passed_list == prev_list_snapshot: # List unchanged - add to mac_history mac_history.append(passed_list) @@ -313,14 +329,22 @@ def test(test_count = 0, bt_usb_id = None) -> None: reset_adapter() mac_history = [] prev_list_snapshot = passed_list - + + # Preparing this scan loop log.info(f" Test run: {test_count} - loop {test_loop_count} start. Initializing run...") + + #################################################### + # Here the primary sensor_data_query query is executed + #################################################### test_run_results = sensor_data_query(test_list) + + # This all governs the loop logic and test variable handling test_loop_count = test_loop_count + 1 for test_run_result in test_run_results: sensors.append(test_run_result) passed_list = [device["name"] for device in sensors] log.info(f" Passed list - {passed_list}") + # If all the MACs have been scanned, then this test is complete. if len(mac_list) == len(passed_list): log.info(f" Test run {test_count} completed, yee-haw") for sensor in sensors: @@ -330,21 +354,26 @@ def test(test_count = 0, bt_usb_id = None) -> None: "found_macs": found_macs, "sensor_results": sensors } - test_result = True + test_complete = True return result + # If not, then re-scan the failed sensors else: log.warning(f" Test run incomplete/failed, attempt {len(mac_history)} of {reset_retries}, re-running, loop {test_loop_count}. Updating test list.") for sensor in sensors: + # Sorry I can't feel celcius, I want my freedom degrees temp_f = (sensor['temp'] * 1.80) + 32.0 log.info(f" {sensor['name']} - Temp: {round(temp_f,1)} | Humidity: {sensor['humidity']}") mac_list_new = [] for mac in mac_list: if mac not in passed_list: mac_list_new.append(mac) + # here is where test_list might shrink test_list = mac_list_new log.warning(f" Re-do list - {test_list}") -# helper function to guess the USB ID. It isn't hard + # end of the big main test loop and test_driver function + +# helper function to guess the USB ID. It isn't usually hard def usb_id_guess(): log.warning("USB ID Not found, I'll try and guess...") result = '' @@ -364,12 +393,12 @@ def usb_id_guess(): log.error(f"USB-ID guess failed: {exc}") return result -# --------------------------------------------------------------------------- # -# If run as a script, execute the legacy test -# Since I intend to run this service for months at a time, -# I need to do some robust long term tests on the driver -# This entire main subroutine is dedicated to testing the driver -# --------------------------------------------------------------------------- # +#################################################### +# If ble.py is run as a script, it will execute a test +# Since I intend to run this driver for months at a time, +# I need to do some robust long term tests on it +# This entire main subroutine is dedicated to testing it +#################################################### if __name__ == "__main__": @@ -439,7 +468,7 @@ if __name__ == "__main__": # it's tricky, the test run is called here inside the json this_test_result = { "test_label": test_label, - "test_result": test(test_count = test_count, bt_usb_id = current_usb_id) + "test_result": test_driver(test_count = test_count, bt_usb_id = current_usb_id) } test_results.append(this_test_result)