cal: Add support for NI ModInst measurement devices

Power measurements for TX calibration can be done using NI-RFSA devices
and the RFmx SpecAn library. Use "rfsa" as meas_device option for this.
The implementation mimics the behaviour of the "RFMXSpecAn TXP (Basic)"
example.

Signal generation for RX calibration can be done using NI-RFSG devices.
Use "rfsg" as meas_device option for this. The implementation mimics
the behaviour of the "RFSG Frequency Sweep" example.

The device can be selected by passing its name in the option string.
This is only necessary if more than one device of the same family is
installed in the system.

The support is limited to Windows operating System. Drivers for RFSA and
RFSG must be installed as well as the RFmx SpecAn library. The "nimodinst"
Python package must be installed to be able to detect the devices.

The implementation uses ctypes to call into the C-API of the device drivers.
Only the bare minimum to fulfill the calibration requirements is implemented.

2 files changed, 455 insertions, 0 deletions

diff --git a/host/python/uhd/usrp/cal/meas_device.py b/host/python/uhd/usrp/cal/meas_device.py
index d6b40eb84..fd4455da0 100644
--- a/host/python/uhd/usrp/cal/meas_device.py
+++ b/host/python/uhd/usrp/cal/meas_device.py
@@ -15,6 +15,7 @@ import numpy
 import uhd
 from .tone_gen import ToneGenerator
 from .visa import get_visa_device
+from .ni_rf_instr import RFSADevice, RFSGDevice
 
 ###############################################################################
 # Base Classes
@@ -214,7 +215,32 @@ class ManualPowerGenerator(SignalGeneratorBase):
               .format(freq/1e6))
     # pylint: enable=no-self-use
 
+##############################################################################
+# RFSA: Run through a NI-RFSA device, using RFmx library
 ###############################################################################
+class RfsaPowerMeter(PowerMeterBase):
+    """
+    Power meter using RFmx TXP measurement on NI-RFSA devices.
+    """
+    key = 'rfsa'
+
+    def __init__(self, options):
+        super().__init__(options)
+        self.device = RFSADevice(options)
+
+    def set_frequency(self, freq):
+        """
+        Set the frequency of the measurement device.
+        """
+        self.device.set_frequency(freq)
+
+    def _get_power(self):
+        """
+        Return the current measured power in dBm.
+        """
+        return self.device.get_power_dbm()
+
+##############################################################################
 # VISA: Run through a VISA device, using SCPI commands
 ###############################################################################
 class VisaPowerMeter(PowerMeterBase):
@@ -323,6 +349,43 @@ class USRPPowerGenerator(SignalGeneratorBase):
         return self._usrp.get_tx_power_reference(self._chan) + self._pwr_dbfs
 
 ###############################################################################
+# RFSG: NI signal generator family
+###############################################################################
+class RFSGPowerGenerator(SignalGeneratorBase):
+    """
+    Power Generator using NI-RFSG devices.
+    """
+    key = 'rfsg'
+
+    def __init__(self, options):
+        super().__init__(options)
+        self.device = RFSGDevice(options)
+
+    def enable(self, enb=True):
+        """
+        Turn tone generation on and off
+        """
+        self.device.enable(enb)
+
+    def set_frequency(self, freq):
+        """
+        Set the center frequency of the generated signal.
+        """
+        self.device.set_frequency(freq)
+
+    def _set_power(self, power_dbm):
+        """
+        Set the output power of the device in dBm.
+        """
+        return self.device.set_power(power_dbm)
+
+    def _get_power(self):
+        """
+        Get the output power of the device in dBm.
+        """
+        return self.device.get_power()
+
+###############################################################################
 # The dispatch function
 ###############################################################################
 def get_meas_device(direction, dev_key, options):
