Processors

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

Increases the sampling rate of a signal by inserting zeros between samples.

Signal Model

Upsampling by an integer factor \(L\) inserts \(L-1\) zeros between consecutive input samples, so that the output rate is \(L\) times the input rate:

\[\begin{split}y[n] = \left\{\begin{array}{cl} \alpha \, x\!\left[\dfrac{n-\tau}{L}\right] & \text{if } (n-\tau) \bmod L = 0,\\[6pt] 0 & \text{otherwise,} \end{array}\right.\end{split}\]

where:

• \(x[n]\) is the input sequence of length \(N\),

• \(y[n]\) is the output sequence of length \(L N\),

• \(L\) is the upsampling factor,

• \(\tau\) is the phase offset of the retained samples,

• \(\alpha\) is an output scaling factor.

Zero insertion creates \(L-1\) spectral images of the input spectrum. When use_filter is set, a lowpass filter of normalized cutoff \(1/L\) is applied to the zero-stuffed sequence, which turns the operation into an interpolation.

Axes: declared axis – the zero stuffing runs along axis (default -1, the sample axis of the Serial layout (..., N)) and the remaining axes are left untouched. The anti-imaging filter of use_filter=True always acts along the last axis, whatever axis says.

param L:

Upsampling factor \(L\). Must be a positive integer.

type L:

int

param phase:

Phase offset \(\tau\) of the retained samples, with \(0 \le \tau < L\). Default is 0.

type phase:

int, optional, keyword-only

param scale:

Output scaling factor \(\alpha\). Default is 1.0.

type scale:

float, optional, keyword-only

param axis:

Axis along which the upsampling is performed. Default is -1.

type axis:

int, optional, keyword-only

param use_filter:

If True, apply the anti-imaging lowpass filter of normalized cutoff \(1/L\) at the output. Default is False.

type use_filter:

bool, optional, keyword-only

param name:

Name of the processor instance. Default is "upsampler".

type name:

str, optional, keyword-only

References

A. V. Oppenheim, R. W. Schafer, Discrete-Time Signal Processing, 3rd ed., Pearson, 2010, Chapter 4 (changing the sampling rate by discrete-time processing).

Examples

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

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

Decreases the sampling rate of a signal by keeping one sample out of every L.

Signal Model

Downsampling (decimation) by an integer factor \(L\) keeps one sample out of every \(L\), starting at the phase \(\tau\):

\[y[n] = \alpha \, x[n L + \tau]\]

where:

• \(x[n]\) is the input sequence of length \(N\),

• \(y[n]\) is the output sequence of length \(\lceil (N-\tau)/L \rceil\),

• \(L\) is the downsampling factor,

• \(\tau\) is the phase of the retained samples,

• \(\alpha\) is an output scaling factor.

Decimation folds the input spectrum \(L\) times; the input must therefore be bandlimited to \(1/L\) beforehand, otherwise aliasing occurs. Setting use_filter applies that anti-aliasing lowpass filter of normalized cutoff \(1/L\) before the decimation – the mirror image of Upsampler, where the anti-imaging filter of the same cutoff comes after the zero stuffing.

Axes: declared axis – the decimation runs along axis (default -1, the sample axis of the Serial layout (..., N)) and the remaining axes are left untouched. The anti-aliasing filter of use_filter=True always acts along the last axis, whatever axis says.

param L:

Downsampling factor \(L\). Must be a positive integer.

type L:

int

param phase:

Phase \(\tau \in \mathbb{N}\) of the retained samples. Default is 0.

type phase:

int, optional, keyword-only

param scale:

Output scaling factor \(\alpha\). Default is 1.0.

type scale:

float, optional, keyword-only

param axis:

Axis along which the downsampling is performed. Default is -1.

type axis:

int, optional, keyword-only

param use_filter:

If True, apply the anti-aliasing lowpass filter of normalized cutoff \(1/L\) (a comnumpy.core.filters.BWFilter) before the decimation. Default is False.

type use_filter:

bool, optional, keyword-only

param name:

Name of the processor instance. Default is "downsampler".

type name:

str, optional, keyword-only

References

A. V. Oppenheim, R. W. Schafer, Discrete-Time Signal Processing, 3rd ed., Pearson, 2010, Chapter 4 (changing the sampling rate by discrete-time processing).

Examples

>>> print(Downsampler(L=2)(np.arange(6)))
[0. 2. 4.]
>>> print(Downsampler(L=2, phase=1)(np.arange(6)))
[1. 3. 5.]
>>> print(Downsampler(L=2)(np.arange(12).reshape(2, 6)))
[[ 0.  2.  4.]
 [ 6.  8. 10.]]

Anti-aliasing: an in-band tone corrupted by an out-of-band one is decimated exactly with use_filter=True, and aliased without it. The cutoff is \(1/L = 1/4\) of the Nyquist frequency, i.e. 0.125 cycle per sample.

>>> n = np.arange(128)
>>> clean = np.cos(2 * np.pi * 2 * n / 128)             # f = 0.016, kept
>>> x = clean + 0.5 * np.cos(2 * np.pi * 30 * n / 128)  # f = 0.234, rejected
>>> y = Downsampler(L=4, use_filter=True)(x)
>>> print(round(float(np.max(np.abs(y - clean[::4]))), 6))
0.0
>>> print(round(float(np.max(np.abs(Downsampler(L=4)(x) - clean[::4]))), 3))
0.5

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

Converts a serial data stream into parallel blocks (Serial to Block layout).

Signal Model

The serial axis of length \(N\) is cut into \(T\) consecutive blocks of \(F\) samples. The sample of index \(n\) of the serial stream becomes the sample of index \(f\) of the block \(t\):

\[y[t, f] = x\!\left[t F + f\right], \qquad 0 \le t < T, \quad 0 \le f < F, \quad T = \left\lceil \frac{N}{F} \right\rceil \; \text{(zero-padding)}, \qquad T = \left\lfloor \frac{N}{F} \right\rfloor \; \text{(truncate)}\]

Equivalently, \(t = \lfloor n / F \rfloor\) and \(f = n \bmod F\): the serial index runs fastest on the last axis. This is exactly a C-order (row-major) reshape, which is the normative Serial \((..., N)\) to Block \((..., T, F)\) conversion of CONVENTIONS.md (decision D2). Any implementation requiring a transposition – for instance a Fortran-order reshape producing \((F, T)\) – violates that convention.

When \(N\) is not a multiple of \(F\), the serial axis is either zero-padded up to \(T F\) samples or truncated down to \(\lfloor N/F \rfloor F\) samples, depending on method.

Axes: declared axis – consumes the serial axis -1 of the Serial layout (..., N) and produces the Block layout (..., T, F); the leading axes are carried through unchanged.

param N_sub:

Block length \(F\) (the number of subcarriers in an OFDM chain), which becomes the size of the last axis of the output. Must be a positive integer.

type N_sub:

int

param method:

How to handle a serial length \(N\) that is not a multiple of \(F\): "zero-padding" appends zeros up to \(T F\) samples, "truncate" drops the trailing incomplete block. Default is "zero-padding".

type method:

{“zero-padding”, “truncate”}, optional, keyword-only

param name:

Name of the processor instance. Default is "S2P".

type name:

str, optional, keyword-only

References

R. van Nee, R. Prasad, OFDM for Wireless Multimedia Communications, Artech House, 2000, Chapter 2 (serial-to-parallel conversion).

Examples

>>> Y = Serial2Parallel(3)(np.arange(5))
>>> print(Y.shape)
(2, 3)
>>> print(Y)
[[0 1 2]
 [3 4 0]]
>>> print(Serial2Parallel(3, method="truncate")(np.arange(5)))
[[0 1 2]]
>>> Y = Serial2Parallel(3)(np.arange(10).reshape(2, 5))
>>> print(Y.shape)
(2, 2, 3)

class Parallel2Serial(name: str = 'P2S')

Converts parallel blocks back into a serial data stream (Block to Serial layout).

Signal Model

Inverse of Serial2Parallel: the \(T\) blocks of \(F\) samples are concatenated back into a single serial axis of length \(T F\),

\[y[t F + f] = x[t, f], \qquad 0 \le t < T, \quad 0 \le f < F\]

that is, the block content index \(f\) runs fastest. This is a pure C-order (row-major) reshape of the last two axes, the normative Block \((..., T, F)\) to Serial \((..., T F)\) conversion of CONVENTIONS.md (decision D2). No transposition is involved, so Parallel2Serial()(Serial2Parallel(F)(x)) returns x whenever \(F\) divides its length.

Axes: declared axis – consumes the Block layout (..., T, F) and produces the Serial layout (..., T*F); the leading axes are carried through unchanged.

param name:

Name of the processor instance. Default is "P2S".

type name:

str, optional

References

R. van Nee, R. Prasad, OFDM for Wireless Multimedia Communications, Artech House, 2000, Chapter 2 (parallel-to-serial conversion).

Examples

>>> print(Parallel2Serial()(np.array([[0, 1, 2], [3, 4, 5]])))
[0 1 2 3 4 5]
>>> Y = Parallel2Serial()(np.arange(12).reshape(2, 2, 3))
>>> print(Y.shape)
(2, 6)
>>> print(Y[0])
[0 1 2 3 4 5]
>>> x = np.arange(6)
>>> print(np.array_equal(Parallel2Serial()(Serial2Parallel(3)(x)), x))
True

class Amplifier(gain: float = 1.0, *, name: str = 'signal_amplifier')

Scales a signal by a constant gain (amplification or attenuation).

Signal Model

The amplifier is a memoryless linear device applying a constant gain \(g\) to every sample:

\[y[n] = g \, x[n]\]

where \(g > 1\) amplifies, \(g < 1\) attenuates, and the output power is \(g^2\) times the input power.

The gain is a single scalar applied to the whole array. To weight the parallel streams of a signal individually, i.e. one gain per entry of a declared axis, use WeightAmplifier.

Axes: element-wise – applied pointwise, shape-agnostic.

param gain:

Gain \(g\) applied to every sample. Default is 1.0.

type gain:

float

param name:

Name of the amplifier instance. Default is "signal_amplifier".

type name:

str, optional, keyword-only

References

S. C. Cripps, RF Power Amplifiers for Wireless Communications, 2nd ed., Artech House, 2006 (memoryless amplitude gain).

Examples

>>> print(Amplifier(gain=2)(np.array([[1, 2], [3, 4]])))
[[2 4]
 [6 8]]
>>> print(Amplifier(gain=0.5)(np.array([2.0, 4.0])))
[1. 2.]

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

Applies one complex weight per parallel stream along a declared axis.

Signal Model

Each stream \(l\) of a parallel (for instance multi-antenna) signal is scaled by its own weight \(w_l\):

\[y_l[n] = w_l \, x_l[n], \qquad 0 \le l < L\]

where:

• \(x_l[n]\) is the \(l\)-th input stream,

• \(w_l\) is the weight applied to that stream,

• \(L\) is the number of streams, i.e. the size of the input along axis.

This is the diagonal (per-stream) case of a linear precoder or of a one-tap equalizer; comnumpy.ofdm.compensators.FrequencyDomainEqualizer derives from it with \(w_k = 1/H[k]\).

Axes: declared axis – the weights are applied along axis (default -1); len(weight) must equal X.shape[axis], which is checked at every call.

param weight:

1D array of weights \(w_l\), of length \(L\).

type weight:

np.ndarray

param axis:

Axis carrying the parallel streams. Default is -1.

type axis:

int, optional, keyword-only

param name:

Name of the processor instance. Default is "parallel_signal_weight".

type name:

str, optional, keyword-only

raises ValueError:

If weight is not a 1D array, or if its length does not match the size of the input along axis.

References

A. Goldsmith, Wireless Communications, Cambridge University Press, 2005, Chapter 10 (multiple-antenna systems, transmit and receive weighting).

Examples

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

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

Extracts the real or the imaginary part of a complex signal.

Signal Model

Writing the input in Cartesian form \(x[n] = x_I[n] + i \, x_Q[n]\), the processor returns one of the two quadrature components:

\[\begin{split}y[n] = \left\{\begin{array}{cl} x_I[n] = \Re\{x[n]\} & \text{if } \texttt{part} = \text{"real"},\\ x_Q[n] = \Im\{x[n]\} & \text{if } \texttt{part} = \text{"imag"}. \end{array}\right.\end{split}\]

The output is real-valued. With validate_input=True the discarded component is checked to be numerically zero (tolerance \(10^{-7}\)), which turns the block into a guard: it then only changes the dtype of a signal that is already known to be real (or purely imaginary), and refuses to silently drop energy.

Axes: element-wise – applied pointwise, shape-agnostic.

param part:

Quadrature component to extract, \(x_I[n]\) or \(x_Q[n]\). Default is "real".

type part:

{“real”, “imag”}, optional

param validate_input:

If True, raise instead of discarding a non-zero component. Default is False.

type validate_input:

bool, optional, keyword-only

raises ValueError:

If validate_input is True and the discarded component is not numerically zero.

Examples

>>> X = np.array([1+2j, 3+4j, 5+0j])
>>> Complex2Real(part="real")(X)
array([1., 3., 5.])
>>> Complex2Real(part="imag")(X)
array([2., 4., 0.])
>>> Complex2Real(part="imag", validate_input=True)(X)
Traceback (most recent call last):
    ...
ValueError: the input data is not imaginary since the real 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 = -1, name: str = 'auto concatenator')

Copies part of a signal and reassembles original and copy through boolean masks.

Signal Model

Let \(m^{\text{in}}\), \(m^{\text{orig}}\) and \(m^{\text{copy}}\) be the three boolean masks, of lengths \(N\) (input) and \(M\) (output). Writing \(\mathcal{I} = \{ i : m^{\text{in}}_i \}\), \(\mathcal{O} = \{ i : m^{\text{orig}}_i \}\) and \(\mathcal{C} = \{ i : m^{\text{copy}}_i \}\) for the sorted positions selected by each mask, the output is

\[y\!\left[\mathcal{O}_j\right] = x[j], \quad 0 \le j < N, \qquad y\!\left[\mathcal{C}_j\right] = g\!\left(x\!\left[\mathcal{I}_j\right]\right), \quad 0 \le j < |\mathcal{I}|,\]

and \(y[i] = 0\) for every position covered by neither mask. The map \(g(\cdot)\) is the hook process_copy (identity by default), which subclasses override to transform the copy.

Consistency requires \(|\mathcal{O}| = N\), \(|\mathcal{I}| = |\mathcal{C}|\), and \(\mathcal{O} \cap \mathcal{C} = \varnothing\); all three are checked. Prepending a copy of the tail of each block is exactly the cyclic prefix, which is why comnumpy.ofdm.processors.CyclicPrefixer derives from this class and only builds the three masks.

Axes: declared axis – the masks index axis (default -1); len(input_copy_mask) must equal X.shape[axis], and the output has size len(output_original_mask) along that axis.

param input_copy_mask:

Mask \(m^{\text{in}}\) selecting, in the input, the entries to be copied. Length X.shape[axis].

type input_copy_mask:

np.ndarray of bool, optional, keyword-only

param output_original_mask:

Mask \(m^{\text{orig}}\) giving, in the output, the positions of the original data. Must select exactly X.shape[axis] entries.

type output_original_mask:

np.ndarray of bool, optional, keyword-only

param output_copy_mask:

Mask \(m^{\text{copy}}\) giving, in the output, the positions of the copied data. Same length as output_original_mask, and must select as many entries as input_copy_mask.

type output_copy_mask:

np.ndarray of bool, optional, keyword-only

param axis:

Axis along which the concatenation is performed. Default is -1.

type axis:

int, optional, keyword-only

param name:

Name of the processor instance. Default is "auto concatenator".

type name:

str, optional, keyword-only

raises ValueError:

If the two output masks have different shapes, select a different number of entries than expected, or overlap; or if the input size along axis does not match input_copy_mask.

Examples

>>> concatenator = AutoConcatenator(
...     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]))
>>> print(concatenator(np.array([1, 2, 3])))
[1 2 3 1 3]
>>> print(concatenator(np.array([[1, 2, 3], [4, 5, 6]])))
[[1 2 3 1 3]
 [4 5 6 4 6]]

