dpd¶

This module implements Digital Pre-Distortion (DPD) techniques for power amplifier linearization.

Configuration¶

class demos.dpd.src.config.Config(seed=42, batch_size=100)[source]¶

Bases: object

Central configuration for the DPD system.

This dataclass enforces a separation between user-configurable parameters (seed, batch_size) and system constants (RF, OFDM, modulation settings). Immutable parameters use underscore-prefixed private fields with read-only properties to prevent accidental modification during experiments.

Parameters:

seed (int) – Random seed for reproducibility. Defaults to 42.
batch_size (int) – Number of OFDM frames per batch. Defaults to 100.

Notes

Design Rationale:

Immutable RF parameters prevent mid-experiment changes that would invalidate DPD coefficient learning.
Guard carrier settings (200, 199) are asymmetric to account for DC null, yielding 624 usable subcarriers.

miscellaneous:

signal_sample_rate = fft_size * subcarrier_spacing
Total subcarriers = fft_size - num_guard_carriers[0]
- num_guard_carriers[1] - dc_null

Example

>>> config = Config(seed=123, batch_size=32)
>>> config.signal_sample_rate
15360000.0
>>> config.fft_size  # Immutable
1024

seed: int = 42¶

batch_size: int = 100¶

property num_ut: int¶

Number of user terminals.

Type:: int

property num_ut_ant: int¶

Number of antennas per user terminal.

Type:: int

property num_streams_per_tx: int¶

Number of spatial streams per transmitter.

Type:: int

property num_ofdm_symbols: int¶

Number of OFDM symbols per frame.

Type:: int

property fft_size: int¶

FFT size for OFDM modulation.

Type:: int

property subcarrier_spacing: float¶

Subcarrier spacing in Hz.

Type:: float

property num_guard_carriers: Tuple[int, int]¶

Number of (lower, upper) guard carriers.

Type:: tuple of int

property dc_null: bool¶

Whether the DC subcarrier is nulled.

Type:: bool

property cyclic_prefix_length: int¶

Cyclic prefix length in samples.

Type:: int

property pilot_pattern: str¶

Pilot pattern type for channel estimation.

Type:: str

property pilot_ofdm_symbol_indices: List[int]¶

OFDM symbol indices containing pilot symbols.

Type:: list of int

property num_bits_per_symbol: int¶

Number of bits per modulation symbol (4 = 16-QAM).

Type:: int

property coderate: float¶

Forward error correction code rate.

Type:: float

property signal_sample_rate: float¶

Baseband signal sample rate in Hz.

Returns:: Sample rate computed as fft_size * subcarrier_spacing. With defaults: 1024 * 15000 = 15.36 MHz.
Return type:: float

Notes

This is the native OFDM sample rate before any upsampling to the PA sample rate. The interpolator handles rate conversion between this rate and the PA’s operating rate of 122.88 MHz.

Transmitter¶

class demos.dpd.src.tx.Tx(*args, **kwargs)[source]¶

Bases: Model

5G NR-style OFDM transmitter for DPD design and evaluation.

Implements a complete transmit chain from random bit generation through OFDM modulation. The output is a baseband time-domain signal ready for upsampling and PA transmission.

Parameters:: config (Config) – Configuration object containing all RF and OFDM parameters.

rg¶

OFDM resource grid defining time-frequency structure.

Type:: sionna.ofdm.ResourceGrid

n¶

LDPC codeword length (coded bits per stream per slot).

Type:: int

k¶

LDPC information bits per stream per slot.

Type:: int

encoder¶

5G NR LDPC encoder.

Type:: LDPC5GEncoder

mapper¶

QAM constellation mapper.

Type:: Mapper

rg_mapper¶

Maps QAM symbols to resource grid locations.

Type:: ResourceGridMapper

ofdm_modulator¶

Converts frequency-domain grid to time-domain signal.

Type:: OFDMModulator

Notes

Signal Flow:

BinarySource generates uniform random bits
LDPC5GEncoder adds redundancy for error correction
Mapper converts bit groups to complex QAM symbols
ResourceGridMapper assigns symbols to subcarriers
OFDMModulator performs IFFT and adds cyclic prefix

LDPC Code Design:

The code dimensions are derived from the resource grid capacity:

n = (number of data symbols) * (bits per symbol)
k = n * code_rate

The 5G NR LDPC encoder automatically selects an appropriate base graph and lifting size for the given (k, n) pair.

miscellaneous:

Config must define valid OFDM parameters
Code rate must yield 0 < k < n
Output x_time has shape [batch, 1, 1, num_time_samples]
Output is complex64 baseband signal normalized for PA input

Example

