driver/quartushpsdriver: add support for Quartus HPS

doc/configuration.rst

@@ -302,6 +302,7 @@ An AlteraUSBBlaster resource describes an Altera USB blaster.
 
 Used by:
   - `OpenOCDDriver`_
+  - `QuartusHPSDriver`_
 
 SNMPEthernetPort
 ~~~~~~~~~~~~~~~~
@@ -690,6 +691,23 @@ Arguments:
   - search (str): include search path for scripts
   - image (str): filename of image to bootstrap onto the device
 
+QuartusHPSDriver
+~~~~~~~~~~~~~~~~
+A QuartusHPSDriver controls the "Quartus Prime Programmer and Tools" to flash
+a target's QSPI.
+
+Binds to:
+  - `AlteraUSBBlaster`_
+
+Implements:
+  - None
+
+Arguments:
+  - image (str): filename of image to flash QSPI
+
+The driver can be used in test cases by calling the `flash` function. An
+example strategy is included in Labgrid.
+
 ManualPowerDriver
 ~~~~~~~~~~~~~~~~~
 A ManualPowerDriver requires the user to control the target power states. This

examples/strategy/quartusstrategy.py

@@ -0,0 +1,67 @@
+import enum
+
+import attr
+
+from labgrid.driver import QuartusHPSDriver, SerialDriver
+from labgrid.factory import target_factory
+from labgrid.protocol import PowerProtocol
+from labgrid.step import step
+from labgrid.strategy.common import Strategy
+
+
+@attr.s(cmp=False)
+class StrategyError(Exception):
+    msg = attr.ib(validator=attr.validators.instance_of(str))
+
+
+class Status(enum.Enum):
+    unknown = 0
+    flashed_xload = 1
+    flashed = 2
+
+
+@target_factory.reg_driver
+@attr.s(cmp=False)
+class QuartusHPSStrategy(Strategy):
+    """QuartusHPSStrategy - Strategy to flash QSPI via 'Quartus Prime Programmer and Tools'"""
+    bindings = {
+        "power": PowerProtocol,
+        "quartushps": QuartusHPSDriver,
+        "serial": SerialDriver,
+    }
+
+    image = attr.ib(validator=attr.validators.instance_of(str))
+    image_xload = attr.ib(validator=attr.validators.instance_of(str))
+    status = attr.ib(default=Status.unknown)
+
+    def __attrs_post_init__(self):
+        super().__attrs_post_init__()
+
+    @step(args=['status'])
+    def transition(self, status, *, step):
+        if not isinstance(status, Status):
+            status = Status[status]
+        if status == Status.unknown:
+            raise StrategyError("can not transition to {}".format(status))
+        elif status == self.status:
+            step.skip("nothing to do")
+            return  # nothing to do
+        elif status == Status.flashed_xload:
+            self.target.activate(self.power)
+            self.power.cycle()
+            self.target.activate(self.quartushps)
+            # flash bootloader xload image to 0x0
+            self.quartushps.flash(self.image_xload, 0x0)
+        elif status == Status.flashed:
+            self.transition(Status.flashed_xload)
+            # flash bootloader image to 0x40000
+            self.quartushps.flash(self.image, 0x40000)
+            self.power.cycle()
+            # activate serial in order to make 'labgrid-client -s $STATE con' work
+            self.target.activate(self.serial)
+        else:
+            raise StrategyError(
+                "no transition found from {} to {}".
+                format(self.status, status)
+            )
+        self.status = status

labgrid/driver/__init__.py

@@ -7,6 +7,7 @@
 from .exception import CleanUpError, ExecutionError
 from .fastbootdriver import AndroidFastbootDriver
 from .openocddriver import OpenOCDDriver
+from .quartushpsdriver import QuartusHPSDriver
 from .onewiredriver import OneWirePIODriver
 from .powerdriver import ManualPowerDriver, ExternalPowerDriver, DigitalOutputPowerDriver, YKUSHPowerDriver
 from .usbloader import MXSUSBDriver, IMXUSBDriver

labgrid/driver/quartushpsdriver.py

@@ -0,0 +1,82 @@
+# pylint: disable=no-member
+import attr
+import subprocess
+import os.path
+import re
+
+from ..factory import target_factory
+from ..resource.remote import NetworkAlteraUSBBlaster
+from ..resource.udev import AlteraUSBBlaster
+from ..step import step
+from .common import Driver, check_file
+from .exception import ExecutionError
+
+
+@target_factory.reg_driver
+@attr.s(cmp=False)
+class QuartusHPSDriver(Driver):
+    bindings = {
+        "interface": {AlteraUSBBlaster, NetworkAlteraUSBBlaster},
+    }
+
+    image = attr.ib(default=None, validator=attr.validators.optional(attr.validators.instance_of(str)))
+    cable_number = 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('quartus_hps') or 'quartus_hps'
+        else:
+            self.tool = 'quartus_hps'
+
+    def on_deactivate(self):
+        # forget cable number as it might change
+        self.cable_number = None
+
+    def _get_cable_number(self):
+        """Returns the JTAG cable numer for the USB path of the device"""
+        # FIXME make sure we always have an environment or config
+        if self.target.env:
+            jtagconfig_tool = self.target.env.config.get_tool('jtagconfig') or 'jtagconfig'
+        else:
+            jtagconfig_tool = 'jtagconfig'
+
+        cmd = self.interface.command_prefix + [jtagconfig_tool]
+        jtagconfig_process = subprocess.Popen(
+            cmd,
+            stdout=subprocess.PIPE
+        )
+        stdout, _ = jtagconfig_process.communicate()
+
+        regex = re.compile(r".*(\d+)\) .* \[(.*)\]")
+        for line in stdout.decode("utf-8").split("\n"):
+            jtag_mapping = regex.match(line)
+            if jtag_mapping:
+                cable_number, usb_path = jtag_mapping.groups()
+                if usb_path == self.interface.path:
+                    return int(cable_number)
+
+        raise ExecutionError("Could not get cable number for USB path {}"
+                             .format(self.interface.path))
+
+    @Driver.check_active
+    @step(args=['filename', 'address'])
+    def flash(self, filename=None, address=0x0):
+        if filename is None and self.image is not None:
+            filename = self.target.env.config.get_image_path(self.image)
+        filename = os.path.abspath(os.path.expanduser(filename))
+        check_file(filename, command_prefix=self.interface.command_prefix)
+
+        assert(isinstance(address, int))
+
+        if self.cable_number is None:
+            self.cable_number = self._get_cable_number()
+
+        cmd = self.interface.command_prefix + [self.tool]
+        cmd += [
+            "--cable={}".format(self.cable_number),
+            "--addr=0x{:X}".format(address),
+            "--operation=P {}".format(filename),
+        ]
+        subprocess.check_call(cmd)