striqt.sensor

striqt.sensor.correct_iq(iq: AcquiredIQ, *, signal_trigger: Trigger | None = None, axis=1, overwrite_x=False) → AcquiredIQ

resample, filter, and apply calibration corrections.

Parameters:

  • iq – IQ dataclass output by a source

  • axis – the axis of x along which to compute the filter

  • overwrite_x – if True, modify the contents of IQ in-place; otherwise, a copy will be returned

Returns:

the filtered IQ waveform

striqt.sensor.iterate_sweep(resources: Resources[SS, SP, SC, PS, PC], *, always_yield: bool = False, yield_values: bool = True, loop: bool = False, **replace: Unpack[AnyResources[SS, SP, SC, PS, PC]]) → Generator[Dataset | DelayedDataset | None, None, None]

an iterator that steps through the execution of a sensor sweep.

Data acquisition, analysis, and sink operations each run in parallel in separate threads. Normally, the iterator yields a result for each of the N captures after it is handled by the sink:

` 0. `(Acquire 0)` 1. `Concurrent (Acquire 1, Analyze 0)`. 2. `Concurrent (Acquire 2, Analyze 1, Sink 0)` ➔ yield (`Result 0)`. (...) N. `Concurrent (Analyze N-1, Sink N-2)` ➔ yield `(Result N-2)`. N+1. `(Sink N-2)` ➔ yield `(Result N-1)`. `

With the always_yield argument, the iterator yields None to support status information in the 2 steps that do not yield anlaysis results:

` 0. `(Acquire 0)` ➔ yield `None` 1. `Concurrent (Acquire 1, Analyze 0)` ➔ yield `(Analysis 0)`. 2. `Concurrent (Acquire 2, Analyze 1, Sink 0)` ➔ yield `(Analysis 1)`. (...) N. `Concurrent (Analyze N-1, Sink N-2)` ➔ yield `(Analysis N-1)`. N+1. `(Sink N-2)` ➔ yield `None` `

The type of each yielded result depends on the return value of sink.append. To to minimize memory usage, the yield values can also be explicitly set to None with yield_values.

Parameters:

  • resources – dictionary of open resources returned by open_resources

  • always_yield – if True, yield None on steps that produce no analysis

  • always_values – if False, yield will return None instead of the sink result

  • loop – if True, the sweep will repeat at the beginning after last capture

Returns:

An iterator of analyzed data or None

striqt.sensor.open_resources(spec: specs.Sweep[SS, SP, SC], spec_path: str | Path | None = None, *, test_only: bool = False, on_source_opened: SourceOpenCallback | None = None) → ConnectionManager[SS, SP, SC, PS, PC]

open the sensor hardware and software contexts needed to run the given sweep.

The returned Connections object contains the resulting context. All of its resources are then open and ready to run the sweep.

striqt.sensor.open_store(spec: specs.Sink, *, format_path: specs.helpers.PathFormatter | None = None, force=False) → ZarrStore
striqt.sensor.read_calibration(path: None, format_path: specs.helpers.PathFormatter | None = None) → None
striqt.sensor.read_calibration(path: str | Path, format_path: specs.helpers.PathFormatter | None = None) → xr.Dataset
striqt.sensor.read_yaml_spec(path: str | Path, *, output_path: str | None = None, store_backend: str | None = None) → Sweep

Build a Sweep specification object from the specified yaml file.

Parameters:

  • path – path to the yaml file

  • output_path – optional override for the specification’s output path

  • store_backend – optional override for the specification’s output store backend

Returns:

an instance of specs.Sweep