Processors#

class CyclicPrefixer(N_cp: int = 10, *, input_copy_mask: ndarray | None = None, output_original_mask: ndarray | None = None, output_copy_mask: ndarray | None = None, axis: int = -1, name: str = 'cp adder')[source]#

Processor for adding a cyclic prefix to combat multipath interference.

Signal Model#

The cyclic prefix is a copy of the last \(N_{cp}\) samples of each block, prepended to the block to combat multipath interference and inter-symbol interference:

\[\begin{split}y[n] = \left\{\begin{array}{cl} x[N - N_{cp} + n] & \text{for } 0 \le n < N_{cp},\\ x[n - N_{cp}] & \text{for } N_{cp} \le n < N + N_{cp}, \end{array}\right.\end{split}\]

where:

\(x[n]\) is the input block of length \(N\),
\(y[n]\) is the output block of length \(N + N_{cp}\),
\(N_{cp}\) is the length of the cyclic prefix.

Axes: axis -1 – the prefix is prepended along the block content axis (Block layout (..., T, F), see CONVENTIONS.md).

param N_cp:: Length \(N_{cp}\) of the cyclic prefix to be added. Must be a non-negative integer. Default is 10.
type N_cp:: int
param name:: Name of the processor instance. Default is "cp adder".
type name:: str, optional, keyword-only

References

R. van Nee, R. Prasad, OFDM for Wireless Multimedia Communications, Artech House, 2000, Chapter 2.

Examples

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

prepare(X: ndarray)[source]#: Initialize the mask based on the signal shape

class CyclicPrefixRemover(N_cp: int, *, name: str = 'cp remover')[source]#

Processor for removing a cyclic prefix from the input data.

Signal Model#

The removal of the cyclic prefix discards the first \(N_{cp}\) samples of each block:

\[y[n] = x[n + N_{cp}], \qquad 0 \le n < N,\]

where:

\(x[n]\) is the input block of length \(N + N_{cp}\) that contains the cyclic prefix,
\(y[n]\) is the output block of length \(N\) after removing the cyclic prefix,
\(N_{cp}\) is the length of the cyclic prefix.

Axes: axis -1 – the prefix is removed along the block content axis (Block layout (..., T, F), see CONVENTIONS.md).

param N_cp:: Length \(N_{cp}\) of the cyclic prefix to be removed. Must be a non-negative integer.
type N_cp:: int
param name:: Name of the processor instance. Default is "cp remover".
type name:: str, optional, keyword-only

References

R. van Nee, R. Prasad, OFDM for Wireless Multimedia Communications, Artech House, 2000, Chapter 2.

Examples

>>> X = np.arange(13)
>>> remover = CyclicPrefixRemover(N_cp=3)
>>> print(remover(X))
[ 3  4  5  6  7  8  9 10 11 12]

class HermitianPrefixer(*, input_copy_mask: ndarray | None = None, output_original_mask: ndarray | None = None, output_copy_mask: ndarray | None = None, axis: int = -1, name: str = 'hermitian prefixer', shift: bool = False)[source]#

Processor for enforcing Hermitian symmetry on frequency-domain blocks, so that the IDFT output is real-valued.

Signal Model#

Each input block of length \(N\) is extended to a block of length \(2(N+1)\) whose spectrum is Hermitian symmetric: the DC and Nyquist subcarriers are set to zero and the conjugate mirror of the data is appended. When shift is False, the output is given by

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

where:

\(x[k]\) is the input block of length \(N\),
\(y[k]\) is the Hermitian-symmetric output block of length \(2(N+1)\).

When shift is True, the two halves are swapped (conjugate copy first), matching an fftshift-ed subcarrier ordering.

Axes: axis -1 – the Hermitian extension is built along the block content axis (Block layout (..., T, F), see CONVENTIONS.md).

param shift:: Whether to swap the original and conjugate halves. Default is False.
type shift:: bool, optional, keyword-only
param name:: Name of the processor instance. Default is "hermitian prefixer".
type name:: str, optional, keyword-only

References

J. Armstrong, “OFDM for Optical Communications”, Journal of Lightwave Technology, vol. 27, no. 3, pp. 189-204, 2009.

Examples

>>> X = np.arange(1, 4) + 1j*np.arange(1, 4)
>>> prefixer = HermitianPrefixer()
>>> print(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]
>>> x = np.arange(1, 7) + 1j*np.arange(1, 7)
>>> X = np.reshape(x, (2, 3))
>>> prefixer = HermitianPrefixer(shift=True)
>>> print(prefixer(X))
[[0.+0.j 3.-3.j 2.-2.j 1.-1.j 0.+0.j 1.+1.j 2.+2.j 3.+3.j]
 [0.+0.j 6.-6.j 5.-5.j 4.-4.j 0.+0.j 4.+4.j 5.+5.j 6.+6.j]]

class FFTProcessor(shift: bool = False, *, norm: Literal['ortho', 'backward', 'forward'] = 'ortho', name: str = 'fft')[source]#

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

Signal Model#

The FFT computes the Discrete Fourier Transform (DFT), which maps a block of time-domain samples to its frequency-domain representation. With the default orthonormal normalization (norm="ortho"), the DFT of a block of length \(N\) is given by:

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

where:

\(x[n]\) is the input block of time-domain samples,
\(y[k]\) is the output value at subcarrier \(k\),
\(N\) is the block length.

Axes: axis -1 – the FFT changes the meaning of axis -1 (time -> frequency), never its position (see CONVENTIONS.md).

param shift:: If True, applies the FFT shift which swaps the low and high frequency components. Default is False.
type shift:: bool, optional
param norm:: Normalization mode for FFT. “ortho” means orthonormal FFT is computed. Default is “ortho”.
type norm:: {“ortho”, “backward”, “forward”}, optional, keyword-only
param name:: Name of the processor instance. Default is "fft".
type name:: str, optional, keyword-only

References

J. G. Proakis, M. Salehi, Digital Communications, 5th ed., McGraw-Hill, 2008, Section 11.2.

Examples

>>> fft_processor = FFTProcessor()
>>> x = np.ones(4)
>>> print(np.round(np.abs(fft_processor(x)), 3) + 0.0)
[2. 0. 0. 0.]

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

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

Signal Model#

The IFFT computes the Inverse Discrete Fourier Transform (IDFT), which maps a block of frequency-domain samples back to the time domain. With the default orthonormal normalization (norm="ortho"), the IDFT of a block of length \(N\) is given by:

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

where:

\(x[k]\) is the input value at subcarrier \(k\),
\(y[n]\) is the output block of time-domain samples,
\(N\) is the block length.

Axes: axis -1 – the IFFT changes the meaning of axis -1 (frequency -> time), never its position (see CONVENTIONS.md).

param shift:: If True, applies the IFFT shift which swaps the low and high frequency components. Default is False.
type shift:: bool, optional
param norm:: Normalization mode for IFFT. “ortho” means orthonormal IFFT is computed. Default is “ortho”.
type norm:: {“ortho”, “backward”, “forward”}, optional, keyword-only
param name:: Name of the processor instance. Default is "ifft".
type name:: str, optional, keyword-only

References

J. G. Proakis, M. Salehi, Digital Communications, 5th ed., McGraw-Hill, 2008, Section 11.2.

Examples

>>> ifft_processor = IFFTProcessor()
>>> X = np.array([2.0, 0.0, 0.0, 0.0])
>>> print(np.round(np.abs(ifft_processor(X)), 3))
[1. 1. 1. 1.]

class CarrierAllocator(carrier_type: object, *, pilots: ndarray | None = None, name: str = 'carrier allocator')[source]#

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 on the pilot subcarriers and zeros on the null subcarriers:

\[\begin{split}y[k] = \left\{\begin{array}{cl} x[m_k] & \text{if } s[k] = 1 \text{ (data)}\\ p[j_k] & \text{if } s[k] = 2 \text{ (pilot)}\\ 0 & \text{if } s[k] = 0 \text{ (null)}\\ \end{array}\right.\end{split}\]

where:

\(x[m]\) is the input data vector, with \(m_k\) indexing the data subcarriers in increasing subcarrier index \(k\),
\(p[j]\) is the pilot value vector, with \(j_k\) indexing the pilot subcarriers in increasing \(k\),
\(s[k]\) is the subcarrier type array, where each element specifies the type of the \(k\)-th subcarrier,
\(y[k]\) is the output data vector over the full set of subcarriers.

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

Axes: declared axis – expects the Block layout (..., T, F) (or (..., N_data) for a period-1 allocation) and validates it in prepare() (decision D18). The mask says where, the pilots argument says what: transmitter and receiver share the same CarrierAllocation object, which removes the “diverging masks” class of bugs.

param carrier_type:: The allocation, providing the subcarrier type array \(s[k]\). A CarrierAllocation is described in physical order and converted once with to_fft_order() (decision D16); a raw 1D array is taken as an FFT-order mask.
type carrier_type:: CarrierAllocation or np.ndarray
param pilots:: Pilot values \(p[j]\), one per PILOT subcarrier of an OFDM symbol (or a scalar broadcast to all of them). Default is None, treated as no pilots.
type pilots:: np.ndarray or scalar, optional, keyword-only
param name:: Name of the processor instance. Default is "carrier allocator".
type name:: str, optional, keyword-only
raises ShapeError:: If the last axis of the input does not hold N_data data subcarriers, or if a periodic (scattered) allocation is fed a 1D signal without the OFDM-symbol axis.

References

R. van Nee, R. Prasad, OFDM for Wireless Multimedia Communications, Artech House, 2000, Chapter 2.

Examples

>>> 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])
>>> print(allocator(X))
[ 1 -1  0  2 -1  3]

