Batched waveform analysis¶

striqt.analysis¶

implementations of channel analysis measurements packaged into xarray.DataArray and xarray.Dataset

class striqt.analysis.AnalysisRegistry¶

Bases: dict[type[Analysis], AnalysisInfo]

a registry of keyword-only arguments for decorated functions

cache_context(capture: Capture, callback: Callable | None = None)¶

caches: dict[AnalysisFunc, list[KwArgCache]]¶

measurement(spec_type: type[specs.Analysis], *, dtype: str, name: str | None = None, dims: tuple[str, ...] | str | None = None, coord_factories: Iterable[WrappedCoord] | WrappedCoord | None = None, depends: Iterable[Callable] | Callable = [], caches: Iterable[KwArgCache] | KwArgCache | None = None, prefer_iq_source: PreferIQSource = 'aligned', store_compressed=True, attrs={}) → AnalysisFuncWrapper¶: add decorated func and its keyword arguments in the self.tostruct() schema

parameter_defaults: dict[str, Any]¶

parameter_fields: dict[str, 'msgspec.structs.FieldInfo|None']¶

tospec(base: type[AnalysisGroup] = <class 'striqt.analysis.specs.AnalysisGroup'>) → type[AnalysisGroup]¶

class striqt.analysis.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.analysis.EvaluationOptions(as_xarray, registry, block_each=True, expand_dims=())¶

Bases: Struct, Generic[TAR]

Parameters:

as_xarray (TAR)
registry (AnalysisRegistry)
block_each (bool)
expand_dims (Sequence[str])

class striqt.analysis.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¶

striqt.analysis.analyze_by_spec(iq: Array | AcquiredIQ, spec: dict[str, Analysis] | AnalysisGroup, capture: Capture, options: EvaluationOptions[Literal[True]]) → Dataset¶
striqt.analysis.analyze_by_spec(iq: Array | AcquiredIQ, spec: dict[str, Analysis] | AnalysisGroup, capture: Capture, options: EvaluationOptions[Literal['delayed']]) → dict[str, DelayedDataArray]
striqt.analysis.analyze_by_spec(iq: Array | AcquiredIQ, spec: dict[str, Analysis] | AnalysisGroup, capture: Capture, options: EvaluationOptions[Literal[False]]) → dict[str, Array]: evaluate a set of different channel analyses on the iq waveform as specified by spec

striqt.analysis.cellular_5g_pss_correlation(iq: Array, capture: Capture, *, subcarrier_spacing: Meta(description='Subcarrier spacing (15e3, 30e3, 60e3, etc)', extra={'standard_name': 'Subcarrier spacing (15e3, 30e3, 60e3, etc)', 'units': 'Hz'})], sample_rate: float = 7680000.0, discovery_periodicity: float = 0.02, frequency_offset: float = 0, shared_spectrum: bool = False, delay: float = 0, symbol_indexes: 1)'})] = 'auto', max_lag_symbols: Meta(ge=1, le=6, description='limit the number of symbols of lag in the 5G sync correlator', extra={'standard_name': 'limit the number of symbols of lag in the 5G sync correlator'})] | None = None, max_block_count: int | None = 1)¶

correlate each channel of the IQ against the cellular primary synchronization signal (PSS) waveform.

Returns a DataArray containing the time-lag for each combination of NID2, symbol, and SSB start time.

Parameters:

iq (Array) – input waveform array
capture (striqt.analysis.specs.Capture) – acquisition specification
ax_xarray (Any) – whether to return an xarray object (True), an array (False), or a delayed xarray object (‘delayed’)
subcarrier_spacing (float) – Subcarrier spacing (15e3, 30e3, 60e3, etc) (Constraints: units: Hz)
sample_rate (float)
discovery_periodicity (float)
frequency_offset (float)
shared_spectrum (bool)
delay (float)
symbol_indexes (one of ['auto', 'a', 'b', 'c', 'd', 'e', 'f', 'g'] | tuple[int, ...]) – index locations in the SSB, or a cell search case from 3GPP TS 38.213 Sec. 4.1)
max_lag_symbols (int | None) – limit the number of symbols of lag in the 5G sync correlator (Constraints: >= 1, <= 6)
max_block_count (int | None)

References

3GPP TS 138 211: Table 7.4.3.1-1, Section 7.4.2.2 3GPP TS 138 213: Section 4.1

