Processors

class Upsampler(L: int, phase: int = 0, scale: float = 1.0, is_mimo: bool = True, axis: int = -1, use_filter: bool = False, name: str = 'upsampler')

Upsampler class for increasing the sampling rate of a signal along a specified axis.

This class increases the sample rate of the incoming signal by inserting \(L – 1\) zeros between samples along the specified axis.

Signal Model

\[\begin{split}y[n] = \begin{cases} \alpha x[(n-\tau)/ L] & \text{if } (n-\tau)\% L = 0 \\\\ 0 & \text{otherwise} \end{cases}\end{split}\]

Attributes

Lint

The upsampling factor.

phaseint, optional

The number of samples \(\tau \in \mathbb{N}\) by which to offset the upsampled sequence (default: 0).

scalefloat, optional

The scaling factor \(\alpha\) applied to the upsampled signal (default: 1.0).

axisint, optional

The axis along which to perform the upsampling operation (default: -1).

use_filter: bool, optional

Apply a lowpass filter at the output (default: false)

namestr, optional

The name of the processor (default: “upsampler”).

Example 1

>>> import numpy as np
>>> X = np.array([1, 2, 3])
>>> upsampler = Upsampler(L=2)
>>> Y = upsampler(X)
>>> print(Y)
[1. 0. 2. 0. 3. 0.]

Example 2

>>> X = np.array([1, 2])
>>> upsampler = Upsampler(L=3, phase=1)
>>> Y = upsampler(X)
>>> print(Y)
[0. 1. 0. 0. 2. 0.]

Example 3

>>> X = np.array([[1, 2], [3, 4]])
>>> print(X[0, :])
[1 2]
>>> upsampler = Upsampler(L=2, axis=-1)
>>> Y = upsampler(X)
>>> print(Y)
[[1. 0. 2. 0.]
[3. 0. 4. 0.]]

class Downsampler(L: int, phase: int = 0, scale: float = 1.0, is_mimo: bool = True, axis: int = -1, use_filter: bool = False, name: str = 'downsampler')

Downsampler class for decreasing the sampling rate of a signal along a specified axis.

This class decreases the sample rate of the input signal by keeping the first sample and then every Lth sample after the first along the specified axis.

Signal Model

The decimation process can be described mathematically as follows:

\[y[n] = x[n \cdot L]\]

where \(L\) is the downsampling factor, \(x\) is the input signal, and \(y\) is the output signal.

Attributes

Lint

The downsampling factor, which determines how many samples are skipped between each retained sample.

phaseint, optional

The number of samples by which to offset the downsampled sequence (default: 0).

scalefloat, optional

The scaling factor applied to the downsampled signal (default: 1.0).

axisint, optional

The axis along which to perform the downsampling operation (default: -1).

use_filter: bool, optional

Apply a lowpass filter at the output (default: false)

namestr, optional

The name of the processor (default: “downsampler”).

Examples

>>> import numpy as np
>>> X = np.array([1, 2, 3, 4, 5, 6])
>>> downsampler = Downsampler(L=2)
>>> Y = downsampler(X)
>>> print(Y)
[1. 3. 5.]

class Serial2Parallel(N_sub: int, order: str = 'F', method: Literal['zero-padding', 'truncate'] = 'zero-padding', name: str = 'S2P')

A class for converting a serial data stream into parallel data streams.

This class reshapes a multi-dimensional array (serial stream) into another multi-dimensional array (parallel data streams) using a specified number of subcarriers and an ordering scheme. If necessary, the input data is either padded with zeros to fit the specified structure or is truncated.

Attributes

N_subint

The number of subcarriers, which defines the size of the last dimension in the reshaped array. Must be a positive integer.

orderstr, optional

The order in which to reshape the array. ‘F’ means to reshape in column-major (Fortran-style) order, and ‘C’ means to reshape in row-major (C-style) order. Default is ‘F’.

