Skip to content

Widgets

Stateless function widgets you call from inside @app.ui. See Widgets concept page for the contract and the widget gallery for a visual index of all of them on one page.


signal_viewer

signal_viewer(ctx: Context, stream_name: str, size: tuple[float, float] = (-1, -1), n_pixels: int = 2000, channel_height: float = 0.0, show_diagnostics: bool = False, selectable: bool = False, scale_mode: str = 'auto', y_range: tuple[float, float] = (-1.0, 1.0), show_markers: bool = False, window_seconds: float = 5.0) -> None

Real-time multi-channel signal viewer.

Includes decimation, pause, auto/manual Y scale, visual-only display filters, channel toggles, stats, stream retargeting, and label markers. The function argument stream_name is the stable widget ID; when selectable=True, the user may switch the active stream from the UI.

scale_mode supports "auto" for ImPlot fitting and "manual" for the user-set y_range.

window_seconds sets the initial display window in seconds — the user can still drag the slider afterwards. Defaults to 5 s, which is wide enough to scan visually across most real-time setups. Pass a smaller value when you want the display to mirror a short analysis window (classification often runs at 0.2 s, for example). The stream's buffer_seconds must be at least this large.

Source code in myogestic/widgets/signal.py
def signal_viewer(
    ctx: Context,
    stream_name: str,
    size: tuple[float, float] = (-1, -1),
    n_pixels: int = 2000,
    channel_height: float = 0.0,
    show_diagnostics: bool = False,
    selectable: bool = False,
    scale_mode: str = "auto",
    y_range: tuple[float, float] = (-1.0, 1.0),
    show_markers: bool = False,
    window_seconds: float = 5.0,
) -> None:
    """Real-time multi-channel signal viewer.

    Includes decimation, pause, auto/manual Y scale, visual-only display
    filters, channel toggles, stats, stream retargeting, and label markers.
    The function argument `stream_name` is the stable widget ID; when
    `selectable=True`, the user may switch the active stream from the UI.

    `scale_mode` supports "auto" for ImPlot fitting and "manual" for the
    user-set `y_range`.

    `window_seconds` sets the *initial* display window in seconds — the user
    can still drag the slider afterwards. Defaults to 5 s, which is wide
    enough to scan visually across most real-time setups. Pass a smaller
    value when you want the display to mirror a short analysis window
    (classification often runs at 0.2 s, for example). The stream's
    ``buffer_seconds`` must be at least this large.
    """
    v = get_viewer_state(
        ctx,
        stream_name,
        n_pixels=n_pixels,
        scale_mode=scale_mode,
        y_range=y_range,
        show_markers=show_markers,
        window_seconds=window_seconds,
    )
    active_stream = v.selected_stream or stream_name
    stream = ctx.streams.get(active_stream)

    panel_header(f"SIGNAL · {active_stream}", fa.ICON_FA_CHART_LINE)
    if stream is None:
        imgui.text(f"{active_stream}: not found")
        return
    if stream.status != "connected" or stream.info is None:
        _disconnected_ui(active_stream, stream)
        return

    render_controls(ctx, stream_name, active_stream, stream, v, selectable)

    frame = build_signal_frame(stream, v)
    if frame is None:
        imgui.text(f"{active_stream}: no data")
        return

    enabled, ch_names, hovered_ch = render_channel_controls(
        stream_name, stream, v, frame.n_channels
    )
    if not enabled:
        imgui.text("No channels enabled")
        return

    if v.display_filter == "rms_env":
        data = frame.data
    else:
        data = apply_display_filter(frame.data, v.display_filter, stream.info.fs)

    channel_ranges = None
    if v.per_channel_scale:
        full_data = apply_display_filter(
            frame.data_full, v.display_filter, stream.info.fs
        )
        channel_ranges = _channel_ranges(full_data, enabled)

    # Honour a "Rescale" button click from the controls bar: snap y_min /
    # y_max to the current visible data range across enabled channels,
    # then switch to Manual so it stays put.
    if v.rescale_pending:
        v.rescale_pending = False
        ranges = _channel_ranges(data, enabled)
        if ranges:
            mins = [lo for lo, _ in ranges.values()]
            maxs = [hi for _, hi in ranges.values()]
            lo = min(mins)
            hi = max(maxs)
            span = hi - lo
            pad = span * 0.1 if span > 0 else 1.0
            v.y_min = lo - pad
            v.y_max = hi + pad
            v.scale_mode = "manual"

    render_plot(
        ctx=ctx,
        stream_name=stream_name,
        stream=stream,
        v=v,
        frame=frame,
        data=data,
        channel_ranges=channel_ranges,
        enabled=enabled,
        ch_names=ch_names,
        hovered_ch=hovered_ch,
        size=size,
        channel_height=channel_height,
    )
    render_footer(
        stream_name=stream_name,
        stream=stream,
        v=v,
        frame=frame,
        enabled=enabled,
        ch_names=ch_names,
        show_diagnostics=show_diagnostics,
    )

signal_viewer


raw_signal_viewer

raw_signal_viewer(ctx: Context, stream_name: str, size: tuple[float, float] = (-1, 300), channel_height: float = 0.0) -> None

Raw signal viewer — every sample, no decimation, zero-alloc render path.

Source code in myogestic/widgets/raw_signal.py
def raw_signal_viewer(
    ctx: Context,
    stream_name: str,
    size: tuple[float, float] = (-1, 300),
    channel_height: float = 0.0,
) -> None:
    """Raw signal viewer — every sample, no decimation, zero-alloc render path."""
    stream = ctx.streams.get(stream_name)
    if stream is None:
        imgui.text(f"{stream_name}: not found")
        return
    if stream.status != "connected" or stream.info is None:
        _disconnected_ui(stream_name, stream)
        return

    r = _raw_viewers.get(stream_name)
    if r is None:
        r = _RawViewerState(window=stream._window)
        _raw_viewers[stream_name] = r

    t_start = _time.perf_counter()

    changed, new_win = imgui.slider_float(
        f"Window (s)##{stream_name}_raw_win", r.window, 0.1, 60.0, "%.1f s",
    )
    if changed:
        r.window = new_win

    snapshot = stream.get_raw_snapshot()
    if snapshot is None:
        imgui.text(f"{stream_name}: no data")
        return
    all_ts, all_data = snapshot
    n_win = int(r.window * stream.info.fs)
    if len(all_data) > n_win:
        data = all_data[-n_win:]
        ts = all_ts[-n_win:]
    else:
        data = all_data
        ts = all_ts

    n_samples, n_channels = data.shape

    if not r.channels_initialized or max(r.channels, default=-1) >= n_channels:
        r.channels = set(range(n_channels))
        r.specs = []
        r.bufs = {}  # invalidate per-channel ys buffers
        r.channels_initialized = True
    enabled = r.channels
    ch_names = stream.info.channel_names if stream.info else None

    for ch in range(n_channels):
        if ch > 0:
            imgui.same_line()
            if imgui.get_content_region_avail().x < 80:
                imgui.new_line()
        is_on = ch in enabled
        color = PALETTE[ch % len(PALETTE)]
        if is_on:
            imgui.push_style_color(
                imgui.Col_.button, imgui.ImVec4(color[0], color[1], color[2], 0.7)
            )
        else:
            imgui.push_style_color(imgui.Col_.button, imgui.ImVec4(0.3, 0.3, 0.3, 0.5))
        label = ch_names[ch] if ch_names and ch < len(ch_names) else f"ch{ch}"
        if imgui.button(f"{label}##{stream_name}_rawtog{ch}"):
            if is_on:
                enabled.discard(ch)
            else:
                enabled.add(ch)
        imgui.pop_style_color()

    if not enabled:
        imgui.text("No channels enabled")
        return

    bufs = r.bufs
    if not bufs or bufs.get("cap", 0) < n_samples:
        cap = n_samples + 1024
        bufs = {
            "cap": cap,
            "xs": np.empty(cap, dtype=np.float64),
            "ys": {ch: np.empty(cap, dtype=np.float64) for ch in range(n_channels)},
        }
        r.bufs = bufs

    xs = bufs["xs"]
    np.subtract(ts, ts[0], out=xs[:n_samples])
    xs_view = xs[:n_samples]

    if channel_height <= 0:
        d_min, d_max = np.inf, -np.inf
        for ch in enabled:
            d_min = min(d_min, float(np.min(data[:, ch])))
            d_max = max(d_max, float(np.max(data[:, ch])))
        data_range = d_max - d_min
        channel_height = data_range * 1.2 if data_range > 0 else 1.0

    if len(r.specs) < n_channels:
        r.specs = []
        for ch in range(n_channels):
            c = PALETTE[ch % len(PALETTE)]
            s = implot.Spec()
            s.line_color = imgui.ImVec4(c[0], c[1], c[2], 0.9)
            s.line_weight = 1.0
            r.specs.append(s)
    specs = r.specs

    if implot.begin_plot(f"{stream_name}##{stream_name}_raw", imgui.ImVec2(size[0], size[1])):
        plot_idx = 0
        for ch in sorted(enabled):
            offset = -plot_idx * channel_height
            if ch not in bufs["ys"]:
                bufs["ys"][ch] = np.empty(bufs["cap"], dtype=np.float64)
            ys = bufs["ys"][ch]
            np.add(data[:, ch], offset, out=ys[:n_samples])
            label = ch_names[ch] if ch_names and ch < len(ch_names) else f"ch{ch}"
            implot.plot_line(f"{label}##{stream_name}_raw", xs_view, ys[:n_samples], specs[ch])
            plot_idx += 1
        implot.end_plot()

    frame_dt = _time.perf_counter() - t_start
    r.fps.append(frame_dt)
    if len(r.fps) > 60:
        r.fps.pop(0)
    avg_ms = np.mean(r.fps) * 1000
    fps = 1000.0 / avg_ms if avg_ms > 0 else 0
    imgui.text_colored(
        imgui.ImVec4(0.5, 0.5, 0.5, 1.0),
        f"{fps:.0f} fps ({avg_ms:.1f} ms) | "
        f"fs={stream.info.fs:.0f} Hz | "
        f"{len(enabled)}/{n_channels} ch | "
        f"{n_samples} pts/ch (raw)",
    )

