Scribd
Upload
10 views17 pages

BCI Framework with Kafka Integration

The document outlines the implementation of a BCI (Brain-Computer Interface) framework using PySide6 for GUI development, with components for Kafka messaging, clock synchronization, and various analysis environments. It includes classes for managing Kafka producers and consumers, handling UI interactions, and managing subprocesses. The framework is designed to facilitate real-time data processing and visualization in a modular manner.

Uploaded by

YeisonEng
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as TXT, PDF, TXT or read online on Scribd
10 views17 pages

BCI Framework with Kafka Integration

The document outlines the implementation of a BCI (Brain-Computer Interface) framework using PySide6 for GUI development, with components for Kafka messaging, clock synchronization, and various analysis environments. It includes classes for managing Kafka producers and consumers, handling UI interactions, and managing subprocesses. The framework is designed to facilitate real-time data processing and visualization in a modular manner.

Uploaded by

YeisonEng
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as TXT, PDF, TXT or read online on Scribd

"""

====
Core
====
"""

import os
import sys
import time
import json
import psutil
import pickle
import platform
import subprocess
from datetime import datetime
from typing import TypeVar, Optional

from [Link] import QUiLoader


from [Link] import Qt, QTimer, QSize, Signal, QThread, Slot
from [Link] import (
QPixmap,
QIcon,
QFontDatabase,
QKeySequence,
QShortcut,
)
from [Link] import (
QWidget,
QMainWindow,
QPushButton,
QLabel,
QCheckBox,
)

from kafka import KafkaProducer, KafkaConsumer


import ntplib

from .widgets import Montage, Projects, Connection, Records, Annotations


from .environments import (
Development,
Visualization,
StimuliDelivery,
TimeLockAnalysis,
)
from .config_manager import ConfigManager
from .configuration import ConfigurationFrame
from .subprocess_handler import run_subprocess
from .raspad import Raspad

KafkaMessage = TypeVar('KafkaMessage')
PathLike = TypeVar('PathLike')
HostLike = TypeVar('HostLike')
Millis = TypeVar('Milliseconds')

########################################################################
class ClockOffset(QThread):
""""""
signal_offset = Signal(object)

# ----------------------------------------------------------------------
def set_host(self, host: HostLike) -> None:
"""Set the host for kafka."""
[Link] = host

# ----------------------------------------------------------------------
def run(self) -> None:
""""""
try:
client = [Link]()
r = [Link]([Link])
clock_offset = [Link] # * 1000
self.signal_offset.emit(clock_offset)
print(f'offset: {clock_offset}')
except Exception:
pass

########################################################################
class Kafka(QThread):
"""Kafka run on a thread."""

signal_kafka_message = Signal(object)
signal_exception_message = Signal()
signal_produser = Signal()
keep_alive = True

# ----------------------------------------------------------------------
def set_host(self, host: HostLike) -> None:
"""Set the host for kafka."""
[Link] = host

# ----------------------------------------------------------------------
def stop(self) -> None:
"""Kill the thread."""
self.keep_alive = False
[Link]()

# ----------------------------------------------------------------------
def run(self) -> None:
"""Start the produser and consumer for Kafka."""
try:
self.create_produser()
self.signal_produser.emit()
[Link] = False
self.create_consumer()
except Exception:
self.signal_exception_message.emit()
[Link]()

# ----------------------------------------------------------------------
def create_consumer(self) -> None:
"""Basic consumer to check stream status and availability."""
bootstrap_servers = [f'{[Link]}:9092']
topics = [
'annotation',
'marker',
'command',
'eeg',
'aux',
'feedback',
]
[Link] = KafkaConsumer(
bootstrap_servers=bootstrap_servers,
value_deserializer=[Link],
auto_offset_reset='latest',
)

[Link](topics)
for message in [Link]:
self.last_message = [Link]()
[Link]['timestamp'] = [Link] / 1000

self.signal_kafka_message.emit(
{'topic': [Link], 'value': [Link]}
)