>>> config = Config()
>>> tx = Tx(config)
>>> outputs = tx(batch_size=16)
>>> outputs["x_time"].shape
TensorShape([16, 1, 1, ...])

call(batch_size)[source]¶

Generate a batch of OFDM transmit waveforms.

Generates random information bits and processes them through the complete transmit chain. All intermediate results are returned to support DPD training which needs access to original bits for BER evaluation.

Parameters:

batch_size (tf.Tensor or int) – Number of independent OFDM signal batches to generate.

Returns:

Dictionary containing all transmit chain outputs:

bitstf.Tensor, shape [B, num_tx, num_streams, k]
Original information bits (for BER calculation).
codewordstf.Tensor, shape [B, num_tx, num_streams, n]
LDPC encoded bits.
x_rgtf.Tensor, shape [B, num_tx, num_streams, num_ofdm_symbols, fft_size]
Frequency-domain resource grid (complex symbols).
x_timetf.Tensor, shape [B, num_tx, num_streams, num_time_samples]
Time-domain baseband signal after OFDM modulation.

Return type:

dict

Notes

The returned bits tensor is essential for end-to-end BER measurement. After the signal passes through DPD and PA, decoded bits are compared against these original bits.

Receiver¶

class demos.dpd.src.rx.Rx(*args, **kwargs)[source]¶

Bases: Layer

Minimal OFDM receiver chain for DPD performance evaluation.

Implements the full receive path from PA output to equalized QAM symbols, with EVM computation for quantifying signal distortion. This receiver is designed for DPD evaluation, not for conducting end-to-end communication link performance evaluation.

Parameters:

signal_fs (float) – Baseband signal sample rate in Hz (e.g., 15.36 MHz for 15 kHz subcarrier spacing with 1024-point FFT).
pa_sample_rate (float) – PA operating sample rate in Hz (typically 8x signal rate for adequate reconstruction of PA nonlinear products).
fft_size (int) – OFDM FFT size (number of subcarriers including guards).
cp_length (int) – Cyclic prefix length in samples.
num_ofdm_symbols (int) – Number of OFDM symbols per slot.
num_guard_lower (int) – Number of guard subcarriers at lower band edge.
num_guard_upper (int) – Number of guard subcarriers at upper band edge.
dc_null (bool) – Whether the DC subcarrier is nulled.
**kwargs – Additional keyword arguments passed to Keras Layer.

_ofdm_demod¶

Sionna OFDM demodulator (FFT + CP removal).

Type:: OFDMDemodulator

_lower_start, _lower_end

Subcarrier indices for lower data band.

Type:: int

_upper_start, _upper_end

Subcarrier indices for upper data band.

Type:: int

Notes

Receiver Processing Steps:

Downsample: Convert from PA rate to baseband rate
Time sync: Cross-correlation to find frame boundary
OFDM demod: Remove CP and apply FFT
Equalize: Zero-forcing per-subcarrier equalization
EVM: Compute error vector magnitude vs. reference

Why Zero-Forcing Equalization?

In this loopback test scenario, the channel is essentially flat (no multipath). ZF equalization corrects only for the PA’s linear gain and phase offset. More sophisticated equalizers (MMSE, etc.) are unnecessary and would complicate DPD performance attribution.

miscellaneous:

PA output signals must be at pa_sample_rate
Reference baseband signal must be at signal_fs
All signals must represent the same transmitted frame
Equalized symbols are normalized to reference constellation
EVM is returned as percentage (0-100+ scale)

Example

>>> rx = Rx(
...     signal_fs=15.36e6,
...     pa_sample_rate=122.88e6,
...     fft_size=1024,
...     cp_length=72,
...     num_ofdm_symbols=14,
...     num_guard_lower=200,
...     num_guard_upper=199,
...     dc_null=True,
... )
>>> results = rx.process_and_compute_evm(
...     pa_input, pa_output_no_dpd, pa_output_with_dpd,
...     tx_baseband, fd_symbols
... )
>>> print(f"EVM with DPD: {results['evm_with_dpd']:.2f}%")

process_and_compute_evm(pa_input, pa_output_no_dpd, pa_output_with_dpd, tx_baseband, fd_symbols)[source]¶

Process PA outputs and compute EVM for DPD performance comparison.

Runs three signal paths (PA input, PA output without DPD, PA output with DPD) through the complete receiver chain and computes EVM for each. This enables direct comparison of DPD effectiveness.

Parameters:

pa_input (tf.Tensor or np.ndarray) – PA input signal at PA sample rate (reference for best-case EVM).
pa_output_no_dpd (tf.Tensor or np.ndarray) – PA output without predistortion at PA sample rate.
pa_output_with_dpd (tf.Tensor or np.ndarray) – PA output with predistortion at PA sample rate.
tx_baseband (tf.Tensor or np.ndarray) – Original baseband transmit signal at signal sample rate (used as timing reference for synchronization).
fd_symbols (tf.Tensor or np.ndarray) – Transmitted frequency-domain symbols, shape [num_data_subcarriers, num_ofdm_symbols]. Used as reference for equalization and EVM calculation.

Returns:

Results dictionary containing:

symbols_inputnp.ndarray
Equalized constellation symbols from PA input path.
symbols_no_dpdnp.ndarray
Equalized symbols from PA output without DPD.
symbols_with_dpdnp.ndarray
Equalized symbols from PA output with DPD.
evm_inputfloat
EVM (%) for PA input (baseline, should be near-zero).
evm_no_dpdfloat
EVM (%) for PA output without DPD (shows PA distortion).
evm_with_dpdfloat
EVM (%) for PA output with DPD (shows DPD improvement).

Return type:

dict

Notes

The PA input path serves as a sanity check: its EVM should be very low since it hasn’t passed through PA nonlinearity.

Power Amplifier¶

class demos.dpd.src.power_amplifier.PowerAmplifier(*args, **kwargs)[source]¶

Bases: Layer

Memory polynomial power amplifier model.

Implements a memory polynomial (MP) PA model with fixed coefficients derived from WARP board measurements. The model captures amplitude- dependent gain compression (AM/AM) and phase distortion (AM/PM), as well as memory effects from previous samples.

The PA transfer function is:

\[y[n] = \sum_{k \in \{1,3,5,7\}} \sum_{m=0}^{3} a_{k,m} \cdot x[n-m] \cdot |x[n-m]|^{k-1}\]

where \(a_{k,m}\) are complex coefficients, \(k\) is the polynomial order (odd-only to model symmetric nonlinearity), and \(m\) is the memory tap index.

Parameters:: **kwargs – Additional keyword arguments passed to the Keras Layer base class.

Notes

miscellaneous:

Input signal should be normalized to appropriate power level (the model expects inputs with RMS around 0.1-1.0)
Output has same shape as input
Output exhibits AM/AM compression and AM/PM distortion

Example

>>> pa = PowerAmplifier()
>>> x = tf.complex(tf.random.normal([16, 1024]), tf.random.normal([16, 1024]))
>>> x = x * 0.3  # Scale to reasonable PA input level
>>> y = pa(x)
>>> y.shape
TensorShape([16, 1024])

property order: int¶

Polynomial order (fixed at 7, odd-only).

Type:: int

property memory_depth: int¶

Memory depth in samples (fixed at 4).

Type:: int

call(x)[source]¶

Apply PA distortion to input signal.

Parameters:: x (tf.Tensor) – Input signal with shape [..., num_samples] and complex dtype. Supports arbitrary batch dimensions.
Returns:: Distorted output signal with same shape as input.
Return type:: tf.Tensor

Notes

The computation is structured as a matrix-vector product for efficiency: y = X @ coeffs where X is the basis matrix containing all polynomial and memory terms, and coeffs is the flattened coefficient vector.

estimate_gain(num_samples=10000)[source]¶

Estimate small-signal (linear) gain of the PA.

Uses a low-amplitude test signal to measure gain in the linear region where higher-order polynomial terms are negligible. This gain value is needed for proper normalization in the indirect learning DPD architecture.

Parameters:: num_samples (int) – Number of random samples for gain estimation. More samples reduce variance. Defaults to 10000.
Returns:: Estimated voltage gain (linear scale, not dB). Typically close to the magnitude of the first-order, zero-memory coefficient (~0.93 for default coefficients).
Return type:: tf.Tensor

Notes

Why estimate gain?

In indirect learning DPD, the PA output is divided by the gain G before being fed to the postdistorter. Without this normalization, the postdistorter would learn to invert both the PA nonlinearity AND its linear gain, which is undesirable.

Why low amplitude?

At low input amplitudes, the PA operates linearly and higher-order terms (|x|^2, |x|^4, etc.) become negligible. The measured gain then reflects only the first-order (linear) coefficient.

Interpolator¶

class demos.dpd.src.interpolator.Interpolator(*args, **kwargs)[source]¶

Bases: Layer

Rational sample rate converter using polyphase-equivalent resampling.

Converts between sample rates by a rational factor L/M, where L is the upsampling factor and M is the downsampling factor. The rate ratio is automatically converted to a reduced fraction for efficiency.

Parameters:

