Improves Setup / Flashing When Multiple USB Devices

README.md

@@ -182,10 +182,14 @@ and let it fail to find the board — it writes `secrets.h` and leaves it in pla
 when the flash doesn't happen. Then flash by hand:
 
 ```sh
+arduino-cli board list   # find your board's port, e.g. /dev/cu.usbmodem1402
 arduino-cli compile --fqbn esp32:esp32:adafruit_qtpy_esp32s3_nopsram:CDCOnBoot=cdc,USBMode=default firmware/dashtouch
-arduino-cli upload --fqbn esp32:esp32:adafruit_qtpy_esp32s3_nopsram:CDCOnBoot=cdc,USBMode=default -p $(arduino-cli board list | grep -o '/dev/cu.usbmodem[0-9]*' | head -1) firmware/dashtouch
+arduino-cli upload --fqbn esp32:esp32:adafruit_qtpy_esp32s3_nopsram:CDCOnBoot=cdc,USBMode=default -p <your board's port> firmware/dashtouch
 ```
 
+Check the port carefully if anything else is plugged in over USB — upload
+will overwrite whatever board is on it.
+
 Delete `firmware/dashtouch/secrets.h` when you're done — it's a plaintext copy
 of the same key that's in your Keychain.
 

helper/dashtouch_helper/cli.py

@@ -112,25 +112,72 @@ def wait_for_port_free(port: str, timeout: float = 5.0) -> list:
         time.sleep(0.25)
 
 
-def find_and_flash(prompt_prefix: str = "") -> str:
-    """Detect the board and, if found, offer to compile+upload the current
+def _helper_holds(port: str) -> bool:
+    return any("dashtouch" in command for _, command in port_holders(port))
+
+
+def _describe_port(port) -> str:
+    return f"{port.device}  {port.label}" if port.label else port.device
+
+
+def choose_port(prompt_prefix: str = "", port_arg: str | None = None):
+    """Pick the board to flash: the UsbPort, None if there's nothing to
+    flash, or "declined".
+
+    Before it's flashed, a dashtouch is just some board, so with more than
+    one plugged in the only reliable answer is the person's. The one the
+    helper already holds is the likeliest answer, so it's the default — a
+    reflash is then one keypress — but it's only a default: the helper
+    picks by port name, and nothing here proves what's on the other end.
+    """
+    ports = serial_link.usb_ports()
+    if port_arg:
+        match = [p for p in ports if p.device == port_arg]
+        return match[0] if match else serial_link.UsbPort(port_arg, "", (port_arg,))
+    if not ports:
+        return None
+    try:
+        if len(ports) == 1:
+            ans = input(f"{prompt_prefix}Found a board at {_describe_port(ports[0])}. "
+                        "Flash the firmware now? [Y/n] ")
+            return ports[0] if ans.strip().lower() in ("", "y", "yes") else "declined"
+
+        held = [i for i, p in enumerate(ports) if _helper_holds(p.device)]
+        default = held[0] + 1 if held else None
+        print(f"{prompt_prefix}More than one USB device is plugged in:")
+        for i, p in enumerate(ports, 1):
+            note = "   <- the helper is connected to this one" if i == default else ""
+            print(f"     {i}) {_describe_port(p)}{note}")
+        hint = f"[{default}]" if default else "(Enter to skip)"
+        for _ in range(3):
+            ans = input(f"   Which one should get the Dashboard Touch firmware? {hint} ").strip()
+            if not ans:
+                return ports[default - 1] if default else "declined"
+            if ans.isdigit() and 1 <= int(ans) <= len(ports):
+                return ports[int(ans) - 1]
+            print(f"   Pick a number from 1 to {len(ports)}.")
+    except EOFError:
+        pass
+    return "declined"
+
+
+def find_and_flash(prompt_prefix: str = "", port_arg: str | None = None) -> str:
+    """Pick the board and, if chosen, compile+upload the current
     firmware/dashtouch/ sources to it. Shared by `setup` and `pairing` so
     there's exactly one flash code path.
 
     Returns one of "flashed", "declined", "not_found", "failed".
