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
 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:
+            self._jesd = jesd(uri, username=username, password=password)
+
         # Get DDC and DUC mappings
         paths = {}
 

adi/jesd_internal.py

@@ -5,7 +5,94 @@
 from .sshfs import sshfs
 
 
-class jesd:
+class eye_scan(object):
+    _jesd_es_duration_ms = 100
+    _jesd_prbs = 7
+
+    _half_rate = {"mode": "Half Fate", "scale": 0.004}
+    _quarter_rate = {"mode": "Quarter Rate", "scale": 0.001}
+
+    lanes = {}
+
+    def get_eye_data(self, lanes=None):
+        """Get JESD204 eye scan data
+
+        Args:
+            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._ctrl.debug_attrs:
+            raise Exception("2D eye scan not supported on platform")
+
+        if not isinstance(lanes, list) and lanes is not None:
+            lanes = [lanes]
+        if lanes is None:
+            if len(self.lanes) == 0:
+                raise Exception("No lanes found. Please run find_lanes() first")
+            lanes = list(self.lanes.keys())
+        for lane in lanes:
+            if lane not in self.lanes.keys():
+                raise Exception(f"Lane {lane} not found.")
+
+        lane_eye_data = {}
+
+        for lane in lanes:
+            # Configure BIST
+            self._set_iio_debug_attr_str(
+                "bist_2d_eyescan_jrx",
+                f"{lane} {self._jesd_prbs} {self._jesd_es_duration_ms}",
+            )
+
+            eye_data = self._get_iio_debug_attr_str("bist_2d_eyescan_jrx_data")
+
+            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] != lane:
+                        print("Invalid lane number for eye data")
+                        print(f"Expected {lane}, got {info[0]}")
+                else:
+                    spo = [int(x) for x in eye_line.split(",")]
+                    x.append(spo[0])
+                    y1.append(spo[1] * scale)
+                    y2.append(spo[2] * scale)
+
+            graph_helpers = {
+                "xlim": [-info[1] / 2, info[1] / 2 - 1],
+                "xlabel": "SPO",
+                "ylabel": "EYE Voltage (V)",
+                "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
+
+
+class jesd(eye_scan):
     """JESD Monitoring"""
 
     def __init__(self, address, username="root", password="analog"):

examples/ad9081_jesd_eye_diagram.py

@@ -0,0 +1,40 @@
+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 lane in 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"]
+
+    plt.subplot(int(num_lanes / 2), 2, int(lane) + 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.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)])