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 (striqt.analysis.lib.typing.TAR)
registry (striqt.analysis.lib.register.AnalysisRegistry)
block_each (bool)
expand_dims (collections.abc.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: float, 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 – the vector of size (N, M) for N channels and M IQ waveform samples
capture – capture structure that describes the iq acquisition parameters
sample_rate (samples/s) – downsample to this rate before analysis (or None to follow capture.sample_rate)
subcarrier_spacing (Hz) – OFDM subcarrier spacing
discovery_periodicity (s) – interval between synchronization blocks
frequency_offset (Hz) – baseband center frequency of the synchronization block
shared_spectrum – whether to assume “shared_spectrum” symbol layout in the SSB according to 3GPP TS 138 213: Section 4.1)
max_block_count – if not None, the number of synchronization blocks to analyze
as_xarray – if True (default), return an xarray.DataArray, otherwise a ChannelAnalysisResult object

References

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

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)
subcarrier_spacing (float) – 3GPP channel subcarrier spacing (Hz)
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
delay – minimum time delay (in s) to start of SSB set

striqt.analysis.cellular_5g_pss_sync(iq: Array, capture: Capture, *, subcarrier_spacing: float, 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

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)
subcarrier_spacing (float) – 3GPP channel subcarrier spacing (Hz)
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
delay – minimum time delay (in s) to start of SSB set

striqt.analysis.cellular_5g_ssb_spectrogram(iq: Array, capture: Capture, *, subcarrier_spacing: float, sample_rate: float = 7680000.0, discovery_periodicity: float = 0.02, frequency_offset: float = 0, max_block_count: int | None = None, window: str | tuple[str, float] = ('kaiser_by_enbw', 2), 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 – the vector of size (N, M) for N channels and M IQ waveform samples
capture – capture structure that describes the iq acquisition parameters
sample_rate (samples/s) – downsample to this rate before analysis (or None to follow capture.sample_rate)
subcarrier_spacing (Hz) – OFDM subcarrier spacing
discovery_periodicity (s) – interval between synchronization blocks
frequency_offset (Hz) – baseband center frequency of the synchronization block
max_block_count – if not None, the number of synchronization blocks to analyze
as_xarray – if True (default), return an xarray.DataArray, otherwise a ChannelAnalysisResult object

References

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

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)
subcarrier_spacing (float) – 3GPP channel subcarrier spacing (Hz)
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

striqt.analysis.cellular_5g_sss_correlation(iq: Array, capture: Capture, *, subcarrier_spacing: float, 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 – the vector of size (N, M) for N channels and M IQ waveform samples
capture – capture structure that describes the iq acquisition parameters
sample_rate (samples/s) – downsample to this rate before analysis (or None to follow capture.sample_rate)
subcarrier_spacing (Hz) – OFDM subcarrier spacing
discovery_periodicity (s) – interval between synchronization blocks
frequency_offset (Hz) – baseband center frequency of the synchronization block
shared_spectrum – whether to assume “shared_spectrum” symbol layout in the SSB according to 3GPP TS 138 213: Section 4.1)
max_block_count – if not None, the number of synchronization blocks to analyze
as_xarray – if True (default), return an xarray.DataArray, otherwise a ChannelAnalysisResult object

References

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

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)
subcarrier_spacing (float) – 3GPP channel subcarrier spacing (Hz)
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
delay – minimum time delay (in s) to start of SSB set