-    `prompt_prefix` is prepended to the confirmation prompt (e.g. step
-    numbering) so each caller keeps its own voice.
+    `prompt_prefix` is prepended to the first prompt (e.g. step numbering)
+    so each caller keeps its own voice. `port_arg` skips the question.
     """
-    port = None
-    try:
-        port = serial_link.find_port()
-    except serial_link.AmbiguousPortError as e:
-        print(f"   Heads up: {e}")
-    if not (port and shutil.which("arduino-cli")):
+    if not shutil.which("arduino-cli"):
         return "not_found"
-    ans = input(f"{prompt_prefix}Found your board at {port}. Flash the firmware now? [Y/n] ")
-    if ans.strip().lower() not in ("", "y", "yes"):
+    chosen = choose_port(prompt_prefix, port_arg)
+    if chosen is None:
+        return "not_found"
+    if chosen == "declined":
         return "declined"
+    port = chosen.device
 
     # The helper keeps the serial port open, so flashing on top of a running
     # one dies with "Resource busy" — and because the launch agent sets
@@ -152,12 +199,12 @@ def find_and_flash(prompt_prefix: str = "") -> str:
             return "failed"
 
         # Releasing the port drops DTR, which can reset the board and bring
-        # it back on a different device node. Re-detect instead of trusting
-        # the name captured before the helper let go.
-        try:
-            port = serial_link.find_port() or port
-        except serial_link.AmbiguousPortError:
-            pass
+        # it back on a different device node. Look it up again by identity
+        # instead of trusting the name captured before the helper let go.
+        for p in serial_link.usb_ports():
+            if p.identity == chosen.identity:
+                port = p.device
+                break
 
         subprocess.run(["arduino-cli", "compile", "--fqbn", FQBN,
                         str(REPO / "firmware" / "dashtouch")], check=True)
@@ -203,7 +250,7 @@ def cmd_setup(args) -> int:
     print("3. Writing the key into the firmware's secrets file.")
     write_secrets(key, SECRETS_PATH)
 
-    result = find_and_flash("4. ")
+    result = find_and_flash("4. ", getattr(args, "port", None))
     if result == "flashed":
         print("   Flashed. Watch for the steady purple ring.")
     elif result == "failed":
@@ -286,7 +333,7 @@ def cmd_pairing(args) -> int:
     write_secrets(key, SECRETS_PATH)
     print("Done.\n")
 
-    result = find_and_flash("")
+    result = find_and_flash("", getattr(args, "port", None))
     if result == "flashed":
         print("\nFlashed. Watch for the steady purple ring.")
         return 0
@@ -562,15 +609,17 @@ def main() -> None:
     p.add_argument("--serial", help="board serial (default: this build's)")
     # Not required: bare `dashtouch` prints the web UI address.
     sub = p.add_subparsers(dest="cmd")
-    sub.add_parser("setup", help="first-time setup, start to finish")
+    p_setup = sub.add_parser("setup", help="first-time setup, start to finish")
+    p_setup.add_argument("--port", help="the board's serial port, if you'd rather not be asked")
     sub.add_parser("run", help="run the helper (launchd uses this)")
     sub.add_parser("enroll", help="open the enrollment page")
     sub.add_parser("doctor", help="quick health check")
     sub.add_parser("install-agent", help="run the helper automatically at login")
     sub.add_parser("uninstall-agent", help="stop running the helper at login")
     p_password = sub.add_parser("password", help="change the password Dashboard Touch types")
     p_password.add_argument("--password", help=argparse.SUPPRESS)
-    sub.add_parser("pairing", help="rotate the pairing key (needs a reflash)")
+    p_pairing = sub.add_parser("pairing", help="rotate the pairing key (needs a reflash)")
+    p_pairing.add_argument("--port", help="the board's serial port, if you'd rather not be asked")
     p_pins = sub.add_parser("pins", help="check or change the sensor's TX/RX pin orientation")
     p_pins.add_argument("--swap", action="store_true",
                         help="swap the sensor UART pins (for boards that can't transmit on the default one)")

helper/dashtouch_helper/serial_link.py

@@ -5,6 +5,8 @@
 """
 from __future__ import annotations
 
+from dataclasses import dataclass
+
 import serial
 import serial.tools.list_ports
 
@@ -15,6 +17,33 @@ class AmbiguousPortError(Exception):
     pass
 
 
+@dataclass(frozen=True)
+class UsbPort:
+    device: str
+    label: str
+    # Stable across a re-enumeration onto a new /dev node, where the board
+    # reports a serial number. Used to find the same board again after a
+    # reset renames its port.
+    identity: tuple
+
+
+def usb_ports() -> list[UsbPort]:
+    """USB serial ports a dashtouch could be on, with a name for each.
+
+    Only /dev/cu.usbmodem*, the same ports find_port() considers: the
+    firmware is built for a native-USB ESP32-S3, and a board that turns up
+    anywhere else is one the helper would never find after flashing it.
+    """
+    out = []
+    for p in serial.tools.list_ports.comports():
+        if p.vid is None or not p.device.startswith("/dev/cu.usbmodem"):
+            continue
+        label = " ".join(x for x in (p.manufacturer, p.product) if x)
+        identity = (p.vid, p.pid, p.serial_number or p.location or p.device)
+        out.append(UsbPort(p.device, label or p.description or "", identity))
+    return out
+
+
 def find_port() -> str | None:
     cands = [p.device for p in serial.tools.list_ports.comports()
              if p.device.startswith("/dev/cu.usbmodem")]

