Atomic Examples

What's in examples/atomic/#

examples/atomic/
├── README.md                       # this folder's standalone README
├── __init__.py
├── minimal_closed_loop.py          # ~60 lines  — Step 1
├── reading_sensor_data.py          # ~85 lines  — Step 2
├── semantic_muscle_control.py      # ~110 lines — Step 3
├── calibration_gate.py             # ~115 lines — Step 4
└── lsl_streaming.py                # ~105 lines — Steps 6-7

Each file imports only from the public fes_framework.* API and runs as a standalone script. Together they cover the Must-priority framework features.

1. minimal_closed_loop.py#

The smallest possible ControlStrategyBase subclass. It counts cycles and emits no stimulation. Hands off to orchestrator.launch() for the standard dual-process scaffold.

class MinimalStrategy(ControlStrategyBase):
    def __init__(self) -> None:
        super().__init__()
        self._cycles = 0

    def process(self) -> None:
        self._cycles += 1
        if self._cycles % 100 == 0:
            print(f"[MinimalStrategy] cycle {self._cycles}")

if __name__ == "__main__":
    launch(MinimalStrategy)

Run:

python examples/atomic/minimal_closed_loop.py

Anchored by: Step 1

2. reading_sensor_data.py#

Reads self.joints (29 joint angles) and self.contacts (foot contacts) every cycle. Prints them once per second. Read-only — no stimulation.

def process(self) -> None:
    self._cycles += 1
    if self._cycles % 100 == 0:
        print(
            f"KneeFlexExtL={self.joints.KneeFlexExtL:+6.1f}° "
            f"KneeFlexExtR={self.joints.KneeFlexExtR:+6.1f}° "
            f"left_contact={self.contacts.left_foot_contact}"
        )

Run:

python examples/atomic/reading_sensor_data.py

Anchored by: Step 2

3. semantic_muscle_control.py#

Two demonstrations in one file:

1. Query the MuscleMap semantically — by_side, by_region, list_muscles, get_channels.
2. Write stimulation by name — self.ems_output.quadriceps_left = EMSParamData(...).

The strategy toggles the left quadriceps on/off at 1 Hz with a conservative 20% amplitude.

def setup(self, muscles=None, config=None) -> None:
    super().setup(muscles=muscles, config=config)
    print(f"hardware:  {self.muscles.hardware_version}")
    print(f"muscles:   {self.muscles.list_muscles()}")
    print(f"upper_leg: {[m.name for m in self.muscles.by_region('upper_leg')]}")
    print(f"channels for quadriceps_left: "
          f"{self.muscles.get_channels('quadriceps_left')}")

def process(self) -> None:
    if self._on:
        self.ems_output.quadriceps_left = EMSParamData(
            IsMuted=False, PulseWidth=120, Amplitude=20, Period=20.0,
        )

Run:

python examples/atomic/semantic_muscle_control.py

⚠️ Safety: delivers real stimulation. Test on yourself first; do not run on someone else without consent and the Safety checklist understood.

Anchored by: Step 3

4. calibration_gate.py#

Demonstrates the headless calibration flow:

1. Construct ClosedLoopEngine (auto-builds CalibrationAPI).
2. Wait for the user to confirm I-pose.
3. Call engine.calibration.calibrate().
4. Check engine.calibration.check_quality().
5. Either start engine.run() or abort with a non-zero exit code.

This example uses ClosedLoopEngine directly (not orchestrator.launch()) so it can abort before the loop starts.

result = engine.calibration.calibrate()
if not result.success:
    return 1

quality = engine.calibration.check_quality()
print(f"quality={quality.overall_score:.0%} "
      f"({quality.sensors_reporting}/{quality.sensors_expected} sensors)")
if not quality.is_acceptable:
    return 2

engine.run()

Run:

python examples/atomic/calibration_gate.py

Anchored by: Step 4

5. lsl_streaming.py#

Two demonstrations in one file:

1. Outlets: lsl_enabled=True flips on the 7 framework outlets — TS_Biomechanics, TS_API_StepDetector, TS_EMSParameters, TS_BonePosition, TS_RawData, AppData_ControlMessage, AppData_UtilityMessage. LabRecorder sees them all.

2. Inlets: external_input_streams=["ExampleForce"] registers an LSL inlet. The strategy reads self.external_data.get("ExampleForce").

launch(
    LSLDemoStrategy,
    lsl_enabled=True,                                # 7 outlets
    external_input_streams=["ExampleForce"],         # 1 inlet
    external_input_timeout=2.0,
)

Run:

# Open LabRecorder, click Update, see 7 streams
python examples/atomic/lsl_streaming.py

To test the inlet, publish a fake stream from a separate Python shell — see Step 6 — How to test without a real device.

Anchored by: Step 6, Step 7

When to use atomic examples vs. full applications#

The atomic examples are intentionally not integrated. They demonstrate one concept each. Real applications combine many. That's what the full examples do.

See also#

• Implementation Guide — the same concepts, but built up incrementally into one application.
• examples/atomic/README.md — the in-folder README with run instructions.
• Concepts — concept-level descriptions for each demonstrated concept.