ad9088: Series of fixes to enable ADI_APOLLO_NCO_CHAN_SEL_DIRECT_GPIO

[gastmaier]

PR Description

• Add adi,gpio-quick-config to select the pre-set quick config profile
• Remap Profile 1 GPIO31 to GPIO15 since 31 is taken.
• Add fast boot template (commented-out).
• Do full chip hop config like the api examples, instead of per slice.

Resolved ai findings:

• Allow 0 in the dts as "apply no pre-set"
• Allow -1 to disable HOP for the FNCO path (there is no enable bit for the CNCO path)

Tested with

"""
Covers:
 - GPIO profile hopping on the TX datapath
     - FNCO GPIO hopping while CNCO is held in regmap mode
     - CNCO GPIO hopping while FNCO is held in regmap mode

 - GPIO profile hopping on the RX datapath
     - FNCO GPIO hopping while CNCO is held in regmap mode
     - CNCO GPIO hopping while FNCO is held in regmap mode

 - Mixed regmap + GPIO coexistence
     - Verifies one block can stay under SPI/regmap profile control while the
       other is under external GPIO control
     - Implicitly exercises the driver's global block_select sharing logic
       (ad9088_ffh_gpio_active(), enter/exit)

 - GPIO address decoding correctness
     - terminal (TX vs RX)
     - side (A vs B)
     - slice (FDUC/FDDC number for FNCO, CDUC/CDDC number for CNCO)
     - block (FNCO vs CNCO)
     - profile index bits
     - All derived from the driver-exposed channel label, so a wrong label/mapping
       would surface

 - Driver sysfs plumbing
     - ffh_{fnco,cnco}_index
     - ffh_{fnco,cnco}_frequency
     - ffh_{fnco,cnco}_mode
     - ffh_{fnco,cnco}_select
     - Writes must succeed without -EINVAL/-EFAULT

 - Device tree GPIO hop mapping
     - adi,gpio-hop-{terminal,slice,side,block,profile} contents

 - End-to-end signal integrity of the hop
     - Real RF/analog loopback TXn -> RXn
     - Actual captured baseband peak moves by the programmed profile delta

 - Profile table persistence across mode changes
     - Profiles are programmed once, then hopped repeatedly, so a HOP_CTRL_INIT
       pulse that wiped profiles would show up

Does not cover:

 - Trigger-based hop modes
     - TRIG_AUTO, TRIG_REGMAP, TRIG_GPIO

 - Auto-hop / increment-decrement behavior
     - The probe-time auto_mode, high_limit, low_limit, hop_ctrl_init settings are
       not validated

 - Hop latency / timing

 - Phase hopping

"""

from os import environ
import math
import struct
import subprocess
from time import sleep

import iio

url = environ.get("URL", "10.44.3.77")
RX_DEVICE = environ.get("RX_DEVICE", "axi-ad9084-rx-hpc")
TX_DEVICE = environ.get("TX_DEVICE", "axi-ad9084-tx-hpc")
CHAN = int(environ.get("CHAN", "1"))
SAMPLES = int(environ.get("SAMPLES", "65536"))
FFT_SAMPLES = int(environ.get("FFT_SAMPLES", "4096"))
DDS_FREQ = int(environ.get("DDS_FREQ", "50000000"))
TX_MAIN_NCO = int(environ.get("TX_MAIN_NCO", "1000000000"))
RX_MAIN_NCO = int(environ.get("RX_MAIN_NCO", "1000000000"))
DDS_SCALE_DB = float(environ.get("DDS_SCALE_DB", "-9"))
FREQ_TOL = float(environ.get("FREQ_TOL", "500000"))

# gpiochip0 line N -> AD9084 GPIO(N + 15)
BLOCK_FNCO = 0
BLOCK_CNCO = 1
TERMINAL = {"out": 0, "in": 1}
SIDE = {"A": 0, "B": 1}
DIRECT_GPIO = 3
DIRECT_REGMAP = 4
PROFILE_PINS = {
    0: "11=0 12=0 13=0 14=0 15=0",
    1: "11=1 12=0 13=0 14=0 15=0",
    2: "11=0 12=1 13=0 14=0 15=0",
    3: "11=1 12=1 13=0 14=0 15=0",
}


def _ssh(cmd):
    result = subprocess.run(["ssh", f"root@{url}", cmd], capture_output=True, text=True)
    print('$', cmd, '\n', result.stdout)
    if result.returncode:
        print(result.stderr)
        raise SystemExit(result.returncode)
    return result.stdout


