Batched sensor acquisition

striqt.sensor

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

run the iterate of resources['runner'] over resources.

Parameters:

  • resources – the open resources of open_resources

  • always_yield – with yield_values False, True yields None at every step and False yields nothing

  • yield_values – True yields the sink result of every step

  • loop – True cycles through the captures from the first after the last

  • **replace – entries that replace those of resources for this run

Yields:

the return value of sink.append for each capture, or None at the steps before the first sink write; with yield_values False, None at every step or nothing as always_yield selects. The iteration never ends when loop is True.

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

open the sensor hardware and software contexts needed to run spec.

The source, peripherals, sink, calibration file, and log file open concurrently for the sweep of the action runner built from spec. The compute warmup runs in the calling thread. The returned ConnectionManager yields the open Resources on entry and closes every resource on exit. An exception from a hook ends the run. Progress is logged at INFO on the sweep logger.

Parameters:

  • spec – the action spec

  • spec_path – the file from which spec was read, or None; sets the process working directory and the {spec_path} path field

  • test_only – if True, open NoPeripherals in place of the peripherals and skip the compute warmup

  • on_source_opened – hook run once with (sweep, source_id) in a worker thread after the source opens and before the first capture

  • on_analyzed – hook run in the analysis thread with the DelayedDataset of each capture, before the sink receives the dataset

  • on_sink_cached – hook run in the analysis thread with (resources, cache, capture, result, **kwargs) each time a measurement fills a KwArgCache

Raises:

  • TypeError – if no action runner is registered for the type of spec

  • KeyError – if no sensor binding is registered for the type of spec

  • ExceptionGroup – if more than one resource fails to open; a single failure raises that exception

striqt.sensor.open_store(spec: specs.ZarrSink, *, format_path: specs.paths.PathFormatter | None = None, force=False) → ZarrStore

open the zarr store at spec.path for appending.

The suffix of the path is set to .zarr for a directory store and .zarr.zip for a zip store. Missing parent directories are created.

Parameters:

  • spec – the sink block with the path and store type

  • format_path – expands the {fields} of spec.path; None leaves the path as is

  • force – if True, truncate an existing store; if False, append

Raises:

ValueError – if spec.path is None

striqt.sensor.read_calibration(path: None, format_path: specs.paths.PathFormatter | None = None) → None
striqt.sensor.read_calibration(path: str | Path, format_path: specs.paths.PathFormatter | None = None) → xr.Dataset
striqt.sensor.read_json_spec(path: str | Path, *, type: type[ActionSpec] | None = None, output_path: str | None = None, store_backend: str | None = None) → ActionSpec

decode the action spec in the json file at path.

The file has the schema of a yaml spec without !include directives. The extensions block of the file is imported before decoding. A relative extensions.import_path is resolved against the directory of the file.

Parameters:

  • path – the json file

  • type – the action spec type to decode into, or None to select the type bound under the sensor_binding tag

  • output_path – replaces sink.path in the returned spec when given

  • store_backend – replaces sink.store of a Sweep spec when given

Returns:

an instance of type, or of the bound class when type is None

Raises:

  • msgspec.ValidationError – if the file does not validate as the spec class, or if the tag names no registered binding

  • TypeError – if store_backend is given and the sink of the spec is not a ZarrSink

  • FileNotFoundError – if extensions.import_path does not exist

striqt.sensor.read_yaml_spec(path: str | Path, *, type: type[ActionSpec] | None = None, output_path: str | None = None, store_backend: str | None = None) → ActionSpec

decode the action spec in the yaml file at path.

The extensions block of the file is imported before decoding. A relative extensions.import_path is resolved against the directory of the file.

Parameters:

  • path – the yaml file

  • type – the action spec type to decode into, or None to select the type bound under the sensor_binding tag

  • output_path – replaces sink.path in the returned spec when given

  • store_backend – replaces sink.store of a Sweep spec when given

Returns:

an instance of type, or of the bound class when type is None

Raises:

  • msgspec.ValidationError – if the file does not validate as the spec class, or if the tag names no registered binding

  • TypeError – if store_backend is given and the sink of the spec is not a ZarrSink

  • FileNotFoundError – if extensions.import_path does not exist

striqt.sensor.read_zarr_spec(path: str | Path, *, type: type[ActionSpec] | None = None, output_path: str | None = None, store_backend: str | None = None, extension_root: str | Path | None = None) → ActionSpec

decode the sweep spec saved in the attrs of the zarr store at path.

The extensions block of the attrs is imported before decoding. Only the attrs that name fields of the sweep class, the sensor_binding tag, and mock_source are read. An extensions.sink attr is dropped with a DeprecationWarning.

Parameters:

  • path – the zarr store, a .zarr directory or .zarr.zip archive

  • type – the sweep class to decode against, or None to select the class bound under the saved sensor_binding tag; a saved mock_source attr selects the mock class of that binding instead

  • output_path – replaces sink.path in the returned spec when given

  • store_backend – replaces sink.store of a Sweep spec when given

  • extension_root – the directory against which a relative extensions.import_path is resolved, or None for the directory of the store

Returns:

an instance of type, or of the bound sweep class when type is None

Raises:

  • msgspec.ValidationError – if the attrs do not validate as the sweep class, or if the tag names no registered binding

  • TypeError – if store_backend is given and the sink of the spec is not a ZarrSink

  • FileNotFoundError – if extensions.import_path or extension_root does not exist

striqt.sensor.bindings

the built-in sensor bindings.

Each binding pairs a source backend with the spec types that configure the backend. The sensor_binding key of a sweep file selects a binding by name.

—

class striqt.sensor.bindings.air7101b(*, master_clock_rate=125000000.0, trigger_strobe=None, signal_trigger=None, array_backend='cupy', calibration=None, time_source='host', time_sync_at='acquire', clock_source='internal', receive_retries=3, adc_overload_limit=-1, if_overload_limit=None, gapless=False)

Bases: Controller

Open and control the backend striqt.sensor.lib.sources.deepwave.Airstack1Source.

Parameters:

  • master_clock_rate (float): Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None):

  • signal_trigger (str | BuiltinTriggers | None):

  • array_backend (one of [‘numpy’, ‘cupy’]): array module to use to set compute device: numpy = cpu, cupy = gpu

  • calibration (str | None):

  • time_source (one of [‘host’, ‘internal’, ‘external’, ‘gps’]): Hardware source for timestamps

  • time_sync_at (one of [‘open’, ‘acquire’]): when to sync the hardware clock: on connection, or before each capture

  • clock_source (one of [‘internal’, ‘external’, ‘gps’]): Hardware source for the frequency reference

  • receive_retries (int): number of acquisition retry attempts on stream error (Constraints: >= 0)

  • adc_overload_limit (float | None): dataset adc_overload=True when the peak ADC level exceeds this threshold (Constraints: <= 0, units: dBfs)

  • if_overload_limit (float | None): the maximum level at maximum gain before IF/baseband overload (Constraints: units: dBfs)

  • gapless (bool): whether to raise an exception on overflows between identical captures

arm(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis=<factory>, external_lo_frequency=None, center_frequency, gain)
Parameters:

  • duration (float): Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float): Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float): Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, …]): Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of [‘left’, ‘right’, ‘none’]): LO shift direction

  • host_resample (bool):

  • backend_sample_rate (float | None): Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]): changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, …]): LO frequency of external frequency converter

  • center_frequency (float | tuple[float, …]): Center frequency for each port (Constraints: units: Hz, > 0, units: Hz, > 0, units: Hz)

  • gain (float | tuple[float, …]): Gain setting for each channel (Constraints: units: dB, units: dB, units: dB)

acquire(overlaps=None)

acquire IQ samples needed for the armed capture.

—

class striqt.sensor.bindings.air7101b_calibration(*, master_clock_rate=125000000.0, trigger_strobe=None, signal_trigger=None, array_backend='cupy', calibration=None, time_source='host', time_sync_at='acquire', clock_source='internal', receive_retries=3, adc_overload_limit=-1, if_overload_limit=None, gapless=False)

Bases: Controller

Open and control the backend striqt.sensor.lib.sources.deepwave.Airstack1Source.

Parameters:

  • master_clock_rate (float): Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None):

  • signal_trigger (str | BuiltinTriggers | None):

  • array_backend (one of [‘numpy’, ‘cupy’]): array module to use to set compute device: numpy = cpu, cupy = gpu

  • calibration (str | None):

  • time_source (one of [‘host’, ‘internal’, ‘external’, ‘gps’]): Hardware source for timestamps

  • time_sync_at (one of [‘open’, ‘acquire’]): when to sync the hardware clock: on connection, or before each capture

  • clock_source (one of [‘internal’, ‘external’, ‘gps’]): Hardware source for the frequency reference

  • receive_retries (int): number of acquisition retry attempts on stream error (Constraints: >= 0)

  • adc_overload_limit (float | None): dataset adc_overload=True when the peak ADC level exceeds this threshold (Constraints: <= 0, units: dBfs)

  • if_overload_limit (float | None): the maximum level at maximum gain before IF/baseband overload (Constraints: units: dBfs)

  • gapless (bool): whether to raise an exception on overflows between identical captures

arm(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis=<factory>, external_lo_frequency=None, center_frequency, gain)
Parameters:

  • duration (float): Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float): Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float): Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, …]): Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of [‘left’, ‘right’, ‘none’]): LO shift direction

  • host_resample (bool):

  • backend_sample_rate (float | None): Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]): changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, …]): LO frequency of external frequency converter

  • center_frequency (float | tuple[float, …]): Center frequency for each port (Constraints: units: Hz, > 0, units: Hz, > 0, units: Hz)

  • gain (float | tuple[float, …]): Gain setting for each channel (Constraints: units: dB, units: dB, units: dB)

acquire(overlaps=None)

acquire IQ samples needed for the armed capture.

—

class striqt.sensor.bindings.air7101b_cellular(*, master_clock_rate=125000000.0, trigger_strobe=None, signal_trigger=None, array_backend='cupy', calibration=None, time_source='host', time_sync_at='acquire', clock_source='internal', receive_retries=3, adc_overload_limit=-1, if_overload_limit=None, gapless=False)

Bases: Controller

Open and control the backend striqt.sensor.lib.sources.deepwave.Airstack1Source.

Parameters:

  • master_clock_rate (float): Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None):

  • signal_trigger (str | BuiltinTriggers | None):

  • array_backend (one of [‘numpy’, ‘cupy’]): array module to use to set compute device: numpy = cpu, cupy = gpu

  • calibration (str | None):

  • time_source (one of [‘host’, ‘internal’, ‘external’, ‘gps’]): Hardware source for timestamps

  • time_sync_at (one of [‘open’, ‘acquire’]): when to sync the hardware clock: on connection, or before each capture

  • clock_source (one of [‘internal’, ‘external’, ‘gps’]): Hardware source for the frequency reference

  • receive_retries (int): number of acquisition retry attempts on stream error (Constraints: >= 0)

  • adc_overload_limit (float | None): dataset adc_overload=True when the peak ADC level exceeds this threshold (Constraints: <= 0, units: dBfs)

  • if_overload_limit (float | None): the maximum level at maximum gain before IF/baseband overload (Constraints: units: dBfs)

  • gapless (bool): whether to raise an exception on overflows between identical captures

arm(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis=<factory>, external_lo_frequency=None, center_frequency, gain)
Parameters:

  • duration (float): Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float): Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float): Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, …]): Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of [‘left’, ‘right’, ‘none’]): LO shift direction

  • host_resample (bool):

  • backend_sample_rate (float | None): Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]): changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, …]): LO frequency of external frequency converter

  • center_frequency (float | tuple[float, …]): Center frequency for each port (Constraints: units: Hz, > 0, units: Hz, > 0, units: Hz)

  • gain (float | tuple[float, …]): Gain setting for each channel (Constraints: units: dB, units: dB, units: dB)

acquire(overlaps=None)

acquire IQ samples needed for the armed capture.

—

class striqt.sensor.bindings.air7201b(*, master_clock_rate=125000000.0, trigger_strobe=None, signal_trigger=None, array_backend='cupy', calibration=None, time_source='host', time_sync_at='acquire', clock_source='internal', receive_retries=3, adc_overload_limit=-1, if_overload_limit=None, gapless=False)

Bases: Controller

Open and control the backend striqt.sensor.lib.sources.deepwave.Airstack1Source.

