308 lines
10 KiB
Matlab
308 lines
10 KiB
Matlab
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
|