start a pickle-less cal read/write

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