Parameters:

  • master_clock_rate (float): Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None):

  • signal_trigger (str | BuiltinTriggers | None):

  • array_backend (one of [‘numpy’, ‘cupy’]): array module to use to set compute device: numpy = cpu, cupy = gpu

  • calibration (str | None):

  • time_source (one of [‘host’, ‘internal’, ‘external’, ‘gps’]): Hardware source for timestamps

  • time_sync_at (one of [‘open’, ‘acquire’]): when to sync the hardware clock: on connection, or before each capture

  • clock_source (one of [‘internal’, ‘external’, ‘gps’]): Hardware source for the frequency reference

  • receive_retries (int): number of acquisition retry attempts on stream error (Constraints: >= 0)

  • adc_overload_limit (float | None): dataset adc_overload=True when the peak ADC level exceeds this threshold (Constraints: <= 0, units: dBfs)

  • if_overload_limit (float | None): the maximum level at maximum gain before IF/baseband overload (Constraints: units: dBfs)

  • gapless (bool): whether to raise an exception on overflows between identical captures

arm(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis=<factory>, external_lo_frequency=None, center_frequency, gain)
Parameters:

  • duration (float): Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float): Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float): Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, …]): Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of [‘left’, ‘right’, ‘none’]): LO shift direction

  • host_resample (bool):

  • backend_sample_rate (float | None): Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]): changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, …]): LO frequency of external frequency converter

  • center_frequency (float | tuple[float, …]): Center frequency for each port (Constraints: units: Hz, > 0, units: Hz, > 0, units: Hz)

  • gain (float | tuple[float, …]): Gain setting for each channel (Constraints: units: dB, units: dB, units: dB)

acquire(overlaps=None)

acquire IQ samples needed for the armed capture.

—

class striqt.sensor.bindings.air7201b_calibration(*, master_clock_rate=125000000.0, trigger_strobe=None, signal_trigger=None, array_backend='cupy', calibration=None, time_source='host', time_sync_at='acquire', clock_source='internal', receive_retries=3, adc_overload_limit=-1, if_overload_limit=None, gapless=False)

Bases: Controller

Open and control the backend striqt.sensor.lib.sources.deepwave.Airstack1Source.

Parameters:

  • master_clock_rate (float): Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None):

  • signal_trigger (str | BuiltinTriggers | None):

  • array_backend (one of [‘numpy’, ‘cupy’]): array module to use to set compute device: numpy = cpu, cupy = gpu

  • calibration (str | None):

  • time_source (one of [‘host’, ‘internal’, ‘external’, ‘gps’]): Hardware source for timestamps

  • time_sync_at (one of [‘open’, ‘acquire’]): when to sync the hardware clock: on connection, or before each capture

  • clock_source (one of [‘internal’, ‘external’, ‘gps’]): Hardware source for the frequency reference

  • receive_retries (int): number of acquisition retry attempts on stream error (Constraints: >= 0)

  • adc_overload_limit (float | None): dataset adc_overload=True when the peak ADC level exceeds this threshold (Constraints: <= 0, units: dBfs)

  • if_overload_limit (float | None): the maximum level at maximum gain before IF/baseband overload (Constraints: units: dBfs)

  • gapless (bool): whether to raise an exception on overflows between identical captures

arm(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis=<factory>, external_lo_frequency=None, center_frequency, gain)
Parameters:

  • duration (float): Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float): Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float): Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, …]): Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of [‘left’, ‘right’, ‘none’]): LO shift direction

  • host_resample (bool):

  • backend_sample_rate (float | None): Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]): changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, …]): LO frequency of external frequency converter

  • center_frequency (float | tuple[float, …]): Center frequency for each port (Constraints: units: Hz, > 0, units: Hz, > 0, units: Hz)

  • gain (float | tuple[float, …]): Gain setting for each channel (Constraints: units: dB, units: dB, units: dB)

acquire(overlaps=None)

acquire IQ samples needed for the armed capture.

—

class striqt.sensor.bindings.air7201b_cellular(*, master_clock_rate=125000000.0, trigger_strobe=None, signal_trigger=None, array_backend='cupy', calibration=None, time_source='host', time_sync_at='acquire', clock_source='internal', receive_retries=3, adc_overload_limit=-1, if_overload_limit=None, gapless=False)

Bases: Controller

Open and control the backend striqt.sensor.lib.sources.deepwave.Airstack1Source.

Parameters:

  • master_clock_rate (float): Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None):

  • signal_trigger (str | BuiltinTriggers | None):

  • array_backend (one of [‘numpy’, ‘cupy’]): array module to use to set compute device: numpy = cpu, cupy = gpu

  • calibration (str | None):

  • time_source (one of [‘host’, ‘internal’, ‘external’, ‘gps’]): Hardware source for timestamps

  • time_sync_at (one of [‘open’, ‘acquire’]): when to sync the hardware clock: on connection, or before each capture

  • clock_source (one of [‘internal’, ‘external’, ‘gps’]): Hardware source for the frequency reference

  • receive_retries (int): number of acquisition retry attempts on stream error (Constraints: >= 0)

  • adc_overload_limit (float | None): dataset adc_overload=True when the peak ADC level exceeds this threshold (Constraints: <= 0, units: dBfs)

  • if_overload_limit (float | None): the maximum level at maximum gain before IF/baseband overload (Constraints: units: dBfs)

  • gapless (bool): whether to raise an exception on overflows between identical captures

arm(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis=<factory>, external_lo_frequency=None, center_frequency, gain)
Parameters:

  • duration (float): Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float): Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float): Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, …]): Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of [‘left’, ‘right’, ‘none’]): LO shift direction

  • host_resample (bool):

  • backend_sample_rate (float | None): Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]): changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, …]): LO frequency of external frequency converter

  • center_frequency (float | tuple[float, …]): Center frequency for each port (Constraints: units: Hz, > 0, units: Hz, > 0, units: Hz)

  • gain (float | tuple[float, …]): Gain setting for each channel (Constraints: units: dB, units: dB, units: dB)

acquire(overlaps=None)

acquire IQ samples needed for the armed capture.

—

class striqt.sensor.bindings.air8201b(*, master_clock_rate=125000000.0, trigger_strobe=None, signal_trigger=None, array_backend='cupy', calibration=None, time_source='host', time_sync_at='acquire', clock_source='internal', receive_retries=3, adc_overload_limit=-1, if_overload_limit=None, gapless=False)

Bases: Controller

Open and control the backend striqt.sensor.lib.sources.deepwave.Airstack1Source.

Parameters:

  • master_clock_rate (float): Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None):

  • signal_trigger (str | BuiltinTriggers | None):

  • array_backend (one of [‘numpy’, ‘cupy’]): array module to use to set compute device: numpy = cpu, cupy = gpu

  • calibration (str | None):

  • time_source (one of [‘host’, ‘internal’, ‘external’, ‘gps’]): Hardware source for timestamps

  • time_sync_at (one of [‘open’, ‘acquire’]): when to sync the hardware clock: on connection, or before each capture

  • clock_source (one of [‘internal’, ‘external’, ‘gps’]): Hardware source for the frequency reference

  • receive_retries (int): number of acquisition retry attempts on stream error (Constraints: >= 0)

  • adc_overload_limit (float | None): dataset adc_overload=True when the peak ADC level exceeds this threshold (Constraints: <= 0, units: dBfs)

  • if_overload_limit (float | None): the maximum level at maximum gain before IF/baseband overload (Constraints: units: dBfs)

  • gapless (bool): whether to raise an exception on overflows between identical captures

arm(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis=<factory>, external_lo_frequency=None, center_frequency, gain)
Parameters:

  • duration (float): Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float): Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float): Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, …]): Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of [‘left’, ‘right’, ‘none’]): LO shift direction

  • host_resample (bool):

  • backend_sample_rate (float | None): Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]): changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, …]): LO frequency of external frequency converter

  • center_frequency (float | tuple[float, …]): Center frequency for each port (Constraints: units: Hz, > 0, units: Hz, > 0, units: Hz)

  • gain (float | tuple[float, …]): Gain setting for each channel (Constraints: units: dB, units: dB, units: dB)

acquire(overlaps=None)

acquire IQ samples needed for the armed capture.

—

class striqt.sensor.bindings.air8201b_calibration(*, master_clock_rate=125000000.0, trigger_strobe=None, signal_trigger=None, array_backend='cupy', calibration=None, time_source='host', time_sync_at='acquire', clock_source='internal', receive_retries=3, adc_overload_limit=-1, if_overload_limit=None, gapless=False)

Bases: Controller

Open and control the backend striqt.sensor.lib.sources.deepwave.Airstack1Source.

Parameters:

  • master_clock_rate (float): Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None):

  • signal_trigger (str | BuiltinTriggers | None):

  • array_backend (one of [‘numpy’, ‘cupy’]): array module to use to set compute device: numpy = cpu, cupy = gpu

  • calibration (str | None):

  • time_source (one of [‘host’, ‘internal’, ‘external’, ‘gps’]): Hardware source for timestamps

  • time_sync_at (one of [‘open’, ‘acquire’]): when to sync the hardware clock: on connection, or before each capture

  • clock_source (one of [‘internal’, ‘external’, ‘gps’]): Hardware source for the frequency reference

  • receive_retries (int): number of acquisition retry attempts on stream error (Constraints: >= 0)

  • adc_overload_limit (float | None): dataset adc_overload=True when the peak ADC level exceeds this threshold (Constraints: <= 0, units: dBfs)

  • if_overload_limit (float | None): the maximum level at maximum gain before IF/baseband overload (Constraints: units: dBfs)

  • gapless (bool): whether to raise an exception on overflows between identical captures

arm(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis=<factory>, external_lo_frequency=None, center_frequency, gain)
Parameters:

  • duration (float): Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float): Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float): Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, …]): Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of [‘left’, ‘right’, ‘none’]): LO shift direction

  • host_resample (bool):

  • backend_sample_rate (float | None): Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]): changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, …]): LO frequency of external frequency converter

  • center_frequency (float | tuple[float, …]): Center frequency for each port (Constraints: units: Hz, > 0, units: Hz, > 0, units: Hz)

  • gain (float | tuple[float, …]): Gain setting for each channel (Constraints: units: dB, units: dB, units: dB)

acquire(overlaps=None)

acquire IQ samples needed for the armed capture.

—

class striqt.sensor.bindings.air8201b_cellular(*, master_clock_rate=125000000.0, trigger_strobe=None, signal_trigger=None, array_backend='cupy', calibration=None, time_source='host', time_sync_at='acquire', clock_source='internal', receive_retries=3, adc_overload_limit=-1, if_overload_limit=None, gapless=False)

Bases: Controller

Open and control the backend striqt.sensor.lib.sources.deepwave.Airstack1Source.

Parameters:

  • master_clock_rate (float): Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None):

  • signal_trigger (str | BuiltinTriggers | None):

  • array_backend (one of [‘numpy’, ‘cupy’]): array module to use to set compute device: numpy = cpu, cupy = gpu

  • calibration (str | None):

  • time_source (one of [‘host’, ‘internal’, ‘external’, ‘gps’]): Hardware source for timestamps

  • time_sync_at (one of [‘open’, ‘acquire’]): when to sync the hardware clock: on connection, or before each capture

  • clock_source (one of [‘internal’, ‘external’, ‘gps’]): Hardware source for the frequency reference

  • receive_retries (int): number of acquisition retry attempts on stream error (Constraints: >= 0)

  • adc_overload_limit (float | None): dataset adc_overload=True when the peak ADC level exceeds this threshold (Constraints: <= 0, units: dBfs)

  • if_overload_limit (float | None): the maximum level at maximum gain before IF/baseband overload (Constraints: units: dBfs)

  • gapless (bool): whether to raise an exception on overflows between identical captures

arm(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis=<factory>, external_lo_frequency=None, center_frequency, gain)
Parameters:

  • duration (float): Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float): Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float): Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, …]): Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of [‘left’, ‘right’, ‘none’]): LO shift direction

  • host_resample (bool):

  • backend_sample_rate (float | None): Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]): changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, …]): LO frequency of external frequency converter

  • center_frequency (float | tuple[float, …]): Center frequency for each port (Constraints: units: Hz, > 0, units: Hz, > 0, units: Hz)

  • gain (float | tuple[float, …]): Gain setting for each channel (Constraints: units: dB, units: dB, units: dB)

acquire(overlaps=None)

acquire IQ samples needed for the armed capture.

—

class striqt.sensor.bindings.dirac_delta(*, master_clock_rate, trigger_strobe=None, signal_trigger=None, array_backend='numpy', num_rx_ports)

Bases: Controller

Open and control the backend striqt.sensor.lib.sources.function.DiracDeltaSource.

Parameters:

  • master_clock_rate (float): Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None):

  • signal_trigger (str | BuiltinTriggers | None):

  • array_backend (one of [‘numpy’, ‘cupy’]): array module to use to set compute device: numpy = cpu, cupy = gpu

  • num_rx_ports (int):

arm(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis=<factory>, external_lo_frequency=None, time=0, power=0)
Parameters:

  • duration (float): Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float): Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float): Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, …]): Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of [‘left’, ‘right’, ‘none’]): LO shift direction

  • host_resample (bool):

  • backend_sample_rate (float | None): Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]): changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, …]): LO frequency of external frequency converter

  • time (float): start time offset (Constraints: units: s)

  • power (float): peak power level (Constraints: units: dB)

