Skip to content

Core API

App lifecycle

App

App(name: str, theme: bool = True, docking: bool = False, ui_scale: float | None = None)

Top-level application object.

Owns the GUI loop, the Context, the run-loop lifecycle hooks, and the recording state machine.

Construct one per process. Register streams via app.streams(...), register your UI via @app.ui, then call app.run(). Optional extensions like Pipeline(app) register themselves via app.before_run_hooks / app.cleanup_hooks - user code rarely needs to touch those lists directly.

On desktop, ImGui multi-viewport is on by default: floating windows (e.g. the signal viewer's channel-grid Edit… window) open as their own native OS windows, square-cornered and opaque. Skipped in the browser (no backend for extra OS windows).

Parameters:

Name Type Description Default
name str

Window title. Also used for the persisted ImGui state file (.imgui_state/<name>.ini) when docking=True.

required
theme bool

Apply MyoGestic's built-in ImGui theme. Set False to keep the Dear ImGui defaults.

True
docking bool

Experimental - enable ImGui docking (a full-screen dockspace) so panels registered via app.popout(...) become tearable DockableWindows. macOS Retina viewport sizing of detached windows can be wrong on the first draw.

False
ui_scale float | None

Global UI zoom factor - scales the font and imgui's style metrics (padding, spacing, rounding). None uses $MYOGESTIC_UI_SCALE then 1.0; the env var overrides an explicit value. Clamped to [0.5, 2.0]. No effect when theme=False.

None

Examples:

>>> from myogestic import App
>>> app = App("EMG demo")
>>> @app.ui
... def ui(ctx):
...     pass
>>> app.run()

Methods:

Name Description
streams

Register one or more streams with the app.

add_stream

Register a stream, and start it if the app is already running.

remove_stream

Stop a stream and unregister it.

bridges

Register one or more Bridge subprocesses with the app.

ui

Decorator. Register the render callback.

popout

Register a dockable window before run().

start_recording

Begin recording all connected streams to a new session.

stop_recording

Stop the active recording and pack the session to a .session.zip.

discard_recording

Stop the active recording and delete it, unsaved and unpacked.

run

Blocking entry point.

streams

streams(*streams: Stream) -> None

Register one or more streams with the app.

Each stream is keyed by its name into ctx.streams. Acquisition threads start when app.run() is called, not at registration time. Calling this with the same name overwrites the previous registration - typically you call it once at setup.

Parameters:

Name Type Description Default
*streams Stream

One or more Stream instances.

()

add_stream

add_stream(stream: Stream) -> bool

Register a stream, and start it if the app is already running.

The counterpart to remove_stream, for an app that lets the operator add a device rather than declaring its streams up front. Before run, this is streams with a return value; after it, it also does the Stream.start that run would have done — nothing else will, because run starts each stream exactly once on the way in.

Refused while recording, and refused for a name already taken. A session sizes one Zarr array per stream at start_recording, so a stream that appears afterwards has nowhere to write; and overwriting a live name would strand the running acquire thread of whatever it replaced.

Returns:

Type Description
bool

False if the stream was refused, with the reason in ctx.status_message.

remove_stream

remove_stream(name: str) -> bool

Stop a stream and unregister it.

Stops the acquire thread and disconnects the source, then drops the name from ctx.streams. Widgets bound to it by name report it missing rather than failing — that is why they look it up every frame.

Refused while recording: stop_recording walks ctx.streams to detach the session, so a stream removed mid-take would keep the session attached and never be finalised.

Returns:

Type Description
bool

False if the stream was refused or was not registered.

bridges

bridges(*bridges: Any) -> None

Register one or more Bridge subprocesses with the app.

Bridges run in their own process (webcam, ultrasound, depth camera, …) and publish an LSL clock stream the main app subscribes to. Registering does not start them — call bridge.start() yourself, unlike streams. Each bridge goes into ctx.bridges under its .name; the same name overwrites.

Nothing renders a bridge for you: bridge.status and bridge.alive are there if you want to show them.

Parameters:

Name Type Description Default
*bridges Any

One or more bridge instances - each must expose a .name attribute and a Bridge-like interface (.start(), .stop()).

()

ui

ui(fn: Callable[[Context], None]) -> Callable[[Context], None]

Decorator. Register the render callback.

@app.ui def my_ui(ctx): imgui.text(f"State: {ctx.state}")

popout

popout(title: str, gui_fn: Callable[[], None], *, default_open: bool = True, can_be_closed: bool = True, remember_is_visible: bool | None = None) -> None

Register a dockable window before run().

Preferred over calling popout_panel(...) inside @app.ui: Hello ImGui gets the complete DockableWindow list before launch rather than on the first frame.

start_recording

start_recording(base_path: str = 'sessions') -> None

Begin recording all connected streams to a new session.

Creates base_path/<timestamp>/ and starts appending each stream's data + timestamps to per-stream Zarr arrays. Streams whose info is still None (disconnected) are skipped - they won't be retroactively captured if they connect later in the recording. Refuses to start if ctx.state isn't "idle"; updates ctx.status_message with the result.

Parameters:

Name Type Description Default
base_path str

Directory where the per-session subfolder is created. Defaults to "sessions".

'sessions'

stop_recording

stop_recording() -> None

Stop the active recording and pack the session to a .session.zip.

Finalises the per-stream Zarr arrays, writes the label track to labels.json, and kicks off a daemon thread that packs the session folder into a single <timestamp>.session.zip archive (the original folder is kept until the pack succeeds). Refuses to stop if ctx.state isn't "recording".

discard_recording

discard_recording() -> None

Stop the active recording and delete it, unsaved and unpacked.

The counterpart to stop_recording for a take the operator threw away — a false start, a bad trial. Detaches every stream, removes the session folder, and returns to "idle". Nothing is written to meta.json and no archive is produced, so a discarded recording leaves no trace to clean up later.

Refuses to run if ctx.state isn't "recording".

run

run(mode: str = 'gui', window_size: tuple[int, int] = (1280, 800), fullscreen: bool = False) -> None

Blocking entry point.

Call tree (top → bottom = runtime order):

App.run()
├─ 1. Stream.start()          per stream → daemon acquire thread
├─ 2. before_run_hooks(app)  extensions register here
│    └─ e.g. myogestic.ml.attach_pipeline → starts predict thread
├─ 3. self._gui_loop()  ← main thread, BLOCKS
│    └─ immapp.run → per frame: self._ui_fn(self.ctx)  (your @app.ui)
└─ 4. [finally] cleanup - always runs, even on startup failure
     ├─ cleanup_hooks(app)   each wrapped in try/except
     ├─ Stream.stop()         per stream
     ├─ Bridge.stop()         per bridge
     └─ process_launcher._cleanup_all()

Core has only idle ↔ recording. myogestic.ml.attach_pipeline(app) adds training/predicting states + their transition methods.

AppState

Bases: StrEnum

Core app-state values. Extensions (e.g. myogestic.ml.PipelineState) add more.

Context.state is a bare str so extensions can introduce their own states without subclassing. Each module validates transitions within its own namespace only.

Examples:

>>> from myogestic import AppState
>>> AppState.RECORDING.value
'recording'

Context dataclass

Context(streams: dict[str, Stream] = dict(), bridges: dict[str, Any] = dict(), state: str = IDLE, session: Session | None = None, class_names: list[str] = list(), control_space: Any = None, current_label: int = -1, status_message: str = '', logs: list[str] = list())

Shared state all threads read/write.

Extensions may add own fields dynamically on the owning App, but Context itself is core-only.

Attributes:

Name Type Description
streams dict[str, Stream]

Every registered Stream, by name. What a widget looks its own stream up in.

bridges dict[str, Any]

Registered remote-target bridges, by name. Unlike streams these are not started for you — see App.bridges.

state str

The recording state machine: "idle" or "recording". Extensions add their own states (myogestic.ml adds "training"). Check it rather than tracking recording yourself.

session Session | None

The Session being written while state == "recording", else None. Set its name before App.stop_recording to label the take.

class_names list[str]

Names for the label class indices, mirrored here by the recording widgets so App.stop_recording can persist them in meta.json.

control_space Any

Optional myogestic.controls.ControlSet this app commands. Set once at setup (app.ctx.control_space = CONTROLS); every recording then stores the space it was made under.

current_label int

Class index a label click would record right now; -1 for rest / no class.

status_message str

One line of transient status, written by the recording lifecycle.

logs list[str]

The app-event lines LogPanel renders. Append via log, which timestamps and bounds them.

Examples:

>>> from myogestic import Context
>>> ctx = Context()
>>> ctx.status_message = "Ready"
>>> ctx.status_message
'Ready'

log

log(message: str, max_lines: int = 500) -> None

Append a one-line app event for the log_panel widget.

Bounded to max_lines (oldest dropped). Use for high-level events - recording saved, training start/done, model load - not per-frame chatter. Safe to call from any thread (list.append/pop are GIL-atomic).

Stream

Stream(name: str, source: Source, window_ms: float, buffer_ms: float = 10000, notch_hz: int = 0)

A named ring-buffered live stream backed by a Source.

Pair a name ("emg") with a source (LSLSource("TestEMG1")) and a window duration, register it with app.streams(...), and the rest of the framework addresses it by stream name.

  • Nothing attaches on its own. Call reconnect — or press the button in a StreamPanel — and the source is opened once. The acquire loop never opens one for you, on the first tick or after a source goes away, so a stream left running by an earlier process is never picked up behind you.
  • One daemon acquisition thread is started per Stream when App.run() begins. Once attached it loops source.read(), appends to the ring buffer and, if a recording session is active, appends to the session's Zarr store. Display and prediction consumers copy only the bounded tail they request. Until attached it waits.
  • get_window and get_display are then readable concurrently from other threads.
  • The ring buffer holds the last buffer_ms of samples so transient consumers (slow extract, momentary GUI hitches) don't lose data.

Examples:

>>> from myogestic import App, Stream
>>> from myogestic.sources import LSLSource
>>> app = App("hello")
>>> app.streams(
...     Stream("emg", source=LSLSource("TestEMG1"),
...            window_ms=1000, buffer_ms=10000),
... )

See Streams concept for the buffer + decimation model in depth, and Add a custom source for the matching source-side contract.

Live ring-buffered stream with display decimation.

Parameters:

Name Type Description Default
name str

Stream label (also used as the recorded zarr stream key).

required
source Source

Anything implementing the Source protocol.

required
window_ms float

Duration in milliseconds of the window returned by get_window.

required
buffer_ms float

Ring-buffer depth in milliseconds. Defaults to 10000 (10 s).

10000
notch_hz int

Mains frequency to notch out of the acquired signal — 50, 60, or 0 to leave it alone. Live-settable.

This conditions the samples themselves, so it reaches the model's windows and the recording, which is the point: train and predict then cannot see different preprocessing. It is not the signal viewer's Notch, which changes only what is drawn.

A recording made with this on holds filtered samples and the raw is not recoverable, so a session's setting is worth storing beside it — training on a mix of filtered and unfiltered takes is a silent inconsistency.

0

Methods:

Name Description
reconnect

Reconnect source. Optionally switch to a different target.

disconnect

Detach the source, leaving the acquire loop running and idle.

start

Start the acquisition loop (a daemon thread, or a per-frame task in the browser).

stop

Stop the acquisition loop and disconnect the source (errors suppressed).

attach_session

Begin recording this stream into session.

detach_session

Stop recording this stream (called by App.stop_recording).

get_window

Return the most recent window_ms as (data, ts).

get_display

M4-decimated display snapshot, computed on demand on the render thread.

get_raw_snapshot

Return an on-demand full, contiguous ring snapshot.

get_raw_snapshot_stable

Locked copy of the (tail of the) display snapshot, tagged with buffer identity.

last_timestamp

Most recent sample timestamp, or None if no samples yet.

reconnect

reconnect(target: str | None = None) -> bool

Reconnect source. Optionally switch to a different target.

Uses the source's own reconnect() if it has one (preserving source-specific logic like LSL resolve or serial port open), else disconnect + connect. Either way the source is connected ONCE, then buffers are (re)allocated from the returned StreamInfo.

One attempt at a time. A second caller while one is in flight is refused rather than queued: an app can offer more than one way to connect a stream (a device picker, a viewer's own button), and two attempts racing used to interleave — the queued one would wake up inside the lock and re-run against whatever source the first had since swapped in, reconnecting a live source out from under the buffers.

The source is connected outside self._lock. That lock is taken by the acquire loop and by every render-side read, while an OTB accept() blocks for accept_timeout — 30 s by default. Holding it across the attempt froze the whole GUI for as long as a device took to answer, or to not answer. It is now taken twice and briefly: once to mark the stream detached, once to publish the new buffers.

disconnect

disconnect() -> None

Detach the source, leaving the acquire loop running and idle.

The counterpart to reconnect. Not stop: that ends the acquire thread, which App.run starts once and owns — a stream stopped that way could not be brought back from the UI.

info is cleared along with the connection. A stream that was deliberately detached has no geometry, and leaving the old one behind makes a viewer report the connection as lost rather than as closed on purpose.

start

start() -> None

Start the acquisition loop (a daemon thread, or a per-frame task in the browser).

stop

stop() -> None

Stop the acquisition loop and disconnect the source (errors suppressed).

attach_session

attach_session(session: Session) -> None

Begin recording this stream into session.

Called by App.start_recording. Set under _session_lock so the acquire loop sees a fully-attached session atomically.

detach_session

detach_session() -> None

Stop recording this stream (called by App.stop_recording).

Waits for any append in flight on the acquire thread and blocks further ones, so once this returns the caller may finalise/clear the session's Zarr stores without racing the acquire loop.

get_window

get_window() -> tuple[ndarray, ndarray]

Return the most recent window_ms as (data, ts).

data is channels-first (n_channels, n_samples) and always float32, whatever dtype the stream buffers in. ts is a view into a reusable per-stream buffer; for a float32 stream data is a view too, otherwise a fresh float32 copy. Copy explicitly to retain either past the next call.

get_display

get_display(n_pixels: int = 800) -> tuple[ndarray, ndarray] | None

M4-decimated display snapshot, computed on demand on the render thread.

Recomputed per call from the current display buffer; the acquire thread must not precompute it (it starves the socket read at high channel counts).

get_raw_snapshot

get_raw_snapshot() -> tuple[ndarray, ndarray] | None

Return an on-demand full, contiguous ring snapshot.

Prefer :meth:get_raw_snapshot_stable with duration_s for live widgets; this compatibility API necessarily copies the whole buffer.

get_raw_snapshot_stable

get_raw_snapshot_stable(duration_s: float | None = None) -> tuple[int, int, float, ndarray, ndarray] | None

Locked copy of the (tail of the) display snapshot, tagged with buffer identity.

Like get_raw_snapshot but returns arrays the acquire thread cannot overwrite, for render-side consumers carrying state across frames (the incremental display notch): the copy so a concurrent buffer refresh cannot tear the samples being filtered (a torn read poisons the IIR state permanently), (epoch, end_seq) to tell new samples from seen ones and detect a reallocation, and fs so the rate matches the copied samples — reading stream.info.fs separately can race a reconnect.

duration_s copies only the newest duration_s seconds instead of the whole buffer (a 60 s buffer at 10 kHz is ~40 MB per frame). The trim uses the locked fs. end_seq stays absolute, so the returned data[i] has sequence end_seq - len(data) + i.

Returns (epoch, end_seq, fs, ts, data) or None if fewer than 2 samples are buffered (or the stream is not connected).

last_timestamp

last_timestamp() -> float | None

Most recent sample timestamp, or None if no samples yet.

Reads the ring tail under _lock, so a concurrent reconnect or append cannot strand the read on a torn buffer.

StreamInfo dataclass

StreamInfo(n_channels: int, fs: float, dtype: dtype = dtype(float32), channel_names: list[str] | None = None, channel_grids: list[ChannelGrid] | None = None)

Describes the shape and dtype of a Source's data.

Returned by Source.connect; sizes the ring buffer and lays out the signal viewer.

Attributes:

Name Type Description
n_channels int

Channel count. Fixed for the life of the source.

fs float

Sample rate in Hz. Used to convert window_ms / buffer_ms into sample counts.

dtype dtype

NumPy dtype of each sample, one of SUPPORTED_DTYPES. Defaults to float32. Accepts anything numpy.dtype understands ("int16", np.int16, np.dtype("int16")) and normalises it. A compact dtype (e.g. int16) keeps the ring buffer and Zarr recording small; the window handed to @pipeline.extract is always upcast to float32 regardless.

channel_names list[str] | None

Optional per-channel labels for the signal viewer legend. None (default) renders as ch0, ch1, ...

channel_grids list[ChannelGrid] | None

Optional list of ChannelGrid electrode topologies for the signal viewer's spatial channel selector. None (default) disables the grid selector. Not validated here — a malformed layout must never block acquisition; the viewer validates it before use.

Examples:

>>> from myogestic import StreamInfo
>>> info = StreamInfo(8, 2048.0, dtype="int16")
>>> (info.n_channels, info.fs, info.dtype.name)
(8, 2048.0, 'int16')

TrainingData dataclass

TrainingData(paths: list[str] = list(), class_names: list[str] = list(), classes: set[int] = set())

Inputs delivered to the user's @pipeline.train callback.

Built by session_manager() and assigned by the user to pipeline.training_data from inside @app.ui::

@app.ui
def ui(ctx):
    pipeline.training_data = session_manager(...)

Attributes:

Name Type Description
paths list[str]

Session locations (folders or .session.zip archives).

class_names list[str]

Human-readable labels — same list passed to recording_controls / session_manager.

classes set[int]

Active class indices to include. Pass as the classes= arg to iter_labeled_windows / iter_aligned_windows.

Examples:

>>> from myogestic import TrainingData
>>> data = TrainingData(
...     paths=["sessions/demo.session.zip"],
...     class_names=["Rest"],
...     classes={0},
... )
>>> data.is_empty
False

is_empty property

is_empty: bool

True when no session paths have been assigned yet.

Layout

Grid

Grid(rows: int, cols: int, row_height: list[Track] | None = None, col_width: list[Track] | None = None)

Grid layout manager. Index with [row, col] or [row, col_start:col_end].

Both axes accept the same Px/Fr track specs. See module docstring for examples.

Parameters:

Name Type Description Default
rows int

Number of rows.

required
cols int

Number of columns.

required
row_height list[Track] | None

Per-row track specs (length must equal rows). Default None → all rows share equally ([Fr(1)] * rows).

None
col_width list[Track] | None

Per-column track specs (length must equal cols). Default None → all columns share equally ([Fr(1)] * cols).

None

Raises:

Type Description
ValueError

if a list length doesn't match rows / cols, or if any track value is non-finite or negative.

TypeError

if a track entry isn't Px, Fr, or a number.

Examples:

>>> from myogestic import Fr, Grid, Px
>>> grid = Grid(2, 3, row_height=[Px(120), Fr(1)])
>>> (grid.rows, grid.cols)
(2, 3)

Px dataclass

Px(value: float)

Fixed pixel size. Px(300) means "exactly 300 px wide/tall".

Examples:

>>> from myogestic import Px
>>> Px(300)
Px(300)

Fr dataclass

Fr(value: float)

Fractional unit (CSS-grid fr).

Fr(1) means "1 share of the space remaining after Px tracks are subtracted". Multiple Fr entries split the remainder proportionally to their values, so [Fr(1), Fr(2)] splits leftover space 1:2.

Examples:

>>> from myogestic import Fr
>>> Fr(2)
Fr(2)

Track module-attribute

Track = Px | Fr

A single grid track size: either a fixed Px or a fractional Fr.

Examples:

>>> from myogestic.grid import Px, Track
>>> track: Track = Px(300)

Event helpers

EdgeTrigger

EdgeTrigger(callback: Callable[[T], None], *, n_stable_ticks: int = 1)

Calls callback(value) only when value differs from the last fire.

Parameters:

Name Type Description Default
callback Callable[[T], None]

Invoked with the new value when an edge fires.

required
n_stable_ticks int

Debounce: the new value must hold for this many consecutive fire_if_changed calls before firing. 1 (default) fires immediately on the first changed value — i.e. dedupe only. Use >1 to swallow tick-to-tick flicker (e.g. a classifier's argmax oscillating during a sliding-window transition) so the side effect isn't re-fired on every flip. It counts calls, not time — convert a duration with the loop rate: n_stable_ticks=math.ceil(seconds * predict_hz).

1
Notes

Thread-safety: the typical pattern is "one writer (predict thread) + occasional rebase() from the UI thread". The whole (last, candidate, count) state is held in one tuple replaced in a single assignment, so under CPython's GIL no lock is needed; a race between the two callers can at worst cost one extra suppressed-or-fired callback — harmless for the intended uses (RPC dedup, audio-cue gating, robot-movement commands).

Examples:

>>> from myogestic import EdgeTrigger
>>> fired = []
>>> trigger = EdgeTrigger(fired.append)
>>> [trigger.fire_if_changed(v) for v in ("Rest", "Rest", "Fist")]
[True, False, True]
>>> fired
['Rest', 'Fist']

last property

last: T | None

The most recently fired (or rebased) value; None before first fire.

fire_if_changed

fire_if_changed(value: T) -> bool

Fire the callback when value becomes a new, stable value.

Fires iff value differs from the last fired value and (when n_stable_ticks > 1) has held for n_stable_ticks consecutive calls. Returns True when the callback ran, False when suppressed.

rebase

rebase(value: T) -> None

Set the "last fired" value without firing.

Discards any pending debounce candidate. Use when another code path already performed the equivalent action; the next different value must then earn the full n_stable_ticks count, so a flicker candidate in progress can't complete on top of the manual one.

Built-in features

features

Classic time-domain EMG features — the starter set every example used to copy-paste.

Use as-is, mix with your own, or replace entirely::

from myogestic.recipes.features import rms, mav, wl
from myogestic.widgets import FeatureSelector

feats = FeatureSelector(
    {"RMS": rms, "MAV": mav, "WL": wl, "MyCustom": my_custom_fn},
    default=["RMS", "MAV"],
)

All take an EMG window of shape (n_channels, n_samples) and return a per-channel scalar vector (n_channels,) of dtype float32.

rms

rms(emg: ndarray) -> ndarray

Root mean square per channel.

Examples:

>>> import numpy as np
>>> from myogestic.recipes.features import rms
>>> emg = np.array([[1, -1], [2, -2]], dtype=np.float32)
>>> rms(emg).tolist()
[1.0, 2.0]

mav

mav(emg: ndarray) -> ndarray

Mean absolute value per channel.

Examples:

>>> import numpy as np
>>> from myogestic.recipes.features import mav
>>> emg = np.array([[1, -1], [2, -2]], dtype=np.float32)
>>> mav(emg).tolist()
[1.0, 2.0]

wl

wl(emg: ndarray) -> ndarray

Waveform length per channel — sum of absolute first differences.

Examples:

>>> import numpy as np
>>> from myogestic.recipes.features import wl
>>> emg = np.array([[1, -1], [2, -2]], dtype=np.float32)
>>> wl(emg).tolist()
[2.0, 4.0]

var

var(emg: ndarray) -> ndarray

Variance per channel.

Examples:

>>> import numpy as np
>>> from myogestic.recipes.features import var
>>> emg = np.array([[1, -1], [2, -2]], dtype=np.float32)
>>> var(emg).tolist()
[1.0, 4.0]

zc

zc(emg: ndarray) -> ndarray

Zero-crossing count per channel.

Examples:

>>> import numpy as np
>>> from myogestic.recipes.features import zc
>>> emg = np.array([[1, -1], [2, -2]], dtype=np.float32)
>>> zc(emg).tolist()
[1.0, 1.0]

External interfaces

virtual_hand

virtual_hand(godot_bin: str | None = None, vhi_path: str | None = None, grpc_host: str | None = None, grpc_port: int | None = None, launch_mode: str | None = None) -> InterfaceSpec

The MyoGestic Virtual Hand Interface (VHI).

Parameters:

Name Type Description Default
godot_bin str | None

Path to the Godot binary, for source-mode launch. Falls back to $GODOT_BIN, then which("godot4") / which("godot"), then platform GUI defaults.

None
vhi_path str | None

Directory containing VHI (binary install OR Godot project). Falls back to $VHI_PATH, then the default install root — <repo>/tools/MyoGestic-VHI in a git checkout, otherwise <user_data>/myogestic/vhi.

None
grpc_host str | None

VHI gRPC host. Falls back to $VHI_GRPC_HOST then 127.0.0.1.

None
grpc_port int | None

VHI gRPC port. Falls back to $VHI_GRPC_PORT then 50051.

None
launch_mode str | None

Launch mode — "binary", "godot", or "auto" (default). Also reads $VHI_LAUNCH_MODE. Explicit launch_mode always wins.

None

Returns:

Type Description
A `myogestic.remote.InterfaceSpec` with the resolved argv, ready to wire into
``process_launcher()``.

Examples:

>>> from myogestic.vhi import virtual_hand
>>> vhi = virtual_hand()
>>> vhi.n_output_channels
9

InterfaceSpec dataclass

InterfaceSpec(name: str, process: list[str], n_output_channels: int, output_hz: float, control_stream_name: str | None = None, n_control_channels: int | None = None, grpc_host: str = '127.0.0.1', grpc_port: int = 50051, install_root: Path | None = None, install_hint: str = '', version_gate: Callable[[], None] | None = None)

Description of a remote target — a separate process MyoGestic drives.

Attributes:

Name Type Description
name str

Human label, used as the process_launcher row title.

process list[str]

argv to spawn the target (passed to subprocess.Popen). An empty list means "nothing here launches it": either it is not installed, or you start it yourself. launcher() then raises, quoting install_hint.

n_output_channels int

Number of channels in the target's full pose vector — the width of the whole-pose read-back a recording consumes, not of a control's own stream.

output_hz float

Outlet send rate.

control_stream_name str | None

LSL inlet name the target publishes when the user drives it manually (used for regression targets). May be None.

n_control_channels int | None

Channel count of the control stream, if known.

grpc_host str

Host the target's gRPC control server listens on.

grpc_port int

Port the target's gRPC control server listens on.

install_root Path | None

The directory process was resolved from; quoted in the "not installed" error.

install_hint str

Appended to that error. How this target is installed is the target's own business — an installer command, an environment variable — and a generic spec has nothing useful to say about it.

version_gate Callable[[], None] | None

Called by launcher just before returning argv, to refuse an installed build this MyoGestic cannot drive. Whatever the check is, it is the target's: it reads a marker only that target leaves behind. None means there is nothing on disk to check, which is the honest default for a target MyoGestic did not install.

Examples:

>>> from myogestic.remote import InterfaceSpec
>>> spec = InterfaceSpec(
...     name="Hand",
...     process=["vhi"],
...     n_output_channels=9,
...     output_hz=32.0,
... )
>>> spec.launcher()
[('Hand', ['vhi'])]

stream_outlet

stream_outlet(name: str, *, n_channels: int | None = None) -> LSLOutlet

Construct an LSLOutlet publishing the target's stream called name.

The name is the target's, not this spec's. Which controls exist is in the manifest a running target answers with, and a streamed control's stream is named for that control's own address — so a stream is named where that answer is read (myogestic.remote.RemoteTarget, which calls this once per address it drives, after negotiation has settled), never guessed here.

Carries a stable source_id so a consumer can re-resolve this stream after a restart. Without one, LSL cannot tell a restarted outlet from a new stream and a consumer that resolved the old one keeps a dead inlet.

Parameters:

Name Type Description Default
name str

The stream's name, as the target's manifest reports it.

required
n_channels int | None

Width. ~myogestic.remote.RemoteTarget passes 1: one control per stream. The default is the target's full pose layout, for the whole-pose read-back a recording consumes.

None

Raises:

Type Description
ValueError

If name is empty. An unnamed LSL stream cannot be resolved, so a nameless control could never be found by the target that exports it.

control_client

control_client() -> RemoteClient

Construct a client for this target's control service.

Hand it to myogestic.remote.RemoteTarget and it asks the remote target which named DOFs it drives, refusing a configuration it cannot place. Required: without one there is nothing to negotiate the control space against.

Imported lazily — a plain install has no [grpc] extra, and stream_outlet / launcher must keep working without it.

Examples:

>>> from myogestic.controls import ControlBus, Continuous, ControlSet
>>> from myogestic.remote import RemoteTarget
>>> from myogestic.vhi import virtual_hand
>>>
>>> vhi = virtual_hand()
>>> controls = ControlSet(dofs={"my_index": Continuous("my_index")})
>>> target = RemoteTarget(client=vhi.control_client(), interface=vhi)
>>> bus = ControlBus(controls, targets=[target])
>>> target.negotiated     # True once the remote end has answered, False until then
False

recording_client

recording_client() -> RecordingClient

Construct a client for this target's recording session gate.

Not a control plane, and nothing it does is a control DOF. It carries the two things a recording session needs: the gate that stops the target's own local input competing as a movement source, and trajectories that cycle its control rig so the recorded kinematics sweep a continuous range.

Imported lazily, like the other gRPC client, so a plain install without the [grpc] extra can still use stream_outlet / launcher.

Examples:

>>> from myogestic.vhi import virtual_hand
>>> aid = virtual_hand().recording_client()
>>> aid.set_recording_session(True)   # False when the target is unreachable
False

launcher

launcher() -> list[tuple[str, list[str]]]

Return the (name, argv) tuple list expected by process_launcher.

Raises FileNotFoundError, quoting install_hint, when nothing can be launched from the resolved location. version_gate runs last, so an install this MyoGestic cannot drive is refused before the process starts rather than by every target at bind.

launchable

launchable() -> list[tuple[str, list[str]]]

Like launcher, but empty instead of raising when nothing can be launched.

For a myogestic.widgets.ProcessLauncher in an application's own UI, where an in-app Launch button is a convenience: launcher raising there takes the whole app down at import, even when a target is already running. This returns no rows and logs why instead. launcher stays strict for a caller whose entire job is to start the thing (tools/launch_vhi.py).

Examples:

>>> from myogestic.vhi import virtual_hand
>>> processes = virtual_hand().launchable()   # [] rather than an exception

Tools

control_outlet

control_outlet(name: str = DEFAULT_CONTROL_STREAM) -> StreamOutlet

LSL outlet for steering the EMG generator from another script.

The generator listens on a stream named name for a single float (channel = 1) that selects the next gesture amplitude — typically 0.0 (rest) … 1.0 (full). Push samples like::

from myogestic.tools.emg_generator import control_outlet
out = control_outlet()
out.push_sample(np.array([0.0], dtype=np.float32))  # rest
out.push_sample(np.array([1.0], dtype=np.float32))  # fist

Matches the protocol the --control flag on python -m myogestic.tools.emg_generator listens for.

Examples:

>>> import numpy as np
>>> from myogestic.tools.emg_generator import control_outlet
>>> outlet = control_outlet()
>>> outlet.push_sample(np.array([1.0], dtype=np.float32))

myogestic.tools.install_vhi

Install the Virtual Hand Interface release binary for this platform.

VHI ships pre-built artifacts on every release at https://github.com/NsquaredLab/MyoGestic-VHI/releases. This CLI picks the right asset for the host OS/arch, downloads it, unpacks it into the location virtual_hand() looks at, and drops a vhi-version.txt marker so a later install knows what's already there.

MyoGestic drives VHI over its control service, asking it what it exports. A release older than MIN_VHI_TAG has no control manifest to answer with, so this refuses to install one rather than leave it to fail at every launch.

Usage: python -m myogestic.tools.install_vhi # latest, default dest python -m myogestic.tools.install_vhi --tag v2.0.0 # pinned version python -m myogestic.tools.install_vhi --dest /custom/path python -m myogestic.tools.install_vhi --force # reinstall over existing

Or after pip install myogestic: myogestic-install-vhi

Pin --tag in production: latest is not reproducible, so a later rebuild may pick up a different VHI version.

main

main() -> None

Console-script entry point (myogestic-install-vhi).