inital commit

.gitignore

@@ -0,0 +1 @@
+__pycache__

Calibration.py

@@ -0,0 +1,385 @@
+#  NanoVNASaver
+#
+#  A python program to view and export Touchstone data from a NanoVNA
+#  Copyright (C) 2019, 2020  Rune B. Broberg
+#  Copyright (C) 2020 NanoVNA-Saver Authors
+#
+#  This program is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU General Public License as published by
+#  the Free Software Foundation, either version 3 of the License, or
+#  (at your option) any later version.
+#
+#  This program is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU General Public License for more details.
+#
+#  You should have received a copy of the GNU General Public License
+#  along with this program.  If not, see <https://www.gnu.org/licenses/>.
+import logging
+import cmath
+import math
+import os
+import re
+from collections import defaultdict, UserDict
+from typing import List
+
+from scipy.interpolate import interp1d
+
+from RFTools import Datapoint
+
+RXP_CAL_LINE = re.compile(r"""^\s*
+    (?P<freq>\d+) \s+
+    (?P<shortr>[-0-9Ee.]+) \s+ (?P<shorti>[-0-9Ee.]+) \s+
+    (?P<openr>[-0-9Ee.]+) \s+ (?P<openi>[-0-9Ee.]+) \s+
+    (?P<loadr>[-0-9Ee.]+) \s+ (?P<loadi>[-0-9Ee.]+)(?: \s
+    (?P<throughr>[-0-9Ee.]+) \s+ (?P<throughi>[-0-9Ee.]+) \s+
+    (?P<isolationr>[-0-9Ee.]+) \s+ (?P<isolationi>[-0-9Ee.]+)
+    )?
+""", re.VERBOSE)
+
+logger = logging.getLogger(__name__)
+
+
+def correct_delay(d: Datapoint, delay: float, reflect: bool = False):
+    mult = 2 if reflect else 1
+    corr_data = d.z * cmath.exp(
+        complex(0, 1) * 2 * math.pi * d.freq * delay * -1 * mult)
+    return Datapoint(d.freq, corr_data.real, corr_data.imag)
+
+
+class CalData(UserDict):
+    def __init__(self):
+        data = {
+            "short": None,
+            "open": None,
+            "load": None,
+            "through": None,
+            "isolation": None,
+            # the frequence
+            "freq": 0,
+            # 1 Port
+            "e00": 0.0,  # Directivity
+            "e11": 0.0,  # Port match
+            "delta_e": 0.0,  # Tracking
+            # 2 port
+            "e30": 0.0,  # Port match
+            "e10e32": 0.0,  # Transmission
+        }
+        super().__init__(data)
+
+    def __str__(self):
+        d = self.data
+        s = (f'{d["freq"]}'
+             f' {d["short"].re} {d["short"].im}'
+             f' {d["open"].re} {d["open"].im}'
+             f' {d["load"].re} {d["load"].im}')
+        if d["through"] is not None:
+            s += (f' {d["through"].re} {d["through"].im}'
+                  f' {d["isolation"].re} {d["isolation"].im}')
+        return s
+
+
+class CalDataSet:
+    def __init__(self):
+        self.data = defaultdict(CalData)
+
+    def insert(self, name: str, dp: Datapoint):
+        if name not in self.data[dp.freq]:
+            raise KeyError(name)
+        self.data[dp.freq]["freq"] = dp.freq
+        self.data[dp.freq][name] = dp
+
+    def frequencies(self) -> List[int]:
+        return sorted(self.data.keys())
+
+    def get(self, freq: int) -> CalData:
+        return self.data[freq]
+
+    def items(self):
+        for item in self.data.items():
+            yield item
+
+    def values(self):
+        for freq in self.frequencies():
+            yield self.get(freq)
+
+    def size_of(self, name: str) -> int:
+        return len([v for v in self.data.values() if v[name] is not None])
+
+    def complete1port(self) -> bool:
+        for val in self.data.values():
+            for name in ("short", "open", "load"):
+                if val[name] is None:
+                    return False
+        return any(self.data)
+
+    def complete2port(self) -> bool:
+        for val in self.data.values():
+            for name in ("short", "open", "load", "through", "isolation"):
+                if val[name] is None:
+                    return False
+        return any(self.data)
+
+
+class Calibration:
+    CAL_NAMES = ("short", "open", "load", "through", "isolation",)
+    IDEAL_SHORT = complex(-1, 0)
+    IDEAL_OPEN = complex(1, 0)
+    IDEAL_LOAD = complex(0, 0)
+
+    def __init__(self):
+
+        self.notes = []
+        self.dataset = CalDataSet()
+        self.interp = {}
+
+        self.useIdealShort = True
+        self.shortL0 = 5.7 * 10E-12
+        self.shortL1 = -8960 * 10E-24
+        self.shortL2 = -1100 * 10E-33
+        self.shortL3 = -41200 * 10E-42
+        self.shortLength = -34.2  # Picoseconfrequenciesds
+        # These numbers look very large, considering what Keysight
+        # suggests their numbers are.
+
+        self.useIdealOpen = True
+        # Subtract 50fF for the nanoVNA calibration if nanoVNA is
+        # calibrated?
+        self.openC0 = 2.1 * 10E-14
+        self.openC1 = 5.67 * 10E-23
+        self.openC2 = -2.39 * 10E-31
+        self.openC3 = 2.0 * 10E-40
+        self.openLength = 0
+
+        self.useIdealLoad = True
+        self.loadR = 25
+        self.loadL = 0
+        self.loadC = 0
+        self.loadLength = 0
+
+        self.useIdealThrough = True
+        self.throughLength = 0
+
+        self.isCalculated = False
+
+        self.source = "Manual"
+
+    def insert(self, name: str, data: List[Datapoint]):
+        for dp in data:
+            self.dataset.insert(name, dp)
+
+    def size(self) -> int:
+        return len(self.dataset.frequencies())
+
+    def data_size(self, name) -> int:
+        return self.dataset.size_of(name)
+
+    def isValid1Port(self) -> bool:
+        return self.dataset.complete1port()
+
+    def isValid2Port(self) -> bool:
+        return self.dataset.complete2port()
+
+    def calc_corrections(self):
+        if not self.isValid1Port():
+            logger.warning(
+                "Tried to calibrate from insufficient data.")
+            raise ValueError(
+                "All of short, open and load calibration steps"
+                "must be completed for calibration to be applied.")
+        logger.debug("Calculating calibration for %d points.", self.size())
+
+        for freq, caldata in self.dataset.items():
+            g1 = self.gamma_short(freq)
+            g2 = self.gamma_open(freq)
+            g3 = self.gamma_load(freq)
+
+            gm1 = caldata["short"].z
+            gm2 = caldata["open"].z
+            gm3 = caldata["load"].z
+
+            try:
+                denominator = (g1 * (g2 - g3) * gm1 +
+                               g2 * g3 * gm2 - g2 * g3 * gm3 -
+                               (g2 * gm2 - g3 * gm3) * g1)
+                caldata["e00"] = - ((g2 * gm3 - g3 * gm3) * g1 * gm2 -
+                                    (g2 * g3 * gm2 - g2 * g3 * gm3 -
+                                     (g3 * gm2 - g2 * gm3) * g1) * gm1
+                                    ) / denominator
+                caldata["e11"] = ((g2 - g3) * gm1 - g1 * (gm2 - gm3) +
+                                  g3 * gm2 - g2 * gm3) / denominator
+                caldata["delta_e"] = - ((g1 * (gm2 - gm3) - g2 * gm2 + g3 *
+                                         gm3) * gm1 + (g2 * gm3 - g3 * gm3) *
+                                        gm2) / denominator
+            except ZeroDivisionError:
+                self.isCalculated = False
+                logger.error(
+                    "Division error - did you use the same measurement"
+                    " for two of short, open and load?")
+                raise ValueError(
+                    f"Two of short, open and load returned the same"
+                    f" values at frequency {freq}Hz.")
+
+            if self.isValid2Port():
+                caldata["e30"] = caldata["isolation"].z
+
+                gt = self.gamma_through(freq)
+                caldata["e10e32"] = (caldata["through"].z / gt - caldata["e30"]
+                                     ) * (1 - caldata["e11"]**2)
+
+        self.gen_interpolation()
+        self.isCalculated = True
+        logger.debug("Calibration correctly calculated.")
+
+    def gamma_short(self, freq: int) -> complex:
+        g = Calibration.IDEAL_SHORT
+        if not self.useIdealShort:
+            logger.debug("Using short calibration set values.")
+            Zsp = complex(0, 1) * 2 * math.pi * freq * (
+                self.shortL0 + self.shortL1 * freq +
+                self.shortL2 * freq**2 + self.shortL3 * freq**3)
+            # Referencing https://arxiv.org/pdf/1606.02446.pdf (18) - (21)
+            g = (Zsp / 50 - 1) / (Zsp / 50 + 1) * cmath.exp(
+                complex(0, 1) * 2 * math.pi * 2 * freq *
+                self.shortLength * -1)
+        return g
+
+    def gamma_open(self, freq: int) -> complex:
+        g = Calibration.IDEAL_OPEN
+        if not self.useIdealOpen:
+            logger.debug("Using open calibration set values.")
+            divisor = (2 * math.pi * freq * (
+                self.openC0 + self.openC1 * freq +
+                self.openC2 * freq**2 + self.openC3 * freq**3))
+            if divisor != 0:
+                Zop = complex(0, -1) / divisor
+                g = ((Zop / 50 - 1) / (Zop / 50 + 1)) * cmath.exp(
+                    complex(0, 1) * 2 * math.pi *
+                    2 * freq * self.openLength * -1)
+        return g
+
+    def gamma_load(self, freq: int) -> complex:
+        g = Calibration.IDEAL_LOAD
+        if not self.useIdealLoad:
+            logger.debug("Using load calibration set values.")
+            Zl = self.loadR + (complex(0, 1) * 2 *
+                               math.pi * freq * self.loadL)
+            g = (Zl / 50 - 1) / (Zl / 50 + 1) * cmath.exp(
+                complex(0, 1) * 2 * math.pi *
+                2 * freq * self.loadLength * -1)
+        return g
+
+    def gamma_through(self, freq: int) -> complex:
+        g = complex(1, 0)
+        if not self.useIdealThrough:
+            logger.debug("Using through calibration set values.")
+            g = cmath.exp(complex(0, 1) * 2 * math.pi *
+                          self.throughLength * freq * -1)
+        return g
+
+    def gen_interpolation(self):
+        freq = []
+        e00 = []
+        e11 = []
+        delta_e = []
+        e30 = []
+        e10e32 = []
+
+        for caldata in self.dataset.values():
+            freq.append(caldata["freq"])
+            e00.append(caldata["e00"])
+            e11.append(caldata["e11"])
+            delta_e.append(caldata["delta_e"])
+            e30.append(caldata["e30"])
+            e10e32.append(caldata["e10e32"])
+
+        self.interp = {
+            "e00": interp1d(freq, e00,
+                            kind="slinear", bounds_error=False,
+                            fill_value=(e00[0], e00[-1])),
+            "e11": interp1d(freq, e11,
+                            kind="slinear", bounds_error=False,
+                            fill_value=(e11[0], e11[-1])),
+            "delta_e": interp1d(freq, delta_e,
+                                kind="slinear", bounds_error=False,
+                                fill_value=(delta_e[0], delta_e[-1])),
+            "e30": interp1d(freq, e30,
+                            kind="slinear", bounds_error=False,
+                            fill_value=(e30[0], e30[-1])),
+            "e10e32": interp1d(freq, e10e32,
+                               kind="slinear", bounds_error=False,
+                               fill_value=(e10e32[0], e10e32[-1])),
+        }
+
+    def correct11(self, dp: Datapoint):
+        i = self.interp
+        s11 = (dp.z - i["e00"](dp.freq)) / (
+            (dp.z * i["e11"](dp.freq)) - i["delta_e"](dp.freq))
+        return Datapoint(dp.freq, s11.real, s11.imag)
+
+    def correct21(self, dp: Datapoint):
+        i = self.interp
+        s21 = (dp.z - i["e30"](dp.freq)) / i["e10e32"](dp.freq)
+        return Datapoint(dp.freq, s21.real, s21.imag)
+
+    # TODO: implement tests
+    def save(self, filename: str):
+        # Save the calibration data to file
+        if not self.isValid1Port():
+            raise ValueError("Not a valid 1-Port calibration")
+        with open(f"{filename}", "w") as calfile:
+            calfile.write("# Calibration data for NanoVNA-Saver\n")
+            for note in self.notes:
+                calfile.write(f"! {note}\n")
+            calfile.write(
+                "# Hz ShortR ShortI OpenR OpenI LoadR LoadI"
+                " ThroughR ThroughI IsolationR IsolationI\n")
+            for freq in self.dataset.frequencies():
+                calfile.write(f"{self.dataset.get(freq)}\n")
+
+    # TODO: implement tests
+    # TODO: Exception should be catched by caller
+    def load(self, filename):
+        self.source = os.path.basename(filename)
+        self.dataset = CalDataSet()
+        self.notes = []
+
+        parsed_header = False
+        with open(filename) as calfile:
+            for i, line in enumerate(calfile):
+                line = line.strip()
+                if line.startswith("!"):
+                    note = line[2:]
+                    self.notes.append(note)
+                    continue
+                if line.startswith("#"):
+                    if not parsed_header:
+                        # Check that this is a valid header
+                        if line == (
+                                "# Hz ShortR ShortI OpenR OpenI LoadR LoadI"
+                                " ThroughR ThroughI IsolationR IsolationI"):
+                            parsed_header = True
+                    continue
+                if not parsed_header:
+                    logger.warning(
+                        "Warning: Read line without having read header: %s",
+                        line)
+                    continue
+
+                m = RXP_CAL_LINE.search(line)
+                if not m:
+                    logger.warning("Illegal data in cal file. Line %i", i)
+                cal = m.groupdict()
+
+                if cal["throughr"]:
+                    nr_cals = 5
+                else:
+                    nr_cals = 3
+
+                for name in Calibration.CAL_NAMES[:nr_cals]:
+                    self.dataset.insert(
+                        name,
+                        Datapoint(int(cal["freq"]),
+                                  float(cal[f"{name}r"]),
+                                  float(cal[f"{name}i"])))

