#!/usr/bin/env python3 #################################################### """ 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 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 edited blurb: 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. """ #################################################### import asyncio import subprocess import time import warnings, logging, sys warnings.filterwarnings("ignore") logging.getLogger('apscheduler').setLevel(logging.ERROR) from dataclasses import dataclass 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" SCAN_TIMEOUT = 10.0 # seconds to scan for each MAC CONNECT_TIMEOUT = 20.0 READ_TIMEOUT = 5.0 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) # Stream (stdout) handler - always present _stdout_handler = logging.StreamHandler(sys.stdout) _stdout_handler.setFormatter( logging.Formatter( fmt="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S", ) ) _logger.addHandler(_stdout_handler) # finally create the log log = _logger #################################################### # Data model #################################################### @dataclass class SensorReading: """Single Mi sensor reading.""" mac: str temperature: float humidity: int ######################################################################################################## ### BLE Driver functions ######################################################################################################## #################################################### # 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]: try: devices = await asyncio.wait_for( BleakScanner.discover(timeout=SCAN_TIMEOUT), timeout=SCAN_TIMEOUT + 2.0, # give a small buffer over scan timeout ) for d in devices: if d.address.lower() == mac.lower(): log.info("Found device %s (%s)", d.name or d.address, d.address) return d.address log.warning("Device %s not found in scan", mac) except asyncio.TimeoutError: log.warning("Discover call timed out after %.1fs for %s", SCAN_TIMEOUT + 2.0, mac) except Exception as e: log.error(f"Dicvoere call failed - {e}") reset_adapter() if "AccessDenied" in str(e): restart_bt_service() 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): sign = raw[1] & (1 << 7) temp = ((raw[1] & 0x7F) << 8 | raw[0]) if sign: temp -= 32767 temp = temp / 100.0 hum = raw[2] return temp, hum #################################################### # 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 str(e): reset_adapter() # return result #if "argument of type 'BleakDBusError' is not iterable" in str(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]: log.info("Starting one-shot BLE scan (timeout = %.1fs)", SCAN_TIMEOUT) try: devices = await BleakScanner.discover(timeout=SCAN_TIMEOUT) except Exception as e: log.error(f"Error reading BLE data - {e}") this_result = None if "org.freedesktop.DBus.Error.AccessDenied" in str(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 for octets and model def _is_match(mac: str, name: str) -> bool: if name == "LYWSD03MMC": return mac.upper().startswith(MAC_PREFIX.upper()) #################################################### # 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: address = await asyncio.wait_for( _discover_device(mac), timeout=DISCOVER_TIMEOUT, ) except asyncio.TimeoutError: log.warning("Device discovery timed out for %s after %.1fs", mac, DISCOVER_TIMEOUT) return None if not address: return None client: BleakClient | None = None try: client = BleakClient(address, timeout=CONNECT_TIMEOUT) await client.connect() # service discovery await asyncio.wait_for(client.get_services(), timeout=SERVICE_DISCOVERY_TIMEOUT) if MI_SERVICE_UUID.lower() not in (s.uuid.lower() for s in client.services): log.error("Mi service not found on %s", mac) return None raw = await asyncio.wait_for(client.read_gatt_char(TEMP_CHAR_UUID), timeout=READ_TIMEOUT) temp, hum = _parse_sensor_data(raw) log.info("Read from %s: %.1f°C, %d%%", mac, temp, hum) return SensorReading(mac=mac, temperature=temp, humidity=hum) except Exception as exc: log.exception("Error reading from %s: %s", mac, exc) return None finally: # Safe disconnect – swallow any EOFError caused by a dropped DBus socket if client is not None: try: await client.disconnect() except Exception as e: # Usually EOFError, just log and ignore log.debug("Disconnect error (ignored): %s", e) ######################################################################################################## ### 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): # 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 #################################################### # 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) #################################################### # 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']) 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 = [] mac_history = [] prev_list_snapshot = None #################################################### # here is the big main test loop # only for second function # it loops until all MACs in mac_list are found #################################################### 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) else: # List changed - reset mac_history mac_history = [] mac_history.append(passed_list) if len(mac_history) == reset_retries and all(lst == mac_history[0] for lst in mac_history): log.warning(f"Last {reset_retries} tests have been identical, resetting adapter") rfkill_bt() 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") print_sensor_data(sensors) test_complete = True result = { "found_macs": found_macs, "sensor_results": sensors, "test_result": "complete" } return result # If not, then re-scan the failed sensors elif test_loop_count > len(mac_list) * 25: log.error(f"Test has reached {len(mac_list) * 25} loops, failing test, rfkill bluetooth, will begin next test run.") rfkill_bt() test_complete = True result = { "found_macs": found_macs, "sensor_results": sensors, "test_result": "failed" } return result else: log.warning(f" Test run incomplete/failed, attempt {len(mac_history) + 1} of {reset_retries}, re-running, loop {test_loop_count}. Updating test list.") print_sensor_data(sensors) # here is where test_list might shrink test_list = get_failed_macs(mac_list, passed_list) log.warning(f" Re-do list - {test_list}") # end of the big main test loop and test_driver function #################################################### # Various other helper functions #################################################### # helper function for processing failed tests def get_failed_macs(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 log.info(f" {sensor['name']} - Temp: {temp_ctof(sensor['temp'])} | Humidity: {sensor['humidity']}") # yes i am an insufferable american that doesn't think in celcius, sue me def temp_ctof(c_temp): f_temp = (c_temp * 1.80) + 32.0 return round(f_temp, 1) # 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(5) else: log.warning("Hmm, this shouldn't happen, the current_usb_id is blank 0_o") log.info("Reset complete. ") # this subroutine will rfkill the bluetooth def rfkill_bt(): log.info(subprocess.run(["rfkill", "block", "bluetooth"], capture_output=True, text=True, check=True).stdout.strip()) time.sleep(5) log.info(subprocess.run(["rfkill", "unblock", "bluetooth"], capture_output=True, text=True, check=True).stdout.strip()) time.sleep(5) # 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()) time.sleep(2) # 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: result = usb_id # if it's not in the list, guess if len(result) == 0: result = usb_id_guess() return result # 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 = '' guess_command = "lsusb | grep -i blue | awk '{print $6'}" try: # Run the command in a shell because of the pipe output = subprocess.run( guess_command, shell=True, capture_output=True, text=True, check=True, ).stdout.strip() result = output log.info(f"USB Best guess - {result}") except subprocess.CalledProcessError as exc: 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, # I need to do some robust long term tests on it # This entire main subroutine is dedicated to testing it #################################################### if __name__ == "__main__": #################################################### # initialize test #################################################### # make a log file in temp import json from datetime import datetime logfile_path = f"/tmp/ble-test-{datetime.now().strftime('%y%m%d-%H%M%S')}.log" _file_handler = logging.FileHandler(logfile_path, mode="a") _file_handler.setLevel(logging.INFO) # same level (or tweak) _file_handler.setFormatter(_stdout_handler.formatter) # reuse formatter _logger.addHandler(_file_handler) log.info(f"Logfile initialized at {logfile_path}") # Make sure the test count value is passed if len(sys.argv) < 2: print("Usage: python3 test.py ") sys.exit(1) try: count = int(sys.argv[1]) except ValueError: print("The count must be an integer.") sys.exit(1) # make sure bluetooth is unblocked in case something weird happened log.info("Unblocking Bluetooth") log.info(subprocess.run(["sudo", "rfkill", "unblock", "bluetooth"], capture_output=True, text=True, check=True).stdout.strip()) time.sleep(2) # We need the usb id for all the resetting this has to do 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() restart_bt_service() #################################################### # Beginning test #################################################### # run the test test_count = 0 test_results = [] log.info(f"Starting BLE Test for {count} iterations") while test_count < count: test_count = test_count + 1 this_test_result = [] log.info(f"Starting test run {test_count} of {count}") # run test_driver here this_test_sensors = test_driver(test_count = test_count, bt_usb_id = current_usb_id) test_label = f"Test {test_count} result - {this_test_sensors['test_result']}" this_test_result = { "test_label": test_label, "test_result": this_test_sensors } test_results.append(this_test_result) # Show test results log.info("Test run completed! Summary:") for test_result in test_results: # Which scan is this log.info(f"{test_result['test_label']}:") # Show summary of MACs found in this scan log.info(summarize_device_scan(test_result)) # Show summary of Sensor data in this scan log.info(summarize_test_results(test_result)) # This is the structure of test_results # [ # { # "test_label": test_label, # "test_result": { # "found_macs": found_macs, # "sensor_results": [ # { # "name": mac, # "temp": this_result.temperature, # "humidity": this_result.humidity # }, # { # "name": mac, # "temp": this_result.temperature, # "humidity": this_result.humidity # } # ] # } # }, # { # "test_label": test_label, # "test_result": { # "found_macs": found_macs, # "sensor_results": [ # { # "name": mac, # "temp": this_result.temperature, # "humidity": this_result.humidity # }, # { # "name": mac, # "temp": this_result.temperature, # "humidity": this_result.humidity # } # ] # } # } # ]