if not self.keep_alive:
return

# ----------------------------------------------------------------------
def create_produser(self) -> None:
"""The produser is used for stream annotations and markers."""
[Link] = KafkaProducer(
bootstrap_servers=[f'{[Link]}:9092'],
compression_type='gzip',
value_serializer=[Link],
)

########################################################################
class BCIFramework(QMainWindow):
""""""

# ----------------------------------------------------------------------
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)

self.load_fonts()
[Link] = QUiLoader().load(
[Link](
[Link]([Link](__file__)),
'qtgui',
'[Link]',
)
)
self.set_icons()

[Link]([Link], [Link])
[Link]([Link], [Link])

[Link](
lambda evt: self.show_interface('Development')
)
[Link](
lambda evt: self.show_interface('Visualizations')
)
[Link].actionStimuli_delivery.[Link](
lambda evt: self.show_interface('Stimuli_delivery')
)
[Link].actionTimelock_analysis.[Link](
lambda evt: self.show_interface('Timelock_analysis')
)
[Link](
lambda evt: self.show_interface('Home')
)
[Link](
lambda evt: self.show_interface('Documentation')
)
[Link].actionRaspad_settings.[Link](
lambda evt: self.show_interface('Raspad_settings')
)

[Link].stackedWidget_montage.setCurrentWidget(
[Link].page_montage
)

self.show_interface('Home')
[Link] = ConfigManager()

for i in range([Link].toolBar_Environs.layout().count()):
tool_button = (
[Link].toolBar_Environs.layout().itemAt(i).widget()
)
tool_button.setMaximumWidth(200)
tool_button.setMinimumWidth(200)

self.set_editor()
self.build_collapse_button()

[Link] = Montage(self)
[Link] = Connection(self)
[Link] = Projects([Link], self)
[Link] = Records([Link], self)
[Link] = Annotations([Link], self)
[Link] = Raspad(self)

[Link] = Development(self)
[Link] = Visualization(self)
self.stimuli_delivery = StimuliDelivery(self)
self.timelock_analysis = TimeLockAnalysis(self)

# self.status_bar('OpenBCI no connected')

[Link].tabWidget_widgets.setCurrentIndex(0)
[Link].tabWidget_data_analysis.setCurrentIndex(0)
[Link].tabWidget_stimuli_delivery.setCurrentIndex(0)
[Link].stackedWidget_noneeg.setCurrentIndex(0)
self.style_home_page()

[Link]()