input_rate (float) – Input sample rate in Hz (e.g., 15.36e6 for OFDM baseband).
output_rate (float) – Desired output sample rate in Hz (e.g., 122.88e6 for PA).
max_denominator (int) – Maximum denominator when approximating the rate ratio as a fraction. Higher values give more accurate rate conversion at the cost of longer filters. Defaults to 1000.
half_len_mult (int) – Filter half-length multiplier. The filter length is 2 * half_len_mult * max(L, M) + 1. Defaults to 20, which provides >100 dB stopband attenuation.
kaiser_beta (float) – Kaiser window shape parameter. Higher values give better stopband attenuation but wider transition band. Defaults to 8.0 for >100 dB.

_upsample_factor¶

Upsampling factor L (numerator of rate ratio).

Type:: int

_downsample_factor¶

Downsampling factor M (denominator of rate ratio).

Type:: int

_output_rate¶

Actual output rate (may differ slightly from requested due to rational approximation).

Type:: float

_filter_coeffs¶

FIR filter coefficients as a TensorFlow constant.

Type:: tf.Tensor

Notes

Algorithm (Polyphase Resampling):

Upsample by L: Insert L-1 zeros between each input sample. This creates spectral images at multiples of the original Nyquist rate.
Anti-imaging filter: Apply a lowpass FIR filter with cutoff at min(input_rate, output_rate) / 2 to remove spectral images.
Downsample by M: Keep every M-th sample.

miscellaneous:

Input must be complex64 with shape [batch_size, num_samples]
Input and output rates must be positive
Output length is approximately num_samples * L / M
Output sample rate equals input_rate * L / M

Filter Design Rationale:

The Kaiser window is chosen because it provides a near-optimal trade-off between main lobe width and side lobe level, controlled by a single parameter (beta). With beta=8.0 and half_len_mult=20, the stopband attenuation exceeds 100 dB, ensuring spectral images are sufficiently suppressed to meet typical PA spurious emission requirements.

Example

>>> # Upsample from 15.36 MHz to 122.88 MHz (8x)
>>> interp = Interpolator(input_rate=15.36e6, output_rate=122.88e6)
>>> x = tf.complex(tf.random.normal([16, 1024]), tf.random.normal([16, 1024]))
>>> y, out_rate = interp(x)
>>> y.shape  # [16, 8192]
>>> out_rate  # 122880000.0

call(x)[source]¶

Resample input signal to the target sample rate.

Parameters:

x (tf.Tensor) – Input signal with shape [batch_size, num_samples] and dtype complex64. Each batch element is resampled independently.

Returns:

x_resampled (tf.Tensor) – Resampled signal with shape [batch_size, num_samples * L / M] and dtype complex64.
output_rate (float) – Actual output sample rate in Hz. This may differ slightly from the requested rate due to rational approximation.

Notes

Implementation Details:

The ‘full’ convolution padding is used to ensure all input samples contribute to the output. The group delay compensation then extracts the correctly aligned portion of the filtered signal.

For a symmetric FIR filter of length K, the group delay is (K-1)/2 samples. The output is trimmed to maintain the expected length relationship between input and output.

Least-Squares DPD¶

class demos.dpd.src.ls_dpd.LeastSquaresDPD(*args, **kwargs)[source]¶

Bases: Layer

Memory Polynomial predistorter with least-squares training.

Implements a polynomial-based predistorter where the output is a weighted sum of nonlinear basis functions of the input signal. The default model is a Memory Polynomial (MP). Setting lag_depth > 0 extends this to a Generalized Memory Polynomial (GMP).

Memory Polynomial:

\[y[n] = \sum_{k} \sum_{m} a_{k,m} \cdot x[n-m] \cdot |x[n-m]|^{k-1}\]

The structure contains parallel branches, each with a static nonlinearity x * |x|^(k-1) followed by an FIR filter (the memory taps).

Generalized Memory Polynomial (lag_depth > 0):

Adds cross-terms where the envelope is sampled at different time instants:

\[y[n] = \text{MP terms} + \sum_{k} \sum_{m} \sum_{l} b_{k,m,l} \cdot x[n-m] \cdot |x[n-m-l]|^{k-1}\]

Parameters:

params (dict, optional) –
Configuration dictionary. Supported keys:
- orderint
  Maximum polynomial order (must be odd). Default: 7.
- memory_depthint
  Number of memory taps per branch. Default: 4.
- lag_depthint
  Lag/lead depth for GMP cross-terms. Default: 0 (standard MP).
- nIterationsint
  Number of indirect learning iterations. Default: 3.
- learning_ratefloat
  Coefficient update rate (0-1). Default: 0.75.
- learning_methodstr
  Update method: ‘newton’ or ‘ema’. Default: ‘newton’.
- use_evenbool
  Include even-order terms. Default: False.
- use_conjbool
  Include conjugate branch (for IQ imbalance). Default: False.
