cal: Add automated port switch

Current implementation needed manual interaction to calibrate each
antenna. More sophisticated setups are able to switch between channels
and antennas programmatically.

This commit introduces a base class that handle the switch behaviour. The
previous implementation moved to a ManualSwitch class which is the
default switch. Without any options the previous flow remains unchanged.

A new class is able to handle NI switch models. The switch port can
be given via options parameter (comA is default). The channels are connected
in ascending order. The user has to ensure that the cable setup matches
the order given for channels and antennas.

Co-authored-by: Martin Braun <martin.braun@ettus.com>

4 files changed, 150 insertions, 20 deletions

diff --git a/host/python/uhd/usrp/cal/__init__.py b/host/python/uhd/usrp/cal/__init__.py
index 53de91114..1e4268a69 100644
--- a/host/python/uhd/usrp/cal/__init__.py
+++ b/host/python/uhd/usrp/cal/__init__.py
@@ -16,4 +16,5 @@ from .libtypes import *
 # pylint: enable=wildcard-import
 
 from .meas_device import get_meas_device
+from .switch import get_switch
 from .usrp_calibrator import get_usrp_calibrator
diff --git a/host/python/uhd/usrp/cal/meas_device.py b/host/python/uhd/usrp/cal/meas_device.py
index fd4455da0..15c82e55f 100644
--- a/host/python/uhd/usrp/cal/meas_device.py
+++ b/host/python/uhd/usrp/cal/meas_device.py
@@ -47,15 +47,6 @@ class PowerMeterBase:
         """
         raise NotImplementedError()
 
-    # pylint: disable=no-self-use
-    def update_port(self, chan, antenna):
-        """
-        Tell the device we're measuring chan + antenna next
-        """
-        input("[TX] Connect your power meter to device channel {}, "
-              "antenna {}. Then, hit Enter.".format(chan, antenna))
-    # pylint: enable=no-self-use
-
 class SignalGeneratorBase:
     """
     Base class for measuring input power (Rx) of the USRP. That means the