Hardware.py

@@ -0,0 +1,148 @@
+#  NanoVNASaver
+#
+#  A python program to view and export Touchstone data from a NanoVNA
+#  Copyright (C) 2019, 2020  Rune B. Broberg
+#  Copyright (C) 2020 NanoVNA-Saver Authors
+#
+#  This program is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU General Public License as published by
+#  the Free Software Foundation, either version 3 of the License, or
+#  (at your option) any later version.
+#
+#  This program is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU General Public License for more details.
+#
+#  You should have received a copy of the GNU General Public License
+#  along with this program.  If not, see <https://www.gnu.org/licenses/>.
+import logging
+import platform
+from collections import namedtuple
+from time import sleep
+from typing import List
+
+import serial
+from serial.tools import list_ports
+
+
+from Serial import drain_serial, Interface
+
+
+logger = logging.getLogger(__name__)
+
+USBDevice = namedtuple("Device", "vid pid name")
+
+USBDEVICETYPES = (
+    USBDevice(0x0483, 0x5740, "NanoVNA"),
+    USBDevice(0x16c0, 0x0483, "AVNA"),
+    USBDevice(0x04b4, 0x0008, "S-A-A-2"),
+)
+RETRIES = 3
+TIMEOUT = 0.2
+WAIT = 0.05
+
+# The USB Driver for NanoVNA V2 seems to deliver an
+# incompatible hardware info like:
+# 'PORTS\\VID_04B4&PID_0008\\DEMO'
+# This function will fix it.
+def _fix_v2_hwinfo(dev):
+    if dev.hwid == r'PORTS\VID_04B4&PID_0008\DEMO':
+        dev.vid, dev.pid = 0x04b4, 0x0008
+    return dev
+
+
+# Get list of interfaces with VNAs connected
+def get_interfaces() -> List[Interface]:
+    interfaces = []
+    # serial like usb interfaces
+    for d in list_ports.comports():
+        if platform.system() == 'Windows' and d.vid is None:
+            d = _fix_v2_hwinfo(d)
+        for t in USBDEVICETYPES:
+            if d.vid != t.vid or d.pid != t.pid:
+                continue
+            logger.debug("Found %s USB:(%04x:%04x) on port %s",
+                         t.name, d.vid, d.pid, d.device)
+            iface = Interface('serial', t.name)
+            iface.port = d.device
+            interfaces.append(iface)
+    return interfaces
+
+
+# def get_VNA(iface: Interface) -> 'VNA':
+#     # serial_port.timeout = TIMEOUT
+
+#     logger.info("Finding correct VNA type...")
+#     with iface.lock:
+#         vna_version = detect_version(iface)
+
+#     if vna_version == 'v2':
+#         logger.info("Type: NanoVNA-V2")
+#         return NanoVNA_V2(iface)
+
+#     logger.info("Finding firmware variant...")
+#     info = get_info(iface)
+#     if info.find("AVNA + Teensy") >= 0:
+#         logger.info("Type: AVNA")
+#         return AVNA(iface)
+#     if info.find("NanoVNA-H 4") >= 0:
+#         logger.info("Type: NanoVNA-H4")
+#         vna = NanoVNA_H4(iface)
+#         return vna
+#     if info.find("NanoVNA-H") >= 0:
+#         logger.info("Type: NanoVNA-H")
+#         vna = NanoVNA_H(iface)
+#         return vna
+#     if info.find("NanoVNA-F_V2") >= 0:
+#         logger.info("Type: NanoVNA-F_V2")
+#         return NanoVNA_F_V2(iface)
+#     if info.find("NanoVNA-F") >= 0:
+#         logger.info("Type: NanoVNA-F")
+#         return NanoVNA_F(iface)
+#     if info.find("NanoVNA") >= 0:
+#         logger.info("Type: Generic NanoVNA")
+#         return NanoVNA(iface)
+#     logger.warning("Did not recognize NanoVNA type from firmware.")
+#     return NanoVNA(iface)
+
+# def detect_version(serial_port: serial.Serial) -> str:
+#     data = ""
+#     for i in range(RETRIES):
+#         drain_serial(serial_port)
+#         serial_port.write("\r".encode("ascii"))
+#         sleep(0.05)
+#         data = serial_port.read(128).decode("ascii")
+#         if data.startswith("ch> "):
+#             return "v1"
+#         # -H versions
+#         if data.startswith("\r\nch> "):
+#             return "vh"
+#         if data.startswith("2"):
+#             return "v2"
+#         logger.debug("Retry detection: %s", i + 1)
+#     logger.error('No VNA detected. Hardware responded to CR with: %s', data)
+#     return ""
+
+# def get_info(serial_port: serial.Serial) -> str:
+#     for _ in range(RETRIES):
+#         drain_serial(serial_port)
+#         serial_port.write("info\r".encode("ascii"))
+#         lines = []
+#         retries = 0
+#         while True:
+#             line = serial_port.readline()
+#             line = line.decode("ascii").strip()
+#             if not line:
+#                 retries += 1
+#                 if retries > RETRIES:
+#                     return ""
+#                 sleep(WAIT)
+#                 continue
+#             if line == "info":  # suppress echo
+#                 continue
+#             if line.startswith("ch>"):
+#                 logger.debug("Needed retries: %s", retries)
+#                 break
+#             lines.append(line)
+#         return "\n".join(lines)

