API Reference

Complete class and method reference for the RapidKit package.

Package version: current (see pyproject.toml) Python requirement: ≥ 3.10

Table of Contents#

1. ClosedLoopEngine
2. ControlStrategyBase
3. orchestrator.launch()
4. CalibrationAPI
5. DataStreamer
6. SuitHandler
7. Stimulator
8. MuscleMap / MuscleInfo
9. LSLStreamer
10. LSLInlet / ExternalInputManager
11. QueueHandler
12. SharedRingBuffer
13. Data Types
14. init_utils — Factory Functions

ClosedLoopEngine#

Location: fes_framework/engine.py

Generic closed-loop FES main loop. Auto-assembles all backend components; only the control strategy is required.

Loop timing is governed by the Teslasuit SDK's blocking data-ready calls (~100 Hz). There is no software throttle — sample_rate is metadata only.

Constructor#

ClosedLoopEngine(
    control_strategy,          # ControlStrategyBase instance — REQUIRED
    *,
    suit_handler=None,         # SuitHandler override (auto-created if None)
    data_streamer=None,        # DataStreamer override (auto-created if None)
    stimulator=None,           # Stimulator override (auto-created if None)
    lsl_streamer=None,         # LSLStreamer override (auto-created if None)
    lsl_enabled: bool = False, # Enable LSL outlet streaming (GDPR default: off)
    external_inputs=None,      # ExternalInputManager | None
    queue_handler=None,        # QueueHandler override
    control_queue=None,        # multiprocessing.Queue (GUI → Backend)
    utility_queue=None,        # multiprocessing.Queue (bidirectional)
    sample_rate: float = 100.0,# Nominal rate Hz (metadata for LSL / diagnostics)
    source_id: str = "ClosedLoopEngine",  # LSL source identifier
)

Attributes#

Methods#

run() → None#

Start the main loop (blocking). Call from the backend subprocess.

The loop runs until stop() is called or Ctrl-C is received. Shutdown order:

1. on_stop() hook — developer cleanup
2. Framework cleanup — stop mocap streaming, mute EMS, close external inputs

stop() → None#

Signal the engine to stop after the current cycle completes.

Overridable Hooks#

Override these in a ClosedLoopEngine subclass:

def on_start(self) -> None:
    """Called once before the first cycle. Default: no-op."""

def on_stop(self) -> None:
    """Called once after the loop exits, before framework cleanup. Default: no-op."""

def on_utility_message(self, message: UtilityMessage) -> None:
    """Called when a new utility message arrives from IPC. Default: no-op."""

def on_control_message(self, message: ControlMessage) -> None:
    """Called when a new control message arrives from IPC. Default: no-op."""

def on_cycle_complete(self) -> None:
    """Called at the end of every cycle, after all framework steps. Default: no-op."""

ControlStrategyBase#

Location: fes_framework/control/strategy_base.py

Abstract base class for FES control strategies. Subclass and implement process().

Constructor#

ControlStrategyBase()

Input Attributes (set by framework before each process() call)#

Output Attribute (written by process())#

Methods#

process() → None (abstract)#

Execute one control cycle. Read sensor inputs, write stimulation commands.

Must override in subclass.

def process(self) -> None:
    if self.contacts.right_foot_contact:
        self.ems_output.quadriceps_right = EMSParamData(
            IsMuted=False, Amplitude=50, PulseWidth=80, Period=20000
        )

setup(muscles=None, config=None) → None#

One-time initialisation called before the first cycle. Override to add custom setup logic (load config, allocate state). Always call super().setup() when overriding.

def setup(self, muscles: MuscleMap = None, config: dict = None) -> None:
    super().setup(muscles, config)
    self.my_state = {}

run_strategy(...) → None#

Orchestrated by the engine. Do not override — implement process() instead.

def run_strategy(
    self,
    control_message: ControlMessage,
    biomechanical_data: BiomechanicalData,
    step_detector_data: StepDetectorData,
    ems_data: EmsData,
    fes_active: bool = True,
) -> None: ...

orchestrator.launch()#

Location: fes_framework/orchestrator.py

