silas/imdd_silas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

restructure and organize

Browse Source
This commit is contained in:
Silas Oettinghaus
2026-06-22 22:58:58 +02:00
parent c4f75b7ec4
commit 33ec5b3116
36 changed files with 1914 additions and 674 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
Functions/Job_Processing/dsp_runid.m Normal file
View File

@@ -0,0 +1,107 @@
function output = dsp_runid(run_id, options)
arguments
run_id
options.append_to_db = 0;
options.max_occurences = 4;
options.start_occurence = 1;
options.userParameters = struct();
options.database_type
options.dataBase
options.server = "134.245.243.254";
options.port = 3306;
options.user = "silas";
options.password = "silas";
options.load_file_path = struct();
options.storage_path
options.mode
options.recipe = @dsp_scope_signal;
options.debug_plots (1,1) logical = false;
end
try
output = initializeOutput();
database = [];
inputSource = normalizeDspInputSource(options.mode);
if inputSource == "run_id" || options.append_to_db
database = DBHandler("dataBase", [options.dataBase], "type", options.database_type, ...
"user", options.user, "password", options.password, ...
"server", options.server, "port", options.port);
end
switch inputSource
case "run_id"
dspInput = loadDspInputFromRunId(run_id, database, options);
case "file_paths"
dspInput = loadDspInputFromFilePaths(run_id, options);
end
num_occurences = length(dspInput.Scpe_cell);
for r = 1:num_occurences
%%%%%%%% CORE EQUALIZATION CALL (Scpe, Symbols, Bits, 'Options') %%%%%%%
dspOutput = options.recipe(dspInput.Scpe_cell{r}, dspInput.Symbols, dspInput.Tx_bits, ...
"fsym", dspInput.fsym, ...
"M", dspInput.M, ...
"duob_mode", dspInput.duob_mode, ...
"dataTable", dspInput.dataTable, ...
"userParameters", options.userParameters, ...
"debug_plots", options.debug_plots);
%%%%%%%% CORE EQUALIZATION CALL %%%%%%%
fieldNames = fieldnames(dspOutput);
for fieldIdx = 1:numel(fieldNames)
fieldName = fieldNames{fieldIdx};
output.(fieldName){r} = dspOutput.(fieldName);
end
if options.append_to_db
appendDspOutputToDatabase(database, run_id, dspOutput);
end
end
catch ME
save('workerError.mat', 'ME');
rethrow(ME);
end
end
function output = initializeOutput()
output.ffe_package = {};
output.dfe_package = {};
output.mlse_package = {};
output.vnle_package = {};
output.dbtgt_package = {};
output.dbenc_package = {};
output.mlmlse_package = {};
end
function inputSource = normalizeDspInputSource(mode)
mode = string(mode);
switch mode
case {"load_run_id", "run_id"}
inputSource = "run_id";
case {"load_files", "file_paths"}
inputSource = "file_paths";
otherwise
error('dsp_runid:UnsupportedMode', ...
'Mode "%s" is not supported. Use "run_id" or "file_paths".', mode);
end
end
function appendDspOutputToDatabase(database, run_id, dspOutput)
packageNames = fieldnames(dspOutput);
for i = 1:numel(packageNames)
package = dspOutput.(packageNames{i});
if isempty(package)
continue
end
database.addProcessingResult(run_id, package.metrics, package.config);
end
end

196
Functions/Job_Processing/loadAndSyncRunSignals.m Normal file
View File

