AD9081 real-mode support

adi/ad9081.py

@@ -5,7 +5,7 @@
 from typing import Dict, List
 
 from adi.context_manager import context_manager
-from adi.rx_tx import rx_tx
+from adi.rx_tx import are_channels_complex, rx_tx
 from adi.sync_start import sync_start
 
 
@@ -22,21 +22,32 @@ def _map_to_dict(paths, ch):
     return paths
 
 
-def _sortconv(chans_names, noq=False, dds=False):
+def _sortconv(chans_names, noq=False, dds=False, complex=False):
     tmpI = filter(lambda k: "_i" in k, chans_names)
     tmpQ = filter(lambda k: "_q" in k, chans_names)
 
+    assert not (
+        dds and complex
+    ), "DDS channels cannot have complex names (voltageX_i, voltageX_q)"
+
     def ignoreadc(w):
         return int(w[len("voltage") : w.find("_")])
 
     def ignorealt(w):
         return int(w[len("altvoltage") :])
 
+    def ignorevoltage(w):
+        return int(w[len("voltage") :])
+
     chans_names_out = []
     if dds:
         filt = ignorealt
         tmpI = chans_names
         noq = True
+    elif not complex:
+        filt = ignorevoltage
+        tmpI = chans_names
+        noq = True
     else:
         filt = ignoreadc
 
@@ -85,6 +96,12 @@ def __init__(self, uri=""):
         self._rxadc = self._ctx.find_device("axi-ad9081-rx-hpc")
         self._txdac = self._ctx.find_device("axi-ad9081-tx-hpc")
 
+        # Update Complex data flags
+        if self._rx_complex_data is None:
+            self._rx_complex_data = are_channels_complex(self._rxadc.channels)
+        if self._tx_complex_data is None:
+            self._tx_complex_data = are_channels_complex(self._txdac.channels)
+
         # Get DDC and DUC mappings
         paths = {}
 
@@ -104,8 +121,12 @@ def __init__(self, uri=""):
                 self._dds_channel_names.append(ch._id)
 
         # Sort channel names
-        self._rx_channel_names = _sortconv(self._rx_channel_names)
-        self._tx_channel_names = _sortconv(self._tx_channel_names)
+        self._rx_channel_names = _sortconv(
+            self._rx_channel_names, complex=self._rx_complex_data
+        )
+        self._tx_channel_names = _sortconv(
+            self._tx_channel_names, complex=self._tx_complex_data
+        )
         self._dds_channel_names = _sortconv(self._dds_channel_names, dds=True)
 
         # Map unique attributes to channel properties
@@ -118,7 +139,9 @@ def __init__(self, uri=""):
                 channels = []
                 for fdc in paths[converter][cdc]:
                     channels += paths[converter][cdc][fdc]["channels"]
-                channels = [name for name in channels if "_i" in name]
+                channels = [
+                    name for name in channels if "_q" not in name and "voltage" in name
+                ]
                 if "ADC" in converter:
                     self._rx_coarse_ddc_channel_names.append(channels[0])
                     self._rx_fine_ddc_channel_names += channels
@@ -236,12 +259,14 @@ def rx_main_nco_phases(self, value):
     @property
     def rx_test_mode(self):
         """rx_test_mode: NCO Test Mode"""
-        return self._get_iio_attr_str_single("voltage0_i", "test_mode", False)
+        return self._get_iio_attr_str_single(
+            self._rx_coarse_ddc_channel_names[0], "test_mode", False
+        )
 
     @rx_test_mode.setter
     def rx_test_mode(self, value):
         self._set_iio_attr_single(
-            "voltage0_i", "test_mode", False, value,
+            self._rx_coarse_ddc_channel_names[0], "test_mode", False, value,
         )
 
     @property
@@ -617,24 +642,33 @@ def tx_ddr_offload(self, value):
     @property
     def rx_sample_rate(self):
         """rx_sampling_frequency: Sample rate after decimation"""
-        return self._get_iio_attr_single("voltage0_i", "sampling_frequency", False)
+        return self._get_iio_attr_single(
+            self._rx_coarse_ddc_channel_names[0], "sampling_frequency", False
+        )
 
     @property
     def adc_frequency(self):
         """adc_frequency: ADC frequency in Hz"""
-        return self._get_iio_attr_single("voltage0_i", "adc_frequency", False)
+        return self._get_iio_attr_single(
+            self._rx_coarse_ddc_channel_names[0], "adc_frequency", False
+        )
 
     @property
     def tx_sample_rate(self):
         """tx_sampling_frequency: Sample rate before interpolation"""
         return self._get_iio_attr_single(
-            "voltage0_i", "sampling_frequency", True, self._txdac
+            self._tx_coarse_duc_channel_names[0],
+            "sampling_frequency",
+            True,
+            self._txdac,
         )
 
     @property
     def dac_frequency(self):
         """dac_frequency: DAC frequency in Hz"""
-        return self._get_iio_attr_single("voltage0_i", "dac_frequency", True)
+        return self._get_iio_attr_single(
+            self._tx_coarse_duc_channel_names[0], "dac_frequency", True
+        )
 
     @property
     def jesd204_fsm_ctrl(self):

adi/ad9172.py

@@ -34,7 +34,7 @@ def __init__(self, uri=""):
             ):
                 self._tx_channel_names.append(chan.id)
 
