Step 6 — External inputs (LSL inlets)

What you're adding#

The framework already lets your strategy fuse Teslasuit sensors with your own algorithm. Many lab setups have additional sensors — force plates, EEG amps, optical mocap — that you want to fold into the same control loop.

The framework supports this via Lab Streaming Layer inlets. You declare which streams you want; the orchestrator builds an ExternalInputManager in the backend subprocess, polls each registered stream every cycle, and writes the latest sample to self.external_data keyed by stream name.

Code#

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


class GRFTriggeredQuadStrategy(ControlStrategyBase):
    """Stimulate right quadriceps when the force plate sees > 100 N vertical load."""

    GRF_STREAM = "ForcePlate_GRF"
    THRESHOLD_N = 100.0

    def process(self) -> None:
        sample = self.external_data.get(self.GRF_STREAM)
        vertical_force = sample.data[2] if sample is not None else 0.0

        if vertical_force > self.THRESHOLD_N:
            self.ems_output.quadriceps_right = EMSParamData(
                IsMuted=False, Amplitude=30, PulseWidth=120, Period=20.0,
            )
        else:
            self.ems_output.quadriceps_right = EMSParamData(IsMuted=True)

# app/main.py
from fes_framework.orchestrator import launch
from app.strategy import GRFTriggeredQuadStrategy

if __name__ == "__main__":
    launch(
        GRFTriggeredQuadStrategy,
        external_input_streams=["ForcePlate_GRF"],
        external_input_timeout=5.0,         # seconds to wait for the stream
    )

Walkthrough#

external_input_streams=["ForcePlate_GRF"] — the orchestrator takes a list of LSL stream names. Each one becomes an entry in self.external_data. The actual ExternalInputManager instance is built inside the backend subprocess (pylsl objects can't be pickled across the process boundary, so the orchestrator passes the stream names as strings and builds the manager in run_engine_process).

external_input_timeout=5.0 — how long the manager waits for each stream to appear on the network. After the timeout it logs a warning and continues. If you depend on the stream, gate it in on_start() (more below).

self.external_data.get(stream_name) — returns either an ExternalSample or None. None means the stream hasn't produced a sample yet (newly connected, or the upstream device pauses between samples). Always handle the None case.

sample.data[2] — data is a numpy array of channel values for the most recent sample. The shape is whatever the stream publishes; for a 6-channel force plate, indices 0–5 are typically Fx, Fy, Fz, Mx, My, Mz. Read the device's documentation for the exact mapping.

sample.timestamp — LSL's synchronised timestamp for the sample. Useful for deciding whether the data is fresh (within 100 ms of pylsl.local_clock()) or stale.

Refusing to start without a stream#

If your application requires the external stream, gate it in on_start():

class GRFGatedEngine(ClosedLoopEngine):
    def on_start(self) -> None:
        if not self.external_inputs.is_connected("ForcePlate_GRF"):
            print("[engine] Force plate not found — refusing to start.")
            self.stop()
            return

This prevents the loop from running blind when the lab equipment is disconnected.

Multiple streams#

Pass a list:

launch(
    MyStrategy,
    external_input_streams=[
        "ForcePlate_GRF",
        "EEG_F4",
        "OpticalMocap_Pelvis",
    ],
    external_input_timeout=10.0,
)

In the strategy, read each by name:

def process(self) -> None:
    grf = self.external_data.get("ForcePlate_GRF")
    eeg = self.external_data.get("EEG_F4")
    pelvis = self.external_data.get("OpticalMocap_Pelvis")
    # ... fuse and act ...

The PRD constrains external inputs to 2 streams. The framework doesn't enforce that limit, but more than two inlets per cycle starts to noticeably affect loop timing — profile carefully.

What an LSL inlet sample looks like#

@dataclass
class ExternalSample:
    data: numpy.ndarray   # shape: (n_channels,) for the latest sample
    timestamp: float      # LSL clock time

If the upstream device publishes at a different rate from the engine (e.g. force plate at 1000 Hz, engine at 100 Hz), the manager keeps the most recent sample. There's no buffering on the inlet side; if you need the full stream, record it via LabRecorder (Step 7) and replay offline.

How to test without a real device#

Publish a fake stream from another Python process:

# fake_force_plate.py
import time, numpy as np
from pylsl import StreamInfo, StreamOutlet

info = StreamInfo("ForcePlate_GRF", "Force", 6, 100, "float32", "fake_grf")
outlet = StreamOutlet(info)

t = 0.0
while True:
    Fz = 80 + 100 * (t % 2 < 1)        # 80 N idle, 180 N during stance
    outlet.push_sample([0, 0, Fz, 0, 0, 0])
    time.sleep(0.01)
    t += 0.01

Run python fake_force_plate.py in one shell, then your app in another. The app should toggle stimulation every second.

Verify#

With the app running and a stream named ForcePlate_GRF publishing on the network:

[Backend] ExternalInputManager: connected to "ForcePlate_GRF"
[Orchestrator] Running headless (Ctrl-C to stop)

If the stream isn't found within external_input_timeout:

[Backend] ExternalInputManager: timeout waiting for "ForcePlate_GRF"
[Orchestrator] Running headless (Ctrl-C to stop)

The strategy still runs; self.external_data["ForcePlate_GRF"] just returns None.

What you've learned#

• External inlets are declared at launch time as a list of stream names.
• self.external_data[stream_name] returns the latest sample (or None).
• Always handle the None case — streams can disappear or pause.
• Use on_start() to refuse to start without the stream when required.

Next#

Step 7 — Recording and LSL outlets. You'll publish every channel of your application as LSL outlets so LabRecorder can record an XDF file of the entire session.