Launch a dual-process FES application. Creates IPC queues, starts the backend engine in a subprocess, and optionally runs a GUI in the main process.

def launch(
    strategy_class: Type[ControlStrategyBase],
    *,
    gui_runner=None,
    lsl_enabled: bool = False,
    hardware_init_delay: float = 5.0,
) -> None

Usage:

from fes_framework.orchestrator import launch
from my_strategy import MyStrategy

# Headless
launch(MyStrategy, lsl_enabled=True)

# With GUI
launch(MyStrategy, gui_runner=my_gui_main, hardware_init_delay=5.0)

run_engine_process()#

def run_engine_process(
    strategy_class: Type[ControlStrategyBase],
    control_queue: Optional[Queue] = None,
    utility_queue: Optional[Queue] = None,
    *,
    lsl_enabled: bool = False,
) -> None

Backend subprocess target. Instantiates ClosedLoopEngine and calls run().

CalibrationAPI#

Location: fes_framework/calibration.py

Motion capture calibration trigger, quality assessment, and export.

Constructor#

CalibrationAPI(suit_handler: SuitHandler)

Accessible as engine.calibration after engine construction.

Attributes#

Methods#

calibrate() → CalibrationResult#

Trigger skeleton calibration (blocking, 1–2 seconds). User must be in I-pose: upright, arms at sides, feet shoulder-width apart, stationary.

@dataclass
class CalibrationResult:
    success: bool
    timestamp: float
    message: str = ""   # human-readable status

export(path: str, filename_prefix: str = "calibration_") → None#

Save the calibration reference frame to a CSV file.

engine.calibration.export(
    path="data/sessions/",
    filename_prefix="mocap_calibration_"
)
# Creates: data/sessions/mocap_calibration_<timestamp>.csv

DataStreamer#

Location: fes_framework/io/data_streamer.py

Three-phase sensor acquisition cycle: collect → process → distribute.

Constructor#

DataStreamer(suit_handler: SuitHandler)

Attributes (populated after each run_cycle() call)#

Methods#

run_cycle() → None#

Execute one complete collect → process → distribute cycle. Called by the engine each iteration.

process() → None (overridable)#

Phase 2 hook. Override to add custom signal processing, filtering, or step detection:

class MyDataStreamer(DataStreamer):
    def process(self) -> None:
        # Low-pass filter knee angle
        self.biomechanical_data.KneeFlexExtR = my_filter(
            self.biomechanical_data.KneeFlexExtR
        )

set_biomech_collection(active: bool) → None#

Toggle the per-cycle SDK inverse-kinematics call that refreshes biomechanical_data. On by default so the strategy's self.joints.* works out of the box. Pass False to skip the call when your app never reads joint angles — it's the single most expensive call in the per- cycle pipeline, so disabling it claws back measurable per-cycle CPU.

# Performance opt-out for a haptic-only app:
engine.data_streamer.set_biomech_collection(False)

Wires naturally to UtilityMessage.BiomechanicalDataCollectionIsActive when the toggle should be controllable from a GUI (see examples/walking_fes/backend_mainloop.py).

calibrate_and_write(path, filename="mocap_calibration_data_") → None#

Trigger calibration and write reference frame to CSV. Prefer CalibrationAPI.export() at the engine level.

SuitHandler#

Location: fes_framework/io/suit_handler.py

Thin hardware interface layer for the Teslasuit device.

Constructor#

SuitHandler(muscle_map_config: str = None)

muscle_map_config defaults to fes_framework/config/muscle_map_4R.json.

Attributes#

Key Methods#

Stimulator#

Location: fes_framework/io/stimulator.py

Maps semantic muscle names to hardware channels and drives EMS output.

Constructor#

Stimulator()

Methods#

run_stimulator(suit: SuitHandler, ems_data: EmsData) → None#

Run one full stimulation cycle: build haptic playables for unmuted muscles, then play/stop them. Called by the engine each cycle.

MuscleMap / MuscleInfo#

Location: fes_framework/muscle_map.py

Semantic muscle-to-hardware-channel mapping.

MuscleInfo#

