Transforms (ria_toolkit_oss.transforms)

The transforms module houses a collection of functions to manipulate and transform radio data.

This module contains various functions that operate on NumPy arrays. These functions are utilized within the machine learning backends to build transforms and functions that seamlessly integrate with those from the respective backend.

All the transforms in this module expect data in the complex 1xN format.

IQ Impairments

This submodule comprises various transforms designed to represent signal impairments. These transforms take a recording as input and return a corresponding recording with the impairment model applied; we call the latter an impaired recording.

Signals travel through transmission media, which are not perfect. The imperfection causes signal impairment, meaning that the signal at the beginning of the medium is not the same as the signal at the end of the medium. What is sent is not what is received. Three causes of impairment are attenuation, distortion, and noise.

ria_toolkit_oss.transforms.iq_impairments.add_awgn_to_signal(signal, snr=1)[source]

Generates additive white gaussian noise (AWGN) relative to the signal-to-noise ratio (SNR) of the provided signal array or Recording.

This function calculates the root mean squared (RMS) power of signal and then finds the RMS power of the noise which matches the specified SNR. Then, the AWGN is generated after calculating the variance and randomly calculating the amplitude and phase of the noise. Then, this generated AWGN is added to the original signal and returned.

Parameters:

signal (array_like or Recording) – Input IQ data as a complex C x N array or Recording, where C is the number of channels and N is the length of the IQ examples.
snr (float, optional) – The signal-to-noise ratio in dB. Default is 1.

Raises:

ValueError – If signal is not CxN complex.

Returns:

A numpy array which is the sum of the noise (which matches the SNR) and the original signal. If signal is a Recording, returns a Recording object with its data attribute containing the noisy signal array.

Return type:

ndarray or Recording

>>> rec = Recording(data=[[1+1j, 2+2j]])
>>> new_rec = add_awgn_to_signal(rec)
>>> new_rec.data
array([[0.83141973+0.32529242j, -1.00909846+2.39282713j]])

ria_toolkit_oss.transforms.iq_impairments.time_shift(signal, shift=1)[source]

Apply a time shift to a signal.

After the time shift is applied, we fill any empty regions with zeros.

Parameters:

signal (array_like or Recording) – Input IQ data as a complex CxN array or Recording, where C is the number of channels and N is the length of the IQ examples.
shift (int, optional) – The number of indices to shift by. Default is 1.

Raises:

ValueError – If signal is not CxN complex.
UserWarning – If shift is greater than length of signal.

Returns:

A numpy array which represents the time-shifted signal. If signal is a Recording, returns a Recording object with its data attribute containing the time-shifted array.

Return type:

ndarray or Recording

>>> rec = Recording(data=[[1+1j, 2+2j, 3+3j, 4+4j, 5+5j]])
>>> new_rec = time_shift(rec, -2)
>>> new_rec.data
array([[3+3j, 4+4j, 5+5j, 0+0j, 0+0j]])

ria_toolkit_oss.transforms.iq_impairments.frequency_shift(signal, shift=0.5)[source]

Apply a frequency shift to a signal.

Note

The frequency shift is applied relative to the sample rate.

Parameters:

signal (array_like or Recording) – Input IQ data as a complex CxN array or Recording, where C is the number of channels and N is the length of the IQ examples.
shift (float, optional) – The frequency shift relative to the sample rate. Must be in the range [-0.5, 0.5]. Default is 0.5.

Raises:

ValueError – If the provided frequency shift is not in the range [-0.5, 0.5].
ValueError – If signal is not CxN complex.

Returns:

A numpy array which represents the frequency-shifted signal. If signal is a Recording, returns a Recording object with its data attribute containing the frequency-shifted array.

Return type:

ndarray or Recording

>>> rec = Recording(data=[[1+1j, 2+2j, 3+3j, 4+4j]])
>>> new_rec = frequency_shift(rec, -0.4)
>>> new_rec.data
array([[1+1j, -0.44246348-2.79360449j, -1.92611857+3.78022053j, 5.04029404-2.56815809j]])

ria_toolkit_oss.transforms.iq_impairments.phase_shift(signal, phase=3.141592653589793)[source]

