Frameworks

ClosedLoopEngine

The orchestrator. It assembles every standard backend component and runs the per-cycle pipeline. Subclass it when you need lifecycle hooks; otherwise just instantiate it.

Source: fes_framework/engine.py API reference: api_reference.md → ClosedLoopEngine

What it is#

ClosedLoopEngine is the top-level object that runs an FES application's control loop. It owns and wires together:

a SuitHandler (hardware connection),
a CalibrationAPI,
a DataStreamer (input layer),
a Stimulator and LibraryStimulator (output layer),
an LSLStreamer (lab integration),
a QueueHandler (IPC, when queues are passed),
and the user's ControlStrategyBase subclass.

You give it a strategy. It gives you a running closed-loop FES application.

Why it exists#

Every FES application needs the same scaffolding: connect to the suit, spin up the data acquisition subsystems, route stimulation, optionally publish LSL streams, optionally listen for messages from a GUI, and clean up unconditionally on shutdown. Without the engine each application would re-implement that scaffolding by hand. With it the only thing that varies between applications is the strategy — which is exactly the point.

The engine also enforces the per-cycle pipeline order. Steps cannot be re-ordered: data acquisition must precede the strategy, the strategy must precede stimulation, the FES kill switch must apply after the strategy, and cleanup must happen on every exit path. By centralising this contract in one class, the framework can guarantee correctness no matter what the user writes.

When to use it#

You always use it. Every RapidKit application instantiates either ClosedLoopEngine directly or orchestrator.launch() (which creates one inside a backend subprocess).

You subclass it when you need any of:

a calibration gate at startup (override on_start()),
per-cycle visualisation (override on_cycle_complete() to write SharedRingBuffer),
custom logging or graceful shutdown (override on_stop()),
reactions to incoming GUI messages (override on_control_message / on_utility_message).

You don't subclass it for control logic — that belongs in ControlStrategyBase.process().

Minimal example#

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

class MyStrategy(ControlStrategyBase):
    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,
            )

engine = ClosedLoopEngine(control_strategy=MyStrategy())
engine.run()  # blocks until Ctrl-C or engine.stop()

In production you'd usually launch this through orchestrator.launch(MyStrategy), which runs the engine in a backend subprocess and lets the main process host a GUI or just wait.

Lifecycle hooks#

Override these in a subclass — they all default to no-ops.

Hook	Called	Typical use
`on_start()`	once, after `strategy.setup()`, before the first cycle	calibration gate; print banner; load config
`on_cycle_complete()`	end of every cycle, after LSL push	write `SharedRingBuffer`, log metrics, count cycles
`on_control_message(msg)`	when a new `ControlMessage` arrives via IPC	log, validate, react
`on_utility_message(msg)`	when a new `UtilityMessage` arrives via IPC	toggle behaviour based on flags
`on_stop()`	once, after the loop exits, before framework cleanup	close log files, save state

Lifecycle order:

__init__()                   ← components auto-assembled
↓
run()
  ├─ strategy.setup(muscles, suit)
  ├─ on_start()              ← your hook
  ├─ while running:          ← per-cycle pipeline (10 ordered steps)
  │     …
  │     on_cycle_complete()  ← your hook
  ├─ on_stop()               ← your hook (try/except — cannot block cleanup)
  └─ _cleanup()              ← framework: stop mocap, mute EMS, close inlets

on_stop() is wrapped in try/except; framework cleanup runs even if your hook raises. That guarantee is non-negotiable — see Safety.

What you can pass to `init`#

Argument	Default	Use it when
`control_strategy`	(required)	Always. Pass an instance of your `ControlStrategyBase` subclass.
`suit_handler`	auto	You need a custom hardware config (e.g. an alternate `MuscleMap` JSON).
`data_streamer`	auto	You're filtering or fusing sensor data — pass a `DataStreamer` subclass.
`stimulator`	auto	Almost never; the default is correct.
`lsl_streamer`	auto (disabled)	You want LSL with non-default settings.
`lsl_enabled`	`False`	You want default-configured LSL outlets.
`external_inputs`	`None`	Bring in LSL inlets from external devices.
`queue_handler` / `control_queue` / `utility_queue`	`None`	A GUI is sending messages to the backend.
`sample_rate`	`100.0`	Metadata only — does not regulate timing.

See engine.py for the full signature.

Per-cycle pipeline#

Every iteration runs these 10 steps in order:

while self._running:
    self.data_streamer.run_cycle()                       # 1–3: collect/process/distribute
    if self.external_inputs:                             # 4: external LSL inlets
        self.control_strategy.external_data = ...
    if self.queue_handler:                               # 5: IPC messages
        self._poll_messages()
    self.control_strategy.run_strategy(...)              # 6: your process()
    self.stimulator.run_stimulator(...)                  # 7: EMS output
    if self.control_strategy.haptic_library:             # 8: custom haptic library
        self.library_stimulator.run_stimulator(...)
    self.lsl_streamer.stream_all_data(...)               # 9: LSL outlets
    self.on_cycle_complete()                             # 10: your hook
    self.cycle_count += 1

The order is fixed. Don't try to re-order by overriding run() — override the relevant hook instead.