striqt.analysis.cellular_5g_pss_sync(iq: Array, capture: Capture, *, subcarrier_spacing: Meta(description='Subcarrier spacing (15e3, 30e3, 60e3, etc)', extra={'standard_name': 'Subcarrier spacing (15e3, 30e3, 60e3, etc)', 'units': 'Hz'})], sample_rate: float = 7680000.0, discovery_periodicity: float = 0.02, frequency_offset: float = 0, shared_spectrum: bool = False, delay: float = 0, symbol_indexes: 1)'})] = 'auto', max_lag_symbols: Meta(ge=1, le=6, description='limit the number of symbols of lag in the 5G sync correlator', extra={'standard_name': 'limit the number of symbols of lag in the 5G sync correlator'})] | None = None, window_fill: Meta(gt=0, le=1, description='Fraction of a symbol to fill with weighting function', extra={'standard_name': 'Fraction of a symbol to fill with weighting function'})] = 1, per_port: Meta(description='whether to evaluate signal synchronization separately on each port', extra={'standard_name': 'whether to evaluate signal synchronization separately on each port'})] = False, max_beams: Meta(gt=0, description='limit the beam count in 5G sync evaluation', extra={'standard_name': 'limit the beam count in 5G sync evaluation'})] | None = None)¶: compute sync index offsets based on correlate_5g_pss