- use_dc_termbool
  Include DC offset term. Default: False.
**kwargs – Additional keyword arguments passed to Keras Layer.

Notes

Why Odd-Order Only (Default)?

PA nonlinearity is predominantly odd-symmetric. Even-order terms model asymmetric distortion which is typically small. Using odd-only terms halves the coefficient count with minimal accuracy loss.

Parallel Hammerstein Interpretation:

The MP model can be viewed as parallel branches:

Branch 1: x -> FIR filter -> (linear path)
Branch 2: x * |x|² -> FIR filter -> (cubic nonlinearity)
Branch 3: x * |x|⁴ -> FIR filter -> (5th order)
…

Each branch’s FIR filter has memory_depth taps. All branch outputs are summed to produce the predistorted signal.

When to Use GMP (lag_depth > 0):

GMP cross-terms help when the PA exhibits strong interaction between the signal and its delayed envelope (e.g., thermal effects with time constants comparable to the signal bandwidth). For most PAs, standard MP (lag_depth=0) is sufficient.

Coefficient Initialization:

Coefficients are initialized to identity (first coefficient = 1, rest = 0), meaning the initial predistorter is a pass-through. Training then adjusts coefficients to learn the PA inverse.

miscellaneous:

Input signal should be complex baseband at PA sample rate
For batched input, all batches are concatenated for basis matrix
Output has same shape as input
Predistortion is differentiable w.r.t. coefficients

Example

>>> # Standard Memory Polynomial
>>> params = {"order": 7, "memory_depth": 4, "lag_depth": 0}
>>> dpd = LeastSquaresDPD(params)
>>> x = tf.complex(tf.random.normal([1024]), tf.random.normal([1024]))
>>> y = dpd(x)  # Apply predistortion
>>> y.shape
TensorShape([1024])

>>> # Generalized Memory Polynomial (with cross-terms)
>>> params_gmp = {"order": 7, "memory_depth": 4, "lag_depth": 2}
>>> dpd_gmp = LeastSquaresDPD(params_gmp)

DEFAULT_PARAMS = {'lag_depth': 0, 'learning_method': 'newton', 'learning_rate': 0.75, 'memory_depth': 4, 'nIterations': 3, 'order': 7, 'use_conj': False, 'use_dc_term': False, 'use_even': False}¶

build(input_shape)[source]¶

Create trainable coefficient weights.

Initializes coefficients to identity predistorter (pass-through): first coefficient is 1+0j, all others are 0+0j.

Parameters:: input_shape (tf.TensorShape) – Shape of input tensor (used by Keras, not directly needed here).

property n_coeffs¶

Total number of DPD coefficients.

Type:: int

property coeffs¶

Complex coefficient vector, shape [n_coeffs, 1].

Raises:: RuntimeError – If layer has not been built yet.
Type:: tf.Tensor

setup_basis_matrix(x)[source]¶

Construct the MP/GMP basis matrix for least-squares fitting.

Each column of the basis matrix corresponds to one coefficient in the model. The predistorter output is y = X @ coeffs.

Parameters:: x (tf.Tensor) – Input signal, shape [num_samples], complex dtype.
Returns:: Basis matrix, shape [num_samples, n_coeffs], complex dtype.
Return type:: tf.Tensor

Notes

Column ordering follows this pattern:

Main MP terms: orders 1,3,5,7 * memory delays 0,1,2,3
Lagging cross-terms (GMP, if lag_depth > 0): orders 3,5,7 * lags * delays
Leading cross-terms (GMP, if lag_depth > 0): orders 3,5,7 * leads * delays
Conjugate terms (if enabled): same structure with conj(x)
DC term (if enabled): column of ones

This method is fully differentiable, enabling gradient-based fine-tuning if desired.

predistort(x)[source]¶

Apply predistortion to input signal.

Computes y = X @ coeffs where X is the GMP basis matrix. This method is fully differentiable.

Parameters:: x (tf.Tensor) – Input signal, shape [num_samples] or [batch, num_samples].
Returns:: Predistorted signal, same shape as input.
Return type:: tf.Tensor
Raises:: ValueError – If input has more than 2 dimensions.

Notes

For batched input, all batches are concatenated before building the basis matrix, then the output is reshaped back to batch form. This ensures consistent processing across the batch.

call(x, training=None)[source]¶

Keras layer call interface.

Parameters:

x (tf.Tensor) – Input signal.
training (bool or None, optional) – Training mode flag (unused, included for Keras compatibility).

Returns:

Predistorted signal.

Return type:

tf.Tensor

Neural Network DPD¶

class demos.dpd.src.nn_dpd.ResidualBlock(*args, **kwargs)[source]¶