-        self._tx_channel_names = _sortconv(self._tx_channel_names)
+        self._tx_channel_names = _sortconv(self._tx_channel_names, complex=True)
 
         tx.__init__(self)
 

adi/obs.py

@@ -42,7 +42,7 @@ def __init__(self, ctx, obs_dev, channel_names, complex_data=True):
         self._ctx = ctx
         self._rxadc = obs_dev
         self._rx_channel_names = channel_names
-        self._complex_data = complex_data
+        self._rx_complex_data = complex_data
         rx.__init__(self)
 
     def __del__(self):

adi/rx_tx.py

@@ -21,6 +21,20 @@
     from adi.compat import compat_libiio_v0_tx as ctx
 
 
+def are_channels_complex(channels: Union[List[str], List[iio.Channel]]) -> bool:
+    """Check if channels are complex or not
+
+    Args:
+        channels: List of channel names or iio.Channel objects
+    """
+    for channel in channels:
+        if isinstance(channel, iio.Channel):
+            channel = channel.id
+        if channel.endswith("_i") or channel.endswith("_q"):
+            return True
+    return False
+
+
 class phy(attribute):
     _ctrl: iio.Device = []
 
@@ -31,6 +45,8 @@ def __del__(self):
 class rx_tx_common(attribute):
     """Common functions for RX and TX"""
 
+    _complex_data = False
+
     def _annotate(self, data, cnames: List[str], echans: List[int]):
         return {cnames[ec]: data[i] for i, ec in enumerate(echans)}
 
@@ -40,7 +56,8 @@ class rx_core(rx_tx_common, metaclass=ABCMeta):
 
     _rxadc: iio.Device = []
     _rx_channel_names: List[str] = []
-    _complex_data = False
+    # Set to True if complex data for RX only, overrides _complex_data
+    _rx_complex_data = None
     _rx_data_type = np.int16
     _rx_data_si_type = np.int16
     _rx_shift = 0
@@ -59,6 +76,16 @@ def __init__(self, rx_buffer_size=1024):
         self.rx_enabled_channels = rx_enabled_channels
         self.rx_buffer_size = rx_buffer_size
 
+    @property
+    def _complex_data(self) -> bool:
+        """Data to/from device is quadrature (Complex).
+        When True ADC channel pairs are used together and the
+        rx method will generate complex data types.
+        """
+        if self._rx_complex_data is None:
+            return super()._complex_data
+        return self._rx_complex_data
+
     @property
     def rx_channel_names(self) -> List[str]:
         """rx_channel_names: List of RX channel names"""
@@ -271,7 +298,8 @@ class tx_core(dds, rx_tx_common, metaclass=ABCMeta):
     _tx_buffer_size = 1024
     _txdac: iio.Device = []
     _tx_channel_names: List[str] = []
-    _complex_data = False
+    # Set to True if complex data for TX only, overrides _complex_data
+    _tx_complex_data = None
     _tx_data_type = None
     _txbuf = None
     _output_byte_filename = "out.bin"
@@ -294,6 +322,16 @@ def __del__(self):
                 v.enabled = False
         self._txdac = []
 
+    @property
+    def _complex_data(self):
+        """Data to device is quadrature (Complex).
+        When True DAC channel pairs are used together and the
+        tx method will assume complex samples.
+        """
+        if self._tx_complex_data is None:
+            return super()._complex_data
+        return self._tx_complex_data
+
     @property
     def tx_cyclic_buffer(self):
         """tx_cyclic_buffer: Enable cyclic buffer for TX"""

requirements.txt

@@ -1,3 +1,3 @@
 numpy>=1.20
-pylibiio==0.23.1
+pylibiio>=0.23.1
 paramiko

requirements_dev.txt

@@ -8,7 +8,7 @@ scapy
 scipy<=1.12.0
 pytest-cov
 coveralls
-pytest-libiio>=0.0.18
+pytest-libiio>=0.0.20
 bump2version
 pytest-html==3.2.0
 plotly-express

test/emu/hardware_map.yml

@@ -125,6 +125,18 @@ ad9081_tdd:
       - data_devices:
           - iio:device3
           - iio:device4
+ad9081_full_bw:
+  - axi-ad9081-tx-hpc
+  - axi-ad9081-rx-hpc
+  - axi-core-tdd
+  - hmc7044
+  - pyadi_iio_class_support:
+      - ad9081
+  - emulate:
+      - filename: ad9081_full_bw_mock.xml
+      - data_devices:
+          - iio:device2
+          - iio:device3
 ad9084:
   - axi-ad9084-tx-hpc
   - axi-ad9084-rx-hpc

test/test_ad9081.py

@@ -356,3 +356,26 @@ def test_ad9081_nco_loopback(
 
 
 #########################################
+@pytest.mark.iio_hardware("ad9081_full_bw")
+def test_full_bw_rx(iio_uri):
+    import adi
+
+    dev = adi.ad9081(uri=iio_uri)
+
+    assert not dev._rx_complex_data
+    assert dev._tx_complex_data
+
+    assert dev._rx_fine_ddc_channel_names == [
+        "voltage0",
+        "voltage1",
+        "voltage2",
+        "voltage3",
+    ]
+    assert dev._rx_coarse_ddc_channel_names == ["voltage0", "voltage2"]
+    assert dev._tx_fine_duc_channel_names == [
+        "voltage0",
+        "voltage1",
+        "voltage2",
+        "voltage3",
+    ]
+    assert dev._tx_coarse_duc_channel_names == ["voltage0", "voltage1"]