@@ -0,0 +1,196 @@
function [Bits, Symbols, Scpe_cell, found_sync] = loadAndSyncRunSignals(dataTable, options)
%LOADANDSYNCRUNSIGNALS Load and synchronize signal files for one run.
%
% Inputs:
% dataTable - one-row table with run metadata and signal file paths
% options - struct with storage_path, start_occurence and max_occurences
%
% Outputs:
% Bits - transmitted bit reference
% Symbols - transmitted symbol reference
% Scpe_cell - synchronized received signal occurrences
% found_sync - true when a valid synchronization was found
found_sync = 0;
tempLocalStorage = 1;
Scpe_cell = {};
loaded_from_cache = false;
storage_dir = fullfile(prefdir, 'temp_sync_data');
if tempLocalStorage == 1
local_filename = fullfile(storage_dir, sprintf('sync_data_run_%s.mat', num2str(dataTable.run_id)));
if exist(local_filename, 'file')
try
cacheData = load(local_filename, 'Bits', 'Symbols', 'Scpe_cell');
if isValidSyncCache(cacheData)
Bits = cacheData.Bits;
Symbols = cacheData.Symbols;
Scpe_cell = cacheData.Scpe_cell;
found_sync = 1;
loaded_from_cache = true;
else
warning('loadAndSyncRunSignals:InvalidSyncCache', ...
'Ignoring incomplete sync cache file "%s".', local_filename);
safeDelete(local_filename);
end
catch
safeDelete(local_filename);
found_sync = 0;
end
end
end
if ~found_sync
Bits = load(composeStoragePath(options.storage_path, dataTable.tx_bits_path));
Bits = Bits.Bits;
M = double(dataTable.pam_level);
fsym = dataTable.symbolrate;
Symbols_mapped = PAMmapper(M,0).map(Bits);
Symbols_mapped.fs = fsym;
Symbols = load(composeStoragePath(options.storage_path, dataTable.tx_symbols_path));
Symbols = Symbols.Symbols;
found_sync = 0;
try
Scpe_load = load(composeStoragePath(options.storage_path, dataTable.rx_sync_path));
Scpe_cell = Scpe_load.S;
[~,~,~,found_sync] = Scpe_cell{1}.tsynch("reference", Symbols, ...
"fs_ref", fsym, ...
"debug_plots", 0);
catch
% Continue with raw data if pre-synchronized data is unavailable.
end
end
if ~found_sync
try
Scpe_sig_raw = load(composeStoragePath(options.storage_path, dataTable.rx_raw_path));
Scpe_sig_raw = Scpe_sig_raw.Scpe_sig_raw;
Scpe_sig_resampled = Scpe_sig_raw.resample("fs_in", Scpe_sig_raw.fs, "fs_out", 2*fsym);
[~, Scpe_cell, ~, found_sync] = Scpe_sig_resampled.tsynch("reference", Symbols, ...
"fs_ref", fsym, ...
"debug_plots", 0);
catch
% Continue to mapped-symbol fallback if raw data sync fails.
end
end
if ~found_sync && exist('Scpe_sig_raw', 'var')
if length(Symbols_mapped.signal) ~= sum(Symbols_mapped.signal == Symbols.signal)
[~, Scpe_cell, ~, found_sync] = Scpe_sig_raw.tsynch("reference", Symbols_mapped, ...
"fs_ref", fsym, ...
"debug_plots", 0);
end
end
if tempLocalStorage == 1 && found_sync && ~loaded_from_cache
if ~exist(storage_dir, 'dir')
mkdir(storage_dir);
end
max_local_files = 10;
files = dir(fullfile(storage_dir, 'sync_data_run_*.mat'));
if length(files) >= max_local_files
[~, fileOrder] = sort([files.datenum]);
delete(fullfile(storage_dir, files(fileOrder(1)).name));
end
writeSyncCache(local_filename, Bits, Symbols, Scpe_cell);
end
if found_sync
Scpe_cell = selectSyncedOccurrences(Scpe_cell, options);
else
warning('Could not synchronize the received signal with the stored symbols!');
end
end
function path = composeStoragePath(storagePath, relativePath)
relativePath = firstValue(relativePath);
path = char(string(storagePath) + string(relativePath));
end
function value = firstValue(value)
if iscell(value)
value = value{1};
elseif ~ischar(value) && ~isscalar(value)
value = value(1);
end
end
function Scpe_cell = selectSyncedOccurrences(Scpe_cell, options)
available_occurences = length(Scpe_cell);
start_occurence = floor(getOption(options, 'start_occurence', 1));
max_occurences = floor(getOption(options, 'max_occurences', available_occurences));
if available_occurences < 1
warning('loadAndSyncRunSignals:NoSyncedOccurrences', ...
'Synchronization reported success, but no synced occurrences are available.');
return
end
if start_occurence < 1
error('loadAndSyncRunSignals:InvalidStartOccurrence', ...
'start_occurence must be >= 1.');
end
if max_occurences < 1
Scpe_cell = Scpe_cell(1:0);
return
end
if start_occurence > available_occurences
warning('loadAndSyncRunSignals:StartOccurrenceTooHigh', ...
['Requested start_occurence %d, but only %d synced occurrences are available. ', ...
'Processing the last occurrence only.'], ...
start_occurence, available_occurences);
Scpe_cell = Scpe_cell(available_occurences);
return
end
stop_occurence = min(available_occurences, start_occurence + max_occurences - 1);
Scpe_cell = Scpe_cell(start_occurence:stop_occurence);
end
function value = getOption(options, name, defaultValue)
if isfield(options, name)
value = options.(name);
else
value = defaultValue;
end
end
function valid = isValidSyncCache(cacheData)
valid = isfield(cacheData, 'Bits') && ...
isfield(cacheData, 'Symbols') && ...
isfield(cacheData, 'Scpe_cell') && ...
iscell(cacheData.Scpe_cell) && ...
~isempty(cacheData.Scpe_cell);
end
function writeSyncCache(local_filename, Bits, Symbols, Scpe_cell)
cache_dir = fileparts(local_filename);
temp_filename = [tempname(cache_dir), '.mat'];
try
save(temp_filename, 'Bits', 'Symbols', 'Scpe_cell');
movefile(temp_filename, local_filename, 'f');
catch ME
safeDelete(temp_filename);
rethrow(ME);
end
end
function safeDelete(filename)
if exist(filename, 'file')
try
delete(filename);
catch
% Another parallel worker may already have removed or replaced it.
end
end
end