NanoVNA_V2.py

@@ -0,0 +1,446 @@
+#  NanoVNASaver
+#
+#  A python program to view and export Touchstone data from a NanoVNA
+#  Copyright (C) 2019, 2020  Rune B. Broberg
+#  Copyright (C) 2020 NanoVNA-Saver Authors
+#
+#  This program is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU General Public License as published by
+#  the Free Software Foundation, either version 3 of the License, or
+#  (at your option) any later version.
+#
+#  This program is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU General Public License for more details.
+#
+#  You should have received a copy of the GNU General Public License
+#  along with this program.  If not, see <https://www.gnu.org/licenses/>.
+import logging
+import platform
+import time
+from struct import pack, unpack_from
+
+from time import sleep
+from typing import List
+from Serial import Interface
+from Calibration import Calibration
+from RFTools import Datapoint
+
+__all__ = ['NanoVNA_V2_H4']
+
+if platform.system() != "Windows":
+    import tty
+
+logger = logging.getLogger(__name__)
+
+_CMD_NOP = 0x00
+_CMD_INDICATE = 0x0D
+_CMD_READ = 0x10
+_CMD_READ2 = 0x11
+_CMD_READ4 = 0x12
+_CMD_READFIFO = 0x18
+_CMD_WRITE = 0x20
+_CMD_WRITE2 = 0x21
+_CMD_WRITE4 = 0x22
+_CMD_WRITE8 = 0x23
+_CMD_WRITEFIFO = 0x28
+
+WRITE_SLEEP = 0.05
+
+_ADF4350_TXPOWER_DESC_MAP = {
+    0: "9dattenuation",
+    1: "6dB attenuation",
+    2: "3dB attenuation",
+    3: "Maximum",
+}
+_ADF4350_TXPOWER_DESC_REV_MAP = {
+    value: key for key, value in _ADF4350_TXPOWER_DESC_MAP.items()
+}
+
+# _NANOVNA_V2_CMDS = {
+#     'NOP'               : {'val': 0x, 'len':, 'resp': },
+#     'INDICATE'               : {'val': 0x, 'len':, 'resp': },
+#     'READ'               : {'val': 0x, 'len':, 'resp': },
+#     'READ2'               : {'val': 0x, 'len':, 'resp': },
+#     'READ4'               : {'val': 0x, 'len':, 'resp': },
+#     'READFIFO'               : {'val': 0x, 'len':, 'resp': },
+#     'NOP'               : {'val': 0x, 'len':, 'resp': },
+#     'NOP'               : {'val': 0x, 'len':, 'resp': },
+
+# }
+
+_NANOVNA_V2_REGS = {
+    "sweepStartHz": {"addr": 0x00, "len": 8},
+    "sweepStepHz": {"addr": 0x10, "len": 8},
+    "sweepPoints": {"addr": 0x20, "len": 2},
+    "valuesPerFrequency": {"addr": 0x22, "len": 2},
+    "rawSamplesMode": {"addr": 0x26, "len": 1},
+    "valuesFIFO": {"addr": 0x30, "len": 4},
+    "deviceVariant": {"addr": 0xF0, "len": 1},
+    "protocolVersion": {"addr": 0xF1, "len": 1},
+    "hardwareRevision": {"addr": 0xF2, "len": 1},
+    "firmwareMajor": {"addr": 0xF3, "len": 1},
+    "firmwareMinor": {"addr": 0xF4, "len": 1},
+}
+
+"""
+NanoVNA FIFO Format (32 bytes)
+
+fwd0Re      (uint32) 
+fw0Im       (uint32)
+rev0Re      (uint32)
+rev0Re      (uint32)
+rev1Re      (uint32)
+rev1Im      (uint32)
+freqIndex   (uint16)
+reserved    (6 bytes)
+"""
+_NANOVAN_V2_FIFO_FORMAT = "<iiiiiiHxxxxxx"
+
+
+class NanoVNA_V2_H4:
+    name = "NanoVNA V2 Plus4"
+
+    def __init__(self, iface: Interface = None):
+        self.iface = iface
+        self.iface.open()
+
+        if platform.system() != "Windows":
+            tty.setraw(self.iface.fd)
+
+        # reset protocol to known state
+        with self.iface.lock:
+            self.iface.write(pack("<Q", 0))
+            sleep(WRITE_SLEEP)
+        # # firmware major version of 0xff indicates dfu mode
+        # if self.version.data["major"] == 0xff:
+        #     raise IOError('Device is in DFU mode')
+
+        # Write Parameters
+        self.write_sleep = 0.05
+
+        # Device Ranges
+        self.valid_freq = range(
+            int(50e3), int(4.4e9 + 1)
+        )  # Frequency Range: 50kHz - 4.4GHz, 1Hz Resolution
+        self.valid_power = None  # Not Implemented / Fixed
+        self.valid_bw = 800  # Fixed
+        self.valid_points = range(1, 10000)  # TODO: Increase to actual limit
+
+        # Device Defaults
+        self.freq_start = 140e6
+        self.freq_step = None
+        self.freq_end = 4.4e9
+        self.bw = 800
+        self.points = 100
+
+        # Device Memory
+        self._fwd = None
+        self._refl = None
+        self._thru = None
+        self._f = None
+        self._s11 = None
+        self._s22 = None
+        self._cal = Calibration()
+        self._cal_valid = False
+
+        # Load Defaults
+        self.set_sweep(self.freq_start, self.freq_end, self.points)
+
+    def __del__(self):
+        self.iface.close()
+
+    def connect(self) -> bool:
+        self.iface.open()
+        return self.iface.isOpen()
+
+    def disconnect(self) -> bool:
+        self.iface.close()
+        return not self.iface.isOpen()
+
+    def reconnect(self) -> bool:
+        self.iface.close()
+        sleep(self.wait)
+        self.iface.open()
+        return self.iface.isOpen()
+
+    def _read_register(self, reg_name: str):
+        if reg_name not in _NANOVNA_V2_REGS:
+            raise ValueError(
+                f"must be name of nanoVNA v2 register: {_NANOVNA_V2_REGS.keys()}"
+            )
+
+        size = _NANOVNA_V2_REGS[reg_name]["len"]
+        addr = _NANOVNA_V2_REGS[reg_name]["addr"]
+
+        if size == 1:
+            packet = pack("<BB", _CMD_READ, addr)
+        elif size == 2:
+            packet = pack("<BB", _CMD_READ2, addr)
+        elif size == 4:
+            packet = pack("<BB", _CMD_READ4, addr)
+        elif size == 8:
+            packet = pack("<BBBB", _CMD_READ4, addr, _CMD_READ4, addr + 4)
+
+        self.iface.flush()
+        self.iface.write(packet)
+        resp = self.iface.read(size)
+
+        if len(resp) != size:
+            return None
+        return int.from_bytes(resp, byteorder="little")
+
+    def _write_register(self, reg_name, value):
+        if reg_name not in _NANOVNA_V2_REGS:
+            raise ValueError("must be name of nanoVNA v2 register")
+
+        size = _NANOVNA_V2_REGS[reg_name]["len"]
+        addr = _NANOVNA_V2_REGS[reg_name]["addr"]
+
+        if size == 1:
+            packet = pack("<BBB", _CMD_WRITE, addr, value)
+        elif size == 2:
+            packet = pack("<BBH", _CMD_WRITE2, addr, value)
+        elif size == 4:
+            packet = pack("<BBI", _CMD_WRITE4, addr, value)
+        elif size == 8:
+            packet = pack("<BBQ", _CMD_WRITE8, addr, value)
+
+        self.iface.write(packet)
+
+    def _read_fifo(self, points: int):
+        # clear something
+        self.iface.write(pack("<Q", 0))
+        
+        # clear FIFO
+        self._write_register("valuesFIFO", 0x00)
+
+        points_remaining = points
+        fifo_bytes = bytes()
+
+        while points_remaining > 0:
+
+            # calculate the number of points to read (255) at a time
+            points_to_read = min(255, points_remaining)
+            bytes_to_read = points_to_read * 32
+
+            # TODO: Fix
+            self.iface.write(pack("<BBB", _CMD_READFIFO, 0x30, points_to_read))
+            timeout = self.iface.timeout
+            self.iface.timeout = points_to_read * 0.032 + 0.1
+            resp = self.iface.read(bytes_to_read)
+
+            if len(resp) != bytes_to_read:
+                self.iface.timeout = timeout
+                return None
+
+            fifo_bytes = fifo_bytes + resp
+            points_remaining = points_remaining - points_to_read
+
+        # restore timeout
+        self.iface.timeout = timeout
+        return fifo_bytes
+
+    def read_registers(self):
+        i = dict()
+        for reg in _NANOVNA_V2_REGS.keys():
+            if reg == "valuesFIFO":  # skip
+                continue
+            i[reg] = self._read_register(reg)
+        return i
+
+    def set_sweep(
+        self, freq_start: float, freq_end: float, points: float = 101, avg: int = 1
+    ):
+
+        # input checking
+        if int(freq_start) not in self.valid_freq:
+            raise ValueError("outside of operating frequency range")
+        if int(freq_end) not in self.valid_freq:
+            raise ValueError("outside of operating frequency range")
+        if freq_end <= freq_start:
+            raise ValueError("start frequency must be < end frequency")
+        if int(points) not in self.valid_points:
+            raise ValueError("invalid number of data points")
+
+        # store and calc registers
+        self.points = int(points)
+        self.freq_start = int(freq_start)
+        self.freq_step = int((int(freq_end) - freq_start) / (self.points - 1))
+        self.freq_end = self.freq_start + self.freq_step * (self.points - 1)
+        self.avg = avg
+
+        # write registers
+        self._write_register("sweepStartHz", self.freq_start)
+        self._write_register("sweepStepHz", self.freq_step)
+        self._write_register("sweepPoints", self.points)
+        self._write_register("valuesPerFrequency", self.avg)
+
+        # allocate memory for sweep data
+        self._fwd = [complex()] * self.points
+        self._refl = [complex()] * self.points
+        self._thru = [complex()] * self.points
+        self._freq_index = [int()] * self.points
+        self._f = [int()] * self.points
+        self._s11 = [complex()] * self.points
+        self._s21 = [complex()] * self.points
+
+    def get_sweep(self) -> List:
+        reg = self.read_registers()
+        self.freq_start = reg["sweepStartHz"]
+        self.freq_step = reg["sweepStepHz"]
+        self.points = reg["sweepPoints"]
+        self.freq_end = self.freq_start + self.freq_step * (self.points - 1)
+        self.avg = reg["valuesPerFrequency"]
+        return (self.freq_start, self.freq_end, self.points, self.avg)
+
+    def wizard_cal(self, port1_only:bool=False):
+        # wipe old cal because I'm scared
+        self._cal = Calibration()
+
+        # short
+        input("connect Short to port 1...")
+        fs, s11_point, _ = self.measure(skip_calibration=True)
+        s11_dp = []
+        for (f, s11_point) in zip(fs, s11_point):
+            s11_dp.append(Datapoint(f, s11_point.real, s11_point.imag))
+        self._cal.insert("short", s11_dp)
+
+        # open
+        input("connect Open to port 1...")
+        fs, s11, _ = self.measure(skip_calibration=True)
+        s11_dp = []
+        for (f, s11_point) in zip(fs, s11):
+            s11_dp.append(Datapoint(f, s11_point.real, s11_point.imag))
+        self._cal.insert("open", s11_dp)
+
+        # load
+        input("connect Load to port 1...")
+        fs, s11, _ = self.measure(skip_calibration=True)
+        s11_dp = []
+        for (f, s11_point) in zip(fs, s11):
+            s11_dp.append(Datapoint(f, s11_point.real, s11_point.imag))
+        self._cal.insert("load", s11_dp)
+
+        # through
+        input("connect port 1 to port 2...")
+        fs, _, s21 = self.measure(skip_calibration=True)
+        s21_dp = []
+        for (f, s21_point) in zip(fs, s21):
+            s21_dp.append(Datapoint(f, s21_point.real, s21_point.imag))
+        self._cal.insert("through", s21_dp)
+
+        # isolation 
+        input("connect load to both ports...")
+        fs, _, s21 = self.measure(skip_calibration=True)
+        s21_dp = []
+        for (f, s21_point) in zip(fs, s21):
+            s21_dp.append(Datapoint(f, s21_point.real, s21_point.imag))
+        self._cal.insert("isolation", s21_dp)
+
+        # calculate corrections 
+        self._cal.calc_corrections()
+
+        if self._cal.isValid1Port(): 
+            self._cal_valid = True 
+        else: 
+            self._cal_valid = False
+
+        # Store Notes 
+        notes = dict() 
+        notes['Date'] = time.ctime() 
+        start, stop, points, avg = self.get_sweep()
+        notes['Sweep Start'] = start 
+        notes['Sweep Stop'] = stop 
+        notes['Sweep Points'] = points 
+        notes['Point Average'] = avg 
+        notes['Device Name'] = self.name 
+        notes['Device Settings'] = self.read_registers()
+        
+        self._cal.notes.append(notes)    # date 
+
+
+        return self._cal_valid
+
+    def save_cal(self, filename: str):
+        if self._cal_valid:
+            self._cal.save(filename)
+
+    def load_cal(self, filename: str) -> bool:
+        self._cal = Calibration()
+        self._cal.load(filename)
+        self._cal.calc_corrections()
+        if self._cal.isValid1Port():
+            self._cal_valid = True
+        else:
+            self._cal_valid = False
+        return self._cal_valid
+
+    def measure(self, skip_calibration=False):
+
+        # retrive data from fifo
+        fifo_bytes = self._read_fifo(self.points)
+
+        if fifo_bytes == None:
+            return None
+
+        # unpack data and convert to complex
+        for n_point in range(self.points):
+
+            # TODO: clean this up
+            (
+                fwd_real,
+                fwd_imag,
+                rev0_real,
+                rev0_imag,
+                rev1_real,
+                rev1_imag,
+                freq_index,
+            ) = unpack_from(_NANOVAN_V2_FIFO_FORMAT, fifo_bytes, n_point * 32)
+            fwd = complex(fwd_real, fwd_imag)
+            refl = complex(rev0_real, rev0_imag)
+            thru = complex(rev1_real, rev1_imag)
+
+            # store internally and calculate s11, s21
+            self._fwd[freq_index] = fwd
+            self._refl[freq_index] = refl
+            self._thru[freq_index] = thru
+
+            self._f[freq_index] = self.freq_start + self.freq_step * freq_index
+            self._s11[freq_index] = refl / fwd
+            self._s21[freq_index] = thru / fwd
+
+        # skip calibration if requested
+        if skip_calibration==True:
+            return self._f, self._s11, self._s21
+
+        # apply calibration
+        # TODO: Implement this better (skrf or custom modifications to calibration import)
+        cal_s11 = None
+        cal_s21 = None
+        if self._cal.isValid1Port():
+            cal_s11 = []
+            for f, s11_point in zip(self._f, self._s11):
+                dp = Datapoint(f, s11_point.real, s11_point.imag)
+                cal_s11.append(self._cal.correct11(dp).z)
+
+        if self._cal.isValid2Port():
+            cal_s21 = []
+            for f, s21_point in zip(self._f, self._s21):
+                dp = Datapoint(f, s21_point.real, s21_point.imag)
+                cal_s21.append(self._cal.correct21(dp).z)
+
+        if cal_s11 == None:
+            return self._f, None, None
+        
+        if cal_s21 == None:
+            return self._f, cal_s11, None
+
+        return self._f, cal_s11, cal_s21
+
+    # def setTXPower(self, freq_range, power_desc):
+    #     if freq_range[0] != 140e6:
+    #         raise ValueError('Invalid TX power frequency range')
+    #     # 140MHz..max => ADF4350
+    #     self._set_register(0x42, _ADF4350_TXPOWER_DESC_REV_MAP[power_desc], 1)

