<jupyter_start><jupyter_text>Data analysisThis feature able the user to develop real-time data analysis, consist of the complete Python-powered environment, with a set of custom methods for agile development.    Extensions > New Extension > Data analysis  Bare minimum<jupyter_code>from bci_framework.extensions.data_analysis import DataAnalysis

class Analysis(DataAnalysis):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

if __name__ == '__main__':
    Analysis()<jupyter_output><empty_output><jupyter_text>Data stream accessThe data stream is accessed asynchronously with the `loop_consumer` decorator from `bci_framework.extensions.data_analysis`, this decorator requires the Kafka topics to access.There is 2 topics availables for `loop_consumer`: `eeg` and `marker`.<jupyter_code>from bci_framework.extensions.data_analysis import DataAnalysis, loop_consumer

class Analysis(DataAnalysis):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.stream()

    @loop_consumer('eeg', 'marker')
    def stream(self):
        print('Incoming data...')

if __name__ == '__main__':
    Analysis()<jupyter_output><empty_output><jupyter_text>The decorated method receives 3 optional arguments: `data`, `topic` and `frame`. **data:** `(eeg, aux)` if topic is `eeg`, `marker_value` if topic is `marker`  **kafka_stream:** The `stream` object from Kafka.  **topic:** The topic of the Kafka stream, this object is available too in the object `data.topic`  **frame:** Incremental flag with the counter of streamed data.  **latency:** The time bewteen acquisition and read.<jupyter_code>@loop_consumer('eeg')
def stream(self, data, topic, frame, latency):
    eeg, aux = data
    
    print(f'Incoming data #{frame}')
    print(f'EEG{eeg.shape}')
    print(f'AUX{aux.shape}')
    print(f'Topic: {topic}')
    print(f'Latency: {latency}')<jupyter_output><empty_output><jupyter_text>The above code will execute every data stream input, and the below code will execute only when a marker is streamed.<jupyter_code>@loop_consumer('marker')
def stream(self, data, topic, frame):
    marker_value = data
    
    print(f'Incoming marker: {marker_value}')<jupyter_output><empty_output><jupyter_text>Is **not possible**, use the decorator `loop_consumer` in more than one place, so the argument `topic` could be used to create a flow control.<jupyter_code>@loop_consumer('eeg', 'marker')
def stream(self, data, topic, frame):
    
    if topic == 'eeg':
        eeg, aux = data
        print("EEG data incomming..")
    
    elif topic == 'marker':
        marker_value = data
        print("Marker incomming..")<jupyter_output><empty_output><jupyter_text>Simulate data streamUsing `fake_loop_consumer` instead of `loop_consumer` is possible to create a fake data stream.<jupyter_code>from bci_framework.extensions.data_analysis import DataAnalysis, fake_loop_consumer
import logging

class Analysis(DataAnalysis):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.stream()

    @fake_loop_consumer('eeg')
    def stream(self):
        logging.debug('Incoming data...')

if __name__ == '__main__':
    Analysis()<jupyter_output><empty_output><jupyter_text>Built in methods  Buffer / Sliding windowWe can use `self.create_buffer` to implement an automatic buffer with a fixed time view, for example, a buffer of 30 seconds:<jupyter_code>self.create_buffer(seconds=30)<jupyter_output><empty_output><jupyter_text>The data can be accesed with `self.buffer_eeg` and `self.buffer_aux`<jupyter_code>class Analysis(DataAnalysis):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.create_buffer(seconds=30)
        self.stream()

    @loop_consumer('eeg')
    def stream(self):
         eeg = self.buffer_eeg
         aux = self.buffer_aux<jupyter_output><empty_output><jupyter_text>The `self.create_buffer` method receives other arguments like `aux_shape`, `fill` and `samples`.<jupyter_code>self.create_buffer(seconds=30, aux_shape=3, fill=0, resampling=1000)<jupyter_output><empty_output><jupyter_text>**aux_shape:** The dimension of the auxiliary data, 3 by default.  **fill:** Initialize buffet with this value, 0 by default.  **resampling:** This value is used to resampling the data.    ResamplingThe resampling is defined when the buffer is created, with the argument `resampling` this value is not strictly used, instead a near and optimal value is calculated based on the sampling rate and the buffer size.<jupyter_code>class Analysis(DataAnalysis):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.create_buffer(seconds=30, resampling=1000)
        self.stream()

    @loop_consumer('eeg')
    def stream(self):
         eeg = self.buffer_eeg_resampled
         aux = self.buffer_aux_resampled
         
         print(f'EEG{eeg.shape}')
         print(f'AUX{aux.shape}')<jupyter_output><empty_output><jupyter_text>The resampling will not affect the buffer, the both data are accessible all the time.  Data slicing referenced by markersConsist of a method that crops the available data with a marker reference. The  decorator `@marker_slicing` do the trick. Receives the `markers`, a `t0` that indicate how many seconds to crop before the marker and the `duration` of the slice.<jupyter_code>from bci_framework.extensions.data_analysis import DataAnalysis, marker_slicing

