Compensators#

class TrainedBasedMixin#
get_target_data()#

Retrieve the target data associated with the model.

Returns#

np.ndarray

The target data array.

Raises#

TypeError

If target_data is neither a numpy array nor an instance of Recorder.

class DCCorrector(value: float = 0.0, axis: int = 0, name: str = 'mean_corrector')#

A class for correcting the mean of a dataset along a specified axis.

This class adjusts the input data so that its mean along the specified axis matches a target value. This is useful for normalizing data or removing bias.

Signal Model#

\[y[n] = x[n] + \alpha\]

where the coefficient \(\alpha\) is adjusted to meet the DC constraint

Attributes#

valuefloat

The target mean value to which the data should be adjusted (default: 0)

axisint

The axis along which to compute the mean.

namestr

Name of the mean corrector instance.

class Normalizer(gain: float = 1, axis: int | None = None, name: str = 'signal_amplifier', method: Literal['amp', 'abs', 'var', 'max'] = 'amp', value: float = 1.0)#

A class for normalizing data based on the specified normalization type.

Signal Model#

\[y[n] = \alpha x[n]\]

where the normalization coefficient \(\alpha\) depends on the normalization technique.

  • ‘amp’: Scales the signal by a constant factor \(\alpha = \text{value}\).

    \[\alpha = \text{value}\]
  • ‘abs’: Normalizes the signal by the maximum absolute value.

    \[\alpha = \frac{\text{value}}{\max(|x[n]|)}\]
  • ‘var’: Normalizes the signal based on its variance.

    \[\alpha = \sqrt{\frac{\text{value}}{\sigma_x^2}}\]
  • ‘max’: Normalizes the signal by the maximum absolute value of the real and imaginary parts.

    \[\alpha = \frac{\text{value}}{\max(\max(|\text{Re}(x[n])|), \max(|\text{Im}(x[n])|))}\]

Attributes#

methodstr

Type of normalization to be applied. Supported types are ‘amp’ for scaling coefficient, ‘max’ for maximum value normalization, ‘var’ for variance-based normalization, and ‘abs’ for absolute maximum value normalization.

valuefloat, optional

The target value for the normalization type. Default is 1.0.

Example#

>>> normalizer = Normalizer(method='max', value=2.0)
>>> X = np.array([1, 2, 3, 4])
>>> Y = normalizer(X)
>>> print(Y)
[0.5 1.  1.5 2. ]
class BlindIQCompensator(should_fit: bool = True, coef: float = 1, name: str = 'iq_compensator')#

Blind IQ compensator based on the diagonalisation of the augmented covariance matrix. This compensation assumes the circularity of compensated signal

Signal Model#

This class implements the following transformation

\[y[n] = \alpha \Re e(x[n]) + \beta \Im m(x[n])\]

Algorithm#

such as :

\[\begin{split}E[\Re e^2(y[n])] &= E[\Im m^2(y[n])] = 1\\ E[\Re e(y[n])\Im m(y[n])] & = 0\end{split}\]
class BlindCFOCompensator(w0_init: float = 0.0, N_iter: int = 3, should_fit: bool = True, grid_search: bool = True, save_history: bool = False, method: Literal['grad', 'newton'] = 'newton', step_size: float = 1e-08, grid_search_tuple: tuple = (-0.1, 0.1, 0.0001), name: str = 'cfo_compensator')#

Blind CFO compensator based on the maximisation of the periodogram of 4th order statistic.

Signal Model#

\[y[n] = x[n]e^{-j\widehat{\omega}_0 n}\]

Algorithm#

\[\widehat{\omega}_0 = \frac{1}{4} \arg \max_{\omega} \left|\sum_{n=0}^{N-1} x^4[n]e^{-j\omega n}\right|^2\]

The maximisation is performed using the Newton Algorithm

Attributes#

w0_initfloat

Initialisation in rad/samples

N_iterint

Number of iterations

methodstr

method used for maximisation

class BlindPhaseCompensation(alphabet: ndarray, theta: float = 0.0, should_fit: bool = True, name: str = 'phase correction')#

Blind phase compensator based on least-squares minimization of the EVM.

Signal Model#

\[y[n] = x[n] e^{j\hat{\theta}}\]

where \(\hat{\theta}\) is the estimated phase offset that minimizes the distance to the nearest constellation point.

Attributes#

alphabetnp.ndarray

