Concepts Overview
The big picture: how the framework's pieces compose into a working closed-loop application (FES or any other input-process-haptic pipeline — see audience.md). Every other page in this section zooms into one piece.
The closed-loop in one diagram#
┌─────────────────────── Backend subprocess ──────────────────────┐
│ │
│ Teslasuit ──▶ SuitHandler ──▶ DataStreamer ──┐ │
│ hardware (collect/process│ │
│ /distribute) │ │
│ ▼ │
│ ┌─── self.joints │
│ ├─── self.contacts │
│ ControlStrategyBase ◀─┤── self.params │
│ .process() ├─── self.external_data ◀──── ExternalInputManager
│ └─── self.muscles (LSL inlets)
│ │ │
│ ▼ │
│ self.ems_output │
│ │ │
│ FES kill switch ─▶ Stimulator ──▶ MuscleMap
│ (UtilityMessage) │ ──▶ Teslasuit
│ │ (haptic SDK)
│ ▼ │
│ LSLStreamer ──▶ LSL outlets
│ │
└─────────────────────────────────┬───────────────────────────────┘
│ multiprocessing.Queue, SharedRingBuffer
│
┌─────────────────────────────────▼───────────────────────────────┐
│ Main process (optional GUI) │
│ PyQt5 widgets ──▶ control_queue (parameters) │
│ ◀── utility_queue (status flags) │
│ ◀── SharedRingBuffer (sensor data, ~30 Hz) │
└─────────────────────────────────────────────────────────────────┘What each piece does#
Read these one-pagers in any order. Each follows the same shape: what it is, why it exists, when you touch it, a code snippet, and links into the API reference.
Hardware and data acquisition (input)#
- SuitHandler — manages the Teslasuit
connection, exposes the haptic / mocap / PPG subsystems and the
MuscleMap. The framework auto-creates one; you rarely instantiate it. - DataStreamer — runs the three-phase
collect → process → distribute cycle each iteration. Override
process()only for custom signal pre-processing. - Data Types — the typed dataclasses
(
BiomechanicalData,EmsData,StepDetectorData, …) that flow through the system. - MuscleMap — anatomical name → SDK channel ID.
Loaded from
muscle_map_4R.jsonat startup; accessed viaself.musclesin your strategy.
Control (your code)#
- ControlStrategyBase — the user extension
point. Subclass it, implement
process(). This is where 95% of application code lives. - ClosedLoopEngine — the orchestrator. It
runs the cycle and offers lifecycle hooks (
on_start,on_stop,on_cycle_complete).
Output#
- Stimulator — translates
EmsDatato Teslasuit haptic SDK calls viaMuscleMap. The FES kill switch is enforced here. - LSL Streaming — 7 LSL outlets at 100 Hz for lab integration (LabRecorder, MATLAB, Python). Off by default.
Cross-cutting#
- Calibration — trigger and quality assessment.
Typically gated in
on_start(). - Messaging —
ControlMessage(per-app params),UtilityMessage(system flags),HapticLibrary(custom playables). - IPC and processes —
SharedRingBufferandmultiprocessing.Queuebetween backend and GUI. - Safety — kill switch, calibration gating, shutdown guarantees.
- Architecture — the multi-process design and full data-flow diagram.
Per-cycle flow#
Every iteration of the engine's main loop, in order:
| # | Step | What happens | Owned by |
|---|---|---|---|
| 1 | Collect | SDK blocks until a fresh sensor frame arrives; ctypes data is parsed into dataclasses | DataStreamer._collect() |
| 2 | Process (data) | Custom signal processing hook (filtering, custom step detection) | DataStreamer.process() (user override) |
| 3 | Distribute | Data is made available to downstream consumers (currently in-place; future hook) | DataStreamer._distribute() |
| 4 | Poll external | LSL inlets are pulled into self.external_data | ExternalInputManager.pull_all() |
| 5 | Poll IPC | control_queue and utility_queue are drained to latest message; hooks fire | QueueHandler |
| 6 | Strategy | process() is called; reads self.joints/contacts/params/external_data, writes self.ems_output | your code |
| 7 | Apply EMS | FES kill switch applied; EmsData → Stimulator → SDK haptic calls | Stimulator.run_stimulator() |
| 8 | Apply haptic library | Custom playables muted/unmuted (only if strategy.haptic_library is set) | LibraryStimulator |
| 9 | Stream | All 7 LSL outlets pushed (only if enabled) | LSLStreamer.stream_all_data() |
| 10 | Post-cycle | on_cycle_complete() hook (write to SharedRingBuffer here, log metrics, etc.) | engine subclass (user) |
The hardware governs timing. There is no software clock.
What you typically write#
For most applications, three files:
my_app/
├── strategy.py # subclass ControlStrategyBase, implement process()
├── messages.py # subclass ControlMessage to declare your parameters
└── main.py # call orchestrator.launch(MyStrategy)If you add a GUI, two more:
├── engine.py # subclass ClosedLoopEngine, override on_start / on_cycle_complete
└── gui/ # PyQt5 widgets that read SharedRingBuffer / send to queuesThat's it. The framework handles everything else.
See Project anatomy for a complete tour of how a real application is laid out.
What you typically don't write#
| Concern | Why you don't touch it |
|---|---|
| Connecting to Teslasuit | SuitHandler does it once at engine construction |
| Reading SDK frames | DataStreamer._collect() runs every cycle |
| Parsing raw ctypes | init_utils.update_inplace populates dataclasses |
| Looking up channel IDs | MuscleMap resolves names to IDs at startup |
| Looped haptic playables | Stimulator creates and re-uses them |
| FES kill switch | Engine applies FesIsActive after every process() |
| Subprocess setup | orchestrator.launch() handles the fork |
| Process cleanup on Ctrl-C | Engine's _cleanup() is unconditional |
| LSL outlet creation | LSLStreamer creates 7 outlets at startup if enabled |
| LSL inlet polling | ExternalInputManager.pull_all() runs every cycle |
| GUI thread / Qt event loop | MainWindow template handles it |
When to step outside the defaults#
| You want to … | What to override |
|---|---|
| React to sensor data | Just ControlStrategyBase.process() |
| Filter sensor data before your strategy sees it | DataStreamer.process() |
| Add a custom step detector | DataStreamer.process() (modify step_detector_data) |
| Tune parameters at runtime from a GUI | Subclass ControlMessage; the strategy reads self.params |
| Refuse to start until calibration is good | Subclass ClosedLoopEngine; override on_start() |
| Bring in external device data | Pass external_inputs=ExternalInputManager(...) to the engine |
| Stream data to LabRecorder | lsl_enabled=True |
| Add real-time visualisation | Subclass ClosedLoopEngine, override on_cycle_complete() to write SharedRingBuffer |
| Fire custom haptic patterns | Subclass HapticLibrary and populate self.haptic_library in setup() |
Each of these is documented as a step in the Implementation Guide.
Mental model checklist#
Before writing your own application, you should be able to answer:
- What does my strategy read each cycle? (joints, contacts, params, external_data)
- What does my strategy write? (ems_output)
- Where does the timing come from? (Teslasuit SDK; never
time.sleep) - Where is the FES kill switch? (
UtilityMessage.FesIsActive, framework-applied) - Where do my parameters come from? (
ControlMessagesubclass, sent via the GUI's queue) - What are the two processes? (backend = engine; main = GUI or just
wait())
If any of those don't have a one-line answer yet, read the page that covers it.
Next#
Choose your path:
- Start the Implementation Guide and build an application step by step.
- Drill into any concept above.
- Look up an API in the API Reference.
