Batched sensor acquisition

striqt.sensor

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

resample, filter, and apply calibration corrections.

Parameters:

  • iq – IQ dataclass output by a source

  • signal_trigger – a signal.analysis.Trigger instance to implement a waveform start alignment, or None to default to iq.info.signal_trigger

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

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

Returns:

the filtered IQ waveform

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

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

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

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

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

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

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

Parameters:

  • resources – dictionary of open resources returned by open_resources

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

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

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

Returns:

An iterator of analyzed data or None

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

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

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

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

Build a specs.Sweep specification object from a json file at the given path.

This supports the same specifications as a yaml file, but without !include directives.

Parameters:

  • path – path to the yaml file

  • type – the type of sweep specification to load, or None to use specs.Sweep

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

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

Returns:

an instance of specs.Sweep

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

Build a Sweep specification object from the specified yaml file.

Parameters:

  • path – path to the yaml file

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

  • type – the type of sweep specification to load, or None to use specs.Sweep

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

Returns:

an instance of specs.Sweep

striqt.sensor.bindings

A registry of data class <-> IQ acquisition control class bindings. These are taken to define sensors. Extension classes can implement peripherals and expanded data fields elsewhere.

—

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 | AnalysisGroup | 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): Analysis sample rate (Constraints: 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 | AnalysisGroup | 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, noise_diode_enabled=False)
Parameters:

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

  • sample_rate (float): Analysis sample rate (Constraints: 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)

  • noise_diode_enabled (bool): Noise diode enabled

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 | AnalysisGroup | 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): Analysis sample rate (Constraints: 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 | AnalysisGroup | 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, noise_diode_enabled=False)
Parameters:

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

  • sample_rate (float): Analysis sample rate (Constraints: 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)

  • noise_diode_enabled (bool): Noise diode enabled

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 | AnalysisGroup | 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): Analysis sample rate (Constraints: 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 | AnalysisGroup | 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, noise_diode_enabled=False)
Parameters:

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

  • sample_rate (float): Analysis sample rate (Constraints: 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)

  • noise_diode_enabled (bool): Noise diode enabled

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 | AnalysisGroup | 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): Analysis sample rate (Constraints: 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 | AnalysisGroup | 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): Analysis sample rate (Constraints: 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 | AnalysisGroup | 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): Analysis sample rate (Constraints: 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 | AnalysisGroup | 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): Analysis sample rate (Constraints: 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 | AnalysisGroup | 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): Analysis sample rate (Constraints: 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 | AnalysisGroup | 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): Analysis sample rate (Constraints: 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 | AnalysisGroup | 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): Analysis sample rate (Constraints: 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 sources of IQ samples from iq_waveform variables in 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 | AnalysisGroup | 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): Analysis sample rate (Constraints: 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, 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()
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, PS, PC]
classmethod from_sweep_spec(spec: Sweep[SS, Any, SC], format_path: PathFormatter | None = None) → Controller[SS, SP, SC, PS, PC]
get_resampler() → ResamplerDesign
is_open(wait=True) → bool
read_iq(overlaps: tuple[int, int] = (0, 0)) → tuple[Array, int | None]

read IQ for the armed capture

schema: specs.Schema[SS, SP, SC, PS, PC]
sensor: bindings.SensorBinding[SS, SP, SC]
set_calibration_formatter(format_path: PathFormatter | None = None)

set the formatter to use to expand calibration file formatting fields.

This is needed if calibration is defined in the source spec with format fields like {name}.

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[SS]', capture: 'type[SC]', peripherals: 'type[SP]', init_like: 'Callable[PS, Any]', arm_like: 'Callable[PC, Any]')

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

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[SourceBackend[SS, SC]]', sink_cls: 'type[sinks.SinkBase[SC]]' = <class 'striqt.sensor.lib.sinks.ZarrCaptureSink'>, sweep_spec_cls: 'type[specs.Sweep[SS, SP, SC]]'=<class 'striqt.sensor.specs.Sweep'>, peripherals_cls: 'type[Peripherals[SP, SC]]'=<class 'striqt.sensor.lib.peripherals.NoPeripherals'>)

Bases: Generic[SS, SP, SC]

peripherals_cls

alias of NoPeripherals

sink_cls

alias of ZarrCaptureSink

source_cls: type[SourceBackend[SS, SC]]
sweep_spec_cls

alias of Sweep

class striqt.sensor.bindings.SensorBinding(source_cls: 'type[SourceBackend[SS, SC]]', sink_cls: 'type[sinks.SinkBase[SC]]' = <class 'striqt.sensor.lib.sinks.ZarrCaptureSink'>, sweep_spec_cls: 'type[BoundSweep[SS, SP, SC]]'=<class 'striqt.sensor.specs.Sweep'>, peripherals_cls: 'type[Peripherals[SP, SC]]'=<class 'striqt.sensor.lib.peripherals.NoPeripherals'>)