striqt.analysis.cellular_5g_ssb_spectrogram(iq: Array, capture: Capture, *, subcarrier_spacing: Meta(description='Subcarrier spacing (15e3, 30e3, 60e3, etc)', extra={'standard_name': 'Subcarrier spacing (15e3, 30e3, 60e3, etc)', 'units': 'Hz'})], sample_rate: float = 7680000.0, discovery_periodicity: float = 0.02, frequency_offset: float = 0, max_block_count: int | None = None, window: get_window`'})] = 'blackmanharris', lo_bandstop: float | None = None)¶

correlate each channel of the IQ against the cellular primary synchronization signal (PSS) waveform.

Returns a DataArray containing the time-lag for each combination of NID2, symbol, and SSB start time.

Parameters:

iq (Array) – input waveform array
capture (striqt.analysis.specs.Capture) – acquisition specification
ax_xarray (Any) – whether to return an xarray object (True), an array (False), or a delayed xarray object (‘delayed’)
subcarrier_spacing (float) – Subcarrier spacing (15e3, 30e3, 60e3, etc) (Constraints: units: Hz)
sample_rate (float)
discovery_periodicity (float)
frequency_offset (float)
max_block_count (int | None)
window (str | tuple[str, float]) – window function specification following scipy.signal.get_window
lo_bandstop (float | None)

References

3GPP TS 138 211: Table 7.4.3.1-1, Section 7.4.2.2 3GPP TS 138 213: Section 4.1

striqt.analysis.cellular_5g_sss_correlation(iq: Array, capture: Capture, *, subcarrier_spacing: Meta(description='Subcarrier spacing (15e3, 30e3, 60e3, etc)', extra={'standard_name': 'Subcarrier spacing (15e3, 30e3, 60e3, etc)', 'units': 'Hz'})], sample_rate: float = 7680000.0, discovery_periodicity: float = 0.02, frequency_offset: float = 0, shared_spectrum: bool = False, delay: float = 0, symbol_indexes: 1)'})] = 'auto', max_lag_symbols: Meta(ge=1, le=6, description='limit the number of symbols of lag in the 5G sync correlator', extra={'standard_name': 'limit the number of symbols of lag in the 5G sync correlator'})] | None = None, max_block_count: int | None = 1)¶

correlate each channel of the IQ against the cellular primary synchronization signal (SSS) waveform.

Returns a DataArray containing the time-lag for each combination of NID2, symbol, and SSB start time.

Parameters:

iq (Array) – input waveform array
capture (striqt.analysis.specs.Capture) – acquisition specification
ax_xarray (Any) – whether to return an xarray object (True), an array (False), or a delayed xarray object (‘delayed’)
subcarrier_spacing (float) – Subcarrier spacing (15e3, 30e3, 60e3, etc) (Constraints: units: Hz)
sample_rate (float)
discovery_periodicity (float)
frequency_offset (float)
shared_spectrum (bool)
delay (float)
symbol_indexes (one of ['auto', 'a', 'b', 'c', 'd', 'e', 'f', 'g'] | tuple[int, ...]) – index locations in the SSB, or a cell search case from 3GPP TS 38.213 Sec. 4.1)
max_lag_symbols (int | None) – limit the number of symbols of lag in the 5G sync correlator (Constraints: >= 1, <= 6)
max_block_count (int | None)

References

3GPP TS 138 211: Table 7.4.3.1-1, Section 7.4.2.2 3GPP TS 138 213: Section 4.1

striqt.analysis.cellular_cyclic_autocorrelation(iq: Array, capture: Capture, *, subcarrier_spacings: Meta(description='One or more subcarrier spacings (15e3, 30e3, 60e3, etc)', extra={'standard_name': 'One or more subcarrier spacings (15e3, 30e3, 60e3, etc)', 'units': 'Hz'})] = (15000.0, 30000.0, 60000.0), frame_range: int | tuple[int, int] = (0, 1), frame_slots: Meta(description="string composed of ('d', 'u', 's') specifying the sequence of slots in 1 TDD cellular frame, or None to fill with downlink", extra={'standard_name': "string composed of ('d', 'u', 's') specifying the sequence of slots in 1 TDD cellular frame, or None to fill with downlink"})] = None, symbol_range: int | tuple[int, int | None] = (0, None), generation: Literal['4G', '5G'] = '5G')¶

evaluate the cyclic autocorrelation of the IQ sequence based on 4G or 5G cellular cyclic prefix sample lag offsets.

The correlation can be configured to evaluate across specified ranges of frame indices, slot indices (across the frames), and symbol indices (across the slots). Each range may be specified as a single number (“first $N$ indices”) or as a tuple that is passed to the python builtin range.

Parameters:

iq (Array) – input waveform array
capture (striqt.analysis.specs.Capture) – acquisition specification
ax_xarray (Any) – whether to return an xarray object (True), an array (False), or a delayed xarray object (‘delayed’)
subcarrier_spacings (float | tuple[float, ...]) – One or more subcarrier spacings (15e3, 30e3, 60e3, etc) (Constraints: units: Hz)
frame_range (int | tuple[int, int])
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
symbol_range (int | tuple[int, int | None])
generation (one of ['4G', '5G'])

Returns:

an float32-valued array with matching the array type of iq

striqt.analysis.cellular_resource_power_histogram(iq: Array, capture: Capture, *, window: get_window`'})], subcarrier_spacing: Meta(description='Subcarrier spacing (15e3, 30e3, 60e3, etc)', extra={'standard_name': 'Subcarrier spacing (15e3, 30e3, 60e3, etc)', 'units': 'Hz'})], power_low: Meta(description='Minimum power bin', extra={'standard_name': 'Minimum power bin', 'units': 'dB power'})], power_high: Meta(description='Maximum power bin', extra={'standard_name': 'Maximum power bin', 'units': 'dB power'})], power_resolution: Meta(description='Power bin resolution', extra={'standard_name': 'Power bin resolution', 'units': 'dB'})], average_rbs: Meta(description='True (or "half") to coarsen spectrogram bins by integrating 1-subcarrier frequency resolution into 1- (or 1/2)-resource block', extra={'standard_name': 'True (or "half") to coarsen spectrogram bins by integrating 1-subcarrier frequency resolution into 1- (or 1/2)-resource block'})] = False, average_slots: Meta(description='True to coarsen spectrogram bins by averaging 1-symbol time resolution', extra={'standard_name': 'True to coarsen spectrogram bins by averaging 1-symbol time resolution'})] = False, guard_bandwidths: Meta(description='Channel guard bandwidths to ignore on the left and right sides', extra={'standard_name': 'Channel guard bandwidths to ignore on the left and right sides', 'units': 'Hz'})] = (0, 0), frame_slots: Meta(description="string composed of ('d', 'u', 's') specifying the sequence of slots in 1 TDD cellular frame, or None to fill with downlink", extra={'standard_name': "string composed of ('d', 'u', 's') specifying the sequence of slots in 1 TDD cellular frame, or None to fill with downlink"})] = None, special_symbols: Meta(description="string composed of the characters ('d', 'u', 'f') indicating the sequence of symbol types (when 's' is in frame_slots)", extra={'standard_name': "string composed of the characters ('d', 'u', 'f') indicating the sequence of symbol types (when 's' is in frame_slots)"})] = None, cyclic_prefix: Meta(description='the 3GPP cyclic prefix type', extra={'standard_name': 'the 3GPP cyclic prefix type'})] = 'normal', lo_bandstop: Meta(description='mask with float("nan") at this bandwidth centered at baseband DC', extra={'standard_name': 'mask with float("nan") at this bandwidth centered at baseband DC', 'units': 'Hz'})] | None = None)¶

Evaluate the spectrograms of a cellular resource grid on each port, and return a flattened histogram of its power levels.

Parameters:

iq (Array) – input waveform array
capture (striqt.analysis.specs.Capture) – acquisition specification
ax_xarray (Any) – whether to return an xarray object (True), an array (False), or a delayed xarray object (‘delayed’)
window (str | tuple[str, float]) – window function specification following scipy.signal.get_window
subcarrier_spacing (float) – Subcarrier spacing (15e3, 30e3, 60e3, etc) (Constraints: units: Hz)
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']) – True (or “half”) to coarsen spectrogram bins by integrating 1-subcarrier frequency resolution into 1- (or 1/2)-resource block
average_slots (bool) – True to coarsen spectrogram bins by averaging 1-symbol time resolution
guard_bandwidths (tuple[float, float]) – Channel guard bandwidths to ignore on the left and right sides (Constraints: units: Hz)
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
special_symbols (str | None) – string composed of the characters (‘d’, ‘u’, ‘f’) indicating the sequence of symbol types (when ‘s’ is in frame_slots)
cyclic_prefix (one of ['normal'] | one of ['extended']) – the 3GPP cyclic prefix type
lo_bandstop (float | None) – mask with float(“nan”) at this bandwidth centered at baseband DC (Constraints: units: Hz)

