DataModel NetCDF Example¶

This notebook shows how to save and load a custom DataModel to NetCDF with nested and mixed array payloads.

In [ ]:

from pathlib import Path

import numpy as np
import tunits
from qxcore.model import DataModel

from pathlib import Path import numpy as np import tunits from qxcore.model import DataModel

In [ ]:

class DemoResult(DataModel):
    """Demo model with nested arrays and tunits values."""

    name: str
    waveform: np.ndarray
    delay: tunits.ValueArray
    iq_blocks: dict[str, np.ndarray]
    traces: list[np.ndarray]
    metadata: dict[str, str]

class DemoResult(DataModel): """Demo model with nested arrays and tunits values.""" name: str waveform: np.ndarray delay: tunits.ValueArray iq_blocks: dict[str, np.ndarray] traces: list[np.ndarray] metadata: dict[str, str]

In [ ]:

result = DemoResult(
    name="demo",
    waveform=np.array([[1 + 2j, 3 + 4j], [5 + 6j, 7 + 8j]]),
    delay=tunits.ValueArray([10, 20], tunits.ns),
    iq_blocks={
        "readout": np.array([0.1 + 0.2j, 0.2 + 0.3j]),
        "reference": np.array([0.0 + 0.0j, 0.5 + 0.1j]),
    },
    traces=[
        np.array([1.0, 2.0, 3.0], dtype=np.float32),
        np.array([1, 2, 3, 4], dtype=np.int32),
    ],
    metadata={"author": "qubex", "experiment": "demo"},
)
print(result)

result = DemoResult( name="demo", waveform=np.array([[1 + 2j, 3 + 4j], [5 + 6j, 7 + 8j]]), delay=tunits.ValueArray([10, 20], tunits.ns), iq_blocks={ "readout": np.array([0.1 + 0.2j, 0.2 + 0.3j]), "reference": np.array([0.0 + 0.0j, 0.5 + 0.1j]), }, traces=[ np.array([1.0, 2.0, 3.0], dtype=np.float32), np.array([1, 2, 3, 4], dtype=np.int32), ], metadata={"author": "qubex", "experiment": "demo"}, ) print(result)

In [ ]:

output = Path(".rawdata/demo_data_model.nc")
output.parent.mkdir(parents=True, exist_ok=True)
saved = result.save_netcdf(output)
print(saved)

output = Path(".rawdata/demo_data_model.nc") output.parent.mkdir(parents=True, exist_ok=True) saved = result.save_netcdf(output) print(saved)

In [ ]:

restored = DemoResult.load_netcdf(saved)
print("name:", restored.name)
print("waveform:", restored.waveform.shape, restored.waveform.dtype)
print("delay:", restored.delay, restored.delay.units)
print("iq_blocks:", {k: (v.shape, v.dtype) for k, v in restored.iq_blocks.items()})
print("trace dtypes:", [t.dtype for t in restored.traces])
print("metadata:", restored.metadata)

restored = DemoResult.load_netcdf(saved) print("name:", restored.name) print("waveform:", restored.waveform.shape, restored.waveform.dtype) print("delay:", restored.delay, restored.delay.units) print("iq_blocks:", {k: (v.shape, v.dtype) for k, v in restored.iq_blocks.items()}) print("trace dtypes:", [t.dtype for t in restored.traces]) print("metadata:", restored.metadata)

NetCDF4 tutorial: attributes, dimensions, variables¶

• Global attributes: file-level metadata (format, model_class, ...)
• Dimensions: named axis definitions (dim_0, ...)
• Variables: actual arrays that reference dimensions

In [ ]:

from netCDF4 import Dataset

with Dataset(saved, mode="r", auto_complex=True) as ds:
    print("[global attributes]")
    for key in ds.ncattrs():
        value = ds.getncattr(key)
        print(f"- {key}: {value}")

    print("\n[dimensions]")
    for name, dim in ds.dimensions.items():
        print(f"- {name}: size={len(dim)}, unlimited={dim.isunlimited()}")

    print("\n[variables]")
    for name, var in ds.variables.items():
        print(f"- {name}: dims={var.dimensions}, shape={var.shape}, dtype={var.dtype}")

from netCDF4 import Dataset with Dataset(saved, mode="r", auto_complex=True) as ds: print("[global attributes]") for key in ds.ncattrs(): value = ds.getncattr(key) print(f"- {key}: {value}") print("\n[dimensions]") for name, dim in ds.dimensions.items(): print(f"- {name}: size={len(dim)}, unlimited={dim.isunlimited()}") print("\n[variables]") for name, var in ds.variables.items(): print(f"- {name}: dims={var.dimensions}, shape={var.shape}, dtype={var.dtype}")

Inspect payload structure¶

payload_json stores the model structure, while array values live in NetCDF variables.

In [ ]:

import json

with Dataset(saved, mode="r", auto_complex=True) as ds:
    payload = ds.getncattr("payload_json")
    obj = json.loads(payload)

print("top-level keys:", list(obj.keys()))
print("waveform ref:", obj["waveform"])
print("first trace ref:", obj["traces"][0])

import json with Dataset(saved, mode="r", auto_complex=True) as ds: payload = ds.getncattr("payload_json") obj = json.loads(payload) print("top-level keys:", list(obj.keys())) print("waveform ref:", obj["waveform"]) print("first trace ref:", obj["traces"][0])