matt/ble_api
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

change prints to logs

Browse Source
This commit is contained in:
matt committed 2026-10-16 21:36:27 -07:00
1 parent 826538e667
commit 5c00f02c55
1 file changed
+24 -26
Download Patch File Download Diff File Expand all files Collapse all files
api/ble.py
+24 -26
View File
@@ -110,11 +110,11 @@ async def _scan_once() -> List[str]:
List[str] List[str]
A list of MAC addresses (upper-case as returned by Bleak). A list of MAC addresses (upper-case as returned by Bleak).
""" """
log.debug("Starting one-shot BLE scan (timeout = %.1fs)", SCAN_TIMEOUT) log.info("Starting one-shot BLE scan (timeout = %.1fs)", SCAN_TIMEOUT)
try: try:
devices = await BleakScanner.discover(timeout=SCAN_TIMEOUT) devices = await BleakScanner.discover(timeout=SCAN_TIMEOUT)
except Exception as e: except Exception as e:
print(f"Error reading BLE data - {e}") log.error(f"Error reading BLE data - {e}")
this_result = None this_result = None
if "org.freedesktop.DBus.Error.AccessDenied" in e: if "org.freedesktop.DBus.Error.AccessDenied" in e:
return [] return []
@@ -129,10 +129,10 @@ def _is_match(mac: str, name: str) -> bool:
# this helper function scans for MACs # this helper function scans for MACs
def scan_macs(): def scan_macs():
print("Scanning for MACs:") log.info("Scanning for MACs:")
new_macs = asyncio.run(_scan_once()) new_macs = asyncio.run(_scan_once())
print(f"Found {len(new_macs)} new MACs.") log.info(f"Found {len(new_macs)} new MACs.")
return new_macs return new_macs
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
@@ -167,7 +167,7 @@ async def read_data(mac: str) -> Optional[SensorReading]:
return None return None
except Exception as e: except Exception as e:
print(f"Discovery failed - {e}") log.error(f"Discovery failed - {e}")
# --------------------------------------------------------------- # ---------------------------------------------------------------
# If we get here, services are available. # If we get here, services are available.
@@ -234,7 +234,7 @@ def sensor_data_query(mac_list: List[str]) -> List[dict]:
try: try:
this_result = asyncio.run(read_data(mac)) this_result = asyncio.run(read_data(mac))
except Exception as e: except Exception as e:
print(f"Error reading BLE data - {e}") log.error(f"Error reading BLE data - {e}")
this_result = None this_result = None
if "org.freedesktop.DBus.Error.AccessDenied" in e: if "org.freedesktop.DBus.Error.AccessDenied" in e:
reset_adapter() reset_adapter()
@@ -243,7 +243,7 @@ def sensor_data_query(mac_list: List[str]) -> List[dict]:
result.append( result.append(
{"name": mac, "temp": this_result.temperature, "humidity": this_result.humidity} {"name": mac, "temp": this_result.temperature, "humidity": this_result.humidity}
) )
print(f"Data query complete - returning result") log.info(f"Data query complete - returning result")
return result return result
@@ -251,8 +251,8 @@ def test(test_count = 0) -> None:
# first test scanner # first test scanner
test_loop_count = 0 test_loop_count = 0
found_macs = scan_macs() found_macs = scan_macs()
print("Found these MACs on a scan:") log.info("Found these MACs on a scan:")
print(found_macs) log.info(found_macs)
test_result = False test_result = False
sensors = [] sensors = []
mac_list = [ mac_list = [
@@ -261,24 +261,22 @@ def test(test_count = 0) -> None:
"A4:C1:38:24:15:B0", "A4:C1:38:24:15:B0",
"A4:C1:38:D7:8D:26", "A4:C1:38:D7:8D:26",
] ]
print("Starting BLE test...") log.info("Starting BLE test...")
print(" Scanning MACs:") log.info(f" Scanning MACs: {mac_list}")
for mac in mac_list:
print(f" {mac}")
test_list = mac_list test_list = mac_list
while not test_result: while not test_result:
print(f" Test run {test_count} loop {test_loop_count} start. Initializing run...") log.info(f" Test run {test_count} loop {test_loop_count} start. Initializing run...")
test_run_results = sensor_data_query(test_list) test_run_results = sensor_data_query(test_list)
test_loop_count = test_loop_count + 1 test_loop_count = test_loop_count + 1
for test_run_result in test_run_results: for test_run_result in test_run_results:
sensors.append(test_run_result) sensors.append(test_run_result)
passed_list = [device["name"] for device in sensors] passed_list = [device["name"] for device in sensors]
print(f" Passed list - {passed_list}") log.info(f" Passed list - {passed_list}")
if len(mac_list) == len(passed_list): if len(mac_list) == len(passed_list):
print(f" Test run {test_count} completed, yee-haw") log.info(f" Test run {test_count} completed, yee-haw")
for sensor in sensors: for sensor in sensors:
temp_f = (sensor['temp'] * 1.80) + 32.0 temp_f = (sensor['temp'] * 1.80) + 32.0
print(f" {sensor['name']} - Temp: {round(temp_f,1)} | Humidity: {sensor['humidity']}") log.info(f" {sensor['name']} - Temp: {round(temp_f,1)} | Humidity: {sensor['humidity']}")
result = { result = {
"found_macs": found_macs, "found_macs": found_macs,
"sensor_results": sensors "sensor_results": sensors
@@ -286,16 +284,16 @@ def test(test_count = 0) -> None:
test_result = True test_result = True
return result return result
else: else:
print(f" Test run incomplete/failed, re-running, loop {test_loop_count}. Updating test list") log.warning(f" Test run incomplete/failed, re-running, loop {test_loop_count}. Updating test list")
for sensor in sensors: for sensor in sensors:
temp_f = (sensor['temp'] * 1.80) + 32.0 temp_f = (sensor['temp'] * 1.80) + 32.0
print(f" {sensor['name']} - Temp: {round(temp_f,1)} | Humidity: {sensor['humidity']}") log.warning(f" {sensor['name']} - Temp: {round(temp_f,1)} | Humidity: {sensor['humidity']}")
mac_list_new = [] mac_list_new = []
for mac in mac_list: for mac in mac_list:
if mac not in passed_list: if mac not in passed_list:
mac_list_new.append(mac) mac_list_new.append(mac)
test_list = mac_list_new test_list = mac_list_new
print(f" Re-do list - {test_list}") log.warning(f" Re-do list - {test_list}")
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
@@ -317,11 +315,11 @@ if __name__ == "__main__":
test_count = 0 test_count = 0
test_results = [] test_results = []
print(f"Starting BLE Test for {count} iterations") log.info(f"Starting BLE Test for {count} iterations")
while test_count < count: while test_count < count:
test_count = test_count + 1 test_count = test_count + 1
this_test_result = [] this_test_result = []
print(f"Starting test run {test_count} of {count}") log.info(f"Starting test run {test_count} of {count}")
try: try:
test_label = f"Test {test_count} result" test_label = f"Test {test_count} result"
this_test_result = { this_test_result = {
@@ -329,14 +327,14 @@ if __name__ == "__main__":
"test_result": test(test_count) "test_result": test(test_count)
} }
except Exception as e: except Exception as e:
print(f"Test failed brah - {e}") log.error(f"Test failed brah - {e}")
if "org.freedesktop.DBus.Error.AccessDenied" in e: if "org.freedesktop.DBus.Error.AccessDenied" in e:
reset_adapter() reset_adapter()
test_results.append(this_test_result) test_results.append(this_test_result)
print("Test run completed") log.info("Test run completed")
for test_result in test_results: for test_result in test_results:
print(f"Test {test_result['test_label']}:") log.info(f"Test {test_result['test_label']}:")
parts: List[str] = [] parts: List[str] = []
for reading in test_result['test_result']: for reading in test_result['test_result']:
mac = reading.get("name", "") mac = reading.get("name", "")
@@ -346,7 +344,7 @@ if __name__ == "__main__":
short_mac = ":".join(octets) short_mac = ":".join(octets)
parts.append(f"{short_mac} - {temp:.1f}") parts.append(f"{short_mac} - {temp:.1f}")
result_summary = " | ".join(parts) result_summary = " | ".join(parts)
print(result_summary) log.info(result_summary)
# Insert before the main guard # Insert before the main guard