114
Functions/Job_Processing/loadAndSyncSignalDataFromDb.m
View File

@@ -1,114 +0,0 @@
function [Bits, Symbols, Scpe_cell, found_sync] = loadAndSyncSignalDataFromDb(dataTable, options)
% LOADSIGNALDATA Loads and synchronizes signal data from storage
%
% Inputs:d
% dataTable - Table with file paths and configuration
% options - Struct with storage_path and max_occurences
%
% Outputs:
% Symbols_mapped - Mapped symbols from bits
% Symbols - Original symbols
% Scpe_cell - Cell array of synchronized signals
% found_sync - Boolean indicating if synchronization was successful
found_sync = 0;
tempLocalStorage = 1;
% Define the fixed storage directory relative to the user's MATLAB preferences directory
storage_dir = fullfile(prefdir, 'temp_sync_data');
% Part A: Check and load from local storage if available-
if tempLocalStorage == 1
local_filename = fullfile(storage_dir, sprintf('sync_data_run_%s.mat', num2str(dataTable.run_id)));
if exist(local_filename, 'file')
% Load from local storage and return
try
load(local_filename, 'Bits', 'Symbols', 'Scpe_cell');
found_sync = 1;
return
catch
delete(local_filename);
end
end
end
% If not locally saved, load from storage
if ~found_sync
% Load transmitted bits
Bits = load(fullfile([options.storage_path, char(dataTable.tx_bits_path)]));
Bits = Bits.Bits;
% Map bits to symbols
M = double(dataTable.pam_level);
fsym = dataTable.symbolrate;
Symbols_mapped = PAMmapper(M,0).map(Bits);
Symbols_mapped.fs = fsym;
% Load original symbols
Symbols = load(fullfile([options.storage_path, char(dataTable.tx_symbols_path)]));
Symbols = Symbols.Symbols;
found_sync = 0;
Scpe_cell = {};
% Try to load pre-synchronized data
try
Scpe_load = load(fullfile([options.storage_path, char(dataTable.rx_sync_path)]));
Scpe_cell = Scpe_load.S;
[~,~,~,found_sync] = Scpe_cell{2}.tsynch("reference", Symbols, "fs_ref", fsym, "debug_plots", 1);
catch
% Continue to next method if this fails
end
end
% If not found, try with raw data
if ~found_sync
try
Scpe_sig_raw = load([options.storage_path, char(dataTable.rx_raw_path(1))]);
Scpe_sig_raw = Scpe_sig_raw.Scpe_sig_raw;
Scpe_sig_resampled = Scpe_sig_raw.resample("fs_in", Scpe_sig_raw.fs, "fs_out", 2*fsym);
[~, Scpe_cell, ~, found_sync] = Scpe_sig_resampled.tsynch("reference", Symbols, "fs_ref", fsym, "debug_plots", 1);
catch
% Continue to next method if this fails
end
end
% Last attempt with mapped symbols
if ~found_sync && exist('Scpe_sig_raw', 'var')
if length(Symbols_mapped.signal) ~= sum(Symbols_mapped.signal == Symbols.signal)
[~, Scpe_cell, ~, found_sync] = Scpe_sig_raw.tsynch("reference", Symbols_mapped, "fs_ref", fsym, "debug_plots", 0);
end
end
% Part B: Save to local storage if data was loaded and synced
if tempLocalStorage == 1 && found_sync
% Create directory if it doesn't exist
if ~exist(storage_dir, 'dir')
mkdir(storage_dir);
end
% local_filename = fullfile(storage_dir, sprintf('sync_data_run_%s.mat', num2str(dataTable.run_id)));
% List existing files and remove oldest if more than N
max_local_files = 10; % Store up to N files
files = dir(fullfile(storage_dir, 'sync_data_run_*.mat'));
if length(files) >= max_local_files
% Sort by date
[~, idx] = sort([files.datenum]);
% Delete oldest file
delete(fullfile(storage_dir, files(idx(1)).name));
end
% Save current data
save(local_filename, 'Bits', 'Symbols', 'Scpe_cell');
end
% Limit number of occurrences
if found_sync
record_realizations = min(options.max_occurences, length(Scpe_cell));
Scpe_cell = Scpe_cell(1:record_realizations);
else
warning('Could not synchronize the received signal with the stored symbols!');
end
end

