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 | striqt.analysis.specs.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 | striqt.analysis.specs.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 | striqt.analysis.specs.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.Controller(reuse_iq=False, rx_ports: tuple[int, ...] | None=None, *args: ~PS, **kwargs: ~PS)

Bases: ControllerBase[SS, SC, PS, PC]

arm(*args: ~PC, **kwargs: ~PC)

stop the stream, apply a capture configuration, and start it

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: SS)

Bases: VirtualSource[MATSource, FileCapture]

returns IQ waveforms from a .mat file

property about
get_waveform(count: int, start: int, offset: int, *, port: int = 0, xp, dtype='complex64')
property id
class striqt.sensor.sources.NoSource(spec: NoSource)

Bases: SourceBackend[NoSource, SensorCapture]

fast paths to acquire empty buffers

property about
property id: str
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.RawController(spec: SS, *, reuse_iq: bool = False, rx_ports: tuple[int, ...] | None = None)

Bases: ControllerBase[SS, SC, PS, PC]

arm(spec: SC)
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: SS)

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.ZarrIQSource(spec: SS)

Bases: VirtualSource[ZarrIQSource, FileCapture]

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

property about
get_waveform(count: int, start: int, offset: int, *, port: int = 0, xp, dtype='complex64')
property id
class striqt.sensor.sources.lookup

Bases: object

asynchronous lookup of controller objects, connection status, or source ID.

This assumes that a source instantiation is in progress or will be within timeout seconds. Otherwise, TimeoutError is raised.

classmethod id(spec: Source, timeout=0.5) → str

lookup a source ID from a source specification.

classmethod instance(spec: Source, timeout=0.5) → ControllerBase
classmethod is_ready(spec: Source, timeout: float, wait: bool = True) → bool