Frameworks

Step 8 — Adding a GUI

Goal: add a real-time PyQt5 operator interface that shows live data and accepts parameter updates. You'll use: SharedRingBuffer, multiprocessing.Queue, QueueHandler Builds on: Step 7 Prerequisites: the implementation guide steps 1–7; pip install PyQt5 (and optionally pyqtgraph for plots). Anchored example: examples/generic_gui/

The framework uses a dual-process architecture — the backend engine runs in its own process and the GUI runs in the main process. They communicate via multiprocessing.Queue.

1. Dual-Process Architecture#

Main Process (GUI)                    Backend Process
┌─────────────────────────────────┐   ┌─────────────────────────────────┐
│  QApplication.exec_()           │   │  ClosedLoopEngine.run()         │
│  ┌───────────────────────────┐  │   │  ┌───────────────────────────┐  │
│  │  MainWindow               │  │   │  │  DataStreamer              │  │
│  │  ┌─────────────────────┐  │  │   │  │  ControlStrategy          │  │
│  │  │  Slider → control_q ├──┼──┼──►├──┤  Stimulator               │  │
│  │  └─────────────────────┘  │  │   │  │  QueueHandler             │  │
│  │  ┌─────────────────────┐  │  │   │  │  SharedRingBuffer.write() │  │
│  │  │  Plot ← ring buffer ◄──┼──┼───┤  └───────────────────────────┘  │
│  │  └─────────────────────┘  │  │   └─────────────────────────────────┘
│  └───────────────────────────┘  │
└─────────────────────────────────┘
        utility_q (bidirectional)

Two multiprocessing.Queue objects connect the processes:

control_queue: GUI → Backend. Carries ControlMessage updates (parameter changes).
utility_queue: Bidirectional. Carries UtilityMessage (FES on/off, recording, calibration).

A SharedRingBuffer in shared memory carries high-frequency sensor data to the GUI at 30 Hz without using queues (the backend writes every cycle, the GUI reads whenever it polls).

2. IPC Messages#

ControlMessage (GUI → Backend)#

Use ControlMessage (or your subclass) to send parameter changes from the GUI. See Step 5 — Runtime parameters for how to subclass it.

UtilityMessage (Bidirectional)#

UtilityMessage carries system state flags:

Field	Type	Direction	Description
`FesIsActive`	`bool`	GUI → Backend	Global FES enable/disable
`RecordingIsActive`	`bool`	GUI → Backend	Toggle data recording
`CalibrationLoopIsActive`	`bool`	GUI → Backend	Request calibration cycle
`FolderPath`	`str`	GUI → Backend	Session data folder path
`TSAPIStepDetectionIsActive`	`bool`	GUI → Backend	Step detector mode
`VUStepDetectionIsActive`	`bool`	GUI → Backend	Step detector mode
`ModelBasedStepDetectionIsActive`	`bool`	GUI → Backend	Step detector mode

QueueHandler#

QueueHandler wraps both queues and provides auto-send on field assignment:

from fes_framework.ipc.queue_handler import QueueHandler

handler = QueueHandler(
    control_queue=control_queue,
    utility_queue=utility_queue,
)

# Assigning a field auto-sends the message to the queue:
handler.utility_message.FesIsActive = True   # → sent immediately to backend
handler.control_message.my_threshold = 25.0  # → sent immediately to backend

No explicit send() call is needed — the __setattr__ callback fires automatically.

3. Minimal PyQt5 GUI Skeleton#

Here is the minimum GUI runner that integrates with orchestrator.launch():

# gui.py
from PyQt5.QtWidgets import QApplication, QMainWindow, QWidget, QVBoxLayout, QPushButton, QCheckBox
from PyQt5.QtCore import QTimer
import sys

from fes_framework.ipc.queue_handler import QueueHandler


def my_gui_main(control_queue, utility_queue):
    """
    GUI entry point called by orchestrator.launch().

    Args:
        control_queue: multiprocessing.Queue for ControlMessage (GUI → backend)
        utility_queue: multiprocessing.Queue for UtilityMessage (bidirectional)
    """
    app = QApplication(sys.argv)
    window = MainWindow(control_queue, utility_queue)
    window.show()
    app.exec_()


