Outlets

An Outlet owns a .push(data) method plus a daemon thread that sends the latest pushed value to its destination at a steady rate. See Publish a data stream.

If something moves, you want a target

An output is a paced sender and nothing else: no aliases, no declared range, no clamp, no neutral frame on shutdown. A hand, a motor, a haptic or a cursor is a Target — see Drive your own device.

An outlet is part of that road rather than an alternative to it: RemoteTarget builds one LSLOutlet per control it drives.

Outlet

The one base class in the public API. Everything else you implement against is a structural Protocol — see design principle 1. Outlet is a class because it is sixty lines of running code rather than a shape: a paced daemon thread, a latest-wins slot, and per-error-kind deduplication. Subclass it to add _send.

(The package stays myogestic.outputs because it also holds the output-side filters and EdgeTrigger, which are not outlets.)

Outlet

Outlet(hz: float = 50)

Base class for "send the latest pushed vector at hz" outputs.

Subclass to define a new transport: override _send with the actual write (LSL push_sample, UDP sendto, serial write, gRPC RPC, ...). The base class handles everything else:

• A daemon output thread is started in __init__ and runs for the lifetime of the outlet. Each tick it reads the latest pushed vector and calls _send.
• push is the caller-facing API: write the latest value to an atomic slot (CPython's GIL guarantees atomic reference assignment). It is latest-wins, not queued - if you push faster than hz, intermediate values are overwritten and never sent.
• Exceptions raised by _send are caught, deduplicated per (error class, message) pair, and logged once. A flapping destination logs one line per failure mode and the send thread keeps running.

Subclassing checklist:

1. Call super().__init__(hz=...) from your __init__ (after opening the underlying socket / serial port / channel - the send thread starts immediately).
2. Implement _send(self, data: np.ndarray) -> None. Treat data as read-only; validate shape; raise on misuse rather than silently mis-sending.
3. Override stop if you need to close a resource (see LSLOutlet.stop for an example).

Outputs are user-owned: instantiate them at module scope, call .push(data) from inside @pipeline.predict. Do not register them with App; the framework does not track them.

Parameters:

Name	Type	Description	Default
hz	float	Send rate of the daemon thread in Hz. Default 50. Tune to match your destination's appetite - LSL subscribers handle high rates well, a serial UART or a gRPC server may not.	50

Examples:

>>> from myogestic.outputs import Outlet
>>> import socket, numpy as np
>>>
>>> class MyOutlet(Outlet):
...     def __init__(self, addr, hz=50):
...         self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
...         self._addr = addr
...         super().__init__(hz=hz)
...     def _send(self, data):
...         self._sock.sendto(data.astype(np.float32).tobytes(),
...                           self._addr)

push

push(data: ndarray) -> None

Set the latest-value slot. Atomic; latest-wins; non-blocking.

Call from inside @pipeline.predict. The send thread picks the value up on its next tick. If you push faster than hz, intermediate values are dropped.

flush

flush() -> None

Send the latest pushed value now, instead of on the next tick.

The send loop is paced, so a value pushed immediately before teardown is normally never sent at all: the thread is mid-sleep, and stop ends the loop before it wakes. Anything guaranteeing rest-on-stop - the neutral "release everything" frame - must flush rather than push.

Errors are swallowed and logged exactly as on a normal tick, and a flush before the first push does nothing.

stop

stop() -> None

Stop the send thread.

Subclasses that hold resources (sockets, serial ports) should override and call super().stop() first.

Built-in outlets

LSLOutlet

LSLOutlet(name: str, n_channels: int, hz: float = 50, *, channel_names: Sequence[str] | None = None, channel_units: Sequence[str] | None = None, source_id: str = '')

Bases: Outlet

Publish a 1-D vector to a Lab Streaming Layer outlet.

The dual of LSLSource - call .push(vec) from inside @pipeline.predict, and the framework's daemon output thread re-sends the latest pushed vector at the configured hz. Channel count is locked at construction time so the LSL metadata matches what subscribers see.

Parameters:

Name	Type	Description	Default
name	str	Outlet name advertised on the LSL network. Typically the stream name that downstream tools (the Virtual Hand, a recorder, another MyoGestic app) resolve by.	required
n_channels	int	Fixed channel count. Push vectors must have this length or _send raises ValueError instead of silently mis-sending.	required
hz	float	Send rate of the daemon thread (Hz). Default 50. Push faster than hz is fine: latest-wins, the slot just gets overwritten.	50
channel_names	Sequence[str] | None	Optional per-channel labels, published in the stream's description so a subscriber can resolve a channel by name instead of by position. Must be exactly n_channels long. Pass ControlSet.channel_labels() to make a control stream self-describing: a reordered configuration then renames channels rather than silently remapping them.	None
channel_units	Sequence[str] | None	Optional per-channel unit strings, same length rule. Control-standard DOFs are normalized, so "normalized" is the honest value for them.	None
source_id	str	Optional stable identifier for this outlet. LSL uses it to recognise the same logical stream across a restart, so a subscriber can reconnect instead of treating it as a new stream.	''

Examples:

>>> outlet = LSLOutlet("VHI_Hand", n_channels=9, hz=32)
>>> @pipeline.predict
... def predict(model, features):
...     pose = model.compose_pose(features)
...     outlet.push(pose)
...     return {"pose": pose}

stop

stop() -> None

Stop the daemon thread and take the stream off the network.

Overridden because the base class only stops the thread, and an LSLOutlet holds a resource: liblsl keeps a stream discoverable for as long as its StreamOutlet is alive, not for as long as anything is pushing to it. Dropping the reference and waiting for the collector is not enough either — the app raises the GC threshold at startup, so a stopped outlet can stay resolvable for a long time.

That matters because a stopped-but-live outlet is not inert. It shares its source_id with the outlet that replaced it, so a consumer sees two equally valid producers of one stream and may resolve the dead one — reading a layout that no longer matches, silently, with every channel after the first change shifted by one.