>>> carrier_type = np.array([1, 0, 0, 1, 1])
>>> allocator = CarrierAllocator(carrier_type=carrier_type)
>>> X = np.array([[1, 4, 7], [2, 5, 8], [3, 6, 9]])
>>> print(allocator(X))
[[1 0 0 4 7]
 [2 0 0 5 8]
 [3 0 0 6 9]]

>>> from comnumpy.ofdm.allocation import get_allocation
>>> allocator = CarrierAllocator(get_allocation("802.11a"), pilots=1.0)
>>> Y = allocator(np.ones((10, 48)))
>>> print(Y.shape)
(10, 64)

plot(ax: Axes | None = None, shift: bool = False) → Axes[source]#: Plot the carrier allocation; returns the axis (decision D25).

class CarrierExtractor(carrier_type: object, *, name: str = 'carrier extractor')[source]#

Processor for extracting data from specific subcarriers.

Signal Model#

The Carrier Extractor extracts the data subcarriers based on a predefined subcarrier type array, undoing the mapping of the CarrierAllocator:

\[y[m] = x[k_m]\]

where:

\(x[k]\) is the input data over the full set of subcarriers,
\(k_m\) is the index of the \(m\)-th subcarrier of type 1 (data subcarrier) in the input vector,
\(y[m]\) is the output data vector.