Bases: Sensor[SS, SP, SC]

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

extend an existing binding with a y-factor calibration

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

register a binding between specifications and controller classes.

Parameters:

  • key – the key used when instantiating from yaml/json

  • sensor – the binding classes

striqt.sensor.calibration

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

Bases: SinkBase

append(capture_result)
close(*exc_info)
flush()
open() → None
striqt.sensor.calibration.bind_manual_yfactor_calibration(name: str, ctrl_cls: type[Controller[SS, SP, Any, PS, PC]]) → type[Controller[SS, SP, Any, PS, PC]]

extend an existing binding with a y-factor calibration

striqt.sensor.calibration.compute_y_factor_corrections(dataset: Dataset, Tref=290.0) → Dataset
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 power spectral density of the receive system noise

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

striqt.sensor.peripherals

striqt.sensor.sinks

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

Bases: SinkBase

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

Bases: Generic[SC]

intake acquisitions one at a time, and parcel data store

append(capture_result: DelayedDataset) → Dataset | DelayedDataset
close(*exc_info)
flush()
open() → None
pop() → list[Dataset]
submit(func, *args, **kws)
wait()
class striqt.sensor.sinks.ZarrCaptureSink(sweep_spec: Sweep[Any, Any, SC], format_path: PathFormatter | None = None, *, force: bool = False)

Bases: ZarrSinkBase

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

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

Bases: SinkBase

close(*exc_info)
get_root_path() → str
open()
class striqt.sensor.sinks.ZarrTimeAppendSink(sweep_spec: Sweep, format_path: PathFormatter | None = None, *, force: bool = False)

Bases: ZarrSinkBase

append(capture_result)
flush()

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 | AnalysisGroup | 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 | AnalysisGroup | 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 | AnalysisGroup | 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: int, offset: int, *, port: int = 0, xp, dtype='complex64')
class striqt.sensor.sources.MATSource(spec)

Bases: VirtualSource[MATSource, FileCapture]

returns IQ waveforms from a .mat file

get_waveform(count: int, start: int, offset: int, *, port: int = 0, xp, dtype='complex64')
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: int, offset: int, *, port: int = 0, xp, dtype='complex64')
class striqt.sensor.sources.SawtoothSource(spec: SS)

Bases: TestSourceBase[FunctionSource, SawtoothCapture]

get_waveform(count: int, start: int, offset: int, *, port: int = 0, xp, dtype='complex64')
class striqt.sensor.sources.SingleToneSource(spec: SS)

Bases: TestSourceBase[FunctionSource, SingleToneCapture]

get_waveform(count: int, start: int, offset: int, *, port: int = 0, xp, dtype='complex64')
class striqt.sensor.sources.SoapySource(spec: SS, **device_kwargs: Any)

Bases: SourceBackend[SS, SoapyCapture]

Applies SoapySDR for device control and acquisition

device: SoapySDR.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: int, offset: int, *, port: int = 0, xp, dtype='complex64')
class striqt.sensor.sources.VirtualSource(spec: SS)

Bases: SourceBackend[SS, SC]

get_waveform(count: int, start: int, offset: int, *, port: int = 0, xp, dtype='complex64') → Array
reset_sample_counter(value=0)
setup_spec: SS
class striqt.sensor.sources.ZarrIQSource(spec)

Bases: VirtualSource[ZarrIQSource, FileCapture]

a sources of IQ samples from iq_waveform variables in a zarr store

get_waveform(count: int, start: int, offset: int, *, port: int = 0, xp, dtype='complex64')

striqt.sensor.specs

schema for the specification of calibration and sweeps

class striqt.sensor.specs.AcquiredIQ(pre_align: Array, pre_filter: Array | None, aligned: Array | None, capture: specs.Capture | None, info: structs.AcquisitionInfo, extra_data: dict[str, Any], source_spec: structs.Source, resampler: sw.ResamplerDesign, format_path: helpers.PathFormatter | None = None, voltage_scale: sw.typing.Array | float = 1, conjugate: tuple[bool, ...] | False = False)

Bases: AcquiredIQ

extra metadata needed for downstream analysis

conjugate: tuple[bool, ...] | False = False
extra_data: dict[str, Any]
format_path: helpers.PathFormatter | None = None
info: structs.AcquisitionInfo
resampler: sw.ResamplerDesign
source_spec: structs.Source
voltage_scale: sw.typing.Array | float = 1
class striqt.sensor.specs.AcquisitionInfo(*, source_id='', sweep_index=None, capture_index=0, signal_trigger=None)