acquire(overlaps=None)

acquire IQ samples needed for the armed capture.

—

class striqt.sensor.bindings.mat_file(*, master_clock_rate, trigger_strobe=None, signal_trigger=None, array_backend='numpy', path, file_format='auto', file_metadata=None, loop=False, key=None)

Bases: Controller

Open and control the backend striqt.sensor.lib.sources.file.MATSource.

returns IQ waveforms from a .mat file

Parameters:

  • master_clock_rate (float): Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None):

  • signal_trigger (str | BuiltinTriggers | None):

  • array_backend (one of [‘numpy’, ‘cupy’]): array module to use to set compute device: numpy = cpu, cupy = gpu

  • path (str): path to the waveform data file

  • file_format (one of [‘auto’, ‘mat’, ‘tdms’]): data format or auto to guess by extension

  • file_metadata (dict[str, Any] | None): any capture fields not included in the file

  • loop (bool): whether to loop the file to create longer IQ waveforms

  • key (str | None):

arm(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis=<factory>, external_lo_frequency=None)
Parameters:

  • duration (float): Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float): Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float): Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, …]): Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of [‘left’, ‘right’, ‘none’]): LO shift direction

  • host_resample (bool):

  • backend_sample_rate (float | None): Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]): changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, …]): LO frequency of external frequency converter

acquire(overlaps=None)

acquire IQ samples needed for the armed capture.

—

class striqt.sensor.bindings.noise(*, master_clock_rate, trigger_strobe=None, signal_trigger=None, array_backend='numpy', num_rx_ports)

Bases: Controller

Open and control the backend striqt.sensor.lib.sources.function.NoiseSource.

Parameters:

  • master_clock_rate (float): Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None):

  • signal_trigger (str | BuiltinTriggers | None):

  • array_backend (one of [‘numpy’, ‘cupy’]): array module to use to set compute device: numpy = cpu, cupy = gpu

  • num_rx_ports (int):

arm(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis=<factory>, external_lo_frequency=None, noise_psd=1e-17)
Parameters:

  • duration (float): Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float): Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float): Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, …]): Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of [‘left’, ‘right’, ‘none’]): LO shift direction

  • host_resample (bool):

  • backend_sample_rate (float | None): Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]): changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, …]): LO frequency of external frequency converter

  • noise_psd (float): noise total channel power (Constraints: >= 0, units: mW/Hz)

acquire(overlaps=None)

acquire IQ samples needed for the armed capture.

—

class striqt.sensor.bindings.sawtooth(*, master_clock_rate, trigger_strobe=None, signal_trigger=None, array_backend='numpy', num_rx_ports)

Bases: Controller

Open and control the backend striqt.sensor.lib.sources.function.SawtoothSource.

Parameters:

  • master_clock_rate (float): Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None):

  • signal_trigger (str | BuiltinTriggers | None):

  • array_backend (one of [‘numpy’, ‘cupy’]): array module to use to set compute device: numpy = cpu, cupy = gpu

  • num_rx_ports (int):

arm(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis=<factory>, external_lo_frequency=None, period=0.01, power=0)
Parameters:

  • duration (float): Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float): Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float): Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, …]): Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of [‘left’, ‘right’, ‘none’]): LO shift direction

  • host_resample (bool):

  • backend_sample_rate (float | None): Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]): changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, …]): LO frequency of external frequency converter

  • period (float): waveform period (Constraints: >= 0, units: s)

  • power (float): peak power level (Constraints: units: dB)

acquire(overlaps=None)

acquire IQ samples needed for the armed capture.

—

class striqt.sensor.bindings.single_tone(*, master_clock_rate, trigger_strobe=None, signal_trigger=None, array_backend='numpy', num_rx_ports)

Bases: Controller

Open and control the backend striqt.sensor.lib.sources.function.SingleToneSource.

Parameters:

  • master_clock_rate (float): Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None):

  • signal_trigger (str | BuiltinTriggers | None):

  • array_backend (one of [‘numpy’, ‘cupy’]): array module to use to set compute device: numpy = cpu, cupy = gpu

  • num_rx_ports (int):

arm(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis=<factory>, external_lo_frequency=None, frequency_offset=0, snr=None)
Parameters:

  • duration (float): Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float): Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float): Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, …]): Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of [‘left’, ‘right’, ‘none’]): LO shift direction

  • host_resample (bool):

  • backend_sample_rate (float | None): Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]): changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, …]): LO frequency of external frequency converter

  • frequency_offset (float): Baseband frequency offset (Constraints: units: Hz)

  • snr (float | None): Add circular white gaussian noise to achieve this SNR (Constraints: units: dB)

acquire(overlaps=None)

acquire IQ samples needed for the armed capture.

—

class striqt.sensor.bindings.tdms_file(*, master_clock_rate, trigger_strobe=None, signal_trigger=None, array_backend='numpy', path)

Bases: Controller

Open and control the backend striqt.sensor.lib.sources.file.TDMSSource.

a source of IQ waveforms from a TDMS file

Parameters:

  • master_clock_rate (float): Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None):

  • signal_trigger (str | BuiltinTriggers | None):

  • array_backend (one of [‘numpy’, ‘cupy’]): array module to use to set compute device: numpy = cpu, cupy = gpu

  • path (str): path to the waveform data file

arm(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis=<factory>, external_lo_frequency=None)
Parameters:

  • duration (float): Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float): Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float): Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, …]): Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of [‘left’, ‘right’, ‘none’]): LO shift direction

  • host_resample (bool):

  • backend_sample_rate (float | None): Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]): changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, …]): LO frequency of external frequency converter

acquire(overlaps=None)

acquire IQ samples needed for the armed capture.

—

class striqt.sensor.bindings.warmup(*, master_clock_rate, trigger_strobe=None, signal_trigger=None, array_backend='numpy', num_rx_ports)

Bases: Controller

Open and control the backend striqt.sensor.lib.sources.base.NoSource.

fast paths to acquire empty buffers

Parameters:

  • master_clock_rate (float): Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None):

  • signal_trigger (str | BuiltinTriggers | None):

  • array_backend (one of [‘numpy’, ‘cupy’]): array module to use to set compute device: numpy = cpu, cupy = gpu

  • num_rx_ports (int):

arm(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis=<factory>, external_lo_frequency=None)
Parameters:

  • duration (float): Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float): Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float): Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, …]): Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of [‘left’, ‘right’, ‘none’]): LO shift direction

  • host_resample (bool):

  • backend_sample_rate (float | None): Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]): changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, …]): LO frequency of external frequency converter

acquire(overlaps=None)

acquire IQ samples needed for the armed capture.

—

class striqt.sensor.bindings.zarr_iq(*, master_clock_rate, trigger_strobe=None, signal_trigger=None, array_backend='numpy', path, center_frequency, select=<factory>)

Bases: Controller

Open and control the backend striqt.sensor.lib.sources.file.ZarrIQSource.

a source of IQ samples from the iq_waveform variable of a zarr store

Parameters:

  • master_clock_rate (float): Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None):

  • signal_trigger (str | BuiltinTriggers | None):

  • array_backend (one of [‘numpy’, ‘cupy’]): array module to use to set compute device: numpy = cpu, cupy = gpu

  • path (str): path to the waveform data file

  • center_frequency (float | tuple[float, …]): Center frequency for each port (Constraints: units: Hz, > 0, units: Hz, > 0, units: Hz)

  • select (dict): dictionary to select in the data as .sel(**select)

arm(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis=<factory>, external_lo_frequency=None)
Parameters:

  • duration (float): Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float): Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float): Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, …]): Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of [‘left’, ‘right’, ‘none’]): LO shift direction

  • host_resample (bool):

  • backend_sample_rate (float | None): Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]): changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, …]): LO frequency of external frequency converter

acquire(overlaps=None)

acquire IQ samples needed for the armed capture.

class striqt.sensor.bindings.Controller(*args: ~PS, **kwargs: ~PS)

Bases: Generic[SS, SP, SC, SA, PS, PC]

Open the source backend for control and acquisition.

acquire(overlaps: tuple[int, int] | None = None) → AcquiredIQ

acquire IQ samples needed for the armed capture.

arm(*args: ~PC, **kwargs: ~PC) → SC | None
backend: SourceBackend[SS, SC]
property capture_spec: SC

return the specification of the currently armed capture

close() → None
discard_iq() → None

drop the acquisition that reuse_iq would serve to the next matching capture.

The next acquire reads the source even when the armed capture is unchanged.

classmethod from_source_spec(spec: SS, reuse_iq: bool = False, rx_ports: tuple[int, ...] | None = None, format_path: PathFormatter | None = None) → Controller[SS, SP, SC, SA, PS, PC]
classmethod from_sweep_spec(spec: Sweep[SS, Any, SC, SA], format_path: PathFormatter | None = None) → Controller[SS, SP, SC, SA, PS, PC]
get_resampler() → ResamplerDesign
is_open(wait: bool = True) → bool

return whether the backend of this instance is open.

Parameters:

wait – if True, block up to the controller timeout for an open in progress; if False, report the state at the time of the call

Returns:

False once close has run on this instance, even if another controller has since been opened for an equal source spec

read_iq(overlaps: tuple[int, int] = (0, 0)) → tuple[Array, int | None]

read IQ for the armed capture

schema: ClassVar['specs.Schema[Any, Any, Any, Any, Any, Any]']
sensor: ClassVar['bindings.SensorBinding[Any, Any, Any]']
set_calibration_formatter(format_path: PathFormatter | None = None)

set the formatter that expands the path fields of source.calibration.

None clears the formatter. A formatter is required when the calibration path has fields such as {name}.

Raises:

TypeError – if format_path is not a PathFormatter or None

property source_id: str
property source_info: SourceInfo
property source_spec: SS
target_analysis(analysis: AnalysisGroup | None)

sets or disables an analysis target to auto-select acquisition overlap

class striqt.sensor.bindings.Schema(source: type[~striqt.sensor.lib.typing.SS], capture: type[~striqt.sensor.lib.typing.SC], peripherals: type[~striqt.sensor.lib.typing.SP], init_like: ~typing.Callable[[~PS], ~typing.Any], arm_like: ~typing.Callable[[~PC], ~typing.Any], action: type[~striqt.sensor.specs.ActionSpec[~striqt.sensor.lib.typing.SS, ~striqt.sensor.lib.typing.SP, ~striqt.sensor.lib.typing.SA]] = <class 'striqt.sensor.specs.Sweep'>, analysis: type[~striqt.sensor.lib.typing.SA] = <class 'striqt.analysis.specs.BuiltinAnalyses'>)

Bases: Generic[SS, SP, SC, SA, PS, PC]

the spec types of a sensor binding.

action

alias of Sweep

analysis

alias of BuiltinAnalyses

arm_like: Callable[PC, Any]
capture: type[SC]
init_like: Callable[PS, Any]
peripherals: type[SP]
source: type[SS]
class striqt.sensor.bindings.Sensor(source_cls: type[~striqt.sensor.lib.typing.SourceBackend[~striqt.sensor.lib.typing.SS, ~striqt.sensor.lib.typing.SC]], sweep_spec_cls: type[~striqt.sensor.specs.ActionSpec[~striqt.sensor.lib.typing.SS, ~striqt.sensor.lib.typing.SP, ~typing.Any]] = <class 'striqt.sensor.specs.Sweep'>, peripherals_cls: type[~striqt.sensor.lib.typing.Peripherals[~striqt.sensor.lib.typing.SP, ~striqt.sensor.lib.typing.SC]] | type[~striqt.sensor.lib.peripherals.NoPeripherals[~striqt.sensor.lib.typing.SP, ~striqt.sensor.lib.typing.SC]] = <class 'striqt.sensor.lib.peripherals.NoPeripherals'>)

Bases: Generic[SS, SP, SC]

the implementation classes of a sensor

peripherals_cls

alias of NoPeripherals

source_cls: type[SourceBackend[SS, SC]]

the source backend class instantiated from the source block of a spec

sweep_spec_cls

alias of Sweep

class striqt.sensor.bindings.SensorBinding(source_cls: type[~striqt.sensor.lib.typing.SourceBackend[~striqt.sensor.lib.typing.SS, ~striqt.sensor.lib.typing.SC]], sweep_spec_cls: type[~striqt.sensor.specs.ActionSpec[~striqt.sensor.lib.typing.SS, ~striqt.sensor.lib.typing.SP, ~typing.Any]] = <class 'striqt.sensor.specs.Sweep'>, peripherals_cls: type[~striqt.sensor.lib.typing.Peripherals[~striqt.sensor.lib.typing.SP, ~striqt.sensor.lib.typing.SC]] | type[~striqt.sensor.lib.peripherals.NoPeripherals[~striqt.sensor.lib.typing.SP, ~striqt.sensor.lib.typing.SC]] = <class 'striqt.sensor.lib.peripherals.NoPeripherals'>)

