Resource: add Segger J-Link support

Add a resource for the Segger J-Link debug probe.
Export the J-Link resource over IP using the J-Link Remote Server.

Signed-off-by: Paul Vittorino <[email protected]>

Author: PaulVittorino

doc/configuration.rst

@@ -746,6 +746,18 @@ Arguments:
 Used by:
   - `OpenOCDDriver`_
 
+JLinkDevice
+~~~~~~~~~~~
+A JLinkDevice resource describes a Segger J-Link Debug Probe.
+
+.. code-block:: yaml
+
+  JLinkDevice:
+    match:
+      ID_SERIAL_SHORT: '000000123456'
+
+- match (dict): key and value for a udev match, see `udev Matching`_
+
 SNMPEthernetPort
 ~~~~~~~~~~~~~~~~
 A SNMPEthernetPort resource describes a port on an Ethernet switch, which is

labgrid/remote/exporter.py

@@ -14,12 +14,14 @@
 from pathlib import Path
 from typing import Dict, Type
 from socket import gethostname, getfqdn
+from pexpect import TIMEOUT
 import attr
 from autobahn.asyncio.wamp import ApplicationRunner, ApplicationSession
 
 from .config import ResourceConfig
 from .common import ResourceEntry, enable_tcp_nodelay
 from ..util import get_free_port, labgrid_version
+from ..util import Timeout
 
 
 __version__ = labgrid_version()
@@ -500,6 +502,80 @@ def __attrs_post_init__(self):
         super().__attrs_post_init__()
         self.data['cls'] = f"Remote{self.cls}".replace("Network", "")
 
+class USBJLinkExport(USBGenericExport):
+    """Export J-Link device using the J-Link Remote Server"""
+
+    def __attrs_post_init__(self):
+        super().__attrs_post_init__()
+        self.child = None
+        self.port = None
+        self.tool = '/opt/SEGGER/JLink/JLinkRemoteServer'
+
+    def _get_params(self):
+        """Helper function to return parameters"""
+        return {
+            'host': self.host,
+            'port': self.port,
+            'busnum': self.local.busnum,
+            'devnum': self.local.devnum,
+            'path': self.local.path,
+            'vendor_id': self.local.vendor_id,
+            'model_id': self.local.model_id,
+        }
+
+    def __del__(self):
+        if self.child is not None:
+            self.stop()
+
+    def _start(self, start_params):
+        """Start ``JLinkRemoteServer`` subprocess"""
+        assert self.local.avail
+        assert self.child is None
+        self.port = get_free_port()
+
+        cmd = [
+            self.tool,
+            "-Port",
+            f"{self.port}",
+            "-select",
+            f"USB={self.local.serial}",
+        ]
+        self.logger.info("Starting JLinkRemoteServer with: %s", " ".join(cmd))
+        self.child = subprocess.Popen(cmd, stdout=subprocess.PIPE, universal_newlines=True)
+
+        # Wait for the server to be ready for incoming connections
+        # Waiting to open the socket with Python does not work
+        timeout = Timeout(10.0)
+        while not timeout.expired:
+            line = self.child.stdout.readline().rstrip()
+            self.logger.debug(line)
+            if "Waiting for client connections..." in line:
+                break
+
+        if timeout.expired:
+            raise TIMEOUT(
+                f"Timeout of {timeout.timeout} seconds exceeded during waiting for J-Link Remote Server startup"
+            )
+        self.logger.info("started JLinkRemoteServer for %s on port %d", self.local.serial, self.port)
+
+    def _stop(self, start_params):
+        """Stop ``JLinkRemoteServer`` subprocess"""
+        assert self.child
+        child = self.child
+        self.child = None
+        port = self.port
+        self.port = None
+        child.terminate()
+        try:
+            child.wait(2.0)  # JLinkRemoteServer takes about a second to react
+        except subprocess.TimeoutExpired:
+            self.logger.warning("JLinkRemoteServer for %s still running after SIGTERM", self.local.serial)
+            log_subprocess_kernel_stack(self.logger, child)
+            child.kill()
+            child.wait(1.0)
+        self.logger.info("stopped JLinkRemoteServer for %s on port %d", self.local.serial, port)
+
+
 exports["AndroidFastboot"] = USBGenericExport
 exports["AndroidUSBFastboot"] = USBGenericRemoteExport
 exports["DFUDevice"] = USBGenericExport
@@ -512,6 +588,7 @@ def __attrs_post_init__(self):
 exports["USBSDMuxDevice"] = USBSDMuxExport
 exports["USBSDWireDevice"] = USBSDWireExport
 exports["USBDebugger"] = USBGenericExport
+exports["JLinkDevice"] = USBJLinkExport
 
 exports["USBMassStorage"] = USBGenericExport
 exports["USBVideo"] = USBGenericExport

