striqt.sensor.sinks

class striqt.sensor.sinks.NoSink(sweep_spec: Sweep[Any, Any, SC], format_path: PathFormatter | None = None, *, force: bool = False)

Bases: SinkBase

append(capture_result) → DelayedDataset

close(*exc_info)

flush()

open()

wait()

class striqt.sensor.sinks.SinkBase(sweep_spec: Sweep[Any, Any, SC], format_path: PathFormatter | None = None, *, force: bool = False)

Bases: Generic[SC]

intake acquisitions one at a time, and parcel data store

append(capture_result: DelayedDataset) → Dataset | DelayedDataset

close(*exc_info)

flush()

open() → None

pop() → list[Dataset]

submit(func, *args, **kws)

wait()

class striqt.sensor.sinks.ZarrCaptureSink(sweep_spec: Sweep[Any, Any, SC], format_path: PathFormatter | None = None, *, force: bool = False)

Bases: ZarrSinkBase

concatenates the data from each capture and dumps to a zarr data store

append(capture_result: DelayedDataset)

flush()

class striqt.sensor.sinks.ZarrSinkBase(sweep_spec: Sweep[Any, Any, SC], format_path: PathFormatter | None = None, *, force: bool = False)

Bases: SinkBase

close(*exc_info)

get_root_path() → str

open()

class striqt.sensor.sinks.ZarrTimeAppendSink(sweep_spec: Sweep, format_path: PathFormatter | None = None, *, force: bool = False)

Bases: ZarrSinkBase

append(capture_result)

flush()