recording_controls

recording_controls(ctx: Context, class_names: list[str] | None = None, *, on_record: Callable[[], None], on_stop: Callable[[], None], on_gesture: Callable[[int], None] | None = None) -> None

Record/Stop + per-class label buttons + state pill.

The widget reads ctx and drives recording via the explicit callbacks — it does not import App. Pass app.start_recording / app.stop_recording if you're using the standard App.

Clicking a class button while recording snaps a label event at that moment (the active class is shown in the "Recording into: …" header). Outside of recording it just sets the next class to be used when Record is clicked.

Parameters:

Name Type Description Default
ctx Context

myogestic Context. Mutated: current_label is clamped to a valid index for class_names, and class_names itself is mirrored into ctx.class_names so App.stop_recording can persist them in the session metadata.

required
class_names list[str] | None

Per-class label-button names.

None
on_record Callable[[], None]

Called when Record is clicked (idle → recording).

required
on_stop Callable[[], None]

Called when Stop is clicked (recording → idle).

required
on_gesture Callable[[int], None] | None

Optional (class_index) -> None for side effects on label-button click (e.g. switching a fake-signal generator).

None
Source code in myogestic/widgets/recording.py
def recording_controls(
    ctx: Context,
    class_names: list[str] | None = None,
    *,
    on_record: Callable[[], None],
    on_stop: Callable[[], None],
    on_gesture: Callable[[int], None] | None = None,
) -> None:
    """Record/Stop + per-class label buttons + state pill.

    The widget reads `ctx` and drives recording via the explicit callbacks —
    it does not import App. Pass `app.start_recording` / `app.stop_recording`
    if you're using the standard App.

    Clicking a class button while recording snaps a label event at that moment
    (the active class is shown in the "Recording into: …" header). Outside of
    recording it just sets the *next* class to be used when Record is clicked.

    Args:
        ctx: myogestic Context. Mutated: `current_label` is clamped to a
            valid index for `class_names`, and `class_names` itself is
            mirrored into `ctx.class_names` so `App.stop_recording` can
            persist them in the session metadata.
        class_names: Per-class label-button names.
        on_record: Called when Record is clicked (idle → recording).
        on_stop:   Called when Stop is clicked (recording → idle).
        on_gesture: Optional `(class_index) -> None` for side effects on
                    label-button click (e.g. switching a fake-signal generator).
    """
    n_classes = len(class_names) if class_names else 0
    # Defensive: clamp stale current_label to a safe range. Users can swap
    # CLASSES between runs — a leftover index could silently corrupt labels.
    if class_names:
        ctx.current_label = _safe_label_index(ctx.current_label, n_classes)
        # Mirror class_names into ctx so save_meta() can persist them in the
        # session's meta.json without the App needing to know about CLASSES.
        ctx.class_names = list(class_names)

    panel_header("RECORDING", fa.ICON_FA_CIRCLE_DOT)

    # Above the buttons: hint of what they do
    if class_names:
        imgui.text("Gesture:")

    # Per-class label buttons — selects current class; while recording also snaps
    # try/finally so an exception raised by on_record / on_stop /
    # on_gesture (or by their downstream effects, e.g. zarr init in a
    # threadless runtime) doesn't leave the ImGui style stack unbalanced,
    # which would then trip an IM_ASSERT on the next end_child further
    # up the call chain.
    imgui.push_style_var(imgui.StyleVar_.frame_padding, imgui.ImVec2(12, 8))
    try:
        if class_names:
            for i, name in enumerate(class_names):
                if i > 0:
                    imgui.same_line()
                selected = ctx.current_label == i
                if selected:
                    imgui.push_style_color(imgui.Col_.button, imgui.ImVec4(0.31, 0.61, 0.98, 0.9))
                try:
                    if imgui.button(f"{name}##rec_gesture{i}", imgui.ImVec2(_LABEL_BTN_W, _LABEL_BTN_H)):
                        ctx.current_label = i
                        if on_gesture is not None:
                            on_gesture(i)
                        if (
                            ctx.state == AppState.RECORDING
                            and ctx.session is not None
                            and 0 <= i < n_classes
                        ):
                            ctx.session.add_label(i)
                finally:
                    if selected:
                        imgui.pop_style_color()
            imgui.spacing()

        # Record / Stop
        if ctx.state == AppState.IDLE:
            if imgui.button(f"{fa.ICON_FA_CIRCLE}  Record##rec_btn", imgui.ImVec2(_RECORD_BTN_W, 0)):
                on_record()
                # Auto-add the current label at the start of the recording, but
                # only if it's a valid index for the current class_names.
                if (
                    ctx.session is not None
                    and class_names
                    and 0 <= ctx.current_label < n_classes
                ):
                    ctx.session.add_label(ctx.current_label)
        elif ctx.state == AppState.RECORDING and imgui.button(
            f"{fa.ICON_FA_STOP}  Stop##rec_btn", imgui.ImVec2(_RECORD_BTN_W, 0)
        ):
            on_stop()
    finally:
        imgui.pop_style_var()

    # Status line — single state pill + status message + (when recording) the
    # snap affordance hint so the user knows clicking a class label snaps it.
    imgui.spacing()
    color = STATE_COLORS.get(ctx.state, _DEFAULT_COLOR)
    _status_pill(ctx.state.upper(), color)
    imgui.same_line()
    message = ctx.status_message or "Ready"
    if message.startswith("Saved"):
        message = f"{fa.ICON_FA_FLOPPY_DISK}  {message}"
    if ctx.state == AppState.RECORDING:
        n_labels = len(ctx.session.label_track) if ctx.session else 0
        active_name = (
            class_names[ctx.current_label]
            if class_names and 0 <= ctx.current_label < n_classes
            else "—"
        )
        message = f"{n_labels} labels · into: {active_name} (click a class to snap)"
    # Vertically center message with the pill (the pill is taller than a text
    # line by 2*_PILL_PAD_Y, so nudge text down by the pad to share a baseline).
    # Don't restore the cursor afterwards — moving cursor without submitting an
    # item next breaks imgui's window-growth assertion in end_child.
    imgui.set_cursor_pos_y(imgui.get_cursor_pos_y() + _PILL_PAD_Y)
    imgui.text(message)

recording_controls


session_manager

Session manager widget: load/select sessions and class filters for training.

add_recorded_session

add_recorded_session(path: str, base_path: str = 'sessions', label: str = 'Sessions') -> None

Register a freshly recorded session as selected.

Source code in myogestic/widgets/_session_manager_state.py
def add_recorded_session(path: str, base_path: str = "sessions", label: str = "Sessions") -> None:
    """Register a freshly recorded session as selected."""
    uid = f"{label}_{base_path}"
    state = get_state(uid)
    if any(s["path"] == path for s in state.sessions):
        return
    for row in scan_sessions(str(Path(path).parent)):
        if row["path"] == path:
            row["selected"] = True
            state.sessions.insert(0, row)
            break

session_manager

session_manager(base_path: str = 'sessions', label: str = 'Sessions', class_names: list[str] | None = None) -> TrainingData

Session picker widget. Returns TrainingData(paths, class_names, classes).

The widget has two training filters: selected session files and selected class indices. Session scanning/state lives in _session_manager_state.py. Assign the returned value to pipeline.training_data to make it visible to @pipeline.train::

@app.ui
def ui(ctx):
    pipeline.training_data = session_manager(...)
Source code in myogestic/widgets/session_manager.py
def session_manager(
    base_path: str = "sessions",
    label: str = "Sessions",
    class_names: list[str] | None = None,
) -> TrainingData:
    """Session picker widget. Returns ``TrainingData(paths, class_names, classes)``.

    The widget has two training filters: selected session files and selected
    class indices. Session scanning/state lives in
    ``_session_manager_state.py``. Assign the returned value to
    ``pipeline.training_data`` to make it visible to ``@pipeline.train``::

        @app.ui
        def ui(ctx):
            pipeline.training_data = session_manager(...)
    """
    uid = f"{label}_{base_path}"
    state = get_state(uid)

    panel_header(label, fa.ICON_FA_FOLDER_OPEN)
    render_summary_and_buttons(uid, base_path, state)
    poll_file_dialog(state)

    classes_in_pool, active_classes = class_pool_and_active(state)
    render_class_buttons(uid, state.deactivated_classes, classes_in_pool, active_classes, class_names)
    render_session_rows(uid, state.sessions, class_names)

    return TrainingData(
        paths=[s["path"] for s in state.sessions if s["selected"]],
        class_names=list(class_names) if class_names else [],
        classes=set(active_classes) if classes_in_pool else set(),
    )

session_manager


process_launcher

Process launcher widget for @app.ui.

Usage

from myogestic.proc import process_launcher

PROCESSES = [ ("8ch EMG", ["mne_lsl_player", "--n_channels", "8", "--fs", "256"]), ("Webcam", [sys.executable, "-m", "myogestic.bridges.webcam", ...]), ]

@app.ui def my_ui(ctx): process_launcher(PROCESSES)

process_launcher

process_launcher(processes: list[Process], label: str = '', log_height: float = -1.0) -> None

Dropdown + Launch/Stop + scrollable log panel.

