ML pipeline

Pipeline

Pipeline

Pipeline(app: App, predict_hz: float = 50.0)

ML lifecycle + state for an App.

Constructor registers the predict thread + cleanup on the App's hook lists; they fire on app.run() start/exit. Decorators set the callbacks. Transition methods flip app.ctx.state.

Parameters:

Name	Type	Description	Default
app	App	The myogestic App.	required
predict_hz	float	Maximum predict-loop tick rate. Set to 0 or negative to remove the cap (run at full speed).	50.0

Examples:

>>> from myogestic import App
>>> from myogestic.ml import Pipeline
>>> pipeline = Pipeline(App("EMG demo"), predict_hz=20)
>>> @pipeline.extract
... def extract(windows):
...     return windows["emg"].mean(axis=1)

Methods:

Name	Description
extract	Decorator: register the feature-extraction callback.
train	Decorator: register the training callback.
predict	Decorator: register the predict callback.
start_training	Run the @pipeline.train callback on a worker thread.
start_predicting	Flip the state to predicting so the predict thread runs.
stop_predicting	Return to idle, pausing the predict loop.

extract

extract(fn: Callable) -> Callable

Decorator: register the feature-extraction callback.

The wrapped function receives windows: dict[str, np.ndarray] keyed by stream name — each array is channels-first (n_channels, n_samples). Return whatever shape your model wants to consume. The same function is invoked from inside train() (over recorded windows) and on the predict thread (over live windows), so keep its return type stable.

train

train(fn: Callable) -> Callable

Decorator: register the training callback.

The wrapped function receives one TrainingData and must return any object — it's stored on pipeline.model and forwarded to every subsequent predict() call. If pipeline.save_model is set, the Save Model button calls it as save_model(pipeline.model, path).

predict

predict(fn: Callable) -> Callable

Decorator: register the predict callback.

The wrapped function is called every 1/predict_hz seconds with (model, features) where features is the return value of the extract callback. Must return a dict[str, Any] — non-dict returns are silently dropped (the previous prediction stays in pipeline.predictions).

start_training

start_training() -> None

Run the @pipeline.train callback on a worker thread.

No-op (sets ctx.status_message) unless the state is idle, a train callback is registered, and non-empty training_data is set. Flips the state to training for the duration and stores the returned object on model.

start_predicting

start_predicting() -> None

Flip the state to predicting so the predict thread runs.

No-op (sets ctx.status_message) unless the state is idle and a model is loaded.

stop_predicting

stop_predicting() -> None

Return to idle, pausing the predict loop.

No-op (sets ctx.status_message) unless the state is currently predicting.

PipelineState

Bases: StrEnum

ML-side extension of AppState.

The core app only knows about "idle" and "recording"; attaching a Pipeline (via Pipeline(app)) adds two more states for the ML lifecycle. Mutually exclusive with each other and with the core states.

The enum is a StrEnum so it compares cleanly against the raw string written to app.ctx.state by the transition methods.

Attributes:

Name	Type	Description
TRAINING		train() is running on a background thread. Predict ticks short-circuit so they don't fight for GPU.
PREDICTING		The predict thread is calling extract + predict each tick at predict_hz and writing the result to pipeline.predictions.

Examples:

>>> from myogestic.ml import PipelineState
>>> PipelineState.PREDICTING.value
'predicting'

Persistence

save_pickle

save_pickle(model: Any, path: str | Path, *, controls: ControlMap | None = None) -> str

Persist model to path via joblib, creating parent dirs as needed.

Returns the path as a string.

Parameters:

Name	Type	Description	Default
model	Any	Any picklable object.	required
path	str | Path	Destination file.	required
controls	ControlMap | None	Optional myogestic.controls.ControlMap the model was trained against, written to a <path>.controls.json sidecar. A model is only meaningful in the output space it was fitted for: loading one trained on a one-way [0, 1] DOF against a signed [-1, 1] configuration produces motion in a direction the model never learned, and nothing in the artifact itself would say so. A sidecar keeps load_pickle's signature and leaves older artifacts loadable.	None

Examples:

>>> from myogestic.ml import save_pickle
>>> save_pickle({"classes": 2}, "models/demo.joblib")
'models/demo.joblib'

load_pickle