class MainWindow(QMainWindow):
    def __init__(self, control_queue, utility_queue):
        super().__init__()
        self.setWindowTitle("FES Application")
        self.setMinimumSize(600, 400)

        # IPC handler — auto-sends messages on field changes
        self.handler = QueueHandler(
            control_queue=control_queue,
            utility_queue=utility_queue,
        )

        self._build_ui()

    def _build_ui(self):
        central = QWidget()
        self.setCentralWidget(central)
        layout = QVBoxLayout(central)

        # FES Active toggle
        self.fes_checkbox = QCheckBox("FES Active")
        self.fes_checkbox.stateChanged.connect(self._on_fes_toggled)
        layout.addWidget(self.fes_checkbox)

        # Calibrate button
        self.calibrate_btn = QPushButton("Calibrate MoCap")
        self.calibrate_btn.clicked.connect(self._on_calibrate)
        layout.addWidget(self.calibrate_btn)

        layout.addStretch()

    def _on_fes_toggled(self, state):
        # state is Qt.Checked (2) or Qt.Unchecked (0)
        self.handler.utility_message.FesIsActive = bool(state)

    def _on_calibrate(self):
        self.handler.utility_message.CalibrationLoopIsActive = True

Launching with the GUI#

# run.py
from fes_framework.orchestrator import launch
from my_strategy import MyFESStrategy
from gui import my_gui_main

if __name__ == "__main__":
    launch(
        MyFESStrategy,
        gui_runner=my_gui_main,
        hardware_init_delay=5.0,  # seconds to wait before showing GUI
    )

orchestrator.launch() creates the queues, starts the backend subprocess, waits hardware_init_delay seconds for hardware to initialise, then calls gui_runner(control_queue, utility_queue) in the main process.

4. Binding GUI Controls to Strategy Parameters#

Sliders, spinboxes, and other widgets can directly update ControlMessage fields. The auto-send callback ensures the backend receives the update immediately.

Example: Amplitude Slider#

# gui.py (continued)
from PyQt5.QtWidgets import QSlider, QLabel, QHBoxLayout
from PyQt5.QtCore import Qt
from dataclasses import dataclass, field
from fes_framework.data.types import ControlMessage

@dataclass
class AppControlMessage(ControlMessage):
    quad_amplitude: int = 40     # initial amplitude %
    knee_threshold: float = 15.0

class MainWindow(QMainWindow):
    def __init__(self, control_queue, utility_queue):
        super().__init__()
        from fes_framework.ipc.queue_handler import QueueHandler
        self.handler = QueueHandler(
            control_queue=control_queue,
            utility_queue=utility_queue,
            control_message=AppControlMessage(),
        )
        self._build_ui()

    def _build_ui(self):
        central = QWidget()
        self.setCentralWidget(central)
        layout = QVBoxLayout(central)

        # Amplitude slider (0–100%)
        row = QHBoxLayout()
        row.addWidget(QLabel("Quad Amplitude:"))

        self.amp_slider = QSlider(Qt.Horizontal)
        self.amp_slider.setRange(0, 100)
        self.amp_slider.setValue(40)
        self.amp_label = QLabel("40 %")

        self.amp_slider.valueChanged.connect(self._on_amplitude_changed)
        row.addWidget(self.amp_slider)
        row.addWidget(self.amp_label)
        layout.addLayout(row)

        # FES Active
        self.fes_checkbox = QCheckBox("FES Active")
        self.fes_checkbox.stateChanged.connect(
            lambda s: setattr(self.handler.utility_message, 'FesIsActive', bool(s))
        )
        layout.addWidget(self.fes_checkbox)

    def _on_amplitude_changed(self, value: int):
        self.amp_label.setText(f"{value} %")
        self.handler.control_message.quad_amplitude = value  # auto-sends to backend

5. Real-Time Data Visualization#

SharedRingBuffer Setup#

SharedRingBuffer uses Python multiprocessing.shared_memory for zero-copy data sharing. The backend writes a frame every cycle; the GUI reads however many frames it needs.

Tip: The framework provides a standard dtype shared_memory_frame in fes_framework.data.types that covers all 29 joint angles, step detector flags, and backend_sample_rate. Import it instead of defining your own unless you need custom fields (e.g. EMS columns, PPG data).

In the backend process (engine subclass):

import numpy as np
from fes_framework.ipc.buffer import SharedRingBuffer
from fes_framework.data.types import shared_memory_frame
from fes_framework.engine import ClosedLoopEngine