@@ -116,16 +107,6 @@ class SignalGeneratorBase:
         """
         raise NotImplementedError()
 
-    # pylint: disable=no-self-use
-    def update_port(self, chan, antenna):
-        """
-        Tell the device we're measuring chan + antenna next
-        """
-        input("[RX] Connect your signal generator to device channel {}, "
-              "antenna {}. Then, hit Enter.".format(chan, antenna))
-    # pylint: enable=no-self-use
-
-
 ###############################################################################
 # Manual Measurement: For masochists, or for small sample sets
 ###############################################################################
diff --git a/host/python/uhd/usrp/cal/switch.py b/host/python/uhd/usrp/cal/switch.py
new file mode 100644
index 000000000..1b6157fe6
--- /dev/null
+++ b/host/python/uhd/usrp/cal/switch.py
@@ -0,0 +1,140 @@
+#
+# Copyright 2020 Ettus Research, a National Instruments Brand
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+"""
+Switch Device Classes for UHD Power Calibration
+"""
+
+import sys
+import inspect
+import importlib
+from .ni_rf_instr import get_modinst_device
+
+# pylint: disable=too-few-public-methods
+class SwitchBase:
+    """
+    Base class to connect ports of measurement equipment and USRP.
+    """
+    key = ""
+
+    def connect(self, chan, antenna):
+        """
+        Connect a port of the USRP (DUT) denoted by chan and antenna to the
+        measurement device.
+        :param chan: channel to connect
+        :param antenna: antenna to connect
+        """
+        raise NotImplementedError()
+
+
+class ManualSwitch(SwitchBase):
+    """
+    ManualSwitch is the fallback implementation if no other switch can be
+    found. It asks the user to change cable setup and halts the calibration
+    until the user confirms the configuration. If `mode=auto` is given in
+    options connect call assumes there is no need to pause for connecting
+    measurement device with DUT (e.g. only one path is measured).
+    """
+    def __init__(self, direction, options = None):
+        self.direction = direction
+        self.mode = options.get('mode', '')
+
+    def connect(self, chan, antenna):
+        """
+        Connect a port of the USRP (DUT) denoted by chan and antenna to the
+        measurement device. In this manual mode the user is responsible to
+        ensure correct wiring. The script waits until the user confirms the
+        setup.
+        :param chan: channel to connect
+        :param antenna: antenna to connect
+        """
+        if self.mode == 'auto':
+            return # no need to wait for manual connection
+        input("[{}] Connect your signal generator to device channel {}, "
+              "antenna {}. Then, hit Enter.".format(self.direction, chan, antenna))
+
+class NISwitch(SwitchBase):
+    """
+    Use NI switch devices to automatically connect measurement devices with
+    DUT.
+    """
+    key = "niswitch"
+
+    def __init__(self, options):
+        # connections stores the connected ports for each chan/antenna
+        # combination. During connect call connection is checked for an
+        # appropriate key. If there is no such key the next channel of the
+        # current port switch is used.
+        # To get a working setup
+        self.connections = {}
+
+        device = get_modinst_device("NI-SWITCH", options.get("name", ""))
+        # pylint: disable=import-outside-toplevel
+        # disable import warning. We do not want to make niswitch a mandatory
+        # package for users who do not use this class
+        import niswitch
+        self.session = niswitch.Session(device.device_name)
+        self.port = options.get("port", "comA")
+
+    def connect(self, chan, antenna):
+        """
+        Connect a port of the USRP (DUT) denoted by chan and antenna to the
+        measurement device.
+        The NI switch connects the channel of the switch (named "chXY") to the
+        port of the switch ("comX") which was given during initialization.
+        The connection is made consecutively, meaning the first requested
+        configuration of channel and antenna is connected to chX1, the second
+        to chX2 and so on. X is the identifier of the port given in
+        initialization. The user has to make sure that the SMA port of the DUT
+        are cabled in the right order. The calibration loops over all
+        (configured) channels and for each channel over the (configured)
+        antennas. The order of channels and antennas can be changed via script
+        parameter. The loop order is fixed.
+        :param chan: channel to connect
+        :param antenna: antenna to connect
+        """
+        key = (chan, antenna)
+        if key not in self.connections:
+            # connection not known yet, generate next connection pair
+            # first item is port used by this instance
+            # second item is next channel, channels are named chXY where
+            # X is the letter of the current port (derived from port name) and
+            # Y is a number starting at 1
+            switch_channel = "ch%d%s" % (len(self.connections) + 1,
+                                         self.port[-1:])
+            self.connections[key] = (self.port, switch_channel)
+
+        connection = self.connections[key]
+        print("Connecting %s-%s at switch to measure %d-%s of DUT" %
+              (connection[0], connection[1], chan, antenna))
+        self.session.disconnect_all()
+        self.session.connect(connection[0], connection[1])
+
+###############################################################################
+# The dispatch function
+###############################################################################
+def get_switch(direction, dev_key, options):
+    """
+    Return the measurement device object
+    """
+    opt_dict = {
+        k[0]: k[1] if len(k) > 1 else None for k in [x.split("=", 1) for x in options]
+    }
+    members = inspect.getmembers(sys.modules[__name__])
+    if 'import' in opt_dict:
+        try:
+            print("Loading external module: {}".format(opt_dict.get('import')))
+            external_module = importlib.import_module(opt_dict.get('import'))
+            members += inspect.getmembers(external_module)
+        except (ModuleNotFoundError, ImportError):
+            print("WARNING: Could not import module '{}'"
+                  .format(opt_dict.get('import')))
+    for _, obj in members:
+        try:
+            if issubclass(obj, SwitchBase) and dev_key == getattr(obj, 'key', ''):
+                return obj(opt_dict)
+        except TypeError:
+            continue
+    return ManualSwitch(direction, opt_dict)
diff --git a/host/utils/uhd_power_cal.py b/host/utils/uhd_power_cal.py
index b71e8f3ab..c598c1058 100644
--- a/host/utils/uhd_power_cal.py
+++ b/host/utils/uhd_power_cal.py
@@ -84,6 +84,12 @@ def parse_args():
         '-o', '--meas-option', default=[], action='append',
         help='Options that are passed to the measurement device')
     parser.add_argument(
+        '--switch', default='manual',
+        help='Type of switch to be used to connect antennas')
+    parser.add_argument(
+        '--switch-option', default=[], action='append',
+        help='Options that are passed to the switch')
+    parser.add_argument(
         '-r', '--rate', type=float,
         help='Sampling rate at which the calibration is performed')
     parser.add_argument(
@@ -194,6 +200,8 @@ def main():
     # not transmitting at full scale
     if args.dir == 'tx':
         meas_dev.power_offset -= 20 * math.log10(args.amplitude)
+    print("=== Initializing port connector...")
+    switch = uhd.usrp.cal.get_switch(args.dir, args.switch, args.switch_option)
     print("=== Initializing USRP calibration object...")
     usrp_cal = uhd.usrp.cal.get_usrp_calibrator(
         usrp, meas_dev, args.dir,
@@ -220,7 +228,7 @@ def main():
                   .format(chan, ant))
             # Set up all the objects
             getattr(usrp, 'set_{}_antenna'.format(args.dir))(ant, chan)
-            meas_dev.update_port(chan, ant)
+            switch.connect(chan, ant)
             usrp_cal.update_port(chan, ant)
             freqs = usrp_cal.init_frequencies(args.start, args.stop, args.step)
             usrp_cal.start() # This will activate siggen