Processors#

class OFDMTransmitter(nb_carrier_data: int, carrier_type: ndarray, nb_cp: int, chain: Sequential = None)#

OFDM Transmitter for converting digital data into an OFDM signal.

Signal Model#

The OFDM Transmitter processes input data through a series of steps:

  1. Serial to Parallel Conversion: Converts the serial data stream into parallel sub-streams.

  2. Carrier Allocation: Allocates data to specific subcarriers.

  3. Inverse Fast Fourier Transform (IFFT): Transforms the frequency-domain data into time-domain signals.

  4. Cyclic Prefixing: Adds a cyclic prefix to combat multipath interference.

  5. Parallel to Serial Conversion: Converts the parallel streams back into a serial data stream.

Attributes#

nb_carrier_dataint

Number of data carriers used in OFDM.

carrier_typenp.ndarray

Array specifying the carrier type.

nb_cpint

Number of samples in the Cyclic Prefix (CP).

chainSequential

Processing chain encapsulating the sequence of OFDM transmission steps.

class CyclicPrefixer(input_copy_mask: ndarray | None = None, output_original_mask: ndarray | None = None, output_copy_mask: ndarray | None = None, axis: int = 0, name: str = 'cp adder', N_cp: int = 10)#

Processor for adding a cyclic prefix to combat multi-path interference.

Signal Model#

The cyclic prefix is a portion of the signal that is prepended to the original data to combat multipath interference and inter-symbol interference in communication systems. The addition of the cyclic prefix involves copying the last N_cp samples of the original data and placing them at the beginning along the specified axis.

Mathematically, if X is the original input signal, the operation can be described as:

\[\begin{split}\mathbf{y}[n] = \begin{bmatrix} \mathbf{0}_{N_{cp},N-N_{cp}} & \mathbf{I}_{N_{cp},N_{cp}} \\ \mathbf{I}_{N-N_{cp},N-N_{cp}} & \mathbf{0}_{N-N_{cp},N_{cp}}\\ \mathbf{0}_{N_{cp},N-N_{cp}} & \mathbf{I}_{N_{cp},N_{cp}}\\ \end{bmatrix} \mathbf{x}[n]\end{split}\]

where:

  • \(N_{cp}\) is the length of the cyclic prefix,

  • \(\mathbf{x}[n]\) is the input signal of size \(N\),

  • \(\mathbf{y}[n]\) is the output signal of size \(N+N_{cp}\) after adding the cyclic prefix.

Attributes#

N_cpint

Length of the cyclic prefix to be added. Must be a non-negative integer.

axisint, optional

Axis along which to add the cyclic prefix. Default is the first dimension

Example 1#

>>> X = np.arange(10)
>>> prefixer = CyclicPrefixer(N_cp=3)
>>> Y = prefixer(X)
[7 8 9 0 1 2 3 4 5 6 7 8 9]

Example 2#

>>> X = np.array([[1, 4, 7], [2, 5, 8], [3, 6, 9]])
>>> prefixer = CyclicPrefixer(N_cp=2)
>>> Y = prefixer(X)
[[2 5 8]
[3 6 9]
[1 4 7]
[2 5 8]
[3 6 9]]      
prepare(X: ndarray)#

Initialize the mask based on the signal shape

class CyclicPrefixRemover(N_cp: int, axis: int = 0, name: str = 'cp remover')#

Processor for removing a cyclic prefix from the input data.

Signal Model#

The cyclic prefix is a portion of the signal that is prepended to the original data to combat multipath interference and inter-symbol interference in communication systems. The removal of the cyclic prefix involves discarding the first N_cp samples along the specified axis.

Mathematically, if X is the input signal with a cyclic prefix, the operation can be described as:

\[\mathbf{y}[n] =\begin{bmatrix} \mathbf{0}_{N,N_{cp}} & \mathbf{I}_{N,N} \end{bmatrix} \mathbf{x}[n]\]

where:

  • \(N_{cp}\) is the length of the cyclic prefix,

  • \(\mathbf{x}[n]\) is the input signal of size \(N+N_{cp}\) that contains the cyclic prefix,

  • \(\mathbf{y}[n]\) is the output signal of size \(N\) after removing the cyclic prefix.

Attributes#

N_cpint

Length of the cyclic prefix to be removed. Must be a non-negative integer.

axisint, optional

Axis along which to remove the cyclic prefix. Default is the first dimension

class HermitianPrefixer(input_copy_mask: ndarray | None = None, output_original_mask: ndarray | None = None, output_copy_mask: ndarray | None = None, axis: int = 0, name: str = 'hermitian prefixer', shift: bool = False)#

