Eye scan support

adi/ad9081.py

@@ -5,6 +5,7 @@
 from typing import Dict, List
 
 from adi.context_manager import context_manager
+from adi.jesd import jesd_eye_scan
 from adi.rx_tx import rx_tx
 from adi.sync_start import sync_start
 
@@ -66,7 +67,9 @@ class ad9081(rx_tx, context_manager, sync_start):
 
     _path_map: Dict[str, Dict[str, Dict[str, List[str]]]] = {}
 
-    def __init__(self, uri=""):
+    def __init__(
+        self, uri="", username="root", password="analog", disable_jesd_control=True
+    ):
 
         # Reset default channel names
         self._rx_channel_names = []
@@ -85,6 +88,9 @@ def __init__(self, uri=""):
         self._rxadc = self._ctx.find_device("axi-ad9081-rx-hpc")
         self._txdac = self._ctx.find_device("axi-ad9081-tx-hpc")
 
+        if not disable_jesd_control and jesd_eye_scan:
+            self._jesd = jesd_eye_scan(self, uri, username=username, password=password)
+
         # Get DDC and DUC mappings
         paths = {}
 

adi/jesd.py

@@ -6,6 +6,6 @@
 
 try:
     from .sshfs import sshfs
-    from .jesd_internal import jesd
+    from .jesd_internal import jesd, jesd_eye_scan
 except ImportError:
     jesd = None

adi/jesd_internal.py

@@ -5,7 +5,7 @@
 from .sshfs import sshfs
 
 
-class jesd:
+class jesd(object):
     """JESD Monitoring"""
 
     def __init__(self, address, username="root", password="analog"):
@@ -24,17 +24,16 @@ def __init__(self, address, username="root", password="analog"):
 
     def find_lanes(self):
         self.lanes = {}
+        if len(self.dirs) == 0:
+            raise Exception("No JESD links found")
         for dr in self.dirs:
             if "-rx" in dr:
                 self.lanes[dr] = []
-                lanIndx = 0
-                while 1:
-                    li = "/lane{}_info".format(lanIndx)
-                    if self.fs.isfile(self.rootdir + dr + li):
-                        self.lanes[dr].append(li)
-                        lanIndx += 1
-                    else:
-                        break
+                subdirs = self.fs.listdir(f"{self.rootdir}{dr}")
+                for subdir in subdirs:
+                    if "lane" in subdir and "info" in subdir:
+                        if self.fs.isfile(f"{self.rootdir}{dr}/{subdir}"):
+                            self.lanes[dr].append(subdir)
 
     def find_jesd_dir(self):
         dirs = self.fs.listdir(self.rootdir)
@@ -76,3 +75,168 @@ def get_all_link_statuses(self):
 
     def get_all_statuses(self):
         return {dr: self.decode_status(self.get_status(dr)) for dr in self.dirs}