# Use the standard frame dtype — or define your own if you need extra fields
FRAME_DTYPE = shared_memory_frame

class MyEngine(ClosedLoopEngine):
    def on_start(self) -> None:
        self._buffer = SharedRingBuffer(
            dtype=FRAME_DTYPE,
            capacity=1000,     # keep 10 seconds at 100 Hz
            create=True,
            name="fes_data",
        )

    def on_cycle_complete(self) -> None:
        frame = np.zeros(1, dtype=FRAME_DTYPE)
        frame['knee_r'] = self.data_streamer.biomechanical_data.KneeFlexExtR
        frame['knee_l'] = self.data_streamer.biomechanical_data.KneeFlexExtL
        frame['contact_r'] = float(self.data_streamer.step_detector_data.right_foot_contact)
        frame['contact_l'] = float(self.data_streamer.step_detector_data.left_foot_contact)
        self._buffer.write_frame(frame[0])

In the GUI process (MainWindow):

import numpy as np
from fes_framework.ipc.buffer import SharedRingBuffer
from PyQt5.QtCore import QTimer

from fes_framework.data.types import shared_memory_frame

FRAME_DTYPE = shared_memory_frame  # same dtype as backend

class MainWindow(QMainWindow):
    def __init__(self, control_queue, utility_queue):
        super().__init__()
        # ... (QueueHandler and UI setup as before)

        # Attach to the buffer created by the backend
        self._buffer = SharedRingBuffer(
            dtype=FRAME_DTYPE,
            name="fes_data",
            create=False,   # attach — backend already created it
        )

        # 30 Hz update timer
        self._update_timer = QTimer()
        self._update_timer.timeout.connect(self._update_plots)
        self._update_timer.start(33)  # 33 ms ≈ 30 Hz

    def _update_plots(self):
        # Read last 300 frames (3 seconds at 100 Hz)
        frames = self._buffer.read_frames(300)
        if len(frames) == 0:
            return

        knee_r = frames['knee_r']
        knee_l = frames['knee_l']
        # Update your pyqtgraph PlotWidget here:
        self.knee_curve_r.setData(knee_r)
        self.knee_curve_l.setData(knee_l)

Important: The buffer name ("fes_data") must match exactly between backend and GUI. The backend must create it (create=True) before the GUI attaches (create=False). The hardware_init_delay in launch() gives the backend time to create the buffer.

Adding pyqtgraph Plots#

# pip install pyqtgraph
import pyqtgraph as pg

class MainWindow(QMainWindow):
    def _build_ui(self):
        layout = QVBoxLayout(central)

        # Create plot widget
        self.plot_widget = pg.PlotWidget(title="Knee Angles")
        self.plot_widget.setLabel('left', 'Angle', units='deg')
        self.plot_widget.setLabel('bottom', 'Frame')
        self.plot_widget.addLegend()

        self.knee_curve_r = self.plot_widget.plot(pen='b', name='Right Knee')
        self.knee_curve_l = self.plot_widget.plot(pen='r', name='Left Knee')

        layout.addWidget(self.plot_widget)

6. FES Active Toggle and Safety#

Emergency Stop#

The FesIsActive flag in UtilityMessage is a global EMS kill switch. When False, the engine's _apply_ems_output function mutes all muscles regardless of what process() writes. Make the emergency stop prominent in your GUI:

# Large, red emergency stop button
stop_btn = QPushButton("STOP FES")
stop_btn.setStyleSheet("background-color: red; color: white; font-size: 16pt; font-weight: bold;")
stop_btn.setMinimumHeight(60)
stop_btn.clicked.connect(self._emergency_stop)

def _emergency_stop(self):
    self.handler.utility_message.FesIsActive = False
    self.fes_checkbox.setChecked(False)  # update checkbox state

Graceful Shutdown#

When the main window closes, the orchestrator terminates the backend process. Override closeEvent to ensure any cleanup happens:

def closeEvent(self, event):
    # Signal backend to stop cleanly
    self.handler.utility_message.FesIsActive = False
    # Let Qt handle the rest — orchestrator catches backend.join()
    event.accept()

Complete Minimal GUI Application#

# minimal_gui_app.py
"""
Complete minimal GUI application:
  - FES Active checkbox
  - Amplitude slider for right quadriceps
  - Real-time knee angle plot
  - Emergency stop button
"""
import sys
import numpy as np
from dataclasses import dataclass