Multiple process_launcher() calls can coexist in the same UI — each gets unique ImGui IDs via the label parameter.

Parameters:

Name Type Description Default
processes list[Process]

List of (name, command) tuples.

required
label str

Unique ID for this launcher instance. Auto-generated if empty.

''
log_height float

Height of the inline log panel in pixels. Pass <= 0 (default) to fill the remaining vertical space of the parent cell — matches the ImGui convention where -1 means "fill available". When the log is popped out (see button), the inline log is replaced by a placeholder and the full log is rendered in a separate floating ImGui window.

-1.0
Source code in myogestic/widgets/process_launcher.py
def process_launcher(
    processes: list[Process],
    label: str = "",
    log_height: float = -1.0,
) -> None:
    """Dropdown + Launch/Stop + scrollable log panel.

    Multiple process_launcher() calls can coexist in the same UI —
    each gets unique ImGui IDs via the label parameter.

    Args:
        processes: List of (name, command) tuples.
        label: Unique ID for this launcher instance. Auto-generated if empty.
        log_height: Height of the inline log panel in pixels. Pass ``<= 0``
            (default) to fill the remaining vertical space of the parent
            cell — matches the ImGui convention where ``-1`` means "fill
            available". When the log is popped out (see ``↗`` button), the
            inline log is replaced by a placeholder and the full log is
            rendered in a separate floating ImGui window.
    """
    if not processes:
        return

    # Auto-generate label from process names
    uid = label or "_".join(n for n, _ in processes)

    # Ensure state exists for all processes (keyed by (uid, name) so
    # two launchers with same-named but different-command processes don't collide).
    for name, cmd in processes:
        key = (uid, name)
        if key not in _procs:
            _procs[key] = _ProcState(name, cmd)

    names = [name for name, _ in processes]

    # Persistent selected index per launcher
    if uid not in _selected:
        _selected[uid] = 0

    # Render every popped-out log window owned by this launcher FIRST,
    # before the inline UI. This makes popouts independent of the dropdown
    # selection: once popped out, a log window stays up even when the user
    # switches the dropdown to a different process. (Codex flag: if the
    # popout were rendered from inside the "currently selected" branch,
    # changing selection would stop submitting the popout's Begin/End and
    # the window would silently disappear.)
    _render_open_popouts(uid)

    panel_header("PROCESS", fa.ICON_FA_TERMINAL)

    # Row 1: dropdown + Launch/Stop button + popout toggle + autoscroll toggle
    # (compact — status text would crop on narrow cells, so it gets its own
    # row below). Reserve the right side dynamically based on the actual
    # button widths at the current font scale, instead of a hardcoded fudge.
    style = imgui.get_style()
    launch_w = imgui.calc_text_size("Launch").x + 2 * style.frame_padding.x
    pop_w = (
        imgui.calc_text_size(fa.ICON_FA_UP_RIGHT_AND_DOWN_LEFT_FROM_CENTER).x
        + 2 * style.frame_padding.x
    )
    auto_w = (
        imgui.calc_text_size(fa.ICON_FA_ANGLES_DOWN).x + 2 * style.frame_padding.x
    )
    # 3 spacings: combo→launch, launch→pop, pop→autoscroll.
    reserved = launch_w + pop_w + auto_w + 3 * style.item_spacing.x
    imgui.push_item_width(-reserved)
    changed, new_idx = imgui.combo(f"##{uid}_select", _selected[uid], names)
    if changed:
        _selected[uid] = new_idx
    imgui.pop_item_width()

    selected_name = names[_selected[uid]]
    state = _procs[(uid, selected_name)]
    proc = state.process

    imgui.same_line()
    if proc is not None and proc.poll() is None:
        imgui.push_style_color(imgui.Col_.button, imgui.ImVec4(0.6, 0.15, 0.15, 1.0))
        if imgui.button(f"Stop##{uid}"):
            state.stop()
        imgui.pop_style_color()
        imgui.set_item_tooltip(f"Kill the running '{selected_name}' process (SIGKILL).")
    else:
        imgui.push_style_color(imgui.Col_.button, imgui.ImVec4(0.15, 0.4, 0.15, 1.0))
        if imgui.button(f"Launch##{uid}"):
            try:
                state.start()
            except Exception as e:
                state.log.append(f"[launch failed: {e}]")
        imgui.pop_style_color()
        imgui.set_item_tooltip(
            f"Spawn '{selected_name}' as a subprocess and stream its stdout into the log."
        )

    # Autoscroll + popout toggles — shared widgets/_log_box helpers, so the
    # buttons look + feel identical to the model panel's log controls.
    imgui.same_line()
    pop_key = (uid, selected_name)
    autoscroll_on = _autoscroll.setdefault(pop_key, True)
    popped = _popout_open.get(pop_key, False)
    autoscroll_on, popped = render_log_buttons(
        f"{uid}_{selected_name}",
        autoscroll=autoscroll_on,
        popped_out=popped,
    )
    _autoscroll[pop_key] = autoscroll_on
    _popout_open[pop_key] = popped

    # Row 2: status text on its own line so it can never get cropped.
    if proc is not None and proc.poll() is None:
        imgui.text_colored(
            imgui.ImVec4(0.2, 0.8, 0.2, 1.0),
            f"Running (PID {proc.pid})",
        )
    else:
        imgui.text_colored(imgui.ImVec4(0.5, 0.5, 0.5, 1.0), "Stopped")

    # Log area: inline if not popped, placeholder otherwise. The popout
    # itself is rendered at the top of the function (see _render_open_popouts).
    h = log_height if log_height > 0 else -1.0
    if _popout_open.get(pop_key, False):
        imgui.text_disabled(
            f"(log popped out — see '{selected_name} log' window)"
        )
    else:
        render_log(
            f"{uid}_{selected_name}",
            state.log,
            height=h,
            autoscroll=_autoscroll.get(pop_key, True),
        )

process_launcher


scatter2d

scatter2d(label: str, points: ndarray, labels: ndarray | None = None, class_names: list[str] | None = None, size: tuple[float, float] = (-1, 300), marker_size: float = 3.0) -> None

2D scatter plot with per-class coloring.

Source code in myogestic/widgets/scatter.py
def scatter2d(
    label: str,
    points: np.ndarray,
    labels: np.ndarray | None = None,
    class_names: list[str] | None = None,
    size: tuple[float, float] = (-1, 300),
    marker_size: float = 3.0,
) -> None:
    """2D scatter plot with per-class coloring."""
    if len(points) == 0:
        imgui.text(f"{label}: no data")
        return

    xs = np.ascontiguousarray(points[:, 0], dtype=np.float64)
    ys = np.ascontiguousarray(points[:, 1], dtype=np.float64)

    if implot.begin_plot(label, imgui.ImVec2(*size)):
        if labels is None:
            spec = implot.Spec()
            spec.marker_size = marker_size
            implot.plot_scatter("##points", xs, ys, spec)
        else:
            for cls in np.unique(labels):
                mask = labels == cls
                name = class_names[int(cls)] if class_names and int(cls) < len(class_names) else str(cls)
                color = PALETTE[int(cls) % len(PALETTE)]
                spec = implot.Spec()
                spec.marker_size = marker_size
                spec.marker_fill_color = imgui.ImVec4(color[0], color[1], color[2], 1.0)
                implot.plot_scatter(name, xs[mask], ys[mask], spec)
        implot.end_plot()

scatter3d

scatter3d(label: str, points: ndarray, labels: ndarray | None = None, class_names: list[str] | None = None, size: tuple[float, float] = (-1, 400), axis_names: tuple[str, str, str] = ('X', 'Y', 'Z')) -> None

3D scatter plot with orbit camera.

Source code in myogestic/widgets/scatter.py
def scatter3d(
    label: str,
    points: np.ndarray,
    labels: np.ndarray | None = None,
    class_names: list[str] | None = None,
    size: tuple[float, float] = (-1, 400),
    axis_names: tuple[str, str, str] = ("X", "Y", "Z"),
) -> None:
    """3D scatter plot with orbit camera."""
    if len(points) == 0:
        imgui.text(f"{label}: no data")
        return

    xs = np.ascontiguousarray(points[:, 0], dtype=np.float64)
    ys = np.ascontiguousarray(points[:, 1], dtype=np.float64)
    zs = np.ascontiguousarray(points[:, 2], dtype=np.float64)

    if implot3d.begin_plot(label, imgui.ImVec2(*size)):
        implot3d.setup_axes(*axis_names)
        if labels is None:
            implot3d.plot_scatter("##points", xs, ys, zs)
        else:
            for cls in np.unique(labels):
                mask = labels == cls
                name = class_names[int(cls)] if class_names and int(cls) < len(class_names) else str(cls)
                color = PALETTE[int(cls) % len(PALETTE)]
                spec = implot3d.Spec()
                spec.marker_fill_color = imgui.ImVec4(color[0], color[1], color[2], 1.0)
                implot3d.plot_scatter(name, xs[mask], ys[mask], zs[mask], spec)
        implot3d.end_plot()

heatmap

Heatmap widget for @app.ui (confusion matrix, correlation matrix, etc.).

from myogestic.widgets.heatmap import heatmap

heatmap

heatmap(label: str, data: ndarray, size: tuple[float, float] = (-1, 300), label_fmt: str = '%.1f') -> None

2D heatmap.

Source code in myogestic/widgets/heatmap.py
def heatmap(
    label: str,
    data: np.ndarray,
    size: tuple[float, float] = (-1, 300),
    label_fmt: str = "%.1f",
) -> None:
    """2D heatmap."""
    if data.size == 0:
        imgui.text(f"{label}: no data")
        return

    values = np.ascontiguousarray(data, dtype=np.float64)

    if implot.begin_plot(label, imgui.ImVec2(*size)):
        implot.plot_heatmap(
            "##heatmap", values,
            scale_min=float(values.min()),
            scale_max=float(values.max()),
            label_fmt=label_fmt,
        )
        implot.end_plot()