load_pickle(path: str | Path, *, controls: ControlMap | None = None, allow_unverified: bool = False) -> Any

Inverse of save_pickle — load a joblib-saved model.

Parameters:

Name	Type	Description	Default
path	str | Path	The model file.	required
controls	ControlMap | None	Optional myogestic.controls.ControlMap to check the model against. When given, the model's sidecar must describe the same control space, or this raises rather than driving a target through a space the model never saw. Wire it in one line:: pipeline.load_model = partial(load_pickle, controls=CONTROLS)	None
allow_unverified	bool	Permit loading a model that carries no sidecar even though controls was supplied. Off by default: an artifact saved before provenance existed cannot be distinguished from one trained in a different space, and silently accepting it is how a polarity change ships.	False

Raises:

Type	Description
ValueError	If the sidecar disagrees with controls, or is absent without allow_unverified.

Examples:

>>> from myogestic.ml import load_pickle
>>> model = load_pickle("models/demo.joblib")
>>> model["classes"]
2

Widgets

TrainButton

TrainButton(pipeline: Pipeline, *, size: tuple[float, float] = (92, 0))

Train button — calls Pipeline.start_training on click.

Examples:

>>> from myogestic.ml.widgets import TrainButton
>>> button = TrainButton(pipeline)
>>> button.ui()

ui

ui() -> None

Render the Train button. Call once per frame inside @app.ui.

PredictButton

PredictButton(pipeline: Pipeline, *, size: tuple[float, float] = (92, 0))

Predict/Stop toggle reflecting the pipeline's predict state.

Enabled to start only when the state is idle, a model is loaded, and both the extract and predict callbacks are wired; shows a Stop button while predicting and is disabled otherwise.

Examples:

>>> from myogestic.ml.widgets import PredictButton
>>> button = PredictButton(pipeline)
>>> button.ui()

ui

ui() -> None

Render the Predict/Stop button. Call once per frame.

TrainingLog

TrainingLog(pipeline: Pipeline, *, height: float = 100.0, widget_id: str = 'ml')

Read-only view of pipeline.train_log.

The popout toggle isn't drawn here — it lives on PipelinePanel's control row, next to Train/Predict.

Examples:

>>> from myogestic.ml.widgets import TrainingLog
>>> log = TrainingLog(pipeline)
>>> log.ui()

ui

ui() -> None

Render the training log. Call once per frame.

SaveModelButton

SaveModelButton(pipeline: Pipeline, path: str, *, size: tuple[float, float] = (92, 0))

Save button — writes the model to path via pipeline.save_model.

Disabled unless both pipeline.save_model and pipeline.model are set.

Examples:

>>> from myogestic.ml.widgets import SaveModelButton
>>> button = SaveModelButton(pipeline, "models/demo.joblib")
>>> button.ui()

ui

ui() -> None

Render the Save button. Call once per frame.

LoadModelButton

LoadModelButton(pipeline: Pipeline, path: str, *, size: tuple[float, float] = (92, 0))

Load button — reads a model from path via pipeline.load_model.

Disabled unless pipeline.load_model is set.

Examples:

>>> from myogestic.ml.widgets import LoadModelButton
>>> button = LoadModelButton(pipeline, "models/demo.joblib")
>>> button.ui()

ui

ui() -> None

Render the Load button. Call once per frame.

---

PipelinePanel

PipelinePanel(pipeline: Pipeline, *, log_height: float = 0.0, widget_id: str = 'ml')

Train + Predict + log as a single titled panel.

Matches the visual style of RecordingControls, SessionManager, and PostProcessor.

The log lives in the ↗ popout, the same way ProcessLauncher's does: a window you can move, resize and leave open. Nothing is drawn inline unless you ask for a height — the panel used to fill its whole cell with a box holding one line of output.

Parameters:

Name	Type	Description	Default
pipeline	Pipeline	The Pipeline to train and predict with.	required
log_height	float	Height in pixels of an optional inline log. <= 0 (the default) draws none.	0.0
widget_id	str	Unique ID for this panel, so two of them keep separate popout state.	'ml'

Examples:

>>> from myogestic.ml.widgets import PipelinePanel
>>> panel = PipelinePanel(pipeline)
>>> panel.ui()

ui

ui() -> None

Render the pipeline panel. Call once per frame inside @app.ui.