Bases: Layer

Fully-connected residual block with layer normalization and PReLU.

Implements the pre-activation residual block pattern where normalization and activation precede each linear transformation. This ordering improves gradient flow and training stability compared to post-activation.

Parameters:

units (int, optional) – Number of units in each dense layer. Default: 64.
num_layers (int, optional) – Number of dense layers in the block. Default: 2.
**kwargs – Additional keyword arguments passed to Keras Layer.

units¶

Number of units per layer.

Type:: int

num_layers¶

Number of layers in block.

Type:: int

Notes

Why PReLU?

DPD corrections can be both positive and negative. PReLU (Parametric ReLU) has a learnable slope for negative inputs, allowing the network to optimize how negative corrections are handled. Standard ReLU would zero out negative values, limiting the correction space.

Why Layer Normalization?

Layer normalization (vs batch normalization) normalizes across features rather than batch dimension. This is important for DPD where:

Batch sizes may be small or variable
Each sample should be processed consistently
Inference behavior should match training exactly

Skip Connection:

The residual connection output = F(x) + x ensures:

Gradient can flow directly through the skip path
Block can learn to output zero (identity mapping)
Deeper networks remain trainable

Example

>>> block = ResidualBlock(units=64, num_layers=2)
>>> x = tf.random.normal([16, 100, 64])  # [batch, time, features]
>>> y = block(x)
>>> y.shape
TensorShape([16, 100, 64])

call(inputs)[source]¶

Forward pass through residual block.

Parameters:: inputs (tf.Tensor) – Input tensor, shape [batch, time, units].
Returns:: Output tensor with residual connection, same shape as input.
Return type:: tf.Tensor

class demos.dpd.src.nn_dpd.NeuralNetworkDPD(*args, **kwargs)[source]¶

Bases: Layer

Fully-connected feedforward neural network predistorter with memory.

Implements a neural network that learns the PA inverse function through gradient-descent and backpropagation. Memory effects are captured via a sliding window over input samples, similar in concept to the memory polynomial approach.

The network receives explicit amplitude features (|x|^2, |x|^4, |x|^6) in addition to I/Q components. This is critical because PA distortion is fundamentally amplitude-driven (AM-AM and AM-PM conversion). Without these features, the network must learn from scratch that amplitude matters, which is inefficient and leads to suboptimal spectral behavior.

Parameters:

memory_depth (int, optional) – Number of samples in sliding window. Larger values capture longer memory effects but increase computation. Default: 4.
num_filters (int, optional) – Number of units in hidden layers. Controls model capacity. Default: 64.
num_layers_per_block (int, optional) – Number of dense layers per residual block. Default: 2.
num_res_blocks (int, optional) – Number of residual blocks. More blocks increase depth and capacity. Default: 3.
**kwargs – Additional keyword arguments passed to Keras Layer.

_memory_depth¶

Sliding window size.

Type:: int

_num_filters¶

Hidden layer width.

Type:: int

_num_res_blocks¶

Number of residual blocks.

Type:: int

Notes

Amplitude Features:

PA nonlinearity is driven by instantaneous signal amplitude. The memory polynomial explicitly models terms like |x[n-m]|^p · x[n-m] where p is even (0, 2, 4, 6…) corresponding to odd-order nonlinearities. By providing |x|^2, |x|^4, |x|^6 as explicit input features, the network is given direct access to the same basis functions that make memory polynomials effective for PA linearization.

Feature layout per sample (5 * memory_depth total features):

[real[n-M+1:n], imag[n-M+1:n], |x|^2[n-M+1:n], |x|^4[n-M+1:n], |x|^6[n-M+1:n]]

Identity Initialization:

The output layer is initialized to zeros, so the initial network output is just the skip connection (identity function). This ensures:

Initial predistorter is pass-through (no distortion)
Training starts from a reasonable point
Network learns corrections relative to identity

Complex Signal Handling:

Complex signals are split into real and imaginary parts for processing. This is necessary because standard neural network layers operate on real-valued tensors. The network learns to process I and Q jointly, capturing their correlations.

Causal Processing:

The sliding window only includes current and past samples (causal). This is appropriate for real-time DPD where future samples are unavailable. Zero-padding handles the initial transient.

miscellaneous:

Input must be complex64 tensor
Batch dimension is optional (will be added if missing)
Output has same shape as input
Initial (untrained) output equals input (identity)

Example

>>> dpd = NeuralNetworkDPD(memory_depth=4, num_filters=64)
>>> x = tf.complex(tf.random.normal([16, 1024]), tf.random.normal([16, 1024]))
>>> y = dpd(x)
>>> y.shape
TensorShape([16, 1024])

System¶