The indices \(k_m\) are determined by the positions in the carrier_type array where the value is 1. The content of the pilot subcarriers (type 2) is exposed after each run as the estimated attribute pilots_ (decision D23).

Axes: declared axis – expects the Block layout (..., T, N_fft) (or (..., N_fft) for a period-1 allocation) and validates it in prepare() (decision D18). Share the same CarrierAllocation object with the transmitter’s CarrierAllocator.

param carrier_type:: The allocation, providing the subcarrier type array used to find the data indices \(k_m\). A CarrierAllocation is described in physical order and converted once with to_fft_order() (decision D16); a raw 1D array is taken as an FFT-order mask.
type carrier_type:: CarrierAllocation or np.ndarray
param name:: Name of the processor instance. Default is "carrier extractor".
type name:: str, optional, keyword-only

pilots_#

Content of the pilot subcarriers seen during the last run (data-dependent, hence the trailing underscore). An array for a period-1 allocation; a list of per-symbol arrays otherwise (pilot counts may differ between symbols).

Type:: np.ndarray or list of np.ndarray

raises ShapeError:: If the last axis of the input does not hold N_fft subcarriers, or if a periodic (scattered) allocation is fed a 1D signal without the OFDM-symbol axis.

References

R. van Nee, R. Prasad, OFDM for Wireless Multimedia Communications, Artech House, 2000, Chapter 2.

Examples

>>> carrier_type = np.array([1, 2, 0, 1, 2, 1])
>>> allocator = CarrierAllocator(carrier_type=carrier_type, pilots=np.array([-1, -2]))
>>> extractor = CarrierExtractor(carrier_type=carrier_type)
>>> Z = allocator(np.array([1, 2, 3]))
>>> print(extractor(Z))
[1 2 3]
>>> print(extractor.pilots_)
[-1 -2]

>>> carrier_type = np.array([1, 0, 0, 1, 1])
>>> allocator = CarrierAllocator(carrier_type=carrier_type)
>>> extractor = CarrierExtractor(carrier_type=carrier_type)
>>> X = np.array([[1, 4, 7], [2, 5, 8]])
>>> Z = allocator(X)
>>> print(Z)
[[1 0 0 4 7]
 [2 0 0 5 8]]
>>> print(extractor(Z))
[[1 4 7]
 [2 5 8]]

plot(ax: Axes | None = None, shift: bool = False) → Axes[source]#: Plot the carrier allocation; returns the axis (decision D25).