API cheatsheet¶
The most-used public symbols, signatures only — hand-maintained and not exhaustive. Each name links to its full entry (prose, Args/Returns, source) in the API reference.
Core¶
App · Stream · Context · StreamInfo · Grid · TrainingData
App(name, theme=True, docking=False)
.streams(*streams)
.ui(fn) # decorator
.popout(title, gui_fn) # App(docking=True)
.start_recording(base_path="sessions")
.stop_recording()
.run(mode="gui" | "headless")
.ctx # Context
Stream(name, source, window_ms, buffer_ms=10000)
.start() / .stop() / .reconnect(target=None)
.get_window() -> (data, ts) # data is channels-first
.get_display(n_pixels) -> (env_min, env_max)
.get_raw_snapshot() -> (ts, data)
Context # flat dataclass
.streams: dict[str, Stream]
.state: str # "idle" / "recording" / ...
.session: Session | None # active recording
.class_names: list[str]
.status_message: str
Grid(rows, cols) # @app.ui layout helper
StreamInfo(n_channels, fs, dtype=float32, channel_names=None)
TrainingData(paths=[], class_names=[], classes=set())
.is_empty
Sources¶
LSLSource · ReplaySource · SerialSource
LSLSource(stream_name)
ReplaySource(session_path, stream_name, speed=1.0) # accepts .session.zip
SerialSource(port, baud, n_channels, fs) # extras:[serial]; from myogestic.sources.serial_source import SerialSource
Targets¶
Anything that moves is a target. Three methods, any transport.
Target · Capability · ControlBus · ControlLink · ControlLinkConnector · RemoteTarget · KeyboardTarget
# The protocol you implement — no base class, no registration.
bind(controls: ControlSet) -> None # main thread, may raise: refuse a bad map here
send(values, changed) -> None # predict thread, must not raise
stop() -> None # idempotent
claims: frozenset[str] # optional: which aliases you drive
capabilities() -> Sequence[Capability] # optional: what you export
Capability(address, kind, lo=-1.0, hi=1.0, rest=0.0, states=(), rest_state="")
load_control_map(mapping) -> ControlMap # a Mapping, not a path
resolve(control_map, capabilities) -> ControlSet # needs a live target's answer
ControlBus(controls, targets=[...], smoothing=None, hz=50, on_warn=None)
.push(values) -> dict .select(name, state) -> bool .stop() -> None
connect_controls(control_map, targets, ...) -> ControlBus | None # None while unreachable
ControlLinkConnector(link, retry_s=2.0).poll() -> bool # non-blocking UI retry
Outlets¶
LSLOutlet — a paced sender, for telemetry or as a target's
transport. It drives nothing on its own: no aliases, no range, no rest-on-stop.
ML pipeline¶
Pipeline · save_pickle · load_pickle
Pipeline(app, predict_hz=50)
@pipeline.extract(windows: dict[str, np.ndarray]) # channels-first
@pipeline.train(data: TrainingData)
@pipeline.predict(model, features) -> dict
.training_data: TrainingData | None
.start_training() / .start_predicting() / .stop_predicting()
.save_model / .load_model # set to enable buttons
.model / .predictions / .train_log
save_pickle(model, path) / load_pickle(path)
Session¶
open_session_store · iter_labeled_windows · iter_aligned_windows
open_session_store(path) -> Session # folder OR .session.zip
iter_labeled_windows(paths, stream_name, window_ms, hop_ms,
classes=None)
-> Iterator[(window, ts, class_index)] # window: (n_ch, n_samp)
iter_aligned_windows(paths, primary_stream_name, aligned_stream_names,
window_ms, hop_ms, n_alignment_samples=1)
-> Iterator[(primary_window, {name: vec}, ts)]
Filters¶
OneEuroFilter · GaussianFilter · IdentityFilter · make_filter
OneEuroFilter(hz=50.0, min_cutoff_hz=1.0, beta=0.02, derivative_cutoff_hz=1.0)
GaussianFilter(n_vectors=5, sigma=1.0)
IdentityFilter()
make_filter(name, hz=50.0, **kwargs) -> VectorFilter
filter(x, timestamp=None) -> np.ndarray # __call__ all filters
Widgets¶
Construct once (setup scope), then call .ui(...) each frame in @app.ui. Full list on the Widgets API page.
SignalViewer · RawSignalViewer · RecordingControls · SessionManager · Heatmap · LinePlot · PostProcessor
SignalViewer(stream_name, selectable=False, scale_mode="auto",
y_range=(-1, 1), show_diagnostics=False, show_markers=False)
.ui(ctx)
RawSignalViewer(stream_name) # every visible sample, bounded copy
.ui(ctx)
RecordingControls(class_names, *, on_record, on_stop, on_gesture=None)
.ui(ctx)
SessionManager(base_path, title="Sessions", class_names=None)
.ui() -> TrainingData(paths, class_names, classes)
ProcessLauncher(processes)
.ui()
Scatter2D(label) / Scatter3D(label)
.ui(points, labels=None, class_names=None)
Heatmap(label, label_fmt="%.1f") / LinePlot(label)
.ui(data) # LinePlot: .ui(data, channel_names=None)
panel_header(title, icon=None)
output_filter = PostProcessor(hz=50)
output_filter.ui()
popout_panel(title, gui_fn)
TemplateInspector(widget_id)
.ui(rows)
TrialPreview(widget_id=widget_id)
.ui(data, fs)
ML widgets¶
TrainButton · PredictButton · TrainingLog · SaveModelButton · LoadModelButton · PipelinePanel