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`'})] = ('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 (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) → 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 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, 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: 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)

National Institute of Standards and Technology

striqt.analysis¶

Table of Contents

Previous topic

Next topic

This Page