Examples¶

All examples are also available as runnable Jupyter notebooks in examples/notebooks/.

Open with Jupyter:

uv sync --all-groups
uv run --active jupyter lab

00 — Settings: create and load¶

from pathlib import Path
import json
import os

from deteqt import load_settings, write_settings_file


def _find_settings(name: str = "settings.toml") -> Path:
    for parent in [Path.cwd(), *Path.cwd().parents]:
        candidate = parent / name
        if candidate.exists():
            return candidate
    return Path.cwd() / name


SETTINGS_PATH = _find_settings()
URL = "https://influx.yourdomain.com"
ORG = "YOUR_ORG"
BUCKET = "YOUR_BUCKET"
TOKEN = os.getenv("DETEQT_TOKEN", "REPLACE_ME")
TIMEOUT_MS = 30_000
DEFAULT_TAGS = {"device_ID": "chip-a", "condition": "4K"}
OVERWRITE = False

written_path = write_settings_file(
    url=URL,
    org=ORG,
    bucket=BUCKET,
    token=TOKEN,
    timeout_ms=TIMEOUT_MS,
    path=SETTINGS_PATH,
    default_tags=DEFAULT_TAGS,
    overwrite=OVERWRITE,
)
print(f"settings written: {written_path}")

settings = load_settings(SETTINGS_PATH)
print(
    json.dumps(
        {
            "url": settings.url,
            "org": settings.org,
            "bucket": settings.bucket,
            "timeout_ms": settings.timeout_ms,
            "default_tags": settings.default_tags,
            "token_loaded": bool(settings.token),
        },
        indent=2,
        sort_keys=True,
    )
)

01 — Write one point¶

from pathlib import Path
import json

from deteqt import InfluxWriteError, TUID, write_point


def _find_settings(name: str = "settings.toml") -> Path:
    for parent in [Path.cwd(), *Path.cwd().parents]:
        candidate = parent / name
        if candidate.exists():
            return candidate
    return Path.cwd() / name


SETTINGS_PATH = _find_settings()
record = {
    "qoi": "qubit_01_frequency",
    "nominal_value": 5.010,
    "uncertainty": 0.005,
    "tuid": TUID("20260213-143319-249-01d39d"),
    "element": "q1",
    "label": "f01",
    "unit": "GHz",
    "element_label": "qubit-01",
    "device_ID": "chip-a",
    "run_ID": "run-001",
    "cycle_ID": "cycle-01",
    "condition": "4K",
    "extra_tags": {
        "area_of_interest": "sweetspot",
        "long_label": "Q1 frequency",
    },
    "extra_fields": {
        "temperature_mK": 12.3,
        "passed_qc": True,
    },
}

try:
    summary = write_point(record, settings_path=SETTINGS_PATH)
    print(json.dumps(summary, indent=2, sort_keys=True))
except InfluxWriteError as exc:
    print(f"Influx write failed: {exc}")

02 — Write batch¶

from pathlib import Path
import json

from deteqt import InfluxWriteError, write_batch


def _find_settings(name: str = "settings.toml") -> Path:
    for parent in [Path.cwd(), *Path.cwd().parents]:
        candidate = parent / name
        if candidate.exists():
            return candidate
    return Path.cwd() / name


SETTINGS_PATH = _find_settings()
records = [
    {
        "qoi": "qubit_01_frequency",
        "nominal_value": 5.01,
        "uncertainty": 0.01,
        "element": "q1",
        "run_ID": "run-001",
        "cycle_ID": "cycle-01",
        "extra_tags": {"sample_id": "S-42"},
        "extra_fields": {"temperature_mK": 12.3},
    },
    {
        "qoi": "t1",
        "nominal_value": 23.4,
        "uncertainty": 0.8,
        "element": "q1",
        "run_ID": "run-001",
        "cycle_ID": "cycle-01",
        "extra_tags": {"sample_id": "S-42"},
        "extra_fields": {"passed_qc": True},
    },
    {
        "qoi": "resonator_qi",
        "nominal_value": 14500.0,
        "uncertainty": 120.0,
        "element": "q1",
        "run_ID": "run-001",
        "cycle_ID": "cycle-01",
    },
]

