diff --git a/api/ble.py b/api/ble.py index acc9c59..450524d 100644 --- a/api/ble.py +++ b/api/ble.py @@ -110,11 +110,11 @@ async def _scan_once() -> List[str]: List[str] 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: devices = await BleakScanner.discover(timeout=SCAN_TIMEOUT) except Exception as e: - print(f"Error reading BLE data - {e}") + log.error(f"Error reading BLE data - {e}") this_result = None if "org.freedesktop.DBus.Error.AccessDenied" in e: return [] @@ -129,10 +129,10 @@ def _is_match(mac: str, name: str) -> bool: # this helper function scans for MACs def scan_macs(): - print("Scanning for MACs:") + log.info("Scanning for MACs:") 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 # --------------------------------------------------------------------------- # @@ -167,7 +167,7 @@ async def read_data(mac: str) -> Optional[SensorReading]: return None except Exception as e: - print(f"Discovery failed - {e}") + log.error(f"Discovery failed - {e}") # --------------------------------------------------------------- # If we get here, services are available. @@ -234,7 +234,7 @@ def sensor_data_query(mac_list: List[str]) -> List[dict]: try: this_result = asyncio.run(read_data(mac)) except Exception as e: - print(f"Error reading BLE data - {e}") + log.error(f"Error reading BLE data - {e}") this_result = None if "org.freedesktop.DBus.Error.AccessDenied" in e: reset_adapter() @@ -243,7 +243,7 @@ def sensor_data_query(mac_list: List[str]) -> List[dict]: result.append( {"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 @@ -251,8 +251,8 @@ def test(test_count = 0) -> None: # first test scanner test_loop_count = 0 found_macs = scan_macs() - print("Found these MACs on a scan:") - print(found_macs) + log.info("Found these MACs on a scan:") + log.info(found_macs) test_result = False sensors = [] mac_list = [ @@ -261,24 +261,22 @@ def test(test_count = 0) -> None: "A4:C1:38:24:15:B0", "A4:C1:38:D7:8D:26", ] - print("Starting BLE test...") - print(" Scanning MACs:") - for mac in mac_list: - print(f" {mac}") + log.info("Starting BLE test...") + log.info(f" Scanning MACs: {mac_list}") test_list = mac_list 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_loop_count = test_loop_count + 1 for test_run_result in test_run_results: sensors.append(test_run_result) 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): - print(f" Test run {test_count} completed, yee-haw") + log.info(f" Test run {test_count} completed, yee-haw") for sensor in sensors: 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 = { "found_macs": found_macs, "sensor_results": sensors @@ -286,16 +284,16 @@ def test(test_count = 0) -> None: test_result = True return result 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: 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 = [] for mac in mac_list: if mac not in passed_list: mac_list_new.append(mac) 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_results = [] - print(f"Starting BLE Test for {count} iterations") + log.info(f"Starting BLE Test for {count} iterations") while test_count < count: test_count = test_count + 1 this_test_result = [] - print(f"Starting test run {test_count} of {count}") + log.info(f"Starting test run {test_count} of {count}") try: test_label = f"Test {test_count} result" this_test_result = { @@ -329,14 +327,14 @@ if __name__ == "__main__": "test_result": test(test_count) } 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: reset_adapter() test_results.append(this_test_result) - print("Test run completed") + log.info("Test run completed") for test_result in test_results: - print(f"Test {test_result['test_label']}:") + log.info(f"Test {test_result['test_label']}:") parts: List[str] = [] for reading in test_result['test_result']: mac = reading.get("name", "") @@ -346,7 +344,7 @@ if __name__ == "__main__": short_mac = ":".join(octets) parts.append(f"{short_mac} - {temp:.1f}") result_summary = " | ".join(parts) - print(result_summary) + log.info(result_summary) # Insert before the main guard