Data Logging

1
from stanza.logger import DataLogger
2
from stanza.routines import RoutineRunner
3
from stanza.models import DeviceConfig
4
5
# 1. Create a logger
6
logger = DataLogger(
7
    name="logger",
8
    routine_name="characterization",
9
    base_dir="./data",
10
    formats=["hdf5", "jsonl"]
11
)
12
13
# 2. Pass it to RoutineRunner
14
config = DeviceConfig.from_yaml("device.yaml")
15
runner = RoutineRunner(configs=[config], logger=logger)
16
17
# 3. Run routines - data logs automatically
18
runner.run("charge_stability")

1
@routine
2
def barrier_sweep(ctx, gate, v_start, v_stop, n_points, contact, session=None):
3
    """Sweep a barrier gate."""
4
    device = ctx.resources.device
5
6
    voltages = np.linspace(v_start, v_stop, n_points)
7
8
    # Session parameter enables automatic logging
9
    v_data, i_data = device.sweep_1d(
10
        gate,
11
        voltages.tolist(),
12
        contact,
13
        session=session  # ← This enables logging
14
    )
15
16
    return {"voltages": v_data, "currents": i_data}

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

1
import h5py
2
import matplotlib.pyplot as plt
3
4
# Open the file
5
with h5py.File("data/characterization/20241003_143022/session.h5", "r") as f:
6
    # Read sweep data
7
    voltages = f["sweeps"]["LEFT_BARRIER_sweep_DRAIN"]["x_data"][:]
8
    currents = f["sweeps"]["LEFT_BARRIER_sweep_DRAIN"]["y_data"][:]
9
10
    # Read metadata
11
    gate = f["sweeps"]["LEFT_BARRIER_sweep_DRAIN"]["metadata"]["gate"][()].decode()
12
    contact = f["sweeps"]["LEFT_BARRIER_sweep_DRAIN"]["metadata"]["contact"][()].decode()
13
14
# Plot
15
plt.plot(voltages, currents * 1e9)  # Convert to nA
16
plt.xlabel("Voltage (V)")
17
plt.ylabel("Current (nA)")
18
plt.title(f"{gate} Pinchoff ({contact})")
19
plt.show()

1
import json
2
3
# Read sweep data
4
sweeps = []
5
with open("data/characterization/20241003_143022/sweep.jsonl", "r") as f:
6
    for line in f:
7
        sweeps.append(json.loads(line))
8
9
# Access the data
10
first_sweep = sweeps[0]
11
print(f"Data name: {first_sweep['data_name']}")
12
print(f"Gate: {first_sweep['metadata']['gate']}")
13
print(f"Points: {len(first_sweep['x_data'])}")

1
@routine
2
def find_pinchoff(ctx, gate, contact, session=None):
3
    """Find pinchoff and log the analysis."""
4
    device = ctx.resources.device
5
6
    # Perform sweep with auto-logging
7
    voltages = np.linspace(-2.0, 0.0, 100)
8
    v_data, i_data = device.sweep_1d(gate, voltages.tolist(), contact, session=session)
9
10
    # Analyze data
11
    threshold_idx = np.where(np.abs(i_data) < 1e-12)[0]
12
    pinchoff_voltage = v_data[threshold_idx[0]] if len(threshold_idx) > 0 else None
13
14
    # Manually log analysis results
15
    if session:
16
        session.log_analysis(
17
            f"{gate}_pinchoff_analysis",
18
            {
19
                "pinchoff_voltage": pinchoff_voltage,
20
                "threshold": 1e-12,
21
                "gate": gate,
22
                "contact": contact
23
            }
24
        )
25
26
    return {
27
        "pinchoff_voltage": pinchoff_voltage,
28
        "voltages": v_data,
29
        "currents": i_data
30
    }

1
@routine
2
def charge_stability(ctx, gate_x, gate_y, v_x_start, v_x_stop, n_x,
3
                      v_y_start, v_y_stop, n_y, contact, session=None):
4
    """Measure 2D charge stability diagram."""
5
    device = ctx.resources.device
6
7
    # Create voltage arrays
8
    v_x = np.linspace(v_x_start, v_x_stop, n_x)
9
    v_y = np.linspace(v_y_start, v_y_stop, n_y)
10
11
    # 2D sweep - automatically logs
12
    v_data, signal = device.sweep_2d(
13
        gate_x, v_x.tolist(),
14
        gate_y, v_y.tolist(),
15
        contact,
16
        session=session
17
    )
18
19
    # Analyze: find charge transitions
20
    grad_x = np.gradient(signal, axis=0)
21
    grad_y = np.gradient(signal, axis=1)
22
    edges = np.sqrt(grad_x**2 + grad_y**2)
23
24
    # Count number of visible charge transitions
25
    n_transitions = np.sum(edges > np.percentile(edges, 95))
26
27
    # Log analysis
28
    if session:
29
        session.log_analysis(
30
            "charge_stability_analysis",
31
            {
32
                "n_transitions": int(n_transitions),
33
                "max_gradient": float(np.max(edges)),
34
                "mean_signal": float(np.mean(signal)),
35
                "gates": [gate_x, gate_y]
36
            }
37
        )
38
39
    return {
40
        "voltages_x": v_data[0],
41
        "voltages_y": v_data[1],
42
        "signal": signal,
43
        "edges": edges,
44
        "n_transitions": n_transitions
45
    }

1
# Configure logger
2
logger = DataLogger(
3
    name="logger",
4
    routine_name="charge_diagrams",
5
    base_dir="./data",
6
    formats=["hdf5", "jsonl"],
7
    compression="gzip"  # Compress to save space
8
)
9
10
runner = RoutineRunner(configs=[config], logger=logger)
11
12
# Run measurement
13
result = runner.run("charge_stability")
14
15
print(f"Found {result['n_transitions']} charge transitions")
16
17
# Plot
18
plt.figure(figsize=(12, 5))
19
20
plt.subplot(1, 2, 1)
21
plt.pcolormesh(result["voltages_x"], result["voltages_y"],
22
               result["signal"].T, cmap="RdBu_r")
23
plt.xlabel("Left Plunger (V)")
24
plt.ylabel("Right Plunger (V)")
25
plt.title("Charge Stability")
26
plt.colorbar(label="Signal")
27
28
plt.subplot(1, 2, 2)
29
plt.pcolormesh(result["voltages_x"], result["voltages_y"],
30
               result["edges"].T, cmap="viridis")
31
plt.xlabel("Left Plunger (V)")
32
plt.ylabel("Right Plunger (V)")
33
plt.title("Charge Transitions")
34
plt.colorbar(label="Gradient")
35
36
plt.tight_layout()
37
plt.show()

1
logger = DataLogger(
2
    name="logger",
3
    routine_name="test",
4
    base_dir="./data",
5
    compression="gzip"  # Options: "gzip", "lzf", None
6
)

1
@routine
2
def full_characterization(ctx, session=None):
3
    """Run multiple measurements in one session."""
4
    device = ctx.resources.device
5
6
    # Barrier pinchoff
7
    v = np.linspace(-2.0, 0.0, 100)
8
    v_data, i_data = device.sweep_1d("LEFT_BARRIER", v.tolist(), "DRAIN", session=session)
9
10
    # Plunger sweep
11
    v = np.linspace(-2.5, -0.5, 150)
12
    v_data, i_data = device.sweep_1d("LEFT_PLUNGER", v.tolist(), "DRAIN", session=session)
13
14
    # Charge stability
15
    vx = np.linspace(-2.5, -0.5, 100)
16
    vy = np.linspace(-2.5, -0.5, 100)
17
    v_data, signal = device.sweep_2d(
18
        "LEFT_PLUNGER", vx.tolist(),
19
        "RIGHT_PLUNGER", vy.tolist(),
20
        "DRAIN",
21
        session=session
22
    )
23
24
    return {"status": "complete"}

1
import h5py
2
from pathlib import Path
3
import numpy as np
4
5
def analyze_all_sessions(routine_dir):
6
    """Analyze all logged sessions for a routine."""
7
    results = []
8
9
    # Find all session directories
10
    for session_dir in Path(routine_dir).iterdir():
11
        if not session_dir.is_dir():
12
            continue
13
14
        h5_file = session_dir / "session.h5"
15
        if not h5_file.exists():
16
            continue
17
18
        # Read sweep data
19
        with h5py.File(h5_file, "r") as f:
20
            for sweep_name in f["sweeps"].keys():
21
                voltages = f["sweeps"][sweep_name]["x_data"][:]
22
                currents = f["sweeps"][sweep_name]["y_data"][:]
23
24
                # Analyze
25
                max_current = np.max(np.abs(currents))
26
27
                results.append({
28
                    "session": session_dir.name,
29
                    "sweep": sweep_name,
30
                    "max_current": max_current
31
                })
32
33
    return results
34
35
# Analyze all barrier sweeps
36
results = analyze_all_sessions("./data/characterization")
37
38
for r in results:
39
    print(f"{r['session']}: {r['sweep']} -> {r['max_current']:.2e} A")

1
session.log_analysis(
2
    "pinchoff_analysis",
3
    data={"pinchoff_voltage": -0.85},
4
    metadata={
5
        "temperature": 10e-3,  # 10 mK
6
        "magnetic_field": 0.0,
7
        "operator": "alice",
8
        "notes": "After cooldown #15"
9
    }
10
)

1
@routine
2
def long_measurement(ctx, session=None):
3
    """Long measurement with periodic flushing."""
4
    device = ctx.resources.device
5
6
    for i in range(100):
7
        # Measure
8
        v_data, i_data = device.sweep_1d(...)
9
10
        # Flush every 10 sweeps
11
        if i % 10 == 0 and session:
12
            session.flush()
13
14
    return {"status": "complete"}