Transforms

Transforms are applied to signals or samples to emulate transmitter and reciever effects, as well as tools for machine learning. There are four types of transforms, that differ in purpose and scope.

  1. torchsig.transforms.signal_transforms.SignalTransform - applied to isolated signals from the signal builder, and typically represent transmitter effects.

  2. torchsig.transforms.dataset_transforms.DatasetTransform - applied to samples, after isolated signals are placed onto a noise floor. Typically represents reciever effects and other machine learning transforms.

  3. Functionals - core logic of both Signal Transforms and Dataset Transforms. Users can use for more fine-grained control of the transform.

  4. torchsig.transforms.impairments.DatasetImpairments - a collection of Signal Transforms and Dastaset Transforms that represent an environment, such as wireless.

Transforms

Base Transforms

Base Transforms

class torchsig.transforms.base_transforms.Transform(**kwargs)[source]

Bases: ABC, Seedable

Transform abstract class.

update(signal: Signal | DatasetSignal) → None[source]

Update bookeeping for signals

Parameters:

signal (Signal | DatasetSignal) – signal to update metadata.

Raises:

NotImplementedError – Inherited classes must override this method.

class torchsig.transforms.base_transforms.Compose(transforms: List[Transform], **kwargs)[source]

Bases: Transform

Composes several transforms together sequentially, in order.

transforms

list of Transform objects.

Type:

List[Transform]

class torchsig.transforms.base_transforms.Lambda(func: Callable, **kwargs)[source]

Bases: Transform

Apply a user-defined lambda as a transform.

Warning: does not automatically update metadata

func

Lambda/function to be used for transform.

Type:

Callable

Example

>>> from torchsig.transforms.base_transforms import Lambda
>>> transform = Lambda(lambda x: x**2)  # A transform that squares all inputs.
class torchsig.transforms.base_transforms.Normalize(norm: int | float | Literal['fro', 'nuc'] | None = 2, flatten: bool = False, seed: int | None = None, **kwargs)[source]

Bases: Transform

Normalize an IQ data vector.

norm

Order of the norm (refer to numpy.linalg.norm).

Type:

str

flatten

Specifies if the norm should be calculated on the flattened representation of the input tensor.

Type:

bool

Example

>>> import torchsig.transforms as ST
>>> transform = ST.Normalize(norm=2) # normalize by l2 norm
>>> transform = ST.Normalize(norm=1) # normalize by l1 norm
>>> transform = ST.Normalize(norm=2, flatten=True) # normalize by l1 norm of the 1D representation
class torchsig.transforms.base_transforms.RandomApply(transform, probability: float, **kwargs)[source]

Bases: Transform

Randomly applies transform with probability p.

transform

Transform to randomly apply.

Type:

Transform

probability

Probability to apply transform in range [0., 1.].

Type:

float

class torchsig.transforms.base_transforms.RandAugment(transforms: List[Transform], choose: int = 2, replace: bool = False, seed: int | None = None, **kwargs)[source]

Bases: Transform