97
Functions/Job_Processing/loadDspInputFromFilePaths.m Normal file
View File

@@ -0,0 +1,97 @@
function dspInput = loadDspInputFromFilePaths(run_id, options)
%LOADDSPINPUTFROMFILEPATHS Load explicit signal files and prepare DSP input.
arguments
run_id
options struct
end
Tx_bits = load(textScalar(options.load_file_path.tx_bits_path));
Symbols = load(textScalar(options.load_file_path.tx_symbols_path));
Scpe_sig_raw = load(textScalar(options.load_file_path.rx_raw_path));
Tx_bits = Tx_bits.Bits;
Symbols = Symbols.Symbols;
Scpe_sig_raw = Scpe_sig_raw.Scpe_sig_raw;
fsym = Symbols.fs;
M = Symbols.logbook.ModifierCopy{1}.M;
duob_mode = Symbols.logbook.ModifierCopy{1}.duobinary_mode;
Scpe_sig_resampled = Scpe_sig_raw.resample("fs_in", Scpe_sig_raw.fs, "fs_out", 2*fsym);
[~, Scpe_cell, ~, found_sync] = Scpe_sig_resampled.tsynch("reference", Symbols, ...
"fs_ref", fsym, ...
"debug_plots", 1);
if found_sync
Scpe_cell = selectSyncedOccurrences(Scpe_cell, options);
else
warning('Could not synchronize the received signal with the stored symbols!');
end
dspInput = struct();
dspInput.run_id = run_id;
dspInput.dataTable = table();
dspInput.Tx_bits = Tx_bits;
dspInput.Symbols = Symbols;
dspInput.Scpe_cell = Scpe_cell;
dspInput.found_sync = found_sync;
dspInput.fsym = fsym;
dspInput.M = M;
dspInput.duob_mode = duob_mode;
end
function value = textScalar(value)
value = firstValue(value);
value = char(string(value));
end
function value = firstValue(value)
if iscell(value)
value = value{1};
elseif ~ischar(value) && ~isscalar(value)
value = value(1);
end
end
function Scpe_cell = selectSyncedOccurrences(Scpe_cell, options)
available_occurences = length(Scpe_cell);
start_occurence = floor(getOption(options, 'start_occurence', 1));
max_occurences = floor(getOption(options, 'max_occurences', available_occurences));
if available_occurences < 1
warning('loadDspInputFromFilePaths:NoSyncedOccurrences', ...
'Synchronization reported success, but no synced occurrences are available.');
return
end
if start_occurence < 1
error('loadDspInputFromFilePaths:InvalidStartOccurrence', ...
'start_occurence must be >= 1.');
end
if max_occurences < 1
Scpe_cell = Scpe_cell(1:0);
return
end
if start_occurence > available_occurences
warning('loadDspInputFromFilePaths:StartOccurrenceTooHigh', ...
['Requested start_occurence %d, but only %d synced occurrences are available. ', ...
'Processing the last occurrence only.'], ...
start_occurence, available_occurences);
Scpe_cell = Scpe_cell(available_occurences);
return
end
stop_occurence = min(available_occurences, start_occurence + max_occurences - 1);
Scpe_cell = Scpe_cell(start_occurence:stop_occurence);
end
function value = getOption(options, name, defaultValue)
if isfield(options, name)
value = options.(name);
else
value = defaultValue;
end
end

