Usage¶

Logging¶

deteqt uses Python's standard logging module. Add this once at the top of your script or notebook to see progress and warnings in your terminal:

import logging
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")

You will then see lines like:

INFO  ingest_data: scanning '/data/20250710'.
INFO  ingest_data: loaded 312 raw records.
INFO  process_data: 308/312 records ready.
INFO  ingest_and_write: 312 raw → 308 processed records; writing now.
INFO  write_batch: starting (chunk_size=500).
INFO  write_batch: done — 308/308 records written.
INFO  ingest_and_write: complete — 308 records written.

If any records fail you get a WARNING instead of INFO on the summary line. Use level=logging.WARNING to see only failures and nothing else.

0) Create settings file¶

from deteqt import write_settings_file

settings_path = write_settings_file(
    url="https://influx.yourdomain.com",
    org="YOUR_ORG",
    bucket="YOUR_BUCKET",
    token="YOUR_WRITE_TOKEN",
    timeout_ms=30000,  # optional
)
print(settings_path)  # settings.toml

1) Write one point¶

from deteqt import TUID, write_point

run_tuid: TUID = TUID("20260213-143319-249-01d39d")

summary = write_point(
    {
        "qoi": "frequency",
        "nominal_value": 1.01,
        "uncertainty": 0.01,
        "tuid": run_tuid,
        "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": {
            "sample_id": "S-42",
            "operator": "alice",
            "area_of_interest": "sweetspot",
            "long_label": "Q1 frequency",
        },
        "extra_fields": {"temperature_mK": 12.3, "passed_qc": True},
    }
)
print(
    summary
)  # {'records_total': 1, 'records_written': 1, 'records_failed': 0}

2) Write a batch¶

from deteqt import write_batch

summary = write_batch(
    [
        {
            "qoi": "frequency",
            "nominal_value": 1.01,
            "uncertainty": 0.01,
            "run_ID": "run-001",
            "cycle_ID": "cycle-01",
            "extra_tags": {"sample_id": "S-42"},
            "extra_fields": {"temperature_mK": 12.3},
        },
        {
            "qoi": "phase",
            "nominal_value": 0.12,
            "run_ID": "run-001",
            "extra_tags": {
                "sample_id": "S-42",
                "area_of_interest": "sweetspot",
            },
            "extra_fields": {"passed_qc": True},
        },
    ]
)

Batch tuning:

chunk_size=500 by default.
Use continue_on_error=True, return_errors=True to keep writing and receive per-record failures in summary["errors"].
On uncertain chunk API failures, records are marked failed without per-record retry to avoid duplicate writes.

Common schema note:

Use common keys for shared entities: qoi, nominal_value, uncertainty, tuid, element, label, unit, element_label, device_ID, run_ID, cycle_ID, condition.
Optional timestamp key: time accepts an RFC3339 string, a datetime object, or a unix epoch number. When omitted, InfluxDB uses server time. Set an explicit time to make writes idempotent — re-sending the same record with the same timestamp overwrites the point instead of duplicating it. Example:

from datetime import datetime, timezone

write_point({
    "qoi": "frequency",
    "nominal_value": 5.01,
    "tuid": "20260213-143319-249-01d39d",
    "time": datetime(2026, 2, 13, 14, 33, 19, tzinfo=timezone.utc),
})

area_of_interest and long_label remain optional extra tags.

3) Ingest recursively (JSON)¶

from deteqt import ingest_data, process_data, write_batch

raw_data = ingest_data(
    folder="quantify-data",
)

processed_data = process_data(raw_data)

summary = write_batch(processed_data)

One-call pipeline:

from deteqt import ingest_and_write

summary = ingest_and_write(
    folder="quantify-data",
    # scans snapshot/QoI JSON recursively.
    # Typically: quantities_of_interest is plain .json,
    # while snapshot is often compressed (.json.xz).
    # When both exist in a run, values come from QoI JSON and
    # metadata/tags come from snapshot JSON.
)

Inspect snapshot/QoI pairing before writing:

from deteqt import debug_ingest_matches

debug_ingest_matches("quantify-data")
# run-1/analysis/quantities_of_interest.json -> run-1/metadata/snapshot.json

Notes:

Hybrid mode uses quantities_of_interest*.json values first and enriches each record with schema metadata from snapshot*.json for the same run.
In most datasets, quantities_of_interest*.json is plain JSON and snapshot*.json is compressed (often .json.xz).
Snapshot file support includes plain .json plus .json.xz, .json.gz, and .json.bz2.
Only standard SCQT transmon QoIs are kept in hybrid mode.
Folders named analysis_BasicAnalysis (case-insensitive) and .ipynb_checkpoints are ignored.
Analysis aliases are normalized to standard names, for example: T1 -> t1, Qi -> resonator_qi, Qc -> resonator_qc, fr -> resonator_freq_low and resonator_freq_high.
If hybrid merge returns no records for a run, that run falls back to snapshot-only parsing.
If dataset TUID does not match snapshot parameter TUIDs, the run-folder element hint (for example ... q2) is used to select metadata.
For quantities_of_interest*.json fallback files, tuid is inferred from a Quantify run folder prefix like 20260213-143319-249-01d39d-* when present.

4) Build and run customizable queries¶

from deteqt import (
    build_query,
    query_flux_rows,
    query_flux_rows_with_meta,
    query_flux_stream,
    query_rows,
    query_rows_with_meta,
    query_stream,
)

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

rows = query_rows(
    start="-7d",
    measurement="frequency",
    fields=("nominal_value", "uncertainty"),
    tags={"run_ID": "run-001"},
    limit=200,
    sort_desc=True,
)

rows_from_flux = query_flux_rows(query=query)

for row in query_flux_stream(query=query):
    print(row["_measurement"])

for row in query_stream(
    start="-7d",
    measurement="frequency",
    fields=("nominal_value",),
    tags={"run_ID": "run-001"},
):
    print(row["_value"])

rows, meta = query_rows_with_meta(
    start="-7d",
    measurement="frequency",
    fields=("nominal_value", "uncertainty"),
    tags={"run_ID": "run-001"},
)
print(
    meta
)  # QueryResultMeta(elapsed_ms=..., row_count=..., query_sha256="...")

rows_from_flux_with_meta, flux_meta = query_flux_rows_with_meta(query=query)
print(
    flux_meta
)  # QueryResultMeta(elapsed_ms=..., row_count=..., query_sha256="...")

Fast troubleshooting¶

Settings file not found: create settings.toml in current working directory or pass settings_path=....
Write rejected (401 unauthorized): token/org/bucket/url do not match, or token is not write-capable.
Unknown record keys: use standard keys only and put custom metadata in extra_tags / extra_fields.
Record must include required field 'nominal_value': provide a numeric nominal_value in each record.