methodLiteral[“zero-padding”, “truncate”], optional

The method to handle data that does not fit perfectly into the reshaped structure. Options are ‘zero-padding’ to pad with zeros or ‘truncate’ to remove excess data. Default is ‘zero-padding’.

namestr, optional

Name of the instance. Default is “S2P”.

Example 1

>>> processor = Serial2Parallel(3)
>>> X = np.arange(5)
>>> print(X)
[0 1 2 3 4]
>>> print(X.shape)
(5,)
>>> Y = processor(X)
>>> print(Y.shape)
(3, 2)
>>> print(Y[:, 0])
[0 1 2]
>>> print(Y[:, 1])
[3 4 0]

Example 2

>>> processor = Serial2Parallel(3)
>>> X = np.arange(10).reshape(2, 5)
>>> print(X)
[[0 1 2 3 4]
[5 6 7 8 9]]
>>> print(X.shape)
(2, 5)
>>> print(X[0, :])
[0 1 2 3 4]
>>> Y = processor(X)
>>> print(Y.shape)
(2, 3, 2)
>>> print(Y[0, :, 0])
[0 1 2]
>>> print(Y[0, :, 1])
[3 4 0]

class Parallel2Serial(order: str = 'F', name: str = 'P2S')

A class for converting parallel data streams into a serial data stream.

This class reshapes a multi-dimensional array (parallel data streams) into another multi-dimensional array where the last dimension is flattened into a serial data stream, using a specified ordering scheme.

Attributes

orderstr, optional

The order in which to reshape the array. ‘F’ means to reshape in column-major (Fortran-style) order, and ‘C’ means to reshape in row-major (C-style) order. Default is ‘F’.

namestr, optional

Name of the instance. Default is “P2S”.

Example 1

>>> processor = Parallel2Serial()
>>> X = np.array([[0, 3], [1, 4], [2, 0]])
>>> print(X.shape)
(3, 2)
>>> Y = processor(X)
>>> print(Y.shape)
(6,)
>>> print(Y)
[0 1 2 3 4 0]

Example 2

>>> processor = Parallel2Serial()
>>> X = np.array([[[0, 3],[1, 4], [2, 0]], [[5, 8], [6, 9], [7, 0]]])
>>> print(X.shape)
(2, 3, 2)
>>> Y =processor(X)
>>> print(Y.shape)
(2, 6)
>>> print(Y[0, :])
[0 1 2 3 4 0]

class Amplifier(gain: float = 1.0, axis: int | None = None, name: str = 'signal_amplifier')

A class for amplifying or attenuating a signal along a specified axis.

This class multiplies an input signal by a specified gain factor, effectively amplifying or attenuating the signal based on the gain value. The gain can be applied to all elements or selectively along a specified axis.

Attributes

gainfloat

The amplification factor by which the signal will be multiplied.

axisint or None, optional

The axis along which to apply the gain. If None, the gain is applied to the entire array. Default is None.

namestr

Name of the signal amplifier instance.

Example 1

>>> amplifier = Amplifier(gain=2)
>>> X = np.array([[1, 2], [3, 4]])
>>> print(X)
[[1 2]
[3 4]]
>>> Y = amplifier(X)
>>> print(Y)
[[2 4]
[6 8]]

Example 2

>>> amplifier = Amplifier(gain=3, axis=-1)
>>> X = np.array([[1, 2], [3, 4]])
>>> print(X)
[[1 2]
[3 4]]
>>> Y = amplifier(X)
>>> print(Y)
[[ 1.  6.]
[ 3. 12.]]

class WeightAmplifier(weight: ndarray | None = None, axis: int = -1, name: str = 'parallel_signal_weight')

Applies weights to a MIMO (Multiple Input Multiple Output) parallel signal along a specified axis.

This class multiplies each parallel stream of the input signal by a corresponding weight along a specified axis. The weights are applied selectively to the elements along the specified axis.