diff --git a/host/python/uhd/usrp/cal/ni_rf_instr.py b/host/python/uhd/usrp/cal/ni_rf_instr.py
new file mode 100644
index 000000000..1b6d34f4e
--- /dev/null
+++ b/host/python/uhd/usrp/cal/ni_rf_instr.py
@@ -0,0 +1,392 @@
+#
+# Copyright 2020 Ettus Research, a National Instruments Brand
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+"""
+RFSA/RFSG-type measurement devices
+"""
+import os
+import platform
+if platform.system() == 'Windows':
+    from ctypes import WinError, CDLL, byref, POINTER
+    from ctypes import c_bool, c_int, c_int32, c_double, c_char_p
+
+# pylint: disable=unused-argument
+def _check(result, func, arguments):
+    """
+    Test result for error. Throw WinError, if result is not 0 using result as
+    error index. This is a convenient default implementation for WIN API calls
+    """
+    assert os.name == "nt" # throwing WinError works on Windows only.
+    if result != 0:
+        error = WinError(result,
+                         "External library call failed with {}".format(result))
+        raise error
+
+def _wrap(lib, func_name, res_type, arg_types, check_error=True):
+    """
+    Ease usage of API functions imported using ctypes library.
+    :param lib: CDLL loaded by ctypes
+    :param func_name: Function import to uses
+    :param res_type: result type of call (as ctype)
+    :param arg_types: argument list of call (as ctype)
+    :param check_error: error checker (default uses _check)
+    :return: Python function
+    """
+    func = lib.__getattr__(func_name)
+    func.restype = res_type
+    func.argtypes = arg_types
+    if check_error:
+        func.errcheck = _check
+    return func
+
+def get_modinst_devices(driver):
+    """
+    Creates a list of instruments matching the current driver. Set
+    self._driver before calling this method
+    :return: list of nimodinst device objects
+    :throws: RuntimeError if called on non-Windows systems.
+    """
+    if platform.system() != 'Windows':
+        raise RuntimeError('Cannot connect to ModInst '
+                           'devices on non-Windows systems.')
+    # make sure NI ModInst package is only loaded when required
+    # pylint: disable=import-outside-toplevel
+    import nimodinst
+    session = nimodinst.Session(driver)
+    return session.devices
+
+def get_modinst_device(driver, name):
+    """
+    Get exact on device for the current driver. Use (optional) name to
+    narrow the list if more than one device is installed for the current
+    driver.
+    :param name: name of devices to get
+    :return: NI ModInst device class
+    :throws: RuntimeError if no or more than one device was found.
+    """
+    devices = [device for device in get_modinst_devices(driver)
+               if not name or name == device.device_name]
+    if len(devices) > 1:
+        print("Found %s devices:" % driver)
+        for device in devices:
+            print("* %s" % device.device_name)
+        raise RuntimeError("Found more than one measurement device. "
+                           "Please limit the query!")
+    if not devices:
+        raise RuntimeError("No measurement device found!")
+    return devices[0]
+
+
+class RFSADevice:
+    """
+    Class to measure power using devices from the NI-RFSA family. This class
+    make use of the RFmx SpecAn TXP measurement. So this library must be
+    installed as well in addition to the RFSA driver.
+    """
+    RFMXSPECAN_ATTR_REFERENCE_LEVEL = 0x00100002
+
+    RFMXSPECAN_VAL_TXP_RBW_FILTER_TYPE_GAUSSIAN = 1
+    RFMXSPECAN_VAL_TXP_AVERAGING_ENABLED_TRUE = 1
+    RFMXSPECAN_VAL_TXP_AVERAGING_TYPE_RMS = 0
+
+    def __init__(self, options):
+        self.session = c_int(0)
+
+        self._generate_specan_calls()
+
+        device = get_modinst_device("NI-RFSA", options.get("name", None))
+
+        self._RFmxSpecAn_Initialize(device.device_name.encode(),
+                                    "".encode(), byref(self.session), 0)
+
+    def __del__(self):
+        if self.session.value > 0:
+            self._RFmxSpecAn_Close(self.session, 0)
+
+    def _generate_specan_calls(self):
+        """
+        Generate needed Python function in class to do power measurements.
+        """
+        lib = CDLL("niRFmxSpecAn.dll")
+        # disable invalid name error to use function names as declared in C-DLL
+        # pylint: disable=invalid-name
+        self._RFmxSpecAn_Initialize = _wrap(
+            lib,
+            "RFmxSpecAn_Initialize",
+            c_int,
+            [c_char_p, c_char_p, POINTER(c_int), c_int]
+        )
+        self._RFmxSpecAn_Close = _wrap(
+            lib,
+            "RFmxSpecAn_Close",
+            c_int,
+            [c_int, c_int]
+        )
+        self._RFmxSpecAn_SetSelectedPorts = _wrap(
+            lib,
+            "RFmxSpecAn_SetSelectedPorts",
+            c_int,
+            [c_int, c_char_p, c_char_p]
+        )
+        self._RFmxSpecAn_CfgRF = _wrap(
+            lib,
+            "RFmxSpecAn_CfgRF",
+            c_int,
+            [c_int, c_char_p, c_double, c_double, c_double]
+        )
+        self._RFmxSpecAn_TXPCfgRBWFilter = _wrap(
+            lib,
+            "RFmxSpecAn_TXPCfgRBWFilter",
+            c_int,
+            [c_int, c_char_p, c_double, c_int32, c_double]
+        )
+        self._RFmxSpecAn_TXPCfgMeasurementInterval = _wrap(
+            lib,
+            "RFmxSpecAn_TXPCfgMeasurementInterval",
+            c_int32,
+            [c_int, c_char_p, c_double]
+        )
+        self._RFmxSpecAn_TXPCfgAveraging = _wrap(
+            lib,
+            "RFmxSpecAn_TXPCfgAveraging",
+            c_int32,
+            [c_int, c_char_p, c_int32, c_int32, c_int32,]
+        )
+        self._RFmxSpecAn_TXPRead = _wrap(
+            lib,
+            "RFmxSpecAn_TXPRead",
+            c_int,
+            [c_int, c_char_p, c_double, POINTER(c_double),
+             POINTER(c_double), POINTER(c_double), POINTER(c_double)]
+        )
+        self._RFmxSpecAn_GetAttributeF64 = _wrap(
+            lib,
+            "RFmxSpecAn_GetAttributeF64",
+            c_int,
+            [c_int, c_char_p, c_int, POINTER(c_double)]
+        )
+        self._RFmxSpecAn_SetAttributeF64 = _wrap(
+            lib,
+            "RFmxSpecAn_SetAttributeF64",
+            c_int,
+            [c_int, c_char_p, c_int, c_double]
+        )
+
+
+    def init_power_meter(self):
+        """
+        Initialize power meter (derived from "RFmxSpecAn TXP(Basic)" example)
+        """
+        rbw = c_double(100E3)
+        filt = c_int32(RFSADevice.RFMXSPECAN_VAL_TXP_RBW_FILTER_TYPE_GAUSSIAN)
+        rrc_alpha = c_double(0)
+        meas_int = c_double(1E-3)
+        avg = c_int32(RFSADevice.RFMXSPECAN_VAL_TXP_AVERAGING_ENABLED_TRUE)
+        avg_count = c_int32(10)
+        avg_type = c_int32(RFSADevice.RFMXSPECAN_VAL_TXP_AVERAGING_TYPE_RMS)
+        self._RFmxSpecAn_SetSelectedPorts(self.session, b"", b"")
+        self._RFmxSpecAn_TXPCfgRBWFilter(self.session, b"", rbw, filt, rrc_alpha)
+        self._RFmxSpecAn_TXPCfgMeasurementInterval(self.session, b"", meas_int)
+        self._RFmxSpecAn_TXPCfgAveraging(self.session, b"", avg, avg_count, avg_type)
+
+    def set_frequency(self, freq):
+        """
+        Set measurement frequency and reset ref level.
+        """
+        freq = c_double(freq)
+        ref_level = c_double(-20) # start with a low ref level
+        ext_att = c_double(0) # calibration script takes responsibility for
+                              # external attenuation, so no need to set it here
+        self._RFmxSpecAn_CfgRF(self.session, b"", freq, ref_level, ext_att)
+
+    def _get_attribute(self, attr_id):
+        """
+        Wrapper to ease read RFSA double attributes.
+        """
+        result = c_double(0)
+        self._RFmxSpecAn_GetAttributeF64(self.session, b"", attr_id, byref(result))
+        return result.value
+
+    def _set_attribute(self, attr_id, value):
+        """
+        Wrapper to ease write RFSA double attributes.
+        """
+        attr = c_double(value)
+        self._RFmxSpecAn_SetAttributeF64(self.session, b"", attr_id, attr)
+
+    def get_reference_level(self):
+        """
+        Return reference level used for measurements
+        """
+        return self._get_attribute(RFSADevice.RFMXSPECAN_ATTR_REFERENCE_LEVEL)
+
+    def set_reference_level(self, value):
+        """
+        Set reference level used for measurements
+        """
+        self._set_attribute(RFSADevice.RFMXSPECAN_ATTR_REFERENCE_LEVEL, value)
+
+    def get_power_dbm(self):
+        """
+        Measure power at current frequency. Reference level starts at -20dBm
+        (see set_frequency) and is increased in 5dB steps until measurement
+        succeeds. Returns the measured averaged power.
+        """
+        timeout = c_double(1.0)
+        mean_val = c_double(0)
+        peak2avg_val = c_double(0)
+        min_val = c_double(0)
+        max_val = c_double(0)
+        while True:
+            try:
+                self._RFmxSpecAn_TXPRead(self.session, b"", timeout,
+                                         byref(mean_val), byref(peak2avg_val),
+                                         byref(min_val), byref(max_val))
+                return mean_val.value
+            except OSError as ex:
+                # increase ref level on ADC or DSA overload
+                if getattr(ex, 'winerror') in [373002, 373003]:
+                    self.set_reference_level(self.get_reference_level() + 5)
+                else:
+                    raise ex
+
+
+class RFSGDevice:
+    """
+    Class to generate CW tone using devices from the NI-RFSG family
+    """
+    NIRFSG_ATTR_POWER_LEVEL = 1250002
+    NIRFSG_VAL_CW = 1000
+
+    def __init__(self, options):
+        super().__init__()
+        self.session = c_int(0)
+
+        self._generate_rfsg_calls()
+
+        device = get_modinst_device("NI-RFSG", options.get("name", None))
+
+        id_query = c_bool(False)
+        reset = c_bool(True)
+        self._niRFSG_init(device.device_name.encode(), id_query, reset,
+                          byref(self.session))
+        self._niRFSG_ConfigureGenerationMode(self.session,
+                                             RFSGDevice.NIRFSG_VAL_CW)
+
+    def __del__(self):
+        if self.session.value > 0:
+            self._niRFSG_close(self.session, 0)
+
+    def _generate_rfsg_calls(self):
+        """
+        Generate needed Python function in class to do power measurements.
+        For RFSGs the library has different names depending on the platform
+        used. So platform is checked to select proper library name.
+        """
+        lib_name = "niRFSG_64.dll" if platform.architecture()[0] == '64bit' \
+              else "niRFSG.dll"
+        lib = CDLL(lib_name)
+        # disable invalid name error to use function names as declared in C-DLL
+        # pylint: disable=invalid-name
+        self._niRFSG_init = _wrap(
+            lib,
+            "niRFSG_init",
+            c_int,
+            [c_char_p, c_bool, c_bool, POINTER(c_int)]
+        )
+        self._niRFSG_close = _wrap(
+            lib,
+            "niRFSG_close",
+            c_int,
+            [c_int, c_int]
+        )
+
+        self._niRFSG_SetAttributeViReal64 = _wrap(
+            lib,
+            "niRFSG_SetAttributeViReal64",
+            c_int,
+            [c_int, c_char_p, c_int, c_double]
+        )
+
+        self._niRFSG_GetAttributeViReal64 = _wrap(
+            lib,
+            "niRFSG_GetAttributeViReal64",
+            c_int,
+            [c_int, c_char_p, c_int, POINTER(c_double)]
+        )
+
+        self._niRFSG_ConfigureRF = _wrap(
+            lib,
+            "niRFSG_ConfigureRF",
+            c_int,
+            [c_int, c_double, c_double]
+        )
+
+        self._niRFSG_ConfigureGenerationMode = _wrap(
+            lib,
+            "niRFSG_ConfigureGenerationMode",
+            c_int,
+            [c_int, c_int]
+        )
+
+        self._niRFSG_Initiate = _wrap(
+            lib,
+            "niRFSG_Initiate",
+            c_int,
+            [c_int]
+        )
+
+        self._niRFSG_Abort = _wrap(
+            lib,
+            "niRFSG_Abort",
+            c_int,
+            [c_int]
+        )
+
+    def enable(self, enb=True):
+        """
+        Switch tone generation on of off
+        """
+        if enb:
+            print("[SigGen] Starting tone generator.")
+            self._niRFSG_Initiate(self.session)
+        else:
+            print("[SigGen] Stopping tone generator.")
+            self._niRFSG_Abort(self.session)
+
+    def set_frequency(self, freq):
+        """
+        Tune tone generator to new center frequency.
+        """
+        print("[SigGen] Tuning signal to {:.3f} MHz.".format(freq/1e6))
+        self._niRFSG_ConfigureRF(self.session, freq, self.get_power())
+
+    def _get_attribute(self, attr_id):
+        """
+        Wrapper to ease reading RFSG double attributes.
+        """
+        result = c_double(0)
+        self._niRFSG_GetAttributeViReal64(self.session, b"", attr_id, byref(result))
+        return result.value
+
+    def _set_attribute(self, attr_id, value):
+        """
+        Wrapper to ease writing RFSG double attributes.
+        """
+        attr = c_double(value)
+        self._niRFSG_SetAttributeViReal64(self.session, b"", attr_id, attr)
+
+    def set_power(self, power_dbm):
+        """
+        Set power of output signal and return actual power.
+        """
+        self._set_attribute(RFSGDevice.NIRFSG_ATTR_POWER_LEVEL, power_dbm)
+        return self.get_power()
+
+    def get_power(self):
+        """
+        Get power of output signal.
+        """
+        return self._get_attribute(RFSGDevice.NIRFSG_ATTR_POWER_LEVEL)