+
+
+class jesd_eye_scan(jesd):
+    _jesd_es_duration_ms = 10
+    _jesd_prbs = 7
+    _max_possible_lanes_index = 24
+
+    _half_rate = {"mode": "Half Rate", "scale": 1}
+    _quarter_rate = {"mode": "Quarter Rate", "scale": 4}
+
+    lanes = {}
+
+    def __init__(self, parent, address, username="root", password="analog"):
+        """JESD204 Eye Scan
+
+        Args:
+            parent (adi.ad9081): Parent AD9081 instance
+            address (str): IP address of the device
+            username (str, optional): Username. Defaults to "root".
+            password (str, optional): Password. Defaults to "analog".
+        """
+        super().__init__(address, username, password)
+        self._parent = parent
+        self._actual_lane_numbers = {}
+        for device in self.lanes.keys():
+            self._actual_lane_numbers[device] = self._get_actual_lane_numbers(device)
+
+    def _get_actual_lane_numbers(self, device: str):
+        """Get actual lane numbers from device
+
+        The sysfs lanes always go 0-(N-1) where N is the number of lanes. But these
+        are not always the actual lane numbers. This function gets the actual lane
+        numbers from the device.
+        """
+        # Check if supported
+        if "bist_2d_eyescan_jrx" not in self._parent._ctrl.debug_attrs:
+            raise Exception("2D eye scan not supported on platform")
+
+        if device not in self.lanes.keys():
+            raise Exception(f"Device {device} not found.")
+        num_lanes = len(self.lanes[device])
+
+        actual_lane_numbers = []
+        for lane_index in range(self._max_possible_lanes_index):
+            try:
+                self._parent._set_iio_debug_attr_str(
+                    "bist_2d_eyescan_jrx",
+                    f"{lane_index} {self._jesd_prbs} {self._jesd_es_duration_ms}",
+                )
+                actual_lane_numbers.append(str(lane_index))
+                if len(actual_lane_numbers) == num_lanes:
+                    break
+            except OSError:
+                continue
+
+        if len(actual_lane_numbers) != num_lanes:
+            raise Exception(
+                f"Could not find all lanes for device {device}. Expected {num_lanes}, found {len(actual_lane_numbers)}."
+            )
+
+        return actual_lane_numbers
+
+    def get_eye_data(self, device=None, lanes=None):
+        """Get JESD204 eye scan data
+
+        Args:
+            device (str, optional): Device to get data for. Defaults to None which will get data for the first device found.
+            lanes (list, optional): List of lanes to get data for. Defaults to None which will get data for all lanes.
+
+        Returns:
+            dict: Dictionary of lane data. Keys are lane numbers, values are dictionaries with keys "x", "y1", "y2", and "mode".
+                where "x" is the x-axis data SPO, "y1" is the y-axis data for the first eye, "y2" is the y-axis data for the second eye,
+                in volts
+
+        """
+        # Check if supported
+        if "bist_2d_eyescan_jrx" not in self._parent._ctrl.debug_attrs:
+            raise Exception("2D eye scan not supported on platform")
+
+        if device is None:
+            device = list(self._actual_lane_numbers.keys())[0]
+        if device not in self._actual_lane_numbers.keys():
+            raise Exception(f"Device {device} not found.")
+
+        available_lanes = self._actual_lane_numbers[device]
+
+        if not isinstance(lanes, list) and lanes is not None:
+            lanes = [lanes]
+        if lanes is None:
+            if len(available_lanes) == 0:
+                raise Exception("No lanes found. Please run find_lanes() first")
+            lanes = available_lanes
+
+        # Check if lanes are valid
+        for lane in lanes:
+            if lane not in available_lanes:
+                raise Exception(f"Lane {lane} not found for device {device}.")
+
+        # Enable PRBS on TX side
+        devices_root = "/sys/bus/platform/devices/"
+        dev_list = self.fs.listdir(devices_root)
+        tx_dev = next((dev for dev in dev_list if "adxcvr-tx" in dev), None)
+        if not tx_dev:
+            raise Exception("No adxcvr-tx device found. Cannot enable PRBS.")
+
+        self.fs.echo_to_fd("7", f"{devices_root}/{tx_dev}/prbs_select")
+
+        lane_eye_data = {}
+
+        print("Hold tight while we get the eye data...")
+
+        for lane in lanes:
+            # Configure BIST
+            print(f"Getting eye data for lane {lane}")
+
+            self._parent._set_iio_debug_attr_str(
+                "bist_2d_eyescan_jrx",
+                f"{lane} {self._jesd_prbs} {self._jesd_es_duration_ms}",
+            )
+
+            eye_data = self._parent._get_iio_debug_attr_str("bist_2d_eyescan_jrx")
+
+            x = []
+            y1 = []
+            y2 = []
+
+            for eye_line in eye_data.splitlines():
+                if "#" in eye_line:
+                    info = [int(s) for s in eye_line.split() if s.isdigit()]
+                    if info[1] == 64:
+                        mode = self._half_rate["mode"]
+                        scale = self._half_rate["scale"]
+                    else:
+                        mode = self._quarter_rate["mode"]
+                        scale = self._quarter_rate["scale"]
+                    if info[0] != int(lane):
+                        print("Invalid lane number for eye data")
+                        print(f"Expected {lane}, got {info[0]}")
+                else:
+                    spo = [float(x) for x in eye_line.split(",")]
+                    x.append(spo[0])
+                    y1.append(spo[1] * scale)
+                    y2.append(spo[2] * scale)
+
+            if len(x) == 0:
+                raise Exception(f"No eye data found for lane {lane}.")
+
+            graph_helpers = {
+                "xlim": [-info[1] / 2, info[1] / 2 - 1],
+                "ylim": [-256, 256],
+                "xlabel": "SPO",
+                "ylabel": "EYE Voltage (mV)",
+                "title": "JESD204 2D Eye Scan",
+                "rate_gbps": info[2] / 1000000,
+            }
+
+            lane_eye_data[lane] = {
+                "x": x,
+                "y1": y1,
+                "y2": y2,
+                "mode": mode,
+                "graph_helpers": graph_helpers,
+            }
+
+        return lane_eye_data

adi/sshfs.py

@@ -51,3 +51,8 @@ def listdir(self, path):
     def gettext(self, path, *kargs, **kwargs):
         stdout, _ = self._run(f"cat {path}")
         return stdout