striqt.analysis.cellular_cyclic_autocorrelation(iq: Array, capture: Capture, *, subcarrier_spacings: float | tuple[float, ...] = (15000.0, 30000.0, 60000.0), frame_range: int | tuple[int, int] = (0, 1), frame_slots: str | None = 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 (numpy.ndarray or cupy.ndarray) – the input waveform
capture – the waveform capture specification
subcarrier_spacings – cellular SCS to evaluate (currently supports 15e3, 30e3, or 60e3)
frame_range – the frame indices to evaluate
frame_slots – string composed of {‘d’, ‘u’, ‘s’} that specify the sequence of link direction of each slot in 1 TDD cellular frame (or None to fill with downlink)
symbol_range – the symbols to evaluate within all indexed slots
iq – the M-channel input waveform of shape (M,N)
capture
as_xarray (bool) – True to return xarray.DataArray or False to match type(iq)

Returns:

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

Returns:

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

striqt.analysis.cellular_resource_power_histogram(iq: Array, capture: Capture, *, window: str | tuple[str, float], subcarrier_spacing: float, power_low: float, power_high: float, power_resolution: float, average_rbs: bool | Literal['half'] = False, average_slots: bool = False, guard_bandwidths: tuple[float, float] = (0, 0), frame_slots: str | None = None, special_symbols: str | None = None, cyclic_prefix: Literal['normal', 'extended'] = 'normal', lo_bandstop: float | None = None)¶

Parameters:

window – window function to use (matching the arguments to scipy.signal.get_window)
subcarrier_spacing (Hz) – 15e3|30e3|60e3|120e3|240e3|480e3|960e3
power_low (dB arb units) – bottom edge of the histogram bins
power_high (dB arb units) – top edge of the histogram bins
power_resolution (dB) – resolution of the histogram
frame_slots – string composed of {‘d’, ‘u’, ‘s’} that speficy the sequence of slots in 1 TDD cellular frame, or None to fill with downlink.
special_symbols – string composed of {‘d’, ‘u’, ‘f’} that indicate the sequence of symbol types (when ‘s’ is in frame_slots).
guard_bandwidths (in Hz) – the channel guard bandwidths on the left and right sides
average_rbs – when True, counts will report power averaged to 1 RB in frequency; if False, resolution is 1 subcarrier.
average_slot – when True, counts will report power averaged to 1 slot in time; if False, resolution is 1 symbol.
cyclic_prefix – the 3GPP cyclic prefix size, one of (‘extended’,’normal’)
lo_bandstop – if specified, null this bandwidth about DC
as_xarray (bool) – if True (the default), returns an xarray with labeled axes and metadata; otherwise, returns an (array, dict) tuple containing the result and metadata
iq (numpy.ndarray or cupy.ndarray) – the M-channel input waveform of shape (M,N)
capture
as_xarray – True to return xarray.DataArray or False to match type(iq)

Returns:

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

Returns:

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

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

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

The outputs correspond to bin centers.

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)

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

None

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)

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'))¶

None

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)

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) → DataArray

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

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

striqt.analysis.dBlinmean(x_dB: DataFrame, axis: int | Sequence[int] | None = None, overwrite_x=False) → 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) → ndarray

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

striqt.analysis.dBlinsum(x_dB: DataFrame, axis: int | Sequence[int] | None = None, overwrite_x=False) → 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, out=None) → _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, out=None) → _ALN¶: compute 20*log10(abs(x) + eps) or 20*log10(x + eps) with speed optimizations

striqt.analysis.envtopow(x: _ALN, out=None) → _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 a clipping of the IQ waveform

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)

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

load a dataset or data array.

Parameters:

path – location of the data store
chunks – None to load the file without dask, or ‘auto’ to return a dask array with automatically selected chunk sizes

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

striqt.analysis.power_spectral_density(iq: Array, capture: Capture, *, window: str | tuple[str, float], frequency_resolution: float, fractional_overlap: Fraction = 0, window_fill: Fraction = 1, integration_bandwidth: float | None = None, trim_stopband: bool = True, lo_bandstop: float | 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.

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

striqt.analysis.powtodB(x: _ALN, abs: bool = True, eps: float = 0, out=None) → _ALN¶: compute 10*log10(abs(x) + eps) or 10*log10(x + eps) with speed optimizations

striqt.analysis.spectrogram(iq: Array, capture: Capture, *, window: str | tuple[str, float], frequency_resolution: float, fractional_overlap: Fraction = 0, window_fill: Fraction = 1, integration_bandwidth: float | None = None, trim_stopband: bool = True, lo_bandstop: float | 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.

National Institute of Standards and Technology

striqt.analysis¶

Table of Contents

Previous topic

Next topic

This Page