docs = [Link](
[Link](
[Link]('BCISTREAM_ROOT'),
'documentation',
'[Link]',
)
)
[Link].webEngineView_documentation.setUrl(f'[Link]

self.subprocess_timer = QTimer()
self.subprocess_timer.[Link](self.save_subprocess)
self.subprocess_timer.setInterval(5000)
self.subprocess_timer.start()

self.clock_offset = 0
self.eeg_size = 0
self.aux_size = 0
self.sample_rate = 0
self.last_update = 0

# QTimer().singleShot(1000, self.calculate_offset)
QTimer().singleShot(3000, self.start_stimuli_server)

self.status_bar(message='', right_message=('disconnected', None))

shortcut_fullscreen = QShortcut(QKeySequence('F11'), [Link])


shortcut_fullscreen.[Link](
lambda: [Link]()
if [Link]() & [Link]
else [Link]()
)

shortcut_docs = QShortcut(QKeySequence('F1'), [Link])


shortcut_docs.[Link](
lambda: self.show_interface('Documentation')
)

shortcut_docs = QShortcut(QKeySequence('F2'), [Link])


shortcut_docs.[Link](self.toggle_dock_collapsed)

shortcut_docs = QShortcut(QKeySequence('F9'), [Link])


shortcut_docs.[Link](
self.timelock_analysis.show_fullscreen
)

[Link].toolBar_Environs.setStyleSheet(
"""
* {
min-width: 200px;
}
"""
)

# ----------------------------------------------------------------------
def set_icons(self) -> None:
"""The Qt resource system has been deprecated."""

def icon(name):
return QIcon(f"bci:/primary/{name}.svg")

[Link](icon("file"))
[Link](icon("latency2"))
[Link].actionStimuli_delivery.setIcon(icon("imagery"))
[Link].actionTimelock_analysis.setIcon(icon("brain"))
[Link](icon("home"))
[Link](icon("documentation"))

if [Link]([Link]('BCISTREAM_RASPAD')):
[Link].actionRaspad_settings.setIcon(icon("brain_settings"))
else:
[Link].actionRaspad_settings.setVisible(False)

[Link].pushButton_file.setIcon(icon("file"))
[Link].pushButton_brain.setIcon(icon("brain"))
[Link].pushButton_imagery.setIcon(icon("imagery"))
[Link].pushButton_docs.setIcon(icon("documentation"))
[Link].pushButton_latency.setIcon(icon("latency"))
[Link].pushButton_annotations.setIcon(icon("annotation"))
[Link].pushButton_timelock.setIcon(icon("latency2"))

[Link].pushButton_stop_preview.setIcon(
icon('media-playback-stop')
)
[Link].pushButton_script_preview.setIcon(
icon('media-playback-start')
)
[Link].pushButton_play_signal.setIcon(
icon('media-playback-start')
)
[Link].pushButton_clear_debug.setIcon(icon('edit-delete'))
[Link].pushButton_remove_record.setIcon(icon('edit-delete'))
[Link].pushButton_remove_annotations.setIcon(icon('edit-delete'))
[Link].pushButton_remove_markers.setIcon(icon('edit-delete'))
[Link].pushButton_remove_commands.setIcon(icon('edit-delete'))
[Link].pushButton_record.setIcon(icon('media-record'))

[Link].pushButton_projects_add_file.setIcon(icon('document-new'))
[Link].pushButton_projects_add_folder.setIcon(icon('folder-add'))
[Link].pushButton_add_project.setIcon(icon('folder-add'))
[Link].pushButton_projects_remove.setIcon(icon('edit-delete'))
[Link].pushButton_remove_project.setIcon(icon('edit-delete'))
[Link].pushButton_remove_montage.setIcon(icon('edit-delete'))

[Link].pushButton_projects.setIcon(icon('go-previous'))
[Link].pushButton_load_visualizarion.setIcon(icon('list-add'))

[Link].pushButton_open_records_folder.setIcon(icon('folder'))
[Link].pushButton_open_extension_folder.setIcon(icon('folder'))
[Link].pushButton_open_project_raspad.setIcon(icon('folder'))

for i in range(1, 5):


getattr([Link], f'pushButton_update_wifi{i}').setIcon(
icon('gtk-convert')
)

[Link](icon("logo"))

# ----------------------------------------------------------------------
def save_subprocess(self) -> None:
"""Save in a file all the child subprocess."""
current_process = [Link]()
children = current_process.children(recursive=True)
[Link](current_process)
file = [Link]([Link]['BCISTREAM_HOME'], '.subprocess')
try:
with open(file, 'w') as file_:
[Link](
{[Link]: [Link]() for ch in children}, file_, indent=2
)
except: # [Link]: [Link] process no longer
exists
pass

# ----------------------------------------------------------------------
def load_fonts(self) -> None:
"""Load custom fonts."""
fonts_path = [Link](
[Link]['BCISTREAM_ROOT'], 'assets', 'fonts'
)

for font_dir in [[Link]('dejavu', 'ttf')]:


for font in filter(
lambda s: [Link]('.ttf'),
[Link]([Link](fonts_path, font_dir)),
):
[Link](
[Link](fonts_path, font_dir, font)
)

# ----------------------------------------------------------------------
def build_collapse_button(self) -> None:
"""Custom collapsible widgets area."""
# This dockWidget has empy name so,
# with in that space is possible to add a custom widget
self.pushButton_collapse_dock = QPushButton(
[Link].dockWidget_global
)
self.pushButton_collapse_dock.[Link](
self.set_dock_collapsed
)

icon = QIcon('bci:/primary/[Link]')
self.pushButton_collapse_dock.setIcon(icon)
self.pushButton_collapse_dock.setCheckable(True)
self.pushButton_collapse_dock.setFlat(True)

w = [Link].tabWidget_widgets.tabBar().width()
self.pushButton_collapse_dock.setMinimumWidth(w)
self.pushButton_collapse_dock.setMaximumWidth(w)
self.pushButton_collapse_dock.move(5, 5)

# ----------------------------------------------------------------------
def set_dock_collapsed(self, collapsed: bool) -> None:
"""Collapse widgets area."""
if collapsed:
w = [Link].tabWidget_widgets.tabBar().width() + 10
icon = QIcon('bci:/primary/[Link]')
self.previous_width = [Link].dockWidget_global.width()
else:
if [Link].dockWidget_global.width() > 50:
return
icon = QIcon('bci:/primary/[Link]')
w = self.previous_width

self.pushButton_collapse_dock.setIcon(icon)
[Link].dockWidget_global.setMaximumWidth(w)
[Link].dockWidget_global.setMinimumWidth(w)

if not collapsed:
[Link].dockWidget_global.setMaximumWidth(w * 99)
[Link].dockWidget_global.setMinimumWidth(0)

# ----------------------------------------------------------------------
def toggle_dock_collapsed(self) -> None:
""""""
if [Link].dockWidget_global.maximumWidth() < 100:
self.set_dock_collapsed(False)
else:
self.set_dock_collapsed(True)

# ----------------------------------------------------------------------
def connect(self) -> None:
"""Connect events."""
[Link].tabWidget_widgets.[Link](
lambda: self.set_dock_collapsed(False)
)

[Link].pushButton_add_project_2.[Link](
[Link].show_create_project_dialog
)

[Link].pushButton_show_visualization.[Link](
lambda evt: self.show_interface('Visualizations')
)
[Link].pushButton_show_stimuli_delivery.[Link](
lambda evt: self.show_interface('Stimuli_delivery', 0)
)
[Link].pushButton_show_timelock.[Link](
lambda evt: self.show_interface('Timelock_analysis', 0)
)

[Link].pushButton_show_documentation.[Link](
lambda evt: self.show_interface('Documentation')
)

[Link].pushButton_show_about.[Link](self.show_about)
[Link].pushButton_show_configurations.[Link](
self.show_configurations
)

[Link].pushButton_show_latency_correction.[Link](
lambda evt: self.show_interface('Stimuli_delivery', 1)
)

[Link].pushButton_show_annotations.[Link](
lambda evt: [Link].tabWidget_widgets.setCurrentIndex(3)
)

[Link].tabWidget_widgets.[Link](self.show_widget)
[Link].pushButton_open_extension_folder.[Link](
lambda: self.open_folder_in_system(
[Link].label_projects_path.text()
)
)
[Link].pushButton_open_records_folder.[Link](
lambda: self.open_folder_in_system(
[Link].label_records_path.text()
)
)

[Link].checkBox_board1.[Link](
lambda checked: not checked
and [Link].checkBox_board1.setChecked(True)
)
# [Link].checkBox_board1.[Link](
# lambda b: self.board_handle([Link].checkBox_board2, b))
[Link].checkBox_board2.[Link](
lambda b: self.board_handle([Link].checkBox_board3, b)
)
[Link].checkBox_board3.[Link](
lambda b: self.board_handle([Link].checkBox_board4, b)
)

# ----------------------------------------------------------------------
def board_handle(self, chbox: QCheckBox, enable: bool) -> None:
""""""
[Link](enable)
[Link](False)

# ----------------------------------------------------------------------
def open_folder_in_system(self, path: PathLike) -> None:
""""""
if [Link]() == "Windows":
[Link](path)
elif [Link]() == "Darwin":
[Link](["open", path])
else:
[Link](["xdg-open", path])

# ----------------------------------------------------------------------
def show_interface(
self, interface: str, sub_widget: Optional[int] = None
) -> None:
"""Switch between environs."""
[Link].stackedWidget_modes.setCurrentWidget(
getattr([Link], f"page{interface}")
)
for action in [Link].toolBar_Environs.actions():
[Link](False)
for action in [Link].toolBar_Documentation.actions():
[Link](False)
if action := getattr([Link], f"action{interface}", False):
[Link](True)

if mod := getattr(
self, f'{[Link]().replace(" ", "_")}', False
):
if on_focus := getattr(mod, 'on_focus'):
on_focus()

if sub_widget != None:
if interface == 'Stimuli_delivery':
[Link].tabWidget_stimuli_delivery.setCurrentIndex(
sub_widget
)

# ----------------------------------------------------------------------
def show_widget(self, index: int) -> None:
"""Call `on_focus` method for all widgets."""
widget = [Link].tabWidget_widgets.tabText(index)
if wg := getattr(self, [Link](), False):
if on_focus := getattr(wg, 'on_focus', False):
on_focus()

# ----------------------------------------------------------------------
def set_editor(self) -> None:
"""Change some styles."""
[Link].plainTextEdit_preview_log.setStyleSheet(
"""
*{
font-weight: normal;
font-family: 'DejaVu Sans Mono';
}
"""
)

[Link](
f"""
*{{
background: {[Link]('QTMATERIAL_SECONDARYDARKCOLOR')};
}}
"""
)

# # ----------------------------------------------------------------------
# def remove_widgets_from_layout(self, layout) -> None:
# """"""
# for i in reversed(range([Link]())):
# if item := [Link](i):
# [Link]().deleteLater()

# ----------------------------------------------------------------------
def status_bar(
self,
message: Optional[str] = None,
right_message: Optional[str] = None,
) -> None:
"""Update messages for status bar."""
statusbar = [Link]()

if not hasattr(statusbar, 'right_label'):


statusbar.right_label = QLabel(statusbar)
statusbar.right_label.setAlignment([Link])
statusbar.right_label.mousePressEvent = (
lambda evt: [Link].tabWidget_widgets.setCurrentWidget(
[Link].tab_connection
)
)
[Link](statusbar.right_label)

[Link] = QPushButton()
[Link]('class', 'connection')
[Link](True)
[Link]([Link])

if message:
[Link](message)

if right_message:
# [Link](False)
message, status = right_message
statusbar.right_label.setText(message)
if status is None:
[Link](True)
if 'light' in [Link]('QTMATERIAL_THEME'):
[Link](
f"""*{{
border: 1px solid
{[Link]('QTMATERIAL_SECONDARYDARKCOLOR')};
background-color:
{[Link]('QTMATERIAL_SECONDARYDARKCOLOR')};}}"""
)
else:
[Link](
f"""*{{
border: 1px solid
{[Link]('QTMATERIAL_SECONDARYLIGHTCOLOR')};
background-color:
{[Link]('QTMATERIAL_SECONDARYLIGHTCOLOR')};}}"""
)
else:
[Link](False)
if status:
[Link](
"""*{
border: 1px solid #3fc55e;
background-color: #3fc55e;}"""
)
else:
[Link](
"""*{
border: 1px solid #dc3545;
background-color: #dc3545;}"""
)

# ----------------------------------------------------------------------
def style_home_page(self) -> None:
"""Set the styles for home page."""

if [Link]([Link]('BCISTREAM_RASPAD')):
size = 50
version = 'RASPAD'
else:
size = 80
version = ''
if '--local' in [Link]:
mode = 'DEBUG'
else:
mode = ''

style = f"""
*{{
width: {size}px;
height: {size}px;
max-width: {size}px;
min-width: {size}px;
max-height: {size}px;
min-height: {size}px;
background-color: {[Link]('secondaryColor')}
}}
"""
[Link].pushButton_file.setStyleSheet(style)
[Link].pushButton_brain.setStyleSheet(style)
[Link].pushButton_imagery.setStyleSheet(style)
[Link].pushButton_docs.setStyleSheet(style)
[Link].pushButton_latency.setStyleSheet(style)
[Link].pushButton_annotations.setStyleSheet(style)
[Link].pushButton_timelock.setStyleSheet(style)

style = f"""
QFrame {{
background-color: {[Link]('secondaryColor')}
}}
"""
[Link].frame_3.setStyleSheet(style)
[Link].frame_2.setStyleSheet(style)
[Link].frame_5.setStyleSheet(style)
[Link].frame_6.setStyleSheet(style)
[Link].frame_7.setStyleSheet(style)
[Link].frame_4.setStyleSheet(style)
[Link](style)

style = """
*{
color: black;
}
"""

# FONTSIZE
labels = [
[Link].label_15,
[Link].label_16,
[Link].label_17,
[Link].label_20,
[Link].label_21,
[Link].label_22,
[Link].label_23,
]
for label in labels:
if [Link]([Link]('BCISTREAM_RASPAD')):
[Link](
[Link]().replace('font-size:12pt', 'font-size:10pt')
)
with open(
[Link]([Link]['BCISTREAM_ROOT'], '_version.txt'), 'r'
) as file:
[Link].label_software_version.setText(
f'{[Link]().strip()} {version} {mode}'
)

# ----------------------------------------------------------------------
def show_about(self, *args, **wargs) -> None:
"""Display about window."""
frame = [Link](
[Link]['BCISTREAM_ROOT'], 'framework', 'qtgui', '[Link]'
)
about = QUiLoader().load(frame, [Link])
[Link](True)

about.pushButton_close.[Link](lambda evt: [Link]())

[Link](QSize(600, 200))
[Link](QSize(600, 200))

banner = [Link](
[Link]['BCISTREAM_ROOT'], 'assets', '[Link]'
)
[Link](QPixmap(banner))

[Link](0)

[Link](
f"""
QPlainTextEdit {{
font-weight: normal;
font-family: 'DejaVu Sans Mono';
font-size: 13px;
}}

QTextEdit {{
color: {[Link]('QTMATERIAL_SECONDARYTEXTCOLOR')};
}}
"""
)

center = [Link](self).availableGeometry().center()
geometry = [Link]()
[Link](center)
[Link]([Link]())

[Link]()

# ----------------------------------------------------------------------
@Slot()
def on_kafka_event(self, value: KafkaMessage) -> None:
"""Register annotations and markers."""

[Link] = True

if value['topic'] == 'marker':
[Link].add_marker(
[Link](value['value']['timestamp']),
value['value']['marker'],
)
elif value['topic'] == 'command':
[Link].add_command(
[Link](value['value']['timestamp']),
value['value']['command'],
)
elif value['topic'] == 'annotation':
[Link].add_annotation(
[Link](value['value']['timestamp']),
value['value']['duration'],
value['value']['description'],
)
elif value['topic'] == 'feedback':
self.handle_feedback(value['value'])

elif value['topic'] in ['eeg', 'aux']:

if [Link]() < self.last_update + 1:


return
self.last_update = [Link]()

# Use only remote times to make the calculation, so the


# differents clocks not affect the measure
binary_created = [Link](
min(value['value']['context']['[Link]'])
)
message_created = [Link](
value['value']['timestamp']
)
since = (message_created - binary_created).total_seconds()
if value['topic'] == 'eeg':
self.eeg_size = value['value']['data'].shape
elif value['topic'] == 'aux':
self.aux_size = value['value']['data'].shape

if since * 1000 > 2000:


color = '#ffc107' # old data
else:
color = '#3fc55e' # recent data

message = f'Last package streamed <b style="color:


{color};">{since*1000:0.2f} ms </b> ago | EEG{self.eeg_size} | AUX{self.aux_size}'

if status := getattr([Link], 'recording_status', None):


message += f' | <b style="color:#dc3545;">{status}</b>'

self.status_bar(right_message=(message, True))
# self.last_update = value['value']['timestamp']

# ----------------------------------------------------------------------
def update_kafka(self, host: HostLike) -> None:
"""Try to restart kafka services."""
if hasattr(self, 'thread_kafka'):
self.thread_kafka.stop()
self.status_bar(right_message=('No streaming', False))
# try:
self.thread_kafka = Kafka()
self.thread_kafka.signal_kafka_message.connect(self.on_kafka_event)
# self.thread_kafka.first_consume.connect(lambda:
[Link].on_connect(
# True))
self.thread_kafka.signal_exception_message.connect(
self.kafka_message
)
self.thread_kafka.signal_produser.connect(
self.kafka_produser_connected
)
self.thread_kafka.set_host(host)
self.thread_kafka.start()

[Link] = QTimer()
[Link](1000)
[Link](self.keep_updated)
[Link]()

# ----------------------------------------------------------------------
def calculate_offset(self) -> None:
""""""
host = self.thread_kafka.host
if host != 'localhost':
self.offset_thread = ClockOffset()
self.offset_thread.signal_offset.connect(self.set_offset)
self.offset_thread.set_host(self.thread_kafka.host)
self.offset_thread.start()
else:
self.set_offset(0)

# ----------------------------------------------------------------------
def set_offset(self, offset: Millis) -> None:
""""""
[Link]['BCISTREAM_OFFSET'] = [Link](offset)

# ----------------------------------------------------------------------
def stop_kafka(self) -> None:
"""Stop kafka."""
[Link] = False
if hasattr(self, 'thread_kafka'):
self.thread_kafka.stop()
self.status_bar(right_message=('No streaming', False))

# ----------------------------------------------------------------------
def keep_updated(self) -> None:
"""The status is update each 3 seconds."""
if hasattr(self, 'thread_kafka') and hasattr(
self.thread_kafka, 'last_message'
):
last = [Link]() - self.thread_kafka.last_message
if [Link] > 3:
self.status_bar(right_message=('No streaming', False))
[Link].set_enable(False)
[Link].pushButton_record.setEnabled(False)
[Link] = False
else:
[Link].set_enable(True)
[Link].pushButton_record.setEnabled(True)
# ----------------------------------------------------------------------
@Slot()
def kafka_message(self) -> None:
"""Error on kafka."""
[Link] = False
if [Link].comboBox_host.currentText() != 'localhost':
self.conection_message = f'* Imposible to connect with remote Kafka on
"{[Link].comboBox_host.currentText()}".'
else:
self.conection_message = (
'* Kafka is not running on this machine.'
)

self.status_bar(right_message=('Disconnected', None))

# ----------------------------------------------------------------------
@Slot()
def kafka_produser_connected(self) -> None:
"""If produser connected is posible the consumer too."""
self.status_bar(right_message=('Connected', False))

# ----------------------------------------------------------------------
def show_configurations(self, *args, **kwargs) -> None:
"""Show configuration window."""
configuration = ConfigurationFrame(self)
[Link]()

# ----------------------------------------------------------------------
def handle_feedback(self, data) -> None:
""""""
if fn := getattr(self, f"feedback_{[Link]('name', False)}", False):
fn(data['value'])
if fn := getattr(self, f"feedback_{[Link]('command', False)}", False):
fn(data['value'])

# ----------------------------------------------------------------------
def feedback_set_latency(self, latency: Millis) -> None:
""""""
[Link]['BCISTREAM_SYNCLATENCY'] = [Link](latency)

# ----------------------------------------------------------------------
def start_stimuli_server(self) -> None:
""""""
if '--local' in [Link]:
self.bciframework_server = run_subprocess(
[
[Link],
[Link](
[Link]['BCISTREAM_ROOT'],
'python_scripts',
'bciframework_server',
'[Link]',
),
]
)
else:
self.bciframework_server = run_subprocess(
[
[Link],
[Link](
[Link]['BCISTREAM_HOME'],
'python_scripts',
'bciframework_server',
'[Link]',
),
]
)

You might also like