typing and defaults

README.md

@@ -1,4 +1,8 @@
 # Charon VNA
+<!-- ![PyPi Downloads](https://img.shields.io/pypi/dm/charon-vna) -->
+<!-- ![Last Commit](https://img.shields.io/gitea/last-commit/brendanhaines/charon-vna?gitea_url=https%3A%2F%2Fgit.brendanhaines.com) -->
+<!-- ![Workflow Status](https://git.brendanhaines.com/brendanhaines/charon-vna/actions/workflows/python_publish.yml/badge.svg) -->
+
 
 Named after [Pluto's moon](https://en.wikipedia.org/wiki/Charon_(moon)), Charon uses the [ADI Pluto SDR](https://www.analog.com/en/resources/evaluation-hardware-and-software/evaluation-boards-kits/adalm-pluto.html) as a vector network analyzer. The basic usage is as a 1 port VNA but this can be extended to arbitrarily many ports with the addition of a couple RF switches.
 
@@ -7,7 +11,8 @@ Named after [Pluto's moon](https://en.wikipedia.org/wiki/Charon_(moon)), Charon
 1. Install LibIIO. This is a dependency of [PyADI-IIO](https://wiki.analog.com/resources/tools-software/linux-software/pyadi-iio).
 On Ubuntu 22.04 just run `sudo apt-get install -y libiio-dev`
 
-2. `pip install charon-vna`
+2. Charon releases are published on [PyPi](https://pypi.org/project/charon-vna/). Install using pip:
+`pip install charon-vna`
 
 ## Hardware Setup
 
@@ -61,4 +66,9 @@ Note that unlike the main calibration, power calibration frequencies do not need
 #### Pluto Default Connection Settings
 user: `root`
 password: `analog`
-ip: `192.168.2.1`
+ip: `192.168.2.1`
+
+## Alternatives
+- [NanoVNA](https://nanovna.com/). 2-ports. 50 kHz - 2.7 GHz. Degraded performance above 1.5 GHz. S11 and S21 only.
+- [pluto-network-analyzer](https://github.com/fromconcepttocircuit/pluto-network-analyzer). 2-ports. 100 MHz - 3 GHz. S11 and S21 only. Uses a [wideband RF bridge](https://www.60dbm.com/product/rf-bridge-1-3000-mhz/) instead of a coupler
+- [LibreVNA](https://github.com/jankae/LibreVNA). 2-ports. 100 KHz - 6 GHz. I've never used this but it is almost certainly faster than Charon. Not sure how the performance compares. $700 on [AliExpress](https://www.aliexpress.us/item/3256802242049773.html?gatewayAdapt=glo2usa4itemAdapt)

charon_vna/config_default.py

@@ -3,14 +3,14 @@ import subprocess
 
 import numpy as np
 
-from charon_vna.gui import DEFAULT_CONFIG
+from charon_vna.gui import PATH_CONFIG_DEFAULT
 
 config = dict(
     frequency=np.linspace(80e6, 500e6, 500).tolist(),
     power=-5,
 )
 
-with open(DEFAULT_CONFIG, "w") as f:
+with open(PATH_CONFIG_DEFAULT, "w") as f:
     json.dump(config, f)
 
 # autoformat
@@ -19,7 +19,7 @@ subprocess.run(
         "python",
         "-m",
         "json.tool",
-        DEFAULT_CONFIG.resolve().as_posix(),
+        PATH_CONFIG_DEFAULT.resolve().as_posix(),
-        DEFAULT_CONFIG.resolve().as_posix(),
+        PATH_CONFIG_DEFAULT.resolve().as_posix(),
     ]
 )

charon_vna/gui.py

@@ -3,6 +3,7 @@ import json
 import pickle
 import re
 import sys
+import webbrowser
 from pathlib import Path
 from typing import List
 
@@ -14,23 +15,23 @@ from numpy import typing as npt
 from PySide6.QtGui import QAction, QKeySequence
 from PySide6.QtWidgets import (
     QApplication,
-    QDialogButtonBox,
     QFileDialog,
     QInputDialog,
     QLineEdit,
     QMainWindow,
     QMenu,
     QProgressBar,
+    QTabWidget,
     QVBoxLayout,
     QWidget,
 )
-from vna import Charon
 
 from charon_vna.plots import PlotWidget
 from charon_vna.util import net2s, s2net
+from charon_vna.vna import Charon
 
 # %%
-DEFAULT_CONFIG = Path(__file__).parent / "config_default.json"
+PATH_CONFIG_DEFAULT = Path(__file__).parent / "config_default.json"
 CONFIG_SUFFIX = ".json"
 
 
@@ -43,7 +44,7 @@ class MainWindow(QMainWindow):
     def __init__(self, ip: str | None = None):
         super().__init__()
 
-        self.config_path = DEFAULT_CONFIG
+        self.config_path = PATH_CONFIG_DEFAULT
         with open(self.config_path, "r") as f:
             config = json.load(f)
         self._frequency = config["frequency"]
@@ -96,6 +97,19 @@ class MainWindow(QMainWindow):
         action_cal_solt.triggered.connect(self.calibrate_solt)
         menu_calibration.addAction(action_cal_solt)
 
+        menu_help = QMenu("&Help")
+        menubar.addMenu(menu_help)
+        action_open_homepage = QAction("&Documentation", self)
+        action_open_homepage.triggered.connect(
+            lambda: webbrowser.open("https://git.brendanhaines.com/brendanhaines/charon_vna")
+        )
+        menu_help.addAction(action_open_homepage)
+        action_report_issue = QAction("&Report an Issue", self)
+        action_report_issue.triggered.connect(
+            lambda: webbrowser.open("https://git.brendanhaines.com/brendanhaines/charon_vna/issues")
+        )
+        menu_help.addAction(action_report_issue)
+
         # Content
         window_layout = QVBoxLayout()
 
@@ -104,19 +118,20 @@ class MainWindow(QMainWindow):
         prog_sweep.setMaximum(100)
         prog_sweep.setFormat("%v / %m")
         # prog_sweep.setTextVisible(False)
-        prog_sweep.setValue(50)
+        prog_sweep.setValue(0)
         window_layout.addWidget(prog_sweep)
         self.prog_sweep = prog_sweep
 
-        # window_widget.se
+        plot_widget = QTabWidget()
-        plot_layout = QVBoxLayout()
-        # TODO: handle plots properly
         self.plots = []
-        for type_ in ["logmag", "phase", "vswr", "smith"]:
+        for type_ in [
+            "logmag",
+            "phase",
+            "vswr",
+            "smith",
+        ]:
             self.plots.append(PlotWidget(type_=type_))
-            plot_layout.addWidget(self.plots[-1])
+            plot_widget.addTab(self.plots[-1], type_)
-        plot_widget = QWidget()
-        plot_widget.setLayout(plot_layout)
         window_layout.addWidget(plot_widget)
 
         # Set the central widget of the Window.
@@ -127,6 +142,7 @@ class MainWindow(QMainWindow):
     def saveas_config(self) -> None:
         print("Prompting for save path...")
         dialog = QFileDialog(self)
+        dialog.setNameFilter(f"*{CONFIG_SUFFIX}")
         dialog.setDefaultSuffix(CONFIG_SUFFIX)
         dialog.setAcceptMode(QFileDialog.AcceptMode.AcceptSave)
         if dialog.exec():
@@ -135,8 +151,8 @@ class MainWindow(QMainWindow):
                 raise ValueError(
                     f"{config_path.name} is not a valid configuration file. Must have extension {CONFIG_SUFFIX}"
                 )
-            if config_path == DEFAULT_CONFIG:
+            if config_path == PATH_CONFIG_DEFAULT:
-                raise ValueError(f"Cannot overwrite default configuration file at {DEFAULT_CONFIG}")
+                raise ValueError(f"Cannot overwrite default configuration file at {PATH_CONFIG_DEFAULT}")
             self.config_path = config_path
             print(f"Config path is now {self.config_path.resolve()}")
 
@@ -160,7 +176,7 @@ class MainWindow(QMainWindow):
             self.load_config(self.config_path)
 
     def save_config(self) -> None:
-        if self.config_path == DEFAULT_CONFIG:
+        if self.config_path == PATH_CONFIG_DEFAULT:
             self.saveas_config()
         else:
             print(f"Saving config to {self.config_path.resolve()}")

charon_vna/io_.py

@@ -1,11 +1,94 @@
+import json
+import shutil
+import zipfile
 from pathlib import Path
+from tempfile import TemporaryDirectory
+from typing import List
 
 import skrf as rf
-from util import net2s
 
 
-# scikit-rf has no way to save files aside from touchstone and pickle
+# scikit-rf has no way to save Calibration objects aside from pickle
-def cal2zarr(cal: rf.calibration.Calibration, outpath: Path):
+def cal2zip(cal: rf.calibration.Calibration, path: Path | str) -> None:
-    ideals = [net2s(net) for net in cal.ideals]
+    path = Path(path)
-    measured = [net2s(net) for net in cal.measured]
+    cal_type = cal.__class__
-    # s.to_zarr(outpath)
+    measured: List[rf.network.Network] = cal.measured
+    ideals: List[rf.network.Network] = cal.ideals
+
+    if cal_type not in [rf.calibration.OnePort]:
+        raise NotImplementedError(f"Calibration {cal_type.__name__} serialization not implemented")
+
+    assert len(ideals) == len(measured)  # this should have already been asserted when cal was instantiated
+
+    with TemporaryDirectory() as temp:
+        dir_temp = Path(temp)
+
+        with zipfile.ZipFile(dir_temp / "archive.zip", "w") as archive:
+            # create a configuration file
+            filename_config = dir_temp / "config.json"
+            with open(filename_config, "w") as f:
+                json.dump(
+                    dict(
+                        cal_type=cal_type.__name__,
+                        num_standards=len(measured),
+                    ),
+                    f,
+                )
+            archive.write(filename_config, str(filename_config.relative_to(dir_temp)))
+
+            # add standard data
+            dir_ideals = dir_temp / "ideals"
+            dir_ideals.mkdir()
+            for ii, ideal in enumerate(ideals):
+                filename = dir_ideals / f"{ii}.s2p"
+                ideal.write_touchstone(filename)
+                archive.write(filename, str(filename.relative_to(dir_temp)))
+
+            # add test data
+            dir_measured = dir_temp / "measured"
+            dir_measured.mkdir()
+            for ii, meas in enumerate(measured):
+                filename = dir_measured / f"{ii}.s2p"
+                meas.write_touchstone(filename)
+                archive.write(filename, str(filename.relative_to(dir_temp)))
+
+            print("Wrote calibration to file")
+            archive.printdir()
+
+        shutil.move(dir_temp / "archive.zip", path)
+
+
+def zip2cal(path: Path | str):
+    path = Path(path)
+    if not path.exists():
+        raise FileNotFoundError(f"Calibration file {path} does not exist")
+
+    with zipfile.ZipFile(path) as archive:
+        archive.printdir()
+
+        config = json.loads(archive.read("config.json"))
+        print(config)
+
+        ideals = list()
+        measured = list()
+
+        with TemporaryDirectory() as temp:
+            dir_temp = Path(temp)
+
+            for ii in range(config["num_standards"]):
+                with open(dir_temp / f"{ii}.s2p", "wb") as f:
+                    f.write(archive.read(f"ideals/{ii}.s2p"))
+                ideals.append(rf.network.Network(dir_temp / f"{ii}.s2p"))
+
+                with open(dir_temp / f"{ii}.s2p", "wb") as f:
+                    f.write(archive.read(f"measured/{ii}.s2p"))
+                measured.append(rf.network.Network(dir_temp / f"{ii}.s2p"))
+
+    cal_type = config["cal_type"]
+    CalClass = getattr(rf.calibration, cal_type)
+    if not issubclass(CalClass, rf.calibration.Calibration):
+        raise ValueError()
+
+    calibration = CalClass(measured=measured, ideals=ideals)
+
+    return calibration

charon_vna/vna.py

@@ -1,11 +1,11 @@
 # %% imports
 import copy
+from enum import IntEnum, unique
 from pathlib import Path
-from typing import Any, Callable, Dict, Tuple
+from typing import Any, Callable, Dict, Literal, Tuple
 
 import adi
-
+import iio
-# import iio
 import numpy as np
 import skrf as rf
 import xarray as xr
@@ -34,14 +34,61 @@ def generate_tone(f: float, fs: float, N: int = 1024, scale: int = 2**14):
     return iq
 
 
+@unique
+class AD9361Register(IntEnum):
+    AUXDAC1_WORD = 0x018
+    AUXDAC2_WORD = 0x019
+    AUXDAC1_CONFIG = 0x01A
+    AUXDAC2_CONFIG = 0x01B
+    AUXADC_CLOCK_DIVIDER = 0x01C
+    AUXADC_CONFIG = 0x01D
+    AUXADC_WORD_MSB = 0x01E
+    AUXADC_WORD_LSB = 0x01F
+    AUTO_GPIO = 0x020
+    AGC_GAIN_LOCK_DELAY = 0x021
+    AGC_ATTACK_DELAY = 0x022
+    AUXDAC_ENABLE_CONTROL = 0x023
+    RX_LOAD_SYNTH_DELAY = 0x024
+    TX_LOAD_SYNTH_DELAY = 0x025
+    EXTERNAL_LNA_CONTROL = 0x026
+    GPO_FORCE_AND_INIT = 0x027
+    GPO0_RX_DELAY = 0x028
+    GPO1_RX_DELAY = 0x029
+    GPO2_RX_DELAY = 0x02A
+    GPO3_RX_DELAY = 0x02B
+    GPO0_TX_DELAY = 0x02C
+    GPO1_TX_DELAY = 0x02D
+    GPO2_TX_DELAY = 0x02E
+    GPO3_TX_DELAY = 0x02F
+    AUXDAC1_RX_DELAY = 0x030
+    AUXDAC1_TX_DELAY = 0x031
+    AUXDAC2_RX_DELAY = 0x032
+    AUXDAC2_TX_DELAY = 0x033
+
+
+@unique
+class AD9361DacVref(IntEnum):
+    VREF_1V0 = 0b00
+    VREF_1V5 = 0b01
+    VREF_2V0 = 0b10
+    VREF_2V5 = 0b11
+
+
+@unique
+class AD9361DacStepFactor(IntEnum):
+    FACTOR_2 = 0b0
+    FACTOR_1 = 0b1
+
+
 class Charon:
     FREQUENCY_OFFSET = 1e6
+    DEFAULT_IP = "192.168.2.1"
 
     calibration: rf.calibration.Calibration | None = None
 
     def __init__(
         self,
-        ip: str = "192.168.2.1",
+        ip: str = DEFAULT_IP,
         frequency: npt.ArrayLike = np.linspace(1e9, 2e9, 3),
         ports: Tuple[int] = (1,),
     ):
@@ -49,7 +96,8 @@ class Charon:
         self.frequency = frequency
 
         # everything RF
-        self.sdr = adi.ad9361(uri=f"ip:{ip}")
+        uri = f"ip:{ip}"
+        self.sdr = adi.ad9361(uri=uri)
         for attr, expected in [
             ("adi,2rx-2tx-mode-enable", True),
             # ("adi,gpo-manual-mode-enable", True),
@@ -81,14 +129,16 @@ class Charon:
         self.sdr.tx_hardwaregain_chan0 = -10
 
         # # switch control
-        # ctx = iio.Context(uri)
+        ctx = iio.Context(uri)
-        # self.ctrl = ctx.find_device("ad9361-phy")
+        self.ctrl = ctx.find_device("ad9361-phy")
-        # # raw ad9361 register accesss:
+        # raw ad9361 register accesss:
-        # # https://ez.analog.com/linux-software-drivers/f/q-a/120853/control-fmcomms3-s-gpo-with-python
+        # https://ez.analog.com/linux-software-drivers/f/q-a/120853/control-fmcomms3-s-gpo-with-python
-        # # https://www.analog.com/media/cn/technical-documentation/user-guides/ad9364_register_map_reference_manual_ug-672.pdf  # noqa: E501
+        # https://www.analog.com/media/cn/technical-documentation/user-guides/ad9364_register_map_reference_manual_ug-672.pdf  # noqa: E501
-        # self.ctrl.reg_write(0x26, 0x90)  # bit 7: AuxDAC Manual, bit 4: GPO Manual
+        self.ctrl.reg_write(AD9361Register.EXTERNAL_LNA_CONTROL, 0x90)  # bit 7: AuxDAC Manual, bit 4: GPO Manual
-        # self._set_gpo(self.ports[0] - 1)
+        self.ctrl.reg_write(AD9361Register.AUXDAC_ENABLE_CONTROL, 0x3F)
-        # # TODO: init AuxDAC
+        self._set_gpo(0b0000)
+        self._set_dac_code(value=0, channel=1)
+        self._set_dac_code(value=0, channel=2)
 
     def get_config(self) -> Dict[str, Any]:
         config = dict()
@@ -113,28 +163,74 @@ class Charon:
         return config
 
     def _get_gpo(self) -> int:
-        return (self.ctrl.reg_read(0x27) >> 4) & 0x0F
+        return (self.ctrl.reg_read(AD9361Register.GPO_FORCE_AND_INIT) >> 4) & 0x0F
 
     def _set_gpo(self, value: int) -> None:
-        self.ctrl.reg_write(0x27, (value & 0x0F) << 4)  # bits 7-4: GPO3-0
+        self.ctrl.reg_write(AD9361Register.GPO_FORCE_AND_INIT, (value & 0x0F) << 4)  # bits 7-4: GPO3-0
+
+    def _set_dac_voltage(self, voltage: float, channel: Literal[1, 2]):
+        raise NotImplementedError()
+
+    def _set_dac_code(
+        self,
+        value: int,
+        channel: Literal[1, 2],
+        vref: AD9361DacVref = AD9361DacVref.VREF_1V0,
+        step_factor: AD9361DacStepFactor = AD9361DacStepFactor.FACTOR_2,
+    ) -> None:
+        if channel not in [1, 2]:
+            raise ValueError(f"Invalid channel {channel}. Must be 1 or 2")
+
+        if value > 0x3FF or value < 0:
+            raise ValueError("Invalid value for 10 bit DAC. Must be between 0 and 0x3FF (inclusive)")
+
+        vref = AD9361DacVref(vref)
+        step_factor = AD9361DacStepFactor(step_factor)
+
+        # https://www.analog.com/media/cn/technical-documentation/user-guides/ad9364_register_map_reference_manual_ug-672.pdf
+        # page 13
+        # vout = 0.97 * vref + (0.000738 + 9e-6 * (vref * 1.6 - 2)) * auxdac_word[9:0] * step_factor - 0.3572 * step_factor + 0.05
+        # vout ~= (vref - 0.3572 * step_factor) + 0.000738 * auxdac_word[9:0] * step_factor
+        # which gives a 1.5V swing with step_factor == 2 and 0.75V swing with step_factor == 1
+        # vref basically just changes the minimum voltage with negligible impact on output scaling
+
+        self.ctrl.reg_write(
+            AD9361Register.__getitem__(f"AUXDAC{channel}_WORD"),
+            (value >> 2) & 0xFF,
+        )
+        self.ctrl.reg_write(
+            AD9361Register.__getitem__(f"AUXDAC{channel}_CONFIG"),
+            (value & 0x3) | (vref.value << 2) | (step_factor << 4),
+        )
+
+    def _get_dac_code(self, channel: Literal[1, 2]) -> Tuple[float, AD9361DacVref, AD9361DacStepFactor]:
+        word = self.ctrl.reg_read(AD9361Register.__getitem__(f"AUXDAC{channel}_WORD"))
+        config = self.ctrl.reg_read(AD9361Register.__getitem__(f"AUXDAC{channel}_CONFIG"))
+
+        value = (word << 2) + (config & 0x3)
+        vref = AD9361DacVref((config >> 2) & 0x3)
+        step_factor = AD9361DacStepFactor((config >> 4) & 0x1)
+
+        return (value, vref, step_factor)
+
+    def _get_dac_voltage(self) -> float:
+        raise NotImplementedError()
 
     def set_output_power(self, power: float):
-        # FIXME: this is a hack because I don't want to go through re-calibration
+        pout = xr.DataArray(
-        if power == 5:
+            [-15, -10, -5, 0, 5],
-            tx_gain = -1
+            dims=["tx_gain"],
-        elif power == 0:
+            coords=dict(
-            tx_gain = -7
+                # TODO: correct over frequency
-        elif power == -5:
+                frequency=1e9,  # FIXME: I'm not sure at what frequency I generated this table
-            tx_gain = -12
+                tx_channel=0,
-        elif power == -10:
+                tx_gain=[-22, -17, -12, -7, -1],
-            tx_gain = -17
+            ),
-        elif power == -15:
+        )
-            tx_gain = -22
+
-        else:
+        tx_gain_idx = np.abs(pout - power).argmin(dim="tx_gain")
-            raise NotImplementedError()
+        tx_gain = pout.coords["tx_gain"][tx_gain_idx]
-            # # TODO: correct over frequency
+
-            # tx_gain_idx = np.abs(pout.sel(tx_channel=0) - power).argmin(dim="tx_gain")
-            # tx_gain = pout.coords["tx_gain"][tx_gain_idx]
         self.sdr.tx_hardwaregain_chan0 = float(tx_gain)
 
     def set_output(self, frequency: float, power: float):
@@ -220,6 +316,7 @@ class Charon:
         )
         for ff, freq in enumerate(s.frequency.data):
             if callback is not None:
+                # report progress during sweep
                 callback(ff, len(s.frequency))
             self.set_output(frequency=freq, power=-5)
             self.sdr.rx_destroy_buffer()
@@ -231,7 +328,9 @@ class Charon:
             self.sdr.rx_hardwaregain_chan1 = 40
             rx = self.sdr.rx()
             s.loc[dict(frequency=freq)] = np.mean(rx[1] / rx[0])
+
         if callback is not None:
+            # mark capture as complete
             callback(len(s.frequency), len(s.frequency))
 
         return s