Apply a phase shift to a signal.

Parameters:

signal (array_like or Recording) – Input IQ data as a complex CxN array or Recording, where C is the number of channels and N is the length of the IQ examples.
phase (float, optional) – The phase angle by which to rotate the IQ samples, in radians. Must be in the range [-π, π]. Default is π.

Raises:

ValueError – If the provided phase rotation is not in the range [-π, π].
ValueError – If signal is not CxN complex.

Returns:

A numpy array which represents the phase-shifted signal. If signal is a Recording, returns a Recording object with its data attribute containing the phase-shifted array.

Return type:

ndarray or Recording

>>> rec = Recording(data=[[1+1j, 2+2j, 3+3j, 4+4j]])
>>> new_rec = phase_shift(rec, numpy.pi/2)
>>> new_rec.data
array([[-1+1j, -2+2j, -3+3j, -4+4j]])

ria_toolkit_oss.transforms.iq_impairments.iq_imbalance(signal, amplitude_imbalance=1.5, phase_imbalance=3.141592653589793, dc_offset=1.5)[source]

Apply an IQ Imbalance to a signal.

Note

Based on MathWorks’ I/Q Imbalance.

Parameters:

signal (array_like or Recording) – Input IQ data as a complex CxN array or Recording, where C is the number of channels and N is the length of the IQ examples.
amplitude_imbalance (float, optional) – The IQ amplitude imbalance to apply, in dB. Default is 1.5.
phase_imbalance (float, optional) – The IQ phase imbalance to apply, in radians. Default is π. Must be in the range [-π, π].
dc_offset (float, optional) – The IQ DC offset to apply, in dB. Default is 1.5.

Raises:

ValueError – If the phase imbalance is not in the range [-π, π].
ValueError – If signal is not CxN complex.

Returns:

A numpy array which is the original signal with an applied IQ imbalance. If signal is a Recording, returns a Recording object with its data attribute containing the IQ imbalanced signal array.

Return type:

ndarray or Recording

>>> rec = Recording(data=[[2+18j, -34+2j, 3+9j]])
>>> new_rec = iq_imbalance(rec, 1, numpy.pi, 2)
>>> new_rec.data
array([[-38.38613587-4.78555031j, -4.26512621+81.35435535j, -19.19306793-7.17832547j]])

ria_toolkit_oss.transforms.iq_impairments.resample(signal, up=4, down=2)[source]

Resample a signal using polyphase filtering.

Uses scipy.signal.resample_poly to upsample the signal by the factor up, apply a zero-phase low-pass FIR filter, and downsample the signal by the factor down.

Parameters:

signal (array_like or Recording) – Input IQ data as a complex CxN array or Recording, where C is the number of channels and N is the length of the IQ examples.
up (int, optional) – The upsampling factor. Default is 4.
down (int, optional) – The downsampling factor. Default is 2.

Raises:

ValueError – If signal is not CxN complex.

Returns:

A numpy array which represents the resampled signal If signal is a Recording, returns a Recording object with its data attribute containing the resampled array.

Return type:

ndarray or Recording

>>> rec = Recording(data=[[1+1j, 2+2j]])
>>> new_rec = resample(rec, 2, 1)
>>> new_rec.data
array([[1.00051747+1.00051747j, 1.90020207+1.90020207j]])

IQ Augmentations

This submodule comprises the functionals of various transforms designed to create new training examples by augmenting existing examples or recordings using a variety of techniques These transforms take an ArrayLike object as input and return a corresponding numpy.ndarray with the impairment model applied; we call the latter the impaired data.

ria_toolkit_oss.transforms.iq_augmentations.generate_awgn(signal, snr=1)[source]

Generates additive white gaussian noise (AWGN) relative to the signal-to-noise ratio (SNR) of the provided signal array or Recording.

This function calculates the root mean squared (RMS) power of signal and then finds the RMS power of the noise which matches the specified SNR. Then, the AWGN is generated after calculating the variance and randomly calculating the amplitude and phase of the noise.

Parameters:

signal (array_like or Recording) – Input IQ data as a complex CxN array or Recording, where C is the number of channels and N is the length of the IQ examples.
snr (float, optional) – The signal-to-noise ratio in dB. Default is 1.

