diff --git a/scripts/ble_client.py b/scripts/ble_client.py new file mode 100644 index 0000000..8d42fdd --- /dev/null +++ b/scripts/ble_client.py @@ -0,0 +1,87 @@ +#!/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 deleted file mode 100644 index 19b2308..0000000 --- a/tools/test_ota_notify.py +++ /dev/null @@ -1,123 +0,0 @@ -#!/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)