Batched waveform analysis

striqt.analysis

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

class striqt.analysis.AcquiredIQ(x: Array, capture: specs.Capture | None)

Bases: object

an acquired IQ waveform x with the capture that describes the acquisition.

capture: specs.Capture | None

the acquisition parameters of x, or None when not known

property preprocess: Preprocessor

return the Preprocessor of this acquisition, one per instance.

A dataclasses.replace copy starts with a Preprocessor of its own.

x: Array

the acquired waveform with shape (port, sample)

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) → AbstractContextManager[ExitStack]
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[CoordFunc] | CoordFunc | None = None, depends: Iterable[Callable] | Callable = [], caches: Iterable[KwArgCache] | KwArgCache | None = None, store_compressed: bool = True, prefer_filtered: bool = True, prefer_aligned: bool = True, attrs: dict[str, Any] = {}, validate: AnalysisValidator | None = None, tolerance: AnalysisTolerance | None = None) → AnalysisFuncWrapper[P]

register the decorated function as the measurement configured by spec_type.

Registration mutates this registry when the decorator is applied. A sweep enables the measurement through an Optional[spec_type] field named name in an analysis group such as specs.BuiltinAnalyses.

The decorated function is called as func(x, capture, **spec.to_dict()) with x an array of shape (port, sample), and returns an array with axes dims or an (array, attrs) tuple. The returned wrapper is called as wrapper(iq, capture, as_xarray=True, **kwargs) with the fields of spec_type as keywords. The __signature__ of the wrapper and the {args} placeholder in the docstring are filled from spec_type. An array iq is passed to the function as x. From an AcquiredIQ, AcquiredIQ.preprocess selects x as prefer_filtered and prefer_aligned request. validate runs before the IQ is read and raises specs.helpers.SpecValidationError on a rejected combination. The wrapper returns, by as_xarray: True, an xarray.DataArray with a leading port dimension; 'delayed', a DelayedDataArray; False, the (array, attrs) pair.

Parameters:

  • spec_type – the specs.Analysis subclass with the keyword arguments of the measurement as fields; one measurement per type

  • dtype – numpy dtype name of the output array

  • name – the output data variable name, or None for the function __name__

  • dims – dimension names of the output axes, or None for the dimensions of coord_factories, in order

  • coord_factories – functions registered with coordinates that produce the coordinate values along dims from (capture, spec)

  • depends – measurements with an intermediate this one reuses; recorded on the AnalysisInfo entry and not otherwise read

  • caches – KwArgCache instances that cache_context enables for one capture; shared with other measurements

  • store_compressed – False to store the output uncompressed in zarr

  • prefer_filtered – False to measure the waveform before the analysis_bandwidth low-pass; applies only when the acquisition filters

  • prefer_aligned – False to measure the waveform before the signal-trigger shift; applies only when the acquisition runs a trigger. A signal_trigger measurement must set False

  • attrs – attrs of the output xarray.DataArray; the spec fields and any attrs returned by the function take precedence

  • validate – a function of (capture, spec) that raises on a rejected combination without reading IQ; capture is the specs.AnalysisCapture projection, and the return value is ignored

  • tolerance – a function of (capture, spec, *, array_backend, input_error) that returns the specs.Tolerance of the output without reading IQ

Returns:

a decorator that registers the decorated function and returns the wrapper

Raises:

  • ValueError – when spec_type is already registered

  • TypeError – when a coord_factories entry is not callable; from the returned decorator, when name is already registered, when caches contains something other than KwArgCache instances, or when a field of spec_type shares a name with a field of another registered spec and differs in type

parameter_defaults: dict[str, Any]
parameter_fields: dict[str, 'msgspec.structs.FieldInfo|None']
tolerances(capture: specs.Capture, analysis: specs.AnalysisGroup, *, array_backend: ArrayBackend = 'numpy', input_error: float = 0.0) → dict[str, specs.Tolerance]

return the roundoff error budget of each measurement in analysis.

Results are cached on the specs.AnalysisCapture projection of capture.

Parameters:

  • array_backend – the array namespace of the measurement evaluation; selects the roundoff model

  • input_error – relative rms amplitude error already present in the IQ handed to the measurements, as a ratio

Returns:

a dict of specs.Tolerance keyed by measurement name; measurements left unset in analysis or registered without a tolerance= function are omitted

validate(capture: Capture, analysis: AnalysisGroup) → None

check each measurement in analysis against capture without IQ.

Measurements left unset in analysis, and those registered without a validate= function, are skipped. Results are cached on the specs.AnalysisCapture projection of capture. Captures equal in that projection share one entry.

Raises:

specs.helpers.SpecValidationError – on the first invalid combination, with a field path rooted at $.analysis

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) – Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

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

class striqt.analysis.EvaluationOptions(as_xarray, registry, block_each=True, expand_dims=('port',))

Bases: Struct, Generic[TAR]

the result type and registry for one evaluation of an analysis group.

