"""Transforms on Signal objects."""
__all__ = [
"AWGN",
"AddSlope",
"AdditiveNoise",
"AdjacentChannelInterference",
"CarrierFrequencyDrift",
"CarrierPhaseNoise",
"CarrierPhaseOffset",
"ChannelSwap",
"ClockDrift",
"ClockJitter",
"CoarseGainChange",
"CochannelInterference",
"ComplexTo2D",
"CutOut",
"DigitalAGC",
"Doppler",
"Fading",
"IQImbalance",
"InterleaveComplex",
"IntermodulationProducts",
"NonlinearAmplifier",
"PassbandRipple",
"PatchShuffle",
"Quantize",
"RandomDropSamples",
"Shadowing",
"SignalTransform",
"SpectralInversion",
"Spectrogram",
"SpectrogramDropSamples",
"SpectrogramImage",
"Spurs",
"TimeReversal",
"TimeVaryingNoise",
]
from copy import copy
import numpy as np
import numpy.typing as npt
import torchsig.transforms.functional as F
from torchsig.signals.signal_types import Signal
from torchsig.transforms.base_transforms import Transform
from torchsig.utils.dsp import TorchSigComplexDataType, TorchSigRealDataType, low_pass
[docs]
class AWGN(SignalTransform):
"""Apply Additive White Gaussian Noise to signal.
This transform adds AWGN to the signal with specified power level.
Attributes:
noise_power_db: Noise AWGN power in dB (absolute).
precise: Measure and update SNR metadata. Default to False.
"""
[docs]
def __init__(self, noise_power_db: float, precise: bool = False, **kwargs):
"""Initialize the AWGN transform.
Args:
noise_power_db: Noise AWGN power in dB (absolute).
precise: Measure and update SNR metadata. Defaults to False.
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(data_dtype=TorchSigComplexDataType, **kwargs)
self.noise_power_db = noise_power_db
self.noise_power_linear = 10 ** (self.noise_power_db / 10)
self.precise = precise
def __apply__(self, signal: Signal) -> Signal:
"""Apply AWGN to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with AWGN added.
"""
if self.precise:
# update SNR for full sampled band, assuming independent noise
snr_linear = 10 ** (signal.snr_db / 10)
total_power = np.sum(np.abs(signal.data) ** 2) / len(signal.data)
sig_power = total_power / (1 + 1 / snr_linear)
noise_power = sig_power / snr_linear
new_snr = sig_power / (noise_power + self.noise_power_linear)
signal.snr_db = 10 * np.log10(new_snr)
signal.data = F.awgn(
signal.data, noise_power_db=self.noise_power_db, rng=self.random_generator
)
return signal
[docs]
class AddSlope(SignalTransform):
"""Add the slope of each sample with its preceding sample to itself.
Creates a weak 0 Hz IF notch filtering effect.
"""
[docs]
def __init__(self, **kwargs):
"""Initialize the AddSlope transform.
Args:
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
def __apply__(self, signal: Signal) -> Signal:
"""Apply slope addition to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with slope added.
"""
signal.data = F.add_slope(signal.data)
return signal
[docs]
class AdditiveNoise(SignalTransform):
"""Adds noise with specified properties to signal.
This transform adds noise with configurable power, color, and continuity to the signal.
Attributes:
power_range: Range bounds for interference power level (W). Defaults to (0.01, 10.0).
power_distribution: Random draw of interference power.
color: Noise color, supports 'white', 'pink', or 'red' noise frequency spectrum types. Defaults to 'white'.
continuous: Sets noise to continuous (True) or impulsive (False). Defaults to True.
precise: Measure and update SNR metadata. Default to False.
"""
[docs]
def __init__(
self,
power_range: tuple = (0.01, 10.0),
color: str = "white",
continuous: bool = True,
precise: bool = False,
**kwargs,
):
"""Initialize the AdditiveNoise transform.
Args:
power_range: Range bounds for interference power level (W). Defaults to (0.01, 10.0).
color: Noise color, supports 'white', 'pink', or 'red' noise frequency spectrum types. Defaults to 'white'.
continuous: Sets noise to continuous (True) or impulsive (False). Defaults to True.
precise: Measure and update SNR metadata. Defaults to False.
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(data_dtype=TorchSigComplexDataType, **kwargs)
self.power_range = power_range
self.power_distribution = self.get_distribution(self.power_range)
self.color = color
self.continuous = continuous
self.precise = precise
def __apply__(self, signal: Signal) -> Signal:
"""Apply additive noise to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with additive noise added.
"""
add_noise_power = self.power_distribution()
if self.precise:
# update SNR for full sampled band, assuming independent noise
snr_linear = 10 ** (signal.snr_db / 10)
total_power = np.sum(np.abs(signal.data) ** 2) / len(signal.data)
sig_power = total_power / (1 + 1 / snr_linear)
noise_power = sig_power / snr_linear
new_snr = sig_power / (noise_power + add_noise_power)
signal.snr_db = 10 * np.log10(new_snr)
signal.data = F.additive_noise(
data=signal.data,
power=add_noise_power,
color=self.color,
continuous=self.continuous,
rng=self.random_generator,
)
return signal
[docs]
class AdjacentChannelInterference(SignalTransform):
"""Apply adjacent channel interference to signal.
This transform adds interference from an adjacent channel with configurable parameters.
Attributes:
sample_rate: Sample rate (normalized). Defaults to 1.0.
power_range: Range bounds for interference power level (W). Defaults to (0.01, 10.0).
power_distribution: Random draw of interference power.
center_frequency_range: Range bounds for interference center frequency (normalized). Defaults to (0.2, 0.3).
center_frequency_distribution: Random draw of interference power.
phase_sigma_range: Range bounds for interference phase sigma. Defaults to (0.0, 1.0).
phase_sigma_distribution: Random draw of phase sigma.
time_sigma_range: Range bounds for interference time sigma. Defaults to (0.0, 10.0).
time_sigma_distribution: Random draw of time sigma.
filter_weights: Predefined baseband lowpass filter, fixed for all calls. Defaults to low_pass(0.125, 0.125, 1.0).
"""
[docs]
def __init__(
self,
sample_rate: float = 1.0,
power_range: tuple = (0.01, 10.0),
center_frequency_range: tuple = (0.2, 0.3),
phase_sigma_range: tuple = (0.0, 1.0),
time_sigma_range: tuple = (0.0, 10.0),
filter_weights: np.ndarray | None = None,
**kwargs,
):
"""Initialize the AdjacentChannelInterference transform.
Args:
sample_rate: Sample rate (normalized). Defaults to 1.0.
power_range: Range bounds for interference power level (W). Defaults to (0.01, 10.0).
center_frequency_range: Range bounds for interference center frequency (normalized). Defaults to (0.2, 0.3).
phase_sigma_range: Range bounds for interference phase sigma. Defaults to (0.0, 1.0).
time_sigma_range: Range bounds for interference time sigma. Defaults to (0.0, 10.0).
filter_weights: Predefined baseband lowpass filter, fixed for all calls. Defaults to low_pass(0.125, 0.125, 1.0).
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.sample_rate = sample_rate
self.power_range = power_range
self.power_distribution = self.get_distribution(self.power_range)
self.center_frequency_range = center_frequency_range
self.center_frequency_distribution = self.get_distribution(
self.center_frequency_range
)
self.phase_sigma_range = phase_sigma_range
self.phase_sigma_distribution = self.get_distribution(self.phase_sigma_range)
self.time_sigma_range = time_sigma_range
self.time_sigma_distribution = self.get_distribution(self.time_sigma_range)
self.filter_weights = low_pass(0.125, 0.125, 1.0) if filter_weights is None else filter_weights
def __apply__(self, signal: Signal) -> Signal:
"""Apply adjacent channel interference to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with adjacent channel interference added.
"""
signal.data = F.adjacent_channel_interference(
data=signal.data,
sample_rate=self.sample_rate,
power=self.power_distribution(),
center_frequency=self.center_frequency_distribution(),
phase_sigma=self.phase_sigma_distribution(),
time_sigma=self.time_sigma_distribution(),
filter_weights=self.filter_weights,
rng=self.random_generator,
)
return signal
[docs]
class CarrierFrequencyDrift(SignalTransform):
"""Apply carrier frequency drift to signal.
This transform simulates frequency drift in the carrier signal.
Attributes:
drift_ppm_range: Drift in parts per million (ppm). Default (0.1,1).
drift_ppm_distribution: Random draw from drift_ppm_range distribution.
"""
[docs]
def __init__(self, drift_ppm: tuple[float, float] = (0.1, 10), **kwargs):
"""Initialize the CarrierFrequencyDrift transform.
Args:
drift_ppm: Drift in parts per million (ppm). Default (0.1,10).
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.drift_ppm = drift_ppm
self.drift_ppm_distribution = self.get_distribution(self.drift_ppm, "log10")
def __apply__(self, signal: Signal) -> Signal:
"""Apply carrier frequency drift to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with carrier frequency drift applied.
"""
drift_ppm = self.drift_ppm_distribution()
signal.data = F.carrier_frequency_drift(
data=signal.data, drift_ppm=drift_ppm, rng=self.random_generator
)
return signal
[docs]
class CarrierPhaseNoise(SignalTransform):
"""Apply Carrier phase noise to signal.
This transform simulates phase noise in the carrier signal.
Attributes:
phase_noise_degrees: Range for phase noise (in degrees). Defaults to (0.25, 1).
phase_noise_degrees_distribution: Random draw from phase_noise_degrees distribution.
"""
[docs]
def __init__(self, phase_noise_degrees: tuple[float, float] = (0.25, 1), **kwargs):
"""Initialize the CarrierPhaseNoise transform.
Args:
phase_noise_degrees: Range for phase noise (in degrees). Defaults to (0.25, 1).
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.phase_noise_degrees = phase_noise_degrees
self.phase_noise_degrees_distribution = self.get_distribution(
self.phase_noise_degrees
)
def __apply__(self, signal: Signal) -> Signal:
"""Apply carrier phase noise to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with carrier phase noise applied.
"""
phase_noise_degrees = self.phase_noise_degrees_distribution()
signal.data = F.carrier_phase_noise(
data=signal.data,
phase_noise_degrees=phase_noise_degrees,
rng=self.random_generator,
)
return signal
[docs]
class CarrierPhaseOffset(SignalTransform):
"""Apply a randomized carrier phase offset to signal.
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.
Attributes:
phase_offset_range: Range bounds for phase offset (radians).
phase_offset_distribution: Random draw from phase offset distribution.
"""
[docs]
def __init__(
self, phase_offset_range: tuple[float, float] = (0, 2 * np.pi), **kwargs
):
"""Initialize the CarrierPhaseOffset transform.
Args:
phase_offset_range: Range bounds for phase offset (radians). Defaults to (0, 2 * np.pi).
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.phase_offset_range = phase_offset_range
self.phase_offset_distribution = self.get_distribution(self.phase_offset_range)
def __apply__(self, signal: Signal) -> Signal:
"""Apply carrier phase offset to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with carrier phase offset applied.
"""
phase_offset = self.phase_offset_distribution()
signal.data = F.phase_offset(signal.data, phase_offset)
return signal
[docs]
class ChannelSwap(SignalTransform):
"""Swaps the I and Q channels of complex input data."""
[docs]
def __init__(self, **kwargs):
"""Initialize the ChannelSwap transform.
Args:
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
def __apply__(self, signal: Signal) -> Signal:
"""Apply channel swap to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with I and Q channels swapped.
"""
signal.data = F.channel_swap(signal.data)
# swapping I/Q channels creates a frequency mirroring
# update metadata: signal
if hasattr(signal, "center_freq"):
signal["center_freq"] *= -1
# update metadata: signal_components
for component in signal.component_signals:
if hasattr(component, "center_freq"):
component["center_freq"] *= -1
return signal
[docs]
class ClockDrift(SignalTransform):
"""Simulates a clock drift effect, which applies a random error to the sampling rate."""
[docs]
def __init__(self, drift_ppm: tuple[float, float] = (1, 10), **kwargs):
"""Initialize the ClockDrift transform.
Args:
drift_ppm: Drift in parts per million (ppm). Default (1,10).
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.drift_ppm = drift_ppm
self.drift_ppm_distribution = self.get_distribution(self.drift_ppm, "log10")
def __apply__(self, signal: Signal) -> Signal:
"""Apply clock drift to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with clock drift applied.
"""
drift_ppm = self.drift_ppm_distribution()
signal.data = F.clock_drift(
data=signal.data, drift_ppm=drift_ppm, rng=self.random_generator
)
return signal
[docs]
class ClockJitter(SignalTransform):
"""Simulates a clock jitter effect, which applies a random error to the sampling phase."""
[docs]
def __init__(self, jitter_ppm: tuple[float, float] = (1, 10), **kwargs):
"""Initialize the ClockJitter transform.
Args:
jitter_ppm: Jitter in parts per million (ppm). Default (1,10).
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.jitter_ppm = jitter_ppm
self.jitter_ppm_distribution = self.get_distribution(self.jitter_ppm, "log10")
def __apply__(self, signal: Signal) -> Signal:
"""Apply clock jitter to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with clock jitter applied.
"""
jitter_ppm = self.jitter_ppm_distribution()
signal.data = F.clock_jitter(
data=signal.data, jitter_ppm=jitter_ppm, rng=self.random_generator
)
return signal
[docs]
class CoarseGainChange(SignalTransform):
"""Apply a randomized instantaneous jump in signal magnitude to model an abrupt receiver gain change.
Attributes:
gain_change_db_range: Sets the (min, max) gain change in dB.
gain_change_db_distribution: Random draw from gain_change_db distribution.
"""
[docs]
def __init__(self, gain_change_db: tuple[float, float] = (-20, 20), **kwargs):
"""Initialize the CoarseGainChange transform.
Args:
gain_change_db: Sets the (min, max) gain change in dB. Defaults to (-20, 20).
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.gain_change_db_distribution = self.get_distribution(gain_change_db)
def __apply__(self, signal: Signal) -> Signal:
"""Apply coarse gain change to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with coarse gain change applied.
"""
# select a gain value change from distribution
gain_change_db = self.gain_change_db_distribution()
# determine which samples gain change will be applied to. minimum index is 1, and maximum
# index is second to last sample, such that at minimum the gain will be applied to one
# sample or at maximum it will be applied to all less 1 samples. applying to zero samples
# or to all samples does not have a practical effect for this specific transform.
start_index = self.random_generator.integers(1, len(signal.data) - 1)
signal.data = F.coarse_gain_change(signal.data, gain_change_db, start_index)
return signal
[docs]
class CochannelInterference(SignalTransform):
"""Apply cochannel interference to signal.
This transform adds interference that shares the same channel as the signal.
Attributes:
power_range: Range bounds for interference power level (W). Default (0.01, 10.0).
power_distribution: Random draw of interference power.
filter_weights: Predefined baseband lowpass filter, fixed for all calls. Default low_pass(0.125, 0.125, 1.0).
noise_color: Base noise color, supports 'white', 'pink', or 'red' noise frequency spectrum types. Default 'white'.
continuous: Sets noise to continuous (True) or impulsive (False). Default True.
precise: Measure and update SNR metadata. Default to False.
"""
[docs]
def __init__(
self,
power_range: tuple = (0.01, 10.0),
filter_weights: np.ndarray | None = None,
color: str = "white",
continuous: bool = True,
**kwargs,
):
"""Initialize the CochannelInterference transform.
Args:
power_range: Range bounds for interference power level (W). Default (0.01, 10.0).
filter_weights: Predefined baseband lowpass filter, fixed for all calls. Default low_pass(0.125, 0.125, 1.0).
color: Base noise color, supports 'white', 'pink', or 'red' noise frequency spectrum types. Default 'white'.
continuous: Sets noise to continuous (True) or impulsive (False). Default True.
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(data_dtype=TorchSigComplexDataType, **kwargs)
self.power_range = power_range
self.power_distribution = self.get_distribution(self.power_range)
self.filter_weights = low_pass(0.125, 0.125, 1.0) if filter_weights is None else filter_weights
self.color = color
self.continuous = continuous
def __apply__(self, signal: Signal) -> Signal:
"""Apply cochannel interference to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with cochannel interference added.
"""
cochan_noise_power = self.power_distribution()
if self.precise:
snr_linear = 10 ** (signal.snr_db / 10)
total_power = np.sum(np.abs(signal.data) ** 2) / len(signal.data)
sig_power = total_power / (1 + 1 / snr_linear)
noise_power = sig_power / snr_linear
new_snr = sig_power / (noise_power + self.noise_power_linear)
signal["snr_db"] = 10 * np.log10(new_snr)
signal.data = F.cochannel_interference(
data=signal.data,
power=cochan_noise_power,
filter_weights=self.filter_weights,
color=self.color,
continuous=self.continuous,
rng=self.random_generator,
)
return signal
[docs]
class ComplexTo2D(SignalTransform):
"""Converts IQ data to two channels (real and imaginary parts)."""
[docs]
def __init__(self, **kwargs):
"""Initialize the ComplexTo2D transform.
Args:
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigRealDataType, **kwargs
)
def __apply__(self, signal: Signal) -> Signal:
"""Convert complex data to 2D real data.
Args:
signal: Signal to be transformed.
Returns:
Signal with complex data converted to 2D real data.
"""
signal.data = F.complex_to_2d(signal.data)
return signal
[docs]
class CutOut(SignalTransform):
"""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" <https://arxiv.org/pdf/1708.04552v2.pdf>`_.
Attributes:
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]).
duration_distribution: Random draw from duration distribution.
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.
cut_type_distribution: Random draw from cut_type distribution.
"""
[docs]
def __init__(
self,
duration=(0.01, 0.2),
cut_type: list[str] = (
["zeros", "ones", "low_noise", "avg_noise", "high_noise"]
),
**kwargs,
):
"""Initialize the CutOut transform.
Args:
duration: cut_dur sets the duration of the region to cut out. Defaults to (0.01, 0.2).
cut_type: cut_fill sets the type of data to fill in the cut region with. Defaults to ["zeros", "ones", "low_noise", "avg_noise", "high_noise"].
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(data_dtype=TorchSigComplexDataType, **kwargs)
self.duration = duration
self.cut_type = cut_type
self.duration_distribution = self.get_distribution(self.duration)
self.cut_type_distribution = self.get_distribution(self.cut_type)
def _determine_overlap(
self, metadata, cut_start: float, cut_duration: float
) -> str:
"""Determine the overlap between cut region and signal.
Args:
metadata: Signal metadata.
cut_start: Start of cut region.
cut_duration: Duration of cut region.
Returns:
Overlap type: "inside", "left", "right", "split", or "outside".
"""
signal_start = metadata.start
signal_stop = metadata.stop
cut_stop = cut_start + cut_duration
# inside
if signal_start > cut_start and signal_stop < cut_stop:
return "inside"
# left
if signal_start < cut_start and signal_stop < cut_stop:
return "left"
# right
if signal_start > cut_start and signal_stop > cut_stop:
return "right"
# split
if signal_start < cut_start and signal_stop > cut_stop:
return "split"
# only remaining type
return "outside"
def __apply__(self, signal: Signal) -> Signal:
"""Apply CutOut to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with CutOut applied.
"""
cut_duration = self.duration_distribution()
cut_type = self.cut_type_distribution()
cut_start = self.random_generator.uniform(low=0.0, high=1.0 - cut_duration)
signal.data = F.cut_out(signal.data, cut_start, cut_duration, cut_type)
# metadata
# CutOut can have complicated signal feature effects in practice.
# Any other desired metadata updates should be made manually.
# update start, duration
cut_stop = cut_start + cut_duration
overlap = self._determine_overlap(signal, cut_start, cut_duration)
if overlap == "left":
signal["stop"] = cut_start
elif overlap == "right":
signal["start"] = cut_stop
elif overlap == "split":
# left half = update current metadata
signal["stop"] = cut_start
signal["start"] = cut_stop
return signal
[docs]
class DigitalAGC(SignalTransform):
"""Automatic Gain Control performing sample-by-sample AGC algorithm.
Attributes:
initial_gain_db: Inital gain value in dB.
alpha_smooth: Alpha for averaging the measure signal level `level_n = level_n * alpha + level_n-1(1-alpha)`
alpha_track: Amount to adjust gain when in tracking state.
alpha_overflow: Amount to adjust gain when in overflow state `[level_db + gain_db] >= max_level`.
alpha_acquire: Amount to adjust gain when in acquire state.
track_range_db: dB range for operating in tracking state.
"""
[docs]
def __init__(
self,
initial_gain_db: tuple[float] = (0, 0),
alpha_smooth: tuple[float] = (1e-7, 1e-6),
alpha_track: tuple[float] = (1e-6, 1e-5),
alpha_overflow: tuple[float] = (1e-1, 3e-1),
alpha_acquire: tuple[float] = (1e-6, 1e-5),
track_range_db: tuple[float] = (0.5, 2),
**kwargs,
):
"""Initialize the DigitalAGC transform.
Args:
initial_gain_db: Inital gain value in dB. Defaults to (0, 0).
alpha_smooth: Alpha for averaging the measure signal level. Defaults to (1e-7, 1e-6).
alpha_track: Amount to adjust gain when in tracking state. Defaults to (1e-6, 1e-5).
alpha_overflow: Amount to adjust gain when in overflow state. Defaults to (1e-1, 3e-1).
alpha_acquire: Amount to adjust gain when in acquire state. Defaults to (1e-6, 1e-5).
track_range_db: dB range for operating in tracking state. Defaults to (0.5, 2).
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.initial_gain_db = initial_gain_db
self.alpha_smooth = alpha_smooth
self.alpha_track = alpha_track
self.alpha_overflow = alpha_overflow
self.alpha_acquire = alpha_acquire
self.track_range_db = track_range_db
self.initial_gain_db_distribution = self.get_distribution(self.initial_gain_db)
self.alpha_smooth_distribution = self.get_distribution(
self.alpha_smooth, "log10"
)
self.alpha_track_distribution = self.get_distribution(self.alpha_track, "log10")
self.alpha_overflow_distribution = self.get_distribution(
self.alpha_track, "log10"
)
self.alpha_acquire_distribution = self.get_distribution(
self.alpha_acquire, "log10"
)
self.track_range_db_distribution = self.get_distribution(self.track_range_db)
def __apply__(self, signal: Signal) -> Signal:
"""Apply DigitalAGC to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with DigitalAGC applied.
"""
initial_gain_db = self.initial_gain_db_distribution()
alpha_smooth = self.alpha_smooth_distribution()
alpha_track = self.alpha_track_distribution()
alpha_overflow = self.alpha_overflow_distribution()
alpha_acquire = self.alpha_acquire_distribution()
track_range_db = self.track_range_db_distribution()
# calculate derived parameters for AGC
# create a copy of the input data since it may need to be
# modified in order to avoid a log10(0)
receive_signal = copy(signal.data)
# get linear magnitude
receive_signal_mag = np.abs(receive_signal)
# find and replace all zeros
zero_sample_index = np.where(np.equal(receive_signal_mag, 0))[0]
# calculate all other values
non_zero_sample_index = np.setdiff1d(
np.arange(0, len(receive_signal)), zero_sample_index
)
# calculate the non-zero minimum
smallest_non_zero_value = np.min(receive_signal_mag[non_zero_sample_index])
# scale to get the "epsilon" to replace the zero values
epsilon = smallest_non_zero_value * 1e-6
# replace zero values
receive_signal[zero_sample_index] = epsilon
# determine average range for input in dB
receive_signal_db = np.log(np.abs(receive_signal))
receive_signal_mean_db = np.mean(receive_signal_db)
# calculate ranges for how to set AGC reference level.
# it is set (roughly) within range of data to provide
# a slight AGC effect
ref_level_max_db = receive_signal_mean_db + 5
ref_level_min_db = receive_signal_mean_db - 5
# randomly select the reference level the AGC will set
ref_level_db = self.random_generator.uniform(ref_level_min_db, ref_level_max_db)
# define the operating bounds of the AGC
low_level_db = ref_level_min_db - 10
high_level_db = ref_level_max_db + 10
signal.data = F.digital_agc(
np.ascontiguousarray(signal.data, dtype=np.complex64),
np.float64(initial_gain_db),
np.float64(alpha_smooth),
np.float64(alpha_track),
np.float64(alpha_overflow),
np.float64(alpha_acquire),
np.float64(ref_level_db),
np.float64(track_range_db),
np.float64(low_level_db),
np.float64(high_level_db),
)
return signal
[docs]
class Doppler(SignalTransform):
"""Apply a wideband Doppler effect to signal.
This transform simulates the Doppler effect caused by relative motion between transmitter and receiver.
Attributes:
velocity_range: Relative velocity bounds in m/s. Default (0.0, 10.0)
velocity_distribution: Random draw from velocity distribution.
propagation_speed: Wave speed in medium. Default 2.9979e8 m/s.
"""
[docs]
def __init__(
self,
velocity_range: tuple[float, float] = (0.0, 10.0),
propagation_speed: float = 2.9979e8,
**kwargs,
):
"""Initialize the Doppler transform.
Args:
velocity_range: Relative velocity bounds in m/s. Defaults to (0.0, 10.0).
propagation_speed: Wave speed in medium. Defaults to 2.9979e8 m/s.
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.velocity_range = velocity_range
self.velocity_distribution = self.get_distribution(self.velocity_range)
self.propagation_speed = propagation_speed
def __apply__(self, signal: Signal) -> Signal:
"""Apply Doppler effect to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with Doppler effect applied.
"""
velocity = self.velocity_distribution()
alpha = self.propagation_speed / (
self.propagation_speed - velocity
) # scaling factor
signal.data = F.doppler(
data=signal.data,
velocity=velocity,
propagation_speed=self.propagation_speed,
)
# update metadata: signal
if hasattr(signal, "center_freq"):
signal["center_freq"] *= alpha
if hasattr(signal, "bandwidth"):
signal["bandwidth"] *= alpha
# update metadata: component_signals
for component in signal.component_signals:
if hasattr(component, "center_freq"):
component["center_freq"] *= alpha
if hasattr(component, "bandwidth"):
component["bandwidth"] *= alpha
return signal
[docs]
class Fading(SignalTransform): # slow, fast, block fading
"""Apply a channel fading model to signal.
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.
Attributes:
coherence_bandwidth: Coherence bandwidth sampling parameters. Defaults to (0.01, 0.1).
coherence_bandwidth_distribution: Random draw from coherence bandwidth distribution.
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).
"""
[docs]
def __init__(
self,
coherence_bandwidth=(0.01, 0.1),
power_delay_profile: tuple | list | np.ndarray = (1, 1),
**kwargs,
):
"""Initialize the Fading transform.
Args:
coherence_bandwidth: Coherence bandwidth sampling parameters. Defaults to (0.01, 0.1).
power_delay_profile: A list of positive values assigning power to taps of the channel model. Defaults to (1, 1).
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.coherence_bandwidth = coherence_bandwidth
self.power_delay_profile = np.asarray(power_delay_profile)
self.coherence_bandwidth_distribution = self.get_distribution(
self.coherence_bandwidth
)
def __apply__(self, signal: Signal) -> Signal:
"""Apply fading to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with fading applied.
"""
coherence_bandwidth = self.coherence_bandwidth_distribution()
signal.data = F.fading(
data=signal.data,
coherence_bandwidth=coherence_bandwidth,
power_delay_profile=self.power_delay_profile,
rng=self.random_generator,
)
return signal
[docs]
class IntermodulationProducts(SignalTransform):
"""Apply simulated basebanded intermodulation products to a signal.
Attributes:
model_order: The choices model order, 3rd or 5th order. Defaults to [3,5].
coeffs_range: Range bounds for each intermodulation coefficient. Defaults to (0., 1.).
"""
[docs]
def __init__(
self,
model_order: list[int] = [3, 5],
coeffs_range: tuple[float, float] = (1e-4, 1e-1),
**kwargs,
):
"""Initialize the IntermodulationProducts transform.
Args:
model_order: The choices model order, 3rd or 5th order. Defaults to [3,5].
coeffs_range: Range bounds for each intermodulation coefficient. Defaults to (1e-4, 1e-1).
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.model_order = model_order
self.model_order_distribution = self.get_distribution(self.model_order)
self.coeffs_range = coeffs_range
self.coeffs_distribution = self.get_distribution(self.coeffs_range, "log10")
def __apply__(self, signal: Signal) -> Signal:
"""Apply intermodulation products to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with intermodulation products applied.
"""
# get randomized choice for model order
model_order = self.model_order_distribution()
# determine how many non-zero coefficients
num_coefficients = len(np.arange(0, model_order, 2))
# pre-allocate with all zeros
non_zero_coeffs = np.zeros(num_coefficients, dtype=TorchSigComplexDataType)
# randomize each coefficient
for index in range(num_coefficients):
if np.equal(index, 0):
non_zero_coeffs[index] = 1
else:
# calculate coefficient
non_zero_coeffs[index] = self.coeffs_distribution()
# run loop to ensure each coefficient must be smaller than the previous
while non_zero_coeffs[index] > non_zero_coeffs[index - 1]:
non_zero_coeffs[index] = self.coeffs_distribution()
# form the coeff array with appropriate zero-based weights
coeffs = np.zeros(model_order, dtype=TorchSigComplexDataType)
inner_index = 0
for outer_index in range(model_order):
if np.equal(np.mod(outer_index, 2), 0):
coeffs[outer_index] = non_zero_coeffs[inner_index]
inner_index += 1
signal.data = F.intermodulation_products(data=signal.data, coeffs=coeffs)
return signal
[docs]
class IQImbalance(SignalTransform):
"""Apply a set of I/Q imbalance effects to a signal: amplitude, phase, and DC offset.
Attributes:
amplitude_imbalance: Range bounds of IQ amplitude imbalance (dB).
amplitude_imbalance_distribution: Random draw from amplitude imbalance distribution.
phase_imbalance: Range bounds of IQ phase imbalance (radians).
phase_imbalance: Random draw from phase imbalance distribution.
dc_offset_db: Range bounds for DC offset in relative power.
dc_offset_db_distribution: Random draw from dc_offset_db distribution.
dc_offset_phase_rads: Range bounds for phase of DC offset
dc_offset_phase_rads_distribution: Random draw from dc_offset_phase_rads distribution.
"""
[docs]
def __init__(
self,
amplitude_imbalance=(-1.0, 1.0),
phase_imbalance=(-2.0 * np.pi / 180.0, 2.0 * np.pi / 180.0),
dc_offset_db=(0, 3),
dc_offset_rads=(0, 2 * np.pi),
**kwargs,
):
"""Initialize the IQImbalance transform.
Args:
amplitude_imbalance: Range bounds of IQ amplitude imbalance (dB). Defaults to (-1.0, 1.0).
phase_imbalance: Range bounds of IQ phase imbalance (radians). Defaults to (-2.0 * np.pi / 180.0, 2.0 * np.pi / 180.0).
dc_offset_db: Range bounds for DC offset in relative power. Defaults to (0, 3).
dc_offset_rads: Range bounds for phase of DC offset. Defaults to (0, 2 * np.pi).
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.amplitude_imbalance = amplitude_imbalance
self.phase_imbalance = phase_imbalance
self.dc_offset_db = dc_offset_db
self.dc_offset_rads = dc_offset_rads
self.amplitude_imbalance_distribution = self.get_distribution(
self.amplitude_imbalance
)
self.phase_imbalance_distribution = self.get_distribution(self.phase_imbalance)
self.dc_offset_db_distribution = self.get_distribution(self.dc_offset_db)
self.dc_offset_phase_rads_distribution = self.get_distribution(
self.dc_offset_rads
)
def __apply__(self, signal: Signal) -> Signal:
"""Apply IQ imbalance to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with IQ imbalance applied.
"""
amplitude_imbalance = self.amplitude_imbalance_distribution()
phase_imbalance = self.phase_imbalance_distribution()
dc_offset_db = self.dc_offset_db_distribution()
dc_offset_rads = self.dc_offset_phase_rads_distribution()
signal.data = F.iq_imbalance(
signal.data,
amplitude_imbalance,
phase_imbalance,
dc_offset_db,
dc_offset_rads,
)
return signal
[docs]
class InterleaveComplex(SignalTransform):
"""Transforms a complex-valued array into a real-valued array of interleaved IQ values."""
[docs]
def __init__(self, **kwargs):
"""Initialize the InterleaveComplex transform.
Args:
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigRealDataType, **kwargs
)
def __apply__(self, signal: Signal) -> Signal:
"""Interleave complex data into real data.
Args:
signal: Signal to be transformed.
Returns:
Signal with complex data interleaved into real data.
"""
signal.data = F.interleave_complex(signal.data)
return signal
[docs]
class NonlinearAmplifier(SignalTransform):
"""Apply a memoryless nonlinear amplifier model to a signal.
Attributes:
gain_range: Small-signal gain range (linear). Defaults to (1.0, 4.0).
gain_distribution: Random draw from gain distribution.
psat_backoff_range: Psat backoff factor (linear) reflecting saturated
power level (Psat) relative to input signal mean power. Defaults to (5.0, 20.0).
psat_backoff_distribution: Random draw from psat_backoff distribution.
phi_max_range: Maximum signal relative phase shift at
saturation power level (radians). Defaults to (-0.05, 0.05).
phi_max_distribution: Random draw from phi_max distribution.
phi_slope_range: Slope of relative phase shift response
(W/radians). Defaults to (-0.1, 0.01).
phi_slope_distribution: Random draw from phi_max distribution.
auto_scale: Automatically rescale output power to match full-scale peak
input power prior to transform, based on peak estimates. Default True.
"""
[docs]
def __init__(
self,
gain_range: tuple[float, float] = (1.0, 1.0),
psat_backoff_range: tuple[float, float] = (5.0, 20.0),
phi_max_range: tuple[float, float] = (-0.05, 0.05),
phi_slope_range: tuple[float, float] = (-0.1, 0.1),
auto_scale: bool = True,
**kwargs,
):
"""Initialize the NonlinearAmplifier transform.
Args:
gain_range: Small-signal gain range (linear). Defaults to (1.0, 1.0).
psat_backoff_range: Psat backoff factor (linear) reflecting saturated power level (Psat) relative to input signal mean power. Defaults to (5.0, 20.0).
phi_max_range: Maximum signal relative phase shift at saturation power level (radians). Defaults to (-0.05, 0.05).
phi_slope_range: Slope of relative phase shift response (W/radians). Defaults to (-0.1, 0.1).
auto_scale: Automatically rescale output power to match full-scale peak input power prior to transform, based on peak estimates. Defaults to True.
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.gain_range = gain_range
self.gain_distribution = self.get_distribution(self.gain_range)
self.psat_backoff_range = psat_backoff_range
self.psat_backoff_distribution = self.get_distribution(self.psat_backoff_range)
self.phi_max_range = phi_max_range
self.phi_max_distribution = self.get_distribution(self.phi_max_range)
self.phi_slope_range = phi_slope_range
self.phi_slope_distribution = self.get_distribution(self.phi_slope_range)
self.auto_scale = auto_scale
def __apply__(self, signal: Signal) -> Signal:
"""Apply nonlinear amplifier to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with nonlinear amplifier applied.
"""
gain = self.gain_distribution()
psat_backoff = self.psat_backoff_distribution()
phi_max = self.phi_max_distribution()
phi_slope = self.phi_slope_distribution()
signal.data = F.nonlinear_amplifier(
data=signal.data,
gain=gain,
psat_backoff=psat_backoff,
phi_max=phi_max,
phi_slope=phi_slope,
auto_scale=self.auto_scale,
)
return signal
[docs]
class PassbandRipple(SignalTransform):
"""Models analog filter passband ripple response for a signal.
Attributes:
max_ripple_db: Range for maximum allowable ripple to simulate. Defaults to (1,2).
num_taps: List of number of taps in simulated filter. Defaults to [2,3].
coefficient_decay_rate: Range for the rate at which the simulated
impulse response goes to zero. Defaults to (1, 5).
"""
[docs]
def __init__(
self,
max_ripple_db: tuple[float] = (1, 2),
num_taps: list[int] = [2, 3],
coefficient_decay_rate: tuple[float] = (1, 5),
**kwargs,
):
"""Initialize the PassbandRipple transform.
Args:
max_ripple_db: Range for maximum allowable ripple to simulate. Defaults to (1, 2).
num_taps: List of number of taps in simulated filter. Defaults to [2, 3].
coefficient_decay_rate: Range for the rate at which the simulated impulse response goes to zero. Defaults to (1, 5).
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.max_ripple_db = max_ripple_db
self.max_ripple_db_distribution = self.get_distribution(self.max_ripple_db)
self.num_taps = num_taps
self.num_taps_distribution = self.get_distribution(self.num_taps)
self.coefficient_decay_rate = coefficient_decay_rate
self.coefficient_decay_rate_distribution = self.get_distribution(
coefficient_decay_rate
)
def __apply__(self, signal: Signal) -> Signal:
"""Apply passband ripple to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with passband ripple applied.
"""
max_ripple_db = self.max_ripple_db_distribution()
num_taps = int(np.round(self.num_taps_distribution()))
coefficient_decay_rate = self.coefficient_decay_rate_distribution()
signal.data = F.passband_ripple(
data=signal.data,
num_taps=num_taps,
max_ripple_db=max_ripple_db,
coefficient_decay_rate=coefficient_decay_rate,
rng=self.random_generator,
)
return signal
[docs]
class PatchShuffle(SignalTransform):
"""Randomly shuffle multiple local regions of samples.
Transform is loosely based on
`"PatchShuffle Regularization" <https://arxiv.org/pdf/1707.07103.pdf>`_.
Attributes:
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]).
patch_size_distribution: Random draw from patch_size distribution.
shuffle_ratio: 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: Random draw from shuffle_ratio distribution.
"""
[docs]
def __init__(
self, patch_size=(3, 10), shuffle_ratio=(0.01, 0.05), **kwargs
) -> None:
"""Initialize the PatchShuffle transform.
Args:
patch_size: patch_size sets the size of each patch to shuffle. Defaults to (3, 10).
shuffle_ratio: shuffle_ratio sets the ratio of the patches to shuffle. Defaults to (0.01, 0.05).
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.patch_size = patch_size
self.shuffle_ratio = shuffle_ratio
self.patch_size_distribution = self.get_distribution(self.patch_size)
self.shuffle_ratio_distribution = self.get_distribution(self.shuffle_ratio)
def __apply__(self, signal: Signal) -> Signal:
"""Apply patch shuffle to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with patch shuffle applied.
"""
patch_size = self.patch_size_distribution()
shuffle_ratio = self.shuffle_ratio_distribution()
num_patches = int(signal.data.shape[0] / patch_size)
num_to_shuffle = int(num_patches * shuffle_ratio)
patches_to_shuffle = self.random_generator.choice(
num_patches,
replace=False,
size=num_to_shuffle,
)
signal.data = F.patch_shuffle(
signal.data, patch_size, patches_to_shuffle, self.random_generator
)
# PatchShuffle can have complicated signal feature effects in practice.
# Any desired metadata updates should be made manually.
return signal
[docs]
class Quantize(SignalTransform):
"""Quantize signal I/Q samples into specified levels with a rounding method.
Attributes:
num_levels: Number of quantization levels.
num_levels_distribution: Random draw from num_levels distribution.
rounding_mode: Quantization rounding method. Must be 'floor' or 'ceiling'.
rounding_mode_distribution: Random draw from rounding_mode distribution.
"""
[docs]
def __init__(
self,
num_bits: tuple[int, int] = (6, 18),
ref_level_adjustment_db: tuple[float, float] = (-10, 3),
rounding_mode: list[str] = ["floor", "ceiling"],
**kwargs,
):
"""Initialize the Quantize transform.
Args:
num_bits: Number of quantization bits. Defaults to (6, 18).
ref_level_adjustment_db: Reference level adjustment in dB. Defaults to (-10, 3).
rounding_mode: Quantization rounding method. Must be 'floor' or 'ceiling'. Defaults to ["floor", "ceiling"].
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.num_bits = num_bits
self.num_bits_distribution = self.get_distribution(self.num_bits)
self.ref_level_adjustment_db = ref_level_adjustment_db
self.ref_level_adjustment_db_distribution = self.get_distribution(
self.ref_level_adjustment_db
)
self.rounding_mode = rounding_mode
self.rounding_mode_distribution = self.get_distribution(self.rounding_mode)
def __apply__(self, signal: Signal) -> Signal:
"""Apply quantization to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with quantization applied.
"""
num_bits = int(np.round(self.num_bits_distribution()))
ref_level_adjustment_db = self.ref_level_adjustment_db_distribution()
rounding_mode = self.rounding_mode_distribution()
# apply quantization
signal.data = F.quantize(
data=signal.data,
num_bits=num_bits,
ref_level_adjustment_db=ref_level_adjustment_db,
rounding_mode=rounding_mode,
)
return signal
[docs]
class RandomDropSamples(SignalTransform):
"""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 <https://github.com/arundo/tsaug/blob/master/src/tsaug/_augmenter/dropout.py>`_.
Attributes:
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]).
drop_rate_distribution: Random draw from drop_rate distribution.
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]).
size_distribution: Random draw from size distribution.
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.
fill_distribution: Random draw from fill distribution.
"""
[docs]
def __init__(
self,
drop_rate=(0.01, 0.05),
size=(1, 10),
fill: list[str] = (["ffill", "bfill", "mean", "zero"]),
**kwargs,
) -> None:
"""Initialize the RandomDropSamples transform.
Args:
drop_rate: drop_rate sets the rate at which to drop samples. Defaults to (0.01, 0.05).
size: size sets the size of each instance of dropped samples. Defaults to (1, 10).
fill: fill sets the method of how the dropped samples should be filled. Defaults to ["ffill", "bfill", "mean", "zero"].
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.drop_rate = drop_rate
self.size = size
self.fill = fill
self.drop_rate_distribution = self.get_distribution(self.drop_rate)
self.size_distribution = self.get_distribution(self.size)
self.fill_distribution = self.get_distribution(self.fill)
def __apply__(self, signal: Signal) -> Signal:
"""Apply random drop samples to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with random drop samples applied.
"""
drop_rate = self.drop_rate_distribution()
fill = self.fill_distribution()
drop_instances = int(signal.data.shape[0] * drop_rate)
if drop_instances < 1:
return signal # drop no samples and return the input signal if we have randomly selected to drop zero samples
drop_sizes = self.size_distribution(size=drop_instances).astype(int)
drop_starts = self.random_generator.uniform(
1, signal.data.shape[0] - max(drop_sizes) - 1, drop_instances
).astype(int)
signal.data = F.drop_samples(signal.data, drop_starts, drop_sizes, fill)
return signal
[docs]
class Shadowing(SignalTransform):
"""Apply channel shadowing effect across entire signal.
This transform models RF shadowing effects by applying lognormal fading to the input data.
Attributes:
mean_db_range: Mean value range in dB. Defaults to (0.0, 4.0).
mean_db_distribution: Random draw from mean_db distribution.
sigma_db_range: Sigma value range in dB. Defaults to (2.0, 6.0).
sigma_db_distribution: Random draw from sigma_db distribution.
"""
[docs]
def __init__(
self,
mean_db_range: tuple[float, float] = (0.0, 4.0),
sigma_db_range: tuple[float, float] = (2.0, 6.0),
**kwargs,
):
"""Initialize the Shadowing transform.
Args:
mean_db_range: Mean value range in dB. Defaults to (0.0, 4.0).
sigma_db_range: Sigma value range in dB. Defaults to (2.0, 6.0).
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.mean_db_range = mean_db_range
self.mean_db_distribution = self.get_distribution(self.mean_db_range)
self.sigma_db_range = sigma_db_range
self.sigma_db_distribution = self.get_distribution(self.sigma_db_range)
def __apply__(self, signal: Signal) -> Signal:
"""Apply shadowing to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with shadowing applied.
"""
mean_db = self.mean_db_distribution()
sigma_db = self.sigma_db_distribution()
signal.data = F.shadowing(
data=signal.data,
mean_db=mean_db,
sigma_db=sigma_db,
rng=self.random_generator,
)
return signal
[docs]
class SpectralInversion(SignalTransform):
"""Inverts spectrum of complex signal data.
This transform performs spectral inversion by complex conjugation of the input data.
"""
[docs]
def __init__(self, **kwargs):
"""Initialize the SpectralInversion transform.
Args:
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
def __apply__(self, signal: Signal) -> Signal:
"""Apply spectral inversion to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with spectral inversion applied.
"""
signal.data = F.spectral_inversion(signal.data)
# update metadata: signal
if hasattr(signal, "center_freq"):
signal["center_freq"] *= -1
# update metadata: signal_components
for component in signal.component_signals:
if hasattr(component, "center_freq"):
component["center_freq"] *= -1
return signal
[docs]
class Spectrogram(SignalTransform):
"""Computes the spectogram of I/Q data.
This transform computes the spectrogram by applying the Short-Time Fourier Transform (STFT)
to the input IQ data.
Attributes:
fft_size: The FFT size (number of bins) in the spectrogram.
"""
[docs]
def __init__(self, fft_size: int, **kwargs):
"""Initialize the Spectrogram transform.
Args:
fft_size: The FFT size (number of bins) in the spectrogram.
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigRealDataType, **kwargs
)
self.fft_size = fft_size
# fft_stride is the number of data points to move or "hop" over when computing the next FF
self.fft_stride = copy(fft_size)
def __apply__(self, signal: Signal) -> Signal:
"""Apply spectrogram computation to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with spectrogram computed.
"""
signal.data = F.spectrogram(
signal.data,
self.fft_size,
self.fft_stride,
)
return signal
[docs]
class SpectrogramDropSamples(SignalTransform):
"""Randomly drop samples from the input data of specified durations and
with specified fill techniques.
Supported 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 <https://github.com/arundo/tsaug/blob/master/src/tsaug/_augmenter/dropout.py>`_.
Attributes:
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]).
drop_rate_distribution: Random draw from drop_rate distribution.
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]).
size_distribution: Random draw from size distribution.
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.
fill_distribution: Random draw from fill distribution.
"""
[docs]
def __init__(
self,
drop_rate=(0.001, 0.005),
size=(1, 10),
fill: list[str] = (
["ffill", "bfill", "mean", "zero", "low", "min", "max", "ones"]
),
**kwargs,
) -> None:
"""Initialize the SpectrogramDropSamples transform.
Args:
drop_rate: drop_rate sets the rate at which to drop samples. Defaults to (0.001, 0.005).
size: size sets the size of each instance of dropped samples. Defaults to (1, 10).
fill: fill sets the method of how the dropped samples should be filled. Defaults to ["ffill", "bfill", "mean", "zero", "low", "min", "max", "ones"].
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigRealDataType, **kwargs
)
self.drop_rate = drop_rate
self.size = size
self.fill = fill
self.drop_rate_distribution = self.get_distribution(self.drop_rate)
self.size_distribution = self.get_distribution(self.size)
self.fill_distribution = self.get_distribution(self.fill)
def __apply__(self, signal: Signal) -> Signal:
"""Apply spectrogram drop samples to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with spectrogram drop samples applied.
"""
drop_rate = self.drop_rate_distribution()
fill = self.fill_distribution()
drop_instances = int(signal.data.shape[0] * drop_rate)
drop_sizes = self.size_distribution(drop_instances).astype(int)
if drop_instances < 1:
return signal # if drop sizes is empty, just return signal
if len(drop_sizes) > 0:
drop_starts = self.random_generator.uniform(
0, signal.data.shape[0] - max(drop_sizes), drop_instances
).astype(int)
signal.data = F.spectrogram_drop_samples(
signal.data,
drop_starts,
drop_sizes,
fill,
)
# SpectrogramDropSamples can have complicated signal feature effects in practice.
# Any desired metadata updates should be made manually.
return signal
[docs]
class SpectrogramImage(SignalTransform):
"""Transforms signal to a spectrogram image.
This transform computes the spectrogram and converts it to a grayscale image.
"""
[docs]
def __init__(self, fft_size: int, black_hot: bool = True, **kwargs) -> None:
"""Initialize the SpectrogramImage transform.
Args:
fft_size: The FFT size (number of bins) in the spectrogram.
black_hot: Toggles black hot spectrogram. Defaults to True (black hot).
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigRealDataType, **kwargs
)
self.fft_size = fft_size
self.fft_stride = fft_size # note: size = stride
self.black_hot = black_hot
def __apply__(self, signal: Signal) -> Signal:
"""Apply spectrogram image transformation to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with spectrogram image computed.
"""
signal.data = F.spectrogram_image(
data=signal.data,
fft_size=self.fft_size,
fft_stride=self.fft_stride,
black_hot=self.black_hot,
)
return signal
[docs]
class TimeReversal(SignalTransform):
"""Apply 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.
Attributes:
allow_spectral_inversion: Whether to allow spectral inversion as a time reversal side effect (True) or not (False). Defaults to True.
* If bool, applied to all signals.
* If float, applied as probability to add signals.
"""
[docs]
def __init__(self, allow_spectral_inversion: bool | float = True, **kwargs) -> None:
"""Initialize the TimeReversal transform.
Args:
allow_spectral_inversion: Whether to allow spectral inversion as a time reversal side effect. Defaults to True.
**kwargs: Additional keyword arguments passed to the parent class.
Raises:
TypeError: If allow_spectral_inversion is not bool or float.
"""
if isinstance(allow_spectral_inversion, bool):
self.allow_spectral_inversion = 1.0 if allow_spectral_inversion else 0.0
elif isinstance(allow_spectral_inversion, float):
self.allow_spectral_inversion = allow_spectral_inversion
else:
raise TypeError(
f"Invalid type for allow_spectral_inversion {type(allow_spectral_inversion)}. Must be bool or float."
)
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
def __apply__(self, signal: Signal) -> Signal:
"""Apply time reversal to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with time reversal applied.
"""
signal.data = F.time_reversal(signal.data)
do_si = self.random_generator.random() > self.allow_spectral_inversion
if do_si:
signal.data = F.spectral_inversion(signal.data)
num_data_samples = len(signal.data)
# update metadata: signal
if hasattr(signal, "stop_in_samples"):
original_stop = signal.stop_in_samples
signal["start_in_samples"] = num_data_samples - original_stop
if hasattr(signal, "center_freq") and not do_si:
signal["center_freq"] *= -1
# update metadata: component_signals
for component in signal.component_signals:
if hasattr(component, "stop_in_samples"):
original_stop = component.stop_in_samples
component["start_in_samples"] = num_data_samples - original_stop
if hasattr(component, "center_freq") and not do_si:
component["center_freq"] *= -1
return signal
[docs]
class TimeVaryingNoise(SignalTransform):
"""Add time-varying noise to signal regions.
This transform adds noise with power levels that vary over time, with specified
minimum and maximum power levels and number of inflection points.
Attributes:
noise_power_low: Range bounds for minimum noise power in dB.
noise_power_low_distribution: Random draw from noise_power_low distribution.
noise_power_high: Range bounds for maximum noise power in dB.
noise_power_high_distribution: Random draw from noise_power_high distribution.
inflections: Number of inflection points over IQ data.
inflections_distribution: Random draw from inflections distribution.
random_regions: Inflections points spread randomly (True) or evenly (False).
random_regions_distribution: Random draw from random_regions distribution.
"""
[docs]
def __init__(
self,
noise_power_low=(-80.0, -60.0),
noise_power_high=(-40.0, -20.0),
inflections=[0, 10],
random_regions: list | bool = True,
**kwargs,
):
"""Initialize the TimeVaryingNoise transform.
Args:
noise_power_low: Range bounds for minimum noise power in dB. Defaults to (-80.0, -60.0).
noise_power_high: Range bounds for maximum noise power in dB. Defaults to (-40.0, -20.0).
inflections: Number of inflection points over IQ data. Defaults to [0, 10].
random_regions: Inflections points spread randomly (True) or evenly (False). Defaults to True.
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.noise_power_low = noise_power_low
self.noise_power_high = noise_power_high
self.inflections = inflections
self.random_regions = random_regions
self.noise_power_low_distribution = self.get_distribution(self.noise_power_low)
self.noise_power_high_distribution = self.get_distribution(
self.noise_power_high
)
self.inflections_distribution = self.get_distribution(self.inflections)
self.random_regions_distribution = self.get_distribution(self.random_regions)
def __apply__(self, signal: Signal) -> Signal:
"""Apply time-varying noise to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with time-varying noise added.
"""
noise_power_low = self.noise_power_low_distribution()
noise_power_high = self.noise_power_high_distribution()
inflections = self.inflections_distribution()
random_regions = self.random_regions_distribution
signal.data = F.time_varying_noise(
signal.data,
noise_power_low,
noise_power_high,
inflections,
random_regions,
rng=self.random_generator,
)
return signal
[docs]
class Spurs(SignalTransform):
"""Simulates spurs by adding tones into the receive signal.
This transform adds spurious signals (tones) at specified frequencies with specified power levels.
Attributes:
num_spurs: The range of numbers of spurs to add. Defaults to (1,4).
relative_power_db: The range of relative power for the spurs. The power is relative to the noise floor. Defaults to (5,15).
"""
[docs]
def __init__(
self,
num_spurs: tuple[int] = (1, 4),
relative_power_db: tuple[float] = (0, 30),
**kwargs,
):
"""Initialize the Spurs transform.
Args:
num_spurs: The range of numbers of spurs to add. Defaults to (1, 4).
relative_power_db: The range of relative power for the spurs. Defaults to (0, 30).
**kwargs: Additional keyword arguments passed to the parent class.
"""
super().__init__(
required_metadata=[], data_dtype=TorchSigComplexDataType, **kwargs
)
self.num_spurs = num_spurs
self.num_spurs_distribution = self.get_distribution(self.num_spurs)
self.relative_power_db = relative_power_db
self.relative_power_db_distribution = self.get_distribution(
self.relative_power_db
)
def __apply__(self, signal: Signal) -> Signal:
"""Apply spurs to the signal.
Args:
signal: Signal to be transformed.
Returns:
Signal with spurs added.
"""
num_spurs = int(np.round(self.num_spurs_distribution()))
sample_rate = 1
# noise floor power value: provided or estimate in function
try:
noise_power_db = signal["noise_power_db"]
except:
noise_power_db = None
# randomize the parameters for each spur
relative_power_db = []
center_freqs = []
for _ in range(num_spurs):
# randomize the relative power in dB
relative_power_db.append(self.relative_power_db_distribution())
# determine the corresponding center frequency
low_freq = -sample_rate / 2
high_freq = sample_rate / 2
center_freqs.append(self.random_generator.uniform(low_freq, high_freq))
# apply spurs
signal.data = F.spurs(
data=signal.data,
sample_rate=sample_rate,
center_freqs=center_freqs,
relative_power_db=relative_power_db,
noise_power_db=noise_power_db,
)
return signal