class Analysis(DataAnalysis):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        
        # Needs to be greater than the duration of the slice.
        self.create_buffer(seconds=30, aux_shape=3)
        self.slicing()

    @marker_slicing(['Right', 'Left'], t0=-2, duration=6)
    def slicing(self, eeg, aux, timestamp, marker):
        
        print(eeg.shape)
        print(aux.shape)
        print(timestamp.shape)
        print(marker)
        print()

if __name__ == '__main__':
    Analysis()<jupyter_output><empty_output><jupyter_text>The above code will be executed each time that and slice is available. The `buffer` must be greater than the duration of the desired slice.**eeg:** The EEG data croped (`channels, time`).  **aux:** The AUX data croped (`aux, time`).  **timestamp:** The `timestamp` vector.  **marker:** The `marker` that trigger the crop.  **latency:** The time bewteen acquisition and read.   Receive markersThe markers can be accessed specifying the topic `marker` in the `loop_consumer`<jupyter_code>@loop_consumer('marker')<jupyter_output><empty_output><jupyter_text>Send commands, annotations and feedbacksThe commands are used to communicate outputs into the real world, or other systems, they can also be read in the **Stimuli delivery** to create neurofeedback applications. To activate this feature just add the `enable_produser` argument as `True` into the `DataAnalysis` subclass.<jupyter_code>if __name__ == '__main__':
    Analysis(enable_produser=True)<jupyter_output><empty_output><jupyter_text>Once activate the producer, the methods `self.send_command`, `self.send_feedback` and `self.end_annotation`are available.<jupyter_code>@loop_consumer('eeg')
def stream(self):
     eeg = self.buffer_eeg_resampled
     aux = self.buffer_aux_resampled

    [...]  # amazing data analysis
    
    self.send_command('MyCommand', value=45)<jupyter_output><empty_output><jupyter_text>The `self.send_annotation` also receive the optional argument `duration`.<jupyter_code>self.send_annotation('Data record start')
self.send_annotation('The subject yawn', duration=5)<jupyter_output><empty_output><jupyter_text>The self.send_feedback receive any kind of Python data structure.<jupyter_code>feed = {'var1': 0,
        'var2': True,
        'var3': 'Right',
       }
self.send_feedback(**feed)

self.send_feedback(a=0, b=2.3, c='Left')<jupyter_output><empty_output><jupyter_text>A generic producer also is available:<jupyter_code>self.generic_produser(topic, data)<jupyter_output><empty_output><jupyter_text>Communication between analysis processLet's build a script that will acts like **Kafka transformer**, this script reads the raw EEG data, calculate their EEG spectrum using Fourier and inject back again into the stream. This can be other advanced processing tasks, like classifications using neural networks.<jupyter_code>from bci_framework.extensions.data_analysis import DataAnalysis, loop_consumer
from bci_framework.extensions import properties as prop
from gcpds.utils.processing import fourier

class Analysis(DataAnalysis):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.create_buffer(seconds=30, resampling=1000)
        self.stream()

    @loop_consumer('eeg')
    def stream(self):
         W, EEG = fourier(self.buffer_eeg, fs=prop.SAMPLE_RATE, axis=1)
         data = {'amplitude': EEG,
                 'frequency': W}
         self.generic_produser('spectrum', data)

if __name__ == '__main__':
    Analysis(enable_produser=True)<jupyter_output><empty_output><jupyter_text>Now, in another script, we will write a **Kafka consumer** this script will consume from the previously created stream.<jupyter_code>from bci_framework.extensions.data_analysis import DataAnalysis, loop_consumer
from bci_framework.extensions import properties as prop
from gcpds.utils.processing import fourier

class Analysis(DataAnalysis):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.stream()

    @loop_consumer('spectrum')
    def stream(self, data):
        data = data.value['data']
        
        EEG = data['amplitude']
        W = data['frequency']

if __name__ == '__main__':
    Analysis()<jupyter_output><empty_output><jupyter_text>This examples are available by default in the framework extensions explorer.The `spectrum` topic must be created before to use the topic:<jupyter_code>kafka-topics.sh --create --bootstrap-server localhost:2181 --replication-factor 1 --partitions 1 --topic feedback<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10825560627108553, 'fraction_numerical': 0.00535820599325263, 'mean_word_length': 4.819284064665127, 'pattern_counts': {'":': 0, '<': 81, '<?xml version=': 0, '>': 83, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '20', 'dir': 'jupyter-structured-clean-dedup', 'id': '34662', 'license': 'BSD-2-Clause', 'n_tokens_mistral': 3101, 'n_tokens_neox': 2995, 'n_words': 884, 'path': 'docs/source/notebooks/70-data_analysis.ipynb', 'repo_name': 'UN-GCPDS/bci-framework', 'size': '10078'}