line_plot

Multi-channel line plot for @app.ui.

from myogestic.widgets.line_plot import line_plot

line_plot

line_plot(label: str, data: ndarray, channel_names: list[str] | None = None, size: tuple[float, float] = (-1, 200)) -> None

Multi-channel line plot.

Source code in myogestic/widgets/line_plot.py
def line_plot(
    label: str,
    data: np.ndarray,
    channel_names: list[str] | None = None,
    size: tuple[float, float] = (-1, 200),
) -> None:
    """Multi-channel line plot."""
    if len(data) == 0:
        imgui.text(f"{label}: no data")
        return

    if data.ndim == 1:
        data = data[:, np.newaxis]

    if implot.begin_plot(label, imgui.ImVec2(*size)):
        for ch in range(data.shape[1]):
            name = channel_names[ch] if channel_names and ch < len(channel_names) else f"ch{ch}"
            implot.plot_line(name, np.ascontiguousarray(data[:, ch]))
        implot.end_plot()

FilterControl

FilterControl(hz: float = 50.0, default: str = 'one_euro')

Stateful holder for a runtime-tunable :class:VectorFilter.

Renders a self-contained panel with a header, button-style filter selector, parameter controls, and a reset button. Parameters update in place where possible (no rebuild) to preserve smoothing history during live tuning.

Parameters:

Name Type Description Default
hz float

Sample rate forwarded to one_euro (ignored by others).

50.0
default str

Initial filter name — "identity" | "gaussian" | "one_euro".

'one_euro'
Source code in myogestic/widgets/filter_controls.py
def __init__(self, hz: float = 50.0, default: str = "one_euro"):
    if default not in _NAMES:
        raise ValueError(f"default must be one of {_NAMES}, got {default!r}")
    self.hz = hz
    self._name = default
    self._params: dict[str, dict[str, Any]] = {
        "identity": {},
        "gaussian": {"window": 5, "sigma": 1.0},
        "one_euro": {"min_cutoff": 1.0, "beta": 0.02, "d_cutoff": 1.0},
    }
    self._filter: VectorFilter = self._build()

reset

reset() -> None

Clear the active filter's smoothing history.

Source code in myogestic/widgets/filter_controls.py
def reset(self) -> None:
    """Clear the active filter's smoothing history."""
    self._filter.reset()

ui

ui(label: str = 'output_filter') -> None

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

Source code in myogestic/widgets/filter_controls.py
def ui(self, label: str = "output_filter") -> None:
    """Render the full panel. Call once per frame inside @app.ui."""
    muted = imgui.get_style().color_(imgui.Col_.text_disabled)
    # Header (shared helper) + right-aligned Reset button on the same row
    panel_header("POST-PROCESSING", fa.ICON_FA_WAVE_SQUARE)
    imgui.same_line()
    # Right-align the reset button. Measure its actual width from the
    # text + frame padding so we don't depend on a hardcoded constant.
    reset_label = f"{fa.ICON_FA_ROTATE_LEFT}  Reset"
    text_w = imgui.calc_text_size(reset_label).x
    frame_pad_x = imgui.get_style().frame_padding.x
    btn_w = text_w + frame_pad_x * 2
    avail = imgui.get_content_region_avail().x
    if avail > btn_w + 4:
        imgui.set_cursor_pos_x(imgui.get_cursor_pos_x() + (avail - btn_w))
    if imgui.small_button(f"{reset_label}##{label}"):
        self.reset()
    if imgui.is_item_hovered():
        imgui.set_tooltip("Clear smoothing history (e.g. on a new session).")

    # Filter type buttons — visually consistent with class buttons
    self._render_buttons(label)

    # Brief description
    imgui.push_style_color(imgui.Col_.text, muted)
    imgui.text_wrapped(_DESCRIPTIONS[self._name])
    imgui.pop_style_color()

    # Parameter controls
    self._render_params(label)

FilterControl


FeatureSelector

FeatureSelector(features: dict[str, FeatureFn], default: Iterable[str] | None = None)

Tickable list of named feature functions.

Parameters:

Name Type Description Default
features dict[str, FeatureFn]

