lots of stuff for bigger bt reset

This commit is contained in:
matt committed 2026-10-17 08:52:29 -07:00
1 parent 61da567070
commit c4e94fce3d
2 files changed
+38 -9

No files matched your search

+33 -8
View File
@@ -39,6 +39,7 @@ 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
@@ -89,6 +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)
return None
def _parse_sensor_data(raw: bytes) -> (float, int):
@@ -231,12 +233,17 @@ def read_data_sync(mac: str) -> Optional[SensorReading]:
# --------------------------------------------------------------------------- #
# Other helper functions
# --------------------------------------------------------------------------- #
def reset_adapter():
def reset_adapter(big_reset = False):
log.info("Restting BT Adapter...")
subprocess.run(["sudo", "rfkill", "block", "bluetooth"], capture_output=True, text=True, check=True)
time.sleep(2)
subprocess.run(["sudo", "rfkill", "unblock", "bluetooth"], capture_output=True, text=True, check=True)
time.sleep(5)
if big_reset:
log.wwarning("Doing big reset:")
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.info("Reset complete. ")
def sensor_data_query(mac_list: List[str]) -> List[dict]:
@@ -259,8 +266,12 @@ def sensor_data_query(mac_list: List[str]) -> List[dict]:
return result
def test(test_count = 0) -> None:
def test(test_count = 0, bt_usb_id = None) -> None:
# first test scanner
if bt_usb_id = None:
log.error(f"Error - no USB ID Passed, this is needed for the test")
return -1
test_loop_count = 0
found_macs = scan_macs()
log.info("Found these MACs on a scan:")
@@ -328,6 +339,20 @@ if __name__ == "__main__":
print("The count must be an integer.")
sys.exit(1)
# calculate the BT ID and pass to test
present_usb_ids_command = "lsusb | awk '{print $6}'"
present_usb_id_output = subprocess.run(
present_usb_ids_command,
shell=True,
capture_output=True,
text=True,
check=True,
).stdout.strip()
current_usb_id = ''
for each id as present_usb_id_output:
if id in bt_usb_ids:
current_usb_id = id
log.info(f"BLE USB ID Detected: {current_usb_id}")
test_count = 0
test_results = []
log.info(f"Starting BLE Test for {count} iterations")
@@ -339,14 +364,14 @@ if __name__ == "__main__":
test_label = f"Test {test_count} result"
this_test_result = {
"test_label": test_label,
"test_result": test(test_count)
"test_result": test(test_count = test_count, bt_usb_id = current_usb_id)
}
except Exception as e:
log.error(f"Test failed brah - {e}")
if "org.freedesktop.DBus.Error.AccessDenied" in e:
reset_adapter()
test_results.append(this_test_result)
# Show test results
log.info("Test run completed! Summary:")
for test_result in test_results:
log.info(f"{test_result['test_label']}:")
+5 -1
View File
@@ -7,6 +7,9 @@ bind_ip = "0.0.0.0"
api_port = 5001
redis_port = 6379
enable_redis = True
bt_usb_ids = [
"0bda:c821"
]
# --- This will store friendly names for MAC addresses -----------------
comment_lookup = [
@@ -26,4 +29,5 @@ comment_lookup = [
"mac": "A4:C1:38:96:53:30",
"comment": "Bedroom 3"
}
]
]