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

ble - move some logic into separate functions to clean things up

Browse Source
This commit is contained in:
matt committed 2026-10-20 04:53:31 -07:00
1 parent 1bd8b95b28
commit f8276f22eb
1 file changed
+107 -99
Download Patch File Download Diff File Expand all files Collapse all files
api/ble.py
+107 -99
View File
@@ -12,17 +12,13 @@ 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:
This module 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.
@@ -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,8 +318,24 @@ 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
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.")
print_sensor_data(sensors)
# here is where test_list might shrink
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
####################################################
# 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,
@@ -368,24 +343,62 @@ def test_driver(test_count = 0, bt_usb_id = None):
}
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']}")
# 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)
# here is where test_list might shrink
test_list = mac_list_new
log.warning(f" Re-do list - {test_list}")
return mac_list_new
# end of the big main test loop and test_driver function
# 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']}")
# helper function to guess the USB ID. It isn't usually hard
# 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
# [