ML 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`

Source code in myogestic/ml/__init__.py

def __init__(self, app: App, predict_hz: float = 50.0):
    """
    Args:
        app: The myogestic App.
        predict_hz: Maximum predict-loop tick rate. Set to 0 or
            negative to remove the cap (run at full speed).
    """
    self.app = app
    self.predict_hz = predict_hz
    self.model: Any = None
    self.predictions: dict[str, Any] = {}
    self.train_log: list[str] = []
    self.on_extract: Callable | None = None
    self.on_train: Callable | None = None
    self.on_predict: Callable | None = None
    # Set if you want save/load buttons to do anything; the
    # `myogestic.models.{save,load}_model` joblib helpers are the
    # obvious default but the library doesn't force them.
    self.save_model: Callable | None = None
    self.load_model: Callable | None = None
    # Set this from inside `@app.ui` to publish what the user picked
    # in `session_manager(...)` to `@pipeline.train`.
    self.training_data: TrainingData | None = None
    self._stop = threading.Event()
    self._thread: threading.Thread | None = None

    app.before_run_hooks.append(self._start_predict_thread)
    app.cleanup_hooks.append(self._cleanup)

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.

Source code in myogestic/ml/__init__.py

def extract(self, 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.
    """
    self.on_extract = fn
    return fn

train ¶

train(fn: Callable) -> Callable

Decorator: register the training callback.

The wrapped function receives one :class: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).

Source code in myogestic/ml/__init__.py

def train(self, fn: Callable) -> Callable:
    """Decorator: register the training callback.

    The wrapped function receives one :class:`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)``.
    """
    self.on_train = fn
    return fn

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).

Source code in myogestic/ml/__init__.py

def predict(self, 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``).
    """
    self.on_predict = fn
    return fn

PipelineState ¶

Bases: StrEnum

ML-side extension of :class:~myogestic.AppState.

The core app only knows about "idle" and "recording"; attaching a :class:Pipeline (via Pipeline(app)) adds two more states for the ML lifecycle. Mutually exclusive with each other and with the core states: a Pipeline cannot be predicting and training at the same time, by design (the train pause exists so the GPU isn't fought over).

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

Members

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.

save_pickle ¶

save_pickle(model: Any, path: str | Path) -> str

Pickle model to path. Creates parent directories if needed.

Source code in myogestic/ml/persistence.py

def save_pickle(model: Any, path: str | Path) -> str:
    """Pickle ``model`` to ``path``. Creates parent directories if needed."""
    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    with p.open("wb") as f:
        pickle.dump(model, f)
    return str(p)

load_pickle ¶

load_pickle(path: str | Path) -> Any

Inverse of save_pickle.

Source code in myogestic/ml/persistence.py

def load_pickle(path: str | Path) -> Any:
    """Inverse of ``save_pickle``."""
    with Path(path).open("rb") as f:
        return pickle.load(f)