ControlStrategy

Source: fes_framework/control/strategy_base.py API reference: api_reference.md → ControlStrategyBase

What it is#

ControlStrategyBase is an abstract base class. Subclassing it and implementing the abstract process() method gives you a complete control strategy that the framework will run at ~100 Hz. There are no other required methods.

Why it exists#

The whole point of the framework is that the only thing varying between applications is the algorithm. The framework makes that explicit by encapsulating exactly that — and nothing else — in ControlStrategyBase. Hardware connection, data parsing, channel routing, and timing all live elsewhere. Your code lives here.

The slot-based design (inputs as self.<name>, outputs as self.ems_output) is deliberate. See Design principle 2 for the reasoning.

When to use it#

You always subclass it. Every FES application has at least one ControlStrategyBase subclass. You can have more than one — e.g. a "warm-up" strategy and a "main" strategy — and switch between them by re-instantiating the engine, but most applications use a single strategy.

You override setup() when you need one-time initialization (load a config, build a haptic library, allocate state). Always call super().setup(muscles, suit) first.

You don't override run_strategy() — that's the framework's orchestration method that calls your process(). The framework documents this explicitly and the source carries a "do NOT override" comment.

Inputs available in process()#

self.params is None until the first ControlMessage arrives — guard with if self.params is not None: if you read from it before the GUI has sent anything.

self.muscles and self.suit are wired automatically by the engine in the call to strategy.setup(muscles=..., suit=...).

Outputs#

EMSParamData has four fields:

• IsMuted (bool) — True to silence this muscle this cycle
• Amplitude (int, 0–100) — stimulation intensity, percent
• PulseWidth (int, μs) — pulse duration, typically 10–140
• Period (float, ms) — time between pulses; 1000 / Period is the pulse frequency in Hz (e.g. Period=20.0 → 50 Hz)

Optional outputs:

• self.haptic_library (a HapticLibrary subclass) — toggle slot.IsMuted to fire/silence custom haptic playables. Set up in setup() via the self.suit.create_haptic_touch() / self.suit.load_haptic_asset() factories.

Minimal example#

from fes_framework.control.strategy_base import ControlStrategyBase
from fes_framework.data.types import EMSParamData

class StanceQuadStrategy(ControlStrategyBase):
    """Stimulate right quadriceps during right stance phase."""

    def process(self) -> None:
        if self.contacts.right_foot_contact:
            self.ems_output.quadriceps_right = EMSParamData(
                IsMuted=False, Amplitude=40, PulseWidth=120, Period=20.0,
            )
        else:
            self.ems_output.quadriceps_right = EMSParamData(IsMuted=True)

That's a complete, runnable strategy. Pair it with orchestrator.launch(StanceQuadStrategy) and you have an application.

Realistic example with setup() and a runtime parameter#

from dataclasses import dataclass
from fes_framework.control.strategy_base import ControlStrategyBase
from fes_framework.data.types import ControlMessage, EMSParamData

@dataclass
class MyControlMessage(ControlMessage):
    threshold_deg: float = 15.0
    quad_amplitude: int = 40

class KneeAngleStrategy(ControlStrategyBase):

    def setup(self, muscles=None, suit=None, config=None) -> None:
        super().setup(muscles, suit, config)
        # one-time init goes here
        self._last_active = False

    def process(self) -> None:
        threshold = self.params.threshold_deg if self.params else 15.0
        amp = self.params.quad_amplitude if self.params else 40

        active = self.joints.KneeFlexExtR > threshold
        if active and not self._last_active:
            print(f"Activating at knee={self.joints.KneeFlexExtR:.1f}°")
        self._last_active = active

        self.ems_output.quadriceps_right = EMSParamData(
            IsMuted=not active, Amplitude=amp, PulseWidth=120, Period=20.0,
        )

Things you should and should not do in process()#

The engine runs at ~100 Hz. A sloppy process() will degrade loop timing or break it entirely. Heavy work belongs in on_cycle_complete() (engine subclass), gated by cycle_count % N, or in a worker thread.

Common patterns#

Always-on stimulation#

def process(self) -> None:
    self.ems_output.quadriceps_left = EMSParamData(
        IsMuted=False, Amplitude=30, PulseWidth=120, Period=20.0,
    )

Threshold-triggered#

def process(self) -> None:
    active = self.joints.KneeFlexExtR > 15.0
    self.ems_output.quadriceps_right = EMSParamData(
        IsMuted=not active, Amplitude=40, PulseWidth=120, Period=20.0,
    )

Bilateral by side query#

def process(self) -> None:
    for muscle in self.muscles.by_side("left"):
        setattr(self.ems_output, muscle.name, EMSParamData(
            IsMuted=False, Amplitude=30, PulseWidth=120, Period=20.0,
        ))

External data fusion#

def process(self) -> None:
    grf = self.external_data.get("ForcePlate_GRF")
    vertical_force = grf.data[2] if grf else 0.0
    if vertical_force > 100:
        self.ems_output.gastrocnemius_right = EMSParamData(
            IsMuted=False, Amplitude=50, PulseWidth=120, Period=20.0,
        )

See also#

• API Reference → ControlStrategyBase
• Data Types — what flows through self.<attr>
• MuscleMap — what self.muscles exposes
• Messaging — ControlMessage for self.params, HapticLibrary for custom playables
• Implementation Guide → Step 1 — first strategy
• Implementation Guide → Step 3 — semantic muscle output
• Source: fes_framework/control/strategy_base.py
• Example: examples/atomic/minimal_closed_loop.py