as_xarray selects the result type of analyze_by_spec: True for an xarray.Dataset, 'delayed' for DelayedDataArray results, or False for the bare arrays in the array namespace of the input. registry supplies the measurement function of each spec type. block_each True materializes each result to numpy as soon as the measurement returns, when as_xarray is True or 'delayed'. expand_dims names the leading dimension of each xarray.DataArray when as_xarray is True, and that dimension indexes the port axis of the input.

Raises:

TypeError – when as_xarray is not True, False, or 'delayed'

Parameters:

  • as_xarray (TAR)

  • registry (AnalysisRegistry)

  • block_each (bool)

  • expand_dims (Sequence[str])

class striqt.analysis.Preprocessor(iq: AcquiredIQ)

Bases: object

the conditioned-waveform selector of one acquisition.

The base class returns x of the acquisition for every preference. A subclass that conditions the waveform lazily overrides __call__.

__init__(iq: AcquiredIQ) → None

store a weak reference to iq.

property iq: AcquiredIQ

the acquisition that owns this preprocessor.

Raises:

RuntimeError – when the acquisition has been garbage collected

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: AnalysisGroup, capture: Capture, options: EvaluationOptions[Literal[True]] | None = None) → Dataset
striqt.analysis.analyze_by_spec(iq: Array | AcquiredIQ, spec: AnalysisGroup, capture: Capture, options: EvaluationOptions[Literal['delayed']]) → dict[str, DelayedDataArray]
striqt.analysis.analyze_by_spec(iq: Array | AcquiredIQ, spec: AnalysisGroup, capture: Capture, options: EvaluationOptions[Literal[False]]) → dict[str, Array]

evaluate the enabled measurements of spec on the IQ waveform.

Parameters:

  • iq – the waveform with shape (port, sample), or an AcquiredIQ from which each measurement selects the conditioned waveform

  • spec – an AnalysisGroup such as specs.BuiltinAnalyses; the None fields are skipped

  • capture – the acquisition parameters of iq

  • options – the result type and the registry, or None for the default registry with as_xarray True

Returns:

  • True: an xarray.Dataset of the measurements, with the capture fields as attrs and options.expand_dims as the leading dimension

  • 'delayed': a dict of DelayedDataArray keyed by measurement name

  • False: a dict of bare arrays keyed by measurement name, in the array namespace of iq

Return type:

by options.as_xarray

Raises:

  • TypeError – when spec is not an AnalysisGroup

  • KeyError – when the spec type of an enabled field is not in options.registry

  • specs.helpers.SpecValidationError – when a measurement validator rejects the (capture, spec) combination

