QuartusPGM: revisit and cleanup implementation

sessl3r · sessl3r · commit 8ac97a9d170d · 2025-11-27T08:40:41.000+01:00
- udev.py/remote.py: cleanup attributes
- exporter.py:
  - move classes to be able to use generic usb class
  - use USBGenericExport and remove duplicated code
  - use tempfile for jtagd config file
  - remove unnecessary env copy
  - use labgrids udev device matching infrastructure
  - cleanup logging and printouts
  - cleanup kill of jtagd
  - add jtag_conf into resource info as driver/client information
- client.py:
  - cleanup exception handling
- quartuspgmdriver.py:
  - remove unnecessary env copy
  - use jtag_conf resource info for temporary jtag.conf
  - cleanup error handling

also some pylint cleanups
diff --git a/labgrid/driver/quartuspgmdriver.py b/labgrid/driver/quartuspgmdriver.py
@@ -1,19 +1,15 @@
 # pylint: disable=no-member
 import subprocess
 import os
-import re
 import attr
 import importlib
+import logging
 import tempfile
 from pathlib import Path
 
 from ..factory import target_factory
 from ..step import step
 from .common import Driver
-from .exception import ExecutionError
-from ..util.helper import processwrapper
-from ..util.managedfile import ManagedFile
-import logging
 
 JTAG_CONF_INTEL = """
 Remote1 {
@@ -56,41 +52,37 @@ def erase(self, filename=None, devnum=1):
 
     @Driver.check_active
     @step(args=['filename', 'operation', 'devnum'])
-    def operate(self, filename=None, operation="P", devnum=1):
+    def operate(self, filename=None, operation="P", devnum=1) -> tuple[str, str]:
         if filename is None and self.image is not None:
             filename = self.target.env.config.get_image_path(self.image)
 
         log = logging.getLogger("QPGM_Driver")
 
-        try:
-            lib_path = importlib.machinery.PathFinder.find_spec('libfilsel').origin
-        except Exception as e:
-            return False, "could not find libfilsel!", str(e)
-
-        my_env = os.environ.copy()
-        my_env["LD_PRELOAD"] = os.pathsep.join(filter(None, [lib_path, os.environ.get('LD_PRELOAD')]))
-        my_env["FILSEL_ORG_PATH"] = str((Path(os.path.expanduser('~')) / ".jtag.conf").resolve())
+        lib_path = importlib.machinery.PathFinder.find_spec('libfilsel').origin
 
-        cable = f"--cable=\"{self.interface.device_name} on " + \
-                f"{self.interface.host}:{self.interface.jtagd_port} " + \
-                f"{self.interface.device_port}\""
+        ld_preload = [lib_path, os.getenv('LD_PRELOAD', "")]
+        os.environ["LD_PRELOAD"] = os.pathsep.join(ld_preload)
+        os.environ["FILSEL_ORG_PATH"] = str((Path(os.path.expanduser('~')) / ".jtag.conf").resolve())
 
-        operation = f"--operation=\"{operation};{filename}@{str(devnum)}\""
-        cmd = f"{self.tool} {cable} --mode=JTAG {operation}"
+        cable = f"'{self.interface.device_name} on {self.interface.host}:{self.interface.jtagd_port} {self.interface.device_port}'"
+        operation = f"'{operation};{filename}@{str(devnum)}'"
+        cmd = f"{self.tool} -c {cable} -m JTAG -o {operation}"
 
         with tempfile.NamedTemporaryFile() as conf_temp:
 
-            cfg = JTAG_CONF_INTEL.replace("HOST", self.interface.host + ":" + str(self.interface.jtagd_port))\
-                                 .replace("PASSWORD", self.interface.jtagd_password)
+            cfg = self.interface.extra['jtag_conf']
             conf_temp.write(cfg.encode("utf-8"))
             conf_temp.flush()
-            log.info("Flashing with command: " + str(cmd))
-            my_env["FILSEL_DEST_PATH"] = conf_temp.name
+            log.info("Flashing with command: %s", cmd)
+            os.environ["FILSEL_DEST_PATH"] = conf_temp.name
 
-            stdout, stderr = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=my_env).communicate()
+            process = subprocess.Popen(cmd, shell=True,
+                    stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            stdout, stderr = process.communicate()
 
         if "Quartus Prime Programmer was successful." in stdout.decode("utf-8"):
-            return True, stdout.decode("utf-8"), stderr.decode("utf-8")
+            return stdout.decode("utf-8"), stderr.decode("utf-8")
         else:
-            return False, stdout.decode("utf-8"), stderr.decode("utf-8")
+            raise subprocess.CalledProcessError(process.returncode,
+                    cmd, output=stdout, stderr=stderr)
 
diff --git a/labgrid/remote/client.py b/labgrid/remote/client.py
@@ -66,17 +66,6 @@ class InteractiveCommandError(Error):
     pass
 
 
-class QuartusPgmDriverError(Exception):
-    """Exception raised for errors in the Quartus PGM Driver.
-    """
-
-    def __init__(self, stdout, stderr):
-        self.stdout = stdout
-        self.stderr = stderr
-        self.message = "QuartusPGM failed with stdout: " + stdout + " and stderr: " + stderr
-        super().__init__(self.message)
-
-
 class ClientSession(ApplicationSession):
     """The ClientSession encapsulates all the actions a Client can Invoke on
     the coordinator."""
@@ -1354,12 +1343,10 @@ def _get_quartus(self, name):
     def intel_program_bitstream(self):
         drv = self._get_quartus(self.args.name)
         processwrapper.enable_print()
-        ret, stdout, stderr = drv.flash(self.args.bitstream)
-        if not ret:
-            print("Flashing failed with:")
-            print(stdout)
-            print(stderr)
-            raise QuartusPgmDriverError(stdout, stderr)
+        try:
+            drv.flash(self.args.bitstream)
+        except subprocess.CalledProcessError as e:
+            raise UserError(f"programming failed with {e.returncode}\n{(e.stdout + e.stderr).decode('utf-8')}")
         processwrapper.disable_print()
 
     def write_image(self):
diff --git a/labgrid/remote/exporter.py b/labgrid/remote/exporter.py
@@ -3,12 +3,14 @@
 
 import argparse
 import asyncio
+import importlib
 import json
 import logging
 import sys
 import os
 import os.path
 import signal
+import tempfile
 import time
 import traceback
 import shutil
@@ -19,7 +21,6 @@
 from socket import gethostname, getfqdn
 import attr
 from autobahn.asyncio.wamp import ApplicationRunner, ApplicationSession
-import importlib
 
 from .config import ResourceConfig
 from .common import ResourceEntry, enable_tcp_nodelay, monkey_patch_max_msg_payload_size_ws_option
@@ -301,75 +302,140 @@ def _stop(self, start_params):
 
 
 @attr.s(eq=False)
-class QuartusServerExport(ResourceExport):
+class NetworkInterfaceExport(ResourceExport):
+    """ResourceExport for a network interface"""
+
+    def __attrs_post_init__(self):
+        super().__attrs_post_init__()
+        if self.cls == "NetworkInterface":
+            from ..resource.base import NetworkInterface
+
+            self.local = NetworkInterface(target=None, name=None, **self.local_params)
+        elif self.cls == "USBNetworkInterface":
+            from ..resource.udev import USBNetworkInterface
+
+            self.local = USBNetworkInterface(target=None, name=None, **self.local_params)
+        self.data["cls"] = "RemoteNetworkInterface"
+
+    def _get_params(self):
+        """Helper function to return parameters"""
+        params = {
+            "host": self.host,
+            "ifname": self.local.ifname,
+        }
+        if self.cls == "USBNetworkInterface":
+            params["extra"] = {
+                "state": self.local.if_state,
+            }
+
+        return params
+
+
+exports["USBNetworkInterface"] = NetworkInterfaceExport
+exports["NetworkInterface"] = NetworkInterfaceExport
+
+
+@attr.s(eq=False)
+class USBGenericExport(ResourceExport):
+    """ResourceExport for USB devices accessed directly from userspace"""
+
+    def __attrs_post_init__(self):
+        super().__attrs_post_init__()
+        local_cls_name = self.cls
+        self.data["cls"] = f"Network{self.cls}"
+        from ..resource import udev
+
+        local_cls = getattr(udev, local_cls_name)
+        self.local = local_cls(target=None, name=None, **self.local_params)
+
+    def _get_params(self):
+        """Helper function to return parameters"""
+        return {
+            "host": self.host,
+            "busnum": self.local.busnum,
+            "devnum": self.local.devnum,
+            "path": self.local.path,
+            "vendor_id": self.local.vendor_id,
+            "model_id": self.local.model_id,
+        }
+
+
+@attr.s(eq=False)
+class QuartusServerExport(USBGenericExport):
     """ ResourceExport for a QuartusUSBJTAG via ``jtagd``"""
 
     def __attrs_post_init__(self):
         super().__attrs_post_init__()
-        self.data['cls'] = "NetworkQuartusUSBJTAG"
-        from ..resource.udev import QuartusUSBJTAG
-        self.local = QuartusUSBJTAG(target=None, name=None, **self.local_params)
         self.child = None
+        self.jtagd_port = self.local.jtagd_port
+        self.cfg_tempfile = None
 
     def __del__(self):
         if self.child is not None:
-            self.release()
-
-    def _get_custom_config_file_name(self):
-        serialNumber = self.local.device_serial
-        return os.path.join(self.local.jtagd_file_locations, f"jtagd_cfg_{serialNumber}_{self.local.jtagd_port}.conf")
+            self.stop()
 
-    def acquire(self, *args, **kwargs):
+    def _start(self, start_params):
         """Start ``jtagd`` subprocess"""
         assert self.local.avail
+        assert self.child is None
 
-        cfgFileName = self._get_custom_config_file_name()
+        if not self.jtagd_port:
+            self.jtagd_port = get_free_port()
 
-        with open(cfgFileName, 'w+') as file:
-            file.write(f"Password = \"{self.local.jtagd_password}\";")
+        self.cfg_tempfile = tempfile.NamedTemporaryFile()
+        self.cfg_tempfile.write(
+                f"Password = \"{self.local.jtagd_password}\";".encode('utf-8'))
+        self.cfg_tempfile.flush()
 
-        #find the right path to the library
+        # find the right path to the library and modify the env
         lib_path = importlib.machinery.PathFinder.find_spec('libhwsf').origin
+        ld_preload = [lib_path, os.getenv('LD_PRELOAD', "")]
+        os.environ["LD_PRELOAD"] = os.pathsep.join(ld_preload)
+        os.environ["HWSF_DEV"] = "path:" + self.local.device.sys_name
 
-        #get the usb path from the device serial number
-        serialNumber = self.local.device_serial
-        dev_grep = "grep {SERIAL} /sys/bus/usb/devices/*/serial | cut -d '/' -f6".format(SERIAL=serialNumber)
-        dev_path, _ = subprocess.Popen(dev_grep, shell=True, stdout=subprocess.PIPE).communicate()
-
-        my_env = os.environ.copy()
-        my_env["LD_PRELOAD"] = os.pathsep.join(filter(None, [lib_path, os.environ.get('LD_PRELOAD')]))
-        my_env["HWSF_DEV"] = "path:" + dev_path.decode("utf-8").replace("\n","")
+        cmd = f"{self.local.jtagd_cmd} --foreground --port {self.jtagd_port} --config {self.cfg_tempfile.name}"
 
-        cmd = " ".join([f"{self.local.jtagd_cmd}",
-                        f"--foreground",
-                        f"--port {self.local.jtagd_port}",
-                        f"--config {self._get_custom_config_file_name()}"])
+        self.logger.info("starting jtagd for %s on port %s with command LD_PRELOAD+=%s HWSF_DEV=%s %s",
+                         self.local.device.sys_name, self.jtagd_port, lib_path,
+                         os.environ['HWSF_DEV'], cmd)
+        self.child = subprocess.Popen(cmd, shell=True, preexec_fn=os.setsid)
 
-        self.logger.info("Starting jtagd with command: " + cmd)
-        self.logger.info("Starting jtagd with LD_PRELOAD=" + lib_path + " HWSF_DEV=" + my_env["HWSF_DEV"])
-
-        self.child = subprocess.Popen(cmd, shell=True, preexec_fn=os.setsid, env=my_env)
-
-    def release(self,  *args, **kwargs):
+    def _stop(self,  start_params):
         """Stop ``jtagd`` subprocess"""
         assert self.child
-
-        os.remove(self._get_custom_config_file_name())
-        os.killpg(os.getpgid(self.child.pid), signal.SIGTERM)
+        child = self.child
         self.child = None
-        self.logger.info(f"stopped jtagd at {self.local.jtagd_port}")
+        self.cfg_tempfile = None
+        child.terminate()
+        try:
+            child.wait(2.0)
+        except subprocess.TimeoutExpired:
+            self.logger.warning("jtagd for %s still running after SIGTERM",
+                                self.local.device.sys_name)
+            child.kill()
+            child.wait(1.0)
+
+        self.logger.info("stopped jtagd for %s on port %s",
+                         self.local.device.sys_name, self.jtagd_port)
 
     def _get_params(self):
         """Helper function to return parameters"""
         return {
-            'host': self.host,
-            'jtagd_password': self.local.jtagd_password,
-            'jtagd_port': self.local.jtagd_port,
-            'jtagd_cmd': self.local.jtagd_cmd,
-            'device_name': self.local.device_name,
-            'device_port': self.local.device_port,
+            **super()._get_params(),
+            "jtagd_cmd": self.local.jtagd_cmd,
+            "jtagd_port": self.jtagd_port,
+            "jtagd_password": self.local.jtagd_password,
+            "device_name": self.local.device_name,
+            "device_port": self.local.device_port,
+            "extra": {
+                "jtag_conf": f"""Remote1 {{
+        Host = \"{self.host}:{self.jtagd_port}\";
+        Password = \"{self.local.jtagd_password}\";
+}}"""
+            }
         }
 
+
 exports["QuartusUSBJTAG"] = QuartusServerExport
 
 
@@ -445,65 +511,6 @@ def _stop(self, start_params):
 exports["XilinxUSBJTAG"] = VivadoHWServerExport
 
 
-@attr.s(eq=False)
-class NetworkInterfaceExport(ResourceExport):
-    """ResourceExport for a network interface"""
-
-    def __attrs_post_init__(self):
-        super().__attrs_post_init__()
-        if self.cls == "NetworkInterface":
-            from ..resource.base import NetworkInterface
-
-            self.local = NetworkInterface(target=None, name=None, **self.local_params)
-        elif self.cls == "USBNetworkInterface":
-            from ..resource.udev import USBNetworkInterface
-
-            self.local = USBNetworkInterface(target=None, name=None, **self.local_params)
-        self.data["cls"] = "RemoteNetworkInterface"
-
-    def _get_params(self):
-        """Helper function to return parameters"""
-        params = {
-            "host": self.host,
-            "ifname": self.local.ifname,
-        }
-        if self.cls == "USBNetworkInterface":
-            params["extra"] = {
-                "state": self.local.if_state,
-            }
-
-        return params
-
-
-exports["USBNetworkInterface"] = NetworkInterfaceExport
-exports["NetworkInterface"] = NetworkInterfaceExport
-
-
-@attr.s(eq=False)
-class USBGenericExport(ResourceExport):
-    """ResourceExport for USB devices accessed directly from userspace"""
-
-    def __attrs_post_init__(self):
-        super().__attrs_post_init__()
-        local_cls_name = self.cls
-        self.data["cls"] = f"Network{self.cls}"
-        from ..resource import udev
-
-        local_cls = getattr(udev, local_cls_name)
-        self.local = local_cls(target=None, name=None, **self.local_params)
-
-    def _get_params(self):
-        """Helper function to return parameters"""
-        return {
-            "host": self.host,
-            "busnum": self.local.busnum,
-            "devnum": self.local.devnum,
-            "path": self.local.path,
-            "vendor_id": self.local.vendor_id,
-            "model_id": self.local.model_id,
-        }
-
-
 @attr.s(eq=False)
 class USBSigrokExport(USBGenericExport):
     """ResourceExport for USB devices accessed directly from userspace"""
diff --git a/labgrid/resource/remote.py b/labgrid/resource/remote.py
@@ -179,10 +179,9 @@ def __attrs_post_init__(self):
 
 @target_factory.reg_resource
 @attr.s(eq=False)
-class NetworkQuartusUSBJTAG(NetworkResource, ManagedResource):
-    host = attr.ib(default="")
+class NetworkQuartusUSBJTAG(RemoteUSBResource):
     jtagd_password = attr.ib(default="password1234")
-    jtagd_port = attr.ib(default=3109)
+    jtagd_port = attr.ib(factory=int)
     jtagd_cmd = attr.ib(default="jtagd")
     device_name = attr.ib(default="Arrow-USB-Blaster")
     device_port = attr.ib(default="")
diff --git a/labgrid/resource/udev.py b/labgrid/resource/udev.py
