diff --git a/api/ble.py b/api/ble.py index 945e3b4..c8b3ad8 100644 --- a/api/ble.py +++ b/api/ble.py @@ -90,7 +90,7 @@ async def _discover_device(mac: str) -> Optional[str]: 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(big_reset = True) + reset_adapter() return None def _parse_sensor_data(raw: bytes) -> (float, int): @@ -233,21 +233,23 @@ def read_data_sync(mac: str) -> Optional[SensorReading]: # --------------------------------------------------------------------------- # # Other helper functions # --------------------------------------------------------------------------- # -def reset_adapter(big_reset = False): +def reset_adapter(): log.info("Restting BT Adapter...") - if big_reset: + if current_usb_id.len() > 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.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.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]: - reset_adapter() result = [] for mac in mac_list: try: @@ -271,7 +273,7 @@ def test(test_count = 0, bt_usb_id = None) -> None: 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...") test_loop_count = 0 found_macs = scan_macs() log.info("Found these MACs on a scan:") @@ -338,7 +340,10 @@ 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 present_usb_ids_command = "lsusb | awk '{print $6}'" # Capture each USB ID as a separate line @@ -353,8 +358,8 @@ if __name__ == "__main__": for usb_id in present_usb_id_output: if usb_id in bt_usb_ids: current_usb_id = usb_id - log.info(f"BLE USB ID Detected: {current_usb_id}") - reset_adapter(big_reset=True) + log.info(f"BLE USB ID Detected: {current_usb_id} - Resetting") + reset_adapter() test_count = 0 test_results = [] log.info(f"Starting BLE Test for {count} iterations") @@ -377,6 +382,12 @@ if __name__ == "__main__": log.info("Test run completed! Summary:") for test_result in test_results: log.info(f"{test_result['test_label']}:") + mac_parts: List[str] = [] + for mac in sorted(test_result['found_macs']): # optional sort + short_mac = ":".join(mac.split(":")[-2:]) + mac_parts.append(short_mac) + found_macs_summary = " | ".join(mac_parts) # e.g. "A1:23 | B4:56" + log.info(f"Found {len(test_result['found_macs'])} MACs - {found_macs_summary}") parts: List[str] = [] # Sort sensor results alphabetically by 'name' before printing for reading in sorted(test_result['test_result']['sensor_results'], key=lambda r: r['name']):