feat: add Ctrl-] x3 hotkey to power cycle board from serial console

python/packages/jumpstarter-driver-pyserial/README.md

@@ -46,6 +46,8 @@ export:
 | check_present | Check if the serial port exists during exporter initialization, disable if you are connecting to a dynamically created port (i.e. USB from your DUT) | bool  | no       | True    |
 | cps            | Characters per second throttling limit. When set, data transmission will be throttled to simulate slow typing. Useful for devices that can't handle fast input | float | no       | None    |
 | disable_hupcl  | Disable HUPCL on POSIX systems to avoid toggling DTR/RTS on close (can prevent MCU reset on serial disconnect)                                       | bool  | no       | False   |
+| power_control_ref | Explicit power device name from DUT tree for Ctrl-] x3 hotkey (skips auto-discovery). Only needed in multi-power setups | str | no | None (auto-discover) |
+| power_control_method | Power cycle method sequence for Ctrl-] x3 hotkey. Supports method names (`cycle`, `reset`, `on`, `off`) and `sleep:N` delays. Set to `[]` or `null` to disable | list[str] | no | `["cycle"]` |
 
 ### NVDemuxSerial Driver
 
@@ -161,7 +163,9 @@ Start an interactive serial console with direct terminal access.
 j serial start-console
 ```
 
-Exit the console by pressing CTRL+B three times.
+**Hotkeys:**
+- **CTRL+B x3**: Exit the console
+- **CTRL+] x3**: Power cycle the board (if a power driver is available in the DUT tree)
 
 ### pipe
 

python/packages/jumpstarter-driver-pyserial/jumpstarter_driver_pyserial/client.py

@@ -1,17 +1,29 @@
+import logging
 import sys
+import time
 from contextlib import contextmanager
 from typing import Optional
 
 import click
-from anyio import BrokenResourceError, EndOfStream, create_task_group, open_file
+from anyio import BrokenResourceError, EndOfStream, create_task_group, open_file, sleep, to_thread
 from anyio.streams.file import FileReadStream
 from jumpstarter_driver_network.adapters import PexpectAdapter
 from pexpect.fdpexpect import fdspawn
 
-from .console import Console
+from .console import Console, ConsoleStreamDrop
 from jumpstarter.client import DriverClient
 from jumpstarter.client.decorators import driver_click_group
 
+logger = logging.getLogger(__name__)
+
+KNOWN_POWER_CLIENTS = frozenset({
+    "jumpstarter_driver_power.client.PowerClient",
+    "jumpstarter_driver_power.client.VirtualPowerClient",
+    "jumpstarter_driver_ridesx.client.RideSXPowerClient",
+    "jumpstarter_driver_noyito_relay.client.NoyitoPowerClient",
+    "jumpstarter_driver_snmp.client.SNMPServerClient",
+})
+
 
 class PySerialClient(DriverClient):
     """
@@ -125,6 +137,94 @@ async def _stdin_to_serial(self, stream) -> tuple[int, int]:
 
         return bytes_read, bytes_sent
 
+    def _find_power_client(self):
+        # Check if hotkey is disabled
+        method_label = self.labels.get("jumpstarter.dev/pyserial/power-control-method", "cycle")
+        if not method_label:
+            return None
+
+        root = getattr(self, 'root', None)
+        if root is None:
+            return None
+
+        # Explicit ref takes precedence
+        ref_label = self.labels.get("jumpstarter.dev/pyserial/power-control-ref")
+        if ref_label:
+            power_client = root.children.get(ref_label)
+            if power_client is None:
+                logger.warning(
+                    "power_control_ref '%s' not found in DUT tree — power cycle hotkey disabled",
+                    ref_label
+                )
+                return None
+            return power_client
+
+        # Auto-discovery: collect all power-capable clients
+        candidates = []
+        self._collect_power_clients(root, candidates)
+
+        if len(candidates) == 0:
+            return None
+        if len(candidates) == 1:
+            return candidates[0]
+
+        # Multiple candidates — ambiguous
+        names = [c.labels.get("jumpstarter.dev/name", "unknown") for c in candidates]
+        logger.warning(
+            "Multiple power drivers found (%s) — power cycle hotkey disabled. "
+            "Set power_control_ref to select one explicitly.",
+            ", ".join(names)
+        )
+        return None
+
+    def _collect_power_clients(self, client, result, seen_uuids=None):
+        if seen_uuids is None:
+            seen_uuids = set()
+        client_uuid = getattr(client, 'uuid', None)
+        if client_uuid and client_uuid in seen_uuids:
+            return
+        if client_uuid:
+            seen_uuids.add(client_uuid)
+        client_class = client.labels.get("jumpstarter.dev/client")
+        if client_class in KNOWN_POWER_CLIENTS:
+            result.append(client)
+        for child in client.children.values():
+            self._collect_power_clients(child, result, seen_uuids)
+
+    def _make_power_cycle(self, power_client):
+        method_label = self.labels.get("jumpstarter.dev/pyserial/power-control-method", "cycle")
+        methods = [m for m in method_label.split(",") if m]
+
+        # Pre-validate and parse all steps
+        steps = []
+        for method in methods:
+            if method.startswith("sleep:"):
+                try:
+                    delay = float(method.split(":", 1)[1])
+                    steps.append(("sleep", delay))
+                except (ValueError, IndexError):
+                    logger.warning("Invalid sleep step '%s' — power cycle hotkey disabled", method)
+                    return None
+            else:
+                operation = getattr(power_client, method, None)
+                if callable(operation):
+                    steps.append(("operation", operation))
+                else:
+                    logger.warning(
+                        "Power client does not have callable method '%s' — power cycle hotkey disabled",
+                        method
+                    )
+                    return None
+
+        async def _cycle():
+            for kind, value in steps:
+                if kind == "sleep":
+                    await sleep(value)
+                else:
+                    await to_thread.run_sync(value)
+
+        return _cycle
+
     def cli(self):  # noqa: C901
         @driver_click_group(self)
         def base():
@@ -134,9 +234,23 @@ def base():
         @base.command()
         def start_console():
             """Start serial port console"""
+            power_client = self._find_power_client()
+            on_power_cycle = self._make_power_cycle(power_client) if power_client is not None else None
             click.echo("\nStarting serial port console ... exit with CTRL+B x 3 times\n")
-            console = Console(serial_client=self)
-            console.run()
+            if on_power_cycle is not None:
+                click.echo("Power cycle: CTRL+] x 3 times\n")
+            retries = 0
+            while retries < 30:
+                console = Console(serial_client=self, on_power_cycle=on_power_cycle)
+                try:
+                    console.run()
+                    break
+                except ConsoleStreamDrop:
+                    click.echo("\r\nSerial connection lost, reconnecting...\n", err=True)
+                    retries += 1
+                    time.sleep(1)
+            else:
+                click.echo("\nSerial connection lost (reconnect attempts exhausted).\n", err=True)
 
         @base.command()
         @click.option(

python/packages/jumpstarter-driver-pyserial/jumpstarter_driver_pyserial/client_test.py

@@ -0,0 +1,188 @@
+import threading
+from unittest.mock import MagicMock
+
+from .driver import PySerial
+from jumpstarter.common.utils import serve
+
+
+def test_find_power_client_no_root():
+    with serve(PySerial(url="loop://", power_control_method=["cycle"])) as client:
+        # No root attribute set → should return None
+        assert client._find_power_client() is None
+
+
+def test_find_power_client_auto_discover():
+    power = MagicMock(spec=["cycle", "children", "labels"])
+    power.children = {}
+    power.labels = {"jumpstarter.dev/client": "jumpstarter_driver_power.client.PowerClient"}
+    root = MagicMock(spec=["children", "labels"])
+    root.children = {"power": power}
+    root.labels = {}
+
+    with serve(PySerial(url="loop://")) as client:
+        object.__setattr__(client, "root", root)
+        assert client._find_power_client() is power
+
+
+def test_make_power_cycle_calls_cycle():
+    called = threading.Event()
+    power = MagicMock()
+    power.cycle = MagicMock(side_effect=lambda: called.set())
+
+    with serve(PySerial(url="loop://")) as client:
+        cycle_fn = client._make_power_cycle(power)
+        client.portal.call(cycle_fn)
+        assert called.is_set()
+
+
+def test_find_power_client_ambiguous():
+    power1 = MagicMock(spec=["children", "labels"])
+    power1.children = {}
+    power1.labels = {
+        "jumpstarter.dev/client": "jumpstarter_driver_power.client.PowerClient",
+        "jumpstarter.dev/name": "power1",
+    }
+    power2 = MagicMock(spec=["children", "labels"])
+    power2.children = {}
+    power2.labels = {
+        "jumpstarter.dev/client": "jumpstarter_driver_power.client.PowerClient",
+        "jumpstarter.dev/name": "power2",
+    }
+    root = MagicMock(spec=["children", "labels"])
+    root.children = {"power1": power1, "power2": power2}
+    root.labels = {}
+
+    with serve(PySerial(url="loop://")) as client:
+        object.__setattr__(client, "root", root)
+        assert client._find_power_client() is None
+
+
+def test_find_power_client_explicit_ref():
+    power = MagicMock(spec=["children", "labels"])
+    power.children = {}
+    power.labels = {"jumpstarter.dev/client": "jumpstarter_driver_power.client.PowerClient"}
+    other = MagicMock(spec=["children", "labels"])
+    other.children = {}
+    other.labels = {"jumpstarter.dev/client": "other.driver.Client"}
+    root = MagicMock(spec=["children", "labels"])
+    root.children = {"power": power, "other": other}
+    root.labels = {}
+
+    with serve(PySerial(url="loop://", power_control_ref="power")) as client:
+        object.__setattr__(client, "root", root)
+        assert client._find_power_client() is power
+
+
+def test_find_power_client_explicit_ref_missing():
+    root = MagicMock(spec=["children", "labels"])
+    root.children = {}
+    root.labels = {}
+
+    with serve(PySerial(url="loop://", power_control_ref="nonexistent")) as client:
+        object.__setattr__(client, "root", root)
+        assert client._find_power_client() is None
+
+
+def test_find_power_client_non_power_ignored():
+    gpio = MagicMock(spec=["children", "labels"])
+    gpio.children = {}
+    gpio.labels = {"jumpstarter.dev/client": "jumpstarter_driver_gpiod.client.DigitalOutputClient"}
+    root = MagicMock(spec=["children", "labels"])
+    root.children = {"gpio": gpio}
+    root.labels = {}
+
+    with serve(PySerial(url="loop://")) as client:
+        object.__setattr__(client, "root", root)
+        assert client._find_power_client() is None
+
+
+def test_make_power_cycle_custom_method():
+    called_sequence = []
+    power = MagicMock()
+    power.off = MagicMock(side_effect=lambda: called_sequence.append("off"))
+    power.on = MagicMock(side_effect=lambda: called_sequence.append("on"))
+
+    with serve(PySerial(url="loop://", power_control_method=["off", "on"])) as client:
+        cycle_fn = client._make_power_cycle(power)
+        client.portal.call(cycle_fn)
+        assert called_sequence == ["off", "on"]
+
+
+def test_find_power_client_disabled_via_empty_list():
+    power = MagicMock(spec=["children", "labels"])
+    power.children = {}
+    power.labels = {"jumpstarter.dev/client": "jumpstarter_driver_power.client.PowerClient"}
+    root = MagicMock(spec=["children", "labels"])
+    root.children = {"power": power}
+    root.labels = {}
+
+    with serve(PySerial(url="loop://", power_control_method=[])) as client:
+        object.__setattr__(client, "root", root)
+        assert client._find_power_client() is None
+
+
+def test_make_power_cycle_missing_method():
+    power = MagicMock(spec=["children", "labels"])
+    power.children = {}
+    power.labels = {}
+
+    with serve(PySerial(url="loop://", power_control_method=["nonexistent_method"])) as client:
+        result = client._make_power_cycle(power)
+        assert result is None
+
+
+def test_make_power_cycle_with_sleep():
+    called_sequence = []
+    power = MagicMock()
+    power.off = MagicMock(side_effect=lambda: called_sequence.append("off"))
+    power.on = MagicMock(side_effect=lambda: called_sequence.append("on"))
+
+    with serve(PySerial(url="loop://", power_control_method=["off", "sleep:0.01", "on"])) as client:
+        cycle_fn = client._make_power_cycle(power)
+        client.portal.call(cycle_fn)
+        assert called_sequence == ["off", "on"]
+
+
+def test_find_power_client_disabled_via_none():
+    power = MagicMock(spec=["children", "labels"])
+    power.children = {}
+    power.labels = {"jumpstarter.dev/client": "jumpstarter_driver_power.client.PowerClient"}
+    root = MagicMock(spec=["children", "labels"])
+    root.children = {"power": power}
+    root.labels = {}
+
+    with serve(PySerial(url="loop://", power_control_method=None)) as client:
+        object.__setattr__(client, "root", root)
+        assert client._find_power_client() is None
+
+
+def test_collect_power_clients_dedup_proxy():
+    # Simulate Proxy scenario: same power driver instance appears twice in tree
+    # (once via proxy delegation, once via direct parent)
+    from uuid import uuid4
+    shared_uuid = uuid4()
+
+    power1 = MagicMock(spec=["children", "labels", "uuid"])
+    power1.children = {}
+    power1.labels = {
+        "jumpstarter.dev/client": "jumpstarter_driver_power.client.PowerClient",
+        "jumpstarter.dev/name": "power",
+    }
+    power1.uuid = shared_uuid
+
+    power2 = MagicMock(spec=["children", "labels", "uuid"])
+    power2.children = {}
+    power2.labels = {
+        "jumpstarter.dev/client": "jumpstarter_driver_power.client.PowerClient",
+        "jumpstarter.dev/name": "power",
+    }
+    power2.uuid = shared_uuid  # Same UUID as power1
+
+    root = MagicMock(spec=["children", "labels"])
+    root.children = {"power1": power1, "power2": power2}
+    root.labels = {}
+
+    with serve(PySerial(url="loop://")) as client:
+        object.__setattr__(client, "root", root)
+        # Should find only one power client despite two references
+        assert client._find_power_client() is power1