Raises:

ValueError – If signal is not CxN complex.

Returns:

A numpy array representing the generated noise which matches the SNR of signal. If signal is a Recording, returns a Recording object with its data attribute containing the generated noise array.

Return type:

ndarray or Recording

>>> rec = Recording(data=[[2 + 5j, 1 + 8j]])
>>> new_rec = generate_awgn(rec)
>>> new_rec.data
array([[2.15991777 + 0.69673915j, 0.2814541 - 0.12111976j]])

ria_toolkit_oss.transforms.iq_augmentations.time_reversal(signal)[source]

Reverses the order of the I (In-phase) and Q (Quadrature) data samples along the time axis of the provided signal array or Recording.

Parameters:: signal (array_like or Recording) – Input IQ data as a complex CxN array or Recording, where C is the number of channels and N is the length of the IQ examples.
Raises:: ValueError – If signal is not CxN complex.
Returns:: A numpy array containing the reversed I and Q data samples if signal is an array. If signal is a Recording, returns a Recording object with its data attribute containing the reversed array.
Return type:: ndarray or Recording

>>> rec = Recording(data=[[1+2j, 3+4j, 5+6j]])
>>> new_rec = time_reversal(rec)
>>> new_rec.data
array([[5+6j, 3+4j, 1+2j]])

ria_toolkit_oss.transforms.iq_augmentations.spectral_inversion(signal)[source]

Negates the imaginary components (Q, Quadrature) of the data samples contained within the provided signal array or Recording.

Parameters:: signal (array_like or Recording) – Input IQ data as a complex CxN array or Recording, where C is the number of channels and N is the length of the IQ examples.
Raises:: ValueError – If signal is not CxN complex.
Returns:: A numpy array containing the original I and negated Q data samples if signal is an array. If signal is a Recording, returns a Recording object with its data attribute containing the inverted array.
Return type:: ndarray or Recording

>>> rec = Recording(data=[[0+45j, 2-10j]])
>>> new_rec = spectral_inversion(rec)
>>> new_rec.data
array([[0-45j, 2+10j]])

ria_toolkit_oss.transforms.iq_augmentations.channel_swap(signal)[source]

Switches the I (In-phase) with the and Q (Quadrature) data samples for each sample within the provided signal array or Recording.

Parameters:: signal (array_like or Recording) – Input IQ data as a complex CxN array or Recording, where C is the number of channels and N is the length of the IQ examples.
Raises:: ValueError – If signal is not CxN complex.
Returns:: A numpy array containing the swapped I and Q data samples if signal is an array. If signal is a Recording, returns a Recording object with its data attribute containing the swapped array.
Return type:: ndarray or Recording

>>> rec = Recording(data=[[10+20j, 7+35j]])
>>> new_rec = channel_swap(rec)
>>> new_rec.data
array([[20+10j, 35+7j]])

ria_toolkit_oss.transforms.iq_augmentations.amplitude_reversal(signal)[source]

Negates the amplitudes of both the I (In-phase) and Q (Quadrature) data samples contained within the provided signal array or Recording.

Parameters:: signal (array_like or Recording) – Input IQ data as a complex CxN array or Recording, where C is the number of channels and N is the length of the IQ examples.
Raises:: ValueError – If signal is not CxN complex.
Returns:: A numpy array containing the negated I and Q data samples if signal is an array. If signal is a Recording, returns a Recording object with its data attribute containing the negated array.
Return type:: ndarray or Recording

>>> rec = Recording(data=[[4-3j, -5-2j, -9+1j]])
>>> new_rec = amplitude_reversal(rec)
>>> new_rec.data
array([[-4+3j, 5+2j, 9-1j]])

ria_toolkit_oss.transforms.iq_augmentations.drop_samples(signal, max_section_size=2, fill_type='zeros')[source]

Randomly drops IQ data samples contained within the provided signal array or Recording.

This function randomly selects sections of the signal and replaces the current data samples in the specified section with another value dependent on the fill type.

Parameters:

signal (array_like or Recording) – Input IQ data as a complex CxN array or Recording, where C is the number of channels and N is the length of the IQ examples.
max_section_size (int, optional) – Maximum allowable size of the section to be dropped and replaced. Default is 2.
fill_type (str, optional) –
Fill option used to replace dropped section of data (back-fill, front-fill, mean, zeros). Default is “zeros”.

”back-fill”: replace dropped section with the data sample occuring before the section.

”front-fill”: replace dropped section with the data sample occuring after the section.

”mean”: replace dropped section with mean of the entire signal.

”zeros”: replace dropped section with constant value of 0+0j.

Raises:

ValueError – If signal is not CxN complex.
ValueError – If max_section_size is less than 1 or greater than or equal to length of signal.

Returns:

A numpy array containing the I and Q data samples with replaced subsections if signal is an array. If signal is a Recording, returns a Recording object with its data attribute containing the array with dropped samples.

Return type:

ndarray or Recording

>>> rec = Recording(data=[[2+5j, 1+8j, 6+4j, 3+7j, 4+9j]])
>>> new_rec = drop_samples(rec)
>>> new_rec.data
array([[2+5j, 0, 0, 0, 4+9j]])

ria_toolkit_oss.transforms.iq_augmentations.quantize_tape(signal, bin_number=4, rounding_type='floor')[source]

Quantizes the IQ data of the provided signal array or Recording by a few bits.