Returns:

xarray.DataArray or (array, dict) based on as_xarray

striqt.analysis.channel_power_histogram(iq: Array, capture: Capture, *, detector_period: Fraction, power_detectors: tuple[str, ...] = ('rms', 'peak'), power_low: Meta(description='Minimum power bin', extra={'standard_name': 'Minimum power bin', 'units': 'dB power'})], power_high: Meta(description='Maximum power bin', extra={'standard_name': 'Maximum power bin', 'units': 'dB power'})], power_resolution: Meta(description='Power bin resolution', extra={'standard_name': 'Power bin resolution', 'units': 'dB'})])¶

evaluate the fraction of channel power readings binned on a uniform grid spacing.

The outputs correspond to bin centers.

Parameters:

iq (Array) – input waveform array
capture (striqt.analysis.specs.Capture) – acquisition specification
ax_xarray (Any) – whether to return an xarray object (True), an array (False), or a delayed xarray object (‘delayed’)
detector_period (Fraction)
power_detectors (tuple[str, ...])
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)

striqt.analysis.channel_power_time_series(iq: Array, capture: Capture, *, detector_period: Fraction, power_detectors: tuple[str, ...] = ('rms', 'peak'))¶

Compute a binned time series of channel power detector measurements.

Parameters:

iq (Array) – input waveform array
capture (striqt.analysis.specs.Capture) – acquisition specification
ax_xarray (Any) – whether to return an xarray object (True), an array (False), or a delayed xarray object (‘delayed’)
detector_period (Fraction)
power_detectors (tuple[str, ...])

striqt.analysis.cyclic_channel_power(iq: Array, capture: Capture, *, cyclic_period: float, detector_period: Fraction, power_detectors: tuple[str, ...] = ('rms', 'peak'), cyclic_statistics: tuple[str | float, ...] = ('min', 'mean', 'max'))¶

Compute cyclic measurements of channel power.

Parameters:

iq (Array) – input waveform array
capture (striqt.analysis.specs.Capture) – acquisition specification
ax_xarray (Any) – whether to return an xarray object (True), an array (False), or a delayed xarray object (‘delayed’)
cyclic_period (float)
detector_period (Fraction)
power_detectors (tuple[str, ...])
cyclic_statistics (tuple[str | float, ...])

striqt.analysis.dBlinmean(x_dB: Dataset, axis: Dims | None = None, overwrite_x=False) → Dataset¶

striqt.analysis.dBlinmean(x_dB: DataArray, axis: Dims | None = None, overwrite_x=False, min_dtype='float32') → DataArray

striqt.analysis.dBlinmean(x_dB: ndarray, axis: int | Sequence[int] | None = None, overwrite_x=False, min_dtype='float32') → ndarray

striqt.analysis.dBlinmean(x_dB: Series, axis: int | Sequence[int] | None = None, overwrite_x=False, min_dtype='float32') → Series

striqt.analysis.dBlinmean(x_dB: DataFrame, axis: int | Sequence[int] | None = None, overwrite_x=False, min_dtype='float32') → DataFrame

evaluate the mean in linear power space given power in dB.

This is equivalent to:: powtodB(dBtopow(x).mean(axis))

Returns:: array-like object with same shape as x_dB, reduced by the dimension at the specified axes

striqt.analysis.dBlinsum(x_dB: Dataset, axis: Dims | None = None, overwrite_x=False) → Dataset¶

striqt.analysis.dBlinsum(x_dB: DataArray, axis: Dims | None = None, overwrite_x=False) → DataArray

striqt.analysis.dBlinsum(x_dB: ndarray, axis: int | Sequence[int] | None = None, overwrite_x=False, min_dtype='float32') → ndarray

striqt.analysis.dBlinsum(x_dB: Series, axis: int | Sequence[int] | None = None, overwrite_x=False, min_dtype='float32') → Series

striqt.analysis.dBlinsum(x_dB: DataFrame, axis: int | Sequence[int] | None = None, overwrite_x=False, min_dtype='float32') → DataFrame

evaluate the sum in linear power space given power in dB.

This is equivalent to:: powtodB(dBtopow(x).sum(axis))

Returns:: array-like object with same shape as x_dB, reduced by the dimension at the specified axes