extract_copy(X: ndarray) → 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 a contiguous run of samples from a serial signal.

Signal Model

A window of \(D\) samples starting at index \(n_0\) is cut out of the input, and the two remaining segments are spliced together:

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

where \(x[n]\) has length \(N\), \(n_0\) is the index of the first deleted sample, \(D\) the number of deleted samples, and \(y[n]\) has length \(N - D\). It is the exact inverse of DataAdder when the same \(n_0\) is used.

Axes: 1D only – operates on a serial signal (N,); the splice uses flat indexing and a batch is refused with ShapeError.

param N_start:

Index \(n_0\) of the first sample to delete. Default is 0.

type N_start:

int

param length:

Number \(D\) of samples to delete. Default is 0.

type length:

int, optional, keyword-only

param name:

Name of the processor instance. Default is "symbol remover".

type name:

str, optional, keyword-only

raises comnumpy.ShapeError:

If the input is not a 1D Serial signal (N,).

Examples

>>> print(SampleRemover(N_start=2, length=3)(np.arange(8)))
[0 1 5 6 7]
>>> print(SampleRemover(N_start=0, length=2)(np.arange(5)))
[2 3 4]
>>> SampleRemover(N_start=1, length=1)(np.arange(6).reshape(2, 3))
Traceback (most recent call last):
    ...
