From 19222328e2acd98807a399ab9aa5f670b1319fdc Mon Sep 17 00:00:00 2001 From: Fabian Pflug Date: Sat, 25 Jul 2026 21:08:09 +0200 Subject: [PATCH 01/10] doc: remove ResetProtocol from SerialDriver the SerialDriver does not support the ResetProtocol, so remove it from documentation. Signed-off-by: Fabian Pflug --- doc/configuration.rst | 1 - 1 file changed, 1 deletion(-) diff --git a/doc/configuration.rst b/doc/configuration.rst index 57b23f13c..766bd72f4 100644 --- a/doc/configuration.rst +++ b/doc/configuration.rst @@ -1971,7 +1971,6 @@ Binds to: Implements: - :any:`ConsoleProtocol` - - :any:`ResetProtocol` Arguments: - txdelay (float, default=0.0): time in seconds to wait before sending a chunk From 91524e1b66a5ee04301f7bb0091ceb0c3eb4d5ee Mon Sep 17 00:00:00 2001 From: Fabian Pflug Date: Sat, 25 Jul 2026 13:21:15 +0200 Subject: [PATCH 02/10] resource: udev: USBDebugger: add keil CMIS-DAPLink A CMSIS DAP cannot start a gdb server, but it can bootstrap a DUT. Signed-off-by: Fabian Pflug --- labgrid/resource/udev.py | 1 + 1 file changed, 1 insertion(+) diff --git a/labgrid/resource/udev.py b/labgrid/resource/udev.py index 7ecd4b01a..517b00bd9 100644 --- a/labgrid/resource/udev.py +++ b/labgrid/resource/udev.py @@ -932,6 +932,7 @@ def filter_match(self, device): ("064b", "2507"), # Analog Devices Onboard Debug Agent ("064b", "2508"), # Analog Devices Onboard Debug Agent ("064b", "250A"), # Analog Devices Onboard Debug Agent + ("c251", "f001"), # Keil CMSIS-DAP-Link ]: return False From 0884f65c1203c44bde9b8606c1987bc699b83379 Mon Sep 17 00:00:00 2001 From: Fabian Pflug Date: Sat, 25 Jul 2026 13:19:02 +0200 Subject: [PATCH 03/10] driver: pyocd: add driver for bootstrapping The PyOCD tool is used for example by the zephyr project to flash the devices. Add support for pyocd to bootstrap and reset a device under test. Signed-off-by: Fabian Pflug --- doc/configuration.rst | 35 +++++++++++ doc/man/device-config.rst | 4 ++ labgrid/driver/__init__.py | 1 + labgrid/driver/pyocddriver.py | 106 ++++++++++++++++++++++++++++++++++ man/labgrid-device-config.5 | 4 ++ 5 files changed, 150 insertions(+) create mode 100644 labgrid/driver/pyocddriver.py diff --git a/doc/configuration.rst b/doc/configuration.rst index 766bd72f4..35ab3ab3b 100644 --- a/doc/configuration.rst +++ b/doc/configuration.rst @@ -991,6 +991,7 @@ Arguments: Used by: - `OpenOCDDriver`_ + - `PyOCDDriver`_ NetworkUSBDebugger ~~~~~~~~~~~~~~~~~~ @@ -2330,6 +2331,40 @@ Arguments: - board_config (str): optional, board config in the ``openocd/scripts/board/`` directory - load_commands (list of str): optional, load commands to use instead of ``init``, ``bootstrap {filename}``, ``shutdown`` +PyOCDDriver +~~~~~~~~~~~~~ +An :any:`PyOCDDriver` controls *PyOCD* to bootstrap a target with a bootloader. + +Binds to: + interface: + - `USBDebugger`_ + - `NetworkUSBDebugger`_ + +Implements: + - :any:`BootstrapProtocol` + - :any:`ResetProtocol` + +.. code-block:: yaml + + PyOCDDriver: + image: 'bitstream' + target_name: 'nrf52832' + frequency: "400khz" + config: 'pyocd.yaml' + serial: '1234567890' + load_commands: + - 'init' + - 'svf -quiet {filename}' + - 'exit' + +Arguments: + - image (str): optional, name of the image to bootstrap onto the device + - target_name (str): optional, set the target type. Should be one of the targets listed by 'pyocd list --targets'. + - frequency (str): optional, SWD/JTAG clock frequency in Hz + - config (str): optional, PyOCD configuration file + - serial (str): optional, USB-Serial to identify device if multiple are present. + - load_commands (list of str): optional, additional command line parameters for load to use instead of ``-e``, ``sector`` + QuartusHPSDriver ~~~~~~~~~~~~~~~~ A :any:`QuartusHPSDriver` controls the "Quartus Prime Programmer and Tools" to diff --git a/doc/man/device-config.rst b/doc/man/device-config.rst index d0e5ff914..d3ca9e0f1 100644 --- a/doc/man/device-config.rst +++ b/doc/man/device-config.rst @@ -110,6 +110,10 @@ TOOLS KEYS Path to the openocd binary, used by the OpenOCDDriver. See: https://openocd.org/ +``pyocd`` + Path to the pyocd binary, used by the PyOCDDriver. + See: https://pyocd.io/ + ``quartus_hps`` Path to the quartus_hps binary, used by the QuartusHPSDriver. See: https://www.intel.com/content/www/us/en/docs/programmable/683039/22-3/hps-flash-programmer.html diff --git a/labgrid/driver/__init__.py b/labgrid/driver/__init__.py index d3cd6f55e..75cb54afb 100644 --- a/labgrid/driver/__init__.py +++ b/labgrid/driver/__init__.py @@ -9,6 +9,7 @@ from .fastbootdriver import AndroidFastbootDriver from .dfudriver import DFUDriver from .openocddriver import OpenOCDDriver +from .pyocddriver import PyOCDDriver from .quartushpsdriver import QuartusHPSDriver from .flashromdriver import FlashromDriver from .onewiredriver import OneWirePIODriver diff --git a/labgrid/driver/pyocddriver.py b/labgrid/driver/pyocddriver.py new file mode 100644 index 000000000..f833f6f2f --- /dev/null +++ b/labgrid/driver/pyocddriver.py @@ -0,0 +1,106 @@ +import attr + +from ..factory import target_factory +from ..protocol import BootstrapProtocol, ResetProtocol +from ..step import step +from ..util.managedfile import ManagedFile +from ..util.helper import processwrapper +from .common import Driver + + +@target_factory.reg_driver +@attr.s(eq=False) +class PyOCDDriver(Driver, BootstrapProtocol, ResetProtocol): + + priorities = {ResetProtocol: 5} + + bindings = { + "interface": { + "USBDebugger", + "NetworkUSBDebugger", + }, + } + + image = attr.ib( + default=None, + validator=attr.validators.optional(attr.validators.instance_of(str)), + ) + load_commands = attr.ib( + default=None, + validator=attr.validators.optional(attr.validators.instance_of((str, list))), + ) + target_name = attr.ib( + default=None, + validator=attr.validators.optional(attr.validators.instance_of(str)), + ) + frequency = attr.ib( + default=None, + validator=attr.validators.optional(attr.validators.instance_of(str)), + ) + config = attr.ib( + default=None, + validator=attr.validators.optional(attr.validators.instance_of(str)), + ) + serial = attr.ib( + default=None, + validator=attr.validators.optional(attr.validators.instance_of(str)), + ) + + def __attrs_post_init__(self): + super().__attrs_post_init__() + + # FIXME make sure we always have an environment or config + if self.target.env: + self.tool = self.target.env.config.get_tool("pyocd") + self.config = self.target.env.config.resolve_path(self.config) + else: + self.tool = "pyocd" + + def _run_commands(self, subcommand: str, commands: list | None = None): + cmd = [self.tool, subcommand] + if self.serial: + cmd += ["--uid", self.serial] + if self.target_name is not None and "--target" not in commands: + cmd += ["--target", self.target_name] + if self.frequency is not None and "--frequency" not in commands: + cmd += ["--frequency", self.frequency] + + if self.config is not None: + mconfig = ManagedFile(self.config, self.interface) + mconfig.sync_to_resource() + cmd += ["--config", mconfig.get_remote_path()] + else: + cmd += ["--no-config"] + + if commands: + cmd += commands + processwrapper.check_output( + self.interface.wrap_command(cmd), print_on_silent_log=True + ) + + @Driver.check_active + @step(args=["filename"]) + def load(self, filename=None): + + if filename is None and self.image is not None: + filename = self.target.env.config.get_image_path(self.image) + + mf = ManagedFile(filename, self.interface) + mf.sync_to_resource() + + if self.load_commands: + if isinstance(self.load_commands, str): + commands = self.load_commands.split() + else: + commands = self.load_commands + else: + commands = ["-e", "sector"] + + commands.append(mf.get_remote_path()) + + self._run_commands("load", commands) + + @Driver.check_active + @step() + def reset(self): + self._run_commands("reset") diff --git a/man/labgrid-device-config.5 b/man/labgrid-device-config.5 index 0755927de..c989e4fe9 100644 --- a/man/labgrid-device-config.5 +++ b/man/labgrid-device-config.5 @@ -125,6 +125,10 @@ See: \X'tty: link https://git.pengutronix.de/cgit/barebox/tree/scripts/mxs-usb-l Path to the openocd binary, used by the OpenOCDDriver. See: \X'tty: link https://openocd.org/'\fI\%https://openocd.org/\fP\X'tty: link' .TP +.B \fBpyocd\fP +Path to the pyocd binary, used by the PyOCDDriver. +See: \X'tty: link https://pyocd.io/'\fI\%https://pyocd.io/\fP\X'tty: link' +.TP .B \fBquartus_hps\fP Path to the quartus_hps binary, used by the QuartusHPSDriver. See: \X'tty: link https://www.intel.com/content/www/us/en/docs/programmable/683039/22-3/hps-flash-programmer.html'\fI\%https://www.intel.com/content/www/us/en/docs/programmable/683039/22\-3/hps\-flash\-programmer.html\fP\X'tty: link' From 114363a2f190fc42ad9654f4e315cfe01fb3d944 Mon Sep 17 00:00:00 2001 From: Fabian Pflug Date: Sun, 26 Jul 2026 00:09:12 +0200 Subject: [PATCH 04/10] tests: add tests for pyocd Add tests for pyocd and fix errors found during testing. Since the PyOCDDriver will call the pyocd binary, most tests mock away the call to pyocd, since it will only work with hardware attached. Signed-off-by: Fabian Pflug --- labgrid/driver/pyocddriver.py | 12 +- tests/test_pyocd.py | 220 ++++++++++++++++++++++++++++++++++ 2 files changed, 227 insertions(+), 5 deletions(-) create mode 100644 tests/test_pyocd.py diff --git a/labgrid/driver/pyocddriver.py b/labgrid/driver/pyocddriver.py index f833f6f2f..39e3118d0 100644 --- a/labgrid/driver/pyocddriver.py +++ b/labgrid/driver/pyocddriver.py @@ -52,7 +52,8 @@ def __attrs_post_init__(self): # FIXME make sure we always have an environment or config if self.target.env: self.tool = self.target.env.config.get_tool("pyocd") - self.config = self.target.env.config.resolve_path(self.config) + if self.config: + self.config = self.target.env.config.resolve_path(self.config) else: self.tool = "pyocd" @@ -60,9 +61,9 @@ def _run_commands(self, subcommand: str, commands: list | None = None): cmd = [self.tool, subcommand] if self.serial: cmd += ["--uid", self.serial] - if self.target_name is not None and "--target" not in commands: + if self.target_name is not None and (not commands or "--target" not in commands): cmd += ["--target", self.target_name] - if self.frequency is not None and "--frequency" not in commands: + if self.frequency is not None and (not commands or "--frequency" not in commands): cmd += ["--frequency", self.frequency] if self.config is not None: @@ -75,14 +76,15 @@ def _run_commands(self, subcommand: str, commands: list | None = None): if commands: cmd += commands processwrapper.check_output( - self.interface.wrap_command(cmd), print_on_silent_log=True + command=self.interface.wrap_command(cmd), + print_on_silent_log=True, ) @Driver.check_active @step(args=["filename"]) def load(self, filename=None): - if filename is None and self.image is not None: + if filename is None and self.image is not None and self.target.env: filename = self.target.env.config.get_image_path(self.image) mf = ManagedFile(filename, self.interface) diff --git a/tests/test_pyocd.py b/tests/test_pyocd.py new file mode 100644 index 000000000..1a12f0a3d --- /dev/null +++ b/tests/test_pyocd.py @@ -0,0 +1,220 @@ +import pytest + +from labgrid import Environment, Target +from labgrid.resource.udev import USBDebugger +from labgrid.driver.pyocddriver import PyOCDDriver + +LABGRID_TEST_SERIAL = "LABGRID_TEST_SER" +LABGRID_TARGET = "target_mcu" +LABGRID_TOOL_CMD = "/some/path/to/nowhere" + + +def test_pyocd_driver_activate(target): + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None) + target.activate(d) + + +def test_pyocd_driver_reset(target, mocker): + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None) + target.activate(d) + + check_output_mock = mocker.patch("labgrid.util.helper.processwrapper.check_output") + + d.reset() + + check_output_mock.assert_called_once() + cmd = check_output_mock.call_args.kwargs["command"] + assert cmd == ["pyocd", "reset", "--no-config"] + assert check_output_mock.call_args.kwargs["print_on_silent_log"] + + +def test_pyocd_driver_load(target, mocker): + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None) + target.activate(d) + + check_output_mock = mocker.patch("labgrid.util.helper.processwrapper.check_output") + + d.load(__file__) + + check_output_mock.assert_called_once() + cmd = check_output_mock.call_args.kwargs["command"] + assert cmd == ["pyocd", "load", "--no-config", "-e", "sector", __file__] + assert check_output_mock.call_args.kwargs["print_on_silent_log"] + + +def test_pyocd_load_error_on_missing_file(target, mocker): + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None) + target.activate(d) + + check_output_mock = mocker.patch("labgrid.util.helper.processwrapper.check_output") + + with pytest.raises(FileNotFoundError): + d.load() + + check_output_mock.assert_not_called() + + +def test_pyocd_load_error_on_missing_target_env(target, mocker): + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None, image="test") + target.activate(d) + + check_output_mock = mocker.patch("labgrid.util.helper.processwrapper.check_output") + + with pytest.raises(FileNotFoundError): + d.load() + + check_output_mock.assert_not_called() + + +def test_pyocd_load_image_on_missing_file(tmpdir, mocker): + p = tmpdir.join("config.yaml") + p.write(f""" + images: + test: {__file__} + """) + env = Environment(str(p)) + target = Target("test", env=env) + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None, image="test") + target.activate(d) + + check_output_mock = mocker.patch("labgrid.util.helper.processwrapper.check_output") + + d.load() + + check_output_mock.assert_called_once() + cmd = check_output_mock.call_args.kwargs["command"] + assert cmd[0] == "pyocd" + assert cmd[1] == "load" + assert cmd[-1] == __file__ + + +@pytest.mark.parametrize( + "args,params,missing", + [ + ({"serial": LABGRID_TEST_SERIAL}, ["--uid", LABGRID_TEST_SERIAL], []), + ({"target_name": LABGRID_TARGET}, ["--target", LABGRID_TARGET], []), + ({"config": __file__}, ["--config", __file__], ["--no-config"]), + # make sure, that it is not the file loaded argument + ({}, ["--no-config"], [__file__]), + ({"frequency": "400000"}, ["--frequency", "400000"], []), + # load commands should take precedens + ({"frequency": "400000", "load_commands": ["--frequency", "100"]}, ["--frequency", "100"], ["400000"]), + # even as a string instead of array + ({"frequency": "400000", "load_commands": "--frequency 100"}, ["--frequency", "100"], ["400000"]), + # load commands overwrite default load commands + ({"load_commands": ["--frequency", "100"]}, ["--frequency", "100"], ["-e", "sector"]), + ], +) +def test_pyocd_driver_params_load(target, mocker, args, params, missing): + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None, **args) + target.activate(d) + + check_output_mock = mocker.patch("labgrid.util.helper.processwrapper.check_output") + + d.load(__file__) + + check_output_mock.assert_called_once() + cmd = check_output_mock.call_args.kwargs["command"] + assert cmd[0] == "pyocd" + assert cmd[1] == "load" + assert cmd[-1] == __file__ + for p in params: + assert p in cmd[2:-1] + for m in missing: + assert m not in cmd[2:-1] + assert check_output_mock.call_args.kwargs["print_on_silent_log"] + + +@pytest.mark.parametrize( + "args,params,missing", + [ + ({"serial": LABGRID_TEST_SERIAL}, ["--uid", LABGRID_TEST_SERIAL], []), + ({"target_name": LABGRID_TARGET}, ["--target", LABGRID_TARGET], []), + ({"config": __file__}, ["--config", __file__], ["--no-config"]), + ({"frequency": "400000"}, ["--frequency", "400000"], []), + # load commands not be used + ({"frequency": "400000", "load_commands": ["--frequency", "100"]}, ["--frequency", "400000"], ["100"]), + ({"load_commands": ["--frequency", "100"]}, [], ["--frequency", "100"]), + ], +) +def test_pyocd_driver_params_reset(tmpdir, mocker, args, params, missing): + p = tmpdir.join("config.yaml") + p.write(""" + dict: {} + """) + env = Environment(str(p)) + target = Target("test", env=env) + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None, **args) + target.activate(d) + + check_output_mock = mocker.patch("labgrid.util.helper.processwrapper.check_output") + + d.reset() + + check_output_mock.assert_called_once() + cmd = check_output_mock.call_args.kwargs["command"] + assert cmd[0] == "pyocd" + assert cmd[1] == "reset" + for p in params: + assert p in cmd[2:] + for m in missing: + assert m not in cmd[2:] + assert check_output_mock.call_args.kwargs["print_on_silent_log"] + + +def test_pyocd_respects_env_tool(mocker, tmpdir): + p = tmpdir.join("config.yaml") + p.write(f""" + tools: + pyocd: {LABGRID_TOOL_CMD} + """) + env = Environment(str(p)) + target = Target("test", env=env) + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None) + target.activate(d) + + assert d.tool == LABGRID_TOOL_CMD + + # Make sure, that the tool actually is used in command + check_output_mock = mocker.patch("labgrid.util.helper.processwrapper.check_output") + + d.reset() + + check_output_mock.assert_called_once() + cmd = check_output_mock.call_args.kwargs["command"] + assert cmd[0] == LABGRID_TOOL_CMD + assert cmd[1] == "reset" + + +def test_pyocd_respects_missing_env_tool(tmpdir): + p = tmpdir.join("config.yaml") + p.write(f""" + tools: + none: {LABGRID_TOOL_CMD} + """) + env = Environment(str(p)) + target = Target("test", env=env) + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None) + target.activate(d) + + assert d.tool == "pyocd" From 7479a7f1b934886851663416854f0077707ebf88 Mon Sep 17 00:00:00 2001 From: Fabian Pflug Date: Sat, 25 Jul 2026 11:10:43 +0200 Subject: [PATCH 05/10] remote/client: bootstrap: add USBDebugger The PyOCDDriver does support the BootstrapProtocol and could be used to flash the device. Signed-off-by: Fabian Pflug --- labgrid/remote/client.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/labgrid/remote/client.py b/labgrid/remote/client.py index 069240b9e..50e0ddd91 100755 --- a/labgrid/remote/client.py +++ b/labgrid/remote/client.py @@ -1176,8 +1176,9 @@ def bootstrap(self): NetworkIMXUSBLoader, NetworkRKUSBLoader, NetworkAlteraUSBBlaster, + NetworkUSBDebugger, ) - from ..driver import OpenOCDDriver + from ..driver import OpenOCDDriver, PyOCDDriver drv = None try: @@ -1199,6 +1200,13 @@ def bootstrap(self): except NoDriverFoundError: drv = OpenOCDDriver(target, name=name, **args) drv.interface.timeout = self.args.wait + elif isinstance(resource, NetworkUSBDebugger): + args = dict(arg.split("=", 1) for arg in self.args.bootstrap_args) + try: + drv = target.get_driver("PyOCDDriver", activate=False, name=name) + except NoDriverFoundError: + drv = PyOCDDriver(target, name=name, **args) + drv.interface.timeout = self.args.wait elif isinstance(resource, NetworkRKUSBLoader): drv = self._get_driver_or_new(target, "RKUSBDriver", activate=False, name=name) drv.loader.timeout = self.args.wait From 248f7ff2d3f9f402b5470eba79d7a89014520b2d Mon Sep 17 00:00:00 2001 From: Fabian Pflug Date: Sat, 25 Jul 2026 21:17:03 +0200 Subject: [PATCH 06/10] remote/client: create function to get PowerDriver Move the search for a power driver into a seperate function, which also respects the priority of power drivers. Not much of an improvement at the moment, as all have the same priority, but could maybe done later. This is in preparation for the reset command in client. Signed-off-by: Fabian Pflug --- labgrid/remote/client.py | 56 +++++++++++++++++++++++----------------- 1 file changed, 33 insertions(+), 23 deletions(-) diff --git a/labgrid/remote/client.py b/labgrid/remote/client.py index 50e0ddd91..e9c67a672 100755 --- a/labgrid/remote/client.py +++ b/labgrid/remote/client.py @@ -932,40 +932,50 @@ def _get_driver_or_new(self, target, cls, *, name=None, activate=True): target.activate(drv) return drv + def _get_power_driver_for_resource(self, target, name=None, priority_cls=None): + from ..resource.power import NetworkPowerPort, PDUDaemonPort + from ..resource.remote import NetworkUSBPowerPort, NetworkSiSPMPowerPort, NetworkSysfsGPIO + from ..resource import TasmotaPowerPort, NetworkYKUSHPowerPort + + drv = None + for resource in target.resources: + if name and resource.name != name: + continue + temp_drv = None + if isinstance(resource, NetworkPowerPort): + temp_drv = self._get_driver_or_new(target, "NetworkPowerDriver", name=name) + elif isinstance(resource, NetworkUSBPowerPort): + temp_drv = self._get_driver_or_new(target, "USBPowerDriver", name=name) + elif isinstance(resource, NetworkSiSPMPowerPort): + temp_drv = self._get_driver_or_new(target, "SiSPMPowerDriver", name=name) + elif isinstance(resource, PDUDaemonPort): + temp_drv = self._get_driver_or_new(target, "PDUDaemonDriver", name=name) + elif isinstance(resource, TasmotaPowerPort): + temp_drv = self._get_driver_or_new(target, "TasmotaPowerDriver", name=name) + elif isinstance(resource, NetworkYKUSHPowerPort): + temp_drv = self._get_driver_or_new(target, "YKUSHPowerDriver", name=name) + elif isinstance(resource, NetworkSysfsGPIO): + self._get_driver_or_new(target, "GpioDigitalOutputDriver", name=name) + temp_drv = self._get_driver_or_new(target, "DigitalOutputPowerDriver", name=name) + if temp_drv: + if not priority_cls: + return temp_drv + if not drv or temp_drv.get_priority(priority_cls) > drv.get_priority(priority_cls): + drv = temp_drv + return drv + def power(self): place = self.get_acquired_place() action = self.args.action delay = self.args.delay name = self.args.name target = self._get_target(place) - from ..resource.power import NetworkPowerPort, PDUDaemonPort - from ..resource.remote import NetworkUSBPowerPort, NetworkSiSPMPowerPort, NetworkSysfsGPIO - from ..resource import TasmotaPowerPort, NetworkYKUSHPowerPort drv = None try: drv = target.get_driver("PowerProtocol", name=name) except NoDriverFoundError: - for resource in target.resources: - if name and resource.name != name: - continue - if isinstance(resource, NetworkPowerPort): - drv = self._get_driver_or_new(target, "NetworkPowerDriver", name=name) - elif isinstance(resource, NetworkUSBPowerPort): - drv = self._get_driver_or_new(target, "USBPowerDriver", name=name) - elif isinstance(resource, NetworkSiSPMPowerPort): - drv = self._get_driver_or_new(target, "SiSPMPowerDriver", name=name) - elif isinstance(resource, PDUDaemonPort): - drv = self._get_driver_or_new(target, "PDUDaemonDriver", name=name) - elif isinstance(resource, TasmotaPowerPort): - drv = self._get_driver_or_new(target, "TasmotaPowerDriver", name=name) - elif isinstance(resource, NetworkYKUSHPowerPort): - drv = self._get_driver_or_new(target, "YKUSHPowerDriver", name=name) - elif isinstance(resource, NetworkSysfsGPIO): - self._get_driver_or_new(target, "GpioDigitalOutputDriver", name=name) - drv = self._get_driver_or_new(target, "DigitalOutputPowerDriver", name=name) - if drv: - break + drv = self._get_power_driver_for_resource(target, name, None) if not drv: raise UserError("target has no compatible resource available") From 1bdb5a3c1b8fad3ee44c2cca6a9e1e094358be49 Mon Sep 17 00:00:00 2001 From: Fabian Pflug Date: Sat, 25 Jul 2026 21:28:15 +0200 Subject: [PATCH 07/10] remote/client: create function to get IODriver Move the search for IO driver into a seperate function, that also handles priorities. This is in preparation of the reset command. Signed-off-by: Fabian Pflug --- labgrid/remote/client.py | 72 +++++++++++++++++++++++----------------- 1 file changed, 41 insertions(+), 31 deletions(-) diff --git a/labgrid/remote/client.py b/labgrid/remote/client.py index e9c67a672..075fe8308 100755 --- a/labgrid/remote/client.py +++ b/labgrid/remote/client.py @@ -964,6 +964,46 @@ def _get_power_driver_for_resource(self, target, name=None, priority_cls=None): drv = temp_drv return drv + def _get_io_driver_for_resource(self, target, name=None, priority_cls=None): + from ..resource import ( + ModbusTCPCoil, + OneWirePIO, + HttpDigitalOutput, + WaveshareModbusTCPCoil, + Eth008DigitalOutput, + ) + from ..resource.remote import NetworkDeditecRelais8, NetworkSysfsGPIO, NetworkLXAIOBusPIO, NetworkHIDRelay + + drv = None + for resource in target.resources: + if name and resource.name != name: + continue + temp_drv = None + if isinstance(resource, WaveshareModbusTCPCoil): + temp_drv = self._get_driver_or_new(target, "WaveShareModbusCoilDriver", name=name) + elif isinstance(resource, ModbusTCPCoil): + temp_drv = self._get_driver_or_new(target, "ModbusCoilDriver", name=name) + elif isinstance(resource, Eth008DigitalOutput): + temp_drv = self._get_driver_or_new(target, "Eth008DigitalOutputDriver", name=name) + elif isinstance(resource, OneWirePIO): + temp_drv = self._get_driver_or_new(target, "OneWirePIODriver", name=name) + elif isinstance(resource, HttpDigitalOutput): + temp_drv = self._get_driver_or_new(target, "HttpDigitalOutputDriver", name=name) + elif isinstance(resource, NetworkDeditecRelais8): + temp_drv = self._get_driver_or_new(target, "DeditecRelaisDriver", name=name) + elif isinstance(resource, NetworkSysfsGPIO): + temp_drv = self._get_driver_or_new(target, "GpioDigitalOutputDriver", name=name) + elif isinstance(resource, NetworkLXAIOBusPIO): + temp_drv = self._get_driver_or_new(target, "LXAIOBusPIODriver", name=name) + elif isinstance(resource, NetworkHIDRelay): + temp_drv = self._get_driver_or_new(target, "HIDRelayDriver", name=name) + if temp_drv: + if not priority_cls: + return temp_drv + if not drv or temp_drv.get_priority(priority_cls) > drv.get_priority(priority_cls): + drv = temp_drv + return drv + def power(self): place = self.get_acquired_place() action = self.args.action @@ -990,42 +1030,12 @@ def digital_io(self): action = self.args.action name = self.args.name target = self._get_target(place) - from ..resource import ( - ModbusTCPCoil, - OneWirePIO, - HttpDigitalOutput, - WaveshareModbusTCPCoil, - Eth008DigitalOutput, - ) - from ..resource.remote import NetworkDeditecRelais8, NetworkSysfsGPIO, NetworkLXAIOBusPIO, NetworkHIDRelay drv = None try: drv = target.get_driver("DigitalOutputProtocol", name=name) except NoDriverFoundError: - for resource in target.resources: - if name and resource.name != name: - continue - if isinstance(resource, WaveshareModbusTCPCoil): - drv = self._get_driver_or_new(target, "WaveShareModbusCoilDriver", name=name) - elif isinstance(resource, ModbusTCPCoil): - drv = self._get_driver_or_new(target, "ModbusCoilDriver", name=name) - elif isinstance(resource, Eth008DigitalOutput): - drv = self._get_driver_or_new(target, "Eth008DigitalOutputDriver", name=name) - elif isinstance(resource, OneWirePIO): - drv = self._get_driver_or_new(target, "OneWirePIODriver", name=name) - elif isinstance(resource, HttpDigitalOutput): - drv = self._get_driver_or_new(target, "HttpDigitalOutputDriver", name=name) - elif isinstance(resource, NetworkDeditecRelais8): - drv = self._get_driver_or_new(target, "DeditecRelaisDriver", name=name) - elif isinstance(resource, NetworkSysfsGPIO): - drv = self._get_driver_or_new(target, "GpioDigitalOutputDriver", name=name) - elif isinstance(resource, NetworkLXAIOBusPIO): - drv = self._get_driver_or_new(target, "LXAIOBusPIODriver", name=name) - elif isinstance(resource, NetworkHIDRelay): - drv = self._get_driver_or_new(target, "HIDRelayDriver", name=name) - if drv: - break + drv = self._get_io_driver_for_resource(target, name, None) if not drv: raise UserError("target has no compatible resource available") From 9ec99741a784e9d15fba305125c05e4c11dd51b4 Mon Sep 17 00:00:00 2001 From: Fabian Pflug Date: Sat, 25 Jul 2026 21:51:07 +0200 Subject: [PATCH 08/10] remote/client: add reset subcommand Add reset subcommand to reset the board. With MCU development, the Debugger, Serial and Power can all be delivered through one USB-Port. In order to see the bootlog messages, a reset command needs to be send to the DUT without powercycling it, as this would also powercycle the USB->UART converter and defeat the point. Since the PyOCDDriver supports reset as a NetworkUSBDebugger it should be the prefered method, but do allow all other sources, that implement the ResetProtocol. Signed-off-by: Fabian Pflug --- labgrid/remote/client.py | 35 +++++++++++++++++++++++++++++++++++ man/labgrid-client.1 | 16 ++++++++++++++++ 2 files changed, 51 insertions(+) diff --git a/labgrid/remote/client.py b/labgrid/remote/client.py index 075fe8308..41f4391ca 100755 --- a/labgrid/remote/client.py +++ b/labgrid/remote/client.py @@ -1046,6 +1046,37 @@ def digital_io(self): elif action == "low": drv.set(False) + def reset(self): + place = self.get_acquired_place() + name = self.args.name + target = self._get_target(place) + from ..resource.remote import NetworkUSBDebugger + from ..protocol import ResetProtocol + + drv = None + try: + drv = target.get_driver(ResetProtocol, name=name) + except NoDriverFoundError: + drv = self._get_power_driver_for_resource(target, name, ResetProtocol) + io_drv = self._get_io_driver_for_resource(target, name, None) + if io_drv: + temp_drv = self._get_driver_or_new(target, "DigitalOutputResetDriver", name=name) + if temp_drv and (not drv or temp_drv.get_priority(ResetProtocol) > drv.get_priority(ResetProtocol)): + drv = temp_drv + for resource in target.resources: + if name and resource.name != name: + continue + temp_drv = None + if isinstance(resource, NetworkUSBDebugger): + temp_drv = self._get_driver_or_new(target, "PyOCDDriver", name=name) + if temp_drv: + if not drv or temp_drv.get_priority(ResetProtocol) > drv.get_priority(ResetProtocol): + drv = temp_drv + + if not drv: + raise UserError("target has no compatible resource available") + drv.reset() + async def _console(self, place, target, timeout, *, logfile=None, loop=False, listen_only=False): name = self.args.name from ..resource import NetworkSerialPort @@ -2017,6 +2048,10 @@ def get_parser(auto_doc_mode=False) -> "argparse.ArgumentParser | AutoProgramArg subparser.add_argument("--name", "-n", help="optional resource name") subparser.set_defaults(func=ClientSession.power) + subparser = subparsers.add_parser("reset", help="reset the target") + subparser.add_argument("--name", "-n", help="optional resource name") + subparser.set_defaults(func=ClientSession.reset) + subparser = subparsers.add_parser("io", help="change (or get) a digital IO status") subparser.add_argument("action", choices=["high", "low", "get"], help="action") subparser.add_argument("name", help="optional resource name", nargs="?") diff --git a/man/labgrid-client.1 b/man/labgrid-client.1 index 45533af05..befda6053 100644 --- a/man/labgrid-client.1 +++ b/man/labgrid-client.1 @@ -651,6 +651,22 @@ format output as shell variables .B \-\-prio priority relative to other reservations (default 0) .UNINDENT +.SS labgrid\-client reset +.sp +reset the target +.INDENT 0.0 +.INDENT 3.5 +.sp +.EX +usage: labgrid\-client reset [\-\-name NAME] +.EE +.UNINDENT +.UNINDENT +.INDENT 0.0 +.TP +.B \-\-name , \-n +optional resource name +.UNINDENT .SS labgrid\-client resources|r .sp list available resources From 72b8d92d0102d78cbfcb7a820150e125e5ce63fc Mon Sep 17 00:00:00 2001 From: Fabian Pflug Date: Sun, 26 Jul 2026 19:42:21 +0200 Subject: [PATCH 09/10] tests: add test for labgrid-client subcommands Mock out the connection to the coordinator and ressource updates, because they are handled inside the testcase. Check that the right ressources get created and used during execution and that an error is raised, if no resources are available. Signed-off-by: Fabian Pflug --- tests/test_client_commands.py | 176 ++++++++++++++++++++++++++++++++++ 1 file changed, 176 insertions(+) create mode 100644 tests/test_client_commands.py diff --git a/tests/test_client_commands.py b/tests/test_client_commands.py new file mode 100644 index 000000000..243382403 --- /dev/null +++ b/tests/test_client_commands.py @@ -0,0 +1,176 @@ +import pytest + +from labgrid.exceptions import NoSupplierFoundError +from labgrid.remote.client import ensure_event_loop, ClientSession, UserError +from labgrid.driver import PyOCDDriver, NetworkPowerDriver, LXAIOBusPIODriver +from labgrid.resource.remote import NetworkUSBDebugger, NetworkLXAIOBusPIO +from labgrid.resource.power import NetworkPowerPort + + +class PseudoArgs: + def __getattr__(self, attr): + if attr == "action": + return "get" + if attr == "bootstrap_args": + return "" + return None + + +class PseudePlace: + def __getattr__(self, attr): + if attr == "name": + return "PseudoPlace" + return None + + +@pytest.fixture(scope="function") +def client(mocker, target): + loop = ensure_event_loop() + client = ClientSession(address="", loop=loop) + client.args = PseudoArgs() + place = PseudePlace() + + def get_acquired_place(): + return place + place_mock = mocker.patch("labgrid.remote.client.ClientSession.get_acquired_place") + place_mock.side_effect = get_acquired_place + + def _get_target(_): + return target + target_mock = mocker.patch("labgrid.remote.client.ClientSession._get_target") + target_mock.side_effect = _get_target + + # disable resource updating + target_resource_update_mock = mocker.patch("labgrid.target.Target.update_resources") + return client + + +@pytest.mark.parametrize( + "command", + [ + "bootstrap", + "digital_io", + "power", + "reset", + "sd_mux", + "usb_mux", + "video", + ], +) +def test_command_without_resources(client, command): + with pytest.raises(UserError, match="target has no compatible resource available"): + c = getattr(client, command) + c() + + +@pytest.mark.parametrize( + "command", + [ + "dfu", + "fastboot", + "flashscript", + "audio", + "write_files", + "write_image", + "ssh", + "scp", + "rsync", + "sshfs", + "telnet", + ], +) +def test_command_without_suplierers(client, command): + with pytest.raises(NoSupplierFoundError): + c = getattr(client, command) + c() + + +def network_usb_ressource(target): + network_args = { + "name": "network_usb_ressource", + "host": "None", + "busnum": 0, + "devnum": 0, + "path": "1-12", + "vendor_id": 0, + "model_id": 0, + } + r = NetworkUSBDebugger(target, **network_args) + r.avail = True + + +def pyocd_driver(target): + network_usb_ressource(target) + PyOCDDriver(target, "pyocd") + + +def network_power_port(target): + network_args = { + "name": "network_power_port", + "host": "None", + "model": "rest", + "index": "0", + } + r = NetworkPowerPort(target, **network_args) + r.avail = True + + +def network_power_driver(target): + network_power_port(target) + NetworkPowerDriver(target, "fake_power") + + +def network_LXA_gpio(target): + network_args = { + "name": "network_lxa_io_bus_name", + "node": "network_lxa_io_bus_node", + "host": "None", + "pin": "0", + "invert": False, + } + r = NetworkLXAIOBusPIO(target, **network_args) + r.avail = True + + +def lxaio_bus_driver(target): + network_LXA_gpio(target) + LXAIOBusPIODriver(target, "lxa_pio_driver") + + +def power_and_gpio_ressource(target): + network_LXA_gpio(target) + network_power_port(target) + + +def power_gpio_and_debug_ressource(target): + network_LXA_gpio(target) + network_power_port(target) + network_usb_ressource(target) + + +@pytest.mark.parametrize( + "command,setup,patch", + [ + ("reset", network_usb_ressource, "labgrid.driver.pyocddriver.PyOCDDriver.reset"), + ("reset", pyocd_driver, "labgrid.driver.pyocddriver.PyOCDDriver.reset"), + ("power", network_power_port, "labgrid.driver.power.rest.power_get"), + ("power", network_power_driver, "labgrid.driver.power.rest.power_get"), + ("digital_io", network_LXA_gpio, "labgrid.driver.lxaiobusdriver.LXAIOBusPIODriver.get"), + ("digital_io", lxaio_bus_driver, "labgrid.driver.lxaiobusdriver.LXAIOBusPIODriver.get"), + ("reset", network_power_port, "labgrid.driver.powerdriver.PowerResetMixin.reset"), + ("reset", network_LXA_gpio, "labgrid.driver.resetdriver.DigitalOutputResetDriver.reset"), + ("reset", power_and_gpio_ressource, "labgrid.driver.resetdriver.DigitalOutputResetDriver.reset"), + ("reset", power_gpio_and_debug_ressource, "labgrid.driver.pyocddriver.PyOCDDriver.reset"), + ("bootstrap", network_usb_ressource, "labgrid.driver.pyocddriver.PyOCDDriver.load"), + ("bootstrap", pyocd_driver, "labgrid.driver.pyocddriver.PyOCDDriver.load"), + ], +) +def test_reset_command_with_remote_resource(target, client, mocker, command, setup, patch): + setup(target) + + driver_mock = mocker.patch(patch) + + c = getattr(client, command) + c() + + driver_mock.assert_called_once() From f2abdc6868aae135558924bfa64e1102c547688f Mon Sep 17 00:00:00 2001 From: Fabian Pflug Date: Mon, 27 Jul 2026 10:22:25 +0200 Subject: [PATCH 10/10] driver: pyocd: fix error on multiple load call When calling load multiple times with load_commands as an array, the file to load was appended to the load_commands array, resulting in the second load command failing. Pointed out in https://github.com/labgrid-project/labgrid/pull/1944#discussion_r3653612080 Signed-off-by: Fabian Pflug --- labgrid/driver/pyocddriver.py | 2 +- tests/test_pyocd.py | 25 +++++++++++++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/labgrid/driver/pyocddriver.py b/labgrid/driver/pyocddriver.py index 39e3118d0..cfeb8f1ef 100644 --- a/labgrid/driver/pyocddriver.py +++ b/labgrid/driver/pyocddriver.py @@ -94,7 +94,7 @@ def load(self, filename=None): if isinstance(self.load_commands, str): commands = self.load_commands.split() else: - commands = self.load_commands + commands = self.load_commands.copy() else: commands = ["-e", "sector"] diff --git a/tests/test_pyocd.py b/tests/test_pyocd.py index 1a12f0a3d..8ed25de5c 100644 --- a/tests/test_pyocd.py +++ b/tests/test_pyocd.py @@ -48,6 +48,31 @@ def test_pyocd_driver_load(target, mocker): assert check_output_mock.call_args.kwargs["print_on_silent_log"] +def test_pyocd_driver_multiple_load_with_load_commands(target, mocker): + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None, load_commands=["--frequency", "100"]) + target.activate(d) + + check_output_mock = mocker.patch("labgrid.util.helper.processwrapper.check_output") + + d.load(__file__) + + check_output_mock.assert_called_once() + cmd = check_output_mock.call_args.kwargs["command"] + assert cmd == ["pyocd", "load", "--no-config", "--frequency", "100", __file__] + assert check_output_mock.call_args.kwargs["print_on_silent_log"] + + check_output_mock.reset_mock() + + d.load(__file__) + + check_output_mock.assert_called_once() + cmd = check_output_mock.call_args.kwargs["command"] + assert cmd == ["pyocd", "load", "--no-config", "--frequency", "100", __file__] + assert check_output_mock.call_args.kwargs["print_on_silent_log"] + + def test_pyocd_load_error_on_missing_file(target, mocker): r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) r.avail = True