Session¶

Read back an on-disk recording session: iterate labeled or aligned windows for training, or open the raw store. See Record and replay.

Reading sessions¶

Three window iterators, differing only in what each window is paired with. iter_labeled_windows gives a class index from the label track. iter_target_windows gives the value of a recorded TargetSource stream at the window's end — the causal choice, and the one to reach for whenever the target is graded rather than cued; see Record for proportional control. iter_aligned_windows gives a sample from one or more arbitrary streams nearest the window's midpoint.

open_session_store ¶

open_session_store(path: str | Path) -> Session

Open a saved session folder or .session.zip as a read-only Session.

Folder sessions keep the existing layout and use zarr.open_array(str(path / "<stream>.zarr"), mode="r"). Zip sessions use a zarr.storage.ZipStore and read the same array paths inside the archive.

Examples:

>>> from myogestic.session import open_session_store
>>> session = open_session_store("sessions/demo.session.zip")
>>> info = session.stream_info("emg")
>>> session.close()

iter_labeled_windows ¶

iter_labeled_windows(paths: list[str] | list[Path], stream_name: str, window_ms: float, hop_ms: float, classes: set[int] | None = None) -> Iterator[tuple[ndarray, ndarray, int]]

Yield (window, ts, class_index) triples from labeled segments.

window is channels-first (n_channels, n_samples) — the library's standard signal layout. ts is the matching 1-D timestamp array. Walks each session's label track, finds the time interval each label covers (this label's timestamp to next label's timestamp), and chops that interval into fixed-size windows. Works for folders and .session.zip sessions.

Examples:

>>> from myogestic.session import iter_labeled_windows
>>> for window, timestamps, class_index in iter_labeled_windows(
...     ["sessions/demo.session.zip"], "emg", 200, 100
... ):
...     print(window.shape, class_index)

iter_target_windows ¶

iter_target_windows(paths: list[str] | list[Path], stream_name: str, target_stream_name: str, window_ms: float, hop_ms: float, phases: set[str] | None = None) -> Iterator[tuple[ndarray, ndarray, float]]

Yield (window, ts, target) triples against a recorded continuous target.

The continuous sibling of iter_labeled_windows. That one walks the label track and hands back a class index, so a session of three cued classes holds three distinct target values — a regressor trained on -1 / 0 / +1 has never seen an intermediate one, and nothing in the data says what half a contraction should produce, so whatever it emits there comes from the estimator's own bias rather than the recording. This one pairs each window with the value of a recorded target stream — a myogestic.sources.target.TargetSource the subject was following — so the targets densely cover the range and interpolation is learned rather than hoped for.

window is channels-first (n_channels, n_samples) and ts the matching 1-D timestamp array, exactly as in iter_labeled_windows. target is channel 0 of the target stream, in whatever units it was recorded in (percent MVC for a force ramp, signed [-1, +1] for a cursor) — this function does not rescale it.

The target is the value at the window's END. The model is causal: it sees a trailing window of EMG and must emit the command for now, so the instant the window ends is the instant it is answering for. Taking the centre would train it to predict the past by half a window, and the mean would train it to output a smeared average that lags every ramp — both produce a model that scores well offline and drags in the hand.

Alignment is by timestamp, never by index. The target stream is typically far slower than the EMG (100 Hz beside 2 kHz) and starts and stops at its own chunk boundaries, so sample i of one is not sample i of the other. The lookup starts from the newest target sample at or before the window's end. A window whose end falls outside the target stream's own time span is dropped, not clamped to the first or last recorded value, because clamping would invent ground truth for the seconds the subject was not being shown anything. A window ending inside a hole in the target stream is dropped for exactly that reason: a gap wider than _MAX_GAP_PERIODS sample periods is a dropout, and the span it covers is as unrecorded as the span past the end.

