classdef FFE_test < IMDDTestCase methods (Test, TestTags = {'unit', 'fast', 'dsp', 'ffe'}) function constructorStoresDefaultsAndInitialState(testCase) ffe = FFE(); testCase.verifyEqual(ffe.sps, 2); testCase.verifyEqual(ffe.order, 15); testCase.verifyEqual(ffe.len_tr, 4096); testCase.verifyEqual(ffe.mu_tr, 0); testCase.verifyEqual(ffe.epochs_tr, 5); testCase.verifyEqual(ffe.adaption_technique, adaption_method.lms); testCase.verifyEqual(ffe.dd_mode, 1); testCase.verifyEqual(ffe.mu_dd, 1e-5); testCase.verifyEqual(ffe.epochs_dd, 5); testCase.verifyFalse(logical(ffe.optimize_dc_tracking_params)); testCase.verifyEqual(ffe.dc_tracking_optimization_delay_weight, 1e-3); testCase.verifyEqual(ffe.dc_tracking_persistence_gain, 0); testCase.verifyEqual(ffe.dc_tracking_power_exponent, 2); testCase.verifyEqual(ffe.dc_avg_bufferlength_a1, 0); testCase.verifyEqual(ffe.dc_avg_update_blocklength_a1, 0); testCase.verifyEqual(ffe.dc_smoothing_a1, 0); testCase.verifyEqual(ffe.dc_level_avg_bufferlength_a2, 0); testCase.verifyEqual(ffe.dc_level_update_blocklength_a2, 0); testCase.verifyEqual(ffe.dc_smoothing_a2, 0); testCase.verifyEqual(ffe.dc_level_weights_a2, 0); testCase.verifyEqual(ffe.dc_tracking_buffer_len, 1); testCase.verifyEqual(ffe.ffe_update_buffer_len, 1); testCase.verifyFalse(logical(ffe.optimize_a2_level_weights)); testCase.verifyEqual(ffe.a2_level_weight_optimization_max_evals, 30); testCase.verifyEqual(ffe.a2_level_weight_max, 1); testCase.verifyFalse(ffe.decide); testCase.verifyEqual(ffe.e, zeros(ffe.order, 1)); testCase.verifyEqual(ffe.error, 0); end function noUpdateModeKeepsWeightsAtZeroAndPreservesSamplingRate(testCase) x = makeSignal([-1; 1; -1; 1], 32e9); d = makeSignal([-1; 1; -1; 1], 8e9); ffe = FFE( ... "sps", 1, ... "order", 1, ... "len_tr", length(x), ... "mu_tr", 0, ... "epochs_tr", 1, ... "mu_dd", 0, ... "epochs_dd", 1, ... "dd_mode", 0, ... "decide", false, ... "adaption_technique", adaption_method.lms); [y, noi] = ffe.process(x, d); testCase.verifyEqual(ffe.e, 0); testCase.verifyEqual(ffe.e_tr, 0); testCase.verifyEqual(y.signal, zeros(size(x.signal)), "AbsTol", 1e-12); testCase.verifyEqual(y.fs, d.fs); testCase.verifyEqual(noi.signal, -d.signal, "AbsTol", 1e-12); testCase.verifyEqual(length(y.signal), length(x.signal)); end function trainingModeLearnsFromTinySyntheticChannel(testCase) x = makeSignal([1; -1; 1; -1], 16e9); d = makeSignal([1; -1; 1; -1], 4e9); ffe = FFE( ... "sps", 1, ... "order", 1, ... "len_tr", length(x), ... "mu_tr", 0.5, ... "epochs_tr", 1, ... "mu_dd", 0, ... "epochs_dd", 1, ... "dd_mode", 0, ... "decide", false, ... "adaption_technique", adaption_method.lms); [y, noi] = ffe.process(x, d); testCase.verifyEqual(y.fs, d.fs); testCase.verifyClass(y, 'Informationsignal'); testCase.verifyClass(noi, 'Informationsignal'); testCase.verifyNotEqual(ffe.e, 0); testCase.verifyNotEqual(ffe.e_tr, 0); testCase.verifyNotEqual(y.signal, zeros(size(x.signal))); testCase.verifyEqual(length(y.signal), length(x.signal)); end function dcTrackingBufferDelaysDcUpdateUntilBlockBoundary(testCase) x = zeros(4, 1); d = ones(4, 1); ffe = FFE( ... "sps", 1, ... "order", 1, ... "dc_tracking_mu", 0.1, ... "dc_tracking_buffer_len", 2, ... "dd_mode", 0, ... "adaption_technique", adaption_method.lms); ffe.constellation = unique(d); ffe.equalize(x, d, 0, 1, numel(x), true, false); testCase.verifyEqual(ffe.e_dc, 0.19, "AbsTol", 1e-12); end function dcTrackingUnbufferedPathUpdatesEverySymbol(testCase) x = zeros(4, 1); d = ones(4, 1); ffe = FFE( ... "sps", 1, ... "order", 1, ... "dc_tracking_mu", 0.1, ... "dc_tracking_buffer_len", 1, ... "dd_mode", 0, ... "adaption_technique", adaption_method.lms); ffe.constellation = unique(d); ffe.equalize(x, d, 0, 1, numel(x), true, false); testCase.verifyEqual(ffe.e_dc, 1 - 0.9^4, "AbsTol", 1e-12); end function dcTrackingBlockUpdateUsesMeanBlockError(testCase) ffe = FFE("dc_tracking_mu", 0.1); [e_dc_next,stats] = ffe.dcTrackingBlockUpdate(0.5, [1; 2; 3]); testCase.verifyEqual(e_dc_next, 0.7, "AbsTol", 1e-12); testCase.verifyEqual(stats.err_mean, 2, "AbsTol", 1e-12); testCase.verifyEqual(stats.mu_eff, 0.1, "AbsTol", 1e-12); testCase.verifyEqual(stats.update, 0.2, "AbsTol", 1e-12); end function dcTrackingBlockUpdateDoesNotSuppressLargeErrorPower(testCase) ffe = FFE("dc_tracking_mu", 0.1); [e_dc_next,stats] = ffe.dcTrackingBlockUpdate(0, [10; 10]); testCase.verifyEqual(e_dc_next, 1, "AbsTol", 1e-12); testCase.verifyEqual(stats.err_mean, 10, "AbsTol", 1e-12); testCase.verifyEqual(stats.mu_eff, 0.1, "AbsTol", 1e-12); end function dcTrackingBlockUpdateCanBoostPersistentMean(testCase) ffe = FFE("dc_tracking_mu", 0.1); [e_dc_next,stats] = ffe.dcTrackingBlockUpdate(0, [2; 2; 2], ... "persistence_gain", 1); testCase.verifyEqual(e_dc_next, 0.4, "AbsTol", 1e-12); testCase.verifyEqual(stats.persistence_scale, 1, "AbsTol", 1e-12); testCase.verifyEqual(stats.mu_eff, 0.2, "AbsTol", 1e-12); end function dcTrackingLoopUsesBlockPersistenceGain(testCase) x = zeros(4, 1); d = ones(4, 1); ffe = FFE( ... "sps", 1, ... "order", 1, ... "dc_tracking_mu", 0.1, ... "dc_tracking_adaptive_enabled", true, ... "dc_tracking_persistence_gain", 1, ... "dc_tracking_buffer_len", 2, ... "dd_mode", 0, ... "adaption_technique", adaption_method.lms); ffe.constellation = unique(d); ffe.equalize(x, d, 0, 1, numel(x), true, false); testCase.verifyEqual(ffe.e_dc, 0.36, "AbsTol", 1e-12); end function ffeBufferDelaysTapUpdateUntilBlockBoundary(testCase) x = ones(4, 1); d = ones(4, 1); ffe = FFE( ... "sps", 1, ... "order", 1, ... "dc_tracking_mu", 0, ... "ffe_update_buffer_len", 2, ... "dd_mode", 0, ... "adaption_technique", adaption_method.lms); ffe.constellation = unique(d); ffe.equalize(x, d, 0.1, 1, numel(x), true, false); testCase.verifyEqual(ffe.e, 0.19, "AbsTol", 1e-12); end function dcAvgA1BufferlengthOneLeavesInputUnchanged(testCase) x = (1:4).'; d = zeros(4, 1); ffe = FFE( ... "sps", 1, ... "order", 1, ... "dc_avg_bufferlength_a1", 1, ... "save_debug", true, ... "dd_mode", 0, ... "adaption_technique", adaption_method.lms); ffe.e = 1; [y,~] = ffe.equalize(x, d, 0, 1, numel(x), true, false); testCase.verifyEqual(y, x, "AbsTol", 1e-12); testCase.verifyEqual(ffe.debug_struct.dc_avg_offset, zeros(1,4), "AbsTol", 1e-12); end function dcAvgA1SubtractsCausalMovingAverageByDefault(testCase) x = (1:4).'; d = zeros(4, 1); expected_offset = [0, 0, 0, 2]; expected_y = [1; 2; 3; 2]; ffe = FFE( ... "sps", 1, ... "order", 1, ... "dc_avg_bufferlength_a1", 3, ... "save_debug", true, ... "dd_mode", 0, ... "adaption_technique", adaption_method.lms); ffe.e = 1; [y,~] = ffe.equalize(x, d, 0, 1, numel(x), true, false); testCase.verifyEqual(y, expected_y, "AbsTol", 1e-12); testCase.verifyEqual(ffe.debug_struct.dc_avg_offset, expected_offset, "AbsTol", 1e-12); end function dcAvgA1SmoothingIsReservedForOfflineVariants(testCase) x = (1:4).'; d = zeros(4, 1); expected_offset = [0, 0, 0, 2]; expected_y = [1; 2; 3; 2]; ffe = FFE( ... "sps", 1, ... "order", 1, ... "dc_avg_bufferlength_a1", 3, ... "dc_smoothing_a1", 0.5, ... "save_debug", true, ... "dd_mode", 0, ... "adaption_technique", adaption_method.lms); ffe.e = 1; [y,~] = ffe.equalize(x, d, 0, 1, numel(x), true, false); testCase.verifyEqual(y, expected_y, "AbsTol", 1e-12); testCase.verifyEqual(ffe.debug_struct.dc_avg_offset, expected_offset, "AbsTol", 1e-12); end function dcAvgA1UpdateBlocklengthOneRefreshesEverySymbol(testCase) x = (1:4).'; d = zeros(4, 1); expected_offset = [0, 1, 1.5, 2]; expected_y = [1; 1; 1.5; 2]; ffe = FFE( ... "sps", 1, ... "order", 1, ... "dc_avg_bufferlength_a1", 3, ... "dc_avg_update_blocklength_a1", 1, ... "save_debug", true, ... "dd_mode", 0, ... "adaption_technique", adaption_method.lms); ffe.e = 1; [y,~] = ffe.equalize(x, d, 0, 1, numel(x), true, false); testCase.verifyEqual(y, expected_y, "AbsTol", 1e-12); testCase.verifyEqual(ffe.debug_struct.dc_avg_offset, expected_offset, "AbsTol", 1e-12); end function dcAvgA1WarnsWhenCombinedWithDcTracking(testCase) testCase.verifyWarning(@() FFE( ... "dc_avg_bufferlength_a1", 3, ... "dc_tracking_mu", 0.1), "FFE:DCAvgWithDcTracking"); end function dcLevelAvgA2SubtractsScalarWeightedResidual(testCase) x = (1:4).'; d = zeros(4, 1); expected_y = [1; 2; 1.5; 2.5]; ffe = FFE( ... "sps", 1, ... "order", 1, ... "dc_level_avg_bufferlength_a2", 2, ... "dc_level_weights_a2", 1, ... "save_debug", true, ... "dd_mode", 0, ... "adaption_technique", adaption_method.lms); ffe.e = 1; ffe.constellation = 0; [y,~] = ffe.equalize(x, d, 0, 1, numel(x), true, false); testCase.verifyEqual(y, expected_y, "AbsTol", 1e-12); testCase.verifyEqual(ffe.debug_struct.dc_level_mpi_est, [0, 0, 1.5, 1.5], "AbsTol", 1e-12); testCase.verifyEqual(ffe.debug_struct.dc_level_weight, [0, 0, 1, 1], "AbsTol", 1e-12); testCase.verifyEqual(ffe.debug_struct.dc_level_valid_count, [0, 0, 2, 2]); end function dcLevelAvgA2UpdateBlocklengthOneRefreshesEverySymbol(testCase) x = (1:4).'; d = zeros(4, 1); ffe = FFE( ... "sps", 1, ... "order", 1, ... "dc_level_avg_bufferlength_a2", 2, ... "dc_level_update_blocklength_a2", 1, ... "dc_level_weights_a2", 1, ... "save_debug", true, ... "dd_mode", 0, ... "adaption_technique", adaption_method.lms); ffe.e = 1; ffe.constellation = 0; [y,~] = ffe.equalize(x, d, 0, 1, numel(x), true, false); testCase.verifyEqual(y, [1; 1.5; 1.5; 1.5], "AbsTol", 1e-12); testCase.verifyEqual(ffe.debug_struct.dc_level_mpi_est, [0, 1, 1.5, 2.5], "AbsTol", 1e-12); testCase.verifyEqual(ffe.debug_struct.dc_level_weight, [0, 0.5, 1, 1], "AbsTol", 1e-12); testCase.verifyEqual(ffe.debug_struct.dc_level_valid_count, [0, 1, 2, 2]); end function dcLevelAvgA2UsesVectorWeightsInConstellationOrder(testCase) x = [1; 3; 1; 4]; d = [0; 2; 0; 2]; ffe = FFE( ... "sps", 1, ... "order", 1, ... "dc_level_avg_bufferlength_a2", 2, ... "dc_level_weights_a2", [0, 1], ... "dd_mode", 0, ... "adaption_technique", adaption_method.lms); ffe.e = 1; ffe.constellation = [0; 2]; [y,~] = ffe.equalize(x, d, 0, 1, numel(x), true, false); testCase.verifyEqual(y, [1; 3; 1; 3.5], "AbsTol", 1e-12); end function dcLevelAvgA2WarnsWhenCombinedWithDcTracking(testCase) testCase.verifyWarning(@() FFE( ... "dc_level_avg_bufferlength_a2", 2, ... "dc_level_weights_a2", 1, ... "dc_tracking_mu", 0.1), "FFE:DCLevelAvgWithOtherDcSuppression"); end function a2InitialWeightGuessFollowsLevelVarianceSlope(testCase) x = [ ... 0.98; 1.02; 1.00; 1.01; ... 1.90; 2.10; 2.00; 2.08; ... 2.50; 3.50; 3.00; 3.40]; d = [ ... ones(4,1); ... 2*ones(4,1); ... 3*ones(4,1)]; ffe = FFE("sps", 1, "a2_level_weight_max", 1); ffe.constellation = [1; 2; 3]; [weights,stats] = ffe.a2LevelWeightInitialGuess(x,d); testCase.verifySize(weights, [3, 1]); testCase.verifyGreaterThan(weights(3), weights(2)); testCase.verifyGreaterThan(weights(2), weights(1)); testCase.verifyEqual(max(weights), 0.7, "AbsTol", 1e-12); testCase.verifyGreaterThan(stats.variance_slope, 0); end end end function sig = makeSignal(values, fs) base = Signal(values); sig = Informationsignal(values, "fs", fs, "logbook", base.logbook); end