comnumpy.exceptions.ShapeError: SampleRemover expects a 1D Serial signal (N,), got 2D (2, 3) -- ...

class DelayRemover(delay: int, *, axis: int = -1, name: str = 'delay remover')

Removes the first samples of a signal to compensate a known delay.

Signal Model

A pure advance by the known integer delay \(d\), i.e. a shift that discards the transient introduced upstream (filter group delay, channel delay):

\[y[n] = x[n + d], \qquad 0 \le n < N - d\]

where \(x[n]\) has length \(N\), \(d\) is the delay to remove, and \(y[n]\) has length \(N - d\).

Axes: declared axis – the samples are dropped along axis (default -1, the sample axis of the Serial layout (..., N)); the remaining axes are untouched.

param delay:

Number \(d\) of leading samples to remove.

type delay:

int

param axis:

Axis along which the delay is removed. Default is -1.

type axis:

int, optional, keyword-only

param name:

Name of the processor instance. Default is "delay remover".

type name:

str, optional, keyword-only

Examples

>>> print(DelayRemover(3)(np.arange(8)))
[3 4 5 6 7]
>>> print(DelayRemover(1)(np.array([[1, 2, 3], [4, 5, 6]])))
[[2 3]
 [5 6]]

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

Inserts a known sequence into a serial signal at a given index.