40
Functions/Job_Processing/loadDspInputFromRunId.m Normal file
View File

@@ -0,0 +1,40 @@
function dspInput = loadDspInputFromRunId(run_id, database, options)
%LOADDSPINPUTFROMRUNID Query run metadata and prepare canonical DSP input.
arguments
run_id
database
options struct
end
dataTable = queryRunid(run_id, database);
% Load signal files referenced by the run metadata, verify/synchronize the
% received signal, optionally cache the sync result, and cap occurrences.
[Tx_bits, Symbols, Scpe_cell, found_sync] = loadAndSyncRunSignals(dataTable, options);
dspInput = struct();
dspInput.run_id = run_id;
dspInput.dataTable = dataTable;
dspInput.Tx_bits = Tx_bits;
dspInput.Symbols = Symbols;
dspInput.Scpe_cell = Scpe_cell;
dspInput.found_sync = found_sync;
dspInput.fsym = dataTable.symbolrate;
dspInput.M = double(dataTable.pam_level);
dspInput.duob_mode = parseDbMode(dataTable.db_mode);
end
function mode = parseDbMode(rawMode)
if isnumeric(rawMode)
mode = db_mode(rawMode);
return
end
if iscell(rawMode)
rawMode = rawMode{1};
end
mode = db_mode(strrep(string(rawMode), '"', ''));
end

36
Functions/Job_Processing/preprocessRunId.m
View File

@@ -1,36 +0,0 @@
function dspInput = preprocessRunId(run_id, database, options)
%PREPROCESSRUNID Load one run_id and prepare the canonical DSP input.
arguments
run_id
database
options struct
end
dataTable = queryRunid(run_id, database);
[Tx_bits, Symbols, Scpe_cell, found_sync] = loadAndSyncSignalDataFromDb(dataTable, options);
dspInput = struct();
dspInput.run_id = run_id;
dspInput.dataTable = dataTable;
dspInput.Tx_bits = Tx_bits;
dspInput.Symbols = Symbols;
dspInput.Scpe_cell = Scpe_cell;
dspInput.found_sync = found_sync;
dspInput.fsym = dataTable.symbolrate;
dspInput.M = double(dataTable.pam_level);
dspInput.duob_mode = parseDbMode(dataTable.db_mode);
end
function mode = parseDbMode(rawMode)
if isnumeric(rawMode)
mode = db_mode(rawMode);
return
end
if iscell(rawMode)
rawMode = rawMode{1};
end
mode = db_mode(strrep(string(rawMode), '"', ''));
end

80
Functions/Job_Processing/runBatch.m
View File