Signal Model

\[y_l[n] = w_l x_l[n]\]

where the coefficient \(w_l\) specifies the weight.

Attributes

weightnp.ndarray

Weights to be applied to each parallel stream of the input signal (1D array).

axisint, optional

The axis along which to apply the weights. Default is -1 (the last axis).

namestr

Name of the weight amplifier instance.

Example 1

>>> weight_amplifier = WeightAmplifier(weight=np.array([2, 3]), axis=0)
>>> X = np.array([[1, 2], [3, 4]])
>>> print(X)
[[1 2]
 [3 4]]
>>> Y = weight_amplifier.forward(X)
>>> print(Y)
[[2 4]
 [9 12]]

Example 2

>>> weight_amplifier = WeightAmplifier(weight=np.array([2, 3]), axis=-1)
>>> X = np.array([[1, 2], [3, 4]])
>>> print(X)
[[1 2]
 [3 4]]
>>> Y = weight_amplifier(X)
>>> print(Y)
[[2 6]
 [6 12]]

class Complex2Real(part: Literal['real', 'imag'] = 'real', validate_input: bool = False)

A processor class to extract the real or imaginary part of a complex array.

Attributes

partLiteral[“real”, “imag”]

Specifies which part of the complex number to extract. Can be either “real” or “imag”. Default is “real”.

validate_inputbool

If True, validates that the input array is purely real or imaginary based on the specified part. Default is False.

Example 1

>>> processor = Complex2Real(part="real")
>>> X = np.array([1+2j, 3+4j, 5+0j])
>>> processor(X)
array([1., 3., 5.])

Example 2

>>> processor_imag = Complex2Real(part="imag")
>>> X = np.array([1+2j, 3+4j, 5+0j])
>>> processor_imag(X)
array([2., 4., 0.])

Example 3

>>> processor_real2 = Complex2Real(part="imag", validate_input=True)
>>> X = np.array([1+2j, 3+4j, 5+0j])
>>> processor(X)  # Raises ValueError
ValueError: The input data is not real since the imaginary part is non-zero.

class AutoConcatenator(input_copy_mask: ndarray | None = None, output_original_mask: ndarray | None = None, output_copy_mask: ndarray | None = None, axis: int = 0, name: str = 'auto concatenator')

A class to automatically concatenate data along a specified axis using masks.

This class facilitates the extraction and concatenation of data from an input array based on predefined masks. It ensures that the output array is constructed correctly by validating the shapes and contents of the masks.

Attributes

input_copy_maskOptional[np.ndarray]

A boolean mask used to extract a portion of the input data to be copied.

output_original_maskOptional[np.ndarray]

A boolean mask indicating where the original data should be placed in the output array.

output_copy_maskOptional[np.ndarray]

A boolean mask indicating where the copied data should be placed in the output array.

axisint

The axis along which to perform the concatenation. Default is 0.

namestr

The name of the processor. Default is “auto concatenator”.

Example 1

>>> input_copy_mask = np.array([True, False, True])
>>> output_original_mask = np.array([True, True, True, False, False])
>>> output_copy_mask = np.array([False, False, False, True, True])
>>> X = np.array([1, 2, 3])
[1 2 3]
>>> concatenator = AutoConcatenator(input_copy_mask, output_original_mask, output_copy_mask)
>>> Y = concatenator(X)
[1 2 3 1 3]

Example 2

>>> input_copy_mask = np.array([True, False])
>>> output_original_mask = np.array([True, True, False, False, False])
>>> output_copy_mask = np.array([False, False, False, True, False])
>>> X = np.array([[1, 2, 3], [4, 5, 6]])
[[1 2 3]
[4 5 6]]
>>> concatenator = AutoConcatenator(input_copy_mask, output_original_mask, output_copy_mask)
>>> Y = concatenator(X)
[[1 2 3]
[4 5 6]
[0 0 0]
[1 2 3]
[0 0 0]]

