Updated existing online writer plugins to support prediction relative to markers (PredictAt option), improved fault tolerance in the online plugin system.

Author: chkothe

RELEASE NOTES.TXT

@@ -105,7 +105,6 @@ Major New Features
 - set_joinepos: new function to join epoched signals across epochs (orders of magnitude faster than set_merge, and handles corner cases)
 - set_makepos: can now extract epochs relative to markers online/incrementally (complete rewrite)
 - onl_***: generally markers can now be streamed through the online system and are visible to filters and the feature extraction
-- 
 
 Minor New Features
 - combine_structs/flatten_structs: new functions to handle structs in name-value pair lists
@@ -118,13 +117,13 @@ Minor New Features
 - onl_newpredictor: new argument PredictAt: allows to define markers relative to which predictions should be generated 
   (and epochs should be extracted)
 - onl_peek: now supports markers
-- onl_read_background: now supports callback functions that also return markers
-- onl_write_background: new options PredictAt, Verbose, and EmptyResultValue
+- onl_read_background: now supports callback functions that also return markers, plus better error handling 
+- onl_write_background: new options PredictAt, Verbose, and EmptyResultValue, also better error handling
 - onl_simulate: aside from specifying the latency of onl_predict calls, now allows to use the pipeline with marker-locked prediction
   (via TargetMarkers argument which takes the same role as PredictAt in some other functions)
 - run_readlsl: now supports reading synchronized markers from marker streams
-- run_writelsl, run_writesnap: added PredictAt option
 - run_readdataset: now supports reading markers
+- run_write***: added PredictAt and Verbose options
 - utl_buffer: new utility function for buffering of subsequent signal structs
 - utl_fileinfo: now supports empty file paths
 - utl_find_filter: now supports returning the subscript path to the matching location in the data structure (for use with subsasgn)
@@ -152,7 +151,7 @@ Minor Fixes
 - utl_resolve_streams: fixed rare crash bug when multiple candidate streams needed to be disambiguated based on supported channels and types
 - vis_filtered: minor documentation fix
 
-
+
 === Changes since 1.1-beta ===
 
 Major New Features
@@ -173,7 +172,7 @@ Minor Changes
 Serious Fixes
 
 Minor Fixes
-- flt_ica, robust_sphere option: was calling an obsolete function
+- flt_ica, robust_sphere option: was calling an obsolete function
 
 === Changes since 1.01 ===
 
@@ -282,10 +281,10 @@ Minor New Features
 - flt_iir: now supports free-form designs via the Yule-Walker method
 - flt_pipeline: now more tolerant when trying to load erroneous plugin functions
 - flt_ica: now supports rica, sphering and robust sphering as methods
-- plugins are now also searched in the respective sub-directory “in_development”, and marked 
+- plugins are now also searched in the respective sub-directory “in_development�?, and marked 
   experimental by default
 - ParadigmFBCSP: now supports robust covariance estimation
-- ParadigmDataflowSimplified: now supports a “feature conditioning” stage between feature 
+- ParadigmDataflowSimplified: now supports a “feature conditioning�? stage between feature 
   extraction and machine learning
 - ParadigmDALERP: now supports application to other data fields than .data, and allows for 
   optional shrinkage-based covariance estimation
@@ -334,7 +333,7 @@ Major Changes
 - set_chanid is now case insensitive at channel matching (note that this should only break
   processing of pathological data sets which have multiple channels with the same name but 
   different case, and otherwise improve things)