from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout,
                              QHBoxLayout, QSlider, QCheckBox, QPushButton, QLabel)
from PyQt5.QtCore import Qt, QTimer
import pyqtgraph as pg

from fes_framework.control.strategy_base import ControlStrategyBase
from fes_framework.data.types import ControlMessage, EMSParamData
from fes_framework.engine import ClosedLoopEngine
from fes_framework.ipc.buffer import SharedRingBuffer
from fes_framework.ipc.queue_handler import QueueHandler
from fes_framework.orchestrator import launch


# ── Data types ────────────────────────────────────────────────────────────────
# Minimal custom dtype for this example.  For a full set of fields use:
#   from fes_framework.data.types import shared_memory_frame

FRAME_DTYPE = np.dtype([
    ('knee_r', np.float32),
    ('contact_r', np.float32),
])

@dataclass
class AppMessage(ControlMessage):
    quad_amplitude: int = 40


# ── Strategy ──────────────────────────────────────────────────────────────────

class StanceStrategy(ControlStrategyBase):
    def process(self) -> None:
        params: AppMessage = self.params
        if self.contacts.right_foot_contact:
            self.ems_output.quadriceps_right = EMSParamData(
                IsMuted=False,
                Amplitude=params.quad_amplitude,
                PulseWidth=120, Period=20.0
            )
        else:
            self.ems_output.quadriceps_right = EMSParamData(IsMuted=True)


# ── Backend engine with ring buffer ───────────────────────────────────────────

class AppEngine(ClosedLoopEngine):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.control_message = AppMessage()

    def on_start(self) -> None:
        self._buf = SharedRingBuffer(dtype=FRAME_DTYPE, capacity=500, create=True, name="app_data")

    def on_cycle_complete(self) -> None:
        f = np.zeros(1, dtype=FRAME_DTYPE)
        f['knee_r'] = self.data_streamer.biomechanical_data.KneeFlexExtR
        f['contact_r'] = float(self.data_streamer.step_detector_data.right_foot_contact)
        self._buf.write_frame(f[0])


# ── GUI ───────────────────────────────────────────────────────────────────────

class MainWindow(QMainWindow):
    def __init__(self, control_queue, utility_queue):
        super().__init__()
        self.setWindowTitle("FES Application")
        self.handler = QueueHandler(
            control_queue=control_queue,
            utility_queue=utility_queue,
            control_message=AppMessage(),
        )
        self._buf = SharedRingBuffer(dtype=FRAME_DTYPE, name="app_data", create=False)
        self._build_ui()
        self._timer = QTimer()
        self._timer.timeout.connect(self._update)
        self._timer.start(33)

    def _build_ui(self):
        central = QWidget()
        self.setCentralWidget(central)
        layout = QVBoxLayout(central)

        # Emergency stop
        stop_btn = QPushButton("■ STOP FES")
        stop_btn.setStyleSheet("background-color: #cc0000; color: white; font-size: 14pt;")
        stop_btn.setMinimumHeight(50)
        stop_btn.clicked.connect(lambda: setattr(self.handler.utility_message, 'FesIsActive', False))
        layout.addWidget(stop_btn)

        # FES Active
        self.fes_cb = QCheckBox("FES Active")
        self.fes_cb.stateChanged.connect(
            lambda s: setattr(self.handler.utility_message, 'FesIsActive', bool(s))
        )
        layout.addWidget(self.fes_cb)

        # Amplitude slider
        row = QHBoxLayout()
        row.addWidget(QLabel("Quad R Amplitude:"))
        self.amp_slider = QSlider(Qt.Horizontal)
        self.amp_slider.setRange(0, 100)
        self.amp_slider.setValue(40)
        self.amp_label = QLabel("40 %")
        self.amp_slider.valueChanged.connect(self._on_amp)
        row.addWidget(self.amp_slider)
        row.addWidget(self.amp_label)
        layout.addLayout(row)

        # Plot
        self.plot = pg.PlotWidget(title="Right Knee Angle")
        self.plot.setLabel('left', 'Angle', units='deg')
        self.knee_curve = self.plot.plot(pen='b')
        layout.addWidget(self.plot)

    def _on_amp(self, v):
        self.amp_label.setText(f"{v} %")
        self.handler.control_message.quad_amplitude = v

    def _update(self):
        frames = self._buf.read_frames(150)
        if len(frames):
            self.knee_curve.setData(frames['knee_r'])