Bases: Sensor[SS, SP, SC]

a bound Sensor with the tagged action spec class as sweep_spec_cls

striqt.sensor.bindings.bind_manual_yfactor_calibration(name: str, ctrl_cls: type[Controller[SS, SP, Any, SA, PS, PC]]) → type[Controller[SS, SP, Any, SA, PS, PC]]

bind the sensor of ctrl_cls for ManualYFactorCalibration under name.

Deprecated in favor of bind_sensor with ctrl_cls.sensor and dataclasses.replace(ctrl_cls.schema, action=specs.ManualYFactorCalibration). A DeprecationWarning is emitted on each call. The __module__ of the returned class is the module of the caller.

striqt.sensor.bindings.bind_sensor(key: str, sensor: Sensor[TS2, TP2, TC2], schema: Schema[SS, SP, SC, SA, PS, PC], register: bool = True, *, module: str | None = None) → type[Controller[SS, SP, SC, SA, PS, PC]]

bind the implementation classes of a sensor to spec types under key.

The returned Controller subclass has the class attributes sensor (a SensorBinding) and schema. Its sensor.sweep_spec_cls is a subclass of the action spec type, tagged with key in the sensor_binding field. In that subclass, source, peripherals, captures (when the action declares the field), and analysis (when the action is a Sweep) are annotated with the types in schema. A sweep file with sensor_binding equal to key decodes to an instance of that subclass. The tagged subclass has the class attributes sensor, schema, controller (the returned class), and sweep_cls (the bound Sweep class with the same spec types, or the tagged subclass when the action is a Sweep). The tagged subclass of an unregistered binding is decoded only by explicit type.

Parameters:

  • key – the sensor_binding value of sweep files, and the registry key

  • sensor – the implementation classes; the sensor of an existing controller binds the same hardware for the action in schema

  • schema – the spec types; schema.action selects the action spec type and schema.analysis the analysis group

  • register – if True, make the binding available by key to get_controller and the tagged subclass an arm of the union from get_tagged_sweep_type

  • module – the __module__ of the returned class; the module of the caller when None

Raises:

TypeError – if sensor or schema has the wrong type, if key is already registered, if sensor.sweep_spec_cls and schema.action name different non-default types, if the check_schema of the action type rejects schema, or if a field of schema.analysis is not a registered measurement

striqt.sensor.calibration

striqt.sensor.calibration.bind_manual_yfactor_calibration(name: str, ctrl_cls: type[Controller[SS, SP, Any, SA, PS, PC]]) → type[Controller[SS, SP, Any, SA, PS, PC]]

bind the sensor of ctrl_cls for ManualYFactorCalibration under name.

Deprecated in favor of bind_sensor with ctrl_cls.sensor and dataclasses.replace(ctrl_cls.schema, action=specs.ManualYFactorCalibration). A DeprecationWarning is emitted on each call. The __module__ of the returned class is the module of the caller.

striqt.sensor.calibration.compute_y_factor_corrections(dataset: Dataset, enr_dB: float, Tref: float = 290.0) → Dataset

return the Y-factor noise figure and power correction of each grid point.

The dimensions of dataset are the calibration fields and noise_diode_enabled. The variables are channel_power_time_series (dB, rms and peak detectors over time_elapsed), analysis_bandwidth (Hz), and backend_sample_rate (S/s). The rms detector is averaged in linear power over time_elapsed before the Y-factor is formed. An infinite analysis_bandwidth counts as the backend sample rate.

Parameters:

  • dataset – the collected captures, indexed by the calibration fields and the diode state

  • enr_dB – the excess noise ratio (dB) of the noise diode

  • Tref – the reference temperature (K) of the ENR and the noise figure

Returns:

a dataset over the calibration fields with noise_figure (dB), temperature (K), power_correction (mW per full-scale power unit), the linear powers p_on and p_off, and the diagnostics papr_on and papr_off (dB)

striqt.sensor.calibration.lookup_power_correction(cal_data: str | Path | None, capture: SoapyCapture, master_clock_rate: float, format_path: PathFormatter | None = None, *, xp=None)
striqt.sensor.calibration.lookup_system_noise_power(cal_data: Path | str | None, capture: SoapyCapture, master_clock_rate: float, format_path: PathFormatter | None = None, *, T=290.0, B=1.0, xp=None)

return the system noise power (dBm) referred to the calibrated input, per port.

The power is k * T * B * F, the noise power that the sensor measures with the input terminated at temperature T. F is the linear noise figure from the calibration file at the port, gain, LO shift, backend sample rate, and analysis bandwidth of capture, interpolated in center frequency.

Parameters:

  • cal_data – the path of the calibration corrections file, or None for no calibration

  • capture – the capture that selects the calibration row, one row per port

  • master_clock_rate – the SDR master clock rate (S/s), which sets the backend sample rate of the row

  • format_path – the formatter of the {field} templates in cal_data, or None

  • T – the source temperature (K) that the noise figure is referred to

  • B – the noise bandwidth (Hz); 1 Hz gives a power spectral density

  • xp – the array namespace of the result, or None for numpy

Returns:

a DataArray on the capture dimension, one value per port, in dBm/Hz when B is 1 and in dBm with a noise_bandwidth attr otherwise; the reference_temperature attr is T. None when cal_data is None or the file has no corrections.

Raises:

  • TypeError – if cal_data is neither a path nor None

  • KeyError – if the port, gain, LO shift, backend sample rate, or analysis bandwidth of a port is not in the calibration grid

  • ValueError – if the center frequency of a port is outside the calibrated range, or no calibrated row remains for the port

striqt.sensor.calibration.read_calibration(path: None, format_path: specs.paths.PathFormatter | None = None) → None
striqt.sensor.calibration.read_calibration(path: str | Path, format_path: specs.paths.PathFormatter | None = None) → xr.Dataset
striqt.sensor.calibration.summarize_calibration(corrections: Dataset, **sel) → DataFrame

striqt.sensor.peripherals

striqt.sensor.sinks

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

Bases: SinkBase

collect the dataset of every capture in memory until pop is called.

Nothing is written. close leaves the collected datasets in place.

close(*exc_info)
open() → None
class striqt.sensor.sinks.NoSink(sweep_spec: Sweep[Any, Any, SC, Any], 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, Any], format_path: PathFormatter | None = None, *, force: bool = False)

Bases: Generic[SC]

base sink that accepts the analyzed dataset of one capture at a time.

A sink is constructed from a Sweep, reads the sink block of that sweep, and is entered as a context manager before the first append. Writes run on 1 worker thread.

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, Any], format_path: PathFormatter | None = None, *, force: bool = False)

Bases: ZarrSinkBase

append the dataset of each capture to a zarr store along capture

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

Bases: SinkBase

write the zarr store at sink.path in batches of batched_write_count or more.

A .zarr.zip path is written through a Zipper.

Raises:

ValueError – if sweep_spec.sink.path is None

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 the spectrogram of each capture along the spectrogram time axis.

The captures of one write batch share one capture index, and each batch is appended along capture. The store is written without compression.

Raises:

ValueError – if sweep_spec.analysis has no spectrogram

append(capture_result)
flush()

striqt.sensor.sources

class striqt.sensor.sources.Air7101BSourceSpec(*, master_clock_rate=125000000.0, trigger_strobe=None, signal_trigger=None, array_backend='cupy', calibration=None, time_source='host', time_sync_at='acquire', clock_source='internal', receive_retries=3, adc_overload_limit=-1, if_overload_limit=None, gapless=False)

Bases: SoapySource

Parameters:

  • master_clock_rate (float) – Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None)

  • signal_trigger (str | BuiltinTriggers | None)

  • array_backend (one of ['numpy', 'cupy']) – array module to use to set compute device: numpy = cpu, cupy = gpu

  • calibration (str | None)

  • time_source (one of ['host', 'internal', 'external', 'gps']) – Hardware source for timestamps

  • time_sync_at (one of ['open', 'acquire']) – when to sync the hardware clock: on connection, or before each capture

  • clock_source (one of ['internal', 'external', 'gps']) – Hardware source for the frequency reference

  • receive_retries (int) – number of acquisition retry attempts on stream error (Constraints: >= 0)

  • adc_overload_limit (float | None) – dataset adc_overload=True when the peak ADC level exceeds this threshold (Constraints: <= 0, units: dBfs)

  • if_overload_limit (float | None) – the maximum level at maximum gain before IF/baseband overload (Constraints: units: dBfs)

  • gapless (bool) – whether to raise an exception on overflows between identical captures

Variables:

  • transient_holdoff_time (float) – (Value: 0.002)

  • stream_all_rx_ports (bool | None) – (Value: True)

  • transport_dtype (one of ['int16', 'float32', 'complex64']) – data transfer type to use inside the source (Value: float32)

rx_enable_delay = 0.35
stream_all_rx_ports: ClassVar[bool | None] = True
transient_holdoff_time: ClassVar[float] = 0.002
class striqt.sensor.sources.Air7201BSourceSpec(*, master_clock_rate=125000000.0, trigger_strobe=None, signal_trigger=None, array_backend='cupy', calibration=None, time_source='host', time_sync_at='acquire', clock_source='internal', receive_retries=3, adc_overload_limit=-1, if_overload_limit=None, gapless=False)

Bases: SoapySource

Parameters:

  • master_clock_rate (float) – Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None)

  • signal_trigger (str | BuiltinTriggers | None)

  • array_backend (one of ['numpy', 'cupy']) – array module to use to set compute device: numpy = cpu, cupy = gpu

  • calibration (str | None)

  • time_source (one of ['host', 'internal', 'external', 'gps']) – Hardware source for timestamps

  • time_sync_at (one of ['open', 'acquire']) – when to sync the hardware clock: on connection, or before each capture

  • clock_source (one of ['internal', 'external', 'gps']) – Hardware source for the frequency reference

  • receive_retries (int) – number of acquisition retry attempts on stream error (Constraints: >= 0)

  • adc_overload_limit (float | None) – dataset adc_overload=True when the peak ADC level exceeds this threshold (Constraints: <= 0, units: dBfs)

  • if_overload_limit (float | None) – the maximum level at maximum gain before IF/baseband overload (Constraints: units: dBfs)

  • gapless (bool) – whether to raise an exception on overflows between identical captures

Variables:

  • transient_holdoff_time (float) – (Value: 0.002)

  • stream_all_rx_ports (bool | None) – (Value: True)

  • transport_dtype (one of ['int16', 'float32', 'complex64']) – data transfer type to use inside the source (Value: float32)

rx_enable_delay = 0.35
stream_all_rx_ports: ClassVar[bool | None] = True
transient_holdoff_time: ClassVar[float] = 0.002
class striqt.sensor.sources.Air8201BSourceSpec(*, master_clock_rate=125000000.0, trigger_strobe=None, signal_trigger=None, array_backend='cupy', calibration=None, time_source='host', time_sync_at='acquire', clock_source='internal', receive_retries=3, adc_overload_limit=-1, if_overload_limit=None, gapless=False)

Bases: SoapySource

Parameters:

  • master_clock_rate (float) – Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None)

  • signal_trigger (str | BuiltinTriggers | None)

  • array_backend (one of ['numpy', 'cupy']) – array module to use to set compute device: numpy = cpu, cupy = gpu

  • calibration (str | None)

  • time_source (one of ['host', 'internal', 'external', 'gps']) – Hardware source for timestamps

  • time_sync_at (one of ['open', 'acquire']) – when to sync the hardware clock: on connection, or before each capture

  • clock_source (one of ['internal', 'external', 'gps']) – Hardware source for the frequency reference

  • receive_retries (int) – number of acquisition retry attempts on stream error (Constraints: >= 0)

  • adc_overload_limit (float | None) – dataset adc_overload=True when the peak ADC level exceeds this threshold (Constraints: <= 0, units: dBfs)

  • if_overload_limit (float | None) – the maximum level at maximum gain before IF/baseband overload (Constraints: units: dBfs)

  • gapless (bool) – whether to raise an exception on overflows between identical captures

Variables:

  • transient_holdoff_time (float) – (Value: 0.002)

  • stream_all_rx_ports (bool | None) – (Value: True)

  • transport_dtype (one of ['int16', 'float32', 'complex64']) – data transfer type to use inside the source (Value: float32)

rx_enable_delay = 1.4
stream_all_rx_ports: ClassVar[bool | None] = True
transient_holdoff_time: ClassVar[float] = 0.002
class striqt.sensor.sources.Airstack1Source(spec: SoapySource, **kwargs)

Bases: SoapySource

read_peripherals() → dict[str, float]

returns the transceiver temperature in Celsius

class striqt.sensor.sources.DiracDeltaSource(spec: SS)

Bases: TestSourceBase[FunctionSource, DiracDeltaCapture]