striqt.analysis.cellular_5g_pss_correlation(iq: Array | AcquiredIQ, 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 port of the IQ against the 5G NR primary synchronization signal.

The input is the IQ before the analysis_bandwidth filter and before trigger alignment. The band at frequency_offset is shifted to baseband and resampled to sample_rate, then cross-correlated with the PSS sequence of each cell sector ID (3GPP TS 38.211 §7.4.2.2). Per capture, the output is the complex64 correlation (in √mW) with dimensions cellular_cell_id2, cellular_ssb_start_time (in s), cellular_ssb_beam_index, and cellular_ssb_lag (in s). The output keeps the first max_block_count sync blocks. A max_block_count of None keeps every block. The PSS template has unit energy over the 127 PSS subcarriers. The squared magnitude of a peak on a lag sample equals the received PSS symbol power. White noise with power spectral density N0 has a mean squared magnitude of N0 * subcarrier_spacing in each lag. The noise_bandwidth attr is subcarrier_spacing.

References

3GPP TS 38.211, “NR; Physical channels and modulation,” Table 7.4.3.1-1 and §7.4.2.2. https://www.3gpp.org/DynaReport/38211.htm

3GPP TS 38.213, “NR; Physical layer procedures for control,” §4.1. https://www.3gpp.org/DynaReport/38213.htm

Parameters:

  • iq (Array or AcquiredIQ) – IQ waveform of shape (port, sample), or an AcquiredIQ from which the conditioned waveform is selected

  • capture (striqt.analysis.specs.Capture) – the acquisition parameters of iq

  • as_xarray ((bool or 'delayed')) – True for an xarray.DataArray, False for an (array, attrs) tuple, or ‘delayed’ for a DelayedDataArray

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

striqt.analysis.cellular_5g_pss_sync(iq: Array | AcquiredIQ, 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)

return the start-of-waveform delay (in s) detected from the PSS correlation.

The input is the IQ before the analysis_bandwidth filter and before trigger alignment. Per capture, the output is one float32 value per port, or one value shared across the ports when per_port is False. The value is the lag chosen from the cellular_5g_pss_correlation output plus delay rounded to a sample at sample_rate. The output is a signal trigger: a source with signal_trigger set to this measurement aligns the capture by the output.

Parameters:

  • iq (Array or AcquiredIQ) – IQ waveform of shape (port, sample), or an AcquiredIQ from which the conditioned waveform is selected

  • capture (striqt.analysis.specs.Capture) – the acquisition parameters of iq

  • as_xarray ((bool or 'delayed')) – True for an xarray.DataArray, False for an (array, attrs) tuple, or ‘delayed’ for a DelayedDataArray

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

striqt.analysis.cellular_5g_ssb_spectrogram(iq: Array | AcquiredIQ, 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: 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 a spectrogram of each 5G NR SSB burst set by OFDM symbol and subcarrier.

The input is the IQ before the analysis_bandwidth filter and after trigger alignment. The STFT hops once per OFDM symbol. Half-subcarrier bins are summed pairwise to one bin per subcarrier of subcarrier_spacing. The frequency axis is trimmed to the sample_rate band about frequency_offset. Only the symbols of the burst set in the first 2 ms of each discovery_periodicity are kept (3GPP TS 38.213 §4.1).

Per capture, the output is power per subcarrier in dBm in a noise bandwidth of one subcarrier_spacing, with dimensions cellular_ssb_index (burst set within the capture), cellular_ssb_symbol_index (OFDM symbol within the burst set), and cellular_ssb_baseband_frequency (in Hz, relative to frequency_offset). Values are rounded to 3 decimals and stored as float16. The noise_bandwidth and resolution_bandwidth attrs follow the spectrogram measurement.

References

3GPP TS 38.213, “NR; Physical layer procedures for control,” §4.1. https://www.3gpp.org/DynaReport/38213.htm

Parameters:

  • iq (Array or AcquiredIQ) – IQ waveform of shape (port, sample), or an AcquiredIQ from which the conditioned waveform is selected

  • capture (striqt.analysis.specs.Capture) – the acquisition parameters of iq

  • as_xarray ((bool or 'delayed')) – True for an xarray.DataArray, False for an (array, attrs) tuple, or ‘delayed’ for a DelayedDataArray

  • 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) – mask with float(“nan”) at this bandwidth centered at baseband DC (Constraints: units: Hz)

striqt.analysis.cellular_5g_sss_correlation(iq: Array | AcquiredIQ, 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 port of the IQ against the 5G NR secondary synchronization signal.

The input is the IQ before the analysis_bandwidth filter and before trigger alignment. The band at frequency_offset is shifted to baseband and resampled to sample_rate, then cross-correlated with the SSS sequences (3GPP TS 38.211 §7.4.2.3). Per capture, the output is the complex64 correlation (in √mW) with dimensions cellular_cell_id2, cellular_ssb_start_time (in s), cellular_ssb_beam_index, and cellular_ssb_lag (in s). The output keeps the first max_block_count sync blocks. A max_block_count of None keeps every block. The SSS template has unit energy over the 127 SSS subcarriers. The squared magnitude of a peak on a lag sample equals the received SSS symbol power. White noise with power spectral density N0 has a mean squared magnitude of N0 * subcarrier_spacing in each lag. The noise_bandwidth attr is subcarrier_spacing.

References

3GPP TS 38.211, “NR; Physical channels and modulation,” Table 7.4.3.1-1 and §7.4.2.3. https://www.3gpp.org/DynaReport/38211.htm

3GPP TS 38.213, “NR; Physical layer procedures for control,” §4.1. https://www.3gpp.org/DynaReport/38213.htm

Parameters:

  • iq (Array or AcquiredIQ) – IQ waveform of shape (port, sample), or an AcquiredIQ from which the conditioned waveform is selected

  • capture (striqt.analysis.specs.Capture) – the acquisition parameters of iq

  • as_xarray ((bool or 'delayed')) – True for an xarray.DataArray, False for an (array, attrs) tuple, or ‘delayed’ for a DelayedDataArray

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

striqt.analysis.cellular_cyclic_autocorrelation(iq: Array | AcquiredIQ, 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-prefix autocorrelation of the IQ at each 4G or 5G symbol lag.

The input is the IQ after the analysis_bandwidth filter and before trigger alignment. For each spacing in subcarrier_spacings, the IQ is correlated with a copy of the IQ delayed by the OFDM symbol length, summed over the cyclic-prefix samples of the symbols that frame_range and symbol_range select. The downlink slots and the uplink slots of frame_slots are summed separately. Per capture, the output is the correlation magnitude in mW as float32 with dimensions (link_direction, subcarrier_spacing, cyclic_sample_lag). cyclic_sample_lag (in s) spans the longest symbol over the spacings. Entries past the symbol length of a spacing are nan, as is the 'uplink' row when frame_slots has no uplink slot. frame_range and symbol_range are a count of indexes from 0, or a (start, stop) pair as for the Python builtin range.

Parameters:

  • iq (Array or AcquiredIQ) – IQ waveform of shape (port, sample), or an AcquiredIQ from which the conditioned waveform is selected

  • capture (striqt.analysis.specs.Capture) – the acquisition parameters of iq

  • as_xarray ((bool or 'delayed')) – True for an xarray.DataArray, False for an (array, attrs) tuple, or ‘delayed’ for a DelayedDataArray

  • 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'])

striqt.analysis.cellular_resource_power_histogram(iq: Array | AcquiredIQ, 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="frequency resolution of the spectrogram: False for 1 subcarrier, True for 1 resource block (12 subcarriers), or 'half' for 6 subcarriers", extra={'standard_name': "frequency resolution of the spectrogram: False for 1 subcarrier, True for 1 resource block (12 subcarriers), or 'half' for 6 subcarriers"})] = 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='Bandwidths excluded at the (low, high) edges of the analysis band', extra={'standard_name': 'Bandwidths excluded at the (low, high) edges of the analysis band', '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: 2-1)'})] = '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 fraction of resource grid elements in each power bin per link.