striqt.analysis.dBtopow(x: _ALN, *, overwrite_x: bool = False, min_dtype: DTypeLike = 'float32') → _ALN¶: compute 10**(x/10) with speed optimizations

striqt.analysis.dump(store: ZarrStore, data: xr.DataArray | xr.Dataset, *, append_dim: str = 'capture', compression: bool = True, zarr_format: ZarrFormat = 'auto', compute: bool = True, chunk_bytes: ChunksSize = 50000000, max_threads: int | None = None, **kwargs) → ZarrStore¶: serialize a dataset into a zarr directory or zipfile

striqt.analysis.envtodB(x: _ALN, *, abs: bool = True, eps: float = 0, overwrite_x: bool = False, min_dtype: DTypeLike = 'float32') → _ALN¶: compute 20*log10(abs(x) + eps) or 20*log10(x + eps) with speed optimizations

striqt.analysis.envtopow(x: _ALN, *, overwrite_x: bool = False, min_dtype: DTypeLike = 'float32') → _ALN¶: Computes abs(x)**2 with speed optimizations

striqt.analysis.iq_waveform(iq: Array, capture: Capture, *, start_time_sec: float | None = None, stop_time_sec: float | None = None)¶

package the IQ waveform as a measurement result.

Parameters:

iq (Array) – input waveform array
capture (striqt.analysis.specs.Capture) – acquisition specification
ax_xarray (Any) – whether to return an xarray object (True), an array (False), or a delayed xarray object (‘delayed’)
start_time_sec (float | None)
stop_time_sec (float | None)

striqt.analysis.load(path: str | Path, chunks: ChunksSize = None, **kwargs) → xr.Dataset¶

load and return a dataset or data array.

Parameters:

path – location of the data store, potentially an fsspec string
chunks – None to load the file without dask, or ‘auto’ to return a dask array with automatically selected chunk sizes
kwargs – Additional arguments passed into xr.open_dataset

Returns:

The opened dataset object.

striqt.analysis.open_store(target: str | Path, *, mode: Literal['r', 'w', 'a']) → ZarrStore¶