This function emulates an analog-to-digital converter (ADC) which is commonly seen in digital RF systems. The relationship between the number of bins and number of bits is: log(# of bins) / log(2) = # of bits.

Parameters:

signal (array_like or Recording) – Input IQ data as a complex CxN array or Recording, where C is the number of channels and N is the length of the IQ examples.
bin_number (int, optional) – The number of bins the signal should be divided into. Default is 4.
rounding_type (str, optional) –
The type of rounding applied during processing. Default is “floor”.

”floor”: rounds down to the lower bound of the bin.

”ceiling”: rounds up to the upper bound of the bin.

Raises:

ValueError – If signal is not CxN complex.
UserWarning – If rounding_type is not “floor” or “ceiling”, “floor” is selected by default.

Returns:

A numpy array containing the quantized I and Q data samples if signal is an array. If signal is a Recording, returns a Recording object with its data attribute containing the quantized array.

Return type:

ndarray or Recording

>>> rec = Recording(data=[[1+1j, 4+4j, 1+2j, 1+4j]])
>>> new_rec = quantize_tape(rec)
>>> new_rec.data
array([[4+4j, 3+3j, 4+1j, 4+3j]])

ria_toolkit_oss.transforms.iq_augmentations.quantize_parts(signal, max_section_size=2, bin_number=4, rounding_type='floor')[source]

Quantizes random parts of the IQ data within the provided signal array or Recording by a few bits.

This function emulates an analog-to-digital converter (ADC) which is commonly seen in digital RF systems. The relationship between the number of bins and number of bits is: log(# of bins) / log(2) = # of bits.

Parameters:

signal (array_like or Recording) – Input IQ data as a complex CxN array or Recording, where C is the number of channels and N is the length of the IQ examples.
max_section_size (int, optional) – Maximum allowable size of the section to be quantized. Default is 2.
bin_number (int, optional) – The number of bins the signal should be divided into. Default is 4.
rounding_type (str, optional) –
Type of rounding applied during processing. Default is “floor”.

”floor”: rounds down to the lower bound of the bin.

”ceiling”: rounds up to the upper bound of the bin.

Raises:

ValueError – If signal is not CxN complex.
UserWarning – If rounding_type is not “floor” or “ceiling”, “floor” is selected by default.

Returns:

A numpy array containing the I and Q data samples with quantized subsections if signal is an array. If signal is a Recording, returns a Recording object with its data attribute containing the partially quantized array.

Return type:

ndarray or Recording

>>> rec = Recording(data=[[2+5j, 1+8j, 6+4j, 3+7j, 4+9j]])
>>> new_rec = quantize_parts(rec)
>>> new_rec.data
array([[2+5j, 1+8j, 3.66666667+3.66666667j, 3+7j, 4+9j]])

ria_toolkit_oss.transforms.iq_augmentations.magnitude_rescale(signal, starting_bounds=None, max_magnitude=1)[source]

Selects a random starting point from within the specified starting bounds and multiplies IQ data of the provided signal array or Recording by a random constant.

Parameters:

signal (array_like or Recording) – Input IQ data as a complex CxN array or Recording, where C is the number of channels and N is the length of the IQ examples.
starting_bounds (tuple, optional) – The bounds (inclusive) as indices in which the starting position of the rescaling occurs. Default is None, but if user does not assign any bounds, the bounds become (random index, N-1).
max_magnitude (int, optional) – The maximum value of the constant that is used to rescale the data. Default is 1.

Raises:

ValueError – If signal is not CxN complex.

Returns:

A numpy array containing the I and Q data samples with the rescaled magnitude after the random starting point if signal is an array. If signal is a Recording, returns a Recording object with its data attribute containing the rescaled array.

Return type:

ndarray or Recording

>>> rec = Recording(data=[[2+5j, 1+8j, 6+4j, 3+7j, 4+9j]])
>>> new_rec = magniute_rescale(rec)
>>> new_rec.data
array([[2+5j, 1+8j, 6+4j, 3+7j, 3.03181761+6.82158963j]])

ria_toolkit_oss.transforms.iq_augmentations.cut_out(signal, max_section_size=3, fill_type='ones')[source]

Cuts out random sections of IQ data and replaces them with either 0s, 1s, or low, average, or high sound-to-noise ratio (SNR) additive white gausssian noise (AWGN) within the provided signal array or Recording.

Parameters:

signal (array_like or Recording) – Input IQ data as a complex CxN array or Recording, where C is the number of channels and N is the length of the IQ examples.
max_section_size (int, optional) – Maximum allowable size of the section to be quantized. Default is 3.
fill_type (str, optional) –
Fill option used to replace cutout section of data (zeros, ones, low-snr, avg-snr-1, avg-snr-2). Default is “ones”.

”zeros”: replace cutout section with 0s.

”ones”: replace cutout section with 1s.

”low-snr”: replace cutout section with AWGN with an SNR of 0.5.

”avg-snr”: replace cutout section with AWGN with an SNR of 1.

”high-snr”: replace cutout section with AWGN with an SNR of 2.

Raises:

ValueError – If signal is not CxN complex.
UserWarning – If fill_type is not “zeros”, “ones”, “low-snr”, “avg-snr”, or “high-snr”, “ones” is selected by default.
ValueError – If max_section_size is less than 1 or greater than or equal to length of signal.

Returns:

A numpy array containing the I and Q data samples with random sections cut out and replaced according to fill_type if signal is an array. If signal is a Recording, returns a Recording object with its data attribute containing the cut out and replaced array.

Return type:

ndarray or Recording

>>> rec = Recording(data=[[2+5j, 1+8j, 6+4j, 3+7j, 4+9j]])
>>> new_rec = cut_out(rec)
>>> new_rec.data
array([[2+5j, 1+8j, 1+1j, 1+1j, 1+1j]])

ria_toolkit_oss.transforms.iq_augmentations.patch_shuffle(signal, max_patch_size=3)[source]

Selects random patches of the IQ data and randomly shuffles the data samples within the specified patch of the provided signal array or Recording.

Parameters:

signal (array_like or Recording) – Input IQ data as a complex CxN array or Recording, where C is the number of channels and N is the length of the IQ examples.
max_patch_size (int, optional) – Maximum allowable patch size of the data that can be shuffled. Default is 3.

Raises:

ValueError – If signal is not CxN complex.
ValueError – If max_patch_size is less than or equal to 1 or greater than length of signal.

Returns:

A numpy array containing the I and Q data samples with randomly shuffled regions if signal is an array. If signal is a Recording, returns a Recording object with its data attribute containing the shuffled array.

Return type:

ndarray or Recording

>>> rec = Recording(data=[[2+5j, 1+8j, 6+4j, 3+7j, 4+9j]])
>>> new_rec = patch_shuffle(rec)
>>> new_rec.data
array([[2+5j, 1+8j, 3+4j, 6+9j, 4+7j]])