Ordered map of feature name → callable. Each callable takes an EMG window (n_channels, n_samples) and returns an array - typically (n_channels, n_features_out) for time-preserving features like sliding RMS, but any consistent shape works as long as every active feature returns the same shape (they're concatenated along axis 0 by :meth:__call__).

required
default Iterable[str] | None

Optional iterable of feature names to start ticked. None (default) ticks every feature; an empty list ticks none.

None

Raises:

Type Description
ValueError

if a name in default isn't in features.

Source code in myogestic/widgets/feature_selector.py
def __init__(
    self,
    features: dict[str, FeatureFn],
    default: Iterable[str] | None = None,
) -> None:
    if not features:
        raise ValueError("FeatureSelector needs at least one feature")
    self._features: dict[str, FeatureFn] = dict(features)
    if default is None:
        default_set = set(self._features)
    else:
        default_set = set(default)
        unknown = default_set - set(self._features)
        if unknown:
            raise ValueError(
                f"default contains unknown feature names: {sorted(unknown)}. "
                f"Available: {list(self._features)}"
            )
    # Insertion order = registration order; preserved for reproducibility.
    self._active: dict[str, bool] = {
        name: (name in default_set) for name in self._features
    }

active_names property

active_names: list[str]

Feature names currently ticked, in registration order.

n_active property

n_active: int

Number of ticked features.

is_active

is_active(name: str) -> bool

Check whether a specific feature is ticked.

Source code in myogestic/widgets/feature_selector.py
def is_active(self, name: str) -> bool:
    """Check whether a specific feature is ticked."""
    return self._active.get(name, False)

set_active

set_active(name: str, on: bool) -> None

Programmatically tick / untick a feature.

Useful for restoring saved selections from a checkpoint, or for scripted training runs that bypass the UI.

Source code in myogestic/widgets/feature_selector.py
def set_active(self, name: str, on: bool) -> None:
    """Programmatically tick / untick a feature.

    Useful for restoring saved selections from a checkpoint, or for
    scripted training runs that bypass the UI.
    """
    if name not in self._features:
        raise ValueError(f"Unknown feature {name!r}")
    self._active[name] = bool(on)

ui

ui() -> None

Render the panel inside an ImGui frame.

Call from inside @app.ui. Renders a header, the feature checkboxes laid out in a wrapping grid that reflows with the panel's current width (one column when narrow, many when wide), and a footer line showing the active count. State updates take effect on the next predict-thread tick.

Source code in myogestic/widgets/feature_selector.py
def ui(self) -> None:
    """Render the panel inside an ImGui frame.

    Call from inside ``@app.ui``. Renders a header, the feature
    checkboxes laid out in a **wrapping grid** that reflows with the
    panel's current width (one column when narrow, many when wide),
    and a footer line showing the active count. State updates take
    effect on the next predict-thread tick.
    """
    panel_header("FEATURES", fa.ICON_FA_LAYER_GROUP)

    # Compute how many checkboxes fit per row at the current panel
    # width. Item width = checkbox indicator (~one frame-height
    # square) + inner spacing + the widest label. We size every
    # column to the widest label so the grid stays a clean alignment.
    style = imgui.get_style()
    spacing = style.item_spacing.x
    inner = style.item_inner_spacing.x
    frame_h = imgui.get_frame_height()
    max_label_w = max(
        imgui.calc_text_size(name).x for name in self._features
    )
    item_w = frame_h + inner + max_label_w
    avail_w = imgui.get_content_region_avail().x
    n_cols = max(
        1, int((avail_w + spacing) / (item_w + spacing)) if item_w > 0 else 1
    )

    for i, name in enumerate(self._features):
        if i > 0 and i % n_cols != 0:
            imgui.same_line()
        on = self._active[name]
        changed, new_val = imgui.checkbox(name, on)
        if changed:
            self._active[name] = new_val

    n = self.n_active
    suffix = "" if n == 1 else "s"
    imgui.text_disabled(f"{n} active feature{suffix}")

FeatureSelector


template_inspector

Generic template-row review table.

A reusable accept/reject + click-to-select widget. Caller owns extraction (the Extract Templates button + worker thread + how to build rows from sessions); this widget renders the resulting rows and lets the user check checkboxes and click rows to select one for inspection.

Reusable from any example that wants an "extract → review → train" workflow (regression target picking, NN onset detection, …).

Design notes: - TemplateInspectorRow is a small mutable dataclass; the widget toggles accepted in place when the user ticks a checkbox. Caller reads [r for r in rows if r.accepted] to pull the selected set. - The widget returns the currently-selected row's key (or None) so the caller can render a preview / details panel for that row. - No alias maps, label classification, or model semantics live here — those stay in user code.

TemplateInspectorRow dataclass

TemplateInspectorRow(key: str, label: str, accepted: bool = True, info_text: str | None = None, energy: float | None = None)

One row in the inspector table.

Parameters:

Name Type Description Default
key str

Stable identity (e.g. "<session>#<trial_idx>"). The widget uses this for selection and for ImGui id disambiguation.

required
label str

Short class/category badge (e.g. "OPEN" / "CLOSED").

required
accepted bool

Mutable. True = include in training. Toggled in place by the checkbox.

True
info_text str | None

Optional secondary text shown in the table (e.g. session name, source path). May be None.

None
energy float | None

Optional scalar shown as a normalised progress bar in the energy column. Caller's choice of metric — RMS energy, peak amplitude, anything monotonic. None hides the bar.

None

template_inspector

template_inspector(uid: str, rows: list[TemplateInspectorRow], *, title: str = 'Templates', height: float = 240.0, label_colors: dict[str, tuple[float, float, float, float]] | None = None) -> str | None

Render the table. Returns the selected row's key (or None).

Parameters:

Name Type Description Default
uid str

Stable identity string. Two calls with the same uid share selection state across frames; different uids are independent.

required
rows list[TemplateInspectorRow]

List of TemplateInspectorRow to render. Mutated in place (only accepted is touched by the widget).

required
title str

Header text shown above the table.

'Templates'
height float

Table height in pixels.

240.0
label_colors dict[str, tuple[float, float, float, float]] | None

Optional {label_text: (r, g, b, a)} for the colored badge in the label column. Unmapped labels render in the default text color.

None

Returns:

Type Description
str | None

The key of the currently-selected row (or None if no row

str | None

is selected, or the previously-selected row was removed).

Source code in myogestic/widgets/template_inspector.py
def template_inspector(
    uid: str,
    rows: list[TemplateInspectorRow],
    *,
    title: str = "Templates",
    height: float = 240.0,
    label_colors: dict[str, tuple[float, float, float, float]] | None = None,
) -> str | None:
    """Render the table. Returns the selected row's key (or ``None``).

    Args:
        uid: Stable identity string. Two calls with the same uid share
            selection state across frames; different uids are independent.
        rows: List of ``TemplateInspectorRow`` to render. Mutated in place
            (only ``accepted`` is touched by the widget).
        title: Header text shown above the table.
        height: Table height in pixels.
        label_colors: Optional ``{label_text: (r, g, b, a)}`` for the
            colored badge in the label column. Unmapped labels render in
            the default text color.

    Returns:
        The ``key`` of the currently-selected row (or ``None`` if no row
        is selected, or the previously-selected row was removed).
    """
    from imgui_bundle import imgui

    label_colors = label_colors or {}
    selected = _SELECTED.get(uid)
    # Drop a stale selection if its row is no longer in the table.
    if selected is not None and not any(r.key == selected for r in rows):
        selected = None
        _SELECTED[uid] = None

    if title:
        imgui.text(title)
    if not rows:
        imgui.text_disabled("(no rows)")
        return selected

    # Energy bar normalisation (relative to current rows).
    e_max = max((r.energy or 0.0) for r in rows) or 1.0

    if imgui.begin_table(
        f"##{uid}_tinsp",
        5,
        imgui.TableFlags_.borders_inner_h
        | imgui.TableFlags_.row_bg
        | imgui.TableFlags_.scroll_y,
        imgui.ImVec2(-1, height),
    ):
        imgui.table_setup_column("Use", imgui.TableColumnFlags_.width_fixed, 30)
        imgui.table_setup_column("Label", imgui.TableColumnFlags_.width_fixed, 70)
        imgui.table_setup_column("Source", imgui.TableColumnFlags_.width_stretch)
        imgui.table_setup_column("Energy", imgui.TableColumnFlags_.width_fixed, 80)
        imgui.table_setup_column("Trial", imgui.TableColumnFlags_.width_fixed, 50)
        for i, row in enumerate(rows):
            imgui.table_next_row()
            imgui.table_next_column()
            changed, new_acc = imgui.checkbox(f"##{uid}_acc{i}", row.accepted)
            if changed:
                row.accepted = new_acc
            imgui.table_next_column()
            color_rgba = label_colors.get(row.label)
            if color_rgba is not None:
                imgui.text_colored(imgui.ImVec4(*color_rgba), row.label)
            else:
                imgui.text(row.label)
            imgui.table_next_column()
            is_selected = row.key == selected
            label_text = row.info_text or row.key
            clicked, _ = imgui.selectable(
                f"{label_text}##{uid}_sel{i}",
                is_selected,
                imgui.SelectableFlags_.span_all_columns,
            )
            if clicked:
                selected = row.key
                _SELECTED[uid] = selected
            imgui.table_next_column()
            if row.energy is not None:
                imgui.progress_bar(
                    min(row.energy / e_max, 1.0), imgui.ImVec2(70, 0), ""
                )
            else:
                imgui.text_disabled("—")
            imgui.table_next_column()
            tail = row.key.split("#")[-1] if "#" in row.key else ""
            imgui.text(tail)
        imgui.end_table()

    return selected

clear_selection

clear_selection(uid: str) -> None

Drop the cached selection for uid (e.g. after Clear button).

Source code in myogestic/widgets/template_inspector.py
def clear_selection(uid: str) -> None:
    """Drop the cached selection for ``uid`` (e.g. after Clear button)."""
    _SELECTED[uid] = None

TemplateInspectorRow dataclass

TemplateInspectorRow(key: str, label: str, accepted: bool = True, info_text: str | None = None, energy: float | None = None)

One row in the inspector table.

Parameters:

Name Type Description Default
key str

Stable identity (e.g. "<session>#<trial_idx>"). The widget uses this for selection and for ImGui id disambiguation.

required
label str

Short class/category badge (e.g. "OPEN" / "CLOSED").

required
accepted bool

Mutable. True = include in training. Toggled in place by the checkbox.

True
info_text str | None

Optional secondary text shown in the table (e.g. session name, source path). May be None.

None
energy float | None

Optional scalar shown as a normalised progress bar in the energy column. Caller's choice of metric — RMS energy, peak amplitude, anything monotonic. None hides the bar.

None

trial_preview

Stacked-channel waveform with optional shaded band overlay.

Generic trial / template / segment preview widget. Renders multi-channel biosignal data as stacked traces (per-channel offsets), optionally with a colored band marking a region of interest (e.g. an extracted template, a labeled gesture, a chosen training window).

Reusable from any example that wants a recorded-trial review surface.

Design notes: - Channels-first (default) or samples-first via data_layout. Live Stream.get_window() is channels-first; Recording.data from Session.get_trials() is samples-first — pass "samples_first" there instead of transposing at the call site. - Auto-gain (lane = data_range * 1.2) like signal_viewer's Auto mode, so channels with different amplitudes stay visible. Manual mode pins lane to y_range. - Display filters are explicit kwargs (rectify / dc_removal / rms_env / none), not coupled to a live viewer's state — the caller decides what to mirror.

trial_preview

trial_preview(uid: str, data: ndarray, fs: float, *, data_layout: Literal['channels_first', 'samples_first'] = 'channels_first', title: str | None = None, size: tuple[float, float] = (-1.0, 240.0), channel_names: list[str] | None = None, band: tuple[float, float] | None = None, band_color: tuple[float, float, float, float] | None = None, gain: float = 1.0, display_filter: Literal['none', 'rectify', 'dc_removal', 'rms_env'] = 'none', scale_mode: Literal['auto', 'manual'] = 'auto', y_range: tuple[float, float] = (-1.0, 1.0), window: bool = False) -> None

Render stacked multi-channel waveform with optional band overlay.

Parameters:

Name Type Description Default
uid str

Stable identity string for ImPlot (combined into plot ids so two trial_preview calls in the same frame don't collide).

required
data ndarray

Multi-channel signal. Shape (n_channels, n_samples) if data_layout == "channels_first" (default) or (n_samples, n_channels) if "samples_first".

required
fs float

Sampling rate in Hz, used for the x-axis labels in seconds.

required
title str | None

Optional header line shown above the plot.

None
size tuple[float, float]

ImPlot size as (width, height). -1 width fills the available content region.

(-1.0, 240.0)
channel_names list[str] | None

Optional per-channel labels. When omitted, channels are shown as ch0..chN-1.

None
band tuple[float, float] | None

Optional (t_start_s, t_end_s) shaded band drawn behind the traces — useful for marking an extracted template, highlighting a labeled segment, etc.

None
band_color tuple[float, float, float, float] | None

RGBA in [0,1]. Defaults to a soft cyan.

None
gain float

Multiplier applied to each channel before plotting. Match this to your live viewer's gain knob if you want the preview to look like what was on screen.

1.0
display_filter Literal['none', 'rectify', 'dc_removal', 'rms_env']

Visual-only transform applied to a copy of data before plotting. Same vocabulary as signal_viewer's display dropdown.

'none'
scale_mode Literal['auto', 'manual']

"auto" (default) computes the per-lane height from the signal's global min/max with 20% padding; "manual" uses y_range directly.

'auto'
y_range tuple[float, float]

(y_min, y_max) used in manual scale mode.

(-1.0, 1.0)
window bool

When True, the widget wraps itself in a free-floating ImGui window with title title. When False (default), it draws inline at the current cursor position.

False
Source code in myogestic/widgets/trial_preview.py
def trial_preview(
    uid: str,
    data: np.ndarray,
    fs: float,
    *,
    data_layout: Literal["channels_first", "samples_first"] = "channels_first",
    title: str | None = None,
    size: tuple[float, float] = (-1.0, 240.0),
    channel_names: list[str] | None = None,
    band: tuple[float, float] | None = None,
    band_color: tuple[float, float, float, float] | None = None,
    gain: float = 1.0,
    display_filter: Literal["none", "rectify", "dc_removal", "rms_env"] = "none",
    scale_mode: Literal["auto", "manual"] = "auto",
    y_range: tuple[float, float] = (-1.0, 1.0),
    window: bool = False,
) -> None:
    """Render stacked multi-channel waveform with optional band overlay.

    Args:
        uid: Stable identity string for ImPlot (combined into plot ids so
            two ``trial_preview`` calls in the same frame don't collide).
        data: Multi-channel signal. Shape ``(n_channels, n_samples)`` if
            ``data_layout == "channels_first"`` (default) or
            ``(n_samples, n_channels)`` if ``"samples_first"``.
        fs: Sampling rate in Hz, used for the x-axis labels in seconds.
        title: Optional header line shown above the plot.
        size: ImPlot size as ``(width, height)``. ``-1`` width fills the
            available content region.
        channel_names: Optional per-channel labels. When omitted, channels
            are shown as ``ch0..chN-1``.
        band: Optional ``(t_start_s, t_end_s)`` shaded band drawn behind
            the traces — useful for marking an extracted template,
            highlighting a labeled segment, etc.
        band_color: RGBA in ``[0,1]``. Defaults to a soft cyan.
        gain: Multiplier applied to each channel before plotting. Match
            this to your live viewer's gain knob if you want the preview
            to look like what was on screen.
        display_filter: Visual-only transform applied to a copy of
            ``data`` before plotting. Same vocabulary as
            ``signal_viewer``'s display dropdown.
        scale_mode: ``"auto"`` (default) computes the per-lane height
            from the signal's global min/max with 20% padding;
            ``"manual"`` uses ``y_range`` directly.
        y_range: ``(y_min, y_max)`` used in manual scale mode.
        window: When ``True``, the widget wraps itself in a free-floating
            ImGui window with title ``title``. When ``False`` (default),
            it draws inline at the current cursor position.
    """
    from imgui_bundle import imgui, implot

    # Normalise to channels-first internally.
    if data_layout == "samples_first":
        arr = np.ascontiguousarray(data.T, dtype=np.float64)
    else:
        arr = np.asarray(data, dtype=np.float64)
    if arr.ndim != 2 or arr.shape[0] == 0 or arr.shape[1] == 0:
        if window:
            return
        imgui.text_disabled("(no data)")
        return
    n_ch, n_samp = arr.shape

    # Apply display filter on a samples-first view (apply_display_filter
    # contract). Round-trip back to channels-first.
    if display_filter != "none":
        arr = apply_display_filter(
            arr.T.astype(np.float32, copy=False), display_filter, fs
        ).T.astype(np.float64)

    if window:
        imgui.set_next_window_size(imgui.ImVec2(960, 480), imgui.Cond_.first_use_ever)
        title_id = (title or "Trial Preview") + f"###trial_preview_{uid}"
        opened, _is_open = imgui.begin(title_id, True)
        if not opened:
            imgui.end()
            return

    if title is not None and not window:
        imgui.text(title)

    # `lane` = per-channel vertical spacing. In manual mode it's the user-
    # provided y_range height; in auto mode it's the global data range with
    # 20 % padding (matches signal_viewer's auto behaviour).
    if scale_mode == "manual":
        lane = max(float(y_range[1]) - float(y_range[0]), 1e-9)
    else:
        finite = arr[np.isfinite(arr)]
        if finite.size:
            d_lo = float(finite.min())
            d_hi = float(finite.max())
            lane = max(d_hi - d_lo, 1e-9) * 1.2
        else:
            lane = 1.0
    # The y-axis extent always spans all channels with a half-lane pad on
    # top and bottom — without this, manual mode would show only channel 0
    # and clip channels 1..n_ch-1 below the visible y range.
    y_hi = lane * 0.6
    y_lo = -lane * (n_ch - 1) - lane * 0.6

    xs = np.arange(n_samp, dtype=np.float64) / fs

    flags = implot.Flags_.no_legend | implot.Flags_.no_title
    if implot.begin_plot(
        f"trial_preview##{uid}",
        imgui.ImVec2(size[0], size[1]),
        flags=flags,
    ):
        implot.setup_axis(implot.ImAxis_.x1, "time (s)")
        implot.setup_axis_limits(implot.ImAxis_.x1, 0.0, n_samp / fs, implot.Cond_.always)  # type: ignore[attr-defined]
        implot.setup_axis(implot.ImAxis_.y1, flags=implot.AxisFlags_.no_tick_labels)
        implot.setup_axis_limits(implot.ImAxis_.y1, y_lo, y_hi, implot.Cond_.always)  # type: ignore[attr-defined]

        # Band overlay below the traces.
        if band is not None:
            bc = band_color if band_color is not None else (0.4, 0.7, 1.0, 1.0)
            spec = implot.Spec()
            spec.fill_color = imgui.ImVec4(bc[0], bc[1], bc[2], 1.0)
            spec.fill_alpha = bc[3] if len(bc) >= 4 else 0.22
            implot.plot_shaded(
                f"band##{uid}",
                np.array([float(band[0]), float(band[1])], dtype=np.float64),
                np.array([y_hi, y_hi], dtype=np.float64),
                np.array([y_lo, y_lo], dtype=np.float64),
                spec,
            )

        for ch in range(n_ch):
            ys = np.ascontiguousarray(arr[ch] * gain - ch * lane, dtype=np.float64)
            label = (
                channel_names[ch] if channel_names and ch < len(channel_names)
                else f"ch{ch}"
            )
            implot.plot_line(f"{label}##{uid}", xs, ys)
        implot.end_plot()

    if window:
        imgui.end()

panel_header

panel_header(title: str, icon: str | None = None) -> None

Render a uniform panel-header line: muted, all-caps, optional FA icon.

Pairs with the button-button + slider styling used by the other widgets in this package. Use it at the top of any custom panel to match the look::

panel_header("MODEL", icons_fontawesome_6.ICON_FA_BRAIN)
train_button(pipeline)
...

The text color follows the active theme's text_disabled slot, so it reads correctly on both light and dark themes without hardcoding.

Source code in myogestic/widgets/_common.py
def panel_header(title: str, icon: str | None = None) -> None:
    """Render a uniform panel-header line: muted, all-caps, optional FA icon.

    Pairs with the button-button + slider styling used by the other widgets in
    this package. Use it at the top of any custom panel to match the look::

        panel_header("MODEL", icons_fontawesome_6.ICON_FA_BRAIN)
        train_button(pipeline)
        ...

    The text color follows the active theme's ``text_disabled`` slot, so it
    reads correctly on both light and dark themes without hardcoding.
    """
    muted = imgui.get_style().color_(imgui.Col_.text_disabled)
    imgui.push_style_color(imgui.Col_.text, muted)
    imgui.text(f"{icon}  {title.upper()}" if icon else title.upper())
    imgui.pop_style_color()

popout_panel

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

Render gui_fn inside a dockable, tearable ImGui window.

Parameters:

Name Type Description Default
title str

Window title — also used as the ImGui id and as the dedup key for repeated calls.

required
gui_fn Callable[[], None]

Zero-arg callable invoked by ImGui every frame. Treat it like the body of a with imgui.begin(...): — call imgui/ implot from inside.

required
default_open bool

Initial visibility of the window on first launch. Subsequent launches restore from .imgui_state.

True
can_be_closed bool

Whether the user can close the window with the X button. Closed windows reappear via the "View" menu.

True
remember_is_visible bool | None

Whether visibility is persisted in the imgui ini file. Defaults to True for existing behavior.

None

When App(docking=True) is not active, this just runs gui_fn() inline so the call site stays the same.

Source code in myogestic/widgets/popout.py
def popout_panel(
    title: str,
    gui_fn: Callable[[], None],
    *,
    default_open: bool = True,
    can_be_closed: bool = True,
    remember_is_visible: bool | None = None,
) -> None:
    """Render `gui_fn` inside a dockable, tearable ImGui window.

    Args:
        title: Window title — also used as the ImGui id and as the dedup
            key for repeated calls.
        gui_fn: Zero-arg callable invoked by ImGui every frame. Treat it
            like the body of a `with imgui.begin(...):` — call ``imgui``/
            ``implot`` from inside.
        default_open: Initial visibility of the window on first launch.
            Subsequent launches restore from ``.imgui_state``.
        can_be_closed: Whether the user can close the window with the X
            button. Closed windows reappear via the "View" menu.
        remember_is_visible: Whether visibility is persisted in the imgui
            ini file. Defaults to True for existing behavior.

    When `App(docking=True)` is not active, this just runs `gui_fn()`
    inline so the call site stays the same.
    """
    from myogestic import core as _core  # late import → no circular

    app = _core._active_app
    if app is None or not getattr(app, "_docking", False):
        # Inline fallback — visually identical to calling gui_fn() directly.
        gui_fn()
        return

    if title in _registered:
        return  # already registered; hello_imgui pumps the gui_function.

    win = _make_dockable_window(
        title,
        gui_fn,
        default_open,
        can_be_closed,
        remember_is_visible,
    )

    # Append to the pending list. If the GUI loop hasn't started yet,
    # `_gui_loop` will drain this into runner_params before launch.
    # If we're already inside the loop (popout_panel called from
    # @app.ui's first frame), append directly to the live params so
    # hello_imgui picks it up on the next frame.
    _core._pending_popouts.append(win)

    rp = getattr(app, "_runner_params", None)
    if rp is not None and rp.docking_params is not None:
        current = list(rp.docking_params.dockable_windows)
        if not any(getattr(w, "label", None) == title for w in current):
            current.append(win)
        rp.docking_params.dockable_windows = current

    _registered[title] = win

Status and logs

stream_panel

Per-stream status panel for @app.ui.

A compact replacement for the MyoGestic "device setup" tab that only shows what's actually true at runtime: source class, connection status, sample rate, channel count, last-sample age, plus inline connect buttons for any target the source's discover() reports. Rendering is one-shot per frame — no hidden state beyond the discovery cache (shared with the signal viewers).

stream_panel

stream_panel(ctx: Context, selectable: bool = True, show_header: bool = True) -> None

Render one row per stream with status + metadata + reconnect.

Parameters:

Name Type Description Default
ctx Context

App context.

required
selectable bool

When True and the stream's source supports discover(), auto-populate available targets as inline connect buttons.

True
show_header bool

Render a uniform panel_header above the rows.

True
Source code in myogestic/widgets/stream_panel.py
def stream_panel(
    ctx: Context,
    selectable: bool = True,
    show_header: bool = True,
) -> None:
    """Render one row per stream with status + metadata + reconnect.

    Args:
        ctx: App context.
        selectable: When True and the stream's source supports `discover()`,
            auto-populate available targets as inline connect buttons.
        show_header: Render a uniform `panel_header` above the rows.
    """
    if show_header:
        panel_header("Streams", fa.ICON_FA_PLUG)

    if not ctx.streams:
        imgui.text_colored(_MUTED, "(no streams registered)")
        return

    for name, stream in ctx.streams.items():
        _stream_row(name, stream, selectable=selectable)

log_panel

General app-event log panel for @app.ui.

Displays whatever lines have been pushed via ctx.log(...). Independent from pipeline.train_log (model-training only) — this is for high-level app events (recording saved, model loaded, stream reconnected, process crashed). The widget is read-only and auto-scrolls to the latest line when the user is already pinned to the bottom.

log_panel

log_panel(ctx: Context, height: float = -1.0, title: str = 'App Log', show_header: bool = True) -> None

Render the app log as a scrollable, read-only panel.

Parameters:

Name Type Description Default
ctx Context

App context; reads from ctx.logs.

required
height float

Panel height in pixels. Pass a value <= 0 (default) to fill the remaining vertical space of the parent cell — matches the ImGui convention where -1 means "fill available".

-1.0
title str

Header label (only shown when show_header=True).

'App Log'
show_header bool

Render the button-style panel_header above the log.

True
Source code in myogestic/widgets/log_panel.py
def log_panel(
    ctx: Context,
    height: float = -1.0,
    title: str = "App Log",
    show_header: bool = True,
) -> None:
    """Render the app log as a scrollable, read-only panel.

    Args:
        ctx: App context; reads from ``ctx.logs``.
        height: Panel height in pixels. Pass a value ``<= 0`` (default) to
            fill the remaining vertical space of the parent cell — matches
            the ImGui convention where ``-1`` means "fill available".
        title: Header label (only shown when ``show_header=True``).
        show_header: Render the button-style ``panel_header`` above the log.
    """
    if show_header:
        panel_header(title, fa.ICON_FA_TERMINAL)

    if imgui.button(f"{fa.ICON_FA_BROOM}  Clear##log_panel"):
        ctx.logs.clear()

    text = "\n".join(ctx.logs) if ctx.logs else "(no events yet)"
    h = height if height > 0 else -1.0
    imgui.input_text_multiline(
        "##log_panel",
        text,
        imgui.ImVec2(-1, h),
        flags=imgui.InputTextFlags_.read_only,
    )

Branding

app_logo(max_size: float | None = None, padding: float = 12.0) -> None

Render the MyoGestic wordmark, fit-to-cell, aspect-preserving.

The widget reads the available content area inside the current panel and renders the wordmark as the largest aspect-preserving rectangle that fits both dimensions (minus the requested padding), then centres it. So in a cell whose aspect matches the wordmark, the image fills edge-to-edge minus the padding margin; in a cell that's a different aspect, the image fills the tighter dimension and leaves extra balanced padding along the other.

Parameters:

Name Type Description Default
max_size float | None

Optional cap on the wordmark's width in pixels. None (default) lets the image grow to fill the cell — appropriate for "logo in a dedicated branding cell" use. Pass a value when the cell can be much larger than the wordmark should ever appear (e.g. a logo embedded in a full-screen splash).

None
padding float

Margin in pixels reserved on every side. Default 12 px gives the wordmark breathing room against the panel border.

12.0

Uses image_and_size_from_asset + raw imgui.image rather than the higher-level image_from_asset(..., size=...) helper, which in this version of hello_imgui ignored the explicit size and rendered at the natural pixel dimensions of the PNG.

Source code in myogestic/widgets/app_logo.py
def app_logo(max_size: float | None = None, padding: float = 12.0) -> None:
    """Render the MyoGestic wordmark, fit-to-cell, aspect-preserving.

    The widget reads the available content area inside the current panel
    and renders the wordmark as the **largest aspect-preserving rectangle
    that fits both dimensions** (minus the requested padding), then
    centres it. So in a cell whose aspect matches the wordmark, the image
    fills edge-to-edge minus the padding margin; in a cell that's a
    different aspect, the image fills the tighter dimension and leaves
    extra balanced padding along the other.

    Args:
        max_size: Optional cap on the wordmark's *width* in pixels. ``None``
            (default) lets the image grow to fill the cell — appropriate for
            "logo in a dedicated branding cell" use. Pass a value when the
            cell can be much larger than the wordmark should ever appear
            (e.g. a logo embedded in a full-screen splash).
        padding: Margin in pixels reserved on every side. Default 12 px
            gives the wordmark breathing room against the panel border.

    Uses ``image_and_size_from_asset`` + raw ``imgui.image`` rather than the
    higher-level ``image_from_asset(..., size=...)`` helper, which in this
    version of hello_imgui ignored the explicit size and rendered at the
    natural pixel dimensions of the PNG.
    """
    if not hello_imgui.asset_exists(_LOGO_ASSET):
        imgui.text_disabled("(myogestic logo asset missing)")
        return
    info: hello_imgui.ImageAndSize = hello_imgui.image_and_size_from_asset(_LOGO_ASSET)
    # imgui.image() wants ImTextureRef; image_and_size_from_asset returns
    # the raw int texture id, so wrap it explicitly.
    tex_ref = imgui.ImTextureRef(info.texture_id)
    natural = info.size
    aspect = natural.x / natural.y if natural.y else 1.0

    avail_w = imgui.get_content_region_avail().x
    avail_h = imgui.get_content_region_avail().y
    # Subtract padding before the fit calculation so the margin is
    # honoured even when the cell is the same aspect as the image.
    usable_w = max(0.0, avail_w - 2 * padding)
    usable_h = max(0.0, avail_h - 2 * padding)
    if usable_w <= 0 or usable_h <= 0:
        return

    # Fit-in-rect: largest aspect-preserving box that fits both dimensions.
    target_w = min(usable_w, usable_h * aspect)
    target_h = target_w / aspect

    # Optional cap so the logo doesn't blow up on a huge cell.
    if max_size is not None and target_w > max_size:
        target_w = max_size
        target_h = max_size / aspect

    # Centre horizontally via indent/unindent — `set_cursor_pos_x` would
    # extend the parent's content extent without submitting an item to
    # claim it, which trips ImGui's IM_ASSERT on the surrounding child.
    offset = max(0.0, (avail_w - target_w) * 0.5)
    # Centre vertically by reserving an empty `Dummy` item before the image
    # equal to half the leftover height. `Dummy` is the documented ImGui way
    # to claim layout space — `set_cursor_pos_y` would trip the same
    # boundary-extension assertion as the horizontal case.
    v_pad = max(0.0, (avail_h - target_h) * 0.5)
    if v_pad:
        imgui.dummy(imgui.ImVec2(1.0, v_pad))
    if offset:
        imgui.indent(offset)
    imgui.image(tex_ref, imgui.ImVec2(target_w, target_h))
    if offset:
        imgui.unindent(offset)

app_logo


ML readout

prediction_label

prediction_label(pipeline: Pipeline, class_names: Sequence[str], *, key: str = 'class', proba_key: str = 'proba', label: str = 'Prediction', show_probability: bool = False, font_scale: float = 2.0) -> None

Render the current predicted class name as a big centred label.

The class index is looked up in pipeline.predictions[key] and the name is taken from class_names. Colour-codes each class with the shared :data:myogestic.widgets._common.PALETTE so the same class is always the same colour (matches the recording / session-manager chips).

Parameters:

Name Type Description Default
pipeline Pipeline

The Pipeline whose predictions to read. Untrained or first-frame state (predictions == {}) renders as a muted "—".

required
class_names Sequence[str]

Class names indexed the same way as the model — class_names[i] is the name for class index i.

required
key str

Dict key in predictions holding the class index. Default "class" matches the convention in the bundled examples.

'class'
proba_key str

Dict key holding the per-class probability vector, consumed only when show_probability is on.

'proba'
label str

Panel header text.

'Prediction'
show_probability bool

When True, render a coloured progress bar of the predicted class's probability below the name.

False
font_scale float

Multiplier applied to the class name's text size. Defaults to 2× the panel font.

2.0
Source code in myogestic/widgets/prediction_label.py
def prediction_label(
    pipeline: Pipeline,
    class_names: Sequence[str],
    *,
    key: str = "class",
    proba_key: str = "proba",
    label: str = "Prediction",
    show_probability: bool = False,
    font_scale: float = 2.0,
) -> None:
    """Render the current predicted class name as a big centred label.

    The class index is looked up in ``pipeline.predictions[key]`` and the
    name is taken from ``class_names``. Colour-codes each class with the
    shared :data:`myogestic.widgets._common.PALETTE` so the same class is
    always the same colour (matches the recording / session-manager
    chips).

    Args:
        pipeline: The Pipeline whose predictions to read. Untrained or
            first-frame state (``predictions == {}``) renders as a muted "—".
        class_names: Class names indexed the same way as the model —
            ``class_names[i]`` is the name for class index ``i``.
        key: Dict key in ``predictions`` holding the class index. Default
            ``"class"`` matches the convention in the bundled examples.
        proba_key: Dict key holding the per-class probability vector,
            consumed only when ``show_probability`` is on.
        label: Panel header text.
        show_probability: When True, render a coloured progress bar of the
            predicted class's probability below the name.
        font_scale: Multiplier applied to the class name's text size.
            Defaults to 2× the panel font.
    """
    panel_header(label, fa.ICON_FA_BRAIN)
    imgui.spacing()

    idx = pipeline.predictions.get(key)
    if not isinstance(idx, int) or not (0 <= idx < len(class_names)):
        imgui.text_disabled("—  (no prediction yet)")
        imgui.spacing()
        return

    name = class_names[idx]
    rgb = PALETTE[idx % len(PALETTE)]
    rgba = imgui.ImVec4(float(rgb[0]), float(rgb[1]), float(rgb[2]), 1.0)

    # Big, centred class name. `set_window_font_scale` was removed upstream;
    # the modern API is `push_font(None, unscaled_base_size)` — pass None to
    # keep the current font, and a size in the *unscaled* base unit (style
    # global scale factors are applied on top automatically).
    base_size = imgui.get_style().font_size_base
    imgui.push_font(None, base_size * font_scale)
    try:
        avail = imgui.get_content_region_avail().x
        text_w = imgui.calc_text_size(name).x
        imgui.set_cursor_pos_x(
            imgui.get_cursor_pos_x() + max(0.0, (avail - text_w) * 0.5)
        )
        imgui.text_colored(rgba, name)
    finally:
        imgui.pop_font()

    if show_probability:
        proba = pipeline.predictions.get(proba_key)
        try:
            p = float(proba[idx])  # type: ignore[index]
        except (TypeError, IndexError):
            p = None
        imgui.spacing()
        if p is None:
            imgui.text_disabled("(no probability vector in predictions)")
        else:
            imgui.push_style_color(imgui.Col_.plot_histogram, rgba)
            imgui.progress_bar(p, imgui.ImVec2(-1, 0), f"{p:.0%}")
            imgui.pop_style_color()

    imgui.spacing()

prediction_label


Virtual Hand integration

VhiMovementPanel

VhiMovementPanel(client: VhiControlClient, *, on_movement: Callable[[str], None] | None = None, refresh_min_interval_s: float = 1.0, label: str = 'VHI Movements')

Stateful widget — instantiate once at module level, call .ui() per frame.

Example::

panel = VhiMovementPanel(vhi_client)

@app.ui
def ui(ctx):
    with grid[8, 0]:
        panel.ui()

Parameters:

Name Type Description Default
client VhiControlClient

The :class:VhiControlClient used to fetch state and dispatch SetMovement commands.

required
on_movement Callable[[str], None] | None

Click handler for a movement button. Defaults to client.set_movement; pass a wrapper to layer side-effects (e.g. snap a session label, fire an edge-trigger).

None
refresh_min_interval_s float

Minimum seconds between background state refreshes. Default 1 s.

1.0
label str

Panel header text rendered above the button grid.

'VHI Movements'
Source code in myogestic/widgets/vhi_movement_panel.py
def __init__(
    self,
    client: VhiControlClient,
    *,
    on_movement: Callable[[str], None] | None = None,
    refresh_min_interval_s: float = 1.0,
    label: str = "VHI Movements",
) -> None:
    self._client = client
    self._cache = VhiStateCache()
    self._on_movement = on_movement or client.set_movement
    self._refresh_interval = refresh_min_interval_s
    self._label = label

ui

ui() -> None

Render the panel — call once per frame inside @app.ui.

Source code in myogestic/widgets/vhi_movement_panel.py
def ui(self) -> None:
    """Render the panel — call once per frame inside ``@app.ui``."""
    request_vhi_state_refresh(
        self._client, self._cache, min_interval_s=self._refresh_interval
    )
    snap = self._cache.snapshot()
    vhi_movement_palette(
        snap.movements,
        connected=snap.connected,
        current_movement=snap.current_movement,
        status=snap.message,
        on_movement=self._on_movement,
        on_refresh=lambda: request_vhi_state_refresh(
            self._client, self._cache, force=True
        ),
        label=self._label,
    )

VhiMovementPanel

Lower-level pieces

VhiMovementPanel wraps these for the common case. Reach for them directly when you want to share one state cache across multiple panels, or render the palette without owning a client.

vhi_movement_palette

vhi_movement_palette(movements: Sequence[str], *, on_movement: Callable[[str], None], on_refresh: Callable[[], None] | None = None, current_movement: str = '', connected: bool = False, status: str = '', label: str = 'VHI Movements') -> None

Render VHI's movement names as a grid of command buttons.

Pure ImGui: performs no RPC and owns no client. movements is the cached list from the last successful GetState; on_movement(name) fires on click (wire it to VhiControlClient.set_movement). If on_refresh is given, a refresh button is drawn. Movement buttons are disabled while connected is False, but a stale list stays visible.

The grid uses as many columns as fit the panel width, so it reflows when the panel is resized; the button matching current_movement is highlighted.

Source code in myogestic/widgets/vhi_movement_palette.py
def vhi_movement_palette(
    movements: Sequence[str],
    *,
    on_movement: Callable[[str], None],
    on_refresh: Callable[[], None] | None = None,
    current_movement: str = "",
    connected: bool = False,
    status: str = "",
    label: str = "VHI Movements",
) -> None:
    """Render VHI's movement names as a grid of command buttons.

    Pure ImGui: performs no RPC and owns no client. ``movements`` is the cached
    list from the last successful ``GetState``; ``on_movement(name)`` fires on
    click (wire it to ``VhiControlClient.set_movement``). If ``on_refresh`` is
    given, a refresh button is drawn. Movement buttons are disabled while
    ``connected`` is False, but a stale list stays visible.

    The grid uses as many columns as fit the panel width, so it reflows when
    the panel is resized; the button matching ``current_movement`` is highlighted.
    """
    panel_header(label, fa.ICON_FA_HAND)

    if on_refresh is not None:
        if imgui.button(f"{fa.ICON_FA_ARROWS_ROTATE}  Refresh from VHI"):
            on_refresh()
        imgui.same_line()
    imgui.text_colored(_DOT_OK if connected else _DOT_BAD, fa.ICON_FA_CIRCLE)
    imgui.same_line()
    imgui.text_disabled(status or ("connected" if connected else "not connected"))

    if not movements:
        imgui.spacing()
        imgui.text_disabled("No movements yet — launch VHI, then refresh.")
        return

    imgui.spacing()
    # As many columns as fit the current panel width (at least one).
    avail = imgui.get_content_region_avail().x
    spacing = imgui.get_style().item_spacing.x
    n_cols = max(1, int((avail + spacing) // (_BTN_W + spacing)))

    imgui.begin_disabled(not connected)
    for i, name in enumerate(movements):
        if i % n_cols != 0:
            imgui.same_line()
        is_current = name == current_movement
        if is_current:
            imgui.push_style_color(imgui.Col_.button, _BTN_CURRENT)
        if imgui.button(f"{name}##vhi_mv_{i}", imgui.ImVec2(_BTN_W, _BTN_H)):
            on_movement(name)
        if is_current:
            imgui.pop_style_color()
    imgui.end_disabled()

VhiStateCache dataclass

VhiStateCache(movements: list[str] = list(), current_movement: str = '', current_state: str = '', mode: str = '', connected: bool = False, refreshing: bool = False, message: str = 'Launch VHI, then refresh.', last_attempt_s: float = 0.0, lock: Lock = Lock())

Last-known VHI state, refreshed off-thread. Use snapshot() to read.

snapshot

snapshot() -> VhiStateSnapshot

Return a consistent, immutable view — safe to read all frame.

Source code in myogestic/widgets/vhi_movement_palette.py
def snapshot(self) -> VhiStateSnapshot:
    """Return a consistent, immutable view — safe to read all frame."""
    with self.lock:
        return VhiStateSnapshot(
            movements=tuple(self.movements),
            current_movement=self.current_movement,
            current_state=self.current_state,
            mode=self.mode,
            connected=self.connected,
            message=self.message,
        )

VhiStateSnapshot dataclass

VhiStateSnapshot(movements: tuple[str, ...], current_movement: str, current_state: str, mode: str, connected: bool, message: str)

An immutable, lock-free view of VhiStateCache for one UI frame.

request_vhi_state_refresh

request_vhi_state_refresh(client: VhiControlClient, cache: VhiStateCache, *, force: bool = False, min_interval_s: float = 1.0) -> None

Start at most one throttled background GetState refresh.

Safe to call every frame from @app.ui: it returns immediately unless a refresh is due (min_interval_s elapsed, or force) and none is already in flight. The blocking get_state() runs on a daemon thread; the result lands in cache under its lock.

Source code in myogestic/widgets/vhi_movement_palette.py
def request_vhi_state_refresh(
    client: VhiControlClient,
    cache: VhiStateCache,
    *,
    force: bool = False,
    min_interval_s: float = 1.0,
) -> None:
    """Start at most one throttled background ``GetState`` refresh.

    Safe to call every frame from ``@app.ui``: it returns immediately unless a
    refresh is due (``min_interval_s`` elapsed, or ``force``) and none is
    already in flight. The blocking ``get_state()`` runs on a daemon thread;
    the result lands in ``cache`` under its lock.
    """
    now = time.monotonic()
    with cache.lock:
        if cache.refreshing or (not force and now - cache.last_attempt_s < min_interval_s):
            return
        cache.refreshing = True
        cache.last_attempt_s = now

    def _worker() -> None:
        reply = client.get_state()
        with cache.lock:
            cache.refreshing = False
            if reply is None:
                cache.connected = False
                cache.message = "VHI not reachable — launch it, then refresh."
                return
            cache.connected = True
            cache.movements = list(reply.available_movements)
            cache.current_movement = reply.current_movement
            cache.current_state = reply.current_state
            cache.mode = reply.mode
            cache.message = f"{reply.mode} mode · {len(cache.movements)} movements"

    threading.Thread(target=_worker, daemon=True, name="vhi-state-refresh").start()