From f8276f22eb7d8cc5b89ce620c1f90f74002741fb Mon Sep 17 00:00:00 2001 From: matt Date: Tue, 20 Oct 2026 04:53:31 -0700 Subject: [PATCH] ble - move some logic into separate functions to clean things up --- api/ble.py | 230 +++++++++++++++++++++++++++-------------------------- 1 file changed, 119 insertions(+), 111 deletions(-) diff --git a/api/ble.py b/api/ble.py index d83c3b1..fc5688a 100644 --- a/api/ble.py +++ b/api/ble.py @@ -12,20 +12,16 @@ 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]: - A lot of this came from LLM assistance, including the next blurb + A lot of this came from LLM assistance, including the next edited 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. + Async BLE reader for Xiaomi Mi temperature / humidity sensors. + + This module exposes a minimal public API that other apps can import: + + * `read_data` - async coroutine, returns SensorReading or None + + All heavy lifting is done in the async functions; the sync wrapper is only + provided for convenience. """ #################################################### @@ -83,6 +79,9 @@ class SensorReading: temperature: float humidity: int +######################################################################################################## +### BLE Driver functions +######################################################################################################## #################################################### # BLE Helpers mined from the javascript @@ -176,24 +175,6 @@ def _is_match(mac: str, name: str) -> bool: if name == "LYWSD03MMC": return mac.upper().startswith(MAC_PREFIX.upper()) -# 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. ") - -# this restarts the bluetooth service -def restart_bt_service(): - log.info("Resetting bluetooth.service...") - log.info(subprocess.run(["systemctl", "stop", "bluetooth.service"], capture_output=True, text=True, check=True).stdout.strip()) - time.sleep(2) - log.info(subprocess.run(["systemctl", "start", "bluetooth.service"], capture_output=True, text=True, check=True).stdout.strip()) - #################################################### # Core async read functions # Mined directly from the javascript @@ -250,34 +231,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]: - 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] - failed = [mac for mac, r in zip(mac_list, results) if r is None] - if failed: - 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]: - return asyncio.run(read_data(mac)) - - -#################################################### -# 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 -#################################################### +######################################################################################################## +### 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_driver(test_count = 0, bt_usb_id = None): @@ -359,33 +318,87 @@ def test_driver(test_count = 0, bt_usb_id = None): # 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: - temp_f = (sensor['temp'] * 1.80) + 32.0 - log.info(f" {sensor['name']} - Temp: {round(temp_f,1)} | Humidity: {sensor['humidity']}") - result = { - "found_macs": found_macs, - "sensor_results": sensors - } - test_complete = True - return result + print_sensor_data(sensors) + return summarize_test_run(sensors) # 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) + print_sensor_data(sensors) # here is where test_list might shrink - test_list = mac_list_new + test_list = get_failed_macs(sensors, mac_list, passed_list) log.warning(f" Re-do list - {test_list}") # end of the big main test loop and test_driver function -# helper function to guess the USB ID. It isn't usually hard +#################################################### +# Various other helper functions +#################################################### + +# helper function for summarizing test run results +def summarize_test_run(sensors): + log.info(f" {sensor['name']} - Temp: {round(temp_f,1)} | Humidity: {sensor['humidity']}") + result = { + "found_macs": found_macs, + "sensor_results": sensors + } + test_complete = True + return result + +# helper function for processing failed tests +def get_failed_macs(sensors, mac_list, passed_list): + mac_list_new = [] + for mac in mac_list: + if mac not in passed_list: + mac_list_new.append(mac) + return mac_list_new + +# helper to print sensor data +def print_sensor_data(sensors): + 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']}") + +# 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. ") + +# this restarts the bluetooth service +def restart_bt_service(): + log.info("Resetting bluetooth.service...") + log.info(subprocess.run(["systemctl", "stop", "bluetooth.service"], capture_output=True, text=True, check=True).stdout.strip()) + time.sleep(2) + log.info(subprocess.run(["systemctl", "start", "bluetooth.service"], capture_output=True, text=True, check=True).stdout.strip()) + +# this will return the USB ID based on systems scans +def get_usb_id(): + present_usb_ids_command = "lsusb | awk '{print $6}'" + # Capture each USB ID as a separate line + present_usb_id_output = subprocess.run( + present_usb_ids_command, + shell=True, + capture_output=True, + text=True, + check=True, + ).stdout.splitlines() + result = '' + # first let's see if it's already in the list + for usb_id in present_usb_id_output: + if usb_id in bt_usb_ids: + current_usb_id = usb_id + # if it's not in the list, guess + if len(current_usb_id) == 0: + current_usb_id = usb_id_guess() + return current_usb_id + +# this will guess the USB ID if it's not found in the list def usb_id_guess(): log.warning("USB ID Not found, I'll try and guess...") result = '' @@ -405,6 +418,29 @@ def usb_id_guess(): log.error(f"USB-ID guess failed: {exc}") return result +# helper function to render BT device scan results for entire test +def summarize_device_scan(test_result): + mac_parts: List[str] = [] + for mac in sorted(test_result['test_result']['found_macs']): + short_mac = ":".join(mac.split(":")[-2:]) + mac_parts.append(short_mac) + found_macs_summary = " | ".join(mac_parts) + result = f"Found {len(test_result['test_result']['found_macs'])} MACs - {found_macs_summary}" + return result + +# helper function to summarize the full test results +def summarize_test_results(test_result): + parts: List[str] = [] + for reading in sorted(test_result['test_result']['sensor_results'], key=lambda r: r['name']): + mac = reading['name'] + temp = reading['temp'] + # Grab the last two octets of the MAC address + octets = mac.split(":")[-2:] + short_mac = ":".join(octets) + parts.append(f"{short_mac} - {temp:.1f}") + result_summary = " | ".join(parts) + return result_summary + #################################################### # 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, @@ -444,24 +480,10 @@ if __name__ == "__main__": time.sleep(2) # We need the usb id for all the resetting this has to do - present_usb_ids_command = "lsusb | awk '{print $6}'" - # Capture each USB ID as a separate line - present_usb_id_output = subprocess.run( - present_usb_ids_command, - shell=True, - capture_output=True, - text=True, - check=True, - ).stdout.splitlines() - current_usb_id = '' - # first let's see if it's already in the list - for usb_id in present_usb_id_output: - if usb_id in bt_usb_ids: - current_usb_id = usb_id - # if it's not in the list, guess - if len(current_usb_id) == 0: - current_usb_id = usb_id_guess() + current_usb_id = get_usb_id() log.info(f"BLE USB ID Detected: {current_usb_id} - Resetting") + + # one last adapter reset before the test reset_adapter() #################################################### @@ -490,23 +512,9 @@ if __name__ == "__main__": # Which scan is this log.info(f"{test_result['test_label']}:") # Show summary of MACs found in this scan - mac_parts: List[str] = [] - for mac in sorted(test_result['test_result']['found_macs']): - short_mac = ":".join(mac.split(":")[-2:]) - mac_parts.append(short_mac) - found_macs_summary = " | ".join(mac_parts) - log.info(f"Found {len(test_result['test_result']['found_macs'])} MACs - {found_macs_summary}") - parts: List[str] = [] + log.info(summarize_device_scan(test_result)) # Show summary of Sensor data in this scan - for reading in sorted(test_result['test_result']['sensor_results'], key=lambda r: r['name']): - mac = reading['name'] - temp = reading['temp'] - # Grab the last two octets of the MAC address - octets = mac.split(":")[-2:] - short_mac = ":".join(octets) - parts.append(f"{short_mac} - {temp:.1f}") - result_summary = " | ".join(parts) - log.info(f" {result_summary}") + log.info(summarize_test_results(test_result)) # This is the structure of test_results # [