Signal Model

A sequence \(s[m]\) of length \(D\) (a preamble, a pilot block, a guard sequence) is spliced into the input at index \(n_0\):

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

where \(x[n]\) has length \(N\) and \(y[n]\) has length \(N + D\). Applying SampleRemover with the same \(n_0\) and \(D\) restores \(x[n]\).

Axes: 1D only – operates on a serial signal (N,); the splice is a concatenation, and a batch is refused with ShapeError.

param symbol:

Sequence \(s[m]\) to insert, of length \(D\).

type symbol:

np.ndarray

param N_start:

Insertion index \(n_0\), in [0, N]. Default is 0.

type N_start:

int, optional, keyword-only

param name:

Name of the processor instance. Default is "Data Adder".

type name:

str, optional, keyword-only

raises comnumpy.ShapeError:

If the input, or symbol, is not a 1D Serial signal (N,).

raises ValueError:

If N_start is outside [0, len(x)].

Examples

>>> print(DataAdder(np.array([-1, -1]), N_start=2)(np.arange(5)))
[ 0  1 -1 -1  2  3  4]
>>> print(DataAdder(np.array([-1, -1]))(np.arange(3)))
[-1 -1  0  1  2]
>>> DataAdder(np.array([-1, -1]))(np.arange(6).reshape(2, 3))
Traceback (most recent call last):
    ...