README.md

@@ -0,0 +1,23 @@
+# VNA
+
+## Objective 
+This library will _eventually_ be resposible for providing a uniform interface for VNA calibartion and control. This will be used for automated scripting of tests. 
+
+# To-Dos 
+    - [ ] Turn nanovnav2_h4 class into metaclass 
+    - [ ] Create regualer VNA class and document 
+    - [ ] Create dict to hold registers for NanoVNA2 
+    - [ ] Optimize calibration proceedure and get away from "datapoints" 
+    - [ ] Turn curiousVNA into a submodule / python package 
+
+# NanoVNAV2 H4 Notes 
+## Sampling Speed 
+Frequency Range (Default): 200MHz - 4.2GHz
+
+| Samples | Acquisition Time | 
+| - | - | 
+| 101 | 304 ms |  
+| 300 | 812 ms | 
+| 500 | 1.31 s | 
+| 1000 | 2.58 s | 
+| 4000 | 10.15 s| 

RFTools.py

@@ -0,0 +1,179 @@
+#  NanoVNASaver
+#
+#  A python program to view and export Touchstone data from a NanoVNA
+#  Copyright (C) 2019, 2020  Rune B. Broberg
+#  Copyright (C) 2020 NanoVNA-Saver Authors
+#
+#  This program is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU General Public License as published by
+#  the Free Software Foundation, either version 3 of the License, or
+#  (at your option) any later version.
+#
+#  This program is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU General Public License for more details.
+#
+#  You should have received a copy of the GNU General Public License
+#  along with this program.  If not, see <https://www.gnu.org/licenses/>.
+import math
+import cmath
+from threading import Lock
+from typing import Iterator, List, NamedTuple, Tuple
+
+import numpy as np
+from scipy.interpolate import interp1d
+
+from SITools import Format, clamp_value
+
+FMT_FREQ = Format()
+FMT_SHORT = Format(max_nr_digits=4)
+FMT_SWEEP = Format(max_nr_digits=9, allow_strip=True)
+
+
+class Datapoint(NamedTuple):
+    freq: int
+    re: float
+    im: float
+
+    @property
+    def z(self) -> complex:
+        """ return the datapoint impedance as complex number """
+        # FIXME: not impedance, but s11 ?
+        return complex(self.re, self.im)
+
+    @property
+    def phase(self) -> float:
+        """ return the datapoint's phase value """
+        return cmath.phase(self.z)
+
+    @property
+    def gain(self) -> float:
+        mag = abs(self.z)
+        if mag > 0:
+            return 20 * math.log10(mag)
+        return -math.inf
+
+    @property
+    def vswr(self) -> float:
+        mag = abs(self.z)
+        if mag == 1:
+            return 1
+        return (1 + mag) / (1 - mag)
+
+    @property
+    def wavelength(self) -> float:
+        return 299792458 / self.freq
+
+    def impedance(self, ref_impedance: float = 50) -> complex:
+        return gamma_to_impedance(self.z, ref_impedance)
+
+    def shuntImpedance(self, ref_impedance: float = 50) -> complex:
+        try:
+            return 0.5 * ref_impedance * self.z / (1 - self.z)
+        except ZeroDivisionError:
+            return math.inf
+
+    def seriesImpedance(self, ref_impedance: float = 50) -> complex:
+        try:
+            return 2 * ref_impedance * (1 - self.z) / self.z
+        except ZeroDivisionError:
+            return math.inf
+
+    def qFactor(self, ref_impedance: float = 50) -> float:
+        imp = self.impedance(ref_impedance)
+        if imp.real == 0.0:
+            return -1
+        return abs(imp.imag / imp.real)
+
+    def capacitiveEquivalent(self, ref_impedance: float = 50) -> float:
+        return impedance_to_capacitance(self.impedance(ref_impedance), self.freq)
+
+    def inductiveEquivalent(self, ref_impedance: float = 50) -> float:
+        return impedance_to_inductance(self.impedance(ref_impedance), self.freq)
+
+
+def gamma_to_impedance(gamma: complex, ref_impedance: float = 50) -> complex:
+    """Calculate impedance from gamma"""
+    try:
+        return ((-gamma - 1) / (gamma - 1)) * ref_impedance
+    except ZeroDivisionError:
+        return math.inf
+
+
+def groupDelay(data: List[Datapoint], index: int) -> float:
+    idx0 = clamp_value(index - 1, 0, len(data) - 1)
+    idx1 = clamp_value(index + 1, 0, len(data) - 1)
+    delta_angle = data[idx1].phase - data[idx0].phase
+    delta_freq = data[idx1].freq - data[idx0].freq
+    if delta_freq == 0:
+        return 0
+    val = -delta_angle / math.tau / delta_freq
+    return val
+
+
+def impedance_to_capacitance(z: complex, freq: float) -> float:
+    """Calculate capacitive equivalent for reactance"""
+    if freq == 0:
+        return -math.inf
+    if z.imag == 0:
+        return math.inf
+    return -(1 / (freq * 2 * math.pi * z.imag))
+
+
+def impedance_to_inductance(z: complex, freq: float) -> float:
+    """Calculate inductive equivalent for reactance"""
+    if freq == 0:
+        return 0
+    return z.imag * 1 / (freq * 2 * math.pi)
+
+
+def impedance_to_norm(z: complex, ref_impedance: float = 50) -> complex:
+    """Calculate normalized z from impedance"""
+    return z / ref_impedance
+
+
+def norm_to_impedance(z: complex, ref_impedance: float = 50) -> complex:
+    """Calculate impedance from normalized z"""
+    return z * ref_impedance
+
+
+def parallel_to_serial(z: complex) -> complex:
+    """Convert parallel impedance to serial impedance equivalent"""
+    z_sq_sum = z.real ** 2 + z.imag ** 2
+    # TODO: Fix divide by zero
+    return complex(z.real * z.imag ** 2 / z_sq_sum,
+                   z.real ** 2 * z.imag / z_sq_sum)
+
+
+def reflection_coefficient(z: complex, ref_impedance: float = 50) -> complex:
+    """Calculate reflection coefficient for z"""
+    return (z - ref_impedance) / (z + ref_impedance)
+
+
+def serial_to_parallel(z: complex) -> complex:
+    """Convert serial impedance to parallel impedance equivalent"""
+    z_sq_sum = z.real ** 2 + z.imag ** 2
+    if z.real == 0 and z.imag == 0:
+        return complex(math.inf, math.inf)
+    # only possible if real and imag == 0, therefor commented out
+    # if z_sq_sum == 0:
+    #     return complex(0, 0)
+    if z.imag == 0:
+        return complex(z_sq_sum / z.real, math.copysign(math.inf, z_sq_sum))
+    if z.real == 0:
+        return complex(math.copysign(math.inf, z_sq_sum), z_sq_sum / z.imag)
+    return complex(z_sq_sum / z.real, z_sq_sum / z.imag)
+
+
+def corr_att_data(data: List[Datapoint], att: float) -> List[Datapoint]:
+    """Correct the ratio for a given attenuation on s21 input"""
+    if att <= 0:
+        return data
+    else:
+        att = 10**(att / 20)
+    ndata = []
+    for dp in data:
+        corrected = dp.z * att
+        ndata.append(Datapoint(dp.freq, corrected.real, corrected.imag))
+    return ndata

