Messaging

Source: fes_framework/data/types.py

What they are#

The framework draws a sharp line between three kinds of "messages" that flow into and around the backend each cycle:

The split is intentional. See Design principle 6 for the rationale.

Why they exist#

Without the split, every application's "state" would be one bag of fields shared with the framework. The framework would need to know about application-specific fields (to decide which ones trigger a re-render or a kill-switch update). And applications would need to know about framework-specific fields (to avoid stomping on them).

By giving each kind its own type, the framework can enforce its flags (FES kill switch, calibration gate) without ever knowing what parameters your application defines, and your application can add parameters without the framework caring.

HapticLibrary is separate because its lifecycle is different: playables are created once in setup(), mutated in place every cycle, and torn down at shutdown. They never travel through queues.

ControlMessage#

The base class is empty. You subclass it to declare your application's parameters:

from dataclasses import dataclass
from fes_framework.data.types import ControlMessage

@dataclass
class MyControlMessage(ControlMessage):
    threshold_deg: float = 15.0
    quad_amplitude: int = 40
    quad_pulse_width: int = 120
    quad_period: float = 20.0
    operator_name: str = ""

The strategy reads it as self.params each cycle:

def process(self) -> None:
    if self.params is None:    # before first message
        return
    threshold = self.params.threshold_deg
    if self.joints.KneeFlexExtR > threshold:
        self.ems_output.quadriceps_right = EMSParamData(
            IsMuted=False,
            Amplitude=self.params.quad_amplitude,
            PulseWidth=self.params.quad_pulse_width,
            Period=self.params.quad_period,
        )

How it gets there#

The GUI process holds the same MyControlMessage instance and pushes it to control_queue. QueueHandler.poll_queues() (called by the engine each cycle) drains the queue, keeps the most recent message, and updates engine.control_message. The strategy reads self.params (= the engine's current control_message) in process().

If your MyControlMessage defines _on_change, the GUI side can auto-send by simply assigning a field — see fes_framework/data/types.py for the pattern.

LSL streaming#

The AppData_ControlMessage outlet (when LSL is enabled) publishes your subclass's fields as channels. Channel count is determined at startup from the dataclass — add or remove fields between sessions and the outlet adapts. Lab tools (LabRecorder) just see the new shape.

Tip#

Define one ControlMessage subclass per application. Define defaults that produce safe but boring behaviour (low amplitude, sane thresholds). Then the GUI's job is just to widen those defaults under operator control.

UtilityMessage#

UtilityMessage is framework-defined. You don't subclass it. Its fields are stable across applications:

The flag set reflects what the framework itself reacts to. If you need additional system flags for your application, those go in your ControlMessage subclass — not in UtilityMessage.

Bidirectional#

UtilityMessage flows both ways. The GUI sends one to toggle FES; the backend can also send one to update the GUI (e.g. the engine reflecting that calibration completed).

The engine's on_utility_message(msg) hook fires whenever a new utility message arrives. Override it in your engine subclass if you need custom reactions.

Why FES kill switch is a utility, not a control#

Because the kill switch is framework concern, not application concern. The framework has to enforce it the same way regardless of what application is running. Putting it in UtilityMessage (a stable type) means the framework can read FesIsActive without knowing anything about the application's parameters.

Step-detection mode flags#

The three step-detection mode flags are Walking FES-specific holdovers. The current RapidKit only honours TSAPIStepDetectionIsActive (the SDK's built-in detector). The other two flags are kept on the dataclass for backwards compatibility with the Walking FES GUI but are not consumed by the engine. New applications should treat them as inert.

HapticLibrary#

The base class is empty. You subclass it to declare named slots for CustomPlayable instances:

from dataclasses import dataclass, field
from fes_framework.data.types import HapticLibrary, CustomPlayable

@dataclass
class NavCues(HapticLibrary):
    cue_left:  CustomPlayable = field(default_factory=CustomPlayable)
    cue_right: CustomPlayable = field(default_factory=CustomPlayable)
    cue_stop:  CustomPlayable = field(default_factory=CustomPlayable)

You populate the slots in your strategy's setup() using factories on self.suit:

class NavStrategy(ControlStrategyBase):

    def setup(self, muscles=None, suit=None, config=None) -> None:
        super().setup(muscles, suit, config)
        self.haptic_library = NavCues()
        self.haptic_library.cue_left  = self.suit.create_haptic_touch(
            bone_id=14, channel_list=[0,1,2], period=20.0, amplitude=30, pulse_width=120,
        )
        self.haptic_library.cue_right = self.suit.create_haptic_touch(
            bone_id=15, channel_list=[0,1,2], period=20.0, amplitude=30, pulse_width=120,
        )
        self.haptic_library.cue_stop  = self.suit.load_haptic_asset(
            "assets/stop_buzz.hpt", looped=False,
        )

In process(), toggle slot.IsMuted to fire/silence:

def process(self) -> None:
    if self.joints.HipFlexExtR > 30:
        self.haptic_library.cue_right.IsMuted = False
    else:
        self.haptic_library.cue_right.IsMuted = True

The engine's LibraryStimulator runs after the main Stimulator and translates slot.IsMuted changes into SDK calls.

Why a separate stimulator?#

Because the lifecycle is different from EmsData. EmsData muscle slots are pre-defined by MuscleMap; the stimulator has full prior knowledge of them. HapticLibrary slots are user-defined and opaque until the strategy populates them. The framework wires the two stimulators in series after every process():

Stimulator.run_stimulator(...)             # EmsData muscles
if strategy.haptic_library is not None:
    LibraryStimulator.run_stimulator(...)  # custom playables

If you don't define a HapticLibrary, LibraryStimulator is a no-op.

LSL outlet#

When you populate strategy.haptic_library, the engine registers the slot layout with LSLStreamer so the haptic library appears as a dedicated outlet — channel labels are the slot field names. See LSL Streaming for outlet details.

When to use each#

See also#

• API Reference → ControlMessage / UtilityMessage / HapticLibrary
• Data Types — concept-level overview of all dataclasses
• IPC and processes — how messages move between backend and GUI
• Implementation Guide → Step 5 — ControlMessage subclass walkthrough
• Examples → Haptic navigation — HapticLibrary in production
• Source: fes_framework/data/types.py