RandAugment transform loosely based on: `”RandAugment: Practical automated data augmentation with a reduced search space”

transforms

list of Transforms to choose from.

Type:

List[Transform]

choose

Number of Transforms to randomly choose. Defaults to 2.

Type:

int, optional

replace

Allow replacement in random choose. Defaults to False.

Type:

bool, optional

Signal Transforms

SignalTransforms on Signal objects.

class torchsig.transforms.signal_transforms.SignalTransform(**kwargs)[source]

Bases: Transform

SignalTransform parent class.

update(signal: Signal) → None[source]

Updates bookkeeping to transforms in Signal’s SignalMetadata and checks signal valididty. Inherited classes should always call self.update() after performing transform operation (inside __call__).

Parameters:

signal (Signal) – Transformed signal.

class torchsig.transforms.signal_transforms.CarrierPhaseOffsetSignalTransform(phase_offset_range: Tuple[float, float] = (0, 6.283185307179586), **kwargs)[source]

Bases: SignalTransform

SignalTransform that applies a randomized carrier phase offset to Signal IQ data.

The randomized phase offset is of the form exp(j * phi) where phi is in the range of 0 to 2pi radians. Real world effects such as time delays as a signal transits the air and others can cause such randomized phase offsets.

The transform does not usually require any arguments due to its simplicity. It is generally unrealistic to have a randomized phase offset of a range less than 0 to 2pi.

phase_offset_range

Range bounds for phase offset (radians).

Type:

Tuple[float, float]

phase_offset_distribution

Random draw from phase offset distribution.

Type:

Callable[[], float]

class torchsig.transforms.signal_transforms.Fading(coherence_bandwidth: float | Tuple[float, float] | List[float] | Callable[[int], float] = (0.01, 0.1), power_delay_profile: Tuple | List | ndarray = (1, 1), **kwargs)[source]

Bases: SignalTransform

SignalTransform that applies a channel fading model.

Note, currently only performs Rayleigh fading:

A Rayleigh fading channel can be modeled as an FIR filter with Gaussian distributed taps which vary over time. The length of the filter determines the coherence bandwidth of the channel and is inversely proportional to the delay spread. The rate at which the channel taps vary over time is related to the coherence time and this is inversely proportional to the maximum Doppler spread. This time variance is not included in this model.

coherence_bandwidth

Coherence bandwidth sampling parameters. Defaults to (0.01, 0.1).

Type:

FloatParameter, optional

coherence_bandwidth_distribution

Random draw from coherence bandwidth distribution.

Type:

Callable[[], float]

power_delay_profile

A list of positive values assigning power to taps of the channel model. When the number of taps exceeds the number of items in the provided power_delay_profile, the list is linearly interpolated to provide values for each tap of the channel. Defaults to (1, 1).

Type:

Tuple | List | np.ndarray, optional

class torchsig.transforms.signal_transforms.IQImbalanceSignalTransform(amplitude_imbalance: float | Tuple[float, float] | List[float] | Callable[[int], float] | int | Tuple[int, int] | List[int] | Callable[[int], int] = (-1.0, 1.0), phase_imbalance: float | Tuple[float, float] | List[float] | Callable[[int], float] | int | Tuple[int, int] | List[int] | Callable[[int], int] = (-0.08726646259971647, 0.08726646259971647), dc_offset: float | Tuple[float, float] | List[float] | Callable[[int], float] | int | Tuple[int, int] | List[int] | Callable[[int], int] = ((-0.1, 0.1), (-0.1, 0.1)), **kwargs)[source]

Bases: SignalTransform

Applies a set of IQImbalance effects to a Signal: amplitude, phase, and DC offset.

amplitude_imbalance

Range bounds of IQ amplitude imbalance (dB).

Type:

NumericParameter, optional

amplitude_imbalance_distribution

Random draw from amplitude imbalance distribution.

Type:

Callable[[], float]

phase_imbalance

Range bounds of IQ phase imbalance (radians).

Type:

NumericParameter, optional

phase_imbalance

Random draw from phase imbalance distribution.

Type:

Callable[[], float]

dc_offset

Range bounds for I and Q component DC offsets (NumericParameters).

Type:

Tuple, optional

dc_offset

Random draw from dc_offset distribution.

Type:

Callable[[], (float, float)]

class torchsig.transforms.signal_transforms.SpectralInversionSignalTransform(**kwargs)[source]

Bases: SignalTransform

Inverts spectrum of complex IQ data.

Dataset Transforms

DatasetTransforms on DatasetSignal objects.

class torchsig.transforms.dataset_transforms.DatasetTransform(**kwargs)[source]

Bases: Transform

Dataset Transform base class

Dataset Transforms are transforms applied to DatasetSignals.

update(signal: DatasetSignal) → None[source]

Updates bookkeeping to transforms in DatasetSignal’s SignalMetadata and checks signal valididty. Inherited classes should always call self.update() after performing transform operation (inside __call__).

Parameters:

signal (DatasetSignal) – transformed DatasetSignal.

class torchsig.transforms.dataset_transforms.AGC(rand_scale: float | Tuple[float, float] | List[float] | Callable[[int], float] = (1.0, 10.0), initial_gain_db: float = 0.0, alpha_smooth: float = 4e-05, alpha_track: float = 0.0004, alpha_overflow: float = 0.3, alpha_acquire: float = 0.04, ref_level_db: float = 0.0, track_range_db: float = 1.0, low_level_db: float = -80.0, high_level_db: float = 6.0, **kwargs)[source]

Bases: DatasetTransform

Automatic Gain Control performing sample-by-sample AGC algorithm.

rand_scale

FloatParameter setting the random scaling bounds for each sample update.

Type:

Tuple

rand_scale_distribution

Random draw from rand_scale distribution.

Type:

Callable[[], float]

initial_gain_db

Inital gain value in dB.

Type:

float

alpha_smooth

Alpha for avergaing the measure signal level level_n = level_n * alpha + level_n-1(1-alpha)

Type:

float

alpha_track

Amount to adjust gain when in tracking state.

Type:

float

alpha_overflow

Amount to adjust gain when in overflow state [level_db + gain_db] >= max_level.

Type:

float

alpha_acquire

Amount to adjust gain when in acquire state.

Type:

float

ref_level_db

Reference level goal for algorithm to achieve, in dB units.

Type:

float

track_range_db

dB range for operating in tracking state.

Type:

float

low_level_db

minimum magnitude value (dB) to perform any gain control adjustment.

Type:

float

high_level_db

magnitude value (dB) to enter overflow state.

Type:

float

class torchsig.transforms.dataset_transforms.AWGN(noise_power_db: float, **kwargs)[source]

Bases: DatasetTransform

Apply Additive White Gaussian Noise to DatasetSignal.

noise_power_db

noise AWGN power in dB (absolute).

Type:

float

class torchsig.transforms.dataset_transforms.BlockAGC(max_gain_change_db: float = 10.0, **kwargs)[source]

Bases: DatasetTransform

Implements a large instantaneous jump in receiver gain.

gain_change_db_range

Sets the (min, max) gain change in dB.

Type:

Tuple

gain_change_db_distribution

Random draw from gain_change_db distribution.

Type:

Callable[[], float]

class torchsig.transforms.dataset_transforms.CarrierPhaseOffsetDatasetTransform(phase_offset_range: Tuple[float, float] = (0, 6.283185307179586), **kwargs)[source]

Bases: DatasetTransform

Apply randomized phase offset to signal I/Q data.

phase_offset_range

Phase range bounds in radians.

Type:

Tuple[float, float]

phase_offset_distribution

Random draw from phase_offset distribution.

Type:

Callable[[], float]

class torchsig.transforms.dataset_transforms.ComplexTo2D(**kwargs)[source]

Bases: DatasetTransform

Converts IQ data to two channels (real and imaginary parts).

class torchsig.transforms.dataset_transforms.IQImbalanceDatasetTransform(amplitude_imbalance: float | Tuple[float, float] | List[float] | Callable[[int], float] | int | Tuple[int, int] | List[int] | Callable[[int], int] = (-1.0, 1.0), phase_imbalance: float | Tuple[float, float] | List[float] | Callable[[int], float] | int | Tuple[int, int] | List[int] | Callable[[int], int] = (-0.08726646259971647, 0.08726646259971647), dc_offset: float | Tuple[float, float] | List[float] | Callable[[int], float] | int | Tuple[int, int] | List[int] | Callable[[int], int] = (-0.1, 0.1), **kwargs)[source]

Bases: DatasetTransform

Applies a set of IQImbalance effects to a Signal: amplitude, phase, and DC offset.

amplitude_imbalance

Range bounds of IQ amplitude imbalance (dB).

Type:

NumericParameter, optional

amplitude_imbalance_distribution

Random draw from amplitude imbalance distribution.

Type:

Callable[[], float]

phase_imbalance

Range bounds of IQ phase imbalance (radians).

Type:

NumericParameter, optional

phase_imbalance

Random draw from phase imbalance distribution.

Type:

Callable[[], float]

dc_offset

Range bounds for I and Q component DC offsets (NumericParameters).

Type:

Tuple, optional

dc_offset

Random draw from dc_offset distribution.

Type:

Callable[[], (float, float)]

class torchsig.transforms.dataset_transforms.Quantize(num_levels: float | Tuple[float, float] | List[float] | Callable[[int], float] | int | Tuple[int, int] | List[int] | Callable[[int], int] = [16, 24, 32, 40, 48, 56, 64], round_type: List[str] = ['floor', 'middle', 'ceiling'], **kwargs)[source]

Bases: DatasetTransform

Quantize signal I/Q samples into specified levels with a rounding method.

num_levels

Number of quantization levels.

Type:

NumericParameter

num_levels_distribution

Random draw from num_levels distribution.

Type:

Callable[[], int]

round_type

Quantization rounding method. Must be ‘floor’, ‘nearest’ or ‘ceiling’. Defaults to ‘ceiling’.

Type:

str, List[str]

round_type_distribution

Random draw from round_type distribution.

Type:

Callable[[], str]

class torchsig.transforms.dataset_transforms.Spectrogram(fft_size: int, **kwargs)[source]

Bases: DatasetTransform

Computes the spectogram of IQ data.

fft_size

The FFT size (number of bins) in the spectrogram

Type:

int

class torchsig.transforms.dataset_transforms.SpectralInversionDatasetTransform(**kwargs)[source]

Bases: DatasetTransform

Invert spectrum of a DatasetSignal.

class torchsig.transforms.dataset_transforms.TimeVaryingNoise(noise_power_low: float | Tuple[float, float] | List[float] | Callable[[int], float] | int | Tuple[int, int] | List[int] | Callable[[int], int] = (-80.0, -60.0), noise_power_high: float | Tuple[float, float] | List[float] | Callable[[int], float] | int | Tuple[int, int] | List[int] | Callable[[int], int] = (-40.0, -20.0), inflections: int | Tuple[int, int] | List[int] | Callable[[int], int] = [0, 10], random_regions: List | bool = True, **kwargs)[source]

Bases: DatasetTransform

Add time-varying noise to DatasetSignal regions.

noise_power_low

Range bounds for minimum noise power in dB.

Type:

NumericParameter

noise_power_low_distribution

Random draw from noise_power_low distribution.

Type:

Callable[[], float]

noise_power_high

Range bounds for maximum noise power in dB.

Type:

NumericParameter

noise_power_high_distribution

Random draw from noise_power_high distribution.

Type:

Callable[[], float]

inflections

Number of inflection points over IQ data.

Type:

IntParameter

inflections_distribution

Random draw from inflections distribution.

Type:

Callable[[], float]

random_regions

Inflections points spread randomly (True) or not (False).

Type:

List | bool

random_regions_distribution

Random draw from random_regions distribution.

Type:

Callable[[], bool]

class torchsig.transforms.dataset_transforms.AddSlope(**kwargs)[source]

Bases: DatasetTransform

Add the slope of each sample with its preceeding sample to itself. Creates a weak 0 Hz IF notch filtering effect.

class torchsig.transforms.dataset_transforms.ChannelSwap(**kwargs)[source]

Bases: DatasetTransform

Swaps the I and Q channels of complex input data.

class torchsig.transforms.dataset_transforms.CutOut(duration: float | Tuple[float, float] | List[float] | Callable[[int], float] | int | Tuple[int, int] | List[int] | Callable[[int], int] = (0.01, 0.2), cut_type: List[str] = ['zeros', 'ones', 'low_noise', 'avg_noise', 'high_noise'], **kwargs)[source]

Bases: DatasetTransform

Applies the CutOut transform operation in the time domain. The cut_dur input specifies how long the cut region should be, and the cut_fill input specifies what the cut region should be filled in with. Options for the cut type include: zeros, ones, low_noise, avg_noise, and high_noise. Zeros fills in the region with zeros; ones fills in the region with 1+1j samples; low_noise fills in the region with noise with -100dB power; avg_noise adds noise at power average of input data, effectively slicing/removing existing signals in the most RF realistic way of the options; and high_noise adds noise with 40dB power. If a list of multiple options are passed in, they are randomly sampled from.

This transform is loosely based on “Improved Regularization of Convolutional Neural Networks with Cutout”.

duration

cut_dur sets the duration of the region to cut out * If float, cut_dur is fixed at the value provided. * If list, cut_dur is any element in the list. * If tuple, cut_dur is in range of (tuple[0], tuple[1]).

Type:

float, list, tuple

duration_distribution

Random draw from duration distribution.

Type:

Callable[[], float]

cut_type

cut_fill sets the type of data to fill in the cut region with from the options: zeros, ones, low_noise, avg_noise, and high_noise * If list, cut_fill is any element in the list. * If str, cut_fill is fixed at the method provided.

Type:

float, list, tuple

cut_type_distribution

Random draw from cut_type distribution.

Type:

Callable[[], str]

class torchsig.transforms.dataset_transforms.PatchShuffle(patch_size: float | Tuple[float, float] | List[float] | Callable[[int], float] | int | Tuple[int, int] | List[int] | Callable[[int], int] = (3, 10), shuffle_ratio: float | Tuple[float, float] | List[float] | Callable[[int], float] = (0.01, 0.05), **kwargs)[source]

Bases: DatasetTransform

Randomly shuffle multiple local regions of samples.

Transform is loosely based on “PatchShuffle Regularization”.

patch_size

patch_size sets the size of each patch to shuffle * If int or float, patch_size is fixed at the value provided. * If list, patch_size is any element in the list. * If tuple, patch_size is in range of (tuple[0], tuple[1]).

Type:

int, float, list, tuple

patch_size_distribution (Callable[[], int]): Random draw from patch_size distribution. shuffle_ratio (int, float, list, tuple):

shuffle_ratio sets the ratio of the patches to shuffle * If int or float, shuffle_ratio is fixed at the value provided. * If list, shuffle_ratio is any element in the list. * If tuple, shuffle_ratio is in range of (tuple[0], tuple[1]).

shuffle_ratio_distribution (Callable[[], float]): Random draw from shuffle_ratio distribution.

class torchsig.transforms.dataset_transforms.RandomDropSamples(drop_rate: float | Tuple[float, float] | List[float] | Callable[[int], float] | int | Tuple[int, int] | List[int] | Callable[[int], int] = (0.01, 0.05), size: float | Tuple[float, float] | List[float] | Callable[[int], float] | int | Tuple[int, int] | List[int] | Callable[[int], int] = (1, 10), fill: List[str] = ['ffill', 'bfill', 'mean', 'zero'], **kwargs)[source]

Bases: DatasetTransform

Randomly drop IQ samples from the input data of specified durations and with specified fill techniques:

  • ffill (front fill): replace drop samples with the last previous value.

  • bfill (back fill): replace drop samples with the next value.

  • mean: replace drop samples with the mean value of the full data.

  • zero: replace drop samples with zeros.

Transform is based off of the TSAug Dropout Transform.

drop_rate

drop_rate sets the rate at which to drop samples * If int or float, drop_rate is fixed at the value provided. * If list, drop_rate is any element in the list. * If tuple, drop_rate is in range of (tuple[0], tuple[1]).

Type:

int, float, list, tuple

drop_rate_distribution

Random draw from drop_rate distribution.

Type:

Callable[[], float]

size

size sets the size of each instance of dropped samples * If int or float, size is fixed at the value provided. * If list, size is any element in the list. * If tuple, size is in range of (tuple[0], tuple[1]).

Type:

int, float, list, tuple

size_distribution

Random draw from size distribution.

Type:

Callable[[], int]

fill

fill sets the method of how the dropped samples should be filled * If list, fill is any element in the list. * If str, fill is fixed at the method provided.

Type:

list, str

fill_distribution

Random draw from fill distribution.

Type:

Callable[[], str]

class torchsig.transforms.dataset_transforms.RandomMagRescale(start: float | Tuple[float, float] | List[float] | Callable[[int], float] | int | Tuple[int, int] | List[int] | Callable[[int], int] = (0.0, 0.9), scale: float | Tuple[float, float] | List[float] | Callable[[int], float] | int | Tuple[int, int] | List[int] | Callable[[int], int] = (-4.0, 4.0), **kwargs)[source]

Bases: DatasetTransform

Randomly apply a magnitude rescaling, emulating a change in a receiver’s gain control.

start

start sets the time when the rescaling kicks in * If int or float, start is fixed at the value provided. * If list, start is any element in the list. * If tuple, start is in range of (tuple[0], tuple[1]).

Type:

int, float, list, tuple

start_distribution (Callable[[], float]): Random draw from start distribution. scale (int, float, list, tuple):

scale sets the magnitude of the rescale * If int or float, scale is fixed at the value provided. * If list, scale is any element in the list. * If tuple, scale is in range of (tuple[0], tuple[1]).

scale_distribution (Callable[[], float]): Random draw from scale distribution.

class torchsig.transforms.dataset_transforms.SpectrogramDropSamples(drop_rate: float | Tuple[float, float] | List[float] | Callable[[int], float] | int | Tuple[int, int] | List[int] | Callable[[int], int] = (0.001, 0.005), size: float | Tuple[float, float] | List[float] | Callable[[int], float] | int | Tuple[int, int] | List[int] | Callable[[int], int] = (1, 10), fill: List[str] = ['ffill', 'bfill', 'mean', 'zero', 'low', 'min', 'max', 'ones'], **kwargs)[source]

Bases: DatasetTransform

Randomly drop samples from the input data of specified durations and with specified fill techniques:

  • ffill (front fill): replace drop samples with the last previous value

  • bfill (back fill): replace drop samples with the next value

  • mean: replace drop samples with the mean value of the full data

  • zero: replace drop samples with zeros

  • low: replace drop samples with low power samples

  • min: replace drop samples with the minimum of the absolute power

  • max: replace drop samples with the maximum of the absolute power

  • ones: replace drop samples with ones

Transform is based off of the TSAug Dropout Transform.

drop_rate

drop_rate sets the rate at which to drop samples * If int or float, drop_rate is fixed at the value provided. * If list, drop_rate is any element in the list. * If tuple, drop_rate is in range of (tuple[0], tuple[1]).

Type:

int, float, list, tuple

drop_rate_distribution

Random draw from drop_rate distribution.

Type:

Callable[[], float]

size

size sets the size of each instance of dropped samples * If int or float, size is fixed at the value provided. * If list, size is any element in the list. * If tuple, size is in range of (tuple[0], tuple[1]).

Type:

int, float, list, tuple

size_distribution

Random draw from size distribution.

Type:

Callable[[], int]

fill

fill sets the method of how the dropped samples should be filled * If list, fill is any element in the list. * If str, fill is fixed at the method provided.

Type:

list, str

fill_distribution

Random draw from fill distribution.

Type:

Callable[[], float]

class torchsig.transforms.dataset_transforms.TimeReversal(allow_spectral_inversion: bool | float = True, **kwargs)[source]

Bases: DatasetTransform

Applies a time reversal to the input.

Note that applying a time reversal inherently also applies a spectral inversion. If a time-reversal without spectral inversion is desired, the undo_spectral_inversion argument can be set to True. By setting this value to True, an additional, manual spectral inversion is applied to revert the time-reversal’s inversion effect.

allow_spectral_inversion

Whether to allow spectral inversion.

Type:

bool | float, optional

as a time reversal side effect
Type:

True) or not (False

\* If bool, applied to all signals.
\* If float, applied as probability to add signals.

Impairments

Base Impairments

Dataset Transform/Impairment base class

class torchsig.transforms.impairments.Impairments(all_levels_signal_transforms: List[SignalTransform], all_levels_dataset_transforms: List[DatasetTransform], level: int, **kwargs)[source]

Bases: Transform

Applies signal and dataset transformations at specific impairment levels.

This class applies a set of signal and dataset transforms based on a given impairment level. The impairment level must be between 0 and 2, where each level corresponds to different sets of transformations for signals and datasets. * Level 0: Perfect * Level 1: Cabled enviornment * Level 2: Wireless environment

Parameters:

  • all_levels_signal_transforms (List[SignalTransform]) – A list of signal transformations for all impairment levels.

  • all_levels_dataset_transforms (List[DatasetTransform]) – A list of dataset transformations for all impairment levels.

  • level (int) – The impairment level (must be between 0 and 2).

  • **kwargs – Additional keyword arguments passed to the parent class Transform.

Raises:

ValueError – If the provided impairment level is outside the valid range (0, 1, 2).

level

The specified impairment level.

Type:

int

signal_transforms

The composed signal transformations corresponding to the given impairment level.

Type:

Compose

dataset_transforms

The composed dataset transformations corresponding to the given impairment level.

Type:

Compose

Narrowband Impairments

Narrowband Transforms and Impairments for Impairment Levels 0-2

Impairments are transforms applied to Signal objects, after the Signal Builder generates an isolated signal. Transforms are applied to DatasetSignal objects, after isolated signals are placed on an IQ cut of noise.

Example

>>> impairments = NarrowbandImpairments(level = 2, dataset_metadata=dm)
>>> sb = SignalBuilder(...)
>>> new_signal = sb.build()
>>> impaired_new_signal = impairments(new_signal)

>>> iq_samples = <random noise>
>>> iq_samples[start:stop] += new_signal.data
>>> new_dataset_signal = DatasetSignal(data=iq_samples, metadata=[impaired_new_signal.metadata])

>>> transforms = NarrowbandTransforms(level = 2, dataset_metadata=dm)
>>> transformed_dataset_signal = transforms(new_dataset_signal)
class torchsig.transforms.impairments_narrowband.NarrowbandImpairments(level: int, **kwargs)[source]

Bases: Impairments

Applies impairments to Narrowband dataset

Wideband Impairments

Wideband Transforms and Impairments for Impairment Levels 0-2

Impairments are transforms applied to Signal objects, after the Signal Builder generates an isolated signal. Transforms are applied to DatasetSignal objects, after isolated signals are placed on an IQ cut of noise.

Example

>>> impairments = WidebandImpairments(level = 2, dataset_metadata=dm)
>>> iq_samples = <random noise>
>>> metadatas = []
>>> for i in range(3): # 3 signals in wideband sample
>>>     sb = SignalBuilder(...)
>>>     new_signal = sb.build()
>>>     impaired_new_signal = impairments(new_signal)
>>>     iq_samples[start:stop] += new_signal.data
>>>     metadatas.append(impaired_new_signal.metadata)

>>> new_dataset_signal = DatasetSignal(data=iq_samples, metadata=metadatas)

>>> transforms = WidebandTransforms(level = 2, dataset_metadata=dm)
>>> transformed_dataset_signal = transforms(new_dataset_signal)
class torchsig.transforms.impairments_wideband.WidebandImpairments(level: int, **kwargs)[source]

Bases: Impairments

Applies impairements to Wideband dataset

Functional Transforms

Functional transforms for reuse and custom fine-grained control

torchsig.transforms.functional.add_slope(data: ndarray) → ndarray[source]

Add slope between each sample and its preceding sample is added to every sample.

Augmentation has the effect of amplifying high frequency component more than lower frequency components.

Parameters:

data (np.ndarray) – IQ data.

Returns:

IQ data with added slope.

Return type:

np.ndarray

torchsig.transforms.functional.agc(data: ndarray, initial_gain_db: float, alpha_smooth: float, alpha_track: float, alpha_overflow: float, alpha_acquire: float, ref_level_db: float, track_range_db: float, low_level_db: float, high_level_db: float) → ndarray[source]

Automatic Gain Control algorithm (deterministic).

Parameters:

  • data (np.ndarray) – IQ data samples.

  • initial_gain_db (float) – Inital gain value in dB.

  • alpha_smooth (float) – Alpha for avergaing the measure signal level level_n = level_n * alpha + level_n-1(1-alpha)

  • alpha_track (float) – Amount to adjust gain when in tracking state.

  • alpha_overflow (float) – Amount to adjust gain when in overflow state [level_db + gain_db] >= max_level.

  • alpha_acquire (float) – Amount to adjust gain when in acquire state.

  • ref_level_db (float) – Reference level goal for algorithm to achieve, in dB units.

  • track_range_db (float) – dB range for operating in tracking state.

  • low_level_db (float) – minimum magnitude value (dB) to perform any gain control adjustment.

  • high_level_db (float) – magnitude value (dB) to enter overflow state.

Returns:

IQ data adjusted sample-by-sample by the AGC algorithm.

Return type:

np.ndarray

torchsig.transforms.functional.block_agc(data: ndarray, gain_change_db: float, start_idx: int) → ndarray[source]

Implements a large instantaneous jump in receiver gain.

Parameters:

  • data (np.ndarray) – IQ data.

  • gain_change_db (float) – Gain value to change in dB.

  • start_idx (np.ndarray) – Start index for IQ data.

Returns:

IQ data with Block AGC applied.

Return type:

np.ndarray

torchsig.transforms.functional.channel_swap(data: ndarray) → ndarray[source]

Swap I and Q channels of IQ data.

Parameters:

data (np.ndarray) – IQ data.

Returns:

IQ data with channels swapped.

Return type:

np.ndarray

torchsig.transforms.functional.complex_to_2d(data: ndarray) → ndarray[source]

Converts IQ data to two channels (real and imaginary parts).

torchsig.transforms.functional.cut_out(data: ndarray, cut_start: float, cut_duration: float, cut_type: str, rng: Generator | None = None) → ndarray[source]

Performs CutOut: replacing values with fill.

Parameters:

  • data (np.ndarray) – IQ data

  • cut_start (float) – Normalized start of cut region [0.0, 1.0)

  • cut_duration (float) – Normalized duration of cut region (0.0, 1.0)

  • cut_type (str) – Type of data to fill cut region.

  • zeros (*)

  • ones (*)

  • low_noise (*)

  • avg_noise (*)

  • high_noise (*)

Raises:

ValueError – Invalid cut_type.

Returns:

CutOut IQ data.

Return type:

np.ndarray

torchsig.transforms.functional.drop_samples(data: ndarray, drop_starts: ndarray, drop_sizes: ndarray, fill: str) → ndarray[source]

Drop samples at given locations/durations with fill technique.

Supported Fill Techniques:

ffill: Forward Fill. Use value at sample one before start. bfill: Backwards Fill. Use value at sample one after end. mean: Mean Fill. Use data mean. zero: Zero Fill. Use 0.

Parameters:

  • data (np.ndarray) – IQ data.

  • drop_starts (np.ndarray) – Start indicies of drops.

  • drop_sizes (np.ndarray) – Durations for each start index.

  • fill (str) – Drop sample replacement method.

Raises:

ValueError – Invalid fill type.

Returns:

data array with fill values during drops.

Return type:

np.ndarray

torchsig.transforms.functional.fading(data: ndarray, coherence_bandwidth: float, power_delay_profile: ndarray, rng: Generator) → ndarray[source]

Apply fading channel to signal. Currently only does Rayleigh fading.

Taps are generated by interpolating and filtering Gaussian taps.

Parameters:

  • data (np.ndarray) – IQ data.

  • coherence_bandwidth (float) – coherence bandwidth relative to sample rate [0, 1.0].

  • power_delay_profile (np.ndarray) – power delay profile assign to channel.

  • rng (Optional[np.random.Generator], optional) – Random Generator to use. Defaults to None (new generator created internally).

Returns:

IQ data with fading applied.

Return type:

np.ndarray

torchsig.transforms.functional.iq_imbalance(data: ndarray, amplitude_imbalance: float, phase_imbalance: float, dc_offset: Tuple[float, float]) → ndarray[source]

Applies IQ imbalance to IQ data.

Parameters:

  • data (np.ndarray) – IQ data.

  • amplitude_imbalance (float) – IQ amplitude imbalance in dB.

  • phase_imbalance (float) – IQ phase imbalance in radians [-pi, pi].

  • dc_offset (Tuple[float, float]) – IQ DC (linear) offsets (In-Phase, Quadrature).

Returns:

IQ data with IQ Imbalance applied.

Return type:

np.ndarray

torchsig.transforms.functional.local_oscillator_frequency_drift()[source]

Unimplemented Functional for modeling Local Oscillator drift in frequency.

torchsig.transforms.functional.local_oscillator_phase_noise()[source]

Unimplemented Functional for modeling Local Oscillator phase noise.

torchsig.transforms.functional.mag_rescale(data: ndarray, start: float | int, scale: float) → ndarray[source]

Apply rescaling of input rescale starting at time start.

Parameters:

  • data (np.ndarray) – IQ data.

  • start (float | int) – Start time of rescaling.

  • int (* If)

  • index. (treated as array)

  • float (* If)

  • time. (treated as normalized start)

  • scale (float) – Scaling factor.

Returns:

data rescaled.

Return type:

np.ndarray

torchsig.transforms.functional.nonlinear_amplifier()[source]

Unimplemented Functional for memoryless nonlinear amplifier response.

torchsig.transforms.functional.normalize(data: ndarray, norm_order: float | int | Literal['fro', 'nuc'] | None = 2, flatten: bool = False) → ndarray[source]
Scale data so that a specfied norm computes to 1. For detailed information, see numpy.linalg.norm.()

  • For norm=1, norm = max(sum(abs(x), axis=0)) (sum of the elements)

  • for norm=2, norm = sqrt(sum(abs(x)^2), axis=0) (square-root of the sum of squares)

  • for norm=np.inf, norm = max(sum(abs(x), axis=1)) (largest absolute value)

Parameters:

  • data (np.ndarray) – (batch_size, vector_length, …)-sized data to be normalized.

  • norm_order (int) – norm order to be passed to np.linalg.norm

  • flatten (bool) – boolean specifying if the input array’s norm should be calculated on the flattened representation of the input data

Returns:

Normalized complex array data.

Return type:

np.ndarray

torchsig.transforms.functional.passband_ripple()[source]

Unimplemented Functional to create passband ripple filter effects within the sampling bandwidth.

torchsig.transforms.functional.patch_shuffle(data: ndarray, patch_size: int, patches_to_shuffle: ndarray, rng: Generator | None = None) → ndarray[source]

Apply shuffling of patches specified by num_patches.

Parameters:

  • data – (np.ndarray): (batch_size, vector_length, …)-sized data.

  • patch_size (int) – Size of each patch to shuffle.

  • patches_to_shuffle (np.ndarray) – Index of each patch of size patch_size to shuffle.

  • random_generator (Optional[np.random.Generator], optional) – Random Generator to use. Defaults to None (new generator created internally).

Returns:

Data that has undergone patch shuffling.

Return type:

np.ndarray

torchsig.transforms.functional.phase_offset(data: ndarray, phase: float) → ndarray[source]

Applies a phase rotation to data.

Parameters:

  • data (np.ndarray) – IQ data.

  • phase (float) – phase to rotate sample in [-pi, pi].

Returns:

Data that has undergone a phase rotation.

Return type:

np.ndarray

torchsig.transforms.functional.quantize(data: ndarray, num_levels: int, round_type: str = 'ceiling') → ndarray[source]

Quantize input to number of levels specified.

Default implementation is ceiling.

Parameters:

  • data (np.ndarray) – IQ data.

  • num_levels (int) – Number of quantization levels

  • round_type (str, optional) – Quantization rounding. Must be one of ‘floor’, ‘nearest’ or ‘ceiling’. Defaults to ‘ceiling’.

Raises:

ValueError – Invalid round type.

Returns:

Quantized IQ data.

Return type:

np.ndarray

torchsig.transforms.functional.spectral_inversion(data: ndarray) → ndarray[source]

Applies a spectral inversion to input data.

Parameters:

data (np.ndarray) – IQ data.

Returns:

Spectrally inverted data.

Return type:

np.ndarray

torchsig.transforms.functional.spectrogram(data: ndarray, fft_size: int, fft_stride: int) → ndarray[source]

Computes spectrogram from IQ data. Directly uses compute_spectrogram inside of utils/dsp.py.

Parameters:

  • data (np.ndarray) – IQ samples.

  • fft_size (int) – The FFT size (number of bins) in the spectrogram.

  • fft_stride (int) – The number of data points to move or “hop” over when computing the next FFT.

  • rng (np.random.Generator) – Optional random generator.

Returns:

Spectrogram computed from IQ data.

Return type:

np.ndarray

torchsig.transforms.functional.spectrogram_drop_samples(data: ndarray, drop_starts: ndarray, drop_sizes: ndarray, fill: str) → ndarray[source]

Drop samples at given locations/durations with fill technique.

Supported Fill Techniques:

ffill: Forward Fill. Use value at sample one before start. bfill: Backwards Fill. Use value at sample one after end. mean: Mean Fill. Use data mean. zero: Zero Fill. Use 0. min: Minimum observed value fill. max: Maximum observed value fill low: Fixed low value fill. Use np.ones * 1e-3. ones: Ones fill. Use np.ones.

Parameters:

  • data (np.ndarray) – IQ data.

  • drop_starts (np.ndarray) – Start indicies of drops.

  • drop_sizes (np.ndarray) – Durations for each start index.

  • fill (str) – Drop sample replacement method.

Raises:

ValueError – Invalid fill type.

Returns:

data array with fill values during drops.

Return type:

np.ndarray

torchsig.transforms.functional.time_reversal(data: ndarray) → ndarray[source]

Applies time reversal to data (flips horizontally).

Parameters:

data (np.ndarray) – IQ data.

Returns:

Time flipped IQ data.

Return type:

np.ndarray

torchsig.transforms.functional.time_varying_noise(data: ~numpy.ndarray, noise_power_low: float, noise_power_high: float, inflections: int, random_regions: bool, rng: ~numpy.random._generator.Generator = Generator(PCG64) at 0x7F88F6E489E0) → ndarray[source]

Adds time-varying complex additive white Gaussian noise with power levels in range (noise_power_low, noise_power_high) dB and with inflections number of inflection points spread over the input iq data randomly if random_regions is True or evenly spread if False.

Parameters:

  • data (np.ndarray) – IQ data.

  • noise_power_low (float) – Minimum noise power in dB.

  • noise_power_high (float) – Maximum noise power in dB.

  • inflections (int) – Number of inflection points over IQ data.

  • random_regions (bool) – Inflections points spread randomly (True) or not (False).

  • rng (np.random.Generator, optional) – Random number generator. Defaults to np.random.default_rng(seed=None).

Returns:

IQ data with time-varying noise.

Return type:

np.ndarray