feat(devices): Android USB-serial support for Adalight/AmbiLED controllers
Adds end-to-end support for driving USB-connected Adalight / AmbiLED LED controllers from Android TV boxes. Android's security model blocks direct USB access from Python, so writes route through a Kotlin UsbSerialBridge singleton via Chaquopy. Python side: - New SerialTransport Protocol (serial_transport.py) with open / write / flush / close. Desktop uses PySerialTransport (wraps pyserial), Android uses AndroidSerialTransport (wraps the Kotlin bridge). - list_serial_ports() factory returns desktop COM ports on desktop, USB devices on Android — callers don't branch. - URL scheme extended: existing COM3[:baud] and /dev/ttyUSB0[:baud] unchanged; new usb:VID:PID[:serial][@baud] for Android (@ is the baud separator since : is already used between VID and PID). - AdalightClient and SerialDeviceProvider refactored to go through the transport — no more direct pyserial imports in hot paths. - 17 new unit tests cover URL parsing, PySerial transport, factory selection, platform-branching discovery. Full suite 750 passing. Kotlin side: - UsbSerialBridge.kt singleton uses com.hoho.android.usbserial (mik3y) which ships drivers for CH340, CP2102, FTDI, Prolific, and CDC-ACM (Arduino). Exposes listDevices, open, write, close via @JvmStatic for Chaquopy. First open() attempt without permission triggers the system USB permission dialog; next call succeeds once user grants. - usb-serial-for-android is distributed via JitPack — added that repo in settings.gradle.kts and the dependency in app/build.gradle.kts. - AndroidManifest declares uses-feature android.hardware.usb.host (required=false so non-USB-host phones still install). - LedGrabApp.onCreate calls UsbSerialBridge.init(this) so the bridge resolves the UsbManager without needing an Activity ref. Verified: ./gradlew compileDebugKotlin succeeds; off-Android import of android_serial_transport works. Real-hardware smoke test on a TV box with a CH340/CP2102/FTDI adapter still pending. ESP-NOW (espnow_client / espnow_provider) still imports pyserial directly because it needs bidirectional reads — separate refactor to extend the transport with read() if that path ever needs Android USB support.
This commit is contained in:
@@ -7,40 +7,21 @@ from typing import Optional, Tuple
|
||||
import numpy as np
|
||||
|
||||
from ledgrab.core.devices.led_client import DeviceHealth, LEDClient
|
||||
from ledgrab.core.devices.serial_transport import (
|
||||
open_transport,
|
||||
parse_serial_url,
|
||||
port_exists,
|
||||
)
|
||||
from ledgrab.utils import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
DEFAULT_BAUD_RATE = 115200
|
||||
ARDUINO_RESET_DELAY = 2.0 # seconds to wait after opening serial for Arduino bootloader
|
||||
|
||||
|
||||
def parse_adalight_url(url: str) -> Tuple[str, int]:
|
||||
"""Parse an Adalight URL into (port, baud_rate).
|
||||
|
||||
Formats:
|
||||
"COM3" -> ("COM3", 115200)
|
||||
"COM3:230400" -> ("COM3", 230400)
|
||||
"/dev/ttyUSB0" -> ("/dev/ttyUSB0", 115200)
|
||||
"""
|
||||
url = url.strip()
|
||||
if ":" in url and not url.startswith("/"):
|
||||
# Windows COM port with baud: "COM3:230400"
|
||||
parts = url.rsplit(":", 1)
|
||||
try:
|
||||
baud = int(parts[1])
|
||||
return parts[0], baud
|
||||
except ValueError:
|
||||
pass
|
||||
elif ":" in url and url.startswith("/"):
|
||||
# Unix path with baud: "/dev/ttyUSB0:230400"
|
||||
parts = url.rsplit(":", 1)
|
||||
try:
|
||||
baud = int(parts[1])
|
||||
return parts[0], baud
|
||||
except ValueError:
|
||||
pass
|
||||
return url, DEFAULT_BAUD_RATE
|
||||
"""Backwards-compatible alias for :func:`parse_serial_url`."""
|
||||
return parse_serial_url(url)
|
||||
|
||||
|
||||
def _build_adalight_header(led_count: int) -> bytes:
|
||||
@@ -81,13 +62,12 @@ class AdalightClient(LEDClient):
|
||||
|
||||
async def connect(self) -> bool:
|
||||
"""Open serial port and wait for Arduino reset."""
|
||||
import serial
|
||||
|
||||
try:
|
||||
self._serial = await asyncio.to_thread(
|
||||
serial.Serial, port=self._port, baudrate=self._baud_rate, timeout=1
|
||||
)
|
||||
# Wait for Arduino to finish bootloader reset (non-blocking)
|
||||
self._serial = open_transport(self._port, baud_rate=self._baud_rate, timeout=1)
|
||||
await asyncio.to_thread(self._serial.open)
|
||||
# Wait for Arduino to finish bootloader reset (non-blocking).
|
||||
# USB-to-TTL adapters without DTR don't reset, but the delay
|
||||
# is harmless on those — keeps the path uniform.
|
||||
await asyncio.sleep(ARDUINO_RESET_DELAY)
|
||||
self._connected = True
|
||||
logger.info(
|
||||
@@ -122,7 +102,7 @@ class AdalightClient(LEDClient):
|
||||
f"led_count={self._led_count}"
|
||||
)
|
||||
self._connected = False
|
||||
if self._serial and self._serial.is_open:
|
||||
if self._serial is not None:
|
||||
try:
|
||||
self._serial.close()
|
||||
except Exception as e:
|
||||
@@ -182,18 +162,13 @@ class AdalightClient(LEDClient):
|
||||
) -> DeviceHealth:
|
||||
"""Check if the serial port exists without opening it.
|
||||
|
||||
Enumerates COM ports to avoid exclusive-access conflicts on Windows.
|
||||
Enumerates COM ports (or USB devices on Android) to avoid
|
||||
exclusive-access conflicts on Windows.
|
||||
"""
|
||||
port, _baud = parse_adalight_url(url)
|
||||
|
||||
try:
|
||||
import serial.tools.list_ports
|
||||
|
||||
available_ports = [p.device for p in serial.tools.list_ports.comports()]
|
||||
port_upper = port.upper()
|
||||
found = any(p.upper() == port_upper for p in available_ports)
|
||||
|
||||
if found:
|
||||
if port_exists(port):
|
||||
return DeviceHealth(
|
||||
online=True,
|
||||
latency_ms=0.0,
|
||||
@@ -202,12 +177,11 @@ class AdalightClient(LEDClient):
|
||||
device_version=None,
|
||||
device_led_count=prev_health.device_led_count if prev_health else None,
|
||||
)
|
||||
else:
|
||||
return DeviceHealth(
|
||||
online=False,
|
||||
last_checked=datetime.now(timezone.utc),
|
||||
error=f"Serial port {port} not found",
|
||||
)
|
||||
return DeviceHealth(
|
||||
online=False,
|
||||
last_checked=datetime.now(timezone.utc),
|
||||
error=f"Serial port {port} not found",
|
||||
)
|
||||
except Exception as e:
|
||||
return DeviceHealth(
|
||||
online=False,
|
||||
|
||||
@@ -0,0 +1,159 @@
|
||||
"""Android USB-serial transport backed by the Kotlin ``UsbSerialBridge``.
|
||||
|
||||
Calls into Java land through Chaquopy; this module only loads on
|
||||
Android. URL format: ``usb:VID:PID`` or ``usb:VID:PID:serial`` (with an
|
||||
optional ``@baud`` suffix).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Optional, Tuple
|
||||
|
||||
from ledgrab.utils import get_logger
|
||||
from ledgrab.utils.platform import is_android
|
||||
|
||||
from .serial_transport import SerialPortInfo
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
def _bridge():
|
||||
"""Return the singleton Kotlin ``UsbSerialBridge`` instance, or raise.
|
||||
|
||||
The bridge exposes static methods ``listDevices()``, ``open(...)``,
|
||||
``write(...)``, ``close(...)``, etc. — see ``UsbSerialBridge.kt``.
|
||||
"""
|
||||
if not is_android():
|
||||
raise RuntimeError("AndroidSerialTransport is only usable on Android")
|
||||
try:
|
||||
from java import jclass # type: ignore[import-not-found]
|
||||
except ImportError as e:
|
||||
raise RuntimeError("Chaquopy java interop not available") from e
|
||||
|
||||
return jclass("com.ledgrab.android.UsbSerialBridge").INSTANCE
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _UsbAddress:
|
||||
vendor_id: int
|
||||
product_id: int
|
||||
serial: Optional[str]
|
||||
|
||||
@classmethod
|
||||
def parse(cls, device: str) -> "_UsbAddress":
|
||||
if not device.startswith("usb:"):
|
||||
raise ValueError(f"Not a USB device URL: {device!r}")
|
||||
body = device[len("usb:") :]
|
||||
parts = body.split(":")
|
||||
if len(parts) < 2:
|
||||
raise ValueError(
|
||||
f"USB URL must be 'usb:VID:PID' or 'usb:VID:PID:serial' (got {device!r})"
|
||||
)
|
||||
try:
|
||||
vid = int(parts[0], 16)
|
||||
pid = int(parts[1], 16)
|
||||
except ValueError as e:
|
||||
raise ValueError(f"VID/PID must be hex: {device!r}") from e
|
||||
serial = parts[2] if len(parts) >= 3 and parts[2] else None
|
||||
return cls(vid, pid, serial)
|
||||
|
||||
|
||||
def _format_url(vid: int, pid: int, serial: Optional[str]) -> str:
|
||||
base = f"usb:{vid:04x}:{pid:04x}"
|
||||
return f"{base}:{serial}" if serial else base
|
||||
|
||||
|
||||
def list_android_usb_devices() -> List[SerialPortInfo]:
|
||||
"""Return USB-serial devices currently attached to the Android host."""
|
||||
try:
|
||||
bridge = _bridge()
|
||||
except Exception as e:
|
||||
logger.debug("UsbSerialBridge unavailable: %s", e)
|
||||
return []
|
||||
|
||||
out: List[SerialPortInfo] = []
|
||||
try:
|
||||
# listDevices() returns a Java List<String> of "vid:pid:serial:description"
|
||||
# tuples — split here to keep the Kotlin DTO trivial.
|
||||
for entry in bridge.listDevices():
|
||||
entry_s = str(entry)
|
||||
parts = entry_s.split("|", 3)
|
||||
if len(parts) < 4:
|
||||
continue
|
||||
vid_str, pid_str, serial, description = parts
|
||||
try:
|
||||
vid = int(vid_str, 16)
|
||||
pid = int(pid_str, 16)
|
||||
except ValueError:
|
||||
continue
|
||||
url = _format_url(vid, pid, serial or None)
|
||||
out.append(SerialPortInfo(device=url, description=description or url))
|
||||
except Exception as e:
|
||||
logger.warning("UsbSerialBridge.listDevices failed: %s", e)
|
||||
return out
|
||||
|
||||
|
||||
class AndroidSerialTransport:
|
||||
"""Serial transport that pipes writes through the Kotlin USB bridge."""
|
||||
|
||||
def __init__(self, device: str, baud_rate: int) -> None:
|
||||
self._url = device
|
||||
self._addr = _UsbAddress.parse(device)
|
||||
self._baud_rate = baud_rate
|
||||
self._handle: Optional[int] = None # opaque token from the bridge
|
||||
|
||||
@property
|
||||
def is_open(self) -> bool:
|
||||
return self._handle is not None
|
||||
|
||||
def open(self) -> None:
|
||||
bridge = _bridge()
|
||||
# open() returns a non-negative int handle on success, or -1 if the
|
||||
# device wasn't found / permission was denied.
|
||||
handle = int(
|
||||
bridge.open(
|
||||
self._addr.vendor_id,
|
||||
self._addr.product_id,
|
||||
self._addr.serial or "",
|
||||
self._baud_rate,
|
||||
)
|
||||
)
|
||||
if handle < 0:
|
||||
raise RuntimeError(
|
||||
f"Failed to open USB serial device {self._url} "
|
||||
f"(handle={handle}; user may have denied permission)"
|
||||
)
|
||||
self._handle = handle
|
||||
logger.info("Android USB serial opened: %s @ %d baud", self._url, self._baud_rate)
|
||||
|
||||
def write(self, data: bytes) -> None:
|
||||
if self._handle is None:
|
||||
raise RuntimeError(f"USB serial {self._url} is not open")
|
||||
bridge = _bridge()
|
||||
# Chaquopy auto-marshals bytes → Java byte[].
|
||||
bridge.write(self._handle, data)
|
||||
|
||||
def flush(self) -> None:
|
||||
# USB endpoints are unbuffered from this side — nothing to flush.
|
||||
return
|
||||
|
||||
def close(self) -> None:
|
||||
if self._handle is None:
|
||||
return
|
||||
try:
|
||||
_bridge().close(self._handle)
|
||||
except Exception as e:
|
||||
logger.warning("Error closing USB serial %s: %s", self._url, e)
|
||||
finally:
|
||||
self._handle = None
|
||||
|
||||
|
||||
__all__ = [
|
||||
"AndroidSerialTransport",
|
||||
"list_android_usb_devices",
|
||||
]
|
||||
|
||||
|
||||
def _used() -> Tuple[type, ...]: # pragma: no cover — re-export marker
|
||||
return (AndroidSerialTransport,)
|
||||
@@ -14,6 +14,10 @@ from ledgrab.core.devices.led_client import (
|
||||
DiscoveredDevice,
|
||||
LEDDeviceProvider,
|
||||
)
|
||||
from ledgrab.core.devices.serial_transport import (
|
||||
list_serial_ports,
|
||||
port_exists,
|
||||
)
|
||||
from ledgrab.utils import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
@@ -47,15 +51,9 @@ class SerialDeviceProvider(LEDDeviceProvider):
|
||||
port, _baud = parse_adalight_url(url)
|
||||
|
||||
try:
|
||||
import serial.tools.list_ports
|
||||
|
||||
available_ports = [p.device for p in serial.tools.list_ports.comports()]
|
||||
port_upper = port.upper()
|
||||
if not any(p.upper() == port_upper for p in available_ports):
|
||||
raise ValueError(
|
||||
f"Serial port {port} not found. "
|
||||
f"Available ports: {', '.join(available_ports) or 'none'}"
|
||||
)
|
||||
if not port_exists(port):
|
||||
available = ", ".join(p.device for p in list_serial_ports()) or "none"
|
||||
raise ValueError(f"Serial port {port} not found. Available ports: {available}")
|
||||
except ValueError:
|
||||
raise
|
||||
except Exception as e:
|
||||
@@ -67,22 +65,19 @@ class SerialDeviceProvider(LEDDeviceProvider):
|
||||
async def discover(self, timeout: float = 3.0) -> List[DiscoveredDevice]:
|
||||
"""Discover serial ports that could be LED devices."""
|
||||
try:
|
||||
import serial.tools.list_ports
|
||||
|
||||
ports = serial.tools.list_ports.comports()
|
||||
results = []
|
||||
for port_info in ports:
|
||||
results.append(
|
||||
DiscoveredDevice(
|
||||
name=port_info.description or port_info.device,
|
||||
url=port_info.device,
|
||||
device_type=self.device_type,
|
||||
ip=port_info.device,
|
||||
mac="",
|
||||
led_count=None,
|
||||
version=None,
|
||||
)
|
||||
ports = list_serial_ports()
|
||||
results = [
|
||||
DiscoveredDevice(
|
||||
name=port_info.description,
|
||||
url=port_info.device,
|
||||
device_type=self.device_type,
|
||||
ip=port_info.device,
|
||||
mac="",
|
||||
led_count=None,
|
||||
version=None,
|
||||
)
|
||||
for port_info in ports
|
||||
]
|
||||
logger.info(f"{self.device_type} serial port scan found {len(results)} port(s)")
|
||||
return results
|
||||
except Exception as e:
|
||||
|
||||
@@ -0,0 +1,174 @@
|
||||
"""Serial transport abstraction for LED clients.
|
||||
|
||||
Wraps the platform-specific way to open a serial line so callers
|
||||
(AdalightClient, etc.) don't import ``pyserial`` directly. The primary
|
||||
motivation is Android: under Chaquopy, ``pyserial`` exists but cannot
|
||||
touch USB ports, so we route writes through a Kotlin
|
||||
``UsbSerialBridge`` instead.
|
||||
|
||||
URL format
|
||||
----------
|
||||
- ``"COM3"`` or ``"COM3:115200"`` — desktop COM port (Windows)
|
||||
- ``"/dev/ttyUSB0"`` or ``"/dev/ttyUSB0:115200"`` — desktop tty
|
||||
- ``"usb:VID:PID"`` or ``"usb:VID:PID:serial@115200"`` — Android USB device
|
||||
|
||||
Selection happens in :func:`open_transport`: anything starting with
|
||||
``usb:`` goes through :class:`AndroidSerialTransport`, everything else
|
||||
through :class:`PySerialTransport`. :func:`list_serial_ports` returns
|
||||
desktop COM ports on desktop, USB devices on Android.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Optional, Protocol
|
||||
|
||||
from ledgrab.utils import get_logger
|
||||
from ledgrab.utils.platform import is_android
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
DEFAULT_BAUD_RATE = 115200
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SerialPortInfo:
|
||||
"""A discovered serial port, regardless of source platform."""
|
||||
|
||||
device: str # URL-like identifier passed back to open_transport()
|
||||
description: str # Human-friendly name for the UI
|
||||
|
||||
|
||||
class SerialTransport(Protocol):
|
||||
"""Minimal serial line — open, write bytes, close. No reads (LED out only)."""
|
||||
|
||||
@property
|
||||
def is_open(self) -> bool: ...
|
||||
def open(self) -> None: ...
|
||||
def write(self, data: bytes) -> None: ...
|
||||
def flush(self) -> None: ...
|
||||
def close(self) -> None: ...
|
||||
|
||||
|
||||
def parse_serial_url(url: str) -> tuple[str, int]:
|
||||
"""Parse a serial URL into (device, baud_rate).
|
||||
|
||||
Recognized formats::
|
||||
|
||||
"COM3" -> ("COM3", 115200)
|
||||
"COM3:230400" -> ("COM3", 230400)
|
||||
"/dev/ttyUSB0" -> ("/dev/ttyUSB0", 115200)
|
||||
"/dev/ttyUSB0:230400"-> ("/dev/ttyUSB0", 230400)
|
||||
"usb:1a86:7523" -> ("usb:1a86:7523", 115200)
|
||||
"usb:1a86:7523@250k" -> raises (only ints accepted for baud)
|
||||
"usb:1a86:7523@230400" -> ("usb:1a86:7523", 230400)
|
||||
|
||||
USB URLs use ``@`` to separate baud because the colon is already
|
||||
the field separator between vendor and product IDs.
|
||||
"""
|
||||
raw = url.strip()
|
||||
if raw.startswith("usb:"):
|
||||
if "@" in raw:
|
||||
device, _, baud_str = raw.rpartition("@")
|
||||
try:
|
||||
return device, int(baud_str)
|
||||
except ValueError:
|
||||
pass
|
||||
return raw, DEFAULT_BAUD_RATE
|
||||
|
||||
# Desktop COM/tty paths use a trailing :BAUD suffix.
|
||||
if ":" in raw:
|
||||
head, _, tail = raw.rpartition(":")
|
||||
try:
|
||||
return head, int(tail)
|
||||
except ValueError:
|
||||
pass
|
||||
return raw, DEFAULT_BAUD_RATE
|
||||
|
||||
|
||||
def list_serial_ports() -> List[SerialPortInfo]:
|
||||
"""Enumerate serial ports available on this host (or USB devices on Android)."""
|
||||
if is_android():
|
||||
try:
|
||||
from ledgrab.core.devices.android_serial_transport import (
|
||||
list_android_usb_devices,
|
||||
)
|
||||
|
||||
return list_android_usb_devices()
|
||||
except Exception as e:
|
||||
logger.warning("Android USB enumeration failed: %s", e)
|
||||
return []
|
||||
|
||||
try:
|
||||
import serial.tools.list_ports
|
||||
except ImportError:
|
||||
logger.warning("pyserial not available — serial enumeration disabled")
|
||||
return []
|
||||
|
||||
return [
|
||||
SerialPortInfo(device=p.device, description=p.description or p.device)
|
||||
for p in serial.tools.list_ports.comports()
|
||||
]
|
||||
|
||||
|
||||
def port_exists(device: str) -> bool:
|
||||
"""Return True if a port with this device id is currently visible."""
|
||||
target = device.upper()
|
||||
return any(p.device.upper() == target for p in list_serial_ports())
|
||||
|
||||
|
||||
def open_transport(
|
||||
url: str,
|
||||
baud_rate: Optional[int] = None,
|
||||
timeout: float = 1.0,
|
||||
) -> SerialTransport:
|
||||
"""Construct an unopened transport for ``url``. Caller invokes ``open()``."""
|
||||
device, parsed_baud = parse_serial_url(url)
|
||||
effective_baud = baud_rate or parsed_baud
|
||||
|
||||
if device.startswith("usb:"):
|
||||
from ledgrab.core.devices.android_serial_transport import (
|
||||
AndroidSerialTransport,
|
||||
)
|
||||
|
||||
return AndroidSerialTransport(device, effective_baud)
|
||||
|
||||
return PySerialTransport(device, effective_baud, timeout)
|
||||
|
||||
|
||||
class PySerialTransport:
|
||||
"""Default serial transport backed by ``pyserial`` on desktop hosts."""
|
||||
|
||||
def __init__(self, device: str, baud_rate: int, timeout: float = 1.0) -> None:
|
||||
self._device = device
|
||||
self._baud_rate = baud_rate
|
||||
self._timeout = timeout
|
||||
self._serial = None
|
||||
|
||||
@property
|
||||
def is_open(self) -> bool:
|
||||
return self._serial is not None and self._serial.is_open
|
||||
|
||||
def open(self) -> None:
|
||||
import serial # imported here so Android (no real pyserial) doesn't fail at import
|
||||
|
||||
self._serial = serial.Serial(
|
||||
port=self._device, baudrate=self._baud_rate, timeout=self._timeout
|
||||
)
|
||||
|
||||
def write(self, data: bytes) -> None:
|
||||
if self._serial is None:
|
||||
raise RuntimeError(f"Serial port {self._device} is not open")
|
||||
self._serial.write(data)
|
||||
|
||||
def flush(self) -> None:
|
||||
if self._serial is not None:
|
||||
self._serial.flush()
|
||||
|
||||
def close(self) -> None:
|
||||
if self._serial is not None and self._serial.is_open:
|
||||
try:
|
||||
self._serial.close()
|
||||
except Exception as e:
|
||||
logger.warning("Error closing %s: %s", self._device, e)
|
||||
self._serial = None
|
||||
@@ -0,0 +1,153 @@
|
||||
"""Tests for the serial transport abstraction."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from ledgrab.core.devices.serial_transport import (
|
||||
DEFAULT_BAUD_RATE,
|
||||
PySerialTransport,
|
||||
SerialPortInfo,
|
||||
list_serial_ports,
|
||||
open_transport,
|
||||
parse_serial_url,
|
||||
port_exists,
|
||||
)
|
||||
|
||||
|
||||
# ── URL parsing ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"url,expected",
|
||||
[
|
||||
("COM3", ("COM3", DEFAULT_BAUD_RATE)),
|
||||
("COM3:230400", ("COM3", 230400)),
|
||||
("/dev/ttyUSB0", ("/dev/ttyUSB0", DEFAULT_BAUD_RATE)),
|
||||
("/dev/ttyUSB0:230400", ("/dev/ttyUSB0", 230400)),
|
||||
# USB URLs use @ for baud since `:` separates VID:PID:serial
|
||||
("usb:1a86:7523", ("usb:1a86:7523", DEFAULT_BAUD_RATE)),
|
||||
("usb:1a86:7523:AB01", ("usb:1a86:7523:AB01", DEFAULT_BAUD_RATE)),
|
||||
("usb:1a86:7523@230400", ("usb:1a86:7523", 230400)),
|
||||
("usb:1a86:7523:AB01@500000", ("usb:1a86:7523:AB01", 500000)),
|
||||
],
|
||||
)
|
||||
def test_parse_serial_url(url, expected):
|
||||
assert parse_serial_url(url) == expected
|
||||
|
||||
|
||||
def test_parse_serial_url_strips_whitespace():
|
||||
assert parse_serial_url(" COM3 ") == ("COM3", DEFAULT_BAUD_RATE)
|
||||
|
||||
|
||||
# ── PySerialTransport ──────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_pyserial_transport_open_and_write(monkeypatch):
|
||||
fake_serial_module = MagicMock()
|
||||
fake_handle = MagicMock(is_open=True)
|
||||
fake_serial_module.Serial.return_value = fake_handle
|
||||
|
||||
monkeypatch.setitem(__import__("sys").modules, "serial", fake_serial_module)
|
||||
|
||||
t = PySerialTransport("COM3", 115200)
|
||||
assert not t.is_open
|
||||
t.open()
|
||||
fake_serial_module.Serial.assert_called_once_with(port="COM3", baudrate=115200, timeout=1.0)
|
||||
assert t.is_open
|
||||
|
||||
t.write(b"hello")
|
||||
fake_handle.write.assert_called_once_with(b"hello")
|
||||
|
||||
t.flush()
|
||||
fake_handle.flush.assert_called_once()
|
||||
|
||||
t.close()
|
||||
fake_handle.close.assert_called_once()
|
||||
|
||||
|
||||
def test_pyserial_transport_write_before_open_raises():
|
||||
t = PySerialTransport("COM3", 115200)
|
||||
with pytest.raises(RuntimeError, match="not open"):
|
||||
t.write(b"x")
|
||||
|
||||
|
||||
# ── Factory ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_open_transport_picks_pyserial_for_com_url():
|
||||
t = open_transport("COM3:230400")
|
||||
assert isinstance(t, PySerialTransport)
|
||||
assert t._device == "COM3"
|
||||
assert t._baud_rate == 230400
|
||||
|
||||
|
||||
def test_open_transport_explicit_baud_overrides_url():
|
||||
t = open_transport("COM3:230400", baud_rate=500000)
|
||||
assert t._baud_rate == 500000
|
||||
|
||||
|
||||
def test_open_transport_picks_android_for_usb_url(monkeypatch):
|
||||
"""usb: URLs should route to the Android transport even off-Android."""
|
||||
# Importing AndroidSerialTransport itself works on any host; only the
|
||||
# bridge call inside .open() fails when not on Android.
|
||||
t = open_transport("usb:1a86:7523@230400")
|
||||
assert type(t).__name__ == "AndroidSerialTransport"
|
||||
assert t._baud_rate == 230400
|
||||
|
||||
|
||||
# ── Discovery ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_list_serial_ports_uses_pyserial_on_desktop(monkeypatch):
|
||||
monkeypatch.setattr("ledgrab.core.devices.serial_transport.is_android", lambda: False)
|
||||
|
||||
fake_module = MagicMock()
|
||||
fake_module.tools.list_ports.comports.return_value = [
|
||||
MagicMock(device="COM3", description="USB Serial CH340"),
|
||||
MagicMock(device="COM4", description=None),
|
||||
]
|
||||
with patch.dict(
|
||||
"sys.modules",
|
||||
{
|
||||
"serial": fake_module,
|
||||
"serial.tools": fake_module.tools,
|
||||
"serial.tools.list_ports": fake_module.tools.list_ports,
|
||||
},
|
||||
):
|
||||
ports = list_serial_ports()
|
||||
|
||||
assert len(ports) == 2
|
||||
assert ports[0] == SerialPortInfo(device="COM3", description="USB Serial CH340")
|
||||
# Falls back to device id when description is None
|
||||
assert ports[1].description == "COM4"
|
||||
|
||||
|
||||
def test_list_serial_ports_routes_to_android_when_on_android(monkeypatch):
|
||||
monkeypatch.setattr("ledgrab.core.devices.serial_transport.is_android", lambda: True)
|
||||
|
||||
fake_devices = [SerialPortInfo(device="usb:1a86:7523", description="CH340 LED")]
|
||||
fake_mod = MagicMock()
|
||||
fake_mod.list_android_usb_devices.return_value = fake_devices
|
||||
monkeypatch.setitem(
|
||||
__import__("sys").modules,
|
||||
"ledgrab.core.devices.android_serial_transport",
|
||||
fake_mod,
|
||||
)
|
||||
|
||||
assert list_serial_ports() == fake_devices
|
||||
|
||||
|
||||
def test_port_exists_is_case_insensitive(monkeypatch):
|
||||
fake_ports = [
|
||||
SerialPortInfo(device="COM3", description="x"),
|
||||
SerialPortInfo(device="/dev/ttyUSB0", description="y"),
|
||||
]
|
||||
monkeypatch.setattr(
|
||||
"ledgrab.core.devices.serial_transport.list_serial_ports", lambda: fake_ports
|
||||
)
|
||||
assert port_exists("com3") is True
|
||||
assert port_exists("/dev/ttyusb0") is True
|
||||
assert port_exists("COM99") is False
|
||||
Reference in New Issue
Block a user