Example 3

>>> input_copy_mask = np.array([False, True, True])
>>> output_original_mask = np.array([False, True, True, True, False])
>>> output_copy_mask = np.array([True, False, False, False, True])
>>> X = np.array([[1, 2, 3], [4, 5, 6]])
[[1 2 3]
[4 5 6]]
>>> concatenator = AutoConcatenator(input_copy_mask, output_original_mask, output_copy_mask, axis=-1)
>>> result = concatenator(X)
[[2 1 2 3 3]
[5 4 5 6 6]]

extract_copy(X: ndarray)

Extract a copy from the input signal on the specified axis

class SampleRemover(N_start: int = 0, length: int = 0, name: str = 'symbol remover')

Deletes samples from a signal.

This class removes a specified number of samples starting from a given index in the signal.

Attributes

N_startint

Index of the first sample to delete.

lengthint

Number of samples to delete.

namestr

Name of the symbol remover instance. Default is “SymbolRemover”.

class Delay_Remover(delay: int, is_mimo: bool = True, axis: int = -1, name: str = 'delay remover')

Removes an initial delay from a signal.

This class removes a specified number of initial samples (delay) from the signal.

Attributes

delayint

Number of initial samples to remove.

namestr

Name of the delay remover instance. Default is “DelayRemover”.

class DataAdder(symbol: ndarray, N_start: int = 0, name: str = 'Data Adder')

Inserts symbol samples into a signal.

This class inserts a specified symbol into a signal at a given index.

Attributes

symbolnp.ndarray

Symbol to be inserted into the signal.

N_startint

Index at which to insert the symbol.

namestr

Name of the data adder instance. Default is “DataAdder”.

class Signal_Extractor(N_start: int = 0, N: int | None = None, name: str = 'Signal Extractor')

A class for extracting a segment from a signal.

This class is used to extract a specific portion of a signal, starting from a specified index. It can extract either until the end of the signal or for a specified number of samples.

Attributes

N_startint

The starting index from where the extraction begins.

NOptional[int]

The number of samples to extract. If None, extracts until the end of the signal.

namestr

Name of the extractor instance. Default is “DataExtractor”.

class Resampler(up: int, down: int, name: str = 'Resampler')

A class for resampling a signal.

This class changes the sampling rate of a signal by a rational factor. It performs upsampling by the specified ‘up’ factor, followed by downsampling by the specified ‘down’ factor, effectively changing the sampling rate by a factor of up/down.

Attributes

upint

The upsampling factor. Must be a positive integer.

downint

The downsampling factor. Must be a positive integer.

namestr

Name of the resampler instance. Default is “Resampler”.

class Clipper(threshold: float, name: str = 'Clipper')

Clipper class for clipping signal values to a specified threshold.

Signal Model

\[y[n] = \frac{x[n]}{|x[n]|} \cdot \min{(|x[n]|, \tau)}\]

Attributes

thresholdfloat

The threshold value \(\tau\) for clipping.

namestr

Name of the clipper instance.

class Blind_Phase_Tracker

A class implementing a blind phase tracking algorithm using a grid search approach.

This algorithm estimates and compensates for unknown phase rotations in a received signal by minimizing the local Error Vector Magnitude (EVM) around each sample. It assumes the signal belongs to a known modulation alphabet (e.g., QAM or PSK).

Parameters

Lint

The number of neighboring symbols on each side used to compute the local EVM cost.

alphabetnp.ndarray

The set of complex symbols representing the modulation constellation.

phase_stepsint, optional

The number of discrete phase candidates evaluated within the range [-π/4, π/4). Default is 10.

Methods

hard_projector(z)

Projects input samples to the nearest symbols from the constellation.

evm_cost(x, n, phi)

Computes the local EVM cost at index n for a candidate phase shift phi.

forward(x)

Applies blind phase correction to the input signal x and plots the estimated phase evolution over time.