class demos.dpd.src.system.DPDSystem(*args, **kwargs)[source]¶

Bases: Layer

Base class for DPD systems implementing indirect learning architecture.

Provides the common infrastructure for all DPD methods:

OFDM signal generation via Sionna
Upsampling from baseband to PA sample rate
Power amplifier model with memory effects
PA gain estimation and normalization
Receiver chain for inference-time EVM measurement

Subclasses must implement:

_forward_signal_path(): Signal flow through DPD and PA
_training_forward(): Training loss computation
_inference_forward(): Inference output generation

Parameters:

training (bool) – Operating mode. True for training (receiver not instantiated), False for inference (receiver instantiated).
config (Config) – Configuration object with RF and OFDM parameters.
rms_input_dbm (float, optional) – Target RMS power for PA input in dBm. Default is 0.5 dBm, which drives the considered PA into compression.
pa_sample_rate (float, optional) – PA operating sample rate in Hz. Default is 122.88 MHz (8x the 15.36 MHz baseband rate for 1024-FFT, 15 kHz spacing).
**kwargs – Additional keyword arguments passed to Keras Layer.

Notes

Indirect Learning Architecture:

The key insight is that at convergence, the predistorter output u equals the ideal PA input that would produce linear output. By training a postdistorter on the PA output to reproduce u, it learns the PA inverse. The trained postdistorter weights are then copied to the predistorter.

The training loop:

Generate baseband signal x
Apply predistorter: u = DPD(x)
Pass through PA: y = PA(u)
Normalize by PA gain: y_norm = y / G
Compute loss: L = ||DPD(y_norm) - u||^2
Update DPD weights via backpropagation

Why Gain Normalization?

The PA has linear gain G in addition to nonlinear distortion. Without normalization, the postdistorter would learn to invert both the gain and nonlinearity. By dividing by G, the nonlinear component is isolated for the postdistorter to learn.

estimate_pa_gain() should be called before training
PA gain is estimated once and remains fixed during training

Example

Subclasses use this base class as follows:

>>> class MyDPDSystem(DPDSystem):
...     def __init__(self, training, config, **kwargs):
...         super().__init__(training, config, **kwargs)
...         self._dpd = MyDPDLayer()  # Set predistorter
...
...     def _forward_signal_path(self, x):
...         # Implement signal flow
...         ...

property dpd¶

The predistorter layer (set by subclass).

Type:: Layer

property minimal_ofdm_receiver¶

OFDM receiver (only available in inference mode).

Type:: demos.dpd.src.rx.Rx or None

property signal_fs¶

Baseband signal sample rate in Hz.

Type:: float

property pa_sample_rate¶

PA operating sample rate in Hz.

Type:: float

property fft_size¶

OFDM FFT size.

Type:: int

property cp_length¶

Cyclic prefix length in samples.

Type:: int

property num_ofdm_symbols¶

Number of OFDM symbols per slot.

Type:: int

estimate_pa_gain(num_samples=10000)[source]¶

Estimate PA small-signal (linear) gain.

Measures the PA’s voltage gain in its linear operating region using a low-amplitude test signal. This gain value is used to normalize the PA output before feeding it to the postdistorter.

Parameters:: num_samples (int, optional) – Number of random samples for gain estimation. More samples reduce variance. Default is 10000.
Returns:: Estimated voltage gain (linear scale, not dB).
Return type:: float

Notes

This method should be called once before training begins. The estimated gain is stored internally and used by the forward pass to normalize PA output. Calling this multiple times will update the stored gain value.

The gain is estimated in the linear region (low amplitude) where higher-order polynomial terms are negligible, giving just the first-order (linear) response.

static normalize_to_rms(data, target_rms)[source]¶

Normalize signal to target RMS power level.

Scales the input signal so its RMS power matches the target, specified in dBm (assuming 50 ohm impedance).

Parameters:

data (tf.Tensor) – Input signal, shape [batch, num_samples], complex dtype.
target_rms (float) – Target RMS power in dBm.

Returns:

normalized (tf.Tensor) – Scaled signal with target RMS power, same shape as input.
scale_factor (tf.Tensor) – Multiplicative scale factor applied (useful for inverse scaling).

Notes

The normalization treats all batches as one concatenated signal (global statistics), ensuring consistent power across the batch.

Power conversion: P_watts = 10^((P_dBm - 30) / 10)

For 50 ohm systems: V_rms = sqrt(50 * P_watts)

generate_signal(batch_size, return_extras=False)[source]¶

Generate a batch of OFDM baseband signals upsampled to PA rate.

Creates random transmit signals through the full Tx chain (bit generation -> LDPC -> QAM -> OFDM), normalizes power, and upsamples to the PA sample rate.