Processor for preparing data to enforce Hermitian symmetry, useful in signal processing applications.

Signal Model#

The HermitianPrefixer generates masks that can be used to enforce Hermitian symmetry on a given signal.

Mathematically, the masks are designed to handle the input signal X and prepare it for Hermitian operations. The process involves creating masks that identify the portions of the signal to be copied and transformed. When shift is false, the output is given by

\[\begin{split}y[n] = \left\{\begin{array}{cl} 0 &\text{if }n=0, N+2,\\ x[n] &\text{for }n=1, \cdots, N+1,\\ x^*[n-N+2)] &\text{for }n=N+2, \cdots, 2N+1. \end{array}\right.\end{split}\]

Attributes#

axisint

The axis along which to apply the Hermitian operation. Default is 0.

shiftbool

Whether to apply a shift to the masks. Default is False.

namestr

The name of the prefixer. Default is “hermitian prefixer”.

Example 1#

>>> X = np.arange(1, 4) + 1j*np.arange(1, 4)
>>> prefixer = HermitianPrefixer()
>>> Y = prefixer(X)
[0.+0.j 1.+1.j 2.+2.j 3.+3.j 0.+0.j 3.-3.j 2.-2.j 1.-1.j]

Example 2#

>>> x = np.arange(1, 7) + 1j*np.arange(1, 7)
>>> X = np.reshape(x, (3, 2), order="F")
>>> prefixer = HermitianPrefixer(shift=True)
>>> Y = prefixer(X)
[[0.+0.j 0.+0.j]
[3.-3.j 6.-6.j]
[2.-2.j 5.-5.j]
[1.-1.j 4.-4.j]
[0.+0.j 0.+0.j]
[1.+1.j 4.+4.j]
[2.+2.j 5.+5.j]
[3.+3.j 6.+6.j]]
class FFTProcessor(axis: int = 0, shift: bool = False, norm: Literal['ortho', 'backward', 'forward'] = 'ortho', name: str = 'fft')#

Processor for performing Fast Fourier Transform (FFT) on the input data.

Signal Model#

The Fast Fourier Transform (FFT) is an algorithm to compute the Discrete Fourier Transform (DFT) and its inverse efficiently. The DFT transforms a sequence of values in the time domain into a sequence of values in the frequency domain.

Mathematically, the DFT of a sequence \(x[n]\) of length \(N\) is given by:

\[y[k] = \frac{1}{\sqrt{N}}\sum_{l=0}^{N-1} x[l] \cdot e^{-i 2 \pi k l / N}\]

The FFT operation can be represented in matrix form as:

\[\mathbf{y}[n] = \mathbf{W} \mathbf{x}[n]\]
  • \(\mathbf{W}\) is the DFT matrix of size ( N times N ),

  • \(\mathbf{x}[n]\) is the input vector of time-domain samples,

  • \(\mathbf{y}[n]\) is the output vector of frequency-domain samples.

Attributes#

axisint, optional

Axis along which to perform the FFT. Default is the first axis

shiftbool, optional

If True, applies the FFT shift which swaps the low and high frequency components. Default is False.

norm{“ortho”, “backward”, “forward”}, optional

Normalization mode for FFT. “ortho” means orthonormal FFT is computed. None means no normalization is applied. Default is “ortho”.

class IFFTProcessor(axis: int = 0, shift: bool = False, norm: Literal['ortho', 'backward', 'forward'] = 'ortho', name: str = 'ifft')#

Processor for performing Inverse Fast Fourier Transform (IFFT) on the input data.

Signal Model#

The Inverse Fast Fourier Transform (IFFT) is an algorithm to compute the Inverse Discrete Fourier Transform (IDFT) efficiently. The IDFT transforms a sequence of values in the frequency domain back into the time domain.

Mathematically, the IDFT of a sequence \(x[k]\) of length \(N\) is given by:

\[y[l] = \frac{1}{\sqrt{N}} \sum_{k=0}^{N-1} x[k] \cdot e^{i 2 \pi k n / N}\]

The IFFT operation can be represented in matrix form as:

\[\mathbf{y}[n] = \mathbf{W}^{H} \mathbf{x}[n]\]
  • \(\mathbf{W}^{H}\) is the inverse DFT matrix of size \(N \times N\),

  • \(\mathbf{x}[n]\) is the input vector of frequency-domain samples,

  • \(\mathbf{y}[n]\) is the output vector of time-domain samples.

Attributes#

axisint, optional

Axis along which to perform the FFT. Default is the first axis

shiftbool, optional

If True, applies the IFFT shift which swaps the low and high frequency components. Default is True.

norm{“ortho”, “backward”, “forward”}, optional