striqt.analysis.power_spectral_density(iq: Array, capture: Capture, *, window: get_window`'})], frequency_resolution: float, fractional_overlap: Fraction = 0, window_fill: Fraction = 1, integration_bandwidth: float | None = None, trim_stopband: bool = True, lo_bandstop: Meta(description='mask with float("nan") at this bandwidth centered at baseband DC', extra={'standard_name': 'mask with float("nan") at this bandwidth centered at baseband DC', 'units': 'Hz'})] | None = None, time_statistic: tuple[str | float, ...] = ('mean',))¶

estimate power spectral density using the Welch method.

A list of statistics can be supplied to evaluate across the frequency axis, including ‘mean’ as applied in the original method.

Parameters:

iq (Array) – input waveform array
capture (striqt.analysis.specs.Capture) – acquisition specification
ax_xarray (Any) – whether to return an xarray object (True), an array (False), or a delayed xarray object (‘delayed’)
window (str | tuple[str, float]) – window function specification following scipy.signal.get_window
frequency_resolution (float)
fractional_overlap (Fraction)
window_fill (Fraction)
integration_bandwidth (float | None)
trim_stopband (bool)
lo_bandstop (float | None) – mask with float(“nan”) at this bandwidth centered at baseband DC (Constraints: units: Hz)
time_statistic (tuple[str | float, ...])

striqt.analysis.powtodB(x: _ALN, *, abs: bool = True, eps: float = 0, overwrite_x: bool = False, min_dtype: DTypeLike = 'float32') → _ALN¶: compute 10*log10(abs(x) + eps) or 10*log10(x + eps) with speed optimizations

striqt.analysis.spectrogram(iq: Array, capture: Capture, *, window: get_window`'})], frequency_resolution: float, fractional_overlap: Fraction = 0, window_fill: Fraction = 1, integration_bandwidth: float | None = None, trim_stopband: bool = True, lo_bandstop: Meta(description='mask with float("nan") at this bandwidth centered at baseband DC', extra={'standard_name': 'mask with float("nan") at this bandwidth centered at baseband DC', 'units': 'Hz'})] | None = None, time_aperture: float | None = None)¶

Evaluate a spectrogram based on an STFT.

The analysis parameters are in physical time and frequency units based on capture.sample_rate. The frequency axis is truncated to ±`capture.analysis_bandwidth`.

The underlying implementation is striqt.waveform.spectrogram. As a result this accepts cupy or numpy arrays interchangably and implements speed optimizations specific to complex-valued IQ waveforms.

Parameters:

iq (Array) – input waveform array
capture (striqt.analysis.specs.Capture) – acquisition specification
ax_xarray (Any) – whether to return an xarray object (True), an array (False), or a delayed xarray object (‘delayed’)
window (str | tuple[str, float]) – window function specification following scipy.signal.get_window
frequency_resolution (float)
fractional_overlap (Fraction)
window_fill (Fraction)
integration_bandwidth (float | None)
trim_stopband (bool)
lo_bandstop (float | None) – mask with float(“nan”) at this bandwidth centered at baseband DC (Constraints: units: Hz)
time_aperture (float | None)

striqt.analysis.specs¶

class striqt.analysis.specs.Analysis¶

Bases: SpecBase

Returns:

Analysis result of type (xarray.DataArray if as_xarray else type(iq))

Parameters:

iq (numpy.ndarray or cupy.ndarray) – the M-channel input waveform of shape (M,N)
capture
as_xarray (bool) – True to return xarray.DataArray or False to match type(iq)

class striqt.analysis.specs.AnalysisFilter(*, nfft=8192, window='hamming', nfft_out=None)¶

Bases: SpecBase

Parameters:

nfft (int)
window (str | tuple[str, float]) – window function specification following scipy.signal.get_window
nfft_out (int | None)

class striqt.analysis.specs.AnalysisGroup¶

Bases: SpecBase

base class for a defining set of Analysis specs

class striqt.analysis.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.analysis.specs.Cellular5GNPSSSync(*, subcarrier_spacing, sample_rate=7680000.0, discovery_periodicity=0.02, frequency_offset=0, shared_spectrum=False, delay=0, symbol_indexes='auto', max_lag_symbols=None, window_fill=1, per_port=False, max_beams=None)¶

Bases: _Cellular5GNRSSBSync

Parameters:

subcarrier_spacing (float) – Subcarrier spacing (15e3, 30e3, 60e3, etc) (Constraints: units: Hz)
sample_rate (float)
discovery_periodicity (float)
frequency_offset (float)
shared_spectrum (bool)
delay (float)
symbol_indexes (one of ['auto', 'a', 'b', 'c', 'd', 'e', 'f', 'g'] | tuple[int, ...]) – index locations in the SSB, or a cell search case from 3GPP TS 38.213 Sec. 4.1)
max_lag_symbols (int | None) – limit the number of symbols of lag in the 5G sync correlator (Constraints: >= 1, <= 6)
window_fill (float) – Fraction of a symbol to fill with weighting function (Constraints: > 0, <= 1)
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.analysis.specs.Cellular5GNRPSSCorrelator(*, subcarrier_spacing, sample_rate=7680000.0, discovery_periodicity=0.02, frequency_offset=0, shared_spectrum=False, delay=0, symbol_indexes='auto', max_lag_symbols=None, max_block_count=1)¶

Bases: _Cellular5GNRSSBCorrelator

Parameters:

subcarrier_spacing (float) – Subcarrier spacing (15e3, 30e3, 60e3, etc) (Constraints: units: Hz)
sample_rate (float)
discovery_periodicity (float)
frequency_offset (float)
shared_spectrum (bool)
delay (float)
symbol_indexes (one of ['auto', 'a', 'b', 'c', 'd', 'e', 'f', 'g'] | tuple[int, ...]) – index locations in the SSB, or a cell search case from 3GPP TS 38.213 Sec. 4.1)
max_lag_symbols (int | None) – limit the number of symbols of lag in the 5G sync correlator (Constraints: >= 1, <= 6)
max_block_count (int | None)

class striqt.analysis.specs.Cellular5GNRSSBSpectrogram(*, subcarrier_spacing, sample_rate=7680000.0, discovery_periodicity=0.02, frequency_offset=0, max_block_count=None, window='blackmanharris', lo_bandstop=None)¶

Bases: Analysis

sample_rate (float): output sample rate for the resampled synchronization waveform (samples/s) discovery_periodicity (float): time period between synchronization blocks (s) frequency_offset (float or dict[float, float]):

center frequency offset (see notes)

shared_spectrum:: whether to follow the 3GPP “shared spectrum” synchronizatio block layout

max_block_count: number of synchronization blocks to evaluate

Parameters:

subcarrier_spacing (float) – Subcarrier spacing (15e3, 30e3, 60e3, etc) (Constraints: units: Hz)
sample_rate (float)
discovery_periodicity (float)
frequency_offset (float)
max_block_count (int | None)
window (str | tuple[str, float]) – window function specification following scipy.signal.get_window
lo_bandstop (float | None)

shared_spectrum = False¶

class striqt.analysis.specs.Cellular5GNRSSSCorrelator(*, subcarrier_spacing, sample_rate=7680000.0, discovery_periodicity=0.02, frequency_offset=0, shared_spectrum=False, delay=0, symbol_indexes='auto', max_lag_symbols=None, max_block_count=1)¶