get_waveform(count: int, start_index: int, *, port: int = 0, xp: ModuleType, dtype: DTypeLike = 'complex64') → Array

return count samples of port starting at source sample start_index.

Index 0 is the first sample of the corrected capture, and negative indices are the lead-overlap pre-roll that conditioning trims. A source with no samples before index 0 fills the pre-roll with zeros.

Parameters:

  • count – the number of samples, at the source sample rate

  • start_index – the source-rate index of the first sample; negative in the pre-roll

  • port – the port number of the armed capture

  • xp – the array namespace of the result

  • dtype – the dtype of the result

Returns:

a 1-D array of count samples in the xp namespace with dtype dtype

Raises:

NotImplementedError – in the base class

class striqt.sensor.sources.MATSource(spec)

Bases: VirtualSource[MATSource, FileCapture]

returns IQ waveforms from a .mat file

get_waveform(count: int, start_index: int, *, port: int = 0, xp: ModuleType, dtype: DTypeLike = 'complex64') → Array

return count samples of port starting at source sample start_index.

Index 0 is the first sample of the corrected capture, and negative indices are the lead-overlap pre-roll that conditioning trims. A source with no samples before index 0 fills the pre-roll with zeros.

Parameters:

  • count – the number of samples, at the source sample rate

  • start_index – the source-rate index of the first sample; negative in the pre-roll

  • port – the port number of the armed capture

  • xp – the array namespace of the result

  • dtype – the dtype of the result

Returns:

a 1-D array of count samples in the xp namespace with dtype dtype

Raises:

NotImplementedError – in the base class

class striqt.sensor.sources.NoSource(spec: NoSource)

Bases: SourceBackend[NoSource, SensorCapture]

fast paths to acquire empty buffers

property about
reset_sample_counter(value=0)
class striqt.sensor.sources.NoiseSource(spec: SS)

Bases: TestSourceBase[FunctionSource, NoiseCapture]

get_waveform(count: int, start_index: int, *, port: int = 0, xp: ModuleType, dtype: DTypeLike = 'complex64') → Array

return count samples of port starting at source sample start_index.

Index 0 is the first sample of the corrected capture, and negative indices are the lead-overlap pre-roll that conditioning trims. A source with no samples before index 0 fills the pre-roll with zeros.

Parameters:

  • count – the number of samples, at the source sample rate

  • start_index – the source-rate index of the first sample; negative in the pre-roll

  • port – the port number of the armed capture

  • xp – the array namespace of the result

  • dtype – the dtype of the result

Returns:

a 1-D array of count samples in the xp namespace with dtype dtype

Raises:

NotImplementedError – in the base class

class striqt.sensor.sources.SawtoothSource(spec: SS)

Bases: TestSourceBase[FunctionSource, SawtoothCapture]

get_waveform(count: int, start_index: int, *, port: int = 0, xp: ModuleType, dtype: DTypeLike = 'complex64') → Array

return count samples of port starting at source sample start_index.

Index 0 is the first sample of the corrected capture, and negative indices are the lead-overlap pre-roll that conditioning trims. A source with no samples before index 0 fills the pre-roll with zeros.

Parameters:

  • count – the number of samples, at the source sample rate

  • start_index – the source-rate index of the first sample; negative in the pre-roll

  • port – the port number of the armed capture

  • xp – the array namespace of the result

  • dtype – the dtype of the result

Returns:

a 1-D array of count samples in the xp namespace with dtype dtype

Raises:

NotImplementedError – in the base class

class striqt.sensor.sources.SingleToneSource(spec: SS)

Bases: TestSourceBase[FunctionSource, SingleToneCapture]

get_waveform(count: int, start_index: int, *, port: int = 0, xp: ModuleType, dtype: DTypeLike = 'complex64') → Array

return count samples of port starting at source sample start_index.

Index 0 is the first sample of the corrected capture, and negative indices are the lead-overlap pre-roll that conditioning trims. A source with no samples before index 0 fills the pre-roll with zeros.

Parameters:

  • count – the number of samples, at the source sample rate

  • start_index – the source-rate index of the first sample; negative in the pre-roll

  • port – the port number of the armed capture

  • xp – the array namespace of the result

  • dtype – the dtype of the result

Returns:

a 1-D array of count samples in the xp namespace with dtype dtype

Raises:

NotImplementedError – in the base class

class striqt.sensor.sources.SoapySource(spec: SS, **device_kwargs: Any)

Bases: SourceBackend[SS, SoapyCapture]

a source backend for device control and acquisition through SoapySDR

device: Device
read_peripherals() → dict[str, Any]
rx_stream: RxStream
sync_time: HardwareTimeSync
class striqt.sensor.sources.TDMSSource(spec)

Bases: VirtualSource[TDMSSource, FileCapture]

a source of IQ waveforms from a TDMS file

get_waveform(count: int, start_index: int, *, port: int = 0, xp: ModuleType, dtype: DTypeLike = 'complex64') → Array

return count samples of port starting at source sample start_index.

Index 0 is the first sample of the corrected capture, and negative indices are the lead-overlap pre-roll that conditioning trims. A source with no samples before index 0 fills the pre-roll with zeros.

Parameters:

  • count – the number of samples, at the source sample rate

  • start_index – the source-rate index of the first sample; negative in the pre-roll

  • port – the port number of the armed capture

  • xp – the array namespace of the result

  • dtype – the dtype of the result

Returns:

a 1-D array of count samples in the xp namespace with dtype dtype

Raises:

NotImplementedError – in the base class

class striqt.sensor.sources.VirtualSource(spec: SS)

Bases: SourceBackend[SS, SC]

a source that reads samples from get_waveform on a synthetic time base.

The time base is 0 at the first arm, and _sync_time_ns is the time of corrected sample 0 of the armed capture. A re-arm continues the time base: the first sample read after arm is stamped at the end time of the samples read before the re-arm.

get_waveform(count: int, start_index: int, *, port: int = 0, xp: ModuleType, dtype: DTypeLike = 'complex64') → Array

return count samples of port starting at source sample start_index.

Index 0 is the first sample of the corrected capture, and negative indices are the lead-overlap pre-roll that conditioning trims. A source with no samples before index 0 fills the pre-roll with zeros.

Parameters:

  • count – the number of samples, at the source sample rate

  • start_index – the source-rate index of the first sample; negative in the pre-roll

  • port – the port number of the armed capture

  • xp – the array namespace of the result

  • dtype – the dtype of the result

Returns:

a 1-D array of count samples in the xp namespace with dtype dtype

Raises:

NotImplementedError – in the base class

reset_sample_counter(value: int = 0) → None

restart the sample counter at value.

The next read is stamped at the end time of the samples read so far.

setup_spec: SS
class striqt.sensor.sources.ZarrIQSource(spec)

Bases: VirtualSource[ZarrIQSource, FileCapture]

a source of IQ samples from the iq_waveform variable of a zarr store

__init__(spec)

set the waveform from an xarray.DataArray containing a single capture of IQ samples

get_waveform(count: int, start_index: int, *, port: int = 0, xp: ModuleType, dtype: DTypeLike = 'complex64') → Array

return count samples of port starting at source sample start_index.

Index 0 is the first sample of the corrected capture, and negative indices are the lead-overlap pre-roll that conditioning trims. A source with no samples before index 0 fills the pre-roll with zeros.

Parameters:

  • count – the number of samples, at the source sample rate

  • start_index – the source-rate index of the first sample; negative in the pre-roll

  • port – the port number of the armed capture

  • xp – the array namespace of the result

  • dtype – the dtype of the result

Returns:

a 1-D array of count samples in the xp namespace with dtype dtype

Raises:

NotImplementedError – in the base class

striqt.sensor.specs

schema for the specification of calibration and sweeps

class striqt.sensor.specs.AcquiredIQ(x: Array, capture: specs.Capture | None, info: structs.AcquisitionInfo, extra_data: dict[str, Any], source_spec: structs.Source, resampler: sw.ResamplerDesign, format_path: paths.PathFormatter | None = None, voltage_scale: sw.typing.Array | float = 1, conjugate: tuple[bool, ...] | Literal[False] = False, signal_trigger: sa.Trigger | None = None, overwrite_x: bool = False)

Bases: AcquiredIQ

an acquisition from a sensor source with the metadata and conditioning of the capture.

x must not change after the first access of preprocess.

conjugate: tuple[bool, ...] | Literal[False] = False
extra_data: dict[str, Any]
format_path: paths.PathFormatter | None = None
info: structs.AcquisitionInfo
overwrite_x: bool = False
property preprocess: corrections.Preprocessor

return the Preprocessor that conditions x through the sensor signal path.

One Preprocessor is created per instance. A dataclasses.replace copy starts with a Preprocessor of its own and every stage empty.

resampler: sw.ResamplerDesign
signal_trigger: sa.Trigger | None = None
source_spec: structs.Source
voltage_scale: sw.typing.Array | float = 1
class striqt.sensor.specs.AcquisitionInfo(*, source_id='', sweep_index=0, capture_index=0, signal_trigger=None)

Bases: Struct

the metadata of one acquisition

Parameters:

  • source_id (str) – Source UUID string

  • sweep_index (int)

  • capture_index (int)

  • signal_trigger (Trigger | None)

classmethod from_dict(d: dict) → _T
replace(**attrs) → _Self

returns a copy of self with changed attributes.

See also

Python standard library copy.replace

to_dict() → dict

return the fields of self as a dict

class striqt.sensor.specs.ActionSpec(*, description=Description(summary=None, version='unversioned'), source, peripherals=Peripherals(), extensions=Extension(import_path=None, import_name=None), sink=Sink(path='{spec_name}-{start_time}', log_path=None, log_level='info'))

Bases: SpecBase, Generic[SS, SP, SA]

the top-level spec that an action runs.

An instance is decoded from a file against the tagged class of a sensor binding or constructed from the API. The source, peripherals, and sink blocks name the resources that every action opens. source and peripherals are the spec types of the binding named by the sensor_binding tag.

Parameters:
Variables:

  • schema (Any)

  • sensor (Any)

  • controller (Any)

  • sweep_cls (Any)

classmethod check_schema(schema: Any) → None

raise if the spec types of schema cannot configure this action.

The base implementation accepts every schema.

Parameters:

schema – the Schema being bound

Raises:

TypeError – if a spec type of schema lacks a field the action requires

class striqt.sensor.specs.AnalysisGroup

Bases: SpecBase

base class for a defining set of Analysis specs

class striqt.sensor.specs.BuiltinAnalyses(*, channel_power_time_series=None, channel_power_histogram=None, cyclic_channel_power=None, spectrogram=None, power_spectral_density=None, spectrogram_histogram=None, spectrogram_ratio_histogram=None, cellular_5g_pss_sync=None, cellular_5g_pss_correlation=None, cellular_5g_sss_sync=None, cellular_5g_sss_correlation=None, cellular_5g_ssb_spectrogram=None, cellular_cyclic_autocorrelation=None, cellular_resource_power_histogram=None, iq_waveform=None)

Bases: AnalysisGroup

the analysis: block of a sweep: one optional field per registered measurement.

A field set to a spec enables the measurement with those parameters, and None leaves the measurement out. to_dict omits the None fields.

Parameters:
class striqt.sensor.specs.BuiltinTriggers(*, cellular_5g_pss_sync=None, cellular_5g_sss_sync=None)

Bases: AnalysisGroup

the struct form of the signal_trigger: field of a source.

One optional field per trigger, and exactly one is set. The set field names the trigger, and the value is the spec of the trigger measurement. With this form, the analysis: block need not enable the trigger measurement. The string form of signal_trigger names a registered trigger and takes the spec from the analysis: block.

Parameters:
striqt.sensor.specs.BundledAnalysis

alias of BuiltinAnalyses

striqt.sensor.specs.BundledTriggers

alias of BuiltinTriggers

class striqt.sensor.specs.Capture(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf)

Bases: SpecBase

bare minimum information about an IQ acquisition

Parameters:

  • duration (float) – Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float) – Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float) – Analysis bandwidth (Constraints: units: Hz)

class striqt.sensor.specs.CapturePlan(*, description=Description(summary=None, version='unversioned'), source, peripherals=Peripherals(), extensions=Extension(import_path=None, import_name=None), sink=Sink(path='{spec_name}-{start_time}', log_path=None, log_level='info'), captures=(), loops=(), adjust_captures={})

Bases: ActionSpec[SS, SP, SA], Generic[SS, SP, SC, SA]

an ActionSpec with the captures, loops, and adjust_captures blocks.

loops expands over captures into the sequence of captures that the action runs (see sequencing.loop_captures), and adjust_captures applies per-source overrides to each expanded capture. Construction rejects a repeat loop that is not the outermost entry, two loops over one field, and a loop over a field that the listed captures set to different values. Lookup keys in adjust_captures are converted to the type of the keyed capture field.

