ble.py - tweaking some vars and log messages

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 = [
+    mac_list = []
-        "A4:C1:38:96:53:30",
+    for mac in comment_lookup:
-        "A4:C1:38:11:F4:2A",
+        mac_list.append(mac['mac'])
-        "A4:C1:38:24:15:B0",
+    #[
-        "A4:C1:38:D7:8D:26",
+    #    "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):
+        if len(mac_history) == reset_retries and all(lst == mac_history[0] for lst in mac_history):
-            log.warning(f"Last 5 tests have been identical, resetting adapter")
+            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 <count>")
         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
 #            [
 #                {

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 = [
     {