Add new functions

This commit is contained in:
Silas Oettinghaus
2025-02-17 21:31:12 +01:00
parent becaf3f6c9
commit d099efea03
21 changed files with 1502 additions and 272 deletions

View File

@@ -0,0 +1,156 @@
function air = air_garcia_implementation(x,r,idx_tx,Px,M_training)
MAX_MEMORY = 200e6; % maximum allowed size for a matrix
if nargin == 3
Px = [];
M_training = [];
end
if nargin == 4
M_training = [];
end
% if input is complex, separate into real and imaginary parts
if any(imag(x(:))~=0) || any(imag(r(:))~=0)
x = [real(x); imag(x)];
r = [real(r); imag(r)];
end
D = size(x, 1); % D = 2 if complex x
N = size(x, 2); % number of constellation points
M = size(r, 2); % number of samples
% set default training set size
if isempty(M_training)
M_training = ceil(0.3*M);
end
M_testing = M - M_training;
% Training: estimate parameters of the conditionally Gaussian model
% sort according to transmit index
[idx_tx_training, idx_sort] = sort(idx_tx(1:M_training));
r_training = r(:, idx_sort);
i_bounds = zeros(1, N+1);
% compute conditional means and covariance matrices
C_n = zeros(D, D, N);
det_n = zeros(1, N);
for n=1:N
% find how many times x(:, n) was transmitted and update i_bounds
N_current_x = find(idx_tx_training((i_bounds(n)+1):end)==n, 1, 'last');
if isempty(N_current_x), N_current_x=0; end
i_bounds(n+1) = i_bounds(n) + N_current_x;
if N_current_x > 0
% Compute mu_n=E[Y|X=x_n] according to Eq. (14) and store it in
% x(:, n) to save space
x(:, n) = sum(r_training(:, (i_bounds(n)+1):i_bounds(n+1)), 2)/(i_bounds(n+1)-i_bounds(n));
% compute C_n=cov[Y|X=x_n] according to Eq. (15)
r_meanfree = r_training(:, (i_bounds(n)+1):i_bounds(n+1)) - x(:, n);
C_n(:, :, n) = (r_meanfree*r_meanfree')/(i_bounds(n+1)-i_bounds(n));
% store also the determinant of C(:, :, n)
det_n(n) = det(C_n(:, :, n));
% if the determinant is 0, or if the matrix is badly conditioned,
% regularize by adding a small identity matrix. Note that we do
% need the check for 0 determinant, in case a cloud has exactly 0
% variance according to the training set
if det_n(n)==0 || cond(C_n(:, :, n))>1e16
C_n(:, :, n) = C_n(:, :, n) + 5 * eps * eye(D);
det_n(n) = (5*eps)^D;
end
end
end
% uniform input pmf Px if not provided
if isempty(Px)
Px = repmat(1/N, [1, N]);
end
% extract testing set and sort it according to transmit index
[idx_tx_testing, idx_sort] = sort(idx_tx((M_training+1):M));
r_testing = r(:, M_training+idx_sort);
% computation of h(Y|X)
h_Y_X = 0;
i_bounds_testing = zeros(1, N+1);
% loop over constellation points to compute h(Y|X)
for n = 1:N
% find how many times x(:, n) was transmitted and update
% i_bounds_testing
N_current_x = find(idx_tx_testing((i_bounds_testing(n)+1):end)==n, 1, 'last');
if isempty(N_current_x), N_current_x=0; end
i_bounds_testing(n+1) = i_bounds_testing(n) + N_current_x;
% add the corresponding contribution to the mutual information (two
% first lines of Eq. (17)). This, together with
% D/2*log2(2*pi) after the end of the loop, gives h(Y|X)
h_Y_X = h_Y_X + N_current_x * log2(det_n(n))/2+...
sum(sum(conj(r_testing(:, (i_bounds_testing(n)+1):i_bounds_testing(n+1))-x(:, n)).*(C_n(:, :, n)\(r_testing(:, (i_bounds_testing(n)+1):i_bounds_testing(n+1))-x(:, n)))))/2/log(2);
end
h_Y_X = D/2*log2(2*pi) + h_Y_X/M_testing;
% When computing log(py), we might run out of memory. If necessary, we
% doe the computation in blocks
logpy = zeros(1, M_testing);
BLOCK_SIZE = floor(MAX_MEMORY/N);
N_blocks = ceil(M_testing/BLOCK_SIZE);
% loop over blocks of symbols. This loop can be replaced by parfor to allow
% parallel computation
for i_block = 1:N_blocks
logpy_cur = zeros(1, M_testing);
% beginning of block
i_start = (i_block-1) * BLOCK_SIZE + 1;
% end of block
i_end = min(M_testing, i_block*BLOCK_SIZE);
% block size
current_block_size = i_end-i_start+1;
% compute exponents of third line of (17)
exponents = zeros(N, current_block_size);
for n = 1:N
exponents(n, :) = -log(det_n(n))/2-real(sum(conj(r_testing(:, i_start:i_end)-x(:, n)).*(C_n(:, :, n)\(r_testing(:, i_start:i_end)-x(:, n))), 1))/2;
%sum über 2 einträge von r
end
% compute third line of Eq. (17). Use a custom function
% that computes log(sum(exp(x))) avoiding overflow errors
logpy_cur(i_start:i_end) = math_logsumexp(log(Px(:))+exponents, 1);
logpy = logpy + logpy_cur;
end
% output entropy h(Y)
h_Y = D/2*log2(2*pi) - mean(logpy)/log(2);%log basis change
% compute mutual information
air = h_Y - h_Y_X;
end
function [y] = math_logsumexp(x, dim)
%[y] = math_logsumexp(x, dim)
% Computes log(sum(exp(x), dim)), avoiding overflow errors when one of the
% x is large.
if nargin<2 || isempty(dim)
m = max(x);
y = m + log(sum(exp(x-m)));
else
m = max(x, [], dim);
y = m + log(sum(exp(x-m), dim));
end
end

View File

@@ -0,0 +1,48 @@
function [ach_inf_rate] = calc_air(test_signal,reference_signal,options)
% Calculation of AIR acc. to J. Kozesnik, Numerically Computing Achievable Rates of Memoryless Channels, Francisco Javier Garcıa-Gomez, doi: 10.1007/978-94-009-9857-5.
% Implementation is not accessible, I mailed TUM to get the code...
arguments(Input)
test_signal;
reference_signal;
options.skip_front = 0;
options.skip_end = 0;
options.returnErrorLocation = 0;
end
options.skip_end = abs(options.skip_end);
options.skip_front = abs(options.skip_front);
assert((options.skip_end+options.skip_front)<length(test_signal),"You can not skip more bits than overall length of data! Set skip_front or skip_end to lower value or check data_in");
if isa(reference_signal,'Signal')
reference_signal = reference_signal.signal;
end
if isa(test_signal,'Signal')
test_signal = test_signal.signal;
end
% TRIM
[test_signal,reference_signal]=trimseq(test_signal,reference_signal,options.skip_front,options.skip_end);
% CALC EVM
%%% new implementation of AIR
constellation = unique(reference_signal);
reference_idx = arrayfun(@(x) find(constellation == x, 1), reference_signal);
ach_inf_rate = air_garcia_implementation(constellation',test_signal',reference_idx');
function [data_,reference_]=trimseq(data,reference,skipstart,skip_end)
data_ = data(skipstart+1:end-skip_end,:);
delta_bits = length(reference) - length(data);
skip_end = delta_bits + skip_end;
reference_ = reference(skipstart+1:end-skip_end,:);
end
end

View File

@@ -0,0 +1,52 @@
function [bits,errors,ber,errorIndice] = calc_ber(data_in,data_ref,options)
arguments(Input)
data_in;
data_ref;
options.skip_front = 0;
options.skip_end = 0;
options.returnErrorLocation = 0;
end
options.skip_end = abs(options.skip_end);
options.skip_front = abs(options.skip_front);
assert((options.skip_end+options.skip_front)<length(data_in),"You can not skip more bits than overall length of data! Set skip_front or skip_end to lower value or check data_in");
errorIndice= [];
% trim sequence to given start; stop. trim reference if signal is shorter
[data_in,data_ref]=trimseq(data_in,data_ref,options.skip_front,options.skip_end);
if length(data_ref) == length(data_in)
bits = numel(data_in(:,options.skip_front+1:end));
if options.returnErrorLocation == 0
errors = sum( data_in ~= data_ref,"all" );
else
errorIndice = sum(data_in ~= data_ref,1);
errors = sum(errorIndice ,"all" );
[~,errorIndice] = find(errorIndice~=0);
errorIndice = errorIndice+options.skip_front;
end
% Determine BER
ber = sum(errors)/sum(bits);
else
error('Sequence length does not match');
end
function [data_,reference_]=trimseq(data,reference,skipstart,skip_end)
data_ = logical(data(skipstart+1:end-skip_end,:))';
delta_bits = length(reference) - length(data);
skip_end = delta_bits + skip_end;
reference_ = logical(reference(skipstart+1:end-skip_end,:))';
end
end

View File

@@ -0,0 +1,66 @@
function [evm_total,evm_lvl] = calc_evm(test_signal,reference_signal,options)
arguments(Input)
test_signal;
reference_signal;
options.skip_front = 0;
options.skip_end = 0;
options.returnErrorLocation = 0;
end
options.skip_end = abs(options.skip_end);
options.skip_front = abs(options.skip_front);
assert((options.skip_end+options.skip_front)<length(test_signal),"You can not skip more bits than overall length of data! Set skip_front or skip_end to lower value or check data_in");
if isa(reference_signal,'Signal')
reference_signal = reference_signal.signal;
end
if isa(test_signal,'Signal')
test_signal = test_signal.signal;
end
% TRIM
[test_signal,reference_signal]=trimseq(test_signal,reference_signal,options.skip_front,options.skip_end);
% CALC EVM
[evm_total,evm_lvl] = calc_evm_(test_signal,reference_signal);
function [evm_total,evm_lvl] = calc_evm_(test_signal,reference_signal)
assert(length(test_signal) == length(reference_signal),"Sequence length does not match");
error_vector = (test_signal-reference_signal);
%%% Overall EVM
evm_total = rms(error_vector);
try
%%% Per Level EVM
k = unique(reference_signal);
for lvl = 1:length(k)
lvl_errors = error_vector(reference_signal==k(lvl));
evm_lvl(lvl) = rms(lvl_errors);
end
catch
evm_lvl = NaN;
warning('No EVM per level calculated')
end
end
function [data_,reference_]=trimseq(data,reference,skipstart,skip_end)
data_ = data(skipstart+1:end-skip_end,:);
delta_bits = length(reference) - length(data);
skip_end = delta_bits + skip_end;
reference_ = reference(skipstart+1:end-skip_end,:);
end
end

View File

@@ -0,0 +1,123 @@
function [GMI,NGMI] = calc_ngmi(test_signal,reference_signal,options)
% Silas implementation of (N)GMI calculation according to: J. Cho, L. Schmalen, und P. J. Winzer,
% Normalized Generalized Mutual Information as a Forward Error Correction Threshold for Probabilistically Shaped QAM,
% in 2017 European Conference on Optical Communication (ECOC), Sep. 2017, doi: 10.1109/ECOC.2017.8345872.
% This implementation assumes the same normal distributed noise for each
% channel (sigma2 is calculated once for the whole signal)
arguments(Input)
test_signal;
reference_signal;
options.skip_front = 0;
options.skip_end = 0;
options.returnErrorLocation = 0;
end
options.skip_end = abs(options.skip_end);
options.skip_front = abs(options.skip_front);
assert((options.skip_end+options.skip_front)<length(test_signal),"You can not skip more bits than overall length of data! Set skip_front or skip_end to lower value or check data_in");
if isa(reference_signal,'Signal')
reference_signal = reference_signal.signal;
end
if isa(test_signal,'Signal')
test_signal = test_signal.signal;
end
% TRIM
[test_signal,reference_signal]=trimseq(test_signal,reference_signal,options.skip_front,options.skip_end);
% CALC EVM
[GMI,NGMI] = calc_ngmi_(test_signal,reference_signal);
function [GMI,NGMI] = calc_ngmi_(test_signal,reference_signal)
assert(length(test_signal) == length(reference_signal),"Sequence length does not match");
%%% implemented according to [1] J. Cho, L. Schmalen, und P. J. Winzer,
% Normalized Generalized Mutual Information as a Forward Error Correction Threshold for Probabilistically Shaped QAM,
% in 2017 European Conference on Optical Communication (ECOC), Sep. 2017, S. 13. doi: 10.1109/ECOC.2017.8345872.
error_vector = (test_signal-reference_signal);
sigma2 = var(error_vector); %noise variance
%%% Separate Classes
constellation = unique(reference_signal);
received_sd = NaN(numel(constellation),length(reference_signal));
lvlcol = cbrewer2('Set1',numel(constellation));
for lvl = 1:numel(constellation)
%Separate the equalized signal into the
%respective levels based on the actually
%transmitted level!
received_sd(lvl,reference_signal==constellation(lvl)) = test_signal(reference_signal==constellation(lvl));
end
N = length(test_signal); % Number of received samples
M = length(constellation); %P
m = log2(M); %bits per symbol
entries = sum(~isnan(received_sd),2)';
P_X = entries./N;
% Parameters
symbols = constellation'; % PAM-4 symbols
gray_bits = PAMmapper(M,0).demap_(constellation); % Gray coding (bits per symbol)
% Conditional probability function for AWGN
q_Y_given_X = @(y, x) (1 / sqrt(2 * pi * sigma2)) * exp(-(y - x).^2 / (2 * sigma2));
% Entropy term
H_X = -sum(P_X .* log2(P_X)); % Entropy of input distribution
% GMI computation
noise_impact_term = 0;
for k = 1:N
y_k = test_signal(k); % Current received sample
[~, closest_symbol_idx] = min(abs(symbols - y_k)); % Closest symbol index
closest_symbol = symbols(closest_symbol_idx); % Closest symbol
for i = 1:m
% Extract i-th bit for each symbol
bit_mask = gray_bits(:, i); % Binary column for i-th bit of all symbols
matching_symbols = symbols(bit_mask == gray_bits(closest_symbol_idx, i));
% Numerator: Sum over x in x_{b_{k, i}}
numerator = sum(q_Y_given_X(y_k, matching_symbols) .* P_X(ismember(symbols, matching_symbols)));
% Denominator: Sum over all x
denominator = sum(q_Y_given_X(y_k, symbols) .* P_X);
% Logarithmic contribution
noise_impact_term = noise_impact_term + log2(numerator / denominator);
end
end
% Normalize the noise impact term by N
noise_impact_term = noise_impact_term / N;
% GMI
GMI = H_X + noise_impact_term;
NGMI = GMI / m;
end
function [data_,reference_]=trimseq(data,reference,skipstart,skip_end)
data_ = data(skipstart+1:end-skip_end,:);
delta_bits = length(reference) - length(data);
skip_end = delta_bits + skip_end;
reference_ = reference(skipstart+1:end-skip_end,:);
end
end