Skip to navigation
StanzaCore Concepts

Data Logging

Stanza automatically logs your measurement data in both HDF5 and JSON formats. You don’t need to write separate logging code for each routine - just pass a DataLogger to RoutineRunner and your sweep data, parameters, and analysis results are saved automatically.

Quick Start

Enable logging in three steps:

from stanza.logger import DataLogger
from stanza.routines import RoutineRunner
from stanza.models import DeviceConfig
# 1. Create a logger
logger = DataLogger(
    name="logger",
    routine_name="characterization",
    base_dir="./data",
    formats=["hdf5", "jsonl"]
)
# 2. Pass it to RoutineRunner
config = DeviceConfig.from_yaml("device.yaml")
runner = RoutineRunner(configs=[config], logger=logger)
# 3. Run routines - data logs automatically
runner.run("charge_stability")

Your data is saved to ./data/characterization/<session_id>/ with both HDF5 and JSONL files.

What Gets Logged Automatically

When you use device.sweep_1d() or device.sweep_2d() with a session parameter, Stanza logs:

  • Sweep data: Voltage arrays and measured signals
  • Labels: Axis labels for plotting
  • Metadata: Gate names, contact information, timestamps
  • Parameters: Routine parameters from YAML or runtime

Example - this routine automatically logs everything:

@routine
def barrier_sweep(ctx, gate, v_start, v_stop, n_points, contact, session=None):
    """Sweep a barrier gate."""
    device = ctx.resources.device
    voltages = np.linspace(v_start, v_stop, n_points)
    # Session parameter enables automatic logging
    v_data, i_data = device.sweep_1d(
        gate,
        voltages.tolist(),
        contact,
        session=session  # ← This enables logging
    )
    return {"voltages": v_data, "currents": i_data}

When you run this:

runner.run("barrier_sweep", gate="LEFT_BARRIER", v_start=-2.0, v_stop=0.0,
           n_points=100, contact="DRAIN")

Stanza automatically logs:

  • The voltage sweep (v_data)
  • The measured currents (i_data)
  • Parameters: gate="LEFT_BARRIER", v_start=-2.0, etc.
  • Metadata: timestamp, gate name, contact name

File Organization

Data is organized by routine and session:

data/
  characterization/
    20241003_143022_barrier_sweep/
      session.h5              # HDF5 format (for analysis)
      sweep.jsonl             # JSON Lines (human-readable)
      measurement.jsonl
      session_metadata.json   # Session info

Each time you run a routine, a new session directory is created with a timestamp.

Reading Your Data Back

Reading HDF5 Files

HDF5 is great for numerical analysis:

import h5py
import matplotlib.pyplot as plt
# Open the file
with h5py.File("data/characterization/20241003_143022/session.h5", "r") as f:
    # Read sweep data
    voltages = f["sweeps"]["LEFT_BARRIER_sweep_DRAIN"]["x_data"][:]
    currents = f["sweeps"]["LEFT_BARRIER_sweep_DRAIN"]["y_data"][:]
    # Read metadata
    gate = f["sweeps"]["LEFT_BARRIER_sweep_DRAIN"]["metadata"]["gate"][()].decode()
    contact = f["sweeps"]["LEFT_BARRIER_sweep_DRAIN"]["metadata"]["contact"][()].decode()
# Plot
plt.plot(voltages, currents * 1e9)  # Convert to nA
plt.xlabel("Voltage (V)")
plt.ylabel("Current (nA)")
plt.title(f"{gate} Pinchoff ({contact})")
plt.show()

Reading JSONL Files

JSON Lines are easier for quick inspection:

import json
# Read sweep data
sweeps = []
with open("data/characterization/20241003_143022/sweep.jsonl", "r") as f:
    for line in f:
        sweeps.append(json.loads(line))
# Access the data
first_sweep = sweeps[0]
print(f"Data name: {first_sweep['data_name']}")
print(f"Gate: {first_sweep['metadata']['gate']}")
print(f"Points: {len(first_sweep['x_data'])}")