@@ -1,6 +1,6 @@
function results = runBatch(workerFcn, jobs, options)
function batchResults = runBatch(workerFcn, jobs, options)
%RUNBATCH Execute a list of jobs in serial or parallel.
% results = runBatch(workerFcn, jobs) executes each jobs(i).args cell
% batchResults = runBatch(workerFcn, jobs) executes each jobs(i).args cell
% with workerFcn and returns a cell array aligned with the input jobs.
%
% Supported job fields:
@@ -24,7 +24,7 @@ function results = runBatch(workerFcn, jobs, options)
mode = normalizeProcessingMode(options.mode);
jobs = normalizeJobs(jobs);
nJobs = numel(jobs);
results = cell(1, nJobs);
batchResults = cell(1, nJobs);
h = [];
if nJobs == 0
@@ -33,7 +33,7 @@ function results = runBatch(workerFcn, jobs, options)
if options.waitbar
h = waitbar(0, char(options.waitbarMessage));
cleanupWaitbar = onCleanup(@() closeWaitbar(h)); %#ok<NASGU>
cleanupWaitbar = onCleanup(@() closeWaitbar(h));
end
switch mode
@@ -41,36 +41,36 @@ function results = runBatch(workerFcn, jobs, options)
pool = setupParallelPool(options.numWorkers, options.idleTimeout, options.cancelExistingQueue);
futures = parallel.FevalFuture.empty(nJobs, 0);
for idx = 1:nJobs
fprintf('[%s] Submitted.\n', jobs(idx).label);
futures(idx) = parfeval(pool, workerFcn, 1, jobs(idx).args{:});
for jobIdx = 1:nJobs
fprintf('[%s] Submitted.\n', jobs(jobIdx).label);
futures(jobIdx) = parfeval(pool, workerFcn, 1, jobs(jobIdx).args{:});
end
consumedIdx = false(nJobs, 1);
for completedCount = 1:nJobs
try
[idx, value] = fetchNext(futures);
consumedIdx(idx) = true;
results{idx} = value;
fprintf('[%s] Completed (%d/%d).\n', jobs(idx).label, completedCount, nJobs);
[jobIdx, jobResult] = fetchNext(futures);
consumedIdx(jobIdx) = true;
batchResults{jobIdx} = jobResult;
fprintf('[%s] Completed (%d/%d).\n', jobs(jobIdx).label, completedCount, nJobs);
if ~isempty(options.resultHandler)
options.resultHandler(value, jobs(idx), idx);
options.resultHandler(jobResult, jobs(jobIdx), jobIdx);
end
catch fetchErr
idxErr = findErroredFuture(futures, consumedIdx);
if isempty(idxErr)
errorJobIdx = findErroredFuture(futures, consumedIdx);
if isempty(errorJobIdx)
rethrow(fetchErr);
end
consumedIdx(idxErr) = true;
errInfo = extractFutureError(futures(idxErr));
results{idxErr} = errInfo;
consumedIdx(errorJobIdx) = true;
errInfo = extractFutureError(futures(errorJobIdx));
batchResults{errorJobIdx} = errInfo;
if ~isempty(options.errorHandler)
options.errorHandler(errInfo, jobs(idxErr), idxErr);
options.errorHandler(errInfo, jobs(errorJobIdx), errorJobIdx);
else
defaultErrorHandler(errInfo, jobs(idxErr), idxErr);
defaultErrorHandler(errInfo, jobs(errorJobIdx), errorJobIdx);
end
end
@@ -78,27 +78,27 @@ function results = runBatch(workerFcn, jobs, options)
end
case processingMode.serial
for idx = 1:nJobs
for jobIdx = 1:nJobs
try
fprintf('[%s] Running.\n', jobs(idx).label);
value = feval(workerFcn, jobs(idx).args{:});
results{idx} = value;
fprintf('[%s] Completed (%d/%d).\n', jobs(idx).label, idx, nJobs);
fprintf('[%s] Running.\n', jobs(jobIdx).label);
jobResult = workerFcn(jobs(jobIdx).args{:});
batchResults{jobIdx} = jobResult;
fprintf('[%s] Completed (%d/%d).\n', jobs(jobIdx).label, jobIdx, nJobs);
if ~isempty(options.resultHandler)
options.resultHandler(value, jobs(idx), idx);
options.resultHandler(jobResult, jobs(jobIdx), jobIdx);
end
catch ME
results{idx} = ME;
batchResults{jobIdx} = ME;
if ~isempty(options.errorHandler)
options.errorHandler(ME, jobs(idx), idx);
options.errorHandler(ME, jobs(jobIdx), jobIdx);
else
defaultErrorHandler(ME, jobs(idx), idx);
defaultErrorHandler(ME, jobs(jobIdx), jobIdx);
end
end
updateWaitbar(options.waitbar, h, idx, nJobs);
updateWaitbar(options.waitbar, h, jobIdx, nJobs);
end
otherwise
@@ -128,23 +128,23 @@ function jobs = normalizeJobs(jobs)
return
end
for idx = 1:numel(jobs)
if ~isfield(jobs, 'args') || isempty(jobs(idx).args)
jobs(idx).args = {};
for jobIdx = 1:numel(jobs)
if ~isfield(jobs, 'args') || isempty(jobs(jobIdx).args)
jobs(jobIdx).args = {};
end
if ~iscell(jobs(idx).args)
error('runBatch:InvalidArgs', 'jobs(%d).args must be a cell array.', idx);
if ~iscell(jobs(jobIdx).args)
error('runBatch:InvalidArgs', 'jobs(%d).args must be a cell array.', jobIdx);
end
if ~isfield(jobs, 'label') || isempty(jobs(idx).label)
jobs(idx).label = sprintf('Job %d', idx);
if ~isfield(jobs, 'label') || isempty(jobs(jobIdx).label)
jobs(jobIdx).label = sprintf('Job %d', jobIdx);
else
jobs(idx).label = string(jobs(idx).label);
jobs(jobIdx).label = string(jobs(jobIdx).label);
end
if ~isfield(jobs, 'meta')
jobs(idx).meta = struct();
jobs(jobIdx).meta = struct();
end
end
end
@@ -189,9 +189,9 @@ function pool = setupParallelPool(numWorkers, idleTimeout, cancelExistingQueue)
end
end
function idxErr = findErroredFuture(futures, consumedIdx)
function errorJobIdx = findErroredFuture(futures, consumedIdx)
readMask = arrayfun(@(future) future.Read, futures).';
idxErr = find(readMask & ~consumedIdx, 1);
errorJobIdx = find(readMask & ~consumedIdx, 1);
end
function errInfo = extractFutureError(future)