The input is the IQ before the analysis_bandwidth filter and after trigger alignment. A spectrogram with one bin per subcarrier (or per resource block, with average_rbs) and one hop per OFDM symbol is split into downlink and uplink symbols by frame_slots and special_symbols. With average_slots, the power is averaged over each slot. The power is converted to dBm before binning. Per capture, the output is float32 with dimensions link_direction and cellular_resource_power_bin (in dBm per bin bandwidth). The fractions of both link directions together sum to 1 on each port. Subcarriers inside the guard_bandwidths and flexible symbols are excluded from the count.

Parameters:

  • iq (Array or AcquiredIQ) – IQ waveform of shape (port, sample), or an AcquiredIQ from which the conditioned waveform is selected

  • capture (striqt.analysis.specs.Capture) – the acquisition parameters of iq

  • as_xarray ((bool or 'delayed')) – True for an xarray.DataArray, False for an (array, attrs) tuple, or ‘delayed’ for a DelayedDataArray

  • 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']) – frequency resolution of the spectrogram: False for 1 subcarrier, True for 1 resource block (12 subcarriers), or ‘half’ for 6 subcarriers

  • average_slots (bool) – True to coarsen spectrogram bins by averaging 1-symbol time resolution

  • guard_bandwidths (tuple[float, float]) – Bandwidths excluded at the (low, high) edges of the analysis band (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', 'extended']) – cyclic prefix length of the 3GPP numerology (TS 38.211 Table 4.2-1)

  • lo_bandstop (float | None) – mask with float(“nan”) at this bandwidth centered at baseband DC (Constraints: units: Hz)

striqt.analysis.channel_power_histogram(iq: Array | AcquiredIQ, capture: Capture, *, detector_period: Meta(description='Power detector bin duration, a whole multiple of the sample period', extra={'standard_name': 'Power detector bin duration, a whole multiple of the sample period', 'units': 's'})], power_detectors: Meta(description="Power detectors evaluated in each bin: 'rms', 'peak', or both", extra={'standard_name': "Power detectors evaluated in each bin: 'rms', 'peak', or both"})] = ('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 that fall in each power bin.

The input is the IQ after the analysis_bandwidth filter and trigger alignment. The readings are those of channel_power_time_series with the same detector fields. Per capture, the output has dimensions (power_detector, channel_power_bin) as float32 and sums to 1 over the bins of each detector. channel_power_bin is the bin centers in dBm. The -inf and inf entries collect the readings more than half a bin below power_low or above power_high. Neither the output nor the bin coordinate has a noise_bandwidth attr. The noise bandwidth of each capture is the analysis_bandwidth coordinate, or sample_rate where analysis_bandwidth is infinite.

Parameters:

  • iq (Array or AcquiredIQ) – IQ waveform of shape (port, sample), or an AcquiredIQ from which the conditioned waveform is selected

  • capture (striqt.analysis.specs.Capture) – the acquisition parameters of iq

  • as_xarray ((bool or 'delayed')) – True for an xarray.DataArray, False for an (array, attrs) tuple, or ‘delayed’ for a DelayedDataArray

  • detector_period (Fraction) – Power detector bin duration, a whole multiple of the sample period (Constraints: units: s)

  • power_detectors (tuple[str, ...]) – Power detectors evaluated in each bin: ‘rms’, ‘peak’, or both

  • 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 | AcquiredIQ, capture: Capture, *, detector_period: Meta(description='Power detector bin duration, a whole multiple of the sample period', extra={'standard_name': 'Power detector bin duration, a whole multiple of the sample period', 'units': 's'})], power_detectors: Meta(description="Power detectors evaluated in each bin: 'rms', 'peak', or both", extra={'standard_name': "Power detectors evaluated in each bin: 'rms', 'peak', or both"})] = ('rms', 'peak'))

evaluate a time series of binned channel power for each power detector.

The input is the IQ after the analysis_bandwidth filter and trigger alignment. Each detector reduces the instantaneous power |iq|**2 over consecutive blocks of detector_period: 'rms' is the mean and 'peak' the maximum. Per capture, the output has dimensions (power_detector, time_elapsed) in dBm as float32. time_elapsed is in s from the start of the capture in steps of detector_period. The capture duration must be a whole number of detector periods. detector_period must be a whole number of samples at sample_rate. The output has no noise_bandwidth attr. The noise bandwidth of each capture is the analysis_bandwidth coordinate, or sample_rate where analysis_bandwidth is infinite.

Parameters:

  • iq (Array or AcquiredIQ) – IQ waveform of shape (port, sample), or an AcquiredIQ from which the conditioned waveform is selected

  • capture (striqt.analysis.specs.Capture) – the acquisition parameters of iq

  • as_xarray ((bool or 'delayed')) – True for an xarray.DataArray, False for an (array, attrs) tuple, or ‘delayed’ for a DelayedDataArray

  • detector_period (Fraction) – Power detector bin duration, a whole multiple of the sample period (Constraints: units: s)

  • power_detectors (tuple[str, ...]) – Power detectors evaluated in each bin: ‘rms’, ‘peak’, or both