try:
    summary = write_batch(
        records,
        settings_path=SETTINGS_PATH,
        continue_on_error=True,
        chunk_size=2,
        return_errors=True,
    )
    print(json.dumps(summary, indent=2, sort_keys=True))
except InfluxWriteError as exc:
    print(f"Influx write failed: {exc}")

03 — Preview recursive ingest matches¶

from pathlib import Path
import json

from deteqt import debug_ingest_matches

FOLDER = Path("/path/to/quantify-data")
plan = debug_ingest_matches(FOLDER)
print(
    json.dumps(
        {
            "pairings_total": len(plan),
            "sample_pairings": plan[:10],
        },
        indent=2,
        sort_keys=True,
    )
)

04 — Manual pipeline: ingest → process → write¶

from pathlib import Path
import json

from deteqt import InfluxWriteError, ingest_data, process_data, write_batch


def _find_settings(name: str = "settings.toml") -> Path:
    for parent in [Path.cwd(), *Path.cwd().parents]:
        candidate = parent / name
        if candidate.exists():
            return candidate
    return Path.cwd() / name


FOLDER = Path("/path/to/quantify-data")
SETTINGS_PATH = _find_settings()
MAX_RECORDS = 200
CONTINUE_ON_ERROR = True

raw_data = ingest_data(FOLDER, continue_on_error=CONTINUE_ON_ERROR)
processed_data = process_data(raw_data, continue_on_error=CONTINUE_ON_ERROR)
processed_for_write = (
    processed_data[:MAX_RECORDS] if MAX_RECORDS > 0 else processed_data
)
print(
    json.dumps(
        {
            "raw_records_total": len(raw_data),
            "processed_records_total": len(processed_data),
            "records_selected_for_write": len(processed_for_write),
        },
        indent=2,
        sort_keys=True,
    )
)

try:
    summary = write_batch(
        processed_for_write,
        settings_path=SETTINGS_PATH,
        continue_on_error=CONTINUE_ON_ERROR,
        return_errors=True,
    )
    print(json.dumps(summary, indent=2, sort_keys=True))
except InfluxWriteError as exc:
    print(f"Influx write failed: {exc}")

05 — One-call ingest and write¶

from pathlib import Path
import json

from deteqt import InfluxWriteError, ingest_and_write


def _find_settings(name: str = "settings.toml") -> Path:
    for parent in [Path.cwd(), *Path.cwd().parents]:
        candidate = parent / name
        if candidate.exists():
            return candidate
    return Path.cwd() / name


FOLDER = Path("/path/to/quantify-data")
SETTINGS_PATH = _find_settings()

try:
    summary = ingest_and_write(
        folder=FOLDER,
        settings_path=SETTINGS_PATH,
        continue_on_error=True,
    )
    print(json.dumps(summary, indent=2, sort_keys=True))
except InfluxWriteError as exc:
    print(f"Influx write failed: {exc}")

06 — Build a Flux query¶

from deteqt import build_query

query = build_query(
    bucket="qoi",
    start="-7d",
    measurement="qubit_01_frequency",
    fields=("nominal_value", "uncertainty"),
    tags={"run_ID": "run-001"},
    limit=200,
    sort_desc=True,
)
print(query)

Output:

from(bucket: "qoi")
  |> range(start: -7d)
  |> filter(fn: (r) => r._measurement == "qubit_01_frequency")
  |> filter(fn: (r) => r._field == "nominal_value" or r._field == "uncertainty")
  |> filter(fn: (r) => r.run_ID == "run-001")
  |> sort(columns: ["_time"], desc: true)
  |> limit(n: 200)

07 — Structured queries¶