@dataclass
class MuscleInfo:
    name: str                  # e.g. "quadriceps_left"
    channel_ids: list[int]     # resolved SDK channel IDs
    side: str                  # "left" or "right"
    body_region: str           # "upper_leg", "lower_leg", "hip", etc.
    default_period_us: int
    default_amplitude_pct: int
    default_pulse_width_us: int

MuscleMap#

Methods#

Available muscles (muscle_map_4R.json)#

Lower body (10): quadriceps_left, quadriceps_right, hamstring_left, hamstring_right, gastrocnemius_left, gastrocnemius_right, tibialis_anterior_left, tibialis_anterior_right, gluteus_left, gluteus_right

Upper body (10): deltoid_left, deltoid_right, biceps_left, biceps_right, triceps_left, triceps_right, wrist_flexors_left, wrist_flexors_right, wrist_extensors_left, wrist_extensors_right

LSLStreamer#

Location: fes_framework/io/lsl_streamer.py

Lab Streaming Layer outlet manager. Creates 7 outlets at startup; pushes data only when enabled=True.

Constructor#

LSLStreamer(
    sample_rate: float = 100.0,
    source_id: str = "TeslaSuit_LSL",
    enabled: bool = False,      # GDPR default: off
)

Attributes#

Outlets#

LSLInlet / ExternalInputManager#

Location: fes_framework/io/lsl_inlet.py

ExternalSample#

@dataclass
class ExternalSample:
    data: list               # channel values (type depends on stream format)
    timestamp: float         # pylsl.local_clock() domain
    stream_name: str
    channel_count: int
    channel_names: list[str] # from LSL stream metadata (may be empty)

LSLInlet#

LSLInlet(
    stream_name: str,
    stream_type: Optional[str] = None,
    timeout: float = 5.0,
)

pull_latest() → Optional[ExternalSample]#

Non-blocking pull of the most recent sample. Drains buffer and keeps only the latest. Returns None if no data ever received. Never blocks the 100 Hz loop.

ExternalInputAdapter#

class ExternalInputAdapter(ABC):
    @property
    @abstractmethod
    def stream_name(self) -> str: ...

    @abstractmethod
    def pull_latest(self) -> Optional[ExternalSample]: ...

    @abstractmethod
    def close(self): ...

Implement for non-LSL devices (serial port, USB HID, etc.).

ExternalInputManager#

class ExternalInputManager:
    MAX_INPUTS = 2   # support at least 2 external sources

QueueHandler#

Location: fes_framework/ipc/queue_handler.py

Bidirectional IPC message handler with auto-send on field assignment.

Constructor#

QueueHandler(
    control_queue=None,
    utility_queue=None,
    control_message=None,   # optional ControlMessage subclass instance
    utility_message=None,
)

Methods#

SharedRingBuffer#

Location: fes_framework/ipc/buffer.py

Circular buffer in shared memory for zero-copy inter-process data streaming.

Constructor#

SharedRingBuffer(
    dtype,              # numpy dtype describing one frame
    capacity=500,
    create=False,       # True = create; False = attach to existing
    name=None,          # shared memory name (must match between processes)
)

Methods#

total property#

Total frames written since creation (includes wraparounds).

Data Types#

Location: fes_framework/data/types.py

UtilityMessage#

@dataclass
class UtilityMessage:
    FesIsActive: bool
    RecordingIsActive: bool
    CalibrationLoopIsActive: bool
    FolderPath: str
    TSAPIStepDetectionIsActive: bool
    VUStepDetectionIsActive: bool
    ModelBasedStepDetectionIsActive: bool

Auto-sends to queue on any field assignment.

ControlMessage#

@dataclass
class ControlMessage:
    """Empty base — subclass to define application parameters."""

Auto-sends to queue on any field assignment.

EMSParamData#

@dataclass
class EMSParamData:
    IsMuted: bool = False
    PulseWidth: int = 0      # μs (10–140)
    Period: float = 0.0      # μs
    Amplitude: int = 0       # % (0–100)

EmsData#

20 muscle fields, all EMSParamData:

Lower body (10): quadriceps_left, quadriceps_right, hamstring_left, hamstring_right, gastrocnemius_left, gastrocnemius_right, tibialis_anterior_left, tibialis_anterior_right, gluteus_left, gluteus_right