def gui_runner(control_queue, utility_queue):
    app = QApplication(sys.argv)
    win = MainWindow(control_queue, utility_queue)
    win.show()
    app.exec_()


# ── Entry point ───────────────────────────────────────────────────────────────

if __name__ == "__main__":
    launch(
        StanceStrategy,
        gui_runner=gui_runner,
        hardware_init_delay=5.0,
    )

The framework ships a reusable widget layer in fes_framework.gui that eliminates most boilerplate. Instead of building a PyQt5 window from scratch (sections 3–6 above), you can assemble complete applications from pre-built components.

FesApp — Application Shell#

FesApp is a QMainWindow subclass with a 30 Hz timer, tab container, and automatic QueueHandler + DataAdapter wiring:

from fes_framework.data.types import shared_memory_frame
from fes_framework.gui.app import FesApp
from fes_framework.gui.data_adapter import DataAdapter
from fes_framework.gui.tabs import OverviewTab, SensorDataTab

def gui_main(control_queue, utility_queue):
    adapter = DataAdapter(
        buffer_name="fes_shared_buffer",
        frame_dtype=shared_memory_frame,
        fields=["KneeFlexExtR", "KneeFlexExtL"],
    )

    app = FesApp(
        control_queue=control_queue,
        utility_queue=utility_queue,
        data_adapter=adapter,
        title="My FES Application",
        tabs=[
            ("Overview", OverviewTab),
            ("Sensor Data", SensorDataTab),
        ],
    )
    app.run()

FesApp refreshes only the visible tab at 30 Hz (lazy refresh). Tabs that set always_update = True refresh even when hidden (useful for the Overview tab that controls FES state).

DataAdapter — Shared Memory Wrapper#

DataAdapter wraps SharedRingBuffer with auto-reconnect and per-field rolling deque buffers. Widgets call adapter.get("KneeFlexExtR") to get numpy arrays for plotting — they never touch shared memory directly.

from fes_framework.data.types import shared_memory_frame

adapter = DataAdapter(
    buffer_name="my_buffer",
    frame_dtype=shared_memory_frame,
    fields=["KneeFlexExtR", "StepDetectorLeft"],
    max_points=300,    # rolling window (~3 s at 100 Hz)
    capacity=1000,     # must match backend SharedRingBuffer
)

Pre-Built Components#

The fes_framework.gui.components package provides strategy-agnostic widgets:

Component	Purpose
`FesToggle`	FES enable/disable toggle — auto-sends `FesIsActive` via `QueueHandler`
`CalibrationPanel`	Calibrate button + status indicator — auto-sends `CalibrationLoopIsActive` via `QueueHandler`
`LivePlot`	Scrolling pyqtgraph plot with auto-range and optional binary overlay shading
`MuscleControlCard`	Per-muscle parameter card (enable, frequency, amplitude, pulse width)
`MuscleActivityPlot`	Timeline showing per-muscle active/inactive state over time
`RangeSlider`	Double-handle slider for selecting a range (e.g. stance phase 20%–80%)
`ParameterRow`	Labelled numeric spinbox with `valueChanged(name, value)` signal
`StatusIndicator`	Colour-coded dot badge: ok (green), warning (yellow), error (red), off (grey)

Pre-Built Tabs#

Tab	Purpose
`OverviewTab`	FES toggle, calibration panel, connection status, dynamic muscle cards (left/right split)
`SensorDataTab`	Bilateral sensor data live plots with optional binary overlay shading

Both accept a data_adapter and queue_handler as keyword arguments. Pass extra configuration with functools.partial:

from functools import partial
from fes_framework.gui.tabs import OverviewTab

MUSCLES = ["quadriceps_left", "quadriceps_right", "hamstring_left", "hamstring_right"]
overview_factory = partial(OverviewTab, muscles=MUSCLES)

Copy any snippet below into a FesWidget.build_ui() (or any QWidget.__init__()). All widgets that send IPC messages accept queue_handler=self.qh.

FesToggle — FES on/off button#

from fes_framework.gui.components import FesToggle

# Wires itself to the backend — no extra code needed.
self._fes = FesToggle(queue_handler=self.qh)
self.layout().addWidget(self._fes)

# Optional: react to state changes
self._fes.toggled.connect(lambda active: print("FES active:", active))