-- some methods are now marked as “experimental” and not shown in the GUI by default (can be 
+- some methods are now marked as “experimental�? and not shown in the GUI by default (can be 
   toggled at startup or in the bcilab_config.m)
 
 Minor Changes

code/online_analysis/onl_newpredictor.m

@@ -80,8 +80,6 @@
 
 % optionally set up the filter graph to generate epochs relative to desired target markers
 if ~isempty(predict_at)
-    if ~iscellstr(predict_at)
-        error('PredictAt must be a cell array of strings.'); end
     % for each pipeline...
     for c=1:length(predictor.tracking.filter_graph)
         % substitute the node set_makepos('signal',x,...,'online_epoching',y,...)

code/online_analysis/onl_predict.m

@@ -9,19 +9,19 @@
 % In:
 %   Name : name of a predictor (under which is was previously created with onl_newpredictor)
 %
-%   Format : the desired form of the prediction (see also ult_formatprediction), can be:       
-%            * 'raw': the raw prediction, as defined by ml_predict (default)
-%            * 'expectation': the output is the expected value (i.e., posterior mean) of the
-%                             quantity to be predicted; can be multi-dimensional [1xD], but D
-%                             equals in most cases 1
-%            * 'distribution': the output is the probability distribution (discrete or
-%                              continuous) of the quantity to be predicted usually, this is a
-%                              discrete distribution - one probability value for every possible
-%                              target outcome [1xV] it can also be the parameters of a
-%                              parametric distribution (e.g., mean, variance) - yielding one
-%                              value for each parameter [DxP]
-%            * 'mode': the mode [1xD], or most likely output value (only supported for discrete
-%                      probability distributions)
+%   Format     : format of the output prediction (in the descriptions, N is the number of
+%                predictions), can be one of:
+%                 * 'expectation': the output is the expected value (i.e., posterior mean) of the
+%                                  quantity to be predicted; can be multi-dimensional [NxD]
+%                 * 'distribution': the output is the probability distribution (discrete or
+%                                   continuous) of the quantity to be predicted usually, this is a
+%                                   discrete distribution - one probability value for every possible
+%                                   target outcome [NxV] it can also be the parameters of a
+%                                   parametric distribution (e.g., mean, variance) - yielding one
+%                                   value for each parameter [NxP]
+%                 * 'mode': the mode [Nx1], or most likely output value (only supported for discrete
+%                           probability distributions)
+%                 * 'raw': the raw prediction, as defined by ml_predict
 %
 %   SuppressOutput : whether to suppress console output (default: true)
 %

code/online_analysis/onl_read_background.m

@@ -53,21 +53,26 @@ function append_data(stream_name,stream_id,timer_handle,read_block)
     x = evalin('base',stream_name);
     if x.streamid ~= stream_id
         error('Stream changed.'); end
-    % get a new block
-    if nargin(read_block) == 1
-        block = read_block(x);
-    else
-        block = read_block();
-    end
-    % append it to the stream
-    if iscell(block)        
-        onl_append(stream_name,block{:});
-    else
-        onl_append(stream_name,block);
+    try
+        % get a new block
+        if nargin(read_block) == 1
+            block = read_block(x);
+        else
+            block = read_block();
+        end
+        % append it to the stream
+        if iscell(block)        
+            onl_append(stream_name,block{:});
+        else
+            onl_append(stream_name,block);
+        end
+    catch e
+        disp('Error in block-reading function:');
+        hlp_handleerror(e);
     end
 catch e
     if ~strcmp(e.identifier,'MATLAB:UndefinedFunction')
-        env_handleerror(e); end
+        hlp_handleerror(e); end
     % stream has been changed (e.g., replaced/deleted) --> stop timer
     stop(timer_handle);
     delete(timer_handle);

code/online_analysis/onl_write_background.m

@@ -8,16 +8,16 @@ function onl_write_background(varargin)
 % computation of updated estimates, and their transfer to the data sink.
 %
 % In:
-%   ResultWriter : function that receives a BCI estimate and writes it to some external device.
+%   ResultWriter : Function that receives one or more BCI estimates and writes them to some external
+%                  device. The format is according to OutputFormat.
 %
-%   MatlabStream : real-time stream name to read from (in MATLAB workspace) (default: 'laststream')
+%   MatlabStream : Real-time stream name to read from (in MATLAB workspace) (default: 'laststream')
 %
-%   Model : predictive model to use, or variable name (in MATLAB workspace)  (default: 'lastmodel')
+%   Model : Predictive model to use, or variable name (in MATLAB workspace)  (default: 'lastmodel')
 %
-%   OutputFormat : output data format, see onl_predict 
-%                  (default: 'distribution')
+%   OutputFormat : Output data format, see onl_predict (default: 'distribution')
 %
-%   UpdateFrequency : frequency at which the device should be queried, in Hz (default: 25)
+%   UpdateFrequency : Frequency at which the device should be queried, in Hz (default: 25)
 %
 %   StartDelay : Delay before real-time processing begins; grace period until user resources are 
 %                created (default: 1)
@@ -86,8 +86,16 @@ function write_data(predictor,stream,fmt,result_writer,pred_id,stream_id,timer_h
     % make a prediction
     y = onl_predict(predictor,fmt,~verbose,empty_result_value);
     % and write it out
-    result_writer(y);
-catch %#ok<CTCH>
+    try
+        result_writer(y);
+    catch e
+        disp('Error in result-writing function:');
+        hlp_handleerror(e);
+    end
+catch e
+    if ~strcmp(e.identifier,'MATLAB:UndefinedFunction')
+        hlp_handleerror(e); end    
+    % stream or predictor have changed (e.g., replaced/deleted) --> stop timer
     stop(timer_handle);
     delete(timer_handle);
 end

code/online_plugins/BioSemi/run_readbiosemi.m

@@ -47,6 +47,9 @@ function run_readbiosemi(varargin)
 % start background acquisition
 onl_read_background(opts.new_stream,@()read_block(conn,opts),opts.update_freq);
 
+disp('Now reading...');
+
+
 
 % background block reader function
 function block = read_block(conn,opts)

code/online_plugins/DataRiver/run_readdatariver.m

@@ -56,6 +56,8 @@ function run_readdatariver(varargin)
 % create background reading job
 onl_read_background(new_stream,@()read_block(datariver_stream),update_freq)
 
+disp('Now reading...');
+
 
 % background data reading function
 function block = read_block(datariver_stream)

code/online_plugins/LSL/run_readlsl.m

@@ -128,9 +128,11 @@ function run_readlsl(varargin)
     timestamps = [];
 
     % start background acquisition
-    disp('Now reading...');
     onl_read_background(opts.new_stream,@()read_data(inlet,marker_inlet,opts.always_double),opts.update_freq);
 
+    disp('Now reading...');
+    
+    
     function result = read_data(data_inlet,marker_inlet,always_double)
         % get a new chunk of data
         [chunk,stamps] = data_inlet.pull_chunk();

code/online_plugins/LSL/run_writelsl.m

@@ -102,8 +102,9 @@ function run_writelsl(varargin)
 info = lsl_streaminfo(lib,opts.out_stream,'MentalState',length(opts.channel_names),opts.update_freq,'cf_float32',uid);
 % ... including some meta-data
 desc = info.desc();
+channels = desc.append_child('channels');
 for c=1:length(opts.channel_names)
-    newchn = desc.append_child('channel');
+    newchn = channels.append_child('channel');
     newchn.append_child_value('name',opts.channel_names{c});
     newchn.append_child_value('type',opts.out_form);
 end
@@ -113,7 +114,7 @@ function run_writelsl(varargin)
 
 % start background writer job
 onl_write_background( ...
-    'ResultWriter',@(y)send_sample(outlet,y),...
+    'ResultWriter',@(y)send_samples(outlet,y),...
     'MatlabStream',opts.in_stream, ...
     'Model',opts.pred_model, ...
     'OutputFormat',opts.out_form, ...
@@ -124,8 +125,10 @@ function run_writelsl(varargin)
     'StartDelay',0,...
     'EmptyResultValue',[]);
 
+disp('Now writing...');
 
-function send_sample(outlet,y)
+
+function send_samples(outlet,y)
 if ~isempty(y)
     outlet.push_chunk(y'); end
 

code/online_plugins/OSC/run_writeosc.m

@@ -47,7 +47,7 @@ function run_writeosc(varargin)
 declare_properties('name','OSC');
 
 % define arguments
-arg_define(varargin, ...
+opts = arg_define(varargin, ...
     arg({'pred_model','Model'}, 'lastmodel', [], 'Predictive model. As obtained via bci_train or the Model Calibration dialog.','type','expression'), ...
     arg({'in_stream','SourceStream'}, 'laststream',[],'Input Matlab stream. This is the stream that shall be analyzed and processed.'), ...
     arg({'out_hostname','OutputHost'}, '127.0.0.1',[],'Destination TCP hostname. Can be a computer name, URL, or IP address.'), ...
@@ -56,28 +56,46 @@ function run_writeosc(varargin)
     arg({'out_form','OutputForm'},'distribution',{'expectation','distribution','mode'},'Output form. Can be the expected value (posterior mean) of the target variable, or the distribution over possible target values (probabilities for each outcome, or parametric distribution), or the mode (most likely value) of the target variable.'), ...
     arg({'msg_format','MessageFormatter'},'@(D) num2cell(single(D(:))))',[],'Message Formatting function. Converts a BCI output (usually scalar or row vector, in some complex cases a matrix) into a cell array of OSC-compatible data types (float32,int32, ...).','type','expression'), ...
     arg({'update_freq','UpdateFrequency'},10,[],'Update frequency. This is the rate at which the graphics are updated.'), ...
-    arg({'pred_name','PredictorName'}, 'lastpredictor',[],'Name of new predictor. This is the workspace variable name under which a predictor will be created.'));
+    arg({'predict_at','PredictAt'}, {},[],'Predict at markers. If nonempty, this is a cell array of online target markers relative to which predictions shall be made. If empty, predictions are always made on the most recently added sample.','type','expression'), ...
+    arg({'pred_name','PredictorName'}, 'lastpredictor',[],'Name of new predictor. This is the workspace variable name under which a predictor will be created.'), ...
+    arg({'verbose_output','Verbose'}, false,[],'Verbose output. Whether to display verbose outputs (e.g., connection failure).'));
 
 % convert format strings to formatting functions
-if ischar(msg_format)
-    msg_format = @(D) sprintf(msg_format,D); end
+if ischar(opts.msg_format)
+    opts.msg_format = @(D) sprintf(opts.msg_format,D); end
 
 % connect to remote address via OSC
 if ~exist('osc_new_address','file')
     try
         build_osc;
-    catch
+    catch e
         error('The OSC library has not been built for your platform yet; see dependencies/OSC* for more info.'); 
     end
 end
-conn = osc_new_address(out_hostname,out_port);
+conn = osc_new_address(opts.out_hostname,opts.out_port);
 deleter = onCleanup(@()osc_free_address(conn)); % if the last reference to this is dropped, the connection is closed (on MATLAB 2008a+)
 
 % start background writer job
-onl_write_background(@(y)send_message(y,conn,out_path,msg_format,deleter),in_stream,pred_model,out_form,update_freq,0,pred_name);
+onl_write_background( ...
+    'ResultWriter',@(y)send_message(y,conn,opts.out_path,opts.msg_format,deleter),...
+    'MatlabStream',opts.in_stream, ...
+    'Model',opts.pred_model, ...
+    'OutputFormat',opts.out_form, ...
+    'UpdateFrequency',opts.update_freq, ...
+    'PredictorName',opts.pred_name, ...
+    'PredictAt',opts.predict_at, ...
+    'Verbose',opts.verbose_output, ...
+    'StartDelay',0,...
+    'EmptyResultValue',[]);
+
+disp('Now writing...');
+
 
 % background message sending function
-function send_message(y,conn,opath,formatter,deleter)
-msg = struct('path',opath,'data',{formatter(y)});
-if osc_send(conn,msg) == 0
-    error('OSC transmission failed.'); end
+function send_message(yy,conn,opath,formatter,deleter)
+for k=1:size(yy,1)
+    y = yy(k,:);
+    msg = struct('path',opath,'data',{formatter(y)});
+    if osc_send(conn,msg) == 0
+        error('OSC transmission failed.'); end
+end

code/online_plugins/SNAP/run_writesnap.m

@@ -81,40 +81,47 @@ function run_writesnap(varargin)
     'StartDelay',0,...
     'EmptyResultValue',[]);
 
+disp('Now writing...');
+
+
 % background message sending function
-function send_message(y,verbose_output,id,host,port,varname)
+function send_message(yy,verbose_output,id,host,port,varname)
 persistent conns;
 try
-    if isempty(y)
-        return; end
     % try to connect if necessary
     if ~isfield(conns,id)
         conns.(id) = connect(host,port); end
-    try
-        strm = conns.(id).strm;
-        if isscalar(y)
-            strm.writeBytes(char([sprintf('setup %s=%.5f',varname,y) 10]));
-        elseif ~isempty(y)
-            strm.writeBytes(char(['setup ' varname '=(' sprintf('%.5f,',y) ')' 10]));
-        end
-        strm.flush();
-    catch e1
-        if strcmp(e1.identifier, 'MATLAB:Java:GenericException')
-            % failed to send: try to re-connect...
-            conns.(id) = connect(host,port);
-        else
-            rethrow(e1);
+    % for each prediction...
+    for k=1:size(yy,1)
+        y = yy(k,:);
+        try
+            % send it off
+            strm = conns.(id).strm;
+            if isscalar(y)
+                strm.writeBytes(char([sprintf('setup %s=%.5f',varname,y) 10]));
+            elseif ~isempty(y)
+                strm.writeBytes(char(['setup ' varname '=(' sprintf('%.5f,',y) ')' 10]));
+            end
+            strm.flush();
+        catch e1
+            if strcmp(e1.identifier, 'MATLAB:Java:GenericException')
+                % failed to send: try to re-connect...
+                conns.(id) = connect(host,port);
+            else
+                rethrow(e1);
+            end
         end
     end
 catch e2
     if strcmp(e2.identifier, 'MATLAB:Java:GenericException')
         if verbose_output
             fprintf('Could not connect to %s:%.0f\n',host,port); end
     else
-        env_handleerror(e2);
+        hlp_handleerror(e2);
     end
 end
 
+
 function newconn = connect(host,port)
 import java.io.*
 import java.net.*

code/online_plugins/Simulated/run_readdataset.m

@@ -47,6 +47,8 @@ function run_readdataset(varargin)
 % start a background reading job
 onl_read_background(new_stream,@(stream)read_block(in_dataset,stream,always_double), update_freq);
 
+disp('Now reading...');
+
 
 % background block reader function
 function result = read_block(in_dataset,stream,always_double)