comnumpy.exceptions.ShapeError: DataAdder expects a 1D Serial signal (N,), got 2D (2, 3) -- ...

class DataExtractor(selector: int | slice | tuple[int, ...] | list[int] | ndarray | None = None, *, name: str = 'Data Extractor')

Extracts a segment of a signal using NumPy-style indexing.

Signal Model

The block keeps the samples whose index belongs to the selection \(\mathcal{S}\) described by selector:

\[y[m] = x\!\left[\mathcal{S}_m\right], \qquad 0 \le m < |\mathcal{S}|\]

With selector = (n_0, n_1, \Delta) the selection is the arithmetic progression \(\mathcal{S} = \{ n_0, n_0 + \Delta, \dots \} \cap [n_0, n_1)\), so \(n_0\) is the start index, \(n_1\) the (excluded) stop index and \(\Delta\) the step. An explicit list or array gives \(\mathcal{S}\) directly, an integer selects a single sample, and None makes the block the identity \(y[n] = x[n]\).

Axes: declared axis – the selector indexes the first axis, which is the sample axis of a 1D Serial signal (N,); on a multidimensional input it selects entries of axis 0 and the trailing axes are carried through.

param selector:

Description of the selection \(\mathcal{S}\): an int for a single index, a tuple (start, stop[, step]) converted to a slice, a slice, a list or array of explicit indices, or None for a pass-through. Default is None.