labgrid/resource/remote.py

@@ -305,6 +305,18 @@ def __attrs_post_init__(self):
         super().__attrs_post_init__()
 
 
+@target_factory.reg_resource
+@attr.s(eq=False)
+class NetworkJLinkDevice(RemoteUSBResource):
+    """The NetworkJLinkDevice describes a remotely accessible USB Segger J-Link device"""
+
+    port = attr.ib(validator=attr.validators.optional(attr.validators.instance_of(int)))
+
+    def __attrs_post_init__(self):
+        self.timeout = 10.0
+        super().__attrs_post_init__()
+
+
 @target_factory.reg_resource
 @attr.s(eq=False)
 class NetworkDeditecRelais8(RemoteUSBResource):

labgrid/resource/suggest.py

@@ -23,6 +23,7 @@
     HIDRelay,
     USBDebugger,
     USBPowerPort,
+    JLinkDevice,
 )
 from ..util import dump
 
@@ -56,6 +57,7 @@ def __init__(self, args):
         self.resources.append(HIDRelay(**args))
         self.resources.append(USBDebugger(**args))
         self.resources.append(USBPowerPort(**args, index=0))
+        self.resources.append(JLinkDevice(**args))
 
     def suggest_callback(self, resource, meta, suggestions):
         cls = type(resource).__name__

labgrid/resource/udev.py

@@ -696,3 +696,19 @@ def filter_match(self, device):
             return False
 
         return super().filter_match(device)
+
+@target_factory.reg_resource
+@attr.s(eq=False)
+class JLinkDevice(USBResource):
+    """The JLinkDevice describes an attached Segger J-Link device,
+    it is identified via USB using udev
+    """
+
+    def __attrs_post_init__(self):
+        self.match["ID_VENDOR_ID"] = "1366"
+        self.match["ID_MODEL_ID"] = "0101"
+        super().__attrs_post_init__()
+
+    def update(self):
+        super().update()
+        self.serial = self.device.properties.get('ID_SERIAL_SHORT')

tests/test_jlink.py

@@ -0,0 +1,83 @@
+import contextlib
+import io
+import subprocess
+from unittest.mock import MagicMock
+
+from labgrid.remote.exporter import USBJLinkExport
+from labgrid.resource.udev import JLinkDevice
+
+FAKE_SERIAL = 123456789
+MATCH = {"ID_SERIAL_SHORT": f"000{FAKE_SERIAL}"}
+
+
+class Popen_mock(object):
+    """Mock of Popen object to mimmic JLinkRemoteServer output"""
+
+    def __init__(self, args, **kwargs):
+        self.wait_called = False
+        assert "JLinkRemoteServer" in args[0]
+        assert args[1] == "-Port"
+        # Since args[2] is dynamic do not check it
+        assert args[3] == "-select"
+        assert args[4] == f"USB={FAKE_SERIAL}"
+
+    stdout = io.StringIO(
+        "SEGGER J-Link Remote Server V7.84a\n"
+        "Compiled Dec 22 2022 16:13:52\n"
+        "\n"
+        "'q' to quit '?' for help\n"
+        "\n"
+        f"Connected to J-Link with S/N {FAKE_SERIAL}\n"
+        "\n"
+        "Waiting for client connections...\n"
+    )
+
+    def kill(self):
+        pass
+
+    def poll(self):
+        return 0
+
+    def terminate(self):
+        pass
+
+    def wait(self, timeout):
+        # Only timeout on the first call to wait to exercise the error handling code.
+        if not self.wait_called:
+            self.wait_called = True
+            raise subprocess.TimeoutExpired("JLinkRemoteServer", timeout)
+
+
+@contextlib.contextmanager
+def subprocess_mock():
+    """Mock of subprocess. Only implements Popen"""
+    import subprocess
+
+    original = subprocess.Popen
+
+    def run(args, **kwargs):
+        assert "JLinkRemoteServer" in args[0]
+        assert args[1] == "-Port"
+        # Since args[2] is dynamic do not check it
+        assert args[3] == "-select"
+        assert args[4] == f"USB={FAKE_SERIAL}"
+        return Popen_mock()
+
+    subprocess.Popen = Popen_mock
+    yield MagicMock()
+    subprocess.Popen = original
+
+
+def test_jlink_resource(target):
+    r = JLinkDevice(target, name=None, match=MATCH)
+
+def test_jlink_export_start(target):
+    config = {'avail': True, 'cls': "JLinkDevice", 'params': {'match': MATCH}, }
+    e = USBJLinkExport(config)
+    e.local.avail = True
+    e.local.serial = FAKE_SERIAL
+
+    with subprocess_mock():
+        e.start()
+        # Exercise the __del__ method which also exercises stop()
+        del e