# Optional: read or set state programmatically
is_on = self._fes.is_active
self._fes.set_active(True)

CalibrationPanel — calibrate button + status badge#

from fes_framework.gui.components import CalibrationPanel

# Wires itself to the backend — no extra code needed.
self._cal = CalibrationPanel(queue_handler=self.qh)
self.layout().addWidget(self._cal)

# Update badge when calibration result arrives (e.g. from utility message):
self._cal.set_calibrated(True)   # shows green "Calibrated"
self._cal.set_calibrated(False)  # shows grey "Not calibrated"

StatusIndicator — colour-coded dot badge#

from fes_framework.gui.components import StatusIndicator

self._status = StatusIndicator(label="Sensor")
self.layout().addWidget(self._status)

# Update from refresh():
self._status.set_status("ok")       # green
self._status.set_status("warning")  # yellow
self._status.set_status("error")    # red
self._status.set_status("off")      # grey
self._status.set_label("Connected")

ParameterRow — labelled spinbox#

from fes_framework.gui.components import ParameterRow

row = ParameterRow(
    name="threshold",       # identifier in valueChanged signal
    label="Knee Threshold",
    minimum=0.0,
    maximum=90.0,
    default=15.0,
    suffix="°",
)
self.layout().addWidget(row)

# Send value changes to backend:
row.valueChanged.connect(
    lambda name, val: setattr(self.qh.control_message, name, val)
)

RangeSlider — dual-handle range picker#

from fes_framework.gui.components import RangeSlider

slider = RangeSlider(minimum=0, maximum=100)
slider.setRange(20, 80)  # initial low / high
self.layout().addWidget(slider)

slider.valueChanged.connect(
    lambda name, val: print(f"Range param {name}: {val}")
)

LivePlot — scrolling sensor data plot#

from fes_framework.gui.components import LivePlot

plot = LivePlot(
    title="Knee Angles",
    ylabel="Angle",
    unit="deg",
    series=["KneeFlexExtR", "KneeFlexExtL"],  # field names from DataAdapter
    overlay_field="StepDetectorRight",          # optional: 0/1 background shading
    overlay_label="Stance",
)
self.layout().addWidget(plot)

# Call from refresh():
if self.data and self.data.has_data:
    t = self.data.get_time()
    plot.update_data(
        t,
        {"KneeFlexExtR": self.data.get("KneeFlexExtR"),
         "KneeFlexExtL": self.data.get("KneeFlexExtL")},
        overlay=self.data.get("StepDetectorRight"),
    )

MuscleControlCard — per-muscle parameter card#

from fes_framework.gui.components import MuscleControlCard

card = MuscleControlCard("quadriceps_left", display_name="Quadriceps L")
self.layout().addWidget(card)

# Send param changes to backend:
card.parametersChanged.connect(
    lambda muscle, params: setattr(
        self.qh.control_message, "stim_params", {muscle: params}
    )
)

# Read current values:
params = card.get_params()
# → {"is_active": bool, "frequency": float, "amplitude": float, "pulse_width": float}

MuscleActivityPlot — active/inactive timeline#

import numpy as np
from fes_framework.gui.components import MuscleActivityPlot

plot = MuscleActivityPlot(
    muscles=["quadriceps_left", "quadriceps_right"],
)
self.layout().addWidget(plot)

# Call from refresh(): drive activity from EMS command params.
# (This is commanded activity, not measured muscle response.)
if self.data and self.data.has_data and self.qh:
    t = self.data.get_time()
    if t is None or len(t) == 0:
        return

    stim_params = getattr(self.qh.control_message, "stim_params", {}) or {}
    fes_on = bool(getattr(self.qh.utility_message, "FesIsActive", True))

    def _is_active(muscle_name: str) -> float:
        params = stim_params.get(muscle_name, {})
        return 1.0 if (fes_on and bool(params.get("is_active", False))) else 0.0

    plot.update_data(t, {
        "quadriceps_left": np.full(len(t), _is_active("quadriceps_left")),
        "quadriceps_right": np.full(len(t), _is_active("quadriceps_right")),
    })

Custom Tabs — `FesWidget` Base Class#

To create your own tab, subclass FesWidget. Override build_ui() to set up widgets (called once) and refresh() to update them at ~30 Hz:

from fes_framework.gui import FesWidget
from fes_framework.gui.components import FesToggle, CalibrationPanel, StatusIndicator