Logging Custom Data

For measurements beyond simple sweeps, log custom data manually:

@routine
def find_pinchoff(ctx, gate, contact, session=None):
    """Find pinchoff and log the analysis."""
    device = ctx.resources.device
    # Perform sweep with auto-logging
    voltages = np.linspace(-2.0, 0.0, 100)
    v_data, i_data = device.sweep_1d(gate, voltages.tolist(), contact, session=session)
    # Analyze data
    threshold_idx = np.where(np.abs(i_data) < 1e-12)[0]
    pinchoff_voltage = v_data[threshold_idx[0]] if len(threshold_idx) > 0 else None
    # Manually log analysis results
    if session:
        session.log_analysis(
            f"{gate}_pinchoff_analysis",
            {
                "pinchoff_voltage": pinchoff_voltage,
                "threshold": 1e-12,
                "gate": gate,
                "contact": contact
            }
        )
    return {
        "pinchoff_voltage": pinchoff_voltage,
        "voltages": v_data,
        "currents": i_data
    }

The analysis results are saved alongside the sweep data.

Example: 2D Charge Stability Diagram

Here’s a complete example of measuring and logging a charge stability diagram:

@routine
def charge_stability(ctx, gate_x, gate_y, v_x_start, v_x_stop, n_x,
                      v_y_start, v_y_stop, n_y, contact, session=None):
    """Measure 2D charge stability diagram."""
    device = ctx.resources.device
    # Create voltage arrays
    v_x = np.linspace(v_x_start, v_x_stop, n_x)
    v_y = np.linspace(v_y_start, v_y_stop, n_y)
    # 2D sweep - automatically logs
    v_data, signal = device.sweep_2d(
        gate_x, v_x.tolist(),
        gate_y, v_y.tolist(),
        contact,
        session=session
    )
    # Analyze: find charge transitions
    grad_x = np.gradient(signal, axis=0)
    grad_y = np.gradient(signal, axis=1)
    edges = np.sqrt(grad_x**2 + grad_y**2)
    # Count number of visible charge transitions
    n_transitions = np.sum(edges > np.percentile(edges, 95))
    # Log analysis
    if session:
        session.log_analysis(
            "charge_stability_analysis",
            {
                "n_transitions": int(n_transitions),
                "max_gradient": float(np.max(edges)),
                "mean_signal": float(np.mean(signal)),
                "gates": [gate_x, gate_y]
            }
        )
    return {
        "voltages_x": v_data[0],
        "voltages_y": v_data[1],
        "signal": signal,
        "edges": edges,
        "n_transitions": n_transitions
    }

Run and analyze:

# Configure logger
logger = DataLogger(
    name="logger",
    routine_name="charge_diagrams",
    base_dir="./data",
    formats=["hdf5", "jsonl"],
    compression="gzip"  # Compress to save space
)
runner = RoutineRunner(configs=[config], logger=logger)
# Run measurement
result = runner.run("charge_stability")
print(f"Found {result['n_transitions']} charge transitions")
# Plot
plt.figure(figsize=(12, 5))
plt.subplot(1, 2, 1)
plt.pcolormesh(result["voltages_x"], result["voltages_y"],
               result["signal"].T, cmap="RdBu_r")
plt.xlabel("Left Plunger (V)")
plt.ylabel("Right Plunger (V)")
plt.title("Charge Stability")
plt.colorbar(label="Signal")
plt.subplot(1, 2, 2)
plt.pcolormesh(result["voltages_x"], result["voltages_y"],
               result["edges"].T, cmap="viridis")
plt.xlabel("Left Plunger (V)")
plt.ylabel("Right Plunger (V)")
plt.title("Charge Transitions")
plt.colorbar(label="Gradient")
plt.tight_layout()
plt.show()

Compression Options

For large datasets, enable compression:

logger = DataLogger(
    name="logger",
    routine_name="test",
    base_dir="./data",
    compression="gzip"  # Options: "gzip", "lzf", None
)

gzip: Best compression (slower) lzf: Fast compression (larger files) None: No compression (fastest)

For a 200×200 charge diagram:

  • Uncompressed: ~640 KB
  • LZF: ~250 KB
  • GZIP: ~180 KB

Working with Multiple Routines

Log data from multiple routines in the same session:

@routine
def full_characterization(ctx, session=None):
    """Run multiple measurements in one session."""
    device = ctx.resources.device
    # Barrier pinchoff
    v = np.linspace(-2.0, 0.0, 100)
    v_data, i_data = device.sweep_1d("LEFT_BARRIER", v.tolist(), "DRAIN", session=session)
    # Plunger sweep
    v = np.linspace(-2.5, -0.5, 150)
    v_data, i_data = device.sweep_1d("LEFT_PLUNGER", v.tolist(), "DRAIN", session=session)
    # Charge stability
    vx = np.linspace(-2.5, -0.5, 100)
    vy = np.linspace(-2.5, -0.5, 100)
    v_data, signal = device.sweep_2d(
        "LEFT_PLUNGER", vx.tolist(),
        "RIGHT_PLUNGER", vy.tolist(),
        "DRAIN",
        session=session
    )
    return {"status": "complete"}

All three measurements are saved in the same session directory.

Batch Processing Logged Data

Process multiple sessions at once:

import h5py
from pathlib import Path
import numpy as np
def analyze_all_sessions(routine_dir):
    """Analyze all logged sessions for a routine."""
    results = []
    # Find all session directories
    for session_dir in Path(routine_dir).iterdir():
        if not session_dir.is_dir():
            continue
        h5_file = session_dir / "session.h5"
        if not h5_file.exists():
            continue
        # Read sweep data
        with h5py.File(h5_file, "r") as f:
            for sweep_name in f["sweeps"].keys():
                voltages = f["sweeps"][sweep_name]["x_data"][:]
                currents = f["sweeps"][sweep_name]["y_data"][:]
                # Analyze
                max_current = np.max(np.abs(currents))
                results.append({
                    "session": session_dir.name,
                    "sweep": sweep_name,
                    "max_current": max_current
                })
    return results
# Analyze all barrier sweeps
results = analyze_all_sessions("./data/characterization")
for r in results:
    print(f"{r['session']}: {r['sweep']} -> {r['max_current']:.2e} A")

Tips and Best Practices

Use both HDF5 and JSONL: HDF5 for analysis scripts, JSONL for quick inspection.

Enable compression for large sweeps: A 200×200 charge diagram compresses 3-4x.

Log analysis results: Don’t just save raw data - save what you learned from it.

Use descriptive routine names: routine_name becomes your top-level directory. Use names like "device_42_characterization" rather than generic names like "test".

Add metadata to analysis: Include information that will help you interpret results later:

session.log_analysis(
    "pinchoff_analysis",
    data={"pinchoff_voltage": -0.85},
    metadata={
        "temperature": 10e-3,  # 10 mK
        "magnetic_field": 0.0,
        "operator": "alice",
        "notes": "After cooldown #15"
    }
)

Flush periodically for long measurements: Force data to disk during long routines:

@routine
def long_measurement(ctx, session=None):
    """Long measurement with periodic flushing."""
    device = ctx.resources.device
    for i in range(100):
        # Measure
        v_data, i_data = device.sweep_1d(...)
        # Flush every 10 sweeps
        if i % 10 == 0 and session:
            session.flush()
    return {"status": "complete"}

Check disk space: HDF5 files can get large. A full day of measurements might generate several GB.

Next Steps

  • Live Plotting - Visualize your logged data in real-time as routines execute
  • Routines - Learn how to write routines that log data automatically
  • Device Configuration - Configure your device for use with logging
  • Drivers - Understand how measurements are performed