Parameters:

batch_size (int or tf.Tensor) – Number of independent signals to generate.
return_extras (bool, optional) – If True, return additional data for constellation plotting and receiver processing. Default is False.

Returns:

If return_extras=False:

Upsampled signal, shape [batch, num_samples].

If return_extras=True:

Dictionary containing:

tx_upsampledtf.Tensor
Upsampled signal at PA rate.
tx_basebandtf.Tensor
Original baseband signal (flattened) for sync reference.
x_rgtf.Tensor
Resource grid (frequency-domain, all subcarriers).
fd_symbolstf.Tensor
Data subcarrier symbols only, shape [num_data_subcarriers, num_symbols].

Return type:

tf.Tensor or dict

Notes

The fd_symbols output excludes guard bands and DC null, containing only the data-bearing subcarriers in the same order as the receiver expects for equalization.

call(batch_size_or_signal, training=None)[source]¶

Forward pass through the DPD system.

Handles both signal generation (from batch size) and processing of pre-generated signals. Dispatches to training or inference path based on mode.

Parameters:

batch_size_or_signal (int, tf.Tensor (scalar), or tf.Tensor (2D)) –
Either:
- Python int or scalar tensor: interpreted as batch size, signal will be generated internally.
- 2D tensor [batch, samples]: pre-generated signal to process.
training (bool or None, optional) – Override the instance’s training mode. If None, uses the mode specified at construction.

Returns:

Training mode:

Scalar loss value (for gradient-based optimization).

Inference mode:

Dictionary with keys:

pa_input : PA input signal (upsampled, before DPD)
pa_output_no_dpd : PA output without predistortion
pa_output_with_dpd : PA output with predistortion

Return type:

tf.Tensor or dict

Raises:

ValueError – If batch_size_or_signal is not a valid type.

Notes

The ability to accept pre-generated signals allows for consistent evaluation: generate once, then compare with/without DPD on the exact same input.

class demos.dpd.src.ls_dpd_system.LS_DPDSystem(*args, **kwargs)[source]¶

Bases: DPDSystem

Complete LS-DPD system with closed-form coefficient estimation.

Extends the base DPDSystem with a Memory Polynomial predistorter trained via least-squares regression. The training uses indirect learning architecture with either Newton or EMA update methods.

Parameters:

training (bool) – Operating mode. True for training, False for inference.
config (Config) – Frozen configuration object with RF and OFDM parameters.
dpd_order (int, optional) – Maximum polynomial order (must be odd). Default: 7.
dpd_memory_depth (int, optional) – Number of memory taps per polynomial branch. Default: 4.
ls_nIterations (int, optional) – Number of indirect learning iterations. Default: 3.
ls_learning_rate (float, optional) – Coefficient update rate (0-1). Higher values give faster but potentially unstable convergence. Default: 0.75.
ls_learning_method (str, optional) –
Coefficient update method:
- 'newton': Update based on error between predistorter output and postdistorter output. More stable.
- 'ema': Exponential moving average of direct LS solutions. Faster convergence but can overshoot.
Default: 'newton'.
rms_input_dbm (float, optional) – Target RMS power for PA input in dBm. Default: 0.5.
pa_sample_rate (float, optional) – PA operating sample rate in Hz. Default: 122.88 MHz.
**kwargs – Additional keyword arguments passed to base DPDSystem.

dpd¶

The Memory Polynomial predistorter layer.

Type:: LeastSquaresDPD

Notes

Newton vs EMA Update Methods:

Newton method (default):

\[\mathbf{c}_{new} = \mathbf{c} + \mu \cdot \text{LS}(\mathbf{Y}, \mathbf{u} - \hat{\mathbf{u}})\]

where \(\hat{\mathbf{u}} = \text{DPD}(\mathbf{y}/G)\) is the postdistorter output. This computes an incremental correction.

EMA method:

\[\mathbf{c}_{new} = (1-\mu) \mathbf{c} + \mu \cdot \text{LS}(\mathbf{Y}, \mathbf{u})\]

This directly averages between old and new LS solutions.

Typical Convergence:

LS-DPD typically converges in 2-4 iterations. More iterations may be needed for highly nonlinear PAs or when starting far from optimal.

miscellaneous:

estimate_pa_gain() must be called before perform_ls_learning()
After perform_ls_learning(), DPD coefficients are optimized
Coefficient history is stored in dpd.coeff_history

Example

>>> config = Config()
>>> system = LS_DPDSystem(training=True, config=config)
>>> system.estimate_pa_gain()
>>> results = system.perform_ls_learning(batch_size=16, verbose=True)
>>> print(f"Final coefficients shape: {results['coeffs'].shape}")