Bases: _Cellular5GNRSSBCorrelator

Parameters:

subcarrier_spacing (float) – Subcarrier spacing (15e3, 30e3, 60e3, etc) (Constraints: units: Hz)
sample_rate (float)
discovery_periodicity (float)
frequency_offset (float)
shared_spectrum (bool)
delay (float)
symbol_indexes (one of ['auto', 'a', 'b', 'c', 'd', 'e', 'f', 'g'] | tuple[int, ...]) – index locations in the SSB, or a cell search case from 3GPP TS 38.213 Sec. 4.1)
max_lag_symbols (int | None) – limit the number of symbols of lag in the 5G sync correlator (Constraints: >= 1, <= 6)
max_block_count (int | None)

class striqt.analysis.specs.Cellular5GNSSSSync(*, subcarrier_spacing, sample_rate=7680000.0, discovery_periodicity=0.02, frequency_offset=0, shared_spectrum=False, delay=0, symbol_indexes='auto', max_lag_symbols=None, window_fill=1, per_port=False, max_beams=None)¶

Bases: _Cellular5GNRSSBSync

Parameters:

subcarrier_spacing (float) – Subcarrier spacing (15e3, 30e3, 60e3, etc) (Constraints: units: Hz)
sample_rate (float)
discovery_periodicity (float)
frequency_offset (float)
shared_spectrum (bool)
delay (float)
symbol_indexes (one of ['auto', 'a', 'b', 'c', 'd', 'e', 'f', 'g'] | tuple[int, ...]) – index locations in the SSB, or a cell search case from 3GPP TS 38.213 Sec. 4.1)
max_lag_symbols (int | None) – limit the number of symbols of lag in the 5G sync correlator (Constraints: >= 1, <= 6)
window_fill (float) – Fraction of a symbol to fill with weighting function (Constraints: > 0, <= 1)
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.analysis.specs.CellularCyclicAutocorrelator(*, subcarrier_spacings=(15000.0, 30000.0, 60000.0), frame_range=(0, 1), frame_slots=None, symbol_range=(0, None), generation='5G')¶

Bases: Analysis

Parameters:

subcarrier_spacings (float | tuple[float, ...]) – One or more subcarrier spacings (15e3, 30e3, 60e3, etc) (Constraints: units: Hz)
frame_range (int | tuple[int, int])
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
symbol_range (int | tuple[int, int | None])
generation (one of ['4G', '5G'])

class striqt.analysis.specs.CellularResourcePowerHistogram(*, window, subcarrier_spacing, power_low, power_high, power_resolution, average_rbs=False, average_slots=False, guard_bandwidths=(0, 0), frame_slots=None, special_symbols=None, cyclic_prefix='normal', lo_bandstop=None)¶

Bases: Analysis

Parameters:

window (str | tuple[str, float]) – window function specification following scipy.signal.get_window
subcarrier_spacing (float) – Subcarrier spacing (15e3, 30e3, 60e3, etc) (Constraints: units: Hz)
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']) – True (or “half”) to coarsen spectrogram bins by integrating 1-subcarrier frequency resolution into 1- (or 1/2)-resource block
average_slots (bool) – True to coarsen spectrogram bins by averaging 1-symbol time resolution
guard_bandwidths (tuple[float, float]) – Channel guard bandwidths to ignore on the left and right sides (Constraints: units: Hz)
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
special_symbols (str | None) – string composed of the characters (‘d’, ‘u’, ‘f’) indicating the sequence of symbol types (when ‘s’ is in frame_slots)
cyclic_prefix (one of ['normal'] | one of ['extended']) – the 3GPP cyclic prefix type
lo_bandstop (float | None) – mask with float(“nan”) at this bandwidth centered at baseband DC (Constraints: units: Hz)

class striqt.analysis.specs.ChannelPowerHistogram(*, detector_period, power_detectors=('rms', 'peak'), power_low, power_high, power_resolution)¶

Bases: ChannelPowerTimeSeries

Parameters:

detector_period (Fraction)
power_detectors (tuple[str, ...])
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)

class striqt.analysis.specs.ChannelPowerTimeSeries(*, detector_period, power_detectors=('rms', 'peak'))¶

Bases: Analysis

Parameters:

detector_period (Fraction)
power_detectors (tuple[str, ...])

class striqt.analysis.specs.CyclicChannelPower(*, cyclic_period, detector_period, power_detectors=('rms', 'peak'), cyclic_statistics=('min', 'mean', 'max'))¶

Bases: Analysis

Parameters:

cyclic_period (float)
detector_period (Fraction)
power_detectors (tuple[str, ...])
cyclic_statistics (tuple[str | float, ...])