SITools.py

@@ -0,0 +1,169 @@
+#  NanoVNASaver
+#
+#  A python program to view and export Touchstone data from a NanoVNA
+#  Copyright (C) 2019, 2020  Rune B. Broberg
+#  Copyright (C) 2020 NanoVNA-Saver Authors
+#
+#  This program is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU General Public License as published by
+#  the Free Software Foundation, either version 3 of the License, or
+#  (at your option) any later version.
+#
+#  This program is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU General Public License for more details.
+#
+#  You should have received a copy of the GNU General Public License
+#  along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from __future__ import annotations
+import math
+import decimal
+from typing import NamedTuple, Union
+from numbers import Number, Real
+
+PREFIXES = ("y", "z", "a", "f", "p", "n", "µ", "m",
+            "", "k", "M", "G", "T", "P", "E", "Z", "Y")
+
+
+def clamp_value(value: Real, rmin: Real, rmax: Real) -> Real:
+    assert rmin <= rmax
+    if value < rmin:
+        return rmin
+    if value > rmax:
+        return rmax
+    return value
+
+
+class Format(NamedTuple):
+    max_nr_digits: int = 6
+    fix_decimals: bool = False
+    space_str: str = ""
+    assume_infinity: bool = True
+    min_offset: int = -8
+    max_offset: int = 8
+    allow_strip: bool = False
+    allways_signed: bool = False
+    printable_min: float = -math.inf
+    printable_max: float = math.inf
+    unprintable_under: str = ""
+    unprintable_over: str = ""
+    parse_sloppy_unit: bool = False
+    parse_sloppy_kilo: bool = False
+    parse_clamp_min: float = -math.inf
+    parse_clamp_max: float = math.inf
+
+
+class Value:
+    CTX = decimal.Context(prec=60, Emin=-27, Emax=27)
+
+    def __init__(self,
+                 value: Union[Number, str] = 0,
+                 unit: str = "",
+                 fmt=Format()):
+        assert 1 <= fmt.max_nr_digits <= 30
+        assert -8 <= fmt.min_offset <= fmt.max_offset <= 8
+        assert fmt.parse_clamp_min < fmt.parse_clamp_max
+        assert fmt.printable_min < fmt.printable_max
+        self._unit = unit
+        self.fmt = fmt
+        if isinstance(value, str):
+            self._value = math.nan
+            self.parse(value)
+        else:
+            self._value = decimal.Decimal(value, context=Value.CTX)
+
+    def __repr__(self) -> str:
+        return (f"{self.__class__.__name__}(" + repr(self._value) +
+                f", '{self._unit}', {self.fmt})")
+
+    def __str__(self) -> str:
+        fmt = self.fmt
+        if fmt.assume_infinity and abs(self._value) >= 10 ** ((fmt.max_offset + 1) * 3):
+            return ("-" if self._value < 0 else "") + "\N{INFINITY}" + fmt.space_str + self._unit
+        if self._value < fmt.printable_min:
+            return fmt.unprintable_under + self._unit
+        if self._value > fmt.printable_max:
+            return fmt.unprintable_over + self._unit
+
+        if self._value == 0:
+            offset = 0
+        else:
+            offset = clamp_value(
+                int(math.log10(abs(self._value)) // 3), fmt.min_offset, fmt.max_offset)
+
+        real = float(self._value) / (10 ** (offset * 3))
+
+        if fmt.max_nr_digits < 3:
+            formstr = ".0f"
+        else:
+            max_digits = fmt.max_nr_digits + (
+                (1 if not fmt.fix_decimals and abs(real) < 10 else 0) +
+                (1 if not fmt.fix_decimals and abs(real) < 100 else 0))
+            formstr = "." + str(max_digits - 3) + "f"
+
+        if self.fmt.allways_signed:
+            formstr = "+" + formstr
+        result = format(real, formstr)
+
+        if float(result) == 0.0:
+            offset = 0
+
+        if self.fmt.allow_strip and "." in result:
+            result = result.rstrip("0").rstrip(".")
+
+        return result + fmt.space_str + PREFIXES[offset + 8] + self._unit
+
+    def __int__(self):
+        return round(self._value)
+
+    def __float__(self):
+        return float(self._value)
+
+    @property
+    def value(self):
+        return self._value
+
+    @value.setter
+    def value(self, value: Number):
+        self._value = decimal.Decimal(value, context=Value.CTX)
+
+    def parse(self, value: str) -> "Value":
+        if isinstance(value, Number):
+            self.value = value
+            return self
+
+        value = value.replace(" ", "")  # Ignore spaces
+
+        if self._unit and (
+                value.endswith(self._unit) or
+                (self.fmt.parse_sloppy_unit and
+                 value.lower().endswith(self._unit.lower()))):  # strip unit
+            value = value[:-len(self._unit)]
+
+        factor = 1
+        # fix for e.g. KHz, mHz gHz as milli-Hertz mostly makes no
+        # sense in NanoVNAs context
+        if self.fmt.parse_sloppy_kilo and value[-1] in ("K", "m", "g"):
+            value = value[:-1] + value[-1].swapcase()
+        if value[-1] in PREFIXES:
+            factor = 10 ** ((PREFIXES.index(value[-1]) - 8) * 3)
+            value = value[:-1]
+
+        if self.fmt.assume_infinity and value == "\N{INFINITY}":
+            self._value = math.inf
+        elif self.fmt.assume_infinity and value == "-\N{INFINITY}":
+            self._value = -math.inf
+        else:
+            try:
+                self._value = (decimal.Decimal(value, context=Value.CTX)
+                               * decimal.Decimal(factor, context=Value.CTX))
+            except decimal.InvalidOperation:
+                raise ValueError
+            self._value = clamp_value(
+                self._value, self.fmt.parse_clamp_min, self.fmt.parse_clamp_max)
+        return self
+
+    @property
+    def unit(self) -> str:
+        return self._unit

Serial.py

@@ -0,0 +1,53 @@
+#  NanoVNASaver
+#
+#  A python program to view and export Touchstone data from a NanoVNA
+#  Copyright (C) 2019, 2020  Rune B. Broberg
+#  Copyright (C) 2020 NanoVNA-Saver Authors
+#
+#  This program is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU General Public License as published by
+#  the Free Software Foundation, either version 3 of the License, or
+#  (at your option) any later version.
+#
+#  This program is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU General Public License for more details.
+#
+#  You should have received a copy of the GNU General Public License
+#  along with this program.  If not, see <https://www.gnu.org/licenses/>.
+import logging
+from threading import Lock
+
+import serial
+
+logger = logging.getLogger(__name__)
+
+
+def drain_serial(serial_port: serial.Serial):
+    """drain up to 64k outstanding data in the serial incoming buffer"""
+    # logger.debug("Draining: %s", serial_port)
+    timeout = serial_port.timeout
+    serial_port.timeout = 0.05
+    for _ in range(512):
+        cnt = len(serial_port.read(128))
+        if not cnt:
+            serial_port.timeout = timeout
+            return
+    serial_port.timeout = timeout
+    logger.warning("unable to drain all data")
+
+
+class Interface(serial.Serial):
+    def __init__(self, interface_type: str, comment, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        assert interface_type in ('serial', 'usb', 'bt', 'network')
+        self.type = interface_type
+        self.comment = comment
+        self.port = None
+        self.baudrate = 115200
+        self.timeout = 0.05
+        self.lock = Lock()
+
+    def __str__(self):
+        return f"{self.port} ({self.comment})"

Touchstone.py

@@ -0,0 +1,290 @@
+#  NanoVNASaver
+#
+#  A python program to view and export Touchstone data from a NanoVNA
+#  Copyright (C) 2019, 2020  Rune B. Broberg
+#  Copyright (C) 2020 NanoVNA-Saver Authors
+#
+#  This program is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU General Public License as published by
+#  the Free Software Foundation, either version 3 of the License, or
+#  (at your option) any later version.
+#
+#  This program is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU General Public License for more details.
+#
+#  You should have received a copy of the GNU General Public License
+#  along with this program.  If not, see <https://www.gnu.org/licenses/>.
+import logging
+import math
+import cmath
+import io
+from operator import attrgetter
+
+from typing import List
+
+from scipy.interpolate import interp1d
+
+from NanoVNASaver.RFTools import Datapoint
+
+logger = logging.getLogger(__name__)
+
+
+class Options:
+    # Fun fact: In Touchstone 1.1 spec all params are optional unordered.
+    # Just the line has to start with "#"
+    UNIT_TO_FACTOR = {
+        "ghz": 10**9,
+        "mhz": 10**6,
+        "khz": 10**3,
+        "hz": 10**0,
+    }
+    VALID_UNITS = UNIT_TO_FACTOR.keys()
+    VALID_PARAMETERS = "syzgh"
+    VALID_FORMATS = ("ma", "db", "ri")
+
+    def __init__(self,
+                 unit: str = "GHZ",
+                 parameter: str = "S",
+                 t_format: str = "ma",
+                 resistance: int = 50):
+        # set defaults
+        assert unit.lower() in Options.VALID_UNITS
+        assert parameter.lower() in Options.VALID_PARAMETERS
+        assert t_format.lower() in Options.VALID_FORMATS
+        assert resistance > 0
+        self.unit = unit.lower()
+        self.parameter = parameter.lower()
+        self.format = t_format.lower()
+        self.resistance = resistance
+
+    @property
+    def factor(self) -> int:
+        return Options.UNIT_TO_FACTOR[self.unit]
+
+    def __str__(self) -> str:
+        return (
+            f"# {self.unit} {self.parameter}"
+            f" {self.format} r {self.resistance}"
+        ).upper()
+
+    def parse(self, line: str):
+        if not line.startswith("#"):
+            raise TypeError("Not an option line: " + line)
+        punit = pparam = pformat = presist = False
+        params = iter(line[1:].lower().split())
+        for p in params:
+            if p in Options.VALID_UNITS and not punit:
+                self.unit = p
+                punit = True
+            elif p in Options.VALID_PARAMETERS and not pparam:
+                self.parameter = p
+                pparam = True
+            elif p in Options.VALID_FORMATS and not pformat:
+                self.format = p
+                pformat = True
+            elif p == "r" and not presist:
+                rstr = next(params)
+                try:
+                    self.resistance = int(rstr)
+                except ValueError:
+                    logger.warning("Non integer resistance value: %s", rstr)
+                    self.resistance = int(float(rstr))
+            else:
+                raise TypeError("Illegal option line: " + line)
+
+
+class Touchstone:
+    FIELD_ORDER = ("11", "21", "12", "22")
+
+    def __init__(self, filename: str):
+        self.filename = filename
+        self.sdata = [[], [], [], []]  # at max 4 data pairs
+        self.comments = []
+        self.opts = Options()
+        self._interp = {}
+
+    @property
+    def s11data(self) -> List[Datapoint]:
+        return self.s("11")
+
+    @s11data.setter
+    def s11data(self, value: List[Datapoint]):
+        self.sdata[0] = value
+
+    @property
+    def s12data(self) -> List[Datapoint]:
+        return self.s("12")
+
+    @s12data.setter
+    def s12data(self, value: List[Datapoint]):
+        self.sdata[2] = value
+
+    @property
+    def s21data(self) -> List[Datapoint]:
+        return self.s("21")
+
+    @s21data.setter
+    def s21data(self, value: List[Datapoint]):
+        self.sdata[1] = value
+
+    @property
+    def s22data(self) -> List[Datapoint]:
+        return self.s("22")
+
+    @s22data.setter
+    def s22data(self, value: List[Datapoint]):
+        self.sdata[3] = value
+
+    @property
+    def r(self) -> int:
+        return self.opts.resistance
+
+    def s(self, name: str) -> List[Datapoint]:
+        return self.sdata[Touchstone.FIELD_ORDER.index(name)]
+
+    def s_freq(self, name: str, freq: int) -> Datapoint:
+        return Datapoint(freq,
+                         float(self._interp[name]["real"](freq)),
+                         float(self._interp[name]["imag"](freq)))
+
+    def min_freq(self) -> int:
+        return self.s("11")[0].freq
+
+    def max_freq(self) -> int:
+        return self.s("11")[-1].freq
+
+    def gen_interpolation(self):
+        for i in Touchstone.FIELD_ORDER:
+            freq = []
+            real = []
+            imag = []
+
+            for dp in self.s(i):
+                freq.append(dp.freq)
+                real.append(dp.re)
+                imag.append(dp.im)
+
+            self._interp[i] = {
+                "real": interp1d(freq, real,
+                                 kind="slinear", bounds_error=False,
+                                 fill_value=(real[0], real[-1])),
+                "imag": interp1d(freq, imag,
+                                 kind="slinear", bounds_error=False,
+                                 fill_value=(imag[0], imag[-1])),
+            }
+
+    def _parse_comments(self, fp) -> str:
+        for line in fp:
+            line = line.strip()
+            if line.startswith("!"):
+                logger.info(line)
+                self.comments.append(line)
+                continue
+            return line
+
+    def _append_line_data(self, freq: int, data: list):
+        data_list = iter(self.sdata)
+        vals = iter(data)
+        for v in vals:
+            if self.opts.format == "ri":
+                next(data_list).append(Datapoint(freq, float(v), float(next(vals))))
+            if self.opts.format == "ma":
+                z = cmath.rect(float(v), math.radians(float(next(vals))))
+                next(data_list).append(Datapoint(freq, z.real, z.imag))
+            if self.opts.format == "db":
+                z = cmath.rect(10 ** (float(v) / 20), math.radians(float(next(vals))))
+                next(data_list).append(Datapoint(freq, z.real, z.imag))
+
+    def load(self):
+        logger.info("Attempting to open file %s", self.filename)
+        try:
+            with open(self.filename) as infile:
+                self.loads(infile.read())
+        except IOError as e:
+            logger.exception("Failed to open %s: %s", self.filename, e)
+
+    def loads(self, s: str):
+        """Parse touchstone 1.1 string input
+           appends to existing sdata if Touchstone object exists
+        """
+        try:
+            self._loads(s)
+        except TypeError as e:
+            logger.exception("Failed to parse %s: %s", self.filename, e)
+
+    def _loads(self, s: str):
+        need_reorder = False
+        with io.StringIO(s) as file:
+            opts_line = self._parse_comments(file)
+            self.opts.parse(opts_line)
+
+            prev_freq = 0.0
+            prev_len = 0
+            for line in file:
+                line = line.strip()
+                # ignore empty lines (even if not specified)
+                if line == "":
+                    continue
+                # accept comment lines after header
+                if line.startswith("!"):
+                    logger.warning("Comment after header: %s", line)
+                    self.comments.append(line)
+                    continue
+
+                # ignore comments at data end
+                data = line.split('!')[0]
+                data = data.split()
+                freq, data = round(float(data[0]) * self.opts.factor), data[1:]
+                data_len = len(data)
+
+                # consistency checks
+                if freq <= prev_freq:
+                    logger.warning("Frequency not ascending: %s", line)
+                    need_reorder = True
+                prev_freq = freq
+
+                if prev_len == 0:
+                    prev_len = data_len
+                    if data_len % 2:
+                        raise TypeError("Data values aren't pairs: " + line)
+                elif data_len != prev_len:
+                    raise TypeError("Inconsistent number of pairs: " + line)
+
+                self._append_line_data(freq, data)
+            if need_reorder:
+                logger.warning("Reordering data")
+                for datalist in self.sdata:
+                    datalist.sort(key=attrgetter("freq"))
+
+    def save(self, nr_params: int = 1):
+        """Save touchstone data to file.
+
+        Args:
+            nr_params: Number of s-parameters. 2 for s1p, 4 for s2p
+        """
+
+        logger.info("Attempting to open file %s for writing",
+                    self.filename)
+        with open(self.filename, "w") as outfile:
+            outfile.write(self.saves(nr_params))
+
+    def saves(self, nr_params: int = 1) -> str:
+        """Returns touchstone data as string.
+
+        Args:
+            nr_params: Number of s-parameters. 1 for s1p, 4 for s2p
+        """
+        assert nr_params in (1, 4)
+
+        ts_str = "# HZ S RI R 50\n"
+        for i, dp_s11 in enumerate(self.s11data):
+            ts_str += f"{dp_s11.freq} {dp_s11.re} {dp_s11.im}"
+            for j in range(1, nr_params):
+                dp = self.sdata[j][i]
+                if dp.freq != dp_s11.freq:
+                    raise LookupError("Frequencies of sdata not correlated")
+                ts_str += f" {dp.re} {dp.im}"
+            ts_str += "\n"
+        return ts_str

VNA.py

@@ -0,0 +1,91 @@
+#  NanoVNASaver
+#
+#  A python program to view and export Touchstone data from a NanoVNA
+#  Copyright (C) 2019, 2020  Rune B. Broberg
+#  Copyright (C) 2020 NanoVNA-Saver Authors
+#
+#  This program is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU General Public License as published by
+#  the Free Software Foundation, either version 3 of the License, or
+#  (at your option) any later version.
+#
+#  This program is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU General Public License for more details.
+#
+#  You should have received a copy of the GNU General Public License
+#  along with this program.  If not, see <https://www.gnu.org/licenses/>.
+import logging
+from collections import OrderedDict
+from time import sleep
+from typing import List, Iterator
+from Serial import Interface, drain_serial
+
+logger = logging.getLogger(__name__)
+
+
+class VNA:
+    name = "VNA"
+    wait = 0.05
+
+    def __init__(self, iface: Interface):
+        self.iface = iface
+        self.valid_freq = None
+        self.valid_power = None
+        self.valid_bw = None
+        self.valid_points = None
+
+        
+        self.freq_start = None
+        self.power = None 
+        self.points = None 
+        self.bw = None
+        
+    def connect(self) -> bool:
+        self.iface.open() 
+        return self.iface.isOpen()
+
+    def disconnect(self) -> bool: 
+        self.iface.close()
+        return not self.iface.isOpen()
+
+    def reconnect(self) -> bool: 
+        self.iface.close()
+        sleep(self.wait)
+        self.iface.open()
+        return self.iface.isOpen()
+
+    def set_bw(self): 
+        pass 
+
+    def set_sweep(self, freq_start: float, freq_end: float, points: int):
+        pass
+
+    def get_bw(self) -> float: 
+        pass 
+
+    def get_sweep(self) -> List: 
+        pass 
+
+    def get_info(self) -> Dict: 
+        pass 
+
+    def run(self) -> List: 
+        pass 
+
+    def set_power(self, power: float): 
+        pass 
+
+    def get_power(self) -> float: 
+        pass
+
+    def set_cal(self): 
+        pass 
+
+    def get_cal(self): 
+        pass 
+
+
+
+