Merge pull request 'fix: remove outdated ble_client.py, add test_ota_notify.py to tools' (#22) from fix/cleanup-outdated-scripts into main

scripts/ble_client.py

@@ -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())

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)