matt/ble_api
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ble.py - lots of comment maintenance

Browse Source
This commit is contained in:
matt committed 2026-10-18 06:38:14 -07:00
1 parent de616a5229
commit 6564d4941f
1 file changed
+160 -131
Download Patch File Download Diff File Expand all files Collapse all files
api/ble.py
+160 -131
View File
@@ -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())
# 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. ")
log.info(f"Found {len(new_macs)} new MACs.")
return new_macs
# --------------------------------------------------------------------------- #
# 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)
@@ -314,13 +330,21 @@ def test(test_count = 0, bt_usb_id = None) -> None:
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)