From de616a5229b94372484b1b508afd5a1dfaa6af4c Mon Sep 17 00:00:00 2001 From: matt Date: Sun, 18 Oct 2026 05:44:44 -0700 Subject: [PATCH] ble.py - tweaking some vars and log messages --- api/ble.py | 65 +++++++++++++++++++++++++++++++++++---------------- api/config.py | 3 +++ 2 files changed, 48 insertions(+), 20 deletions(-) diff --git a/api/ble.py b/api/ble.py index 8f48b04..5c30251 100644 --- a/api/ble.py +++ b/api/ble.py @@ -267,7 +267,10 @@ def sensor_data_query(mac_list: List[str]) -> List[dict]: log.info(f"Data query complete - returning result") return result - +#################################################### +# Test Function +# This runs a single full test of the driver +#################################################### def test(test_count = 0, bt_usb_id = None) -> None: # first test scanner if bt_usb_id == None: @@ -280,12 +283,15 @@ def test(test_count = 0, bt_usb_id = None) -> None: log.info(found_macs) test_result = False sensors = [] - mac_list = [ - "A4:C1:38:96:53:30", - "A4:C1:38:11:F4:2A", - "A4:C1:38:24:15:B0", - "A4:C1:38:D7:8D:26", - ] + mac_list = [] + 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 = mac_list @@ -302,8 +308,8 @@ def test(test_count = 0, bt_usb_id = None) -> None: # List changed - reset mac_history mac_history = [] mac_history.append(passed_list) - if len(mac_history) == 5 and all(lst == mac_history[0] for lst in mac_history): - log.warning(f"Last 5 tests have been identical, resetting adapter") + 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 @@ -327,7 +333,7 @@ def test(test_count = 0, bt_usb_id = None) -> None: test_result = True return result else: - log.warning(f" Test run incomplete/failed, re-running, loop {test_loop_count}. Updating test list") + 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: temp_f = (sensor['temp'] * 1.80) + 32.0 log.info(f" {sensor['name']} - Temp: {round(temp_f,1)} | Humidity: {sensor['humidity']}") @@ -338,6 +344,7 @@ def test(test_count = 0, bt_usb_id = None) -> None: test_list = mac_list_new log.warning(f" Re-do list - {test_list}") +# helper function to guess the USB ID. It isn't hard def usb_id_guess(): log.warning("USB ID Not found, I'll try and guess...") result = '' @@ -355,16 +362,22 @@ def usb_id_guess(): log.info(f"USB Best guess - {result}") except subprocess.CalledProcessError as exc: 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 __name__ == "__main__": - # if running a test, make a log file in temp + + #################################################### + # 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" @@ -373,7 +386,8 @@ if __name__ == "__main__": _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 is passed + + # Make sure the test count value is passed if len(sys.argv) < 2: print("Usage: python3 test.py ") sys.exit(1) @@ -382,11 +396,13 @@ if __name__ == "__main__": 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) - # calculate the BT ID and pass to test + + # 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( @@ -397,15 +413,22 @@ if __name__ == "__main__": 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() log.info(f"BLE USB ID Detected: {current_usb_id} - Resetting") reset_adapter() + + #################################################### + # Beginning test + #################################################### + + # run the test test_count = 0 - reset_count = 0 test_results = [] log.info(f"Starting BLE Test for {count} iterations") while test_count < count: @@ -413,23 +436,27 @@ if __name__ == "__main__": 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(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 mac_parts: List[str] = [] - for mac in sorted(test_result['test_result']['found_macs']): # optional sort + 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) # e.g. "A1:23 | B4:56" + found_macs_summary = " | ".join(mac_parts) log.info(f"Found {len(test_result['test_result']['found_macs'])} MACs - {found_macs_summary}") parts: List[str] = [] - # Sort sensor results alphabetically by 'name' before printing + # 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'] @@ -440,8 +467,6 @@ if __name__ == "__main__": result_summary = " | ".join(parts) log.info(f" {result_summary}") - - # This is the structure of test_results # [ # { diff --git a/api/config.py b/api/config.py index 4897db5..3146345 100644 --- a/api/config.py +++ b/api/config.py @@ -11,6 +11,9 @@ bt_usb_ids = [ "0bda:c821" ] +# test settings +reset_retries = 5 + # --- This will store friendly names for MAC addresses ----------------- comment_lookup = [ {