The value is interpolated only within one phase. Inside a phase the trajectory is smooth and slow, so numpy.interp between the two bracketing target samples reads it more finely than the target's own rate. Across a phase change the value is held at the same sample that supplied the phase, because the value can step there — TargetSource.stop drops the level and switches to idle in a single sample — and interpolating through a step produces a level the subject was never shown: a plausible-looking label for effort that never happened.

A window must also span the time it claims to. Windows are cut by sample index, so wherever the signal stream dropped out, win_samples consecutive recorded samples reach back further than window_ms. Such a window is dropped once it spans more than _MAX_SPAN_SLACK times its nominal length: the target at its end is not the answer for samples much older than it, which is the causal argument above failing rather than an edge case of it.

Phases. Channel 1 carries the phase code (PHASE_CODES). By default only windows ending in a phase where the subject is actually following the target are yielded — every name in PHASE_CODES except "idle" (no block running) and "done" (block finished). Both sit at target 0 while the subject does whatever they like, which is exactly the data that teaches a model that silence means zero and that zero means silence. Pass phases to override, e.g. phases=set(PHASE_CODES) to keep everything, or {"hold"} for plateaus only.

A window straddling a phase boundary is classified by its end, the same instant that gives it its target, and both are read from the same target sample, so the two can never disagree. A window reaching from rest into ramp_up is a ramp_up window and is kept; one whose last sample lands past the end of the block is done and is dropped, along with the EMG that trails it.

Parameters:

Name	Type	Description	Default
`paths`	`list[str] \| list[Path]`	Session locations — folders or `.session.zip` archives.	required
`stream_name`	`str`	The signal to window, e.g. `"emg"`.	required
`target_stream_name`	`str`	The recorded `TargetSource` stream, e.g. `"target"`.	required
`window_ms`	`float`	Window length in milliseconds.	required
`hop_ms`	`float`	Step between consecutive window starts, in milliseconds.	required
`phases`	`set[str] \| None`	Phase names (not codes) to keep. `None` means every phase except `"idle"` and `"done"`.	`None`

Raises:

Type Description

ValueError

If window_ms or hop_ms is not positive, if phases names a phase that is not in PHASE_CODES, or if a session is missing the target stream, recorded it empty, or did not record it from a TargetSource — fewer than that source's two channels, or channel names that are not its ["target_pct", "phase"]. Unlike a missing signal stream — which is skipped with a log line, as in iter_labeled_windows — a missing target is raised: this iterator exists for the target, and silently dropping the one session recorded without it means training on two thirds of the data and never being told.

Examples:

>>> from myogestic.session import iter_target_windows
>>> for window, timestamps, target in iter_target_windows(
...     ["sessions/demo.session.zip"], "emg", "target", 200, 100
... ):
...     print(window.shape, target)

iter_aligned_windows ¶

iter_aligned_windows(paths: list[str] | list[Path], primary_stream_name: str, aligned_stream_names: list[str], window_ms: float, hop_ms: float, n_alignment_samples: int = 1, *, with_names: bool = False) -> Iterator[tuple[ndarray, dict[str, Any], ndarray]]

Yield (primary_window, aligned, ts) for regression training.

primary_window is channels-first (n_channels, n_samples). For each primary window, find the nearest sample in every aligned stream at the window midpoint and average n_alignment_samples around that index.

With with_names=True each aligned stream's value is a dict[channel_name, float] instead of a bare vector, so a training script selects a target by name rather than by wire position. That is the difference between a script that keeps working when a configuration is reordered and one that silently trains on the wrong channel. Requires the recording to carry channel names; it raises naming the stream if it does not.

Examples:

>>> from myogestic.session import iter_aligned_windows
>>> for window, aligned, timestamps in iter_aligned_windows(
...     ["sessions/demo.session.zip"], "emg", ["vhi_control"], 200, 50
... ):
...     target = aligned["vhi_control"]

split_sessions_by_stream ¶

split_sessions_by_stream(paths: Iterable[P], stream: str) -> SessionSplit[P]

Sort session paths by whether they recorded stream, without holding them open.