+
+    def echo_to_fd(self, data, path):
+        if not self.isfile(path):
+            raise FileNotFoundError(f"No such file: {path}")
+        self._run(f"echo '{data}' > {path}")

examples/ad9081_jesd_eye_diagram.py

@@ -0,0 +1,52 @@
+import time
+
+import adi
+import matplotlib.pyplot as plt
+from scipy import signal
+
+dev = adi.ad9081("ip:10.44.3.92", disable_jesd_control=False)
+
+# Configure properties
+print("--Setting up chip")
+
+dev._ctx.set_timeout(90000)
+
+fig = plt.figure()
+
+eye_data_per_lane = dev._jesd.get_eye_data()
+
+num_lanes = len(eye_data_per_lane.keys())
+
+for i, lane in enumerate(eye_data_per_lane):
+
+    x = eye_data_per_lane[lane]["x"]
+    y1 = eye_data_per_lane[lane]["y1"]
+    y2 = eye_data_per_lane[lane]["y2"]
+
+    ax1 = plt.subplot(int(num_lanes / 2), 2, int(i) + 1)
+    plt.scatter(x, y1, marker="+", color="blue")
+    plt.scatter(x, y2, marker="+", color="red")
+    plt.xlim(eye_data_per_lane[lane]["graph_helpers"]["xlim"])
+    plt.xlabel(eye_data_per_lane[lane]["graph_helpers"]["xlabel"])
+    plt.ylabel(eye_data_per_lane[lane]["graph_helpers"]["ylabel"])
+    plt.rcParams["axes.titley"] = 1.0  # y is in axes-relative coordinates.
+    plt.rcParams["axes.titlepad"] = -14  # pad is in points...
+    plt.title(f" Lane {lane}", loc="left", fontweight="bold")
+    fig.suptitle(
+        f"JESD204 MxFE 2D Eye Scan ({eye_data_per_lane[lane]['mode']}) Rate {eye_data_per_lane[lane]['graph_helpers']['rate_gbps']} Gbps"
+    )
+    plt.axvline(0, color="black")  # vertical
+    plt.axhline(0, color="black")  # horizontal
+    plt.grid(True)
+    # Add secondary x-axis
+    x_norm = [round(n * 0.1, 2) for n in range(11)]
+    x.sort()
+    x = np.linspace(min(x), max(x), 11)
+
+    ax2 = ax1.twiny()
+    ax2.set_xlim(ax1.get_xlim())
+    ax2.set_xticks(x)
+    ax2.set_xticklabels(x_norm)
+    ax2.set_xlabel("Unit Interval (UI)")
+
+plt.show()

test/conftest.py

@@ -14,6 +14,7 @@
 from test.generics import iio_attribute_single_value
 from test.globals import *
 from test.html import pytest_html_report_title, pytest_runtest_makereport
+from test.jesd import check_jesd_links
 
 import adi
 import numpy as np
@@ -202,3 +203,8 @@ def test_verify_overflow(request):
 @pytest.fixture()
 def test_verify_underflow(request):
     yield verify_underflow
+
+
+@pytest.fixture()
+def test_check_jesd_links(request):
+    yield check_jesd_links

test/jesd.py

@@ -0,0 +1,26 @@
+import time
+
+import adi
+import pytest
+
+
+def check_jesd_links(classname, uri, iterations=4):
+    """Check that the JESD links are up and in DATA mode
+
+    Args:
+        classname (str): The name of the class to instantiate
+        uri (str): The URI of the device to connect to
+        iterations (int): The number of times to check the JESD links
+    """
+
+    sdr = eval(f"{classname}(uri='{uri}', disable_jesd_control=False)")
+
+    for _ in range(iterations):
+        # Check that the JESD links are up
+        links = sdr._jesd.get_all_statuses()
+        for link in links:
+            print(f"Link {link} status: \n{links[link]}")
+            assert links[link]["enabled"] == "enabled", f"Link {link} is down"
+            assert links[link]["Link status"] == "DATA", f"Link {link} not in DATA mode"
+
+        time.sleep(1)

test/test_ad9081.py

@@ -19,6 +19,12 @@ def scale_field(param_set, iio_uri):
     return param_set
 
 
+#########################################
+@pytest.mark.iio_hardware(hardware)
+def test_ad9081_jesd_links(test_check_jesd_links, iio_uri):
+    test_check_jesd_links(classname, iio_uri)
+
+
 #########################################
 @pytest.mark.iio_hardware(hardware)
 @pytest.mark.parametrize("classname", [(classname)])