class striqt.analysis.specs.FilteredCapture(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, analysis_filter=AnalysisFilter(nfft=8192, window='hamming', nfft_out=None))¶

Bases: Capture

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)
analysis_filter (AnalysisFilter)

class striqt.analysis.specs.FrequencyAnalysisSpecBase(*, window, frequency_resolution, fractional_overlap=0, window_fill=1, integration_bandwidth=None, trim_stopband=True, lo_bandstop=None)¶

Bases: Analysis

window (specs.types.WindowType): a window specification, following scipy.signal.get_window frequency_resolution (float): the STFT resolution (in Hz) fractional_overlap (float):

fraction of each FFT window that overlaps with its neighbor

window_fill (float):: fraction of each FFT window that is filled with the window function (leaving the rest zeroed)

integration_bandwidth (float): bin bandwidth for RMS averaging in the frequency domain trim_stopband (bool):

whether to trim the frequency axis to capture.analysis_bandwidth

Parameters:

window (str | tuple[str, float]) – window function specification following scipy.signal.get_window
frequency_resolution (float)
fractional_overlap (Fraction)
window_fill (Fraction)
integration_bandwidth (float | None)
trim_stopband (bool)
lo_bandstop (float | None) – mask with float(“nan”) at this bandwidth centered at baseband DC (Constraints: units: Hz)

class striqt.analysis.specs.IQWaveform(*, start_time_sec=None, stop_time_sec=None)¶

Bases: Analysis

Parameters:

start_time_sec (float | None)
stop_time_sec (float | None)

striqt.analysis.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.analysis.specs.PowerSpectralDensity(*, window, frequency_resolution, fractional_overlap=0, window_fill=1, integration_bandwidth=None, trim_stopband=True, lo_bandstop=None, time_statistic=('mean',))¶

Bases: FrequencyAnalysisSpecBase

Parameters:

window (str | tuple[str, float]) – window function specification following scipy.signal.get_window
frequency_resolution (float)
fractional_overlap (Fraction)
window_fill (Fraction)
integration_bandwidth (float | None)
trim_stopband (bool)
lo_bandstop (float | None) – mask with float(“nan”) at this bandwidth centered at baseband DC (Constraints: units: Hz)
time_statistic (tuple[str | float, ...])

class striqt.analysis.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.analysis.specs.Spectrogram(*, window, frequency_resolution, fractional_overlap=0, window_fill=1, integration_bandwidth=None, trim_stopband=True, lo_bandstop=None, time_aperture=None)¶

Bases: FrequencyAnalysisSpecBase

time_aperture (float):: if specified, binned RMS averaging is applied along time axis in the spectrogram to yield this coarser resolution (s)

dB (bool): if True, returned power is transformed into dB units

Parameters:

window (str | tuple[str, float]) – window function specification following scipy.signal.get_window
frequency_resolution (float)
fractional_overlap (Fraction)
window_fill (Fraction)
integration_bandwidth (float | None)
trim_stopband (bool)
lo_bandstop (float | None) – mask with float(“nan”) at this bandwidth centered at baseband DC (Constraints: units: Hz)
time_aperture (float | None)

dB = True¶

class striqt.analysis.specs.SpectrogramHistogram(*, window, frequency_resolution, fractional_overlap=0, window_fill=1, integration_bandwidth=None, trim_stopband=True, lo_bandstop=None, time_aperture=None, power_low, power_high, power_resolution)¶

Bases: Spectrogram

Parameters:

window (str | tuple[str, float]) – window function specification following scipy.signal.get_window
frequency_resolution (float)
fractional_overlap (Fraction)
window_fill (Fraction)
integration_bandwidth (float | None)
trim_stopband (bool)
lo_bandstop (float | None) – mask with float(“nan”) at this bandwidth centered at baseband DC (Constraints: units: Hz)
time_aperture (float | None)
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)

class striqt.analysis.specs.SpectrogramHistogramRatio(*, window, frequency_resolution, fractional_overlap=0, window_fill=1, integration_bandwidth=None, trim_stopband=True, lo_bandstop=None, time_aperture=None, power_low, power_high, power_resolution)¶

Bases: SpectrogramHistogram

Parameters:

window (str | tuple[str, float]) – window function specification following scipy.signal.get_window
frequency_resolution (float)
fractional_overlap (Fraction)
window_fill (Fraction)
integration_bandwidth (float | None)
trim_stopband (bool)
lo_bandstop (float | None) – mask with float(“nan”) at this bandwidth centered at baseband DC (Constraints: units: Hz)
time_aperture (float | None)
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)

class striqt.analysis.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¶

National Institute of Standards and Technology

Batched waveform analysis¶

striqt.analysis¶

striqt.analysis.specs¶

Table of Contents

Previous topic

Next topic

This Page