Final MPI Calculations here

This commit is contained in:
Silas Oettinghaus
2026-07-14 11:50:22 +02:00
parent f421348e5b
commit 8c4edf2490
8 changed files with 1003 additions and 12 deletions

View File

@@ -0,0 +1,254 @@
%% BER over SIR from MpiReductionResults
% 1) gather data from the database
% 2) clean data and remove per-curve outliers
% 3) plot BER over SIR grouped by algorithm
clear; clc;
%% 1) Gather data
studyName = "pam4_112_greater_3153";
pathLengthToPlot = 0;
selectedBlockUpdate = 1; % set [] to pool all block_update values
selectedPamLevels = []; % set [] to use all PAM levels in the query result
selectedAlgorithms = []; % set [] to use all algorithms in the query result
useBoundedLines = true;
usePolyfit = true;
polyfitOrderMax = 4;
maxBerForPlot = 0.1;
db = DBHandler( ...
"dataBase", "labor", ...
"type", "mysql", ...
"server", "192.168.178.192", ...
"user", "silas", ...
"password", "silas");
db.refresh();
fp = QueryFilter();
fp.where('Runs', 'interference_path_length', 'EQUALS', pathLengthToPlot);
fp.where('Runs', 'fiber_length', 'EQUALS', 0);
fp.where('Runs', 'db_mode', 'EQUALS', '"no_db"');
fp.where('MpiReductionResults', 'study_name', 'EQUALS', char(studyName));
if ~isempty(selectedBlockUpdate)
fp.where('MpiReductionResults', 'block_update', 'EQUALS', selectedBlockUpdate);
end
selectedFields = { ...
'Runs.run_id', ...
'Runs.sir', ...
'Runs.pam_level', ...
'Runs.symbolrate', ...
'Runs.interference_path_length', ...
'Runs.power_mpi_interference', ...
'MpiReductionResults.occurrence_idx', ...
'MpiReductionResults.storage_name', ...
'MpiReductionResults.algorithm', ...
'MpiReductionResults.algorithm_variant', ...
'MpiReductionResults.eq_class', ...
'MpiReductionResults.block_update', ...
'MpiReductionResults.BER', ...
'MpiReductionResults.BER_precoded', ...
'MpiReductionResults.SNR', ...
'MpiReductionResults.GMI', ...
'MpiReductionResults.AIR'};
selectedFields = selectedFields(:);
[rawData, query] = db.queryDB(fp, selectedFields);
disp(query);
fprintf("Fetched %d MPI reduction rows.\n", height(rawData));
%% 2) Clean data
data = rawData;
numericFields = ["run_id", "sir", "pam_level", "symbolrate", ...
"interference_path_length", "occurrence_idx", "block_update", ...
"power_mpi_interference", "BER", "BER_precoded", "SNR", "GMI", "AIR"];
for fieldIdx = 1:numel(numericFields)
fieldName = numericFields(fieldIdx);
if ismember(fieldName, string(data.Properties.VariableNames))
data.(char(fieldName)) = numericColumn(data.(char(fieldName)));
end
end
stringFields = ["storage_name", "algorithm", "algorithm_variant", "eq_class"];
for fieldIdx = 1:numel(stringFields)
fieldName = stringFields(fieldIdx);
if ismember(fieldName, string(data.Properties.VariableNames))
data.(char(fieldName)) = stringColumn(data.(char(fieldName)));
end
end
data = data(isfinite(data.BER) & data.BER > 0 & data.BER < maxBerForPlot, :);
data.sir_exact = -7 - data.power_mpi_interference;
if isempty(data)
warning("PLOT_mpi_reduction_db_ber_vs_sir:NoRows", ...
"No rows remain after initial BER/path/study/block filtering.");
return
end
if isempty(selectedPamLevels)
selectedPamLevels = unique(data.pam_level(isfinite(data.pam_level))).';
else
data = data(ismember(data.pam_level, selectedPamLevels), :);
end
if isempty(selectedAlgorithms)
selectedAlgorithms = unique(data.algorithm, "stable").';
else
selectedAlgorithms = string(selectedAlgorithms);
data = data(ismember(data.algorithm, selectedAlgorithms), :);
end
if isempty(data)
warning("PLOT_mpi_reduction_db_ber_vs_sir:NoSelectedRows", ...
"No rows remain after selectedPamLevels/selectedAlgorithms filtering.");
return
end
data.clean_keep = true(height(data), 1);
[groupId, groupPam, groupAlgorithm, groupSir] = findgroups(data.pam_level, data.algorithm, data.sir);
for groupIdx = 1:max(groupId)
rowMask = groupId == groupIdx;
berValues = data.BER(rowMask);
if nnz(rowMask) > 3
data.clean_keep(rowMask) = ~isoutlier(berValues);
end
end
cleanData = data(data.clean_keep, :);
summaryTable = groupsummary(cleanData, ["pam_level", "algorithm", "sir"], ...
{"mean", "min", "max"}, "BER");
sirExactTable = groupsummary(cleanData, ["pam_level", "algorithm", "sir"], ...
"median", "sir_exact");
summaryTable.sir_exact = sirExactTable.median_sir_exact;
summaryTable = sortrows(summaryTable, ["pam_level", "algorithm", "sir"]);
fprintf("Cleaned to %d rows across %d PAM/algorithm/SIR groups.\n", ...
height(cleanData), height(summaryTable));
disp(groupcounts(cleanData, ["pam_level", "algorithm"]));
%% 3) Plot data
if exist("linspecer", "file")
algColors = linspecer(numel(selectedAlgorithms));
else
algColors = lines(numel(selectedAlgorithms));
end
figure(); clf;
tiledlayout(numel(selectedPamLevels), 1, "TileSpacing", "compact");
for pamIdx = 1:numel(selectedPamLevels)
pamLevel = selectedPamLevels(pamIdx);
nexttile; hold on;
for algIdx = 1:numel(selectedAlgorithms)
algorithmName = selectedAlgorithms(algIdx);
algColor = algColors(algIdx, :);
rawMask = cleanData.pam_level == pamLevel & cleanData.algorithm == algorithmName;
curveMask = summaryTable.pam_level == pamLevel & summaryTable.algorithm == algorithmName;
if ~any(curveMask)
continue
end
scatter(cleanData.sir_exact(rawMask), cleanData.BER(rawMask), ...
30, ...
"Marker", ".", ...
"MarkerEdgeColor", algColor, ...
"HandleVisibility", "off");
sirValues = summaryTable.sir_exact(curveMask).';
meanBer = summaryTable.mean_BER(curveMask).';
minBer = summaryTable.min_BER(curveMask).';
maxBer = summaryTable.max_BER(curveMask).';
valid = isfinite(sirValues) & isfinite(meanBer) & meanBer > 0;
if useBoundedLines && exist("boundedline", "file") && any(valid)
yLower = max(meanBer - minBer, 0);
yUpper = max(maxBer - meanBer, 0);
yBounds = [yLower(:), yUpper(:)];
[hl, hp] = boundedline(sirValues(valid).', meanBer(valid).', yBounds(valid, :), ...
'alpha', 'transparency', 0.08, ...
'cmap', algColor, ...
'nan', 'fill', ...
'orientation', 'vert');
set(hl, "LineStyle", "none", "Marker", "none", "HandleVisibility", "off");
set(hp, "LineStyle", "none", "HandleVisibility", "off");
end
plot(sirValues(valid), meanBer(valid), ...
"LineStyle", "none", ...
"Marker", "o", ...
"MarkerSize", 5, ...
"LineWidth", 1.2, ...
"Color", algColor, ...
"MarkerFaceColor", algColor, ...
"DisplayName", char(algorithmName));
if usePolyfit
fitMask = valid & meanBer > 0;
if nnz(fitMask) >= 2
fitOrder = min(polyfitOrderMax, nnz(fitMask) - 1);
fitCoeff = polyfit(sirValues(fitMask), log10(meanBer(fitMask)), fitOrder);
xFit = linspace(min(sirValues(fitMask)), max(sirValues(fitMask)), 300);
yFit = 10 .^ polyval(fitCoeff, xFit);
plot(xFit, yFit, ...
"LineStyle", "--", ...
"LineWidth", 1.1, ...
"Color", algColor, ...
"HandleVisibility", "off");
end
end
end
yline(2.2e-4, "LineWidth", 1, "LineStyle", "--", "HandleVisibility", "off");
yline(3.8e-3, "LineWidth", 1, "LineStyle", "--", "HandleVisibility", "off");
yline(2e-2, "LineWidth", 1, "LineStyle", "--", "HandleVisibility", "off");
title(sprintf("PAM %.0f, path %.0f m, block update %s", ...
pamLevel, pathLengthToPlot, blockUpdateLabel(selectedBlockUpdate)));
xlabel("SIR (dB)");
ylabel("BER");
set(gca, "YScale", "log");
ylim([9e-5, maxBerForPlot]);
grid on;
box on;
legend("Location", "best", "Interpreter", "none");
end
if exist("beautifyBERplot", "file")
beautifyBERplot("logscale", true, "setcolors", false, "setmarkers", false);
end
%% Local helpers
function values = numericColumn(values)
if iscell(values)
values = string(values);
end
if isstring(values) || ischar(values)
values = str2double(values);
end
values = double(values);
end
function values = stringColumn(values)
if iscell(values)
values = string(values);
elseif ischar(values)
values = string(values);
end
values = strip(string(values));
end
function label = blockUpdateLabel(selectedBlockUpdate)
if isempty(selectedBlockUpdate)
label = "pooled";
else
label = string(selectedBlockUpdate);
end
end