Parameters:
Variables:

  • schema (Any)

  • sensor (Any)

  • controller (Any)

  • sweep_cls (Any)

class striqt.sensor.specs.CaptureRemap(*, key, lookup, required=True, default=UNSET)

Bases: SpecBase

Parameters:

  • key (str | tuple[str, ...])

  • lookup (dict[tuple[Any, ...] | Any, Any])

  • required (bool)

  • default (Any)

class striqt.sensor.specs.CellChannel(*, center_frequency, channel_bandwidth, subcarrier_spacing=30000.0, ssb_frequency_offset=0.0, ssb_case=None, channel_guard_bandwidths=(0.0, 0.0), tdd_frame_slots=None, tdd_special_symbols=None, min_sample_rate=None)

Bases: SpecBase

one 5G NR channel of the cell_channels table of a Cellular5GNRSweep.

A capture with the center_frequency of the channel is acquired at the sample rate that striqt.waveform.comms_5gnr.nr_sample_rate derives from channel_bandwidth and subcarrier_spacing, at or above min_sample_rate, with analysis_bandwidth equal to channel_bandwidth. The remaining fields set the cellular measurement parameters through the adjust_analysis of the capture, with ssb_case None selecting the TS 38.213 §4.1 case for the spacing ('c' at 30 kHz). Only a subcarrier_spacing of 30 kHz is accepted, and tdd_frame_slots is 1 character per slot of the 10 ms frame or 1 character for every slot.

Parameters:

  • center_frequency (float) – RF center frequency (Constraints: > 0, units: Hz)

  • channel_bandwidth (float) – nominal 5G NR channel bandwidth (3GPP TS 38.104 §5.3) (Constraints: > 0, units: Hz)

  • subcarrier_spacing (float) – Subcarrier spacing (15e3, 30e3, 60e3, etc) (Constraints: units: Hz)

  • ssb_frequency_offset (float) – offset of the SSB center from the channel center frequency (Constraints: units: Hz)

  • ssb_case (one of ['auto', 'a', 'b', 'c', 'd', 'e', 'f', 'g'] | tuple[int, ...] | None) – index locations in the SSB, or a cell search case from 3GPP TS 38.213 Sec. 4.1)

  • channel_guard_bandwidths (tuple[float, float]) – guard bandwidths excluded at the (low, high) edges of the channel (Constraints: units: Hz)

  • tdd_frame_slots (str | None) – string composed of (‘d’, ‘u’, ‘s’) specifying the sequence of slots in 1 TDD cellular frame, or None to fill with downlink

  • tdd_special_symbols (str | None) – string composed of the characters (‘d’, ‘u’, ‘f’) indicating the sequence of symbol types (when ‘s’ is in frame_slots)

  • min_sample_rate (float | None) – lower bound on the sample rate derived for the channel; None applies no bound (Constraints: > 0, units: S/s)

class striqt.sensor.specs.Cellular5GNRAnalysis(*, spectrogram=None, power_spectral_density=None, cellular_resource_power_histogram=None, cellular_5g_pss_correlation=None, cellular_5g_pss_sync=None, cellular_5g_ssb_spectrogram=None, cyclic_channel_power=None, channel_power_time_series=None, channel_power_histogram=None, cellular_cyclic_autocorrelation=None, iq_waveform=None)

Bases: SpecBase

the per-measurement settings of a Cellular5GNRSweep.

Each field is named for a measurement, and None disables the measurement. A block enables the measurement and exposes only the parameters that the channel does not fix. The remaining parameters are set from the cell_channels entry at the center frequency of the capture and from the subcarrier spacing, or stay at the measurement default.

Parameters:
class striqt.sensor.specs.Cellular5GNRSweep(*, description=Description(summary=None, version='unversioned'), source, peripherals=Peripherals(), extensions=Extension(import_path=None, import_name=None), sink=ZarrSink(path='{spec_name}-{start_time}', log_path=None, log_level='info', store='directory', max_threads=None, batched_write_count=1, max_chunk_bytes=50000000), captures=(), loops=(), adjust_captures={}, plot_hint=None, options=SweepOptions(reuse_iq=False, loop_only_nyquist=False, skip_warmup=True), cell_channels, analysis=Cellular5GNRAnalysis(spectrogram=None, power_spectral_density=None, cellular_resource_power_histogram=None, cellular_5g_pss_correlation=None, cellular_5g_pss_sync=None, cellular_5g_ssb_spectrogram=None, cyclic_channel_power=None, channel_power_time_series=None, channel_power_histogram=None, cellular_cyclic_autocorrelation=None, iq_waveform=None))

Bases: CapturePlan[SS, SP, SC, SA]

the spec of a sweep over 5G NR channels.

Each entry of captures names a channel by center_frequency, and cell_channels maps a label to the CellChannel at that frequency. A capture leaves sample_rate and analysis_bandwidth at the class defaults, and the matched channel sets both. A loop over center_frequency is rejected, and every channel in use must share one subcarrier spacing.

Parameters:
Variables:

  • schema (Any)

  • sensor (Any)

  • controller (Any)

  • sweep_cls (Any)

classmethod check_schema(schema: Any) → None

raise unless schema can configure a sweep over cell channels.

The capture type must have a center_frequency field, and schema.analysis must declare every field of Cellular5GNRAnalysis with the spec type of the BuiltinAnalyses field of that name.

Raises:

TypeError – if either requirement fails

class striqt.sensor.specs.CellularChannelPowerHistogramSpec(*, power_low, power_high, power_resolution, detector_period=None, power_detectors=('rms', 'peak'))

Bases: SpecBase

Parameters:

  • power_low (float) – Minimum power bin (Constraints: units: dB power)

  • power_high (float) – Maximum power bin (Constraints: units: dB power)

  • power_resolution (float) – Power bin resolution (Constraints: units: dB)

  • detector_period (Fraction | None) – Power detector bin duration, a whole multiple of the sample period (Constraints: units: s)

  • power_detectors (tuple[str, ...]) – Power detectors evaluated in each bin: ‘rms’, ‘peak’, or both

class striqt.sensor.specs.CellularChannelPowerTimeSeriesSpec(*, detector_period=None, power_detectors=('rms', 'peak'))

Bases: SpecBase

Parameters:

  • detector_period (Fraction | None) – Power detector bin duration, a whole multiple of the sample period (Constraints: units: s)

  • power_detectors (tuple[str, ...]) – Power detectors evaluated in each bin: ‘rms’, ‘peak’, or both

class striqt.sensor.specs.CellularCyclicAutocorrelationSpec(*, frame_range=(0, 1), symbol_range=(0, None))

Bases: SpecBase

Parameters:

  • frame_range (int | tuple[int, int])

  • symbol_range (int | tuple[int, int | None])

class striqt.sensor.specs.CellularCyclicChannelPowerSpec(*, power_detectors=('rms', 'peak'), cyclic_statistics=('min', 'mean', 'max'))

Bases: SpecBase

Parameters:

  • power_detectors (tuple[str, ...]) – Power detectors evaluated in each bin: ‘rms’, ‘peak’, or both

  • cyclic_statistics (tuple[str | float, ...]) – Statistics evaluated across cycles at each cycle lag: ‘min’, ‘mean’, ‘max’, ‘median’, or a quantile in (0, 1)

class striqt.sensor.specs.CellularIQWaveformSpec(*, start_time_sec=None, stop_time_sec=None)

Bases: SpecBase

Parameters:

  • start_time_sec (float | None)

  • stop_time_sec (float | None)

class striqt.sensor.specs.CellularPSSCorrelationSpec(*, discovery_periodicity=0.02, delay=0.0, shared_spectrum=False, max_block_count=1)

Bases: SpecBase

Parameters:

  • discovery_periodicity (float)

  • delay (float)

  • shared_spectrum (bool)

  • max_block_count (int | None)

class striqt.sensor.specs.CellularPSSSyncSpec(*, discovery_periodicity=0.02, delay=0.0, shared_spectrum=False, per_port=False, max_beams=None)

Bases: SpecBase

Parameters:

  • discovery_periodicity (float)

  • delay (float)

  • shared_spectrum (bool)

  • per_port (bool) – whether to evaluate signal synchronization separately on each port

  • max_beams (int | None) – limit the beam count in 5G sync evaluation (Constraints: > 0)

class striqt.sensor.specs.CellularPowerSpectralDensitySpec(*, window, time_statistic=('mean',), lo_bandstop=None, trim_stopband=True)

Bases: SpecBase

Parameters:

  • window (str | tuple[str, float]) – window function specification following scipy.signal.get_window

  • time_statistic (tuple[str | float, ...])

  • lo_bandstop (float | None) – mask with float(“nan”) at this bandwidth centered at baseband DC (Constraints: units: Hz)

  • trim_stopband (bool)

class striqt.sensor.specs.CellularResourcePowerHistogramSpec(*, window, power_low, power_high, power_resolution, average_rbs=False, average_slots=False, cyclic_prefix='normal', lo_bandstop=None)

Bases: SpecBase

Parameters:

  • window (str | tuple[str, float]) – window function specification following scipy.signal.get_window

  • power_low (float) – Minimum power bin (Constraints: units: dB power)

  • power_high (float) – Maximum power bin (Constraints: units: dB power)

  • power_resolution (float) – Power bin resolution (Constraints: units: dB)

  • average_rbs (bool | one of ['half']) – frequency resolution of the spectrogram: False for 1 subcarrier, True for 1 resource block (12 subcarriers), or ‘half’ for 6 subcarriers

  • average_slots (bool) – True to coarsen spectrogram bins by averaging 1-symbol time resolution

  • cyclic_prefix (one of ['normal', 'extended']) – cyclic prefix length of the 3GPP numerology (TS 38.211 Table 4.2-1)

  • lo_bandstop (float | None) – mask with float(“nan”) at this bandwidth centered at baseband DC (Constraints: units: Hz)

class striqt.sensor.specs.CellularSSBSpectrogramSpec(*, window='blackmanharris', discovery_periodicity=0.02, max_block_count=None, lo_bandstop=None)

Bases: SpecBase

Parameters:

  • window (str | tuple[str, float]) – window function specification following scipy.signal.get_window

  • discovery_periodicity (float)

  • max_block_count (int | None)

  • lo_bandstop (float | None) – mask with float(“nan”) at this bandwidth centered at baseband DC (Constraints: units: Hz)

class striqt.sensor.specs.CellularSpectrogramSpec(*, window, lo_bandstop=None, time_aperture=None, trim_stopband=True)

Bases: SpecBase

Parameters:

  • window (str | tuple[str, float]) – window function specification following scipy.signal.get_window

  • lo_bandstop (float | None) – mask with float(“nan”) at this bandwidth centered at baseband DC (Constraints: units: Hz)

  • time_aperture (float | None)

  • trim_stopband (bool)

class striqt.sensor.specs.DataOptions(*, groupby_dims=(), sweep_index, query=None, select={})

Bases: SpecBase

Parameters:

  • groupby_dims (tuple[str, ...])

  • sweep_index (int)

  • query (str | None)

  • select (dict[str, Any])

class striqt.sensor.specs.Description(*, summary=None, version='unversioned')

Bases: SpecBase

Parameters:

  • summary (str | None)

  • version (str)

class striqt.sensor.specs.DiracDeltaCapture(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis={}, external_lo_frequency=None, time=0, power=0)

Bases: SensorCapture

Parameters:

  • duration (float) – Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float) – Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float) – Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, ...]) – Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of ['left', 'right', 'none']) – LO shift direction

  • host_resample (bool)

  • backend_sample_rate (float | None) – Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]) – changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, ...]) – LO frequency of external frequency converter

  • time (float) – start time offset (Constraints: units: s)

  • power (float) – peak power level (Constraints: units: dB)

class striqt.sensor.specs.Extension(*, import_path=None, import_name=None)

Bases: SpecBase

the extensions: block of an action spec.

import_path is inserted into sys.path and import_name is imported before the rest of the spec is decoded. The module may bind the sensor_binding that the spec names.

Parameters:

  • import_path (str | None) – path to append to sys.path before extension imports

  • import_name (str | None) – name of the extension module that calls bind_sensor

class striqt.sensor.specs.FileAcquisitionInfo(*, source_id='', sweep_index=0, capture_index=0, signal_trigger=None, center_frequency=nan, backend_sample_rate, port=0, gain=nan)

Bases: AcquisitionInfo

Parameters:

  • source_id (str) – Source UUID string

  • sweep_index (int)

  • capture_index (int)

  • signal_trigger (Trigger | None)

  • center_frequency (float | tuple[float, ...]) – Center frequency for each port (Constraints: units: Hz, > 0, units: Hz, > 0, units: Hz)

  • backend_sample_rate (float) – Source sample rate (Constraints: > 0, units: S/s)

  • port (int | tuple[int, ...]) – Input port indices (Constraints: >= 0, >= 0)

  • gain (float | tuple[float, ...]) – Gain setting for each channel (Constraints: units: dB, units: dB, units: dB)