striqt.analysis.cyclic_channel_power(iq: Array | AcquiredIQ, capture: Capture, *, cyclic_period: Meta(description='Cycle period of the channel power statistics, a whole multiple of `detector_period` and a common multiple of the expected signal periods', extra={'standard_name': 'Cycle period of the channel power statistics, a whole multiple of `detector_period` and a common multiple of the expected signal periods', 'units': 's'})], detector_period: Meta(description='Power detector bin duration, a whole multiple of the sample period', extra={'standard_name': 'Power detector bin duration, a whole multiple of the sample period', 'units': 's'})], power_detectors: Meta(description="Power detectors evaluated in each bin: 'rms', 'peak', or both", extra={'standard_name': "Power detectors evaluated in each bin: 'rms', 'peak', or both"})] = ('rms', 'peak'), cyclic_statistics: Meta(description="Statistics evaluated across cycles at each cycle lag: 'min', 'mean', 'max', 'median', or a quantile in (0, 1)", extra={'standard_name': "Statistics evaluated across cycles at each cycle lag: 'min', 'mean', 'max', 'median', or a quantile in (0, 1)"})] = ('min', 'mean', 'max'))

evaluate cyclic statistics of channel power across the cycles in a capture.

The input is the IQ after the analysis_bandwidth filter and trigger alignment. Each power detector bins the capture on detector_period. The binned series is folded into cycles of cyclic_period. Each cyclic statistic reduces across the cycles (Kuester et al.). Per capture, the output has dimensions (power_detector, cyclic_statistic, cyclic_lag) in dBm as float32. The cyclic_lag coordinate in s starts at 0 and spans one cyclic_period in steps of detector_period. Those 2 fields alone set the trace length, independent of the capture duration.

A signal with a period that divides cyclic_period (10 ms covers TDD cellular frames, 5 ms WiMAX frames, and the 1 ms CBRS test bin 1 pulse repetition interval) resolves at fixed lags with little spread between the 'min' and 'max' statistics. The uplink and downlink levels of a TDD network then occupy disjoint lag windows. Occupancy with an incommensurate period smears across all lags.

Statistics are evaluated in linear power and converted to dBm afterward. The capture duration must be a whole number of cycles. cyclic_period must be a whole number of detector periods, and detector_period a whole number of samples at sample_rate. A detector_period written as a fraction (1/28000 s is one OFDM symbol at 30 kHz subcarrier spacing) stays exact. The output has no noise_bandwidth attr, as for channel_power_time_series.

References

    1. Kuester et al., “Cyclic Analysis of Power in Radio Channels.”

Parameters:

  • iq (Array or AcquiredIQ) – IQ waveform of shape (port, sample), or an AcquiredIQ from which the conditioned waveform is selected

  • capture (striqt.analysis.specs.Capture) – the acquisition parameters of iq

  • as_xarray ((bool or 'delayed')) – True for an xarray.DataArray, False for an (array, attrs) tuple, or ‘delayed’ for a DelayedDataArray

  • cyclic_period (float) – Cycle period of the channel power statistics, a whole multiple of detector_period and a common multiple of the expected signal periods (Constraints: units: s)

  • detector_period (Fraction) – Power detector bin duration, a whole multiple of the sample period (Constraints: units: s)

  • power_detectors (tuple[str, ...]) – Power detectors evaluated in each bin: ‘rms’, ‘peak’, or both

  • cyclic_statistics (tuple[str | float, ...]) – Statistics evaluated across cycles at each cycle lag: ‘min’, ‘mean’, ‘max’, ‘median’, or a quantile in (0, 1)

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

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: _ALD, axis: Dims | int | Sequence[int] | None = None, overwrite_x: bool = False, min_dtype: DTypeLike = 'float32') → _ALD

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

The computation is elementwise in a floating-point dtype no narrower than min_dtype.

Parameters:

  • x – values in dB, as a number, array, or pandas/xarray object

  • overwrite_x – True to write the result into x when the dtype of x is at least min_dtype

  • min_dtype – the narrowest floating-point dtype of the computation; a narrower x is widened to min_dtype

Returns:

a Python number for a number, otherwise the array, pandas, or xarray type of x. An xarray units attribute is converted from dB to linear power.

Return type:

the values in linear power units, in the container type of x

Raises:

TypeError – if min_dtype is None or narrower than float32, or if no array can be associated with the type of x

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, bounded: bool = False, overwrite_x: bool = False, min_dtype: DTypeLike = 'float32') → _ALN

compute 20*log10(abs(x)), or 20*log10(x) when abs is False.

The computation is elementwise in a floating-point dtype no narrower than min_dtype.