View File

@@ -0,0 +1,99 @@
function [summary, rows] = appendMpiReductionDspOutput(db, run_id, occurrence_idx, dspOutput, dsp_options, options)
%appendMpiReductionDspOutput Store one DSP occurrence in MpiReductionResults.
arguments
db
run_id
occurrence_idx
dspOutput struct
dsp_options struct = struct()
options.study_name string = "mpi_reduction_v1"
options.dry_run (1,1) logical = false
options.verbose (1,1) logical = false
end
summary = emptySummary();
rows = {};
packageNames = fieldnames(dspOutput);
if isempty(packageNames)
return
end
block_update = resolveBlockUpdate(dspOutput, dsp_options);
wh = DataStorage(struct( ...
"run_id", double(run_id), ...
"block_update", double(block_update)));
hasPackages = false;
for packageIdx = 1:numel(packageNames)
packageName = packageNames{packageIdx};
package = dspOutput.(packageName);
if isempty(package)
continue
end
if iscell(package)
packageCell = package;
else
packageCell = {package};
end
wh.addStorage(packageName);
wh.addValueToStorageByLinIdx(packageCell, packageName, 1);
hasPackages = true;
end
if ~hasPackages
return
end
dsp_options_local = dsp_options;
dsp_options_local.start_occurence = occurrence_idx;
dsp_options_local.userParameters.block_update = block_update;
[summary, rows] = appendMpiReductionWarehouse(db, wh, run_id, dsp_options_local, ...
"study_name", options.study_name, ...
"dry_run", options.dry_run, ...
"verbose", options.verbose);
end
function summary = emptySummary()
summary = struct( ...
"rows_considered", 0, ...
"rows_inserted", 0, ...
"rows_skipped_duplicate", 0, ...
"rows_skipped_existing_key", 0, ...
"rows_skipped_empty", 0);
end
function block_update = resolveBlockUpdate(dspOutput, dsp_options)
if isfield(dsp_options, "userParameters") && ...
isfield(dsp_options.userParameters, "block_update") && ...
isscalar(dsp_options.userParameters.block_update)
block_update = dsp_options.userParameters.block_update;
return
end
packageNames = fieldnames(dspOutput);
for packageIdx = 1:numel(packageNames)
package = dspOutput.(packageNames{packageIdx});
if isempty(package)
continue
end
if iscell(package)
package = package{1};
end
if isstruct(package) && ...
isfield(package, "mpi_reduction_config") && ...
isfield(package.mpi_reduction_config, "params") && ...
isfield(package.mpi_reduction_config.params, "block_update")
block_update = package.mpi_reduction_config.params.block_update;
return
end
end
error("appendMpiReductionDspOutput:MissingBlockUpdate", ...
"Could not resolve scalar block_update from dsp_options or package metadata.");
end