class striqt.sensor.specs.FileCapture(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis={}, external_lo_frequency=None)

Bases: SensorCapture

Capture specification read from a file, with support for None sentinels

Parameters:

  • duration (float) – Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float) – Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float) – Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, ...]) – Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of ['left', 'right', 'none']) – LO shift direction

  • host_resample (bool)

  • backend_sample_rate (float | None) – Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]) – changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, ...]) – LO frequency of external frequency converter

class striqt.sensor.specs.FrequencyBinRange(*, field, isin='capture', start, stop, step)

Bases: LoopBase

Parameters:

  • field (str)

  • isin (one of ['capture', 'analysis']) – selects whether to loop a capture or analysis parameters

  • start (float)

  • stop (float)

  • step (float)

get_points() → list[float]
class striqt.sensor.specs.FunctionSource(*, master_clock_rate, trigger_strobe=None, signal_trigger=None, array_backend='numpy', num_rx_ports)

Bases: Source

Parameters:

  • master_clock_rate (float) – Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None)

  • signal_trigger (str | BuiltinTriggers | None)

  • array_backend (one of ['numpy', 'cupy']) – array module to use to set compute device: numpy = cpu, cupy = gpu

  • num_rx_ports (int)

Variables:

  • transient_holdoff_time (float) – (Value: 0)

  • stream_all_rx_ports (bool) – (Value: False)

  • transport_dtype (one of ['int16', 'float32', 'complex64']) – data transfer type to use inside the source (Value: complex64)

class striqt.sensor.specs.List(*, field, isin='capture', values)

Bases: LoopBase

Parameters:

  • field (str)

  • isin (one of ['capture', 'analysis']) – selects whether to loop a capture or analysis parameters

  • values (tuple[dict[str | one of ['defaults'], dict[str, CaptureRemap | float | int | str | bool | None]] | str | float | int | bool | None, ...]) – Source UUID string

get_points() → list
class striqt.sensor.specs.LoopBase(*, field, isin='capture')

Bases: SpecBase

Parameters:

  • field (str | None)

  • isin (one of ['capture', 'analysis']) – selects whether to loop a capture or analysis parameters

get_points() → list
class striqt.sensor.specs.MATSource(*, master_clock_rate, trigger_strobe=None, signal_trigger=None, array_backend='numpy', path, file_format='auto', file_metadata=None, loop=False, key=None)

Bases: Source

Parameters:

  • master_clock_rate (float) – Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None)

  • signal_trigger (str | BuiltinTriggers | None)

  • array_backend (one of ['numpy', 'cupy']) – array module to use to set compute device: numpy = cpu, cupy = gpu

  • path (str) – path to the waveform data file

  • file_format (one of ['auto', 'mat', 'tdms']) – data format or auto to guess by extension

  • file_metadata (dict[str, Any] | None) – any capture fields not included in the file

  • loop (bool) – whether to loop the file to create longer IQ waveforms

  • key (str | None)

Variables:

  • transient_holdoff_time (float) – (Value: 0)

  • stream_all_rx_ports (bool | None) – (Value: False)

  • transport_dtype (one of ['int16', 'float32', 'complex64']) – data transfer type to use inside the source (Value: complex64)

class striqt.sensor.specs.ManualYFactorCalibration(*, description=Description(summary=None, version='unversioned'), source, peripherals=Peripherals(), extensions=Extension(import_path=None, import_name=None), sink=Sink(path='{spec_name}-{start_time}', log_path=None, log_level='info'), captures=(), loops=(), adjust_captures={}, calibration, analysis=None)

Bases: CapturePlan[SS, SP, SC, SA]

the spec of a Y-factor calibration with a hand-switched noise diode.

An analysis block is accepted, discarded, and reported with a DeprecationWarning. More than one capture is accepted only when calibration.implied_loops names the fields that the captures vary, and source.calibration must be None. A bound spec is also rejected when the runner cannot derive the calibration sweep.

Parameters:
Variables:

  • schema (Any)

  • sensor (Any)

  • controller (Any)

  • sweep_cls (Any)

classmethod check_schema(schema: Any) → None

raise unless schema.analysis declares channel_power_time_series.

striqt.sensor.specs.ManualYFactorPeripheral

alias of ManualYFactorSpec

class striqt.sensor.specs.ManualYFactorSpec(*, enr, ambient_temperature, implied_loops=(), detector_period=None)

Bases: Peripherals

the calibration: block of a ManualYFactorCalibration.

implied_loops names the capture fields that the operator varies by hand between captures, and the saved corrections are indexed by these fields ahead of the loop fields. detector_period None selects the one duration of the captures and loops as the bin of the generated channel_power_time_series analysis, and a bound spec with more than one duration is then rejected. enr is referred to a diode-off source temperature of 290 K, and ambient_temperature is recorded with the spec without entering the corrections.

Parameters:

  • enr (float) – Excess noise ratio (Constraints: units: dB)

  • ambient_temperature (float) – Ambient temperature (Constraints: units: K)

  • implied_loops (tuple[str, ...])

  • detector_period (Fraction | None) – Power detector bin duration, a whole multiple of the sample period (Constraints: units: s)

striqt.sensor.specs.Meta(standard_name: str, units: str | None = None, **kws) → Meta

annotation that is used to generate ‘standard_name’ and ‘units’ fields of xarray attrs objects

class striqt.sensor.specs.NoPeripherals

Bases: Peripherals

class striqt.sensor.specs.NoSource(*, master_clock_rate, trigger_strobe=None, signal_trigger=None, array_backend='numpy', num_rx_ports)

Bases: Source

Parameters:

  • master_clock_rate (float) – Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None)

  • signal_trigger (str | BuiltinTriggers | None)

  • array_backend (one of ['numpy', 'cupy']) – array module to use to set compute device: numpy = cpu, cupy = gpu

  • num_rx_ports (int)

Variables:

  • transient_holdoff_time (float) – (Value: 0)

  • stream_all_rx_ports (bool) – (Value: False)

  • transport_dtype (one of ['int16', 'float32', 'complex64']) – data transfer type to use inside the source (Value: float32)

class striqt.sensor.specs.NoiseCapture(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis={}, external_lo_frequency=None, noise_psd=1e-17)

Bases: SensorCapture

Parameters:

  • duration (float) – Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float) – Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float) – Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, ...]) – Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of ['left', 'right', 'none']) – LO shift direction

  • host_resample (bool)

  • backend_sample_rate (float | None) – Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]) – changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, ...]) – LO frequency of external frequency converter

  • noise_psd (float) – noise total channel power (Constraints: >= 0, units: mW/Hz)

class striqt.sensor.specs.Peripherals

Bases: SpecBase

class striqt.sensor.specs.PlotOptions(*, data, plotter, variables={})

Bases: SpecBase

Parameters:
class striqt.sensor.specs.Range(*, field, isin='capture', start, stop, step)

Bases: LoopBase

Parameters:

  • field (str)

  • isin (one of ['capture', 'analysis']) – selects whether to loop a capture or analysis parameters

  • start (float)

  • stop (float)

  • step (float)

get_points() → list
class striqt.sensor.specs.Repeat(*, field=None, isin='capture', count=1)

Bases: LoopBase

Parameters:

  • field (None)

  • isin (one of ['capture', 'analysis']) – selects whether to loop a capture or analysis parameters

  • count (int)

get_points() → list[int]
class striqt.sensor.specs.SawtoothCapture(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis={}, external_lo_frequency=None, period=0.01, power=0)

Bases: SensorCapture

Parameters:

  • duration (float) – Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float) – Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float) – Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, ...]) – Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of ['left', 'right', 'none']) – LO shift direction

  • host_resample (bool)

  • backend_sample_rate (float | None) – Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]) – changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, ...]) – LO frequency of external frequency converter

  • period (float) – waveform period (Constraints: >= 0, units: s)

  • power (float) – peak power level (Constraints: units: dB)

class striqt.sensor.specs.Schema(source: type[~striqt.sensor.lib.typing.SS], capture: type[~striqt.sensor.lib.typing.SC], peripherals: type[~striqt.sensor.lib.typing.SP], init_like: ~typing.Callable[[~PS], ~typing.Any], arm_like: ~typing.Callable[[~PC], ~typing.Any], action: type[~striqt.sensor.specs.ActionSpec[~striqt.sensor.lib.typing.SS, ~striqt.sensor.lib.typing.SP, ~striqt.sensor.lib.typing.SA]] = <class 'striqt.sensor.specs.Sweep'>, analysis: type[~striqt.sensor.lib.typing.SA] = <class 'striqt.analysis.specs.BuiltinAnalyses'>)

Bases: Generic[SS, SP, SC, SA, PS, PC]

the spec types of a sensor binding.

action

alias of Sweep

analysis

alias of BuiltinAnalyses

arm_like: Callable[PC, Any]
capture: type[SC]
init_like: Callable[PS, Any]
peripherals: type[SP]
source: type[SS]
class striqt.sensor.specs.SensorCapture(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis={}, external_lo_frequency=None)

Bases: Capture

Capture specification for a generic waveform with resampling support

Parameters:

  • duration (float) – Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float) – Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float) – Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, ...]) – Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of ['left', 'right', 'none']) – LO shift direction

  • host_resample (bool)

  • backend_sample_rate (float | None) – Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]) – changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, ...]) – LO frequency of external frequency converter

class striqt.sensor.specs.SharedPlotOptions(*, style=None, col='port', row=None, col_label_format='Port {port}', row_label_format=None, col_wrap=None, suptitle_fmt='{start_time}', filename_fmt='{name} {start_time}.svg')

Bases: SpecBase

Parameters:

  • style (str | None)

  • col (str | None)

  • row (str | None)

  • col_label_format (str)

  • row_label_format (str | None)

  • col_wrap (int | None)

  • suptitle_fmt (str)

  • filename_fmt (str)

class striqt.sensor.specs.SingleToneCapture(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis={}, external_lo_frequency=None, frequency_offset=0, snr=None)

Bases: SensorCapture

Parameters:

  • duration (float) – Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float) – Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float) – Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, ...]) – Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of ['left', 'right', 'none']) – LO shift direction

  • host_resample (bool)

  • backend_sample_rate (float | None) – Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]) – changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, ...]) – LO frequency of external frequency converter

  • frequency_offset (float) – Baseband frequency offset (Constraints: units: Hz)

  • snr (float | None) – Add circular white gaussian noise to achieve this SNR (Constraints: units: dB)

class striqt.sensor.specs.Sink(*, path='{spec_name}-{start_time}', log_path=None, log_level='info')

Bases: SpecBase

the output block of every action spec.

path is the output location, with the {field} templates of paths.get_path_fields expanded, and None writes nothing. log_path and log_level configure the log file independently of path.

Parameters:

  • path (str | None)

  • log_path (str | None)

  • log_level (str)

class striqt.sensor.specs.SoapyAcquisitionInfo(*, source_id='', sweep_index=0, capture_index=0, signal_trigger=None, sweep_start_time=None, start_time, backend_sample_rate)

Bases: AcquisitionInfo

extra coordinate information returned from an acquisition

Parameters:

  • source_id (str) – Source UUID string

  • sweep_index (int)

  • capture_index (int)

  • signal_trigger (Trigger | None)

  • sweep_start_time (Timestamp | None) – Capture acquisition start time

  • start_time (Timestamp | None) – Acquisition start time of the first capture

  • backend_sample_rate (float | None) – Source sample rate (Constraints: > 0, units: S/s)

class striqt.sensor.specs.SoapyCapture(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, port, lo_shift='none', host_resample=True, backend_sample_rate=None, adjust_analysis={}, external_lo_frequency=None, center_frequency, gain)

Bases: SensorCapture

Parameters:

  • duration (float) – Duration of the analysis waveform (Constraints: units: s)

  • sample_rate (float) – Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

  • analysis_bandwidth (float) – Analysis bandwidth (Constraints: units: Hz)

  • port (int | tuple[int, ...]) – Input port indices (Constraints: >= 0, >= 0)

  • lo_shift (one of ['left', 'right', 'none']) – LO shift direction

  • host_resample (bool)

  • backend_sample_rate (float | None) – Source sample rate (Constraints: > 0, units: S/s)

  • adjust_analysis (dict[str, Any]) – changes to apply across each analysis for any matching parameter

  • external_lo_frequency (float | None | tuple[float | None, ...]) – LO frequency of external frequency converter

  • center_frequency (float | tuple[float, ...]) – Center frequency for each port (Constraints: units: Hz, > 0, units: Hz, > 0, units: Hz)

  • gain (float | tuple[float, ...]) – Gain setting for each channel (Constraints: units: dB, units: dB, units: dB)