Parameters:

  • x – values in linear amplitude (envelope) units, as a number, array, or pandas/xarray object

  • abs – True to convert abs(x); a complex x is converted by magnitude in either case

  • bounded – True to floor the result at the same dB value as powtodB with bounded, 10*log10(tiny) for the smallest normal value tiny of the computation dtype; zeros then map to the floor instead of -inf

  • overwrite_x – True to write the result into x when the dtype of x is at least min_dtype; a complex x is then returned as the real view of the buffer of x

  • min_dtype – the narrowest floating-point dtype of the computation; a narrower x is widened to min_dtype

Returns:

a Python number for a number, otherwise the array, pandas, or xarray type of x. An xarray units attribute is converted from amplitude to dB.

Return type:

the dB values, real-valued, in the container type of x

Raises:

TypeError – if min_dtype is None or narrower than float32, or if no array can be associated with the type of x

striqt.analysis.envtopow(x: _ALN, *, overwrite_x: bool = False, min_dtype: DTypeLike = 'float32') → _ALN

compute abs(x)**2.

The computation is elementwise in a floating-point dtype no narrower than min_dtype.

Parameters:

  • x – values in linear amplitude (envelope) units, real or complex, as a number, array, or pandas/xarray object

  • overwrite_x – True to write the result into x when the dtype of x is at least min_dtype; a complex x is then returned as the real view of the buffer of x

  • min_dtype – the narrowest floating-point dtype of the computation; a narrower x is widened to min_dtype

Returns:

a Python number for a number, otherwise the array, pandas, or xarray type of x. An xarray units attribute is converted from amplitude to linear power.

Return type:

the power values, real-valued, in the container type of x

Raises:

TypeError – if min_dtype is None or narrower than float32, or if no array can be associated with the type of x

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

return the IQ samples between start_time_sec and stop_time_sec.

The input is the IQ after the analysis_bandwidth filter and trigger alignment. Per capture, the output has the dimension (iq_index,) of sample indexes into the capture. The values are complex64 in √mW. The squared magnitude |iq|**2 is power in mW. A None time bound selects the start or the end of the capture. A bound outside the capture is clamped to the capture. The output has no noise_bandwidth attr. The noise bandwidth of each capture is the analysis_bandwidth coordinate, or sample_rate where analysis_bandwidth is infinite.

Parameters:

  • iq (Array or AcquiredIQ) – IQ waveform of shape (port, sample), or an AcquiredIQ from which the conditioned waveform is selected

  • capture (striqt.analysis.specs.Capture) – the acquisition parameters of iq

  • as_xarray ((bool or 'delayed')) – True for an xarray.DataArray, False for an (array, attrs) tuple, or ‘delayed’ for a DelayedDataArray

  • 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 | AcquiredIQ, 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 the power spectral density from time statistics of the spectrogram.

The input is the IQ before the analysis_bandwidth filter and after trigger alignment. Each time_statistic reduces the spectrogram output with the same STFT fields (linear power, unrounded) over the STFT windows at each frequency bin. 'mean' is the Welch estimate (Welch, 1967). A float in [0, 1] is a quantile. Per capture, the output has dimensions (time_statistic, baseband_frequency) in dBm per noise bandwidth as float32. The noise_bandwidth attr is that bandwidth in Hz. A power spectral density in dBm/Hz is the output minus 10 * log10(noise_bandwidth).

References

P. D. Welch, “The use of fast Fourier transform for the estimation of power spectra: A method based on time averaging over short, modified periodograms,” IEEE Trans. Audio Electroacoust., vol. 15, no. 2, pp. 70-73, 1967. doi:10.1109/TAU.1967.1161901

Parameters:

  • iq (Array or AcquiredIQ) – IQ waveform of shape (port, sample), or an AcquiredIQ from which the conditioned waveform is selected

  • capture (striqt.analysis.specs.Capture) – the acquisition parameters of iq

  • as_xarray ((bool or 'delayed')) – True for an xarray.DataArray, False for an (array, attrs) tuple, or ‘delayed’ for a DelayedDataArray

  • 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, bounded: bool = False, overwrite_x: bool = False, min_dtype: DTypeLike = 'float32') → _ALN

compute 10*log10(abs(x)), or 10*log10(x) when abs is False.

The computation is elementwise in a floating-point dtype no narrower than min_dtype.

Parameters:

  • x – values in linear power units, as a number, array, or pandas/xarray object

  • abs – True to convert abs(x); a complex x is converted by magnitude in either case

  • bounded – True to floor the result at 10*log10(tiny), where tiny is the smallest normal value of the computation dtype (about -379 dB in float32 and -3077 dB in float64); zeros then map to the floor instead of -inf

  • overwrite_x – True to write the result into x when the dtype of x is at least min_dtype; a complex x is then returned as the real view of the buffer of x

  • min_dtype – the narrowest floating-point dtype of the computation; a narrower x is widened to min_dtype

Returns:

a Python number for a number, otherwise the array, pandas, or xarray type of x. An xarray units attribute is converted from linear power to dB.

Return type:

the dB values, real-valued, in the container type of x

Raises:

TypeError – if min_dtype is None or narrower than float32, or if no array can be associated with the type of x

