Run BLE autoresearch (--ble). 1. Send startBle RPC over serial -> device starts BLE GATT server + advertising 2. Scan for "FastLED-C6" via Bleak, connect 3. Send ping JSON-RPC over BLE (write to RX characteristic) 4. Receive pong via NOTIFY or READ on TX characteristic -> PASS 5
(
upload_port: str,
serial_iface: "SerialInterface | None",
timeout: float = 60.0,
)
| 26 | |
| 27 | |
| 28 | async def run_ble_autoresearch( |
| 29 | upload_port: str, |
| 30 | serial_iface: "SerialInterface | None", |
| 31 | timeout: float = 60.0, |
| 32 | ) -> int: |
| 33 | """Run BLE autoresearch (--ble). |
| 34 | |
| 35 | 1. Send startBle RPC over serial -> device starts BLE GATT server + advertising |
| 36 | 2. Scan for "FastLED-C6" via Bleak, connect |
| 37 | 3. Send ping JSON-RPC over BLE (write to RX characteristic) |
| 38 | 4. Receive pong via NOTIFY or READ on TX characteristic -> PASS |
| 39 | 5. Send stopBle RPC over serial -> device stops BLE |
| 40 | |
| 41 | Args: |
| 42 | upload_port: Serial port for RPC communication |
| 43 | serial_iface: Pre-created serial interface |
| 44 | timeout: RPC timeout in seconds |
| 45 | |
| 46 | Returns: |
| 47 | Exit code (0 = success, 1 = failure) |
| 48 | """ |
| 49 | print() |
| 50 | print("=" * 60) |
| 51 | print("BLE AUTORESEARCH MODE") |
| 52 | print("=" * 60) |
| 53 | print() |
| 54 | |
| 55 | serial_client: RpcClient | None = None |
| 56 | ble_iface: BleInterface | None = None |
| 57 | |
| 58 | try: |
| 59 | # Step 1: Connect to device via serial RPC |
| 60 | print(f"--- Step 1: Connect to device on {upload_port} (serial) ---") |
| 61 | serial_client = RpcClient( |
| 62 | upload_port, timeout=timeout, serial_interface=serial_iface |
| 63 | ) |
| 64 | await serial_client.connect(boot_wait=3.0, drain_boot=True) |
| 65 | print(f" {Fore.GREEN}Connected to device (serial){Style.RESET_ALL}") |
| 66 | |
| 67 | # Step 2: Start BLE on device via serial RPC |
| 68 | print("\n--- Step 2: Start BLE GATT server on ESP32 ---") |
| 69 | response = await serial_client.send("startBle", timeout=30.0) |
| 70 | ble_info = response.data |
| 71 | |
| 72 | if not isinstance(ble_info, dict) or not ble_info.get("success"): |
| 73 | error = ( |
| 74 | ble_info.get("error", "Unknown error") |
| 75 | if isinstance(ble_info, dict) |
| 76 | else str(ble_info) |
| 77 | ) |
| 78 | print(f" {Fore.RED}Failed to start BLE: {error}{Style.RESET_ALL}") |
| 79 | return 1 |
| 80 | |
| 81 | device_name = ble_info.get("device_name", "FastLED-C6") |
| 82 | print( |
| 83 | f" {Fore.GREEN}BLE started: name={device_name}, " |
| 84 | f"service={ble_info.get('service_uuid', '?')}{Style.RESET_ALL}" |
| 85 | ) |
no test coverage detected