class striqt.sensor.specs.SoapySource(*, master_clock_rate, trigger_strobe=None, signal_trigger=None, array_backend='numpy', calibration=None, time_source='host', time_sync_at='acquire', clock_source='internal', receive_retries=0, adc_overload_limit=-1, if_overload_limit=None, gapless=False)

Bases: Source

Parameters:

  • master_clock_rate (float) – Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None)

  • signal_trigger (str | BuiltinTriggers | None)

  • array_backend (one of ['numpy', 'cupy']) – array module to use to set compute device: numpy = cpu, cupy = gpu

  • calibration (str | None)

  • time_source (one of ['host', 'internal', 'external', 'gps']) – Hardware source for timestamps

  • time_sync_at (one of ['open', 'acquire']) – when to sync the hardware clock: on connection, or before each capture

  • clock_source (one of ['internal', 'external', 'gps']) – Hardware source for the frequency reference

  • receive_retries (int) – number of acquisition retry attempts on stream error (Constraints: >= 0)

  • adc_overload_limit (float | None) – dataset adc_overload=True when the peak ADC level exceeds this threshold (Constraints: <= 0, units: dBfs)

  • if_overload_limit (float | None) – the maximum level at maximum gain before IF/baseband overload (Constraints: units: dBfs)

  • gapless (bool) – whether to raise an exception on overflows between identical captures

Variables:

  • transient_holdoff_time (float) – (Value: 0)

  • stream_all_rx_ports (bool | None) – (Value: False)

  • transport_dtype (one of ['int16', 'float32', 'complex64']) – data transfer type to use inside the source (Value: float32)

rx_enable_delay = 0.0
shared_rx_sample_clock = True
class striqt.sensor.specs.Source(*, master_clock_rate, trigger_strobe=None, signal_trigger=None, array_backend='numpy')

Bases: SpecBase

run-time characteristics of the radio that are left invariant during a sweep

Parameters:

  • master_clock_rate (float) – Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None)

  • signal_trigger (str | BuiltinTriggers | None)

  • array_backend (one of ['numpy', 'cupy']) – array module to use to set compute device: numpy = cpu, cupy = gpu

Variables:

  • transient_holdoff_time (float) – (Value: 0)

  • stream_all_rx_ports (bool | None) – (Value: False)

  • transport_dtype (one of ['int16', 'float32', 'complex64']) – data transfer type to use inside the source (Value: float32)

calibration = None
gapless = False
class striqt.sensor.specs.SourceInfo(*, num_rx_ports, retries=None)

Bases: SpecBase

Parameters:

  • num_rx_ports (int | None)

  • retries (int | None)

min_port_count(tuple_size: int)
class striqt.sensor.specs.SpecBase

Bases: Struct

the base of every spec struct.

An instance is a frozen, keyword-only msgspec.Struct that rejects unknown fields. Construction converts list and dict field values to tuple and frozendict. An instance round-trips through to_dict and from_dict unchanged.

classmethod from_dict(d: dict | frozendict) → _T

return an instance of cls converted from the field values in d.

Parameters:

d – field values keyed by name; a float for a fractions.Fraction field converts to the fraction of its decimal representation

Raises:

msgspec.ValidationError – when a required field is missing, a key is not a field, or a value does not convert to the annotated type

classmethod from_spec(other: SpecBase) → _T
replace(**attrs) → Self

returns a copy of self with changed attributes.

See also

Python standard library copy.replace

to_dict(unfreeze: bool = False, allow_tuple_keys: bool = True) → dict

return the field values of self as a dict of builtin values.

Parameters:

  • unfreeze – True converts nested tuple and frozendict values to list and dict

  • allow_tuple_keys – False renders tuple dict keys as JSON array strings

Returns:

a dict keyed by field name; fractions.Fraction values are strings and frozendict values are dicts

validate() → Self

return an equal instance rebuilt through to_dict and from_dict.

Raises:

msgspec.ValidationError – when a field value does not satisfy the annotated type or constraints

exception striqt.sensor.specs.SpecValidationError(message: str, path: tuple[str, ...] = (), locations: tuple[str, ...] = ())

Bases: ValidationError

a spec validation failure with a msgspec-style field path.

path is the chain of field accesses from the document root, rendered as msgspec renders a path ($.analysis.spectrogram) ahead of message. locations are further roots at which the same failure is located, rendered as an at ... on ... trailer in the order given. msgspec propagates an instance raised in __post_init__ unchanged.

at(*locations: str) → SpecValidationError

return a copy with locations prepended to the at ... on ... trailer

prepend(*parts: str) → SpecValidationError
class striqt.sensor.specs.Sweep(*, description=Description(summary=None, version='unversioned'), source, peripherals=Peripherals(), extensions=Extension(import_path=None, import_name=None), sink=ZarrSink(path='{spec_name}-{start_time}', log_path=None, log_level='info', store='directory', max_threads=None, batched_write_count=1, max_chunk_bytes=50000000), captures=(), loops=(), adjust_captures={}, plot_hint=None, analysis=BuiltinAnalyses(channel_power_time_series=None, channel_power_histogram=None, cyclic_channel_power=None, spectrogram=None, power_spectral_density=None, spectrogram_histogram=None, spectrogram_ratio_histogram=None, cellular_5g_pss_sync=None, cellular_5g_pss_correlation=None, cellular_5g_sss_sync=None, cellular_5g_sss_correlation=None, cellular_5g_ssb_spectrogram=None, cellular_cyclic_autocorrelation=None, cellular_resource_power_histogram=None, iq_waveform=None), options=SweepOptions(reuse_iq=False, loop_only_nyquist=False, skip_warmup=True))

Bases: CapturePlan[SS, SP, SC, SA]

the spec of a sweep, and the form every action opens resources for.

analysis is an instance of the group that Schema.analysis names for the binding, and is evaluated on each expanded capture with the adjust_analysis of the capture applied on top. options selects the warmup, IQ reuse, and the loop_only_nyquist filter of the expansion. Construction validates each unique capture against the registered measurements and the conditioning signal path (see action.validate_sweep).

Parameters:
Variables:

  • schema (Any)

  • sensor (Any)

  • controller (Any)

  • sweep_cls (Any)

class striqt.sensor.specs.SweepOptions(*, reuse_iq=False, loop_only_nyquist=False, skip_warmup=True)

Bases: SpecBase

Parameters:

  • reuse_iq (bool)

  • loop_only_nyquist (bool)

  • skip_warmup (bool) – if True, suppress empty buffer runs for GPU backends

class striqt.sensor.specs.TDMSSource(*, master_clock_rate, trigger_strobe=None, signal_trigger=None, array_backend='numpy', path)

Bases: Source

Parameters:

  • master_clock_rate (float) – Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None)

  • signal_trigger (str | BuiltinTriggers | None)

  • array_backend (one of ['numpy', 'cupy']) – array module to use to set compute device: numpy = cpu, cupy = gpu

  • path (str) – path to the waveform data file

Variables:

  • transient_holdoff_time (float) – (Value: 0)

  • stream_all_rx_ports (bool | None) – (Value: False)

  • transport_dtype (one of ['int16', 'float32', 'complex64']) – data transfer type to use inside the source (Value: complex64)

class striqt.sensor.specs.Trigger(name_or_func: str | Callable, spec: Analysis, registry: AnalysisRegistry)

Bases: object

classmethod from_spec(name: str, analysis: specs.AnalysisGroup, registry: AnalysisRegistry) → Self
info: SyncInfo
max_lag(capture: Capture) → int
class striqt.sensor.specs.ZarrIQSource(*, master_clock_rate, trigger_strobe=None, signal_trigger=None, array_backend='numpy', path, center_frequency, select={})

Bases: Source

Parameters:

  • master_clock_rate (float) – Base sample rate used inside the source (Constraints: > 0, units: Hz)

  • trigger_strobe (float | None)

  • signal_trigger (str | BuiltinTriggers | None)

  • array_backend (one of ['numpy', 'cupy']) – array module to use to set compute device: numpy = cpu, cupy = gpu

  • path (str) – path to the waveform data file

  • center_frequency (float | tuple[float, ...]) – Center frequency for each port (Constraints: units: Hz, > 0, units: Hz, > 0, units: Hz)

  • select (dict) – dictionary to select in the data as .sel(**select)

Variables:

  • transient_holdoff_time (float) – (Value: 0)

  • stream_all_rx_ports (bool | None) – (Value: False)

  • transport_dtype (one of ['int16', 'float32', 'complex64']) – data transfer type to use inside the source (Value: complex64)

class striqt.sensor.specs.ZarrSink(*, path='{spec_name}-{start_time}', log_path=None, log_level='info', store='directory', max_threads=None, batched_write_count=1, max_chunk_bytes=50000000)

Bases: Sink

the output block of a Sweep: a zarr store appended along capture.

store selects a directory or zip store. The remaining fields tune the write batching and chunking of the store.

Parameters:

  • path (str | None)

  • log_path (str | None)

  • log_level (str)

  • store (one of ['zip', 'directory']) – serialization output format of the sink

  • max_threads (int | None)

  • batched_write_count (int)

  • max_chunk_bytes (int)

class striqt.sensor.specs.frozendict(*args: Any, **kwargs: Any)

Bases: Mapping[_K, _V]

An immutable dictionary that supports hashing

copy() → frozendict[_K, _V]
classmethod fromkeys(seq: Iterable[_K], value: _V | None = None) → frozendict[_K, _V]
items() → a set-like object providing a view on D's items
keys() → a set-like object providing a view on D's keys
update(other: dict[_K, _V], /)
values() → an object providing a view on D's values

striqt.sensor.util

class striqt.sensor.util.DebugOnException(enable: bool = False, verbose: bool = False)

Bases: object

run(etype, exc, tb)
class striqt.sensor.util.ExceptionStack(group_label: str | None = None, cancel_on_except: bool = False)

Bases: object

context manager that collects the exceptions raised in defer blocks.

The collected exceptions are raised by handle, on exit from the context, and on garbage collection. With cancel_on_except, an exception in a defer block also calls cancel_threads.

defer()
handle() → None

raise the deferred exceptions and clear the stack.

No deferred exception: nothing is raised. One exception other than an interrupt: that exception is raised. Several: an ExceptionGroup of those exceptions is raised, labeled with group_label. ThreadInterruptRequest and KeyboardInterrupt are raised only when no other exception was deferred, and never inside an ExceptionGroup. A KeyboardInterrupt takes precedence over a ThreadInterruptRequest. The stack is empty after the call.

exception striqt.sensor.util.ThreadInterruptRequest

Bases: Exception

Raised in a thread to indicate the owning thread requested termination

striqt.sensor.util.await_and_ignore(futures: Iterable[concurrent.futures.Future], except_msg: str | None = None)
striqt.sensor.util.cancel_threads()
striqt.sensor.util.log_capture_context(name_suffix, /, capture_index=0, capture_count=None)
striqt.sensor.util.log_to_file(log_path: str | Path, level_name: str)
striqt.sensor.util.log_verbosity(verbose: int = 0)
striqt.sensor.util.propagate_thread_interrupts()
striqt.sensor.util.retry(excs: type[BaseException] | Iterable[type[BaseException]], tries: int, *, delay: float = 0, backoff: float = 0, exception_func: Callable[..., None] = <function <lambda>>, logger: logging.Logger | logging.LoggerAdapter | None = None) → PassThroughWrapper[P, R]

return a decorator that repeats a call on the exceptions excs.

The decorated function is called up to tries times. The result of the first call that raises no exception in excs is returned. An exception in excs on the last call is raised. Before each repeated call, exception_func is called with the arguments of the call. The thread then sleeps for the current delay.

Parameters:

  • excs – the exception class, or classes, that trigger a repeat

  • tries – the maximum number of calls, at least 1

  • delay – the sleep (in s) before the second call

  • backoff – the factor applied to the sleep after each repeat

  • exception_func – called with the arguments of the call before each repeat

  • logger – logs at INFO on the first repeat when given

Raises:

ValueError – if tries is less than 1

Examples

Repeat a telnet connection up to 5 times on ConnectionRefusedError:

>>> @retry(ConnectionRefusedError, tries=5)
... def open(host, port):
...     t = telnetlib.Telnet()
...     t.open(host, port, 5)
...     return t
striqt.sensor.util.share_thread_interrupts()
striqt.sensor.util.zip_offsets(seq: Iterable[_T], shifts: tuple[int, ...] | list[int], fill: Any, *, squeeze: bool = True) → Iterator[tuple[_T, ...]]

a generator that yields from seq at multiple index shifts.

Shifts that would yield an invalid index (i.e., before or beyond the end of seq) are replaced with fill.

Parameters:

  • seq – iterable of values to shift

  • shifts – a sequence of integers indicating the index shifts to apply

  • fill – the value to yield before or after seq yields values

Yields:

tuples of length shifts composed of delayed/advanced values from seq