Upper body (10): deltoid_left, deltoid_right, biceps_left, biceps_right, triceps_left, triceps_right, wrist_flexors_left, wrist_flexors_right, wrist_extensors_left, wrist_extensors_right

StepDetectorData#

@dataclass
class StepDetectorData:
    left_foot_contact: bool = True   # True = foot on ground (stance phase)
    right_foot_contact: bool = True

BiomechanicalData#

29 float fields (degrees). Suffix R = right, L = left:

Pelvis: PelvisTilt, PelvisList, PelvisRotation

Legs (×2): HipFlexExt, HipAddAbd, HipRot, KneeFlexExt, AnkleFlexExt, AnkleProSup

Arms (×2): ElbowFlexExt, ForearmProSup, WristFlexExt, WristDeviation

Shoulders (×2): ShoulderAddAbd, ShoulderRot, ShoulderFlexExt

RawData / ProcessedData#

20 body segments each. RawData fields are SensorData; ProcessedData fields are MocapBoneData. See class definitions in data/types.py.

Segments: Hips, LeftUpperLeg, RightUpperLeg, LeftLowerLeg, RightLowerLeg, LeftFoot, RightFoot, Spine, Chest, UpperChest, Neck, Head, LeftShoulder, RightShoulder, LeftUpperArm, RightUpperArm, LeftLowerArm, RightLowerArm, LeftHand, RightHand

_default_stim_params() → dict#

Per-muscle parameter dict factory:

{
    'is_active': False, 'frequency': 0, 'amplitude': 0, 'pulse_width': 0,
    'stance_start': 0.0, 'stance_end': 0.0, 'swing_start': 0.0, 'swing_end': 0.0,
}

shared_memory_frame#

Standard numpy dtype for SharedRingBuffer, suitable for most applications. Contains all 29 BiomechanicalData joint angles, step detector flags, and backend_sample_rate. Field names match the Teslasuit SDK conventions (PascalCase with R/L suffix).

from fes_framework.data.types import shared_memory_frame

Applications that need a custom layout (e.g. extra EMS columns or PPG data) can build their own dtype — see examples/walking_fes/walking_types.py.

init_utils — Factory Functions#

Location: fes_framework/data/init_utils.py

GUI Widget Toolkit#

Location: fes_framework/gui/

Reusable PyQt5 components for building FES application interfaces. All widgets are strategy-agnostic — they communicate with the backend through DataAdapter (read) and QueueHandler (write).

FesApp#

Location: fes_framework/gui/app.py

Generic QMainWindow shell with 30 Hz update timer.

FesApp(
    control_queue=None,
    utility_queue=None,
    *,
    data_adapter: Optional[DataAdapter] = None,
    control_message=None,
    title: str = "fes_framework",
    tabs: Optional[List[Tuple[str, Type]]] = None,
    refresh_rate: int = 30,
    stylesheet: Optional[str] = None,
)

Methods#

DataAdapter#

Location: fes_framework/gui/data_adapter.py

SharedRingBuffer wrapper with auto-reconnect and rolling per-field deques.

DataAdapter(
    buffer_name: str,
    frame_dtype,
    fields: List[str],
    max_points: int = 150,
    capacity: int = 1000,
)

LivePlot#

Location: fes_framework/gui/components/live_plot.py

Scrolling pyqtgraph plot with auto y-range and optional binary overlay shading.

LivePlot(
    title: str = "",
    y_label: str = "",
    series: Optional[List[str]] = None,
    *,
    show_background: bool = False,
    background_label: str = "Active",
    max_height: int = 250,
)

Push fresh data. series_data is {series_key: y_values}. Optional overlay_data is a 0/1 array for background fill (e.g. step phase, trigger region).

MuscleControlCard#

Location: fes_framework/gui/components/muscle_control_card.py

Per-muscle parameter card with enable checkbox and frequency / amplitude / pulse-width spinboxes. Application-specific controls (e.g. phase-timing sliders) should be added at the application layer.