striqt.analysis.spectrogram(iq: Array | AcquiredIQ, 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 power spectrogram of the IQ by STFT.

The input is the IQ before the analysis_bandwidth filter and after trigger alignment. The output array namespace (numpy or cupy) follows the input. Per capture, the output has dimensions (spectrogram_time, spectrogram_baseband_frequency). spectrogram_time in s runs from the start of the capture at the STFT hop, which fractional_overlap and frequency_resolution set. spectrogram_baseband_frequency in Hz is the FFT grid at frequency_resolution, truncated to the analysis_bandwidth about DC when trim_stopband is set and the bandwidth is finite. Values are power in dBm per noise bandwidth, which is frequency_resolution or, when set, integration_bandwidth. The noise_bandwidth attr is that bandwidth in Hz. long_name names the bandwidth. White noise with power spectral density N0 has a mean of N0 * noise_bandwidth in every bin for every window. A CW tone has frequency_resolution / resolution_bandwidth of the tone power in one bin, where the resolution_bandwidth attr is the equivalent noise bandwidth (in Hz) of the zero-padded window. Values are rounded to 2 decimals and stored as float16.

Parameters:

  • iq (Array or AcquiredIQ) – IQ waveform of shape (port, sample), or an AcquiredIQ from which the conditioned waveform is selected

  • capture (striqt.analysis.specs.Capture) – the acquisition parameters of iq

  • as_xarray ((bool or 'delayed')) – True for an xarray.DataArray, False for an (array, attrs) tuple, or ‘delayed’ for a DelayedDataArray

  • 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.spectrogram_histogram(iq: Array | AcquiredIQ, 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, 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 spectrogram pixels that fall in each power bin.

The input is the IQ before the analysis_bandwidth filter and after trigger alignment. The pixels are those of the spectrogram output with the same STFT fields (float32, unrounded), flattened over time and frequency. Per capture, the output has the dimension (spectrogram_power_bin,) as float32 and sums to 1. spectrogram_power_bin is the bin centers in dBm per noise bandwidth. The -inf and inf entries collect the pixels more than half a bin below power_low or above power_high. The noise_bandwidth attr (in Hz) is set on the output and on the bin coordinate.

Parameters:

  • iq (Array or AcquiredIQ) – IQ waveform of shape (port, sample), or an AcquiredIQ from which the conditioned waveform is selected

  • capture (striqt.analysis.specs.Capture) – the acquisition parameters of iq

  • as_xarray ((bool or 'delayed')) – True for an xarray.DataArray, False for an (array, attrs) tuple, or ‘delayed’ for a DelayedDataArray

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

striqt.analysis.spectrogram_ratio_histogram(iq: Array | AcquiredIQ, 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, 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 a histogram of the spectrogram power ratio between 2 ports.

The input is the IQ before the analysis_bandwidth filter and after trigger alignment. The spectrograms are the spectrogram output with the same STFT fields (float32, unrounded). The ratio is the pixel-by-pixel difference of the two spectrograms in dB. Per capture, the output has the dimension (spectrogram_ratio_power_bin,) as float32, with bin centers in dB. The first output port is the ratio of the first input port to the second. The second output port is the negative of the first. Normalization and the catch-all bins follow spectrogram_histogram. Neither the output nor the bin coordinate has a noise_bandwidth attr.

Raises:

ValueError – if the capture has a port count other than 2

Parameters:

  • iq (Array or AcquiredIQ) – IQ waveform of shape (port, sample), or an AcquiredIQ from which the conditioned waveform is selected

  • capture (striqt.analysis.specs.Capture) – the acquisition parameters of iq

  • as_xarray ((bool or 'delayed')) – True for an xarray.DataArray, False for an (array, attrs) tuple, or ‘delayed’ for a DelayedDataArray

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

striqt.analysis.specs

class striqt.analysis.specs.Analysis

Bases: SpecBase

the base of every measurement spec.

The fields of a subclass are the keyword arguments of one measurement.

class striqt.analysis.specs.AnalysisCapture(*, duration=0.1, sample_rate=15360000.0, analysis_bandwidth=inf, center_frequency=None)

Bases: Capture

the projection of a capture onto the fields that a measurement may read.

Two captures with equal values in these fields project to equal instances.

Parameters:

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

  • sample_rate (float) – Sample rate of the analyzed IQ waveform (Constraints: > 0, units: S/s)

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

  • center_frequency (float | tuple[float, ...] | None) – Center frequency for each port (Constraints: units: Hz, > 0, units: Hz, > 0, units: Hz)

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.BuiltinAnalyses(*, channel_power_time_series=None, channel_power_histogram=None, cyclic_channel_power=None, spectrogram=None, power_spectral_density=None, spectrogram_histogram=None, spectrogram_ratio_histogram=None, cellular_5g_pss_sync=None, cellular_5g_pss_correlation=None, cellular_5g_sss_sync=None, cellular_5g_sss_correlation=None, cellular_5g_ssb_spectrogram=None, cellular_cyclic_autocorrelation=None, cellular_resource_power_histogram=None, iq_waveform=None)

Bases: AnalysisGroup

the analysis: block of a sweep: one optional field per registered measurement.

A field set to a spec enables the measurement with those parameters, and None leaves the measurement out. to_dict omits the None fields.

Parameters:
class striqt.analysis.specs.BuiltinTriggers(*, cellular_5g_pss_sync=None, cellular_5g_sss_sync=None)

Bases: AnalysisGroup

the struct form of the signal_trigger: field of a source.

One optional field per trigger, and exactly one is set. The set field names the trigger, and the value is the spec of the trigger measurement. With this form, the analysis: block need not enable the trigger measurement. The string form of signal_trigger names a registered trigger and takes the spec from the analysis: block.

Parameters:
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) – Sample rate of the analyzed IQ waveform (Constraints: > 0, 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)

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) – mask with float(“nan”) at this bandwidth centered at baseband DC (Constraints: units: Hz)

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']) – frequency resolution of the spectrogram: False for 1 subcarrier, True for 1 resource block (12 subcarriers), or ‘half’ for 6 subcarriers

  • average_slots (bool) – True to coarsen spectrogram bins by averaging 1-symbol time resolution

  • guard_bandwidths (tuple[float, float]) – Bandwidths excluded at the (low, high) edges of the analysis band (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', 'extended']) – cyclic prefix length of the 3GPP numerology (TS 38.211 Table 4.2-1)

  • 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 detector bin duration, a whole multiple of the sample period (Constraints: units: s)

  • power_detectors (tuple[str, ...]) – Power detectors evaluated in each bin: ‘rms’, ‘peak’, or both

  • 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 detector bin duration, a whole multiple of the sample period (Constraints: units: s)

  • power_detectors (tuple[str, ...]) – Power detectors evaluated in each bin: ‘rms’, ‘peak’, or both

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

Bases: Analysis

Parameters:

  • cyclic_period (float) – Cycle period of the channel power statistics, a whole multiple of detector_period and a common multiple of the expected signal periods (Constraints: units: s)

  • detector_period (Fraction) – Power detector bin duration, a whole multiple of the sample period (Constraints: units: s)

  • power_detectors (tuple[str, ...]) – Power detectors evaluated in each bin: ‘rms’, ‘peak’, or both

  • cyclic_statistics (tuple[str | float, ...]) – Statistics evaluated across cycles at each cycle lag: ‘min’, ‘mean’, ‘max’, ‘median’, or a quantile in (0, 1)

class striqt.analysis.specs.ErrorBound(*, rms, peak)

Bases: SpecBase

an error bound on the elements of one output.

rms bounds the error over all elements and peak the error at any one element, both in the units of the Tolerance field.

Parameters:

  • rms (float)

  • peak (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) – Sample rate of the analyzed IQ waveform (Constraints: > 0, 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

the STFT settings shared by the spectrum measurements.

The FFT size is the capture sample_rate divided by frequency_resolution, and the quotient must be a whole number. fractional_overlap and window_fill are the hop and the zero padding of each window as fractions of that size. integration_bandwidth is the width of the adjacent bins summed after the FFT, a multiple of frequency_resolution, or None for no summing. trim_stopband True limits the frequency axis to the analysis_bandwidth of the capture when that bandwidth is finite.

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

the base of every spec struct.

An instance is a frozen, keyword-only msgspec.Struct that rejects unknown fields. Construction converts list and dict field values to tuple and frozendict. An instance round-trips through to_dict and from_dict unchanged.

classmethod from_dict(d: dict | frozendict) → _T

return an instance of cls converted from the field values in d.

Parameters:

d – field values keyed by name; a float for a fractions.Fraction field converts to the fraction of its decimal representation

Raises:

msgspec.ValidationError – when a required field is missing, a key is not a field, or a value does not convert to the annotated type

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 the field values of self as a dict of builtin values.

Parameters:

  • unfreeze – True converts nested tuple and frozendict values to list and dict

  • allow_tuple_keys – False renders tuple dict keys as JSON array strings

Returns:

a dict keyed by field name; fractions.Fraction values are strings and frozendict values are dicts

validate() → Self

return an equal instance rebuilt through to_dict and from_dict.

Raises:

msgspec.ValidationError – when a field value does not satisfy the annotated type or constraints

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)

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)

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.Tolerance(*, units, rtol, on_peak, off_peak_dBc=None)

Bases: SpecBase

the roundoff error budget of one measurement output.

The budget depends on the capture and the spec alone, not on the output values. on_peak bounds the error of the elements at the peak of the output. off_peak_dBc bounds the error of the other elements as a function of the level of each below the peak. Elements more than off_peak_dBc.peak below the peak are not bounded. The peak is the matched-filter bin of the matched input, e.g., the FFT bin at the frequency of a tone or the detector sample at the time of an impulse.

Parameters:

  • units (str) – units of on_peak: ‘dB’, or the linear units of the output

  • rtol (float) – relative error bound on each output value, as a ratio

  • on_peak (ErrorBound) – absolute error bound, in units, on the elements at the peak of the output

  • off_peak_dBc (ErrorBound | None) – levels below the output peak, as negative values, at which the rms and the worst-case amplitude errors equal the amplitude of the element, or None to apply on_peak to every element (Constraints: units: dBc)

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