Step 5 — Runtime parameters

What you're adding#

Up to now, every parameter (AMPLITUDE = 30, PULSE_WIDTH = 120, Period = 20.0) has been hardcoded. To tune them you'd edit the file and restart the application — fine while developing, painful during a session.

You'll subclass ControlMessage to declare your application's parameters as fields. The strategy reads them from self.params each cycle. In Step 8 you'll build a GUI that pushes new ControlMessage instances over the IPC queue; for now, you'll seed defaults from the orchestrator and demonstrate that the strategy respects them.

Code#

# app/messages.py
from dataclasses import dataclass
from fes_framework.data.types import ControlMessage


@dataclass
class MyControlMessage(ControlMessage):
    # When True, the strategy emits stimulation in stance phase.
    quad_active: bool = True

    # Stimulation parameters for the right quadriceps.
    quad_amplitude: int = 30        # 0–100 %
    quad_pulse_width: int = 120     # μs
    quad_period: float = 20.0       # ms (50 Hz)

# app/strategy.py
from fes_framework.control.strategy_base import ControlStrategyBase
from fes_framework.data.types import EMSParamData

from app.messages import MyControlMessage


class StanceQuadStrategy(ControlStrategyBase):

    def process(self) -> None:
        # The first few cycles may run before the GUI has sent any
        # ControlMessage. Guard with a default.
        params = self.params if isinstance(self.params, MyControlMessage) else MyControlMessage()

        if params.quad_active and self.contacts.right_foot_contact:
            self.ems_output.quadriceps_right = EMSParamData(
                IsMuted=False,
                Amplitude=params.quad_amplitude,
                PulseWidth=params.quad_pulse_width,
                Period=params.quad_period,
            )
        else:
            self.ems_output.quadriceps_right = EMSParamData(IsMuted=True)

# app/engine.py — extend the CalibratedEngine from Step 4
from fes_framework.engine import ClosedLoopEngine

from app.messages import MyControlMessage


class CalibratedParamEngine(ClosedLoopEngine):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # Replace the framework's default ControlMessage with our subclass.
        self.control_message = MyControlMessage()

    def on_start(self) -> None:
        self.data_streamer.set_biomech_collection(True)
        # ... calibration gate from Step 4 ...
        result = self.calibration.calibrate()
        if not result.success or not self.calibration.check_quality().is_acceptable:
            self.stop()

# app/main.py
from fes_framework.orchestrator import launch
from app.engine import CalibratedParamEngine
from app.strategy import StanceQuadStrategy

if __name__ == "__main__":
    launch(StanceQuadStrategy, engine_class=CalibratedParamEngine)

Walkthrough#

@dataclass class MyControlMessage(ControlMessage) — the base ControlMessage is empty. Subclass it; declare your parameters as fields with defaults. Use plain Python types (int, float, bool, str) — anything more complex is harder to pickle across the IPC queue.

self.params — the engine assigns its current control_message to strategy.params before each process() call. So self.params.quad_amplitude reads whatever the GUI most recently sent (or your default if nothing's been sent).

The isinstance guard. Until the GUI sends its first message, self.params is whatever the engine was constructed with. We construct a MyControlMessage() instance in CalibratedParamEngine.__init__() so the default is correct from cycle 1. The isinstance check is belt-and-braces — it costs nothing and protects against engines that were constructed without the override.

self.control_message = MyControlMessage() in __init__() — replaces the framework's default. The framework needs some ControlMessage instance from cycle 1; we give it ours so subsequent reads of self.params see the right shape.

How a GUI would update parameters (preview)#

You won't build the GUI until Step 8. For Step 5, just trust this preview: the GUI side will hold its own MyControlMessage instance and push it onto control_queue whenever a slider moves:

# Inside the GUI process (Step 8)
self.control_message = MyControlMessage()
control_queue.put(self.control_message)            # push initial defaults

def on_slider_change(self, value: int) -> None:
    self.control_message.quad_amplitude = value
    control_queue.put(self.control_message)        # push update

The backend's QueueHandler drains control_queue each cycle and keeps the latest message. By the next process(), self.params reflects the new value.

For now you can simulate this manually by writing a small script that puts a MyControlMessage onto a queue you pass to the engine — but it's easier to just wait until Step 8 and use a real GUI.

LSL streaming and ControlMessage#

When LSL is enabled (Step 7), the framework publishes an AppData_ControlMessage outlet whose channels are exactly your subclass's fields. So MyControlMessage with three numeric fields becomes a 3-channel LSL outlet. LabRecorder captures every parameter change in synchrony with the rest of the data.

Verify#

Run the app with the existing MyControlMessage() defaults (quad_active=True, quad_amplitude=30). It should behave exactly like Step 3.

Modify MyControlMessage's defaults — e.g. quad_active: bool = False — and rerun. The strategy should produce no stimulation. That confirms self.params.quad_active is being read.

Common patterns#

Per-muscle parameter blocks#

For applications that target many muscles, group parameters into sub-dataclasses or per-muscle dicts:

@dataclass
class MuscleParams:
    active:      bool = False
    amplitude:   int = 30
    pulse_width: int = 120
    period:      float = 20.0

@dataclass
class WalkingControlMessage(ControlMessage):
    quad_right:        MuscleParams = field(default_factory=MuscleParams)
    quad_left:         MuscleParams = field(default_factory=MuscleParams)
    gastrocnemius_right: MuscleParams = field(default_factory=MuscleParams)
    # ...

_on_change auto-send (advanced)#

The base ControlMessage supports an _on_change callback that fires on field assignment, so the GUI side can simply do self.control_message.quad_amplitude = 35 and have the framework auto-send. See the source in fes_framework/data/types.py for the pattern. Use this when you have many parameters and don't want to remember to call queue.put() after every assignment.

What you've learned#

• ControlMessage is the application's parameter type. You always subclass it.
• self.params is how the strategy reads them. The framework keeps it up to date.
• Replace the engine's default control_message in __init__() so defaults are correct from cycle 1.

Next#

Step 6 — External inputs. You'll bring in data from external LSL devices (force plate, EEG, optical mocap) into your strategy.