The question every mixed training callback asks first: sessions with a kinematics stream train against it, the rest fall back to synthetic targets from their labels.

Each session is opened only to read its store list and is closed again straight away — an open zarr.storage.ZipStore keeps a lock on the .session.zip, which on Windows blocks deleting or moving the file afterwards.

Parameters:

Name	Type	Description	Default
`paths`	`Iterable[P]`	Session locations — folders or `.session.zip` archives, e.g. `data.paths`.	required
`stream`	`str`	The stream name to test for, e.g. `"vhi_control"`.	required

Returns:

Type	Description
`SessionSplit`	The three-way outcome, each list in the order the paths came in.

Examples:

>>> from myogestic.session import split_sessions_by_stream
>>> kin, labels, unreadable = split_sessions_by_stream([], "vhi_control")
>>> kin, labels, unreadable
([], [], [])

SessionSplit ¶

Bases: NamedTuple

Which sessions carry a stream, which do not, and which could not be opened.

Attributes:

Name	Type	Description
`with_stream`	`list[P]`	Paths whose recording contains the stream — the `iter_aligned_windows` set.
`without_stream`	`list[P]`	Paths that opened fine and simply do not have it — the `iter_labeled_windows` fallback set.
`unreadable`	`list[tuple[P, Exception]]`	`(path, exception)` for every path that would not open at all. Returned rather than logged, because where a skipped session is reported belongs to the caller: a training callback puts it in the app's own log, not in `logging`.

Data model¶

Session ¶

Session(base_path: str = 'sessions')

One recording session on disk: per-stream Zarr arrays + a label track.

Created when the user clicks Record, finalised when they click Stop. While active, every acquisition thread with a registered stream appends to the session's Zarr stores; UI label clicks emit LabelEvent entries onto the label track. Closing writes meta.json and labels.json alongside the Zarr folders, and optionally packs the tree into a portable .session.zip.

Layout on disk (one folder per recording, named with the start timestamp)::

sessions/2026-05-17_14-23-05/
    emg.zarr/                  # shape (n_samples, n_channels)
    emg_timestamps.zarr/       # shape (n_samples,) float64
    vhi_control.zarr/          # any additional stream
    vhi_control_timestamps.zarr/
    meta.json                  # streams_info, app_name, class_names
    labels.json                # the LabelEvent list

Read sessions back with open_session_store, which handles both folders and .session.zip archives.

Parameters:

Name	Type	Description	Default
`base_path`	`str`	Parent directory; the session creates a timestamp-named subdirectory inside. Default `"sessions"` (created if missing).	`'sessions'`

Examples:

>>> from myogestic.session import Session
>>> session = Session("sessions")
>>> session.add_label(0, timestamp=123.0)

Methods:

Name	Description
`init_stream`	Called once per stream when recording starts.
`append`	Called from acquire loop when recording. data: (n_samples, n_channels).
`add_label`	Append a label event to the session's label track.
`save_meta`	Write meta.json + labels.json to the session folder.
`pack_to_zip`	Pack the session folder into a single `<name>.session.zip` file.
`discard`	Delete this session's folder and everything recorded into it.
`get_trials`	Extract discrete labeled windows for classification training.
`get_continuous`	Return full stream data + timestamps for regression training.
`stream_info`	Public accessor for a stream's StreamInfo.
`close`	Release file handles held by this session.

init_stream ¶

init_stream(stream_name: str, info: StreamInfo) -> None

Called once per stream when recording starts.

append ¶

append(stream_name: str, data: ndarray, timestamps: ndarray) -> None

Called from acquire loop when recording. data: (n_samples, n_channels).

add_label ¶

add_label(class_index: int, timestamp: float | None = None) -> None

Append a label event to the session's label track.

Parameters:

Name	Type	Description	Default
`class_index`	`int`	Class index for the event; `-1` marks a rest / no-class boundary.	required
`timestamp`	`float \| None`	Event time in seconds (`mne_lsl` clock). Defaults to the current `local_clock()` when omitted.	`None`

