575 lines
21 KiB
Python
575 lines
21 KiB
Python
#!/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 "org.freedesktop.DBus.Error.AccessDenied" in 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 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
|
|
|
|
try:
|
|
async with BleakClient(address, timeout=CONNECT_TIMEOUT) as client:
|
|
# NEW BLOCK ----------------------------------------------
|
|
try:
|
|
await asyncio.wait_for(client.get_services(),
|
|
timeout=SERVICE_DISCOVERY_TIMEOUT)
|
|
except asyncio.TimeoutError:
|
|
log.warning("Service discovery timed out for %s", mac)
|
|
return None
|
|
|
|
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
|
|
):
|
|
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 asyncio.TimeoutError:
|
|
# This is the *connect* timeout (BleakClient)
|
|
log.error("Connection to %s timed out after %.1fs", mac, CONNECT_TIMEOUT)
|
|
except Exception as exc:
|
|
log.exception("Error reading from %s: %s", mac, exc)
|
|
return None
|
|
|
|
########################################################################################################
|
|
### 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'])
|
|
mac_list = [
|
|
"A4:C1:38:11:F4:2A",
|
|
"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 = []
|
|
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")
|
|
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")
|
|
print("Printing sensor results: ")
|
|
print(sensors)
|
|
print_sensor_data(sensors)
|
|
test_complete = True
|
|
return summarize_test_run(sensors, found_macs)
|
|
# 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(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, found_macs):
|
|
sensor_count = 1
|
|
for sensor in sensors:
|
|
print(f"Sensor count: {sensor_count}")
|
|
log.info(f" {sensor['name']} - Temp: {temp_ctof(sensor['temp'])} | Humidity: {sensor['humidity']}")
|
|
sensor_count = sensor_count + 1
|
|
result = {
|
|
"found_macs": found_macs,
|
|
"sensor_results": sensors
|
|
}
|
|
return result
|
|
|
|
# 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(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 = ''
|
|
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 <count>")
|
|
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()
|
|
|
|
####################################################
|
|
# 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}")
|
|
test_label = f"Test {test_count} result"
|
|
# it's tricky, the test run is called here inside the json
|
|
this_test_result = {
|
|
"test_label": test_label,
|
|
"test_result": test_driver(test_count = test_count, bt_usb_id = current_usb_id)
|
|
}
|
|
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
|
|
# }
|
|
# ]
|
|
# }
|
|
# }
|
|
# ]
|
|
|
|
|