type selector:

int, slice, tuple, list, np.ndarray or None, optional

param name:

Name of the processor instance. Default is "Data Extractor".

type name:

str, optional, keyword-only

Examples

>>> x = np.arange(10)
>>> print(DataExtractor(3)(x))
3
>>> DataExtractor((2, 8))(x)
array([2, 3, 4, 5, 6, 7])
>>> DataExtractor((1, 9, 2))(x)
array([1, 3, 5, 7])
>>> DataExtractor((1, 3))(np.arange(20).reshape(4, 5))
array([[ 5,  6,  7,  8,  9],
       [10, 11, 12, 13, 14]])

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

Changes the sampling rate of a signal by a rational factor.

Signal Model

The rate is changed by the rational factor \(P/Q\): the signal is upsampled by \(P\), lowpass filtered, then decimated by \(Q\). Writing \(h[k]\) for the anti-imaging / anti-aliasing filter,

\[\begin{split}y[n] = \sum_{k} h[k] \; v\!\left[n Q - k\right], \qquad v[m] = \left\{\begin{array}{cl} x[m / P] & \text{if } m \bmod P = 0,\\ 0 & \text{otherwise,} \end{array}\right.\end{split}\]

so that an input of \(N\) samples yields \(\lceil N P / Q \rceil\) output samples and, for a signal bandlimited below \(\min(1/P, 1/Q)\), \(y[n] \approx x_c\!\left(n Q / P\right)\) where \(x_c\) is the underlying continuous-time waveform. The three stages are evaluated jointly by a polyphase implementation (scipy.signal.resample_poly), which never computes the discarded samples.

Axes: axis -1 – the resampling runs along the last axis, the sample axis of the Serial layout (..., N); leading axes are batch, every row resampled by the same factors.

param up:

Interpolation factor \(P\). Must be a positive integer.

type up:

int

param down:

Decimation factor \(Q\). Must be a positive integer.

type down:

int

param name:

Name of the processor instance. Default is "Resampler".

type name:

str, optional, keyword-only

raises ValueError:

If up or down is not a positive integer.

References

A. V. Oppenheim, R. W. Schafer, Discrete-Time Signal Processing, 3rd ed., Pearson, 2010, Chapter 4 (changing the sampling rate by a non-integer factor, polyphase implementation).

Examples

>>> x = np.cos(2 * np.pi * 0.02 * np.arange(48))
>>> y = Resampler(up=3, down=2)(x)
>>> print(y.shape)
(72,)
>>> m = np.arange(20, 52)
>>> ideal = np.cos(2 * np.pi * 0.02 * (2 / 3) * m)
>>> print(round(float(np.max(np.abs(y[20:52] - ideal))), 4))
0.0006

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

Saturates a real signal to a symmetric amplitude interval.

Signal Model

The clipper is a memoryless nonlinearity: it is transparent below the threshold \(\tau\) and saturates above it,

\[\begin{split}y[n] = \max\left(-\tau, \min\left(x[n], \tau\right)\right) = \left\{\begin{array}{cl} x[n] & \text{if } |x[n]| \le \tau,\\ \tau \, \mathrm{sign}(x[n]) & \text{if } |x[n]| > \tau, \end{array}\right.\end{split}\]

where \(\tau > 0\) is the saturation threshold. Clipping reduces the peak-to-average power ratio at the cost of in-band distortion and spectral regrowth.

Note

The interval \([-\tau, \tau]\) is a real interval: this block is meant for real-valued signals. For the polar clipping \(\tau e^{i \angle x[n]}\) of a complex baseband envelope, use comnumpy.ofdm.predistorders.HardClipper.

Axes: element-wise – applied pointwise, shape-agnostic.

param threshold:

Saturation threshold \(\tau\).

type threshold:

float

param name:

Name of the processor instance. Default is "Clipper".

type name:

str, optional, keyword-only

References

S. C. Cripps, RF Power Amplifiers for Wireless Communications, 2nd ed., Artech House, 2006 (memoryless saturating nonlinearity).

Examples

>>> print(Clipper(threshold=1.5)(np.array([-3.0, -1.0, 0.5, 2.0])))
[-1.5 -1.   0.5  1.5]
>>> print(Clipper(threshold=1.0)(np.array([[0.2, 5.0], [-5.0, -0.2]])))
[[ 0.2  1. ]
 [-1.  -0.2]]

class BlindPhaseTracker(L: int, alphabet: ndarray, phase_steps: int = 10)

Blind carrier phase recovery by exhaustive phase search (BPS).

Signal Model

The input carries an unknown, slowly varying carrier phase \(\theta[n]\):

\[x[n] = s[n] \, e^{i \theta[n]} + b[n]\]

where \(s[n]\) belongs to the known alphabet \(\mathcal{A}\). For each sample, the \(B\) test phases \(\varphi_p = \frac{\pi}{4}\left(\frac{2p}{B} - 1\right)\), \(p = 0, \dots, B-1\), are applied, and the one minimizing the local squared error over a sliding window of \(2L + 1\) samples is retained:

\[\hat{\theta}[n] = \arg\min_{\varphi_p} \frac{1}{2L+1} \sum_{m=-L}^{L} \left| x[n+m] e^{-i \varphi_p} - \mathcal{D}\!\left(x[n+m] e^{-i \varphi_p}\right) \right|^2\]

where \(\mathcal{D}(z) = \arg\min_{a \in \mathcal{A}} |z - a|^2\) is the hard decision on the alphabet. The compensated output is

\[y[n] = x[n] \, e^{-i \hat{\theta}[n]}\]

The search range is limited to \([-\pi/4, \pi/4)\), the phase ambiguity of a square QAM constellation; residual \(\pi/2\) cycle slips must be resolved elsewhere (differential encoding or pilots). The window half-length \(L\) sets the usual trade-off: a long window averages the noise, a short one follows a fast phase.

The estimand is per path (D49): a (..., P, N) signal gets one phase trajectory per path, and theta_ has the shape of the input. After a butterfly equalizer each output carries its own residual rotation, so tracking the paths jointly would fit a single phase to two different ones.

The search range is the ambiguity of a square QAM. On a constellation of finer rotational symmetry – 8-PSK repeats every \(\pi/4\), so the range covers two of its periods – several test phases give mathematically identical costs and which one wins is arbitrary. That is a property of the criterion, not of the search.

Axes: axis -1 – one independent trajectory per leading-axis row (D49): a (..., P, N) signal gets one phase estimate per path. The decision of a sample under a test phase does not depend on any other sample, so the whole search is a table of decision errors plus a moving sum over it, and nothing is done in order.

param L:

Half-length \(L\) of the sliding window; the local cost is averaged over \(2L + 1\) samples.

type L:

int

param alphabet:

Constellation \(\mathcal{A}\) used by the hard decision \(\mathcal{D}(\cdot)\).

type alphabet:

np.ndarray

param phase_steps:

Number \(B\) of test phases uniformly spanning \([-\pi/4, \pi/4)\). Default is 10.

type phase_steps:

int, optional

phases

The \(B\) test phases \(\varphi_p\), precomputed from phase_steps alone (parameter-derived, hence no trailing underscore).

Type:

np.ndarray

theta_

The estimated phase trajectory \(\hat{\theta}[n]\) of the last call, of the same shape as the input (data-derived, hence the trailing underscore, D23). A block never draws (D25, D42): plot it from the caller, for instance with matplotlib.pyplot.plot(tracker.theta_).

Type:

np.ndarray

References

T. Pfau, S. Hoffmann, R. Noe, “Hardware-Efficient Coherent Digital Receiver Concept With Feedforward Carrier Recovery for M-QAM Constellations”, Journal of Lightwave Technology, vol. 27, no. 8, pp. 989-999, 2009, doi: 10.1109/JLT.2008.2010511.

Examples

>>> alphabet = np.array([1+1j, -1+1j, -1-1j, 1-1j]) / np.sqrt(2)
>>> rng = np.random.default_rng(42)
>>> s = alphabet[rng.integers(0, 4, size=12)]
>>> x = s * np.exp(1j * 0.3)
>>> tracker = BlindPhaseTracker(L=3, alphabet=alphabet, phase_steps=16)
>>> y = tracker(x)
>>> print(round(float(np.max(np.abs(y - s))), 3))
0.005
>>> print(tracker.theta_.shape, round(float(np.mean(tracker.theta_)), 3))
(12,) 0.295