save_meta ¶

save_meta(app_name: str, class_names: list[str] | None = None, control_space: Mapping[str, object] | None = None) -> None

Write meta.json + labels.json to the session folder.

Parameters:

Name	Type	Description	Default
`app_name`	`str`	Identifier for the producing app.	required
`class_names`	`list[str] \| None`	Optional human-readable names for label class indices. Persisting them keeps old sessions self-describing: readers can render labels without an external lookup.	`None`
`control_space`	`Mapping[str, object] \| None`	Optional control configuration this recording was made under, as produced by `ControlMap.as_control_space()`. Records what each number meant — the alias it came from and the target control it drove. Carries a `format` tag; read it back with `myogestic.controls.read_control_space`, which refuses a pre-alias control space by name instead of reinterpreting it.	`None`

pack_to_zip ¶

pack_to_zip() -> Path

Pack the session folder into a single <name>.session.zip file.

Uses ZIP_STORED: zarr chunks are already compressed internally, so an outer compression layer would add CPU for little gain.

discard ¶

discard() -> None

Delete this session's folder and everything recorded into it.

For a recording the operator threw away — a false start, a bad trial. Releases the Zarr handles first, exactly as pack_to_zip does: on Windows the folder cannot be removed while any chunk file is still open.

get_trials ¶

get_trials(stream_name: str, pre_s: float = 0.0, post_s: float = 0.0, class_names: list[str] | None = None) -> list[Recording]

Extract discrete labeled windows for classification training.

get_continuous ¶

get_continuous(stream_name: str) -> tuple[ndarray, ndarray]

Return full stream data + timestamps for regression training.

stream_info ¶

stream_info(stream_name: str) -> StreamInfo

Public accessor for a stream's StreamInfo.

close ¶

close() -> None

Release file handles held by this session.

Closes the ZipStore opened by open_session_store for a .session.zip and drops the array references. Safe to call more than once. On Windows an open ZipStore keeps the archive locked, so close the session before moving or deleting the .session.zip — use it as a context manager (with open_session_store(p) as s: ...).

LabelEvent `dataclass` ¶

LabelEvent(timestamp: float, class_index: int)

One entry in a session's label track: "at LSL time T, the user picked class N".

Recorded whenever the user clicks a class button in RecordingControls. The label track is a chronological list of these events; the recording-window iterators (iter_labeled_windows, iter_aligned_windows) walk it to decide which sample range gets which class index.

Attributes:

Name	Type	Description
`timestamp`	`float`	LSL clock time (seconds) when the label was emitted. Mint one by hand with `mne_lsl.lsl.local_clock()`.
`class_index`	`int`	Index into the session's `class_names` list. `-1` is the unlabeled sentinel (the iterators skip it).

Examples:

>>> from myogestic.session import LabelEvent
>>> event = LabelEvent(timestamp=123.0, class_index=1)
>>> event.class_index
1

Recording `dataclass` ¶

Recording(class_index: int, class_name: str, data: ndarray, timestamps: ndarray)

A single labeled trial, extracted from a Session.

Examples:

>>> import numpy as np
>>> from myogestic.session import Recording
>>> recording = Recording(
...     1, "Fist", np.zeros((3, 2), dtype=np.float32), np.arange(3, dtype=np.float64)
... )
>>> (recording.class_name, recording.data.shape)
('Fist', (3, 2))

Session¶

Reading sessions¶

open_session_store ¶

iter_labeled_windows ¶

iter_target_windows ¶

iter_aligned_windows ¶

split_sessions_by_stream ¶

SessionSplit ¶

Data model¶

Session ¶

init_stream ¶

append ¶

add_label ¶

save_meta ¶

pack_to_zip ¶

discard ¶

get_trials ¶

get_continuous ¶

stream_info ¶

close ¶

LabelEvent dataclass ¶

Recording dataclass ¶

LabelEvent `dataclass` ¶

Recording `dataclass` ¶