From c36ca10edd1493ecd4ca6b2fccf1faba8b8485a4 Mon Sep 17 00:00:00 2001 From: eddy Date: Tue, 4 Aug 2026 18:33:35 +0200 Subject: [PATCH] fix: remove outdated ble_client.py, add test_ota_notify.py to tools - Remove scripts/ble_client.py (uses old UUIDs, incompatible) - Move test_ota_notify.py to tools/ directory - Make serial port configurable via SYLDA_CLIENT_PORT env var --- scripts/ble_client.py | 87 --------------------------- tools/test_ota_notify.py | 123 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 123 insertions(+), 87 deletions(-) delete mode 100644 scripts/ble_client.py create mode 100644 tools/test_ota_notify.py diff --git a/scripts/ble_client.py b/scripts/ble_client.py deleted file mode 100644 index 8d42fdd..0000000 --- a/scripts/ble_client.py +++ /dev/null @@ -1,87 +0,0 @@ -#!/usr/bin/env python3 -"""BLE client for SylDa_Server — send color commands from laptop.""" - -import asyncio -import sys -from bleak import BleakScanner, BleakClient - -SERVICE_UUID = "0000abcd-0000-1000-8000-00805f9b34fb" -CHAR_UUID = "0000abce-0000-1000-8000-00805f9b34fb" -SERVER_NAME = "SylDa_Server" - -COLORS = { - "r": ("red", b"\x00"), - "g": ("green", b"\x01"), - "y": ("yellow", b"\x02"), - "o": ("orange", b"\x03"), - "b": ("blue", b"\x04"), -} - - -async def find_server(): - print(f"Scanning for {SERVER_NAME}...") - device = await BleakScanner.find_device_by_name(SERVER_NAME, timeout=10) - if not device: - print("Not found.") - return None - print(f"Found: {device.name} ({device.address})") - return device - - -async def send_color(client, data): - await client.write_gatt_char(CHAR_UUID, data, response=True) - print(f"Sent: {COLORS[[k for k, v in COLORS.items() if v[1] == data][0]][0]}") - - -def show_menu(): - print("\n┌─────────────────────┐") - print("│ r = red │") - print("│ g = green │") - print("│ y = yellow │") - print("│ o = orange │") - print("│ b = blue │") - print("│ q = quit │") - print("└─────────────────────┘") - - -async def connect_handler(client): - print(f"Connected: {client.is_connected}") - - -async def disconnect_handler(client): - print("Disconnected from server.") - - -async def main(): - device = await find_server() - if not device: - return - - async with BleakClient( - device, - disconnected_callback=disconnect_handler, - timeout=10, - ) as client: - await connect_handler(client) - - while True: - try: - show_menu() - cmd = await asyncio.get_event_loop().run_in_executor( - None, lambda: input("> ").strip().lower() - ) - except (EOFError, KeyboardInterrupt): - break - - if cmd == "q": - break - elif cmd in COLORS: - await send_color(client, COLORS[cmd][1]) - else: - print(f"Unknown: {cmd}") - - print("Bye.") - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/tools/test_ota_notify.py b/tools/test_ota_notify.py new file mode 100644 index 0000000..19b2308 --- /dev/null +++ b/tools/test_ota_notify.py @@ -0,0 +1,123 @@ +#!/usr/bin/env python3 +"""Test OTA notify flow: writes 0x02 to server's CLIENT_OTA_STATUS via BLE, + then checks client serial for peripheral-mode entry log.""" +import sys, time, asyncio, serial, threading, os +from bleak import BleakScanner, BleakClient + +SERVICE_UUID = "4fafc201-1fb5-459e-8fcc-c5c9c331914b" +STATUS_UUID = "beb5483e-36e1-4688-b7f5-ea07361f26df" +SERVER_NAME = "SylDa_Server" + +CLIENT_PORT = os.environ.get("SYLDA_CLIENT_PORT", "/dev/serial/by-id/usb-Espressif_USB_JTAG_serial_debug_unit_44:1B:F6:F5:D1:E0-if00") +TIMEOUT_S = 25 + +class TestResult: + def __init__(self): + self.pass_marker = False + self.fail_marker = False + self.lines = [] + +def monitor_serial(stop_event, result): + try: + ser = serial.Serial(CLIENT_PORT, 115200, timeout=0.5) + print(f"[TEST] Serial opened: {CLIENT_PORT}") + while not stop_event.is_set(): + try: + data = ser.read(1024) + if data: + text = data.decode("ascii", errors="replace") + for line in text.splitlines(): + if line.strip(): + result.lines.append(line) + if "peripheral" in line.lower(): + result.pass_marker = True + stop_event.set() + print(f"[TEST] PASS: {line}") + if "OTA: received trigger" in line or "entering peripheral" in line: + result.pass_marker = True + stop_event.set() + print(f"[TEST] PASS: {line}") + except serial.SerialException: + result.fail_marker = True + stop_event.set() + print("[TEST] FAIL: serial exception") + except Exception as e: + result.fail_marker = True + stop_event.set() + print(f"[TEST] FAIL: {e}") + finally: + try: ser.close() + except: pass + +async def ble_write_02(result): + print("[TEST] Scanning for SylDa_Server...") + try: + device = await BleakScanner.find_device_by_name(SERVER_NAME, timeout=10) + except Exception as e: + print(f"[TEST] FAIL: BLE scan error: {e}") + result.fail_marker = True + return + + if device is None: + print("[TEST] FAIL: SylDa_Server not found via BLE scan") + result.fail_marker = True + return + + print(f"[TEST] Found: {device.name} ({device.address}), connecting...") + try: + async with BleakClient(device, timeout=10) as client: + print(f"[TEST] Connected, writing 0x02 to {STATUS_UUID}...") + svc = client.services.get_service(SERVICE_UUID) + if svc is None: + print("[TEST] FAIL: service not found") + result.fail_marker = True + return + char = svc.get_characteristic(STATUS_UUID) + if char is None: + print("[TEST] FAIL: CLIENT_OTA_STATUS characteristic not found") + result.fail_marker = True + return + await client.write_gatt_char(char, b'\x02', response=True) + print("[TEST] Wrote 0x02 OK, waiting for client to enter peripheral mode...") + except Exception as e: + print(f"[TEST] FAIL: BLE write error: {e}") + result.fail_marker = True + return + +async def main(): + result = TestResult() + stop = threading.Event() + + serial_thread = threading.Thread(target=monitor_serial, args=(stop, result), daemon=True) + serial_thread.start() + + await asyncio.sleep(2) + + await ble_write_02(result) + + if not stop.is_set(): + print(f"[TEST] Waiting up to {TIMEOUT_S}s for client response...") + stop.wait(TIMEOUT_S) + + stop.set() + serial_thread.join(timeout=3) + + if result.pass_marker: + print("[TEST] ===== PASS =====") + return 0 + elif result.fail_marker: + print("[TEST] ===== FAIL =====") + print("[TEST] Serial lines captured:") + for line in result.lines[-20:]: + print(f" {line}") + return 1 + else: + print("[TEST] ===== INDETERMINATE (timeout, no pass/fail marker) =====") + print("[TEST] Serial lines captured:") + for line in result.lines[-20:]: + print(f" {line}") + return 2 + +if __name__ == "__main__": + rc = asyncio.run(main()) + sys.exit(rc)