120
Functions/Job_Processing/submitJobs.m
View File

@@ -9,9 +9,11 @@ function [results, wh] = submitJobs(run_ids, dsp_options, submit_mode, submit_op
% submit_options : struct with fields
% .waitbar (logical)
% .wh (DataStorage object)
% .storePackages (string array, optional package whitelist)
%
% results : cell(nJobsPerRunId, nRunIds)
% wh : updated DataStorage
% wh : updated DataStorage. For multiple run_ids, run_id is added as
% the first storage axis.
arguments
run_ids int32 = 0
@@ -19,15 +21,19 @@ arguments
submit_mode processingMode = processingMode.serial
submit_options.waitbar (1,1) logical = true
submit_options.wh = DataStorage(struct())
submit_options.storePackages string = string.empty()
end
% Normalize
run_ids = run_ids(:)';
nRunIds = numel(run_ids);
nJobsPerRunId = submit_options.wh.getLastLinIndice();
sweepWh = submit_options.wh;
validateNoRunIdSweepParameter(sweepWh);
nJobsPerRunId = sweepWh.getLastLinIndice();
totalJobs = nRunIds * nJobsPerRunId;
wh = submit_options.wh;
wh = buildStorageWarehouse(sweepWh, run_ids);
results = cell(nJobsPerRunId, nRunIds);
jobs = repmat(struct('args', {{}}, 'label', "", 'meta', struct()), 1, totalJobs);
@@ -35,14 +41,16 @@ jobCounter = 0;
for r = 1:nRunIds
for k = 1:nJobsPerRunId
jobCounter = jobCounter + 1;
optionalVars = buildOptionalVars(k, submit_options.wh);
userParameters = buildUserParameters(k, sweepWh);
jobOptions = dsp_options;
jobOptions.parameters = optionalVars;
jobOptions.userParameters = userParameters;
jobs(jobCounter).args = [{run_ids(r)}, structToNameValue(jobOptions)];
jobs(jobCounter).label = sprintf('RunID %d, Job %d', run_ids(r), k);
jobs(jobCounter).meta.runIndex = r;
jobs(jobCounter).meta.jobIndex = k;
jobs(jobCounter).meta.sweepJobIndex = k;
jobs(jobCounter).meta.storageJobIndex = buildStorageJobIndex(run_ids(r), userParameters, wh);
jobs(jobCounter).meta.run_id = run_ids(r);
end
end
@@ -74,16 +82,69 @@ end
fprintf('Full report:\n%s\n', getReport(ME,'extended'));
end
function optionalVars = buildOptionalVars(jobIndex, wh)
optionalVars = struct();
function userParameters = buildUserParameters(jobIndex, wh)
userParameters = struct();
if ~isempty(wh.getDimension())
[vals, names] = wh.getPhysIndicesByLinIndex(jobIndex);
for pi = 1:numel(names)
optionalVars.(names{pi}) = vals{pi};
userParameters.(names{pi}) = vals{pi};
end
end
end
function validateNoRunIdSweepParameter(wh)
if isfield(wh.inputParams, "run_id")
error("submitJobs:RunIdParameterConflict", ...
"Do not define userParameters.run_id. submitJobs manages run_id as a storage axis.");
end
end
function storageWh = buildStorageWarehouse(sweepWh, runIds)
if isscalar(runIds)
storageWh = sweepWh;
return
end
storageParameters = struct();
storageParameters.run_id = runIds;
sweepParameterNames = fieldnames(sweepWh.inputParams);
for parameterIdx = 1:numel(sweepParameterNames)
parameterName = sweepParameterNames{parameterIdx};
storageParameters.(parameterName) = sweepWh.inputParams.(parameterName);
end
storageWh = DataStorage(storageParameters);
end
function storageJobIndex = buildStorageJobIndex(runId, userParameters, wh)
storageParameterNames = wh.fn;
if isempty(storageParameterNames)
storageJobIndex = 1;
return
end
storageSubscripts = cell(1, numel(storageParameterNames));
for parameterIdx = 1:numel(storageParameterNames)
parameterName = char(storageParameterNames(parameterIdx));
if strcmp(parameterName, "run_id")
parameterValue = runId;
else
parameterValue = userParameters.(parameterName);
end
storageSubscripts{parameterIdx} = wh.getIndexByPhys(parameterName, parameterValue);
end
if isscalar(storageSubscripts)
storageJobIndex = storageSubscripts{1};
else
storageJobIndex = sub2ind(wh.getStorageSize(), storageSubscripts{:});
end
end
function nameValue = structToNameValue(options)
names = fieldnames(options);
nameValue = cell(1, 2*numel(names));
@@ -95,14 +156,41 @@ end
end
function storeResult(val, job, ~)
if ~isempty(wh)
jobIndex = job.meta.jobIndex;
wh.addValueToStorageByLinIdx(val.ffe_package, 'ffe_package', jobIndex);
wh.addValueToStorageByLinIdx(val.mlse_package, 'mlse_package', jobIndex);
wh.addValueToStorageByLinIdx(val.vnle_package, 'vnle_package', jobIndex);
wh.addValueToStorageByLinIdx(val.dbtgt_package,'dbtgt_package',jobIndex);
wh.addValueToStorageByLinIdx(val.dbenc_package,'dbenc_package',jobIndex);
wh.addValueToStorageByLinIdx(val.mlmlse_package,'mlmlse_package',jobIndex);
if isempty(wh) || ~isstruct(val)
return
end
storageJobIndex = job.meta.storageJobIndex;
resultFields = fieldnames(val);
for resultIdx = 1:numel(resultFields)
storageName = resultFields{resultIdx};
if ~shouldStore(storageName)
continue
end
if isempty(val.(storageName))
continue
end
ensureStorage(storageName);
wh.addValueToStorageByLinIdx(val.(storageName), storageName, storageJobIndex);
end
end
function tf = shouldStore(storageName)
if isempty(submit_options.storePackages)
tf = true;
return
end
tf = any(string(storageName) == submit_options.storePackages);
end
function ensureStorage(storageName)
if ~isfield(wh.sto, storageName)
wh.addStorage(storageName);
end
end