helper/tests/test_cli.py

@@ -133,7 +133,8 @@ def test_pairing_does_not_flash_when_declined(tmp_path, monkeypatch):
     answers = iter(["y", "n"])
     monkeypatch.setattr("builtins.input", lambda *_: next(answers))
     monkeypatch.setattr(cli.keychain, "set_pairing_key", mock.Mock())
-    monkeypatch.setattr(cli.serial_link, "find_port", lambda: "/dev/cu.usbmodem1")
+    monkeypatch.setattr(cli.serial_link, "usb_ports",
+                        lambda: [usb("/dev/cu.usbmodem1")])
     monkeypatch.setattr(cli.shutil, "which", lambda _: "/usr/local/bin/arduino-cli")
     run = mock.Mock()
     monkeypatch.setattr(cli.subprocess, "run", run)
@@ -366,20 +367,34 @@ def test_bare_dashtouch_survives_a_corrupt_url_file(tmp_path, monkeypatch, capsy
     assert "isn't running" in capsys.readouterr().out
 
 
-def _flashable(monkeypatch, ports=("/dev/cu.usbmodem1",)):
-    """Common stubs so find_and_flash gets as far as the flash itself."""
-    seq = iter(ports)
-    last = [ports[-1]]
+def usb(device, label="", identity=None):
+    return cli.serial_link.UsbPort(device, label, identity or (device,))
 
-    def next_port():
-        try:
-            return next(seq)
-        except StopIteration:
-            return last[0]
 
-    monkeypatch.setattr(cli.serial_link, "find_port", next_port)
+def _flashable(monkeypatch, listings=(("/dev/cu.usbmodem1",),), answers=("y",),
+               helper_on=None):
+    """Common stubs so find_and_flash gets as far as the flash itself.
+
+    `listings` is what successive usb_ports() calls see; the last one
+    repeats. `helper_on` is the device the running helper holds, if any.
+    """
+    seq = [[usb(d) if isinstance(d, str) else d for d in ports] for ports in listings]
+    calls = iter(range(10**6))
+    monkeypatch.setattr(cli.serial_link, "usb_ports",
+                        lambda: seq[min(next(calls), len(seq) - 1)])
     monkeypatch.setattr(cli.shutil, "which", lambda name: "/usr/bin/" + name)
-    monkeypatch.setattr("builtins.input", lambda *_: "y")
+    replies = iter(answers)
+    prompts = []
+
+    def fake_input(prompt=""):
+        prompts.append(prompt)
+        return next(replies)
+
+    monkeypatch.setattr("builtins.input", fake_input)
+    monkeypatch.setattr(cli, "port_holders",
+                        lambda port: [(1, "python -m dashtouch_helper.daemon")]
+                        if port == helper_on else [])
+    return prompts
 
 
 def test_find_and_flash_stops_the_agent_before_flashing_and_restarts_it(monkeypatch):
@@ -435,7 +450,10 @@ def test_find_and_flash_redetects_the_port_after_the_helper_lets_go(monkeypatch)
     # Releasing the port drops DTR, which can reset the board onto a new
     # device node. Flashing the pre-release name then hits a port that no
     # longer exists.
-    _flashable(monkeypatch, ports=("/dev/cu.usbmodem83402", "/dev/cu.usbmodem83401"))
+    board = "68:ee:8f"
+    _flashable(monkeypatch, listings=(
+        (usb("/dev/cu.usbmodem83402", identity=(board,)),),
+        (usb("/dev/cu.usbmodem83401", identity=(board,)),)))
     monkeypatch.setattr(cli, "agent_is_loaded", lambda: False)
     monkeypatch.setattr(cli, "wait_for_port_free", lambda port, **kw: [])
     captured = []
@@ -445,3 +463,97 @@ def test_find_and_flash_redetects_the_port_after_the_helper_lets_go(monkeypatch)
     upload = [c for c in captured if "upload" in c][0]
     assert "/dev/cu.usbmodem83401" in upload
     assert "/dev/cu.usbmodem83402" not in upload
+
+
+def test_find_and_flash_redetects_the_port_with_another_board_plugged_in(monkeypatch):
+    # Re-detecting by "the only usbmodem port" gave up when there were two,
+    # and flashed the stale name. Identity finds the same board regardless.
+    board = "68:ee:8f"
+    other = usb("/dev/cu.usbmodem1234562", identity=("123456",))
+    _flashable(monkeypatch, listings=(
+        (usb("/dev/cu.usbmodem83402", identity=(board,)), other),
+        (other, usb("/dev/cu.usbmodem83401", identity=(board,)))),
+        answers=("1",))
+    monkeypatch.setattr(cli, "agent_is_loaded", lambda: False)
+    monkeypatch.setattr(cli, "wait_for_port_free", lambda port, **kw: [])
+    captured = []
+    monkeypatch.setattr(cli.subprocess, "run", lambda a, **k: captured.append(a))
+
+    assert cli.find_and_flash() == "flashed"
+    assert _uploaded_to(captured) == "/dev/cu.usbmodem83401"
+
+
+def _uploaded_to(captured):
+    upload = [c for c in captured if "upload" in c][0]
+    return upload[upload.index("-p") + 1]
+
+
+def _flash_stubs(monkeypatch):
+    monkeypatch.setattr(cli, "agent_is_loaded", lambda: False)
+    monkeypatch.setattr(cli, "wait_for_port_free", lambda port, **kw: [])
+    captured = []
+    monkeypatch.setattr(cli.subprocess, "run", lambda a, **k: captured.append(a))
+    return captured
+
+
+def test_find_and_flash_asks_which_board_when_several_are_plugged_in(monkeypatch, capsys):
+    # Nothing tells a not-yet-flashed dashtouch from any other board, so
+    # with more than one plugged in the person picks.
+    _flashable(monkeypatch, listings=(("/dev/cu.usbmodemA", "/dev/cu.usbmodemB"),),
+               answers=("2",))
+    captured = _flash_stubs(monkeypatch)
+    assert cli.find_and_flash() == "flashed"
+    assert _uploaded_to(captured) == "/dev/cu.usbmodemB"
+    out = capsys.readouterr().out
+    assert "1) /dev/cu.usbmodemA" in out and "2) /dev/cu.usbmodemB" in out
+
+
+def test_find_and_flash_defaults_to_the_board_the_helper_is_using(monkeypatch, capsys):
+    # The helper only holds a port whose board answered HELLO, so a reflash
+    # is one keypress.
+    _flashable(monkeypatch, listings=(("/dev/cu.usbmodemA", "/dev/cu.usbmodemB"),),
+               answers=("",), helper_on="/dev/cu.usbmodemB")
+    captured = _flash_stubs(monkeypatch)
+    assert cli.find_and_flash() == "flashed"
+    assert _uploaded_to(captured) == "/dev/cu.usbmodemB"
+    assert "helper is connected to this one" in capsys.readouterr().out
+
+
+def test_find_and_flash_with_no_obvious_choice_skips_on_enter(monkeypatch):
+    _flashable(monkeypatch, listings=(("/dev/cu.usbmodemA", "/dev/cu.usbmodemB"),),
+               answers=("",))
+    captured = _flash_stubs(monkeypatch)
+    assert cli.find_and_flash() == "declined"
+    assert captured == []
+
+
+def test_find_and_flash_reasks_on_a_bad_choice(monkeypatch):
+    prompts = _flashable(monkeypatch,
+                         listings=(("/dev/cu.usbmodemA", "/dev/cu.usbmodemB"),),
+                         answers=("7", "banana", "1"))
+    captured = _flash_stubs(monkeypatch)
+    assert cli.find_and_flash() == "flashed"
+    assert _uploaded_to(captured) == "/dev/cu.usbmodemA"
+    assert len(prompts) == 3
+
+
+def test_find_and_flash_gives_up_quietly_when_input_ends(monkeypatch):
+    _flashable(monkeypatch, listings=(("/dev/cu.usbmodemA", "/dev/cu.usbmodemB"),))
+    monkeypatch.setattr("builtins.input", mock.Mock(side_effect=EOFError))
+    captured = _flash_stubs(monkeypatch)
+    assert cli.find_and_flash() == "declined"
+    assert captured == []
+
+
+def test_find_and_flash_takes_a_port_without_asking(monkeypatch):
+    _flashable(monkeypatch, listings=(("/dev/cu.usbmodemA", "/dev/cu.usbmodemB"),),
+               answers=())
+    captured = _flash_stubs(monkeypatch)
+    assert cli.find_and_flash(port_arg="/dev/cu.usbmodemA") == "flashed"
+    assert _uploaded_to(captured) == "/dev/cu.usbmodemA"
+
+
+def test_find_and_flash_with_nothing_plugged_in(monkeypatch):
+    _flashable(monkeypatch, listings=((),))
+    assert cli.find_and_flash() == "not_found"
+