class MyControlTab(FesWidget):
    always_update = True   # refresh even when hidden (needed for control widgets)

    def build_ui(self) -> None:
        # self.qh  → QueueHandler (IPC)
        # self.data → DataAdapter (sensor data)

        self._toggle = FesToggle(queue_handler=self.qh)
        self.layout().addWidget(self._toggle)

        self._cal = CalibrationPanel(queue_handler=self.qh)
        self.layout().addWidget(self._cal)

        self._conn = StatusIndicator(label="Disconnected")
        self.layout().addWidget(self._conn)

        self.layout().addStretch()

    def refresh(self) -> None:
        if self.data and self.data.is_connected:
            self._conn.set_status("ok")
            self._conn.set_label("Connected")
        else:
            self._conn.set_status("error")
            self._conn.set_label("Disconnected")

tabs = [("Control", MyControlTab)]

FesWidget provides:

self.data — DataAdapter instance (shared memory reader)
self.qh — QueueHandler instance (IPC sender)
self.layout() — pre-created QVBoxLayout
always_update = False — override to True if your tab must refresh when hidden

Theme#

The fes_framework.gui.theme module defines colour tokens (COLORS dict), a 6-colour PLOT_PALETTE, and a DEFAULT_STYLESHEET (TeslaSuit dark design system). Override the stylesheet via FesApp(stylesheet=my_css).

Generic Example#

See examples/generic_gui/main.py for a complete runnable example using all the built-in widgets plus a custom ConnectionTab (FesWidget subclass) — no application-specific code, just DummyStrategy, 8 muscles, and 4 sensor data groups.

Next#

Step 9 — Full application. All 9 steps brought together.

Step 8 — Adding a GUI

1. Dual-Process Architecture#

2. IPC Messages#

ControlMessage (GUI → Backend)#

UtilityMessage (Bidirectional)#

QueueHandler#

3. Minimal PyQt5 GUI Skeleton#

Launching with the GUI#

4. Binding GUI Controls to Strategy Parameters#

Example: Amplitude Slider#

5. Real-Time Data Visualization#

SharedRingBuffer Setup#

Adding pyqtgraph Plots#

6. FES Active Toggle and Safety#

Emergency Stop#

Graceful Shutdown#

Complete Minimal GUI Application#

7. Using the Built-In Widget Toolkit#

FesApp — Application Shell#

DataAdapter — Shared Memory Wrapper#

Pre-Built Components#

Pre-Built Tabs#

Widget Quick-Start Templates#

FesToggle — FES on/off button#

CalibrationPanel — calibrate button + status badge#

StatusIndicator — colour-coded dot badge#

ParameterRow — labelled spinbox#

RangeSlider — dual-handle range picker#

LivePlot — scrolling sensor data plot#

MuscleControlCard — per-muscle parameter card#

MuscleActivityPlot — active/inactive timeline#

Custom Tabs — `FesWidget` Base Class#

Theme#

Generic Example#

Next#

See also#

Step 8 — Adding a GUI

1. Dual-Process Architecture#

2. IPC Messages#

ControlMessage (GUI → Backend)#

UtilityMessage (Bidirectional)#

QueueHandler#

3. Minimal PyQt5 GUI Skeleton#

Launching with the GUI#

4. Binding GUI Controls to Strategy Parameters#

Example: Amplitude Slider#

5. Real-Time Data Visualization#

SharedRingBuffer Setup#

Adding pyqtgraph Plots#

6. FES Active Toggle and Safety#

Emergency Stop#

Graceful Shutdown#

Complete Minimal GUI Application#

7. Using the Built-In Widget Toolkit#

FesApp — Application Shell#

DataAdapter — Shared Memory Wrapper#

Pre-Built Components#

Pre-Built Tabs#

Widget Quick-Start Templates#

FesToggle — FES on/off button#

CalibrationPanel — calibrate button + status badge#

StatusIndicator — colour-coded dot badge#

ParameterRow — labelled spinbox#

RangeSlider — dual-handle range picker#

LivePlot — scrolling sensor data plot#

MuscleControlCard — per-muscle parameter card#

MuscleActivityPlot — active/inactive timeline#

Custom Tabs — FesWidget Base Class#

Theme#

Generic Example#

Next#

See also#

Custom Tabs — `FesWidget` Base Class#