metabci.brainflow.workers module

Start another process, define a framework for offline modeling and online processing with three functions:

pre(): for offline modeling;

consume(): for online prediction;

post(): for subsequent custom operations.

In the actual usage process, you only need to customize the operations of the above functions.

class metabci.brainflow.workers.ProcessWorker(timeout: float = 0.001, name: str | None = None)[source]

Bases: Process

Online processing.

author: Lichao Xu

Created on: 2021-04-01

update log:

2022-08-10 by Wei Zhao

Parameters:

  • timeout (float) – Timer setting.

  • name (str) – Custom name for the online processing process.

daemon
Type:

bool

_exit

Multiprocess event handling.

_in_queue

Data sharing between the online processing process and the main process.

Type:

queue

Tip

A example using brainflow. worker
 1from brainflow. worker import ProcessWorker
 2class FeedbackWorker(ProcessWorker):
 3    def __init__():
 4        #Initialization
 5
 6    def pre(self):
 7        #Off-line modeling
 8
 9        #Online processing of data flow between stimulus interfaces
10        info = StreamInfo(
11            name='meta_feedback',
12            type='Markers',
13            channel_count=1,
14            nominal_srate=0,
15            channel_format='int32',
16            source_id=self.lsl_source_id)
17        self.outlet = StreamOutlet(info)
18        print('waiting connection...')
19        while not self._exit:
20            if self.outlet.wait_for_consumers(1e-3):
21                break
22        print('Connected')
23
24    def consume(self, data) :
25        #Online processing
26        if self.outlet.have_consumers ():
27            self.outlet.push_sample(“online results，list")
28
29    def post(self):
30        pass
clear_queue()[source]

Clear the queue.

author: Lichao Xu

Created on: 2021-04-01

update log:

2022-08-10 by Wei Zhao

abstract consume(data)[source]

Custom function to process online data.

author: Lichao Xu

Created on: 2021-04-01

update log:

2022-08-10 by Wei Zhao

Parameters:

data (ndarray, shape(n_samples, n_channels+1)) – Single trial of online data.

abstract post()[source]
abstract pre()[source]

Custom function to build a model using offline data.

author: Lichao Xu

Created on: 2021-04-01

update log:

2022-08-10 by Wei Zhao

put(data)[source]

Put the data in the queue

author: Lichao Xu

Created on: 2021-04-01

update log:

2022-08-10 by Wei Zhao

Parameters:

data (ndarray, shape(n_samples, n_channels+1)) – Single trial of online data.

run()[source]

author: Lichao Xu

Created on: 2021-04-01

update log:

2022-08-10 by Wei Zhao

Online processing process:

① Customize the pre() function to build a model using offline data.

② Clear the queue and wait for data retrieval thread in the main process to get data within a fixed time.

③ Customize the consume() function to process online data and provide feedback.

④ Customize the post() function to perform subsequent operations.

⑤ Wait for the next online label to start the next online processing.

⑥ Close the online processing process, clear the queue, and stop online experiments.

settimeout(timeout=0.01)[source]

Set the timer.

author: Lichao Xu

Created on: 2021-04-01

update log:

2022-08-10 by Wei Zhao

stop()[source]

Stop the online processing process.

author: Lichao Xu

Created on: 2021-04-01

update log:

2022-08-10 by Wei Zhao