DataStreamer

Source: fes_framework/io/data_streamer.py API reference: api_reference.md → DataStreamer

What it is#

DataStreamer is the framework's input pipeline. Every cycle it runs a strict three-phase routine:

1. Collect (framework-owned) — block on the SDK's get_*_on_ready calls, parse the raw ctypes structures into typed dataclasses.
2. Process (user-overridable) — your custom signal-processing hook. Default is no-op.
3. Distribute (framework-owned) — currently a no-op (data is updated in place); reserved as an explicit slot for future distribution logic.

Per cycle it produces:

• BiomechanicalData — 29 joint angles (on by default; toggle off for performance — see below)
• ProcessedData — 20 bone positions and rotations
• StepDetectorData — left_foot_contact, right_foot_contact
• RawData — raw IMU sensor data from 20 body segments
• HeartRateData, HRVData, RawPPGData — only if a PPG sensor is detected

Why it exists#

Reading the Teslasuit SDK directly means working with ctypes structs, multi-call data acquisition (get_raw_data_on_ready + get_skeleton_data_on_ready + get_biomechanical_angles_on_ready + get_foot_contacts_on_ready), and per-call validity checks. Without a streaming layer, every application would re-implement that. The DataStreamer does it once: each cycle ends with a fresh, fully-typed view of every sensor on the suit.

The three-phase split exists so that user code can intercept data after it has been parsed but before it is consumed by the strategy. That's the right place for filtering, custom step detection, and signal fusion. The framework owns Phase 1 (so collection rules stay consistent) and Phase 3 (so the framework can grow new distribution targets without breaking user code).

When to use it#

You don't normally instantiate it. The engine creates one for you, configured against the SuitHandler. You access its current outputs indirectly via your strategy's self.joints, self.contacts, etc.

You subclass it when you need custom signal processing — e.g.:

• low-pass filter the joint angles before the strategy reads them,
• override the foot-contact source (use a knee-angle threshold instead of the SDK detector),
• compute a derived signal (joint velocity from the angle history).

You pass your subclass into the engine like this:

engine = ClosedLoopEngine(
    control_strategy=MyStrategy(),
    data_streamer=MyDataStreamer(suit_handler=engine.suit_handler),
)

(or simply MyDataStreamer(SuitHandler()) and let the engine pick it up).

Minimal example — custom step detection#

from fes_framework.io.data_streamer import DataStreamer

class KneeAngleStepDetector(DataStreamer):
    """Override foot contact detection using knee angle threshold."""

    KNEE_THRESHOLD_DEG = 20.0

    def process(self):
        # Override SDK foot-contact with a knee-flexion heuristic
        self.step_detector_data.right_foot_contact = (
            self.biomechanical_data.KneeFlexExtR < self.KNEE_THRESHOLD_DEG
        )
        self.step_detector_data.left_foot_contact = (
            self.biomechanical_data.KneeFlexExtL < self.KNEE_THRESHOLD_DEG
        )

This reads self.biomechanical_data.KneeFlexExt*, which is refreshed every cycle by default — no extra wiring needed. If you've explicitly disabled biomech collection (see next section) you'd see zero angles and the heuristic would always report "foot in contact".

Biomechanical-angle collection: on by default, opt-out for performance#

The SDK call get_biomechanical_angles_on_ready() runs the inverse- kinematics solver and is the single most expensive call in the per- cycle pipeline. We still leave it ON by default: a strategy that reads self.joints.* should just work without ceremony. Surprising the user with all-zero joint angles is a much worse default than paying the IK cost.

When to turn it off: your strategy never reads self.joints.* (haptic-only, foot-contact-only, raw-IMU-only). Skipping the SDK IK call claws back the per-cycle wait time. Note that processing biomechanical angles is cheap — the cost is in the SDK call itself.

Trade-off summary:

Two ways to turn it off:

# 1) Programmatic — for apps that always run without joint angles.
#    Put this in your ClosedLoopEngine subclass's __init__.
class HapticOnlyEngine(ClosedLoopEngine):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.data_streamer.set_biomech_collection(False)

# 2) Dynamic — wired to a GUI checkbox via UtilityMessage.
#    See examples/walking_fes/backend_mainloop.py for the wiring.
def on_utility_message(self, message):
    self.data_streamer.set_biomech_collection(
        message.BiomechanicalDataCollectionIsActive
    )

If you accidentally turn it off, the symptom is unambiguous: every field of self.joints reads 0.0 forever, and any GUI plot of joint angles is a flat horizontal line even while the user is moving. See Troubleshooting → joint angles are all 0.0.

What you can read in your process() override#

self.biomechanical_data  # BiomechanicalData (29 joint angles)
self.processed_data      # ProcessedData (20 bones, position + quaternion)
self.step_detector_data  # StepDetectorData (foot contacts)
self.raw_data            # RawData (raw IMU per segment)
self.heart_rate_data     # HeartRateData (only if ppg_available)
self.hrv_data            # HRVData (only if ppg_available)
self.raw_ppg_data        # RawPPGData (only if ppg_available)
self.suit_handler        # for hardware access (rare; read-only recommended)
self.ppg_available       # bool — True if PPG sensor detected at startup

Modify in place. The strategy reads the same instances on the next step of the pipeline.

What you can not do#

• Don't override _collect() or _distribute() — they are framework-owned. The leading underscore signals that.
• Don't override run_cycle() — it sequences the three phases.
• Don't allocate large arrays per cycle. Pre-allocate in __init__ (call super().__init__(suit_handler) first) and update in place.

PPG (heart-rate) auto-detection#

DataStreamer probes for a PPG sensor at startup. If it sees real heart-rate or photodiode data within a 2-second window, it sets self.ppg_available = True and continues streaming. If not, it stops the PPG stream and disables the feature for the session.

The probe exists because the SDK silently accepts start_raw_streaming() even on suits without PPG hardware (returning all-zero data forever). Without the probe the framework would emit non-stop zero-valued heart-rate samples, which is worse than missing data.

LSL outlets for PPG (HeartRate, HRV, RawPPG) are only created by LSLStreamer when ppg_available is true.

See also#

• API Reference → DataStreamer
• Data Types — what DataStreamer populates
• ControlStrategy — the consumer of streamed data
• Implementation Guide → Step 2 — reading sensor data in process()
• Source: fes_framework/io/data_streamer.py
• Example: examples/atomic/reading_sensor_data.py