Qimchi Connect

qimchi-connect publishes real-time xarray.Dataset snapshots to Qimchi without requiring a specific measurement framework or storage format.

To publish a measurement, provide a callback that returns the current xarray.Dataset. Built-in providers support QCoDeS, qanary, and Quantify.

---

• Qimchi Connect
  • Installation
  • Publishing a measurement
  • Supported frameworks
  • Runnable examples
  • Repairing discovery records
  • Runtime behavior
  • Development
  • License

---

Installation

Install the latest release from PyPI:

pip install qimchi-connect

For a project managed with uv:

uv add qimchi-connect

To install the latest development version from the preview branch:

pip install "qimchi-connect @ git+https://gitlab.com/squad-lab/qimchi-connect.git@preview"

For a project managed with uv:

uv add "qimchi-connect @ git+https://gitlab.com/squad-lab/qimchi-connect.git@preview"

Qimchi Connect requires Python 3.13 or later. Python 3.13 is recommended, and CI also tests Python 3.14.

Optional extras are listed below. For example, install the Quantify extra with pip install "qimchi-connect[quantify]".

Extra	Adds	For
examples	qcodes	QCoDeSSnapshotProvider and examples/qcodes_measurement.py
quantify	quantify-core	QuantifySnapshotProvider and examples/quantify_measurement.py
dev	pre-commit, ruff, pytest	Development
test	pytest, pytest-cov	Tests

Publishing a measurement

The snapshot callback is evaluated when a client requests data. It should return a consistent view of the measurement and use the producer's own lock when the underlying data can change concurrently. The server takes ownership of each returned dataset, materializes it, and closes its backing resources; return a snapshot rather than the producer's owned dataset.

from qimchi_connect import live_measurement

with live_measurement(
    measurement_id="experiment-42",
    snapshot=lambda: current_dataset.copy(deep=True),
    disk_path="data/experiment-42.nc",
):
    perform_measurement()

Following a measurement incrementally

For a grid that grows along one shared leading dimension, clients can avoid downloading the rows they already hold. The first response is a full snapshot; its row metadata supplies the offset for the next request:

from qimchi_connect import open_live_measurement_sync

snapshot = open_live_measurement_sync("experiment-42", ws_url)
progress = snapshot.encoding.get("qimchi_connect_rows")

if progress is not None:
    delta = open_live_measurement_sync(
        "experiment-42",
        ws_url,
        since_rows=progress["rows_written"],
    )

The returned delta contains only the missing rows. Its encoding["qimchi_connect_rows"] records the append dimension, the first row returned, the current written frontier, and the preallocated total. The client does not merge the delta automatically. Older servers ignore since_rows, and datasets without a common leading dimension continue to return full snapshots.

Supported frameworks

Pass a framework provider to live_measurement instead of a callback. A provider can prepare the first snapshot, set source_package, and derive the measurement ID and disk path.

QCoDeSSnapshotProvider manages the cache, locking, write flushing, and refresh rate:

# QCoDeS
from qimchi_connect import QCoDeSSnapshotProvider, live_measurement

with meas.run() as datasaver:
    live = QCoDeSSnapshotProvider(datasaver)
    with live_measurement(f"qcodes-{datasaver.dataset.captured_run_id}", live):
        for voltage in voltages:
            dac.ch1.set(voltage)
            live.add_result((dac.ch1, voltage), (dmm.v1, dmm.v1.get()))

The provider prepares a snapshot before registration and sets the run's source_package. See examples/qcodes_measurement.py for a complete example.

QuantifySnapshotProvider derives the measurement ID and disk path from a tuid. It retries when dataset.hdf5 is not yet readable at the start of a run. Install it with pip install "qimchi-connect[quantify]".

# Quantify
from qimchi_connect import QuantifySnapshotProvider, live_measurement

live = QuantifySnapshotProvider(tuid)
with live_measurement(live.measurement_id, live, disk_path=live.disk_path):
    MC.run(experiment_name)

QanarySnapshotProvider reads the in-memory Zarr store used by a qanary sweep:

# Qanary
from qimchi_connect import QanarySnapshotProvider, register_live_measurement

register_live_measurement(measurement_id, QanarySnapshotProvider(memory_store))

Qanary was called qcutils up to and including qimchi-connect 0.1.0, where this provider was named QCUtilsSnapshotProvider. That name was removed. Please import QanarySnapshotProvider instead.

See Adding a framework to support another framework.

Runnable examples

The examples do not require hardware. All three call qimchi_connect.live_measurement; only the snapshot argument differs.

• examples/qcodes_measurement.py uses QCoDeSSnapshotProvider with a standard QCoDeS Measurement and datasaver. The provider manages the snapshot cache, locking, write flushing, and refresh rate. Install its dependency with pip install qimchi-connect[examples].
• examples/quantify_measurement.py uses QuantifySnapshotProvider with a two-dimensional Quantify MeasurementControl sweep. A discovery thread publishes the TUID that Quantify creates inside run(). Install its dependency with pip install "qimchi-connect[quantify]".
• examples/generic_producer.py uses an acquisition loop that updates a locked NumPy array and exposes it as an xarray.Dataset. Use this pattern for frameworks without a provider.

tests/test_examples.py runs all three scripts. When adding an example, add a test for it in that file. test_every_example_is_covered_by_a_test checks this requirement.

There is no separate qanary script in this directory because qanary already includes higher-level examples based on qanary.measure.run and Sweep.

Repairing discovery records

Registration checks records marked as live. It marks a record as ended when the endpoint remains unavailable after retries or no longer advertises the recorded dataset. It then deletes ended records older than seven days.

Run the same maintenance manually with:

qimchi-connect cleanup --retention-days 7

The discovery database is stored in Qimchi's application directory -- $QIMCHI_HOME/live_measurements.db, or ~/.qimchi/live_measurements.db by default. Every producer publishes into the one file Qimchi reads, so it belongs with the viewer rather than under any single measurement package.

Up to 0.1.0 it lived at ~/.qcutils/live_measurements.db. Nothing is migrated: the registry only holds live and recently-ended runs and repopulates as soon as producers publish again, so upgrade your producers and Qimchi together and delete the old ~/.qcutils afterwards.

Runtime behavior

One server per process. The WebSocket server is process-wide. The first live_measurement call starts it. Later calls use the same server. The host and port arguments apply only to the call that starts the server. Each registration advertises the active server.

Snapshots are read-only. Numeric and boolean variables are transferred as raw bytes and received as read-only NumPy arrays. Call .copy() before modifying a variable.

The callback runs on the server thread. It runs when a client requests data. Use a lock and return a copy instead of a live buffer. See examples/generic_producer.py. Use QCoDeSSnapshotProvider for QCoDeS.

Snapshot caching is configurable. Responses for the same measurement share one serialized snapshot for 0.25 seconds by default. Set QIMCHI_CONNECT_SNAPSHOT_TTL to a non-negative number of seconds to tune the freshness/serialization tradeoff. Zero disables cache hits. Invalid values are ignored with a warning and use the default.

Development

uv sync --extra dev --extra test --extra examples
uv run pre-commit install     # once per clone
uv run pytest

ruff is pinned to the same version in pyproject.toml, .pre-commit-config.yaml and the CI lint job. Bump all three together.

See CONTRIBUTING.md for branch and release rules and for instructions on adding framework support.

License

AGPL-3.0-or-later. See LICENSE.