Normalization mode for IFFT. “ortho” means orthonormal FFT is computed. None means no normalization is applied. Default is “ortho”.

class CarrierAllocator(carrier_type: ndarray, pilots: ndarray | None = None, axis: int = 0, name: str = 'carrier allocator')#

Processor for allocating data to specific subcarriers.

Signal Model#

The Carrier Allocator assigns data to specific subcarriers based on a predefined subcarrier type array. It supports the insertion of pilot values and ensures Hermitian symmetry for certain subcarriers.

Mathematically, the allocation can be described as:

\[\begin{split}y[n] = \left\{\begin{array}{cl} x[m] & \text{if } s[n] = 1 \\ p[k] & \text{if } s[n] = 2 \\ 0 & \text{if } s[n] = 0 \\ \end{array}\right.\end{split}\]

where:

  • \(x[m]\) is the input data vector, with \(m\) indexing the data subcarriers,

  • \(p[k]\) is the pilot value vector, with \(k\) indexing the pilot subcarriers,

  • \(s[n]\) is the subcarrier type array, where each element specifies the type of the \(n\)-th subcarrier,

  • \(y[n]\) is the output data vector, with allocated subcarriers.

The indices \(m\) and \(k\) are determined by the positions in the carrier_type array where the values are 1 and 2, respectively.

Attributes#

carrier_typenp.ndarray

Array specifying the type of each subcarrier.

pilotsnp.ndarray, optional

Array of pilot values to be inserted into the subcarriers. Default is an empty array.

axisint, optional

Axis along which to allocate subcarriers. Default is the first dimension.

Example 1#

>>> carrier_type = np.array([1, 2, 0, 1, 2, 1])
>>> pilots = np.array([-1, -1])
>>> allocator = CarrierAllocator(carrier_type=carrier_type, pilots=pilots)
>>> X = np.array([1, 2, 3])
[1 2 3]
>>> Y = allocator(X)
[ 1 -1  0  2 -1  3]

Example 2#

>>> carrier_type = np.array([1, 0, 0, 1, 1])
>>> allocator = CarrierAllocator(carrier_type=carrier_type, axis=-1)
>>> X = np.array([[1, 4, 7], [2, 5, 8], [3, 6, 9]])
[[1 4 7]
[2 5 8]
[3 6 9]]
>>> Y = allocator(X)
[[1 0 0 4 7]
[2 0 0 5 8]
[3 0 0 6 9]]
plot(shift=False)#

Plot the carrier allocation

class CarrierExtractor(carrier_type: ndarray, pilot_recorder: Callable | None = None, axis: int = 0, name: str = 'carrier extractor')#

Processor for extracting data from specific subcarriers.

Signal Model#

The Carrier Extractor extracts data from specific subcarriers based on a predefined subcarrier type array. It supports the extraction of pilot values and ensures Hermitian symmetry for certain subcarriers.

Mathematically, the extraction can be described as:

\[y[n] = x[k_n]\]

where:

  • \(x[n]\) is the input data,

  • \(k_m\) is the index of the \(m\)-th subcarrier of type 1 (data subcarrier) in the input vector,

  • \(y[n]\) is the output data.

The indices \(k_m\) are determined by the positions in the carrier_type array where the value is 1.

Attributes#

carrier_typenp.ndarray

Array specifying the type of each subcarrier.

pilot_recordercallable, optional

Function to record the content associated to pilot values if required. Default is None.

axisint, optional

Axis along which to extract subcarriers. Default is the first dimension

Example 1#

>>> carrier_type = np.array([1, 2, 0, 1, 2, 1])
>>> pilots = np.array([-1, -2])
>>> pilot_recorder = Recorder()
>>> allocator = CarrierAllocator(carrier_type=carrier_type, pilots=pilots)
>>> extractor = CarrierExtractor(carrier_type=carrier_type, pilot_recorder=pilot_recorder)
>>> X = np.array([1, 2, 3])
>>> Z = allocator(X)
[ 1 -1  0  2 -2  3]
>>> Y = extractor(Z)
[1 2 3]
>>> pilot_recorded = pilot_recorder.get_data()
[-1 -2]

Example 2#

>>> carrier_type = np.array([1, 0, 0, 1, 1])
>>> allocator = CarrierAllocator(carrier_type=carrier_type, axis=-1)
>>> extractor = CarrierExtractor(carrier_type=carrier_type, axis=-1)
>>> X = np.array([[1, 4, 7], [2, 5, 8]])
>>> Z = allocator(X)
[[1 0 0 4 7]
[2 0 0 5 8]]
>>> Y = extractor(Z)
[[1 4 7]
[2 5 8]]
plot(shift=False)#

Plot the carrier allocation