View File

@@ -0,0 +1,307 @@
function [summary, rows] = appendMpiReductionWarehouse(db, wh, run_ids, dsp_options, options)
%appendMpiReductionWarehouse Store MPI reduction warehouse entries in MySQL.
arguments
db
wh
run_ids
dsp_options struct
options.study_name string = "mpi_reduction_v1"
options.dry_run (1,1) logical = false
options.verbose (1,1) logical = true
end
ensureTableVisible(db);
storageNames = fieldnames(wh.sto);
summary = struct( ...
"rows_considered", 0, ...
"rows_inserted", 0, ...
"rows_skipped_duplicate", 0, ...
"rows_skipped_existing_key", 0, ...
"rows_skipped_empty", 0);
rows = {};
for storageIdx = 1:numel(storageNames)
storageName = storageNames{storageIdx};
storageArray = wh.sto.(storageName);
for linIdx = 1:numel(storageArray)
packageCell = storageArray{linIdx};
if isempty(packageCell)
summary.rows_skipped_empty = summary.rows_skipped_empty + 1;
continue
end
phys = physicalCoordinates(wh, linIdx);
run_id = resolveRunId(phys, run_ids);
block_update = resolveBlockUpdate(phys, dsp_options);
packageCell = normalizePackageCell(packageCell);
for packageIdx = 1:numel(packageCell)
package = packageCell{packageIdx};
if isempty(package) || ~isstruct(package) || ~isfield(package, "metrics")
summary.rows_skipped_empty = summary.rows_skipped_empty + 1;
continue
end
occurrence_idx = startOccurrence(dsp_options) + packageIdx - 1;
row = buildRow(db, package, storageName, run_id, occurrence_idx, ...
block_update, options.study_name);
summary.rows_considered = summary.rows_considered + 1;
rows{end+1,1} = row; %#ok<AGROW>
if options.dry_run
continue
end
if resultHashExists(db, row.result_hash)
summary.rows_skipped_duplicate = summary.rows_skipped_duplicate + 1;
if options.verbose
fprintf("MPI reduction DB: skipped duplicate %s, run_id=%d, occurrence=%d, block_update=%d\n", ...
row.storage_name, row.run_id, row.occurrence_idx, row.block_update);
end
continue
end
if uniqueResultKeyExists(db, row)
summary.rows_skipped_existing_key = summary.rows_skipped_existing_key + 1;
if options.verbose
fprintf("MPI reduction DB: skipped existing key %s, run_id=%d, occurrence=%d, block_update=%d\n", ...
row.storage_name, row.run_id, row.occurrence_idx, row.block_update);
end
continue
end
db.appendToTable("MpiReductionResults", row);
summary.rows_inserted = summary.rows_inserted + 1;
end
end
end
if options.verbose
fprintf("MPI reduction DB: considered %d rows, inserted %d, duplicates %d, existing keys %d, empty %d\n", ...
summary.rows_considered, summary.rows_inserted, ...
summary.rows_skipped_duplicate, summary.rows_skipped_existing_key, ...
summary.rows_skipped_empty);
end
end
function ensureTableVisible(db)
if ~isfield(db.tables, "MpiReductionResults")
db.refresh();
end
if ~isfield(db.tables, "MpiReductionResults")
error("appendMpiReductionWarehouse:MissingTable", ...
"Table MpiReductionResults is not visible to DBHandler. Create the table and call db.refresh().");
end
end
function phys = physicalCoordinates(wh, linIdx)
phys = struct();
[physValues, physNames] = wh.getPhysIndicesByLinIndex(linIdx);
for physIdx = 1:numel(physNames)
name = char(physNames{physIdx});
phys.(name) = physValues{physIdx};
end
end
function run_id = resolveRunId(phys, run_ids)
if isfield(phys, "run_id")
run_id = phys.run_id;
return
end
if isscalar(run_ids)
run_id = run_ids;
return
end
error("appendMpiReductionWarehouse:MissingRunId", ...
"Warehouse has no run_id axis, but run_ids is not scalar.");
end
function block_update = resolveBlockUpdate(phys, dsp_options)
if isfield(phys, "block_update")
block_update = phys.block_update;
return
end
if isfield(dsp_options, "userParameters") && ...
isfield(dsp_options.userParameters, "block_update") && ...
isscalar(dsp_options.userParameters.block_update)
block_update = dsp_options.userParameters.block_update;
return
end
error("appendMpiReductionWarehouse:MissingBlockUpdate", ...
"Could not resolve block_update from warehouse coordinates or scalar userParameters.");
end
function packageCell = normalizePackageCell(packageCell)
if ~iscell(packageCell)
packageCell = {packageCell};
end
end
function occurrence = startOccurrence(dsp_options)
occurrence = 1;
if isfield(dsp_options, "start_occurence")
occurrence = dsp_options.start_occurence;
end
end
function row = buildRow(db, package, storageName, run_id, occurrence_idx, block_update, study_name)
config = fieldOr(package, "mpi_reduction_config", fallbackConfig(storageName, block_update));
params = fieldOr(config, "params", struct());
if isempty(params)
params = struct();
end
metrics = package.metrics;
if isobject(metrics) && ismethod(metrics, "toStruct")
metrics = metrics.toStruct();
end
param_json = jsonencode(params);
if strlength(string(param_json)) == 0
param_json = "{}";
end
row = struct();
row.run_id = double(run_id);
row.occurrence_idx = double(occurrence_idx);
row.study_name = char(study_name);
row.storage_name = char(fieldOr(config, "storage_name", storageName));
row.algorithm = char(fieldOr(config, "algorithm", "unknown"));
row.algorithm_variant = char(fieldOr(config, "algorithm_variant", ""));
row.eq_class = char(fieldOr(config, "eq_class", "unknown"));
row.block_update = double(block_update);
row.sps = scalarParam(params, "sps");
row.eq_order = scalarParam(params, "order");
row.len_tr = scalarParam(params, "len_tr");
row.epochs_tr = scalarParam(params, "epochs_tr");
row.mu_tr = scalarParam(params, "mu_tr");
row.dd_mode = logicalParam(params, "dd_mode");
row.epochs_dd = scalarParam(params, "epochs_dd");
row.mu_dd = scalarParam(params, "mu_dd");
row.numBits = metricScalar(metrics, "numBits", 0);
row.numBitErr = metricScalar(metrics, "numBitErr", 0);
row.BER = metricScalar(metrics, "BER", NaN);
row.numBitErr_precoded = metricScalar(metrics, "numBitErr_precoded", NaN);
row.BER_precoded = metricScalar(metrics, "BER_precoded", NaN);
row.SNR = metricScalar(metrics, "SNR", NaN);
row.SNR_level = metricJson(metrics, "SNR_level");
row.GMI = metricScalar(metrics, "GMI", NaN);
row.AIR = metricScalar(metrics, "AIR", NaN);
row.EVM = metricScalar(metrics, "EVM", NaN);
row.EVM_level = metricJson(metrics, "EVM_level");
row.STD = metricScalar(metrics, "STD", NaN);
row.STD_level = metricJson(metrics, "STD_level");
row.STDrx = metricScalar(metrics, "STDrx", NaN);
row.STDrx_level = metricJson(metrics, "STDrx_level");
row.Alpha = firstFiniteScalar(fieldOr(metrics, "Alpha", NaN));
row.param_json = char(param_json);
row.param_hash = db.calcHash(params);
row.result_hash = db.calcHash(row);
end
function config = fallbackConfig(storageName, block_update)
params = struct("block_update", block_update);
config = struct( ...
"storage_name", storageName, ...
"algorithm", "unknown", ...
"algorithm_variant", "", ...
"eq_class", "unknown", ...
"params", params);
end
function value = fieldOr(s, fieldName, defaultValue)
fieldName = char(fieldName);
if isstruct(s) && isfield(s, fieldName) && ~isempty(s.(fieldName))
value = s.(fieldName);
else
value = defaultValue;
end
end
function value = scalarParam(params, fieldName)
value = firstFiniteScalar(fieldOr(params, fieldName, NaN));
end
function value = logicalParam(params, fieldName)
value = firstFiniteScalar(fieldOr(params, fieldName, false));
if isnan(value)
value = false;
else
value = logical(value);
end
end
function value = metricScalar(metrics, fieldName, defaultValue)
value = firstFiniteScalar(fieldOr(metrics, fieldName, defaultValue));
end
function value = metricJson(metrics, fieldName)
metricValue = fieldOr(metrics, fieldName, []);
value = char(jsonencode(metricValue));
end
function value = firstFiniteScalar(valueIn)
if islogical(valueIn)
valueIn = double(valueIn);
end
if isempty(valueIn)
value = NaN;
return
end
if isnumeric(valueIn)
finiteValues = valueIn(isfinite(valueIn));
if isempty(finiteValues)
value = NaN;
else
value = double(finiteValues(1));
end
return
end
value = str2double(string(valueIn));
if isnan(value)
value = NaN;
end
end
function exists = resultHashExists(db, resultHash)
query = sprintf("SELECT mpi_result_id FROM MpiReductionResults WHERE result_hash = '%s' LIMIT 1", ...
char(resultHash));
existing = db.fetch(query);
exists = ~isempty(existing);
end
function exists = uniqueResultKeyExists(db, row)
query = sprintf( ...
"SELECT mpi_result_id FROM MpiReductionResults " + ...
"WHERE run_id = %d " + ...
"AND occurrence_idx = %d " + ...
"AND study_name = '%s' " + ...
"AND storage_name = '%s' " + ...
"AND block_update = %d " + ...
"AND param_hash = '%s' LIMIT 1", ...
row.run_id, row.occurrence_idx, ...
sqlString(row.study_name), sqlString(row.storage_name), ...
row.block_update, sqlString(row.param_hash));
existing = db.fetch(query);
exists = ~isempty(existing);
end
function value = sqlString(value)
value = strrep(char(value), '''', '''''');
end

View File

@@ -7,6 +7,12 @@ dsp_options.start_occurence = 1;
dsp_options.max_occurences = 15;
dsp_options.debug_plots = false;
write_mpi_reduction_db = 1;
stream_mpi_reduction_db = 1;
mpi_reduction_study_name = "block_update_sweep";
mpi_reduction_writer_path = fullfile(fileparts(mfilename('fullpath')),"db");
addpath(mpi_reduction_writer_path);
dsp_options.database_type = "mysql";
dsp_options.dataBase = "labor";
@@ -16,6 +22,9 @@ dsp_options.server = "192.168.178.192";
dsp_options.port = 3306;
dsp_options.user = "silas";
dsp_options.password = "silas";
dsp_options.append_mpi_reduction_db = write_mpi_reduction_db && stream_mpi_reduction_db;
dsp_options.mpi_reduction_study_name = mpi_reduction_study_name;
dsp_options.mpi_reduction_writer_path = mpi_reduction_writer_path;
db = DBHandler("dataBase", [dsp_options.dataBase],...
"type", dsp_options.database_type,...
@@ -45,7 +54,6 @@ for i = 1
[dataTable,~] = db.queryDB(fp, db.getTableFieldNames('Runs'));
[~, sortIdx] = sort(dataTable.sir, 'descend');
dataTable = dataTable(sortIdx, :);
@@ -55,11 +63,10 @@ for i = 1
%% === Warehouse setup ===
dsp_options.userParameters = struct();
dsp_options.userParameters.block_update = 1;%[1,2,4,8,16,32,64,112,224,448,512,1024,2048,2048*2,2048*4];%%logspace(-3.8,-1,22);%[linspace(2,4096,22)];
dsp_options.userParameters.block_update = [1,2,4,8,16,32,64,112,224,448,448*2,1024,2048,4096,8192,16384];%%logspace(-3.8,-1,22);%[linspace(2,4096,22)];
% dsp_options.userParameters.block_update = linspace(1,224,22);
wh = DataStorage(dsp_options.userParameters);
%%
n_realizations = (dsp_options.max_occurences - dsp_options.start_occurence + 1);
n_userparams = prod(wh.dim);
@@ -76,6 +83,14 @@ for i = 1
[results, wh] = submitJobs(run_ids, dsp_options, processingMode.parallel, ...
"wh", wh, ...
"waitbar", true);
%%
if write_mpi_reduction_db && ~stream_mpi_reduction_db
db.refresh();
appendMpiReductionWarehouse(db, wh, run_ids, dsp_options, ...
"study_name", mpi_reduction_study_name);
end
% savepath = fullfile("C:","Users","Silas","Documents","MATLAB","imdd_simulation","projects","Diss","MPI_revisit","algorithms",sprintf("pam_%d_results.mat",M));
% save(wh,savepath)