Safety

The three guarantees#

1. The FES kill switch is unbypassable. When UtilityMessage.FesIsActive is False, the framework writes muted parameters to every muscle regardless of what the strategy produced. No user code path can override this.
2. Cleanup runs on every exit. Whether the engine exits normally, on Ctrl-C, or because process() raised an exception, the framework stops mocap streaming, stops the haptic player, and closes all external inputs.
3. The hardware can always be reached. As long as Teslasuit Control Center is running, engine.stop() (or the Ctrl-C handler) will halt the loop within one cycle (~10 ms) and put the device in a known-safe state.

These guarantees are implemented by the framework so the application doesn't have to. You can write strategies that ignore safety entirely and still produce a safe application, because the safety machinery sits outside process().

What the framework is not#

• It is not a medical device. There is no certification, no electrical isolation guarantee beyond what the Teslasuit hardware itself provides, no clinical safety case.
• It does not detect adverse reactions. A user reporting pain or involuntary movement is a human decision, not a framework one.
• It does not enforce stimulation parameter ranges. You can pass Amplitude=100 and the framework will deliver it. Sane upper bounds are the application's responsibility.

This means the framework can guarantee that the system does what the operator told it to. It cannot guarantee that what the operator told it to do is appropriate for the user.

Guarantee 1 — FES kill switch#

How it works#

UtilityMessage carries a FesIsActive boolean. The framework applies it after every process() call, before the stimulator runs:

# fes_framework/control/strategy_base.py

def run_strategy(self, control_message, biomechanical_data,
                 step_detector_data, ems_data, fes_active=True):
    self.joints = biomechanical_data
    self.contacts = step_detector_data
    self.params = control_message

    self.process()                                         # user logic

    _apply_ems_output(fes_active, self.ems_output, ems_data)

def _apply_ems_output(fes_active, source, target):
    for muscle in EmsData.__dataclass_fields__:
        if fes_active:
            setattr(target, muscle, getattr(source, muscle))
        else:
            setattr(target, muscle, _MUTED)               # always muted

_MUTED is a pre-allocated EMSParamData(IsMuted=True, ...) reused every cycle. By the time the stimulator sees ems_data, every muscle has either the strategy's intent (when fes_active=True) or the muted value (when fes_active=False). The strategy never touches fes_active; the framework does.

How the operator triggers it#

In a GUI application, set utility_message.FesIsActive = False on the GUI side. The change is sent over utility_queue and applied on the backend's next cycle (~10 ms latency).

Recommended GUI pattern: a prominent emergency-stop button that sets FesIsActive = False immediately on click, and sends the UtilityMessage over the queue.

What you should still build into your application#

• An emergency-stop UI that's bigger and more obvious than any other button.
• A keyboard shortcut (Esc, space, etc.) that flips FesIsActive=False even when the mouse is unreachable.
• A power-loss policy: if the backend process dies, the SDK stops streaming, which stops stimulation. Verify this in your particular setup.

Guarantee 2 — Cleanup always runs#

How it works#

ClosedLoopEngine.run() wraps the loop in try/finally, with on_stop() (your hook) and _cleanup() (framework) inside:

def run(self) -> None:
    self._running = True
    self.cycle_count = 0
    try:
        self.control_strategy.setup(...)
        self.on_start()
        while self._running:
            # ... per-cycle pipeline ...
    except KeyboardInterrupt:
        logger.info("ClosedLoopEngine interrupted by user")
    finally:
        # Developer hook first — may raise; framework cleanup still runs
        try:
            self.on_stop()
        except Exception:
            logger.exception("Error in on_stop() hook")

        self._cleanup()
        self._running = False

_cleanup() itself wraps each step in its own try/except so a failure in one step doesn't prevent the others:

def _cleanup(self) -> None:
    try: self.suit_handler.stop_mocap_streaming()
    except Exception: logger.exception("Error stopping mocap streaming")
    if self.data_streamer.ppg_available:
        try: self.suit_handler.stop_ppg_streaming()
        except Exception: logger.exception("Error stopping PPG streaming")
    try: self.suit_handler.stop_player()
    except Exception: logger.exception("Error stopping stimulation player")
    if self.external_inputs is not None:
        try: self.external_inputs.close_all()
        except Exception: logger.exception("Error closing external inputs")

The result: even on a buggy on_stop() or a partial-hardware-failure shutdown, the framework still:

• stops mocap streaming,
• stops PPG streaming (if it was running),
• stops the haptic player (which mutes any active stimulation),
• closes any LSL inlets.

Why on_stop() is not allowed to block cleanup#

If a user's on_stop() does something risky (network call, file flush) and raises, the framework still has to leave the suit safe. Wrapping on_stop() in try/except is the only way to guarantee that. The framework documents this contract explicitly in the docstring.

What you can rely on#

Every engine.run() invocation, no matter how it ends:

Guarantee 3 — Stop is one cycle away#

How it works#

engine.stop() flips self._running = False. The main loop checks this flag at the top of every iteration:

while self._running:
    ...

Once a cycle is in progress, it completes — stop() doesn't interrupt mid-cycle. This bounds the worst-case latency from "asked to stop" to "stopped" at one cycle (~10 ms at 100 Hz).

Where stop() is called#

• Ctrl-C — KeyboardInterrupt is caught in run(), which falls through to finally: (calling on_stop() and _cleanup()).
• STOP_SENTINEL on the utility queue — the orchestrator sends this on shutdown. _poll_messages() recognises it and calls self.stop().
• Manual call — your code (or hook) can call self.stop() any time. The next cycle exits the loop.

What you can rely on#

If you've passed Ctrl-C, hit the GUI's emergency-stop, or the orchestrator received SIGTERM, the suit will stop stimulating within one cycle. The cleanup guarantees follow.

Calibration as an explicit pre-condition#

Calibration is not a framework guarantee — it's the application's responsibility. The framework only logs a warning when run() is called without calibration:

Starting ClosedLoopEngine without calibration. MoCap data may be
inaccurate. Call engine.calibration.calibrate() before engine.run().

For applications where joint accuracy matters, the canonical pattern is a calibration gate inside on_start():

class CalibratedEngine(ClosedLoopEngine):
    def on_start(self) -> None:
        result = self.calibration.calibrate()
        if not result.success:
            self.stop(); return
        if not self.calibration.check_quality().is_acceptable:
            self.stop(); return

If the gate calls self.stop(), the engine exits the loop on the next cycle (i.e. before any stimulation has happened) and runs cleanup as usual. See Implementation Guide → Step 4.

Operator-side practice#

The framework's guarantees are necessary but not sufficient for safe operation. Operators should:

• Start at low amplitude (20–30%) for any new strategy or new user, increase incrementally with explicit consent.
• Watch the user for the entire session. Never leave a powered suit unsupervised.
• Verify the kill switch works at the start of every session. Toggle FES off, see stimulation stop within ~10 ms, toggle back on.
• Have a hardware off path — a power switch, an unplug procedure, a way to disconnect the user from the suit if something fails catastrophically.
• Document the session — what parameters were used, what was observed. The LSL outlets give you the raw data; what they don't capture is operator judgement.

See also#

• Stimulator — where the kill switch is enforced
• ClosedLoopEngine — lifecycle hooks — on_stop() contract
• Calibration — how to add a quality gate
• Implementation Guide → Step 4
• Implementation Guide → Step 5 — putting an emergency-stop button in the GUI
• Design principle 10
• Source: fes_framework/engine.py, fes_framework/control/strategy_base.py