Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,5 +4,5 @@
- The CLI and firmware engine are Python in `cli/`. Run its tests from the repo root with `PYTHONPATH=cli/src python -m unittest discover -s cli/tests` (about 80 seconds). Use a venv with the CLI's dependencies installed: the system `python3` fails 11 of them.
- `python3 tools/test_cli_workflows.py` drives attached boards, and skips flashing unless you pass `--install-latest-firmware`.
- Prefer the extension's TCP RPC over opening a second serial connection to the same board ([how commands find a session](docs/guides/drive-a-board-from-an-agent.md#how-commands-find-a-session)).
- On WSL, serial and ESP32 flashing need Windows Python (`python.exe`), not the WSL interpreter ([install](docs/get-started/install.md#vs-code-and-compatible-editors)).
- On WSL, a `COMn` port and ESP32 flashing need Windows Python (`python.exe`); a `/dev/tty*` port (usbipd) runs in the Linux interpreter ([install](docs/get-started/install.md#vs-code-and-compatible-editors)).
- Board operations, CLI workflows, firmware build and flash, and troubleshooting: [drive a board from an agent](docs/guides/drive-a-board-from-an-agent.md).
46 changes: 41 additions & 5 deletions cli/src/mpftp/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -848,10 +848,20 @@ def _probe_python(cand: str, timeout: float) -> Optional[str]:
return lines[-1] if lines else f"exit status {r.returncode}"


def resolve_python() -> str:
# The -d device of the command being run, so get_client() can pick the
# interpreter without every command passing it down.
_device_hint: Optional[str] = None


def _is_linux_device(device: Optional[str]) -> bool:
return bool(device) and str(device).startswith("/dev/")


def resolve_python(device: Optional[str] = None) -> str:
"""The interpreter the sidecar runs in: one that has mpremote and pyserial.

On WSL that's a Windows Python, since only Windows can open COM ports.
On WSL that's a Windows Python, since only Windows can open COM ports; a
``/dev/...`` device is a Linux port, so it gets the Linux interpreter.
Elsewhere it's the interpreter running mpftp, so the sidecar gets the
mpremote mpftp was installed with rather than whatever ``python.exe`` is
first on PATH.
Expand All @@ -870,7 +880,9 @@ def resolve_python() -> str:
"python",
]
wsl = bool(os.environ.get("WSL_DISTRO_NAME") or os.environ.get("WSL_INTEROP"))
groups = [windows, own] if wsl else [own, windows]
if device is None:
device = _device_hint
groups = [windows, own] if wsl and not _is_linux_device(device) else [own, windows]
tried: list[str] = []
seen: set[str] = set()
for group in groups:
Expand All @@ -892,7 +904,7 @@ def resolve_python() -> str:
cand = next((c for c in slow if _probe_python(c, 45) is None), "")
if not cand:
continue
if wsl and group is own:
if wsl and group is own and not _is_linux_device(device):
print(
"mpftp: no Windows Python with mpremote and pyserial answered ("
+ "; ".join(tried)
Expand Down Expand Up @@ -1011,9 +1023,31 @@ def cmd_ports(_: argparse.Namespace) -> None:
client, _ = get_client()
try:
ports = client.call("list_ports")
out(ports)
finally:
client.close()
out(ports + _linux_ports(ports))


def _linux_ports(listed: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Under WSL the sidecar is Windows Python and sees only COM ports; ask a
Linux one for the /dev ports (a board attached with usbipd)."""
if not (os.environ.get("WSL_DISTRO_NAME") or os.environ.get("WSL_INTEROP")):
return []
try:
python = resolve_python("/dev/")
if python == resolve_python("COM1"):
return []
client = SidecarClient(python)
try:
found = client.call("list_ports")
finally:
client.close()
except BaseException as e: # a missing Linux pyserial must not break `ports`
if isinstance(e, KeyboardInterrupt):
raise
return []
have = {p.get("device") for p in listed}
return [p for p in found if str(p.get("device", "")).startswith("/dev/") and p.get("device") not in have]


def connect_params(device: str, baud: int) -> dict[str, Any]:
Expand Down Expand Up @@ -3099,6 +3133,8 @@ def main(argv: Optional[list[str]] = None) -> None:
ns.device = ns.device_pos
elif not hasattr(ns, "device"):
ns.device = None
global _device_hint
_device_hint = ns.device
if not hasattr(ns, "baud"):
ns.baud = 115200
try:
Expand Down
7 changes: 7 additions & 0 deletions cli/src/mpftp/sidecar.py
Original file line number Diff line number Diff line change
Expand Up @@ -1616,6 +1616,13 @@ def _friendly_port_open_error(device: str, exc: BaseException) -> str:
if isinstance(exc, webrepl.WebReplError):
return str(exc) # already says what to check on the network
raw = str(exc.args[0] if getattr(exc, "args", None) else exc)
# mpremote wraps the serial error in a bare TransportError; the real
# OSError (no such device, permission, ...) is its cause.
cause = exc.__cause__ or exc.__context__
while cause is not None and not isinstance(cause, OSError):
cause = cause.__cause__ or cause.__context__
if cause is not None and str(cause) and str(cause) not in raw:
raw = f"{raw}; {type(cause).__name__}: {cause}"
low = raw.lower()
locked = (
"exclusively lock" in low
Expand Down
52 changes: 50 additions & 2 deletions cli/tests/test_resolve_python.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@


class ResolvePythonTests(unittest.TestCase):
def _resolve(self, answers, wsl):
def _resolve(self, answers, wsl, device=None, env_python=None):
"""``answers`` maps a candidate to its probe result (None = usable)."""
calls = []

Expand All @@ -26,13 +26,15 @@ def probe(cand, timeout):
env = {k: v for k, v in os.environ.items() if not k.startswith("WSL_") and k != "MPFTP_PYTHON"}
if wsl:
env["WSL_DISTRO_NAME"] = "Ubuntu"
if env_python:
env["MPFTP_PYTHON"] = env_python
err = io.StringIO()
with mock.patch.dict(os.environ, env, clear=True), \
mock.patch.object(cli.config, "resolve", return_value=None), \
mock.patch.object(cli, "_probe_python", side_effect=probe), \
redirect_stderr(err):
try:
return cli.resolve_python(), calls, err.getvalue()
return cli.resolve_python(device), calls, err.getvalue()
except SystemExit as e:
return None, calls, str(e)

Expand All @@ -54,6 +56,20 @@ def test_wsl_slow_windows_python_gets_a_longer_try(self):
self.assertEqual(got, WIN)
self.assertIn((WIN, 45), calls)

def test_wsl_linux_device_runs_the_sidecar_in_linux(self):
got, calls, err = self._resolve({WIN: None, sys.executable: None}, wsl=True, device="/dev/ttyACM0")
self.assertEqual(got, sys.executable)
self.assertNotIn(WIN, [c for c, _ in calls])
self.assertEqual(err, "")

def test_wsl_com_device_still_uses_windows_python(self):
got, _, _ = self._resolve({WIN: None, sys.executable: None}, wsl=True, device="COM17")
self.assertEqual(got, WIN)

def test_mpftp_python_overrides_for_a_linux_device(self):
got, _, _ = self._resolve({}, wsl=True, device="/dev/ttyACM0", env_python="/opt/py")
self.assertEqual(got, "/opt/py")

def test_elsewhere_the_running_interpreter_comes_first(self):
got, calls, _ = self._resolve({sys.executable: None, "python.exe": None}, wsl=False)
self.assertEqual(got, sys.executable)
Expand All @@ -66,5 +82,37 @@ def test_no_usable_python_names_each_reason(self):
self.assertIn("python.exe: not found", msg)


class FakeClient:
def __init__(self, python, ports):
self.python, self.ports = python, ports

def call(self, method):
return list(self.ports)

def close(self):
pass


class PortsUnderWslTests(unittest.TestCase):
def test_ports_adds_linux_ports_to_the_windows_list(self):
win = [{"device": "COM31"}]
lin = [{"device": "/dev/ttyACM0"}]

def make(python):
return FakeClient(python, lin if python == "linux-py" else win)

def resolve(device=None):
return "linux-py" if device == "/dev/" else "win-py"

out = []
with mock.patch.dict(os.environ, {"WSL_DISTRO_NAME": "Ubuntu"}), \
mock.patch.object(cli, "resolve_python", side_effect=resolve), \
mock.patch.object(cli, "get_client", return_value=(make("win-py"), "sidecar")), \
mock.patch.object(cli, "SidecarClient", side_effect=make), \
mock.patch.object(cli, "out", side_effect=out.append):
cli.cmd_ports(None)
self.assertEqual([p["device"] for p in out[0]], ["COM31", "/dev/ttyACM0"])


if __name__ == "__main__":
unittest.main()
12 changes: 12 additions & 0 deletions cli/tests/test_serial_recovery.py
Original file line number Diff line number Diff line change
Expand Up @@ -257,6 +257,18 @@ def test_native_cdc_port_says_the_board_may_be_stuck(self):
self.assertIn("hard-reset", msg)
self.assertIn("Access is denied", msg)

def test_hidden_cause_is_shown(self):
from mpremote.transport import TransportError
try:
try:
raise OSError(2, "No such file or directory: '/dev/ttyACM0'")
except OSError as inner:
raise TransportError("failed to access /dev/ttyACM0") from inner
except TransportError as e:
with mock.patch("serial.tools.list_ports.comports", return_value=self.PORTS):
msg = self.mod.Session._friendly_port_open_error("/dev/ttyACM0", e)
self.assertIn("No such file or directory", msg)

def test_uart_bridge_and_serial_jtag_keep_the_plain_message(self):
for device in ("COM4", "COM9"):
msg = self._message(device)
Expand Down
Loading