MuscleControlCard(
    muscle_name: str,
    *,
    display_name: str = "",
    default_enabled: bool = True,
    default_amplitude: int = 50,
    default_pulse_width: int = 120,
    default_frequency: float = 30.0,
)

Signals#

get_params() → dict#

Returns dict with keys: is_active, frequency, amplitude, pulse_width.

RangeSlider#

Location: fes_framework/gui/components/range_slider.py

Double-handle slider for selecting a value range.

RangeSlider(minimum: int = 0, maximum: int = 100)

ParameterRow#

Location: fes_framework/gui/components/parameter_row.py

Labelled numeric spinbox.

ParameterRow(
    name: str,
    label: str = "",
    *,
    minimum: float = 0,
    maximum: float = 100,
    default: float = 0,
    suffix: str = "",
    decimals: int = 0,
)

StatusIndicator#

Location: fes_framework/gui/components/status_indicator.py

Coloured dot badge.

StatusIndicator(label: str = "")

FesToggle#

Location: fes_framework/gui/components/fes_toggle.py

Self-contained FES enable/disable toggle button. When a queue_handler is provided it automatically sends FesIsActive to the backend on every toggle — no extra signal connections required.

FesToggle(
    *,
    queue_handler: Optional[QueueHandler] = None,
    parent=None,
)

Styling: ON state renders in #50fa7b (green); OFF state in #44475a (grey). Both states use the application's default dark-theme font.

CalibrationPanel#

Location: fes_framework/gui/components/calibration_panel.py

Calibrate button + status indicator. When a queue_handler is provided it automatically sends CalibrationLoopIsActive = True to the backend on click — no extra signal connections required.

CalibrationPanel(
    *,
    queue_handler: Optional[QueueHandler] = None,
    parent=None,
)

MuscleActivityPlot#

Location: fes_framework/gui/components/muscle_activity_plot.py

Timeline plot showing per-muscle active/inactive state over time. Each muscle gets a horizontal lane — filled when active, empty when inactive.

MuscleActivityPlot(
    muscles: Optional[List[str]] = None,
    *,
    title: str = "Muscle Activity",
    max_height: int = 300,
)

Display names are auto-formatted: quadriceps_left → Quadriceps L.

FesWidget#

Location: fes_framework/gui/base_widget.py

Base class for custom GUI widgets and tabs. Subclass it to build panels that integrate with FesApp's 30 Hz refresh loop, DataAdapter, and QueueHandler.

FesWidget(
    *,
    data_adapter=None,
    queue_handler=None,
    parent=None,
)

Minimal example:

from fes_framework.gui import FesWidget
from PyQt5 import QtWidgets

class StatusTab(FesWidget):
    always_update = True

    def build_ui(self):
        self._label = QtWidgets.QLabel("Waiting…")
        self.layout().addWidget(self._label)

    def refresh(self):
        if self.data and self.data.is_connected:
            self._label.setText("Connected")
        else:
            self._label.setText("Disconnected")

OverviewTab#

Location: fes_framework/gui/tabs/overview_tab.py

FES toggle, calibration, status, and dynamic muscle control cards.

OverviewTab(
    *,
    data_adapter=None,
    queue_handler=None,
    muscles: Optional[List[str]] = None,
    show_activity_plot: bool = False,
)

always_update = True — refreshes even when hidden.

Muscles ending in _left / _right are split into two columns automatically. When show_activity_plot is enabled, the timeline reflects each muscle card's is_active state. Sends UtilityMessage.FesIsActive and ControlMessage.stim_params via QueueHandler on user interaction.

SensorDataTab#

Location: fes_framework/gui/tabs/biomechanics_tab.py

Bilateral sensor data live plots with optional binary overlay shading. A backward-compatible alias BiomechanicsTab is available in fes_framework.gui.tabs.

SensorDataTab(
    *,
    data_adapter=None,
    queue_handler=None,
    series: Optional[List[Tuple[str, str, str, str]]] = None,
    overlay_field_left: Optional[str] = None,
    overlay_field_right: Optional[str] = None,
    overlay_label: str = "Active",
)

always_update = False — only refreshes when visible (lazy).

Theme#

Location: fes_framework/gui/theme.py