Bases: Struct

information about an acquired acquisition

Parameters:

  • source_id (str) – Source UUID string

  • sweep_index (int | None)

  • capture_index (int)

  • signal_trigger (str | 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 a dictinary representation of self

class striqt.sensor.specs.AnalysisGroup

Bases: SpecBase

base class for a defining set of Analysis specs

striqt.sensor.specs.BundledAnalysis

alias of Analysis

striqt.sensor.specs.BundledTriggers

alias of Analysis

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

Bases: Sweep[SS, SP, SC], Generic[SS, SP, SC, SPC]

This specialized sweep is fed to the YAML file loader to specify the change in expected capture structure.

Parameters:
Variables:

  • schema (Any)

  • sensor (Any)

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) – Analysis sample rate (Constraints: units: S/s)

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

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.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) – Analysis sample rate (Constraints: 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(*, sink=None, import_path=None, import_name=None)

Bases: SpecBase

Parameters:

  • sink (str | one of ['striqt.sensor.sinks.CaptureAppender'] | None) – Data sink class that implements data storage

  • 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=None, 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 | None)

  • capture_index (int)

  • signal_trigger (str | 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) – Analysis sample rate (Constraints: 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 | AnalysisGroup | 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] | tuple[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 | AnalysisGroup | 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.ManualYFactorPeripheral(*, enr, ambient_temperature, implied_loops=())

Bases: Peripherals

Parameters:

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

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

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

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 | AnalysisGroup | 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) – Analysis sample rate (Constraints: 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) – Analysis sample rate (Constraints: 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[SS]', capture: 'type[SC]', peripherals: 'type[SP]', init_like: 'Callable[PS, Any]', arm_like: 'Callable[PC, Any]')

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

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) – Analysis sample rate (Constraints: 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) – Analysis sample rate (Constraints: 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='{yaml_name}-{start_time}', log_path=None, log_level='info', store='directory', max_threads=None, batched_write_count=1, max_chunk_bytes=50000000)

Bases: SpecBase

Parameters:

  • path (str)

  • 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.SoapyAcquisitionInfo(*, source_id='', sweep_index=None, 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 | None)

  • capture_index (int)

  • signal_trigger (str | 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) – Analysis sample rate (Constraints: 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 | AnalysisGroup | 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 | AnalysisGroup | 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

Base type for structures that support validated (de)serialization.

It is a msgspec.Struct class with some often-used utility methods that fix encoding and decoding hooks for extra types.

classmethod from_dict(d: dict) → _T
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 a dictinary representation of self

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

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

Parameters:
Variables:

  • schema (Any)

  • sensor (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 | AnalysisGroup | 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: float32)

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 | AnalysisGroup | 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.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

Creates a context manager object that accumulates exceptions.

Any exception raised within a defer() context is stashed to be raised later. An exception (possibly an ExceptionGroup) is raised when the ExceptionDeferral context exits, or on a call to handle().

defer()
handle()

raise an exception based on any deferred exceptions.

The raised exception type follows these rules: - No exceptions: no exception is raised - Exactly one exception: raise that exception - More than one exception: raise an ExceptionGroup

A ThreadInterruptRequest will only be raised if it was the only deferred exception. ThreadInterruptRequest is never included in an ExceptionGroup.

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=<function <lambda>>, logger: logging.Logger | logging.LoggerAdapter | None = None) → PassThroughWrapper

calls to the decorated function are repeated, suppressing specified exception(s), until a maximum number of retries has been attempted.

If the function raises the exception the specified number of times, the underlying exception is raised. Otherwise, return the result of the function call.

Example

The following retries the telnet connection 5 times on ConnectionRefusedError:

import telnetlib


# Retry a telnet connection 5 times if the telnet library raises ConnectionRefusedError
@retry(ConnectionRefusedError, tries=5)
def open(host, port):
    t = telnetlib.Telnet()
    t.open(host, port, 5)
    return t
Parameters:

  • exception_or_exceptions – Exception (sub)class (or tuple of exception classes) to watch for

  • tries – number of times to try before giving up

  • delay – initial delay between retries in seconds

  • backoff – backoff to multiply to the delay for each retry

  • exception_func – function to call on exception before the next retry

  • logger – if specified, a log info message is emitted on the first retry

striqt.sensor.util.share_thread_interrupts()
striqt.sensor.util.zip_offsets(seq: Iterable[_T], shifts: tuple[int, ...] | list[int], fill: Any, *, squeeze: bool = True) → Generator[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