from pathlib import Path
import json

from deteqt import query_rows, query_rows_with_meta, query_stream


def _find_settings(name: str = "settings.toml") -> Path:
    for parent in [Path.cwd(), *Path.cwd().parents]:
        candidate = parent / name
        if candidate.exists():
            return candidate
    return Path.cwd() / name


SETTINGS_PATH = _find_settings()
START = "-7d"
MEASUREMENT = "qubit_01_frequency"
RUN_ID = "run-001"
LIMIT = 50

rows = query_rows(
    settings_path=SETTINGS_PATH,
    start=START,
    measurement=MEASUREMENT,
    fields=("nominal_value", "uncertainty"),
    tags={"run_ID": RUN_ID},
    limit=LIMIT,
    sort_desc=True,
)
rows_with_meta, meta = query_rows_with_meta(
    settings_path=SETTINGS_PATH,
    start=START,
    measurement=MEASUREMENT,
    fields=("nominal_value", "uncertainty"),
    tags={"run_ID": RUN_ID},
    limit=LIMIT,
    sort_desc=True,
)
stream_preview: list[dict[str, object]] = []
for idx, row in enumerate(
    query_stream(
        settings_path=SETTINGS_PATH,
        start=START,
        measurement=MEASUREMENT,
        fields=("nominal_value",),
        tags={"run_ID": RUN_ID},
        limit=LIMIT,
        sort_desc=True,
    )
):
    if idx >= 3:
        break
    stream_preview.append(row)

print(
    json.dumps(
        {
            "rows_count": len(rows),
            "rows_with_meta_count": len(rows_with_meta),
            "meta": {
                "elapsed_ms": meta.elapsed_ms,
                "row_count": meta.row_count,
                "query_sha256": meta.query_sha256,
            },
            "stream_preview": stream_preview,
        },
        indent=2,
        sort_keys=True,
        default=str,
    )
)

08 — Raw Flux queries¶

from pathlib import Path
import json

from deteqt import (
    build_query,
    load_settings,
    query_flux_rows,
    query_flux_rows_with_meta,
    query_flux_stream,
)


def _find_settings(name: str = "settings.toml") -> Path:
    for parent in [Path.cwd(), *Path.cwd().parents]:
        candidate = parent / name
        if candidate.exists():
            return candidate
    return Path.cwd() / name


SETTINGS_PATH = _find_settings()
query = build_query(
    bucket=load_settings(SETTINGS_PATH).bucket,
    start="-7d",
    measurement="qubit_01_frequency",
    fields=("nominal_value", "uncertainty"),
    tags={"run_ID": "run-001"},
    limit=50,
    sort_desc=True,
)
print(query)

rows = query_flux_rows(query=query, settings_path=SETTINGS_PATH)
rows_with_meta, meta = query_flux_rows_with_meta(
    query=query,
    settings_path=SETTINGS_PATH,
)
stream_preview: list[dict[str, object]] = []
for idx, row in enumerate(
    query_flux_stream(query=query, settings_path=SETTINGS_PATH)
):
    if idx >= 3:
        break
    stream_preview.append(row)

print(
    json.dumps(
        {
            "rows_count": len(rows),
            "rows_with_meta_count": len(rows_with_meta),
            "meta": {
                "elapsed_ms": meta.elapsed_ms,
                "row_count": meta.row_count,
                "query_sha256": meta.query_sha256,
            },
            "stream_preview": stream_preview,
        },
        indent=2,
        sort_keys=True,
        default=str,
    )
)

09 — Load one JSON file into records¶

from pathlib import Path
import json

from deteqt.ingest_data import load_json_file_records

FILE_PATH = Path("/path/to/quantities_of_interest.json")
records = load_json_file_records(FILE_PATH)
print(
    json.dumps(
        {
            "records_total": len(records),
            "preview": records[:2],
        },
        indent=2,
        sort_keys=True,
        default=str,
    )
)