halfway merged and pulled?!
This commit is contained in:
Silas Labor Zizou
114
Functions/Job_Processing/loadAndSyncSignalDataFromDb.m
Normal file
114
Functions/Job_Processing/loadAndSyncSignalDataFromDb.m
Normal file
@@ -0,0 +1,114 @@
|
||||
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
|
||||
32
Functions/Job_Processing/preprocessSignal.m
Normal file
32
Functions/Job_Processing/preprocessSignal.m
Normal file
@@ -0,0 +1,32 @@
|
||||
function Scpe_sig = preprocessSignal(Scpe_sig, Symbols, fsym)
|
||||
% PREPROCESSSIGNAL Performs standard preprocessing on a signal
|
||||
%
|
||||
% Inputs:
|
||||
% Scpe_sig - Input signal
|
||||
% Symbols - Reference symbols for synchronization
|
||||
% fsym - Symbol frequency
|
||||
%
|
||||
% Outputs:
|
||||
% Scpe_sig - Preprocessed signal
|
||||
|
||||
% Resample to 2x symbol rate
|
||||
Scpe_sig = Scpe_sig.resample("fs_out", 2*fsym);
|
||||
|
||||
% Synchronize with reference
|
||||
[Scpe_sig, ~] = Scpe_sig.tsynch("reference", Symbols, "fs_ref", fsym, "debug_plots", 0);
|
||||
|
||||
% Apply Gaussian filter
|
||||
if 1
|
||||
Scpe_sig = Filter('filtdegree', 8, "f_cutoff", Symbols.fs.*0.52, ...
|
||||
"fs", Scpe_sig.fs, "filterType", filtertypes.gaussian, ...
|
||||
"active", true).process(Scpe_sig);
|
||||
else
|
||||
Scpe_sig = Filter('filtdegree', 4, "f_cutoff", Symbols.fs.*0.6, ...
|
||||
"fs", Scpe_sig.fs, "filterType", filtertypes.gaussian, ...
|
||||
"active", true).process(Scpe_sig);
|
||||
end
|
||||
|
||||
%Remove DC offset
|
||||
Scpe_sig = Scpe_sig - mean(Scpe_sig.signal);
|
||||
|
||||
end
|
||||
47
Functions/Job_Processing/printResults.m
Normal file
47
Functions/Job_Processing/printResults.m
Normal file
@@ -0,0 +1,47 @@
|
||||
function printResults(run_id, M, Symbols, vnle_pf_package, dbtgt_package, occ)
|
||||
% PRINTRESULTS Prints formatted results from equalization
|
||||
%
|
||||
% Inputs:
|
||||
% run_id - Run identifier
|
||||
% M - PAM level
|
||||
% Symbols - Symbol data with fs property
|
||||
% vnle_pf_package - VNLE+PF results package
|
||||
% dbtgt_package - DB target results package (optional)
|
||||
% occ - Occurrence index
|
||||
|
||||
% Print header
|
||||
fprintf("==== EQUALIZATION RUN-ID %d | PAM-%d | %.2f GBd ====\n\n", run_id, M, Symbols.fs.*1e-9);
|
||||
|
||||
% VNLE Results
|
||||
if ~isempty(vnle_pf_package)
|
||||
vnle_result = vnle_pf_package{occ}.resultsVNLE;
|
||||
mlse_result = vnle_pf_package{occ}.resultsMLSE;
|
||||
|
||||
fprintf(">> VNLE Results:\n");
|
||||
fprintf(" BER: %.2e\n", vnle_result.BER);
|
||||
fprintf(" BER (pre-code): %.2e\n", vnle_result.BER_precoded);
|
||||
fprintf(" SNR: %.2f dB\n", vnle_result.SNR);
|
||||
fprintf(" GMI: %.4f\n", vnle_result.GMI);
|
||||
fprintf(" Linerate: %.2f Gbps\n", Symbols.fs .* floor(log2(M)*10)/10 .*1e-9);
|
||||
fprintf(" AIR: %.2f Gbps\n", vnle_result.AIR.*1e-9);
|
||||
fprintf("\n");
|
||||
|
||||
% MLSE Results
|
||||
fprintf(">> MLSE Results:\n");
|
||||
fprintf(" BER: %.2e\n", mlse_result.BER);
|
||||
fprintf(" BER (pre-code): %.2e\n", mlse_result.BER_precoded);
|
||||
fprintf(" Channel Alpha: %.2f\n", mlse_result.Alpha);
|
||||
fprintf("\n");
|
||||
end
|
||||
|
||||
% DB Target Results
|
||||
if ~isempty(dbtgt_package)
|
||||
dbtgt = dbtgt_package{occ}.resultsDBtgt;
|
||||
fprintf(">> DB Target Results:\n");
|
||||
fprintf(" BER: %.2e\n", dbtgt.BER);
|
||||
fprintf(" BER (pre-code): %.2e\n", dbtgt.BER_precoded);
|
||||
fprintf("\n");
|
||||
end
|
||||
|
||||
fprintf("- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - \n\n");
|
||||
end
|
||||
11
Functions/Job_Processing/queryRunid.m
Normal file
11
Functions/Job_Processing/queryRunid.m
Normal file
@@ -0,0 +1,11 @@
|
||||
function dataTable = queryRunid(run_id, database)
|
||||
|
||||
% Query database for run configuration
|
||||
filterParams = database.tables;
|
||||
filterParams.Runs = struct('run_id', run_id);
|
||||
|
||||
[dataTable, ~] = database.queryDB(filterParams, database.getTableFieldNames('Runs'));
|
||||
[~, uniqueIdx] = unique(dataTable.run_id); % Get unique run_id indices
|
||||
dataTable = dataTable(uniqueIdx, :); % Extract unique configurations for each run_id
|
||||
|
||||
end
|
||||
235
Functions/Job_Processing/submitJobs.m
Normal file
235
Functions/Job_Processing/submitJobs.m
Normal file
@@ -0,0 +1,235 @@
|
||||
function [results, wh] = submitJobs(run_ids, dsp_options, submit_mode, submit_options)
|
||||
% SUBMITJOBS Submits dsp_runid jobs for processing (parallel or serial)
|
||||
%
|
||||
% [results, wh] = submitJobs(run_ids, dsp_options, submit_mode, submit_options)
|
||||
%
|
||||
% run_ids : scalar or vector of run IDs
|
||||
% dsp_options : struct of dsp_runid name/value options
|
||||
% submit_mode : processingMode.parallel or .serial
|
||||
% submit_options : struct with fields
|
||||
% .waitbar (logical)
|
||||
% .wh (DataStorage object)
|
||||
%
|
||||
% results : cell(nJobsPerRunId, nRunIds)
|
||||
% wh : updated DataStorage
|
||||
|
||||
arguments
|
||||
run_ids int32 = 0
|
||||
dsp_options struct = struct()
|
||||
submit_mode processingMode = processingMode.serial
|
||||
submit_options.waitbar (1,1) logical = true
|
||||
submit_options.wh = DataStorage(struct())
|
||||
end
|
||||
|
||||
% Normalize
|
||||
run_ids = run_ids(:)';
|
||||
nRunIds = numel(run_ids);
|
||||
nJobsPerRunId = submit_options.wh.getLastLinIndice();
|
||||
totalJobs = nRunIds * nJobsPerRunId;
|
||||
|
||||
% Preallocate
|
||||
results = cell(nJobsPerRunId, nRunIds);
|
||||
futures = parallel.FevalFuture.empty(totalJobs,0);
|
||||
jobIndices = zeros(totalJobs,2);
|
||||
|
||||
% Optional waitbar
|
||||
if submit_options.waitbar
|
||||
h = waitbar(0, 'Processing Jobs...');
|
||||
cleanupObj = onCleanup(@() delete(h));
|
||||
end
|
||||
|
||||
switch submit_mode
|
||||
case processingMode.parallel
|
||||
%—– SET UP POOL & QUEUE —–
|
||||
p = setupParallelPool(11, 300); % 10 workers, 300s idle timeout
|
||||
|
||||
% === submit all futures ===
|
||||
jobCounter = 0;
|
||||
for r = 1:nRunIds
|
||||
for k = 1:nJobsPerRunId
|
||||
jobCounter = jobCounter + 1;
|
||||
jobIndices(jobCounter,:) = [r,k];
|
||||
|
||||
opt = buildOptionalVars(k, submit_options.wh);
|
||||
futures(jobCounter) = parfeval( ...
|
||||
p, @dsp_runid, 1, ...
|
||||
run_ids(r), ...
|
||||
"database_type", dsp_options.database_type, ...
|
||||
"dataBase", dsp_options.dataBase, ...
|
||||
"append_to_db", dsp_options.append_to_db, ...
|
||||
"load_file_path", dsp_options.load_file_path, ...
|
||||
"max_occurences", dsp_options.max_occurences, ...
|
||||
"storage_path", dsp_options.storage_path, ...
|
||||
"mode", dsp_options.mode, ...
|
||||
"parameters", opt ...
|
||||
);
|
||||
|
||||
fprintf('[RunID %d, Job %d] Submitted to pool.\n', run_ids(r), k);
|
||||
end
|
||||
end
|
||||
|
||||
%—– W A I T B A R U P D A T E —–
|
||||
if submit_options.waitbar
|
||||
futureArray = futures;
|
||||
updateWB = @() waitbar( ...
|
||||
sum(arrayfun(@(f) strcmp(f.State,'finished'), futureArray))/totalJobs, ...
|
||||
h, sprintf('Completed %d/%d jobs', ...
|
||||
sum(arrayfun(@(f) strcmp(f.State,'finished'), futureArray)), totalJobs) ...
|
||||
);
|
||||
afterEach(futureArray, updateWB, 0);
|
||||
end
|
||||
|
||||
% —– before the loop —–
|
||||
% Keep track of which futures we've already handled:
|
||||
consumedIdx = false(totalJobs,1);
|
||||
|
||||
% —– fetch in completion order, handling successes and errors —–
|
||||
for n = 1:totalJobs
|
||||
try
|
||||
% This returns the value AND the linear index in 'futures'
|
||||
[idx, val] = fetchNext(futures);
|
||||
|
||||
duration = futures(idx).RunningDuration;
|
||||
startDT = futures(idx).StartDateTime;
|
||||
finishDT = futures(idx).FinishDateTime;
|
||||
duration = finishDT - startDT;
|
||||
|
||||
% Mark it consumed
|
||||
consumedIdx(idx) = true;
|
||||
|
||||
% Map back to (r,k) and store
|
||||
r = jobIndices(idx,1);
|
||||
k = jobIndices(idx,2);
|
||||
|
||||
fprintf('[%s] JobID %d/%d (%.1f%%) — %s — RunID %d — Subjob %d — fetched.\n', ...
|
||||
datestr(now,'yyyy-mm-dd HH:MM:SS'), ...
|
||||
r, totalJobs, 100*n/totalJobs,char(duration), ...
|
||||
run_ids(r), k);
|
||||
|
||||
% Update waitbar
|
||||
if submit_options.waitbar
|
||||
waitbar(n/totalJobs, h, ...
|
||||
sprintf('Fetched %d/%d (%.1f% Percent)', n, totalJobs, 100*n/totalJobs));
|
||||
drawnow; % force the GUI to refresh
|
||||
end
|
||||
|
||||
storeResult(val, k, submit_options.wh);
|
||||
results{k,r} = val;
|
||||
|
||||
catch fetchErr
|
||||
% fetchNext has already set Read=true on the errored future.
|
||||
% Find the one Read==true that we have _not_ yet consumed.
|
||||
readMask = arrayfun(@(f) f.Read, futures);
|
||||
idxErr = find(readMask & ~consumedIdx', 1);
|
||||
consumedIdx(idxErr) = true;
|
||||
|
||||
% Pull the _real_ exception out of the future object
|
||||
errInfo = futures(idxErr).Error;
|
||||
if iscell(errInfo)
|
||||
origME = errInfo{1};
|
||||
else
|
||||
origME = errInfo;
|
||||
end
|
||||
|
||||
% Map back to (r,k) and log
|
||||
r = jobIndices(idxErr,1);
|
||||
k = jobIndices(idxErr,2);
|
||||
handleError(origME, k, run_ids(r));
|
||||
results{k,r} = origME;
|
||||
end
|
||||
end
|
||||
|
||||
|
||||
case processingMode.serial
|
||||
%—– SERIAL EXECUTION —–
|
||||
jobCounter = 0;
|
||||
for r = 1:nRunIds
|
||||
for k = 1:nJobsPerRunId
|
||||
jobCounter = jobCounter + 1;
|
||||
optionalVars = buildOptionalVars(k, submit_options.wh);
|
||||
try
|
||||
fprintf('[RunID %d, Job %d] Running in linear mode...\n', run_ids(r), k);
|
||||
val = dsp_runid( run_ids(r), ...
|
||||
"database_type", dsp_options.database_type, ...
|
||||
"dataBase", dsp_options.dataBase, ...
|
||||
"append_to_db", dsp_options.append_to_db, ...
|
||||
"load_file_path", dsp_options.load_file_path, ...
|
||||
"max_occurences", dsp_options.max_occurences, ...
|
||||
"storage_path", dsp_options.storage_path, ...
|
||||
"mode", dsp_options.mode, ...
|
||||
"parameters", optionalVars );
|
||||
|
||||
fprintf('[RunID %d, Job %d] Completed successfully.\n', run_ids(r), k);
|
||||
storeResult(val, k, submit_options.wh);
|
||||
results{k,r} = val;
|
||||
|
||||
catch ME
|
||||
handleError(ME, k, run_ids(r));
|
||||
results{k,r} = ME;
|
||||
end
|
||||
|
||||
if submit_options.waitbar
|
||||
waitbar(jobCounter/totalJobs, h, ...
|
||||
sprintf('Completed %d/%d jobs', jobCounter, totalJobs));
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
otherwise
|
||||
error('Unknown submit_mode "%s".', string(submit_mode))
|
||||
end
|
||||
|
||||
wh = submit_options.wh;
|
||||
|
||||
%% Local helpers %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
|
||||
|
||||
function handleError(ME, jobIndex, run_id)
|
||||
fprintf('[RunID %d, Job %d] ERROR [%s]: %s\n', run_id, jobIndex, ...
|
||||
ME.identifier, ME.message);
|
||||
for st = ME.stack'
|
||||
fprintf(' %s:%d (%s)\n', st.file, st.line, st.name);
|
||||
end
|
||||
fprintf('Full report:\n%s\n', getReport(ME,'extended'));
|
||||
end
|
||||
|
||||
function optionalVars = buildOptionalVars(jobIndex, wh)
|
||||
optionalVars = struct();
|
||||
if ~isempty(wh.getDimension())
|
||||
[vals, names] = wh.getPhysIndicesByLinIndex(jobIndex);
|
||||
for pi = 1:numel(names)
|
||||
optionalVars.(names{pi}) = vals{pi};
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
function storeResult(val, jobIndex, wh)
|
||||
if ~isempty(wh)
|
||||
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);
|
||||
end
|
||||
end
|
||||
function p = setupParallelPool(numWorkers, idleTimeout)
|
||||
% Ensure a pool exists at the right size & timeout
|
||||
p = gcp('nocreate');
|
||||
if isempty(p) || p.NumWorkers~=numWorkers
|
||||
if ~isempty(p)
|
||||
delete(p);
|
||||
end
|
||||
p = parpool('local', numWorkers, 'IdleTimeout', idleTimeout);
|
||||
end
|
||||
|
||||
% Cancel anything left in the pool's default queue
|
||||
q = p.FevalQueue;
|
||||
if ~isempty(q.QueuedFutures) || ~isempty(q.RunningFutures)
|
||||
cancelAll(q);
|
||||
fprintf('Canceled %d unfetched jobs from old queue.\n', ...
|
||||
numel(q.QueuedFutures)+numel(q.RunningFutures));
|
||||
end
|
||||
end
|
||||
|
||||
|
||||
end
|
||||
Reference in New Issue
Block a user