The modulation alphabet used for phase compensation.

thetafloat, optional

Initial phase angle in radians. Default is 0.

should_fitbool, optional

Whether to estimate the phase from the input signal. Default is True.

namestr, optional

Name of the processor. Default is "phase correction".

class LinearEqualizer(h: ndarray, method: Literal['zf', 'mmse'] = 'zf', sigma2: float = 0.0, name: str = 'equalizer')#

Linear equalizer for the signal model with inter-symbol interference (ISI).

Signal Model#

\[z[n] = \sum_{l=0}^{L-1} h[l] x[n-l] + b[n]\]

The equalizer constructs a Toeplitz channel matrix from the impulse response and applies ZF or MMSE equalization.

Attributes#

hnp.ndarray

Channel impulse response.

methodLiteral[“zf”, “mmse”], optional

Equalization method. Default is "zf".

sigma2float, optional

Noise variance (used only for MMSE). Default is 0.

namestr, optional

Name of the equalizer instance. Default is "equalizer".

class DataAidedFIRCompensator(h: array, should_fit: bool = True, name: str = 'data_aided_fir')#

Data-aided FIR compensator using Zero Forcing estimation.

Estimates the channel impulse response from a known reference (target data) and applies deconvolution to equalize the received signal.

Attributes#

hnp.ndarray

Initial or estimated impulse response.

target_datanp.ndarray or Recorder

Known reference data for channel estimation.

should_fitbool, optional

Whether to estimate the channel on each call. Default is True.

namestr, optional

Name of the processor. Default is "data_aided_fir".

class TrainedBasedPhaseCompensator(target_data: array | Recorder, name: str = 'data_aided_phase')#

Trained-based phase compensator using cross-correlation.

Estimates a phase offset \(\theta\) from the cross-correlation between the input and reference signals, then applies a correction.

\[y[n] = x[n] e^{j\hat{\theta}}\]

Attributes#

target_datanp.ndarray or Recorder

Reference signal for phase estimation.

namestr, optional

Name of the processor. Default is "data_aided_phase".

class TrainedBasedComplexGainCompensator(extractor: Field(name = None, type=None, default=<dataclasses._MISSING_TYPE object at 0x7f3d8b312660>, default_factory=<function TrainedBasedComplexGainCompensator.<lambda> at 0x7f3d60605c70>, init=True, repr=True, hash=None, compare=True, metadata=mappingproxy({}), kw_only=<dataclasses._MISSING_TYPE object at 0x7f3d8b312660>, doc=None, _field_type=None), gain: complex = (1+0j), should_fit: bool = True, name: str = 'complex_gain_compensator')#

This class performs the complex-gain channel estimation and compensation

Attributes#

recorder_preamblendarray

The recorded preamble used to compute the complex gain of the channel.

extractorData Extractor

How to extract the data in the received samples.

class TrainedBasedSimpleSynchronizer(scale_correction: bool = True, save_cross_correlation: bool = True, signal_len: int | None = None, name: str = 'synchronizer')#

Implements a simple synchronizer using cross-correlation to determine time delay and scaling between signals.

Attributes#

target_datandarray

The reference preamble signal to which the input signals will be synchronized.

scale_correctionbool, optional

If True, applies a scaling correction based on the peak of the cross-correlation. Default is True.

save_cross_corrbool, optional

If True, saves the computed cross-correlation and the associated lag vector. Default is True.

signal_leninteger, optional

Truncates the signal to the given length after synchronization

namestr, optional

Name of the synchronizer instance. Default is “synchronizer”.

class TrainedBasedFineSynchronizer(scale_correction: bool = True, save_cross_correlation: bool = True, signal_len: int | None = None, d_max: int | None = None, name: str = 'synchronizer')#

Implements a simple synchronizer using cross-correlation to determine time delay and scaling between signals.

Attributes#

target_datandarray

The reference preamble signal to which the input signals will be synchronized.

up_factorint

The upsampling factor made before cross-correlation

scale_correctionbool, optional

If True, applies a scaling correction based on the peak of the cross-correlation. Default is True.

save_cross_corrbool, optional

If True, saves the computed cross-correlation and the associated lag vector. Default is True.

signal_leninteger, optional

Truncates the signal to the given length after synchronization

d_max :integer, optional

The maximum expected delay [number of samples of x]

namestr, optional

Name of the synchronizer instance. Default is “synchronizer”.