def _validate_device_tree():
    base = "/sys/firmware/devicetree/base/fpga-axi@0/spi@a4a80000/ad9084@0"
    checks = {
        "adi\\,gpio-hop-terminal": "0000 0011 0000 0012 ",
        "adi\\,gpio-hop-slice": "0000 0013 0000 0014 0000 0015",
        "adi\\,gpio-hop-side": "0000 0016",
        "adi\\,gpio-hop-block": "0000 0017 0000 0018 0000 0019 0000 000f",
        "adi\\,gpio-hop-profile": "00000000: 0000 001a 0000 001b 0000 001c 0000 001d",
    }
    for prop, needle in checks.items():
        exports = _ssh(f"cat {base}/{prop} | xxd")
        assert needle in exports, exports


def _gpio_addr_from_label(iiodev, dir_, chan_, fnc_sel):
    label = _ssh(f"cd {iiodev}; cat {dir_}_voltage{chan_}_i_label").strip()
    side = SIDE[label.split(":", 1)[0].split("-")[1]]
    blocks = label.split(":", 1)[1].split("->")
    fine = int(blocks[0][4:])
    coarse = int(blocks[1][4:])

    block = BLOCK_FNCO if fnc_sel == "fnco" else BLOCK_CNCO
    slice_ = fine if fnc_sel == "fnco" else coarse
    terminal = TERMINAL[dir_]

    bits = {
        0: (block >> 3) & 1,       # profile_fcn_sel[3]
        1: 0,                      # unused, keep low
        2: terminal & 1,           # profile_tx_rxn[0]
        3: (terminal >> 1) & 1,    # profile_tx_rxn[1]
        4: slice_ & 1,             # profile_txrx_slice[0]
        5: (slice_ >> 1) & 1,      # profile_txrx_slice[1]
        6: (slice_ >> 2) & 1,      # profile_txrx_slice[2]
        7: side,                   # profile_txrx_BA
        8: block & 1,              # profile_fcn_sel[0]
        9: (block >> 1) & 1,       # profile_fcn_sel[1]
        10: (block >> 2) & 1,      # profile_fcn_sel[2]
    }
    addr = " ".join(f"{line}={val}" for line, val in bits.items())
    print(f"{dir_}_voltage{chan_} {fnc_sel}: {label} -> {addr}")
    return addr


def _set_gpio_profile(profile, iiodev, dir_, chan_, fnc_sel):
    pins = PROFILE_PINS[profile]
    addr = _gpio_addr_from_label(iiodev, dir_, chan_, fnc_sel)
    _ssh(f"pkill gpioset || : ; gpioset -zc /dev/gpiochip0 {addr} {pins}")


def _config_profile(iiodev, dir_, chan_, fnc_sel, freqs):
    chan = f"{dir_}_voltage{chan_}_i"
    for i, f in enumerate(freqs):
        _ssh(f"cd {iiodev}; echo {i} > {chan}_ffh_{fnc_sel}_index")
        _ssh(f"cd {iiodev}; echo {f} > {chan}_ffh_{fnc_sel}_frequency")


def _set_ffh_mode(iiodev, dir_, chan_, fnc_sel, mode):
    _ssh(f"cd {iiodev}; echo {mode} > {dir_}_voltage{chan_}_i_ffh_{fnc_sel}_mode")


def _set_regmap_profile(iiodev, dir_, chan_, fnc_sel, profile):
    _ssh(f"cd {iiodev}; echo {profile} > {dir_}_voltage{chan_}_i_ffh_{fnc_sel}_select")


def _dbfs_to_linear(dbfs):
    return 0.0 if dbfs <= -91 else 10 ** (dbfs / 20.0)


def _write_attr(attrs, name, value):
    if name in attrs:
        attrs[name].value = str(value)
        return True
    return False


def _find_dev(ctx, name):
    dev = ctx.find_device(name)
    if dev is None:
        available = ", ".join(d.name or d.id for d in ctx.devices)
        raise RuntimeError(f"Could not find IIO device {name!r}; available: {available}")
    return dev


def _configure_tx_tone(tx, chan_):
    scale = _dbfs_to_linear(DDS_SCALE_DB)
    tone_base = chan_ * 4
    tone_cfg = [
        (tone_base + 0, DDS_FREQ, scale, 0),
        (tone_base + 1, DDS_FREQ, 0, 0),
        (tone_base + 2, DDS_FREQ, scale, 270000),
        (tone_base + 3, DDS_FREQ, 0, 0),
    ]
    for tone, freq, tone_scale, phase in tone_cfg:
        chn = tx.find_channel(f"altvoltage{tone}", True)
        if not chn:
            raise RuntimeError(f"Missing TX DDS altvoltage{tone}")
        _write_attr(chn.attrs, "raw", 1)
        _write_attr(chn.attrs, "frequency", freq)
        _write_attr(chn.attrs, "scale", tone_scale)
        _write_attr(chn.attrs, "phase", phase)


def _write_channel_attr(dev, chan_name, attr, value, is_output):
    chn = dev.find_channel(chan_name, is_output)
    chn.attrs[attr].value = str(value)
    print(f"{'out' if is_output else 'in'}_{chan_name}_{attr} = {chn.attrs[attr].value}")


def _configure_ncos(ad9084_dev, tx_dac, chan_):
    _write_channel_attr(ad9084_dev, f"voltage{chan_}_i", "main_nco_frequency", TX_MAIN_NCO, True)
    _write_channel_attr(ad9084_dev, f"voltage{chan_}_i", "main_nco_frequency", RX_MAIN_NCO, False)
    _configure_tx_tone(tx_dac, chan_)


def _sample_rate(dev, channels):
    for chn in channels:
        for attr in ("sampling_frequency", "sampling_frequency_available"):
            if attr in chn.attrs:
                return float(chn.attrs[attr].value.split()[0])
    if "sampling_frequency" in dev.attrs:
        return float(dev.attrs["sampling_frequency"].value.split()[0])
    return float(environ.get("SAMPLE_RATE", "500000000"))


def _complex_samples(i_bytes, q_bytes):
    count = min(len(i_bytes), len(q_bytes)) // 2
    i_vals = struct.unpack_from("<" + "h" * count, i_bytes)
    q_vals = struct.unpack_from("<" + "h" * count, q_bytes)
    return [complex(i_vals[n], q_vals[n]) for n in range(count)]


def _peak_frequency(samples, sample_rate):
    n = min(len(samples), FFT_SAMPLES)
    data = samples[:n]
    best_bin = 0
    best_mag = -1.0
    for k in range(n):
        re = im = 0.0
        for m, x in enumerate(data):
            ang = -2.0 * math.pi * k * m / n
            c = math.cos(ang)
            s = math.sin(ang)
            re += x.real * c - x.imag * s
            im += x.real * s + x.imag * c
        mag = re * re + im * im
        if mag > best_mag:
            best_bin, best_mag = k, mag
    freq = best_bin * sample_rate / n
    if best_bin > n // 2:
        freq -= sample_rate
    return freq, math.sqrt(best_mag) / n


def _capture_pair(rx, rx_idx):
    i_ch = rx.find_channel(f"voltage{rx_idx}_i") or rx.find_channel(f"voltage{rx_idx}")
    q_ch = rx.find_channel(f"voltage{rx_idx}_q")
    if not i_ch or not q_ch:
        raise RuntimeError(f"Missing RX I/Q scan channels for voltage{rx_idx}")

    mask = iio.ChannelsMask(rx)
    mask.channels = [i_ch, q_ch]
    buf = rx.get_buffer(0)
    if buf is None:
        raise RuntimeError(f"Device {rx.name or rx.id} has no RX buffer")

    stream = iio.Stream(buf, mask, SAMPLES, 4)
    block = next(stream)
    try:
        i_data = i_ch.read(block, raw=True)
        q_data = q_ch.read(block, raw=True)
        return _complex_samples(i_data, q_data), _sample_rate(rx, [i_ch, q_ch])
    finally:
        stream.cancel()


def _capture_peak(rx, chan_):
    sleep(0.1)
    samples, fs = _capture_pair(rx, chan_)
    freq, peak = _peak_frequency(samples, fs)
    print(f"capture: peak={peak:.3f} freq={freq:.3f} Hz")
    return freq


def _assert_delta(measured, baseline, expected_delta, tolerance=FREQ_TOL):
    delta = measured - baseline
    diff = abs(abs(delta) - abs(expected_delta))
    assert diff <= tolerance, f"delta {delta} != {expected_delta} (diff {diff})"


def _pin_regmap(iiodev, dir_, chan_, profile=0):
    """Hold both blocks of @dir_ on a known regmap profile."""
    for fnc in ("fnco", "cnco"):
        _set_ffh_mode(iiodev, dir_, chan_, fnc, DIRECT_REGMAP)
        _set_regmap_profile(iiodev, dir_, chan_, fnc, profile)


def _test_gpio_sweep(rx, iiodev, dir_, chan_, gpio_fnc, regmap_fnc, gpio_freqs, regmap_freqs):
    print(f"=== Test {dir_.upper()} {gpio_fnc.upper()} GPIO + {regmap_fnc.upper()} regmap ===")
    # Keep the opposite terminal from hopping while this one is swept.
    _pin_regmap(iiodev, "in" if dir_ == "out" else "out", chan_)
    _set_ffh_mode(iiodev, dir_, chan_, regmap_fnc, DIRECT_REGMAP)
    _set_regmap_profile(iiodev, dir_, chan_, regmap_fnc, 0)
    _set_ffh_mode(iiodev, dir_, chan_, gpio_fnc, DIRECT_GPIO)
    _set_gpio_profile(0, iiodev, dir_, chan_, gpio_fnc)
    baseline = _capture_peak(rx, CHAN)
    for i, f in enumerate(gpio_freqs):
        _set_gpio_profile(i, iiodev, dir_, chan_, gpio_fnc)
        measured = _capture_peak(rx, CHAN)
        _assert_delta(measured, baseline, f - gpio_freqs[0])
    _set_gpio_profile(0, iiodev, dir_, chan_, gpio_fnc)


def hop_gpio_single_rx_tx():
    _validate_device_tree()
    iiodev = _ssh("grep -rw /sys/bus/iio/devices/*/name -e axi-ad9084-rx-hpc -l | xargs dirname").strip()

    ctx = iio.Context(f"ip:{url}")
    ctx.set_timeout(int(environ.get("IIO_TIMEOUT_MS", "10000")))
    ad9084 = _find_dev(ctx, RX_DEVICE)       # AD9084 RX/TX controls live here
    tx_dac = _find_dev(ctx, TX_DEVICE)       # AXI-DAC DDS source

    _configure_ncos(ad9084, tx_dac, CHAN)

    tx_fnco = [500000000, 510000000, 520000000]
    tx_cnco = [480000000, 500000000, 520000000]
    rx_fnco = [100000000, 110000000, 120000000]
    rx_cnco = [1000000000, 1010000000, 1020000000]

    for dir_, fnco, cnco in (("out", tx_fnco, tx_cnco), ("in", rx_fnco, rx_cnco)):
        _set_gpio_profile(0, iiodev, dir_, CHAN, "fnco")
        _set_gpio_profile(0, iiodev, dir_, CHAN, "cnco")
        _config_profile(iiodev, dir_, CHAN, "fnco", fnco)
        _config_profile(iiodev, dir_, CHAN, "cnco", cnco)

    # TX path GPIO hopping, observed through RX buffer over TXn->RXn loopback.
    _test_gpio_sweep(ad9084, iiodev, "out", CHAN, "fnco", "cnco", tx_fnco, tx_cnco)
    _test_gpio_sweep(ad9084, iiodev, "out", CHAN, "cnco", "fnco", tx_cnco, tx_fnco)

    # RX path GPIO hopping, driven by the same TX DDS/AD9084 source.
    _test_gpio_sweep(ad9084, iiodev, "in", CHAN, "fnco", "cnco", rx_fnco, rx_cnco)
    _test_gpio_sweep(ad9084, iiodev, "in", CHAN, "cnco", "fnco", rx_cnco, rx_fnco)


if __name__ == "__main__":
    hop_gpio_single_rx_tx()

PR Type

• Bug fix (a change that fixes an issue)
• New feature (a change that adds new functionality)
• Breaking change (a change that affects other repos or cause CIs to fail)

PR Checklist

• I have conducted a self-review of my own code changes
• I have compiled my changes, including the documentation
• I have tested the changes on the relevant hardware
• I have updated the documentation outside this repo accordingly
• I have provided links for the relevant upstream lore

[nunojsa]

The above handling is very annoying 😄. Not related to the current PR and just something I'm sharing if you ever want to change it. In navassa, I added this exactly to avoid handling like the above.

[nunojsa]

nowadays you can do for (u8 t = 0; ...) in the kernel