Skip to content

API reference

Generated from the docstrings in src/ssvep by mkdocstrings, so it cannot drift from the code the way a hand-written reference does. If something here is wrong, fix the docstring.

The decode, spectral and metrics layers take plain arrays plus sfreq — shape (n_channels, n_times) — and nothing else. That boundary is deliberate: the same function runs offline over a whole recording and online over a single sliding window, so an online result and an offline one cannot disagree because they went through different code.

ssvep.classification — the decode engine

Decoding

ssvep.classification.decoding

SSVEP frequency-recognition decoders (domain-generic, array-based).

All functions take plain NumPy arrays and a sampling rate, so the same code runs offline (batch) and online (per-window in an LSL loop). No MNE / BIDS / study dependencies.

  • CCA, FBCCA: calibration-free (sinusoidal reference templates; no training).
  • TRCA: template-trained; gate it with trca_feasible() — it needs several phase-consistent trials per class and degrades to chance otherwise.

Trial array convention: (n_channels, n_times); batches: (n_trials, n_channels, n_times). Reference frequencies are in Hz.

TRCA

Ensemble Task-Related Component Analysis (Nakanishi et al. 2018).

Calibration-based: one spatial filter + template per class. Requires several phase-consistent trials per class — check trca_feasible(y) first.

Epoch-length rules (every epoch must be time-locked to stimulus onset): - Within a class, training epochs must be equal length — the cross-trial covariance is sample-aligned, so ragged within-class input raises. - Across classes, templates may differ in length: each class's filter and template are learned independently, with no cross-frequency coupling. So a whole-number-of-cycles (per-frequency) window is fine, and there is no need for a common length across classes. - predict compares a test window to each class over their shared onset-locked interval (both cropped to the shorter), so the test window need not match the training length either. - Caveat: a correlation over more samples trends higher, so scoring classes at different lengths biases the argmax toward the longer templates. Prefer a common decision-window length unless you specifically want per-frequency lengths.

Source code in src/ssvep/classification/decoding.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
class TRCA:
    """Ensemble Task-Related Component Analysis (Nakanishi et al. 2018).

    Calibration-based: one spatial filter + template per class. Requires several
    phase-consistent trials per class — check `trca_feasible(y)` first.

    Epoch-length rules (every epoch must be time-locked to stimulus onset):
    - Within a class, training epochs must be equal length — the cross-trial
      covariance is sample-aligned, so ragged within-class input raises.
    - Across classes, templates may differ in length: each class's filter and
      template are learned independently, with no cross-frequency coupling. So a
      whole-number-of-cycles (per-frequency) window is fine, and there is no need
      for a common length across classes.
    - `predict` compares a test window to each class over their shared
      onset-locked interval (both cropped to the shorter), so the test window need
      not match the training length either.
    - Caveat: a correlation over more samples trends higher, so scoring classes at
      different lengths biases the argmax toward the longer templates. Prefer a
      common decision-window length unless you specifically want per-frequency
      lengths.
    """

    def __init__(self, n_harmonics: int = 3):
        self.n_harmonics = n_harmonics
        self.classes_ = self.templates_ = self.W_ = None

    @staticmethod
    def _trca_filter(trials):
        n_tr, n_ch, _ = trials.shape
        S = np.zeros((n_ch, n_ch))
        for i in range(n_tr):
            xi = trials[i] - trials[i].mean(1, keepdims=True)
            for j in range(n_tr):
                if i != j:
                    xj = trials[j] - trials[j].mean(1, keepdims=True)
                    S += xi @ xj.T
        cat = np.concatenate([t - t.mean(1, keepdims=True) for t in trials], axis=1)
        Q = cat @ cat.T
        try:
            vals, vecs = np.linalg.eig(np.linalg.solve(Q, S))
        except np.linalg.LinAlgError:
            vals, vecs = np.linalg.eig(np.linalg.pinv(Q) @ S)
        return vecs[:, np.argmax(vals.real)].real

    @staticmethod
    def _class_trials(X, y, c):
        """Trials for class ``c`` as ``(n_tr, n_ch, n_times)``.

        Accepts an ``(n_trials, n_ch, n_times)`` array or a sequence of
        ``(n_ch, n_times)`` trials (lengths may vary across classes). Unequal
        lengths *within* the class are an error, not a silent truncation.
        """
        trials = [np.asarray(X[i]) for i in np.where(y == c)[0]]
        try:
            return np.stack(trials)
        except ValueError as e:
            raise ValueError(
                f"TRCA: class {c!r} has unequal-length epochs; within a class, "
                "training epochs must be equal length and onset-aligned"
            ) from e

    def fit(self, X, y):
        y = np.asarray(y)
        self.classes_ = np.unique(y)
        filters, templates = [], {}
        for c in self.classes_:
            trials = self._class_trials(X, y, c)
            filters.append(self._trca_filter(trials))
            templates[c] = trials.mean(0)
        self.W_ = np.column_stack(filters)
        self.templates_ = templates
        return self

    def predict(self, X):
        preds = []
        for trial in X:
            trial = np.asarray(trial)
            scores = []
            for c in self.classes_:
                tmpl = self.templates_[c]
                n = min(trial.shape[-1], tmpl.shape[-1])   # shared onset-locked span
                r = np.corrcoef((self.W_.T @ trial[:, :n]).ravel(),
                                (self.W_.T @ tmpl[:, :n]).ravel())[0, 1]
                scores.append(0.0 if np.isnan(r) else r)
            preds.append(self.classes_[int(np.argmax(scores))])
        return np.array(preds)

rendered_frequency 

rendered_frequency(nominal, refresh_hz, method)

Actual displayed flicker frequency given the render method.

  • sine: continuous-phase -> exact nominal frequency.
  • square/even: integer frames per cycle -> refresh/round(refresh/f) (drifts from nominal at high freq / low frames-per-cycle).
Source code in src/ssvep/classification/decoding.py 
23
24
25
26
27
28
29
30
31
32
33
def rendered_frequency(nominal: float, refresh_hz: float, method: str) -> float:
    """Actual displayed flicker frequency given the render method.

    - ``sine``: continuous-phase -> exact nominal frequency.
    - ``square``/``even``: integer frames per cycle -> ``refresh/round(refresh/f)``
      (drifts from nominal at high freq / low frames-per-cycle).
    """
    if method.startswith("sine"):
        return float(nominal)
    fpc = max(int(round(refresh_hz / nominal)), 2)
    return float(refresh_hz / fpc)

expand_subepochs 

expand_subepochs(X, y, win_samples, step_samples)

Split each trial into shorter sub-epochs; returns (Xs, ys, groups).

groups is the originating trial index so cross-validation can keep a trial's sub-epochs together (no leakage). NOTE: sub-epochs cut at arbitrary offsets have different SSVEP phase — fine for CCA, but they break TRCA's time-domain template averaging unless aligned to the stimulus cycle.

Source code in src/ssvep/classification/decoding.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def expand_subepochs(X: np.ndarray, y: np.ndarray, win_samples: int,
                     step_samples: int):
    """Split each trial into shorter sub-epochs; returns (Xs, ys, groups).

    ``groups`` is the originating trial index so cross-validation can keep a
    trial's sub-epochs together (no leakage). NOTE: sub-epochs cut at arbitrary
    offsets have different SSVEP phase — fine for CCA, but they break TRCA's
    time-domain template averaging unless aligned to the stimulus cycle.
    """
    n_times = X.shape[-1]
    starts = range(0, n_times - win_samples + 1, step_samples)
    Xs, ys, groups = [], [], []
    for i, (trial, label) in enumerate(zip(X, y)):
        for s in starts:
            Xs.append(trial[:, s:s + win_samples]); ys.append(label); groups.append(i)
    return np.array(Xs), np.array(ys), np.array(groups)

reference_signals 

reference_signals(freq, n_harmonics, n_times, sfreq)

Sin/cos reference matrix (n_times, 2*n_harmonics) for one frequency.

Source code in src/ssvep/classification/decoding.py 
57
58
59
60
61
62
63
64
65
def reference_signals(freq: float, n_harmonics: int, n_times: int,
                      sfreq: float) -> np.ndarray:
    """Sin/cos reference matrix ``(n_times, 2*n_harmonics)`` for one frequency."""
    t = np.arange(n_times) / sfreq
    cols = []
    for h in range(1, n_harmonics + 1):
        cols.append(np.sin(2 * np.pi * h * freq * t))
        cols.append(np.cos(2 * np.pi * h * freq * t))
    return np.column_stack(cols)

cca_scores 

cca_scores(trial, freqs, sfreq, n_harmonics=3)

Max canonical correlation of one trial (n_ch, n_times) to each freq.

Source code in src/ssvep/classification/decoding.py 
79
80
81
82
83
84
85
def cca_scores(trial: np.ndarray, freqs, sfreq: float,
               n_harmonics: int = 3) -> np.ndarray:
    """Max canonical correlation of one trial ``(n_ch, n_times)`` to each freq."""
    X = trial.T
    n_times = X.shape[0]
    return np.array([_canon_corr_max(X, reference_signals(f, n_harmonics, n_times, sfreq))
                     for f in freqs])

fbcca_scores 

fbcca_scores(trial, freqs, sfreq, band_low, band_high, n_harmonics=3, n_subbands=5, a=1.25, b=0.25)

Filter-bank CCA combined scores (Chen et al. 2015), one per candidate freq.

Source code in src/ssvep/classification/decoding.py 
111
112
113
114
115
116
117
118
119
def fbcca_scores(trial, freqs, sfreq, band_low, band_high, n_harmonics=3,
                 n_subbands=5, a=1.25, b=0.25) -> np.ndarray:
    """Filter-bank CCA combined scores (Chen et al. 2015), one per candidate freq."""
    subbands = _fb_subbands(band_low, band_high, n_subbands, sfreq)
    weights = np.array([k ** (-a) + b for k in range(1, len(subbands) + 1)])
    total = np.zeros(len(freqs))
    for w, (lo, hi) in zip(weights, subbands):
        total += w * cca_scores(_bandpass(trial, lo, hi, sfreq), freqs, sfreq, n_harmonics) ** 2
    return total

trca_feasible 

trca_feasible(y, min_trials_per_class=4)

True if every class has >= min_trials_per_class trials (TRCA pre-check).

Source code in src/ssvep/classification/decoding.py 
125
126
127
128
def trca_feasible(y, min_trials_per_class: int = 4) -> bool:
    """True if every class has >= min_trials_per_class trials (TRCA pre-check)."""
    _, counts = np.unique(y, return_counts=True)
    return len(counts) >= 2 and counts.min() >= min_trials_per_class

classify 

classify(trial, freqs, sfreq, method='fbcca', n_harmonics=3, band=None, **fb_kw)

Predict the attended-frequency index for one trial. Calibration-free.

method='cca' or 'fbcca' (fbcca needs band=(low, high)).

Source code in src/ssvep/classification/decoding.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def classify(trial, freqs, sfreq, method="fbcca", n_harmonics=3,
             band=None, **fb_kw) -> int:
    """Predict the attended-frequency index for one trial. Calibration-free.

    method='cca' or 'fbcca' (fbcca needs band=(low, high)).
    """
    if method == "cca":
        s = cca_scores(trial, freqs, sfreq, n_harmonics)
    elif method == "fbcca":
        if band is None:
            band = (min(freqs), max(freqs))
        s = fbcca_scores(trial, freqs, sfreq, band[0], band[1], n_harmonics, **fb_kw)
    else:
        raise ValueError(f"unknown calibration-free method: {method}")
    return int(np.argmax(s))

Spectral analysis

ssvep.classification.spectral

Spectral analysis and SSVEP SNR (array-based, scipy only — no MNE).

SNR follows the standard SSVEP definition (Norcia et al. 2015): power at the stimulus frequency divided by the mean power of neighbouring bins, excluding a small guard band, reported in dB. A harmonic-summed variant aggregates f,2f,3f.

Data convention: last axis is time. (n_times,), (n_ch, n_times) or (n_trials, n_ch, n_times) all work; PSD/SNR keep the leading axes.

compute_psd 

compute_psd(data, sfreq, fmin=1.0, fmax=None, nperseg_s=4.0)

Welch PSD. Returns (freqs, psd) with frequency on the last axis.

Source code in src/ssvep/classification/spectral.py 
16
17
18
19
20
21
22
23
24
def compute_psd(data, sfreq, fmin=1.0, fmax=None, nperseg_s=4.0):
    """Welch PSD. Returns (freqs, psd) with frequency on the last axis."""
    n_times = data.shape[-1]
    nperseg = min(int(nperseg_s * sfreq), n_times)
    freqs, psd = welch(data, fs=sfreq, nperseg=nperseg,
                       noverlap=nperseg // 2, axis=-1)
    fmax = fmax or sfreq / 2.0
    keep = (freqs >= fmin) & (freqs <= fmax)
    return freqs[keep], psd[..., keep]

snr_spectrum 

snr_spectrum(psd, freqs, neighbor_hz=1.0, guard_hz=0.2)

SNR at every bin: power / mean(neighbouring bins, guard excluded). Same shape.

Source code in src/ssvep/classification/spectral.py 
27
28
29
30
31
32
33
34
35
36
def snr_spectrum(psd, freqs, neighbor_hz=1.0, guard_hz=0.2):
    """SNR at every bin: power / mean(neighbouring bins, guard excluded). Same shape."""
    bin_w = float(freqs[1] - freqs[0])
    n_neigh = max(int(round(neighbor_hz / bin_w)), 1)
    n_guard = max(int(round(guard_hz / bin_w)), 0)
    kernel = np.concatenate([np.ones(n_neigh), np.zeros(n_guard * 2 + 1), np.ones(n_neigh)])
    kernel /= kernel.sum()
    noise = np.apply_along_axis(lambda m: np.convolve(m, kernel, mode="same"), -1, psd)
    with np.errstate(divide="ignore", invalid="ignore"):
        return psd / noise

snr_at 

snr_at(freqs, snr_spec, f)

Index an SNR spectrum at the bin nearest frequency f.

Source code in src/ssvep/classification/spectral.py 
39
40
41
def snr_at(freqs, snr_spec, f):
    """Index an SNR spectrum at the bin nearest frequency ``f``."""
    return snr_spec[..., int(np.argmin(np.abs(freqs - f)))]

target_snr_db 

target_snr_db(psd, freqs, f, neighbor_hz=1.0, guard_hz=0.2, n_harmonics=1)

SNR (dB) at target f, optionally harmonic-summed over f,2f,3f...

Source code in src/ssvep/classification/spectral.py 
44
45
46
47
48
49
def target_snr_db(psd, freqs, f, neighbor_hz=1.0, guard_hz=0.2, n_harmonics=1):
    """SNR (dB) at target ``f``, optionally harmonic-summed over f,2f,3f..."""
    harmonics = [f * k for k in range(1, n_harmonics + 1) if f * k <= freqs[-1]]
    spec = snr_spectrum(psd, freqs, neighbor_hz, guard_hz)
    vals = np.stack([snr_at(freqs, spec, h) for h in harmonics], axis=0)
    return 10.0 * np.log10(vals.mean(axis=0))

target_snr_from_data 

target_snr_from_data(data, sfreq, f, n_harmonics=3, neighbor_hz=1.0, guard_hz=0.2, average_trials=True)

Convenience: per-channel SNR (dB) at f straight from a data array.

data (n_trials, n_ch, n_times) or (n_ch, n_times). With trials and average_trials, the PSD is averaged across trials first.

Source code in src/ssvep/classification/spectral.py 
52
53
54
55
56
57
58
59
60
61
62
def target_snr_from_data(data, sfreq, f, n_harmonics=3, neighbor_hz=1.0,
                         guard_hz=0.2, average_trials=True):
    """Convenience: per-channel SNR (dB) at ``f`` straight from a data array.

    data ``(n_trials, n_ch, n_times)`` or ``(n_ch, n_times)``. With trials and
    average_trials, the PSD is averaged across trials first.
    """
    freqs, psd = compute_psd(data, sfreq, fmin=max(1.0, f - 5), fmax=f * (n_harmonics + 1) + 5)
    if psd.ndim == 3 and average_trials:
        psd = psd.mean(axis=0)
    return target_snr_db(psd, freqs, f, neighbor_hz, guard_hz, n_harmonics)

Metrics

ssvep.classification.metrics

Classification metrics for SSVEP BCI: accuracy, macro precision, Wolpaw ITR.

Pure NumPy + scikit-learn. Rigor helper: always pass chance = 1/n_classes alongside any accuracy you report.

wolpaw_itr 

wolpaw_itr(accuracy, n_classes, selection_time_s)

Information transfer rate in bits/min (Wolpaw et al. 1998).

selection_time_s = decision window + inter-selection overhead. B = log2(N) + Plog2(P) + (1-P)log2((1-P)/(N-1)); ITR = B*60/selection_time.

Source code in src/ssvep/classification/metrics.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def wolpaw_itr(accuracy: float, n_classes: int, selection_time_s: float) -> float:
    """Information transfer rate in bits/min (Wolpaw et al. 1998).

    selection_time_s = decision window + inter-selection overhead.
    B = log2(N) + P*log2(P) + (1-P)*log2((1-P)/(N-1)); ITR = B*60/selection_time.
    """
    p = float(np.clip(accuracy, 0.0, 1.0))
    n = int(n_classes)
    if n <= 1 or selection_time_s <= 0:
        return 0.0
    if p >= 1.0:
        bits = np.log2(n)
    elif p <= 0.0:
        bits = 0.0
    else:
        bits = np.log2(n) + p * np.log2(p) + (1 - p) * np.log2((1 - p) / (n - 1))
    return float(max(bits, 0.0) * 60.0 / selection_time_s)

classification_metrics 

classification_metrics(y_true, y_pred, n_classes, selection_time_s, labels=None)

Accuracy, macro precision, ITR, and n for one set of predictions.

Source code in src/ssvep/classification/metrics.py 
31
32
33
34
35
36
37
38
39
40
41
def classification_metrics(y_true, y_pred, n_classes, selection_time_s, labels=None):
    """Accuracy, macro precision, ITR, and n for one set of predictions."""
    acc = accuracy_score(y_true, y_pred)
    return {
        "accuracy": float(acc),
        "precision_macro": float(precision_score(y_true, y_pred, labels=labels,
                                                 average="macro", zero_division=0)),
        "itr_bits_per_min": wolpaw_itr(acc, n_classes, selection_time_s),
        "n_trials": int(len(y_true)),
        "chance": 1.0 / n_classes,
    }

Visualisation

ssvep.classification.viz

Domain-generic SSVEP plotting primitives (matplotlib only).

  • plot_channel_array: discrete per-channel map (NO interpolation) for sparse / single-region montages where interpolated topomaps mislead. Bichromatic diverging for signed values (SNR), monochromatic sequential for non-negative values (impedance).
  • plot_window_sweep: mean ± SD vs decision-window length, with a chance line.
  • plot_confusion: normalised confusion matrix.

plot_channel_array 

plot_channel_array(positions, ch_names, values, ax=None, title=None, vlim=None, cbar=True, label='value', diverging=True, cmap=None, show_values=None, norm=None, cbar_ticks=None)

One coloured disc per channel at its 2-D montage position; no interpolation.

positions: dict ch -> (x, y) (e.g. cm from a reference landmark). Discs are drawn in data coordinates with a radius just under the nearest-neighbour spacing, so they never overlap regardless of channel count (a 64-ch cap reads as cleanly as an 8-ch montage); the figure grows with the montage. show_values prints the numeric value inside each disc — defaults to on for sparse montages (≤20 ch) and off for dense ones (colour + colourbar carry the value there); the label font shrinks with channel count so it fits the disc on a dense cap. diverging=True -> bichromatic, centred at 0 (default RdBu_r) for signed values. diverging=False -> sequential from 0 (default Greens) for non-negative values. norm (+ a cmap that may be a Colormap object) overrides the auto scale — e.g. a BoundaryNorm + ListedColormap for discrete pass/marginal/fail zones; cbar_ticks then labels the boundaries.

Source code in src/ssvep/classification/viz.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def plot_channel_array(positions, ch_names, values, ax=None, title=None, vlim=None,
                       cbar=True, label="value", diverging=True, cmap=None, show_values=None,
                       norm=None, cbar_ticks=None):
    """One coloured disc per channel at its 2-D montage position; no interpolation.

    positions: dict ``ch -> (x, y)`` (e.g. cm from a reference landmark). Discs are drawn in **data
    coordinates** with a radius just under the nearest-neighbour spacing, so they never overlap
    regardless of channel count (a 64-ch cap reads as cleanly as an 8-ch montage); the figure grows
    with the montage. ``show_values`` prints the numeric value inside each disc — defaults to on for
    sparse montages (≤20 ch) and off for dense ones (colour + colourbar carry the value there); the
    label font shrinks with channel count so it fits the disc on a dense cap.
    diverging=True  -> bichromatic, centred at 0 (default RdBu_r) for signed values.
    diverging=False -> sequential from 0 (default Greens) for non-negative values.
    ``norm`` (+ a ``cmap`` that may be a Colormap object) overrides the auto scale — e.g. a
    ``BoundaryNorm`` + ``ListedColormap`` for discrete pass/marginal/fail zones; ``cbar_ticks`` then
    labels the boundaries.
    """
    vals = np.asarray(values, float)
    finite = vals[np.isfinite(vals)]
    if norm is not None:
        cmap = cmap or "Greens"                                 # caller supplied the scale; use as-is
    elif diverging:
        cmap = cmap or "RdBu_r"
        m = (np.nanmax(np.abs(finite)) if finite.size else 1.0) if vlim is None else max(abs(vlim[0]), abs(vlim[1]))
        norm = TwoSlopeNorm(vcenter=0.0, vmin=-m or -1e-6, vmax=m or 1e-6)
    else:
        cmap = cmap or "Greens"
        vlim = vlim or (0.0, np.nanmax(finite) if finite.size else 1.0)
        norm = Normalize(vmin=vlim[0], vmax=max(vlim[1], vlim[0] + 1e-6))
    cmap_obj = cmap if isinstance(cmap, Colormap) else plt.get_cmap(cmap)
    xs = np.array([positions[c][0] for c in ch_names], float)
    ys = np.array([positions[c][1] for c in ch_names], float)
    n = len(ch_names)
    if show_values is None:
        show_values = n <= 20

    # Declutter for display: some montages place electrodes almost on top of each other (the actiCAP
    # schematic puts TP9/TP7 ~0.6 cm apart), and one near-coincident pair would force a tiny disc radius
    # for the WHOLE map. Nudge only the too-close pairs apart to a floor of 0.8×(typical spacing); a
    # well-spaced montage (8-ch) is untouched. Positions here are display-only — the montage is unchanged.
    xs, ys = _declutter(xs, ys, 0.8 * _typical_spacing(xs, ys))
    radius = 0.46 * _nearest_neighbour_dist(xs, ys)            # data units → discs (near-)touch at most
    if ax is None:
        span_x = (xs.max() - xs.min()) + 3 * radius
        span_y = (ys.max() - ys.min()) + 3 * radius
        base = float(np.clip(4.6 + 0.05 * n, 4.6, 9.0))         # bigger canvas for denser montages
        big = max(span_x, span_y)
        _, ax = plt.subplots(figsize=(base * span_x / big + (1.2 if cbar else 0.3),
                                      base * span_y / big + 0.6))

    faces = [cmap_obj(norm(v)) if np.isfinite(v) else (0.85, 0.85, 0.85, 1.0) for v in vals]
    ax.add_collection(PatchCollection([Circle((x, y), radius) for x, y in zip(xs, ys)],
                                      facecolors=faces, edgecolors="k", linewidths=0.8, zorder=2))

    # Shrink the label font as the montage densifies so it fits inside the (smaller) discs. The old
    # 200/√n constant saturated at the 9 pt cap for every montage ≤493 ch, so a 64-ch cap got the same
    # 9 pt as an 8-ch one and the 10-10 names overflowed. 40/√n gives ~9 pt at 8-16 ch, ~7 at 32, ~5 at 64.
    fontsize = float(np.clip(40.0 / np.sqrt(max(n, 1)), 4.5, 9.0))
    for c, x, y, v, face in zip(ch_names, xs, ys, vals, faces):
        r, g, b = face[:3]
        tcol = "white" if np.isfinite(v) and (0.299 * r + 0.587 * g + 0.114 * b) < 0.5 else "black"
        if show_values:
            txt = f"{c}\n{v:.1f}" if np.isfinite(v) else f"{c}\nn/a"
        else:
            txt = c
        ax.text(x, y, txt, ha="center", va="center", fontsize=fontsize, color=tcol, zorder=3)

    pad = 1.5 * radius
    ax.set_xlim(xs.min() - pad, xs.max() + pad)
    ax.set_ylim(ys.min() - pad, ys.max() + pad)
    ax.set_aspect("equal")
    ax.set_xticks([]); ax.set_yticks([])
    if title:
        ax.set_title(title, fontsize=10)
    if cbar:
        sm = plt.cm.ScalarMappable(norm=norm, cmap=cmap_obj)
        sm.set_array([])
        cb = ax.figure.colorbar(sm, ax=ax, fraction=0.046, label=label,   # figure-safe (embedded canvases)
                                ticks=cbar_ticks)
        if cbar_ticks is not None:
            cb.ax.tick_params(labelsize=8)
    return ax.figure

plot_impedance 

plot_impedance(positions, ch_names, kohm, ax=None, vmax=None, title='Electrode impedance', cmap=IMPEDANCE_CMAP, zones=None)

Discrete channel-array map of impedance (kΩ).

Thin wrapper over :func:plot_channel_array (sequential, non-negative). kohm aligns to ch_names; NaN renders as an 'n/a' (unmeasured) marker. vmax is the top of the colour scale; None auto-scales to at least 100 kΩ, expanding (rounded to the next 100) to cover the worst channel — so gel montages sit on a 0–100 scale while dry electrodes (which read many hundreds of kΩ) aren't all pinned to the top colour.

zones (e.g. :data:ACTICHAMP_IMPEDANCE_ZONES) switches the gradient for discrete traffic-light bands — a (upper_kohm, colour) list, worst bound last. Each channel takes the colour of the first band its value falls under; the colourbar shows the bands with ticks at the boundaries. vmax is ignored when zones is given.

Source code in src/ssvep/classification/viz.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def plot_impedance(positions, ch_names, kohm, ax=None, vmax=None, title="Electrode impedance",
                   cmap=IMPEDANCE_CMAP, zones=None):
    """Discrete channel-array map of impedance (kΩ).

    Thin wrapper over :func:`plot_channel_array` (sequential, non-negative). ``kohm`` aligns to
    ``ch_names``; NaN renders as an 'n/a' (unmeasured) marker. ``vmax`` is the top of the colour
    scale; ``None`` auto-scales to **at least 100 kΩ**, expanding (rounded to the next 100) to cover
    the worst channel — so gel montages sit on a 0–100 scale while dry electrodes (which read many
    hundreds of kΩ) aren't all pinned to the top colour.

    ``zones`` (e.g. :data:`ACTICHAMP_IMPEDANCE_ZONES`) switches the gradient for **discrete
    traffic-light bands** — a ``(upper_kohm, colour)`` list, worst bound last. Each channel takes the
    colour of the first band its value falls under; the colourbar shows the bands with ticks at the
    boundaries. ``vmax`` is ignored when ``zones`` is given.
    """
    if zones:
        ups = [float(u) for u, _ in zones]
        colours = [c for _, c in zones]
        worst = np.asarray(kohm, float)
        worst = worst[np.isfinite(worst)]
        top = float(np.nanmax(worst)) if worst.size else 0.0
        # A finite top boundary the colourbar can render; the last band covers everything above it.
        finite_top = max((ups[-2] if len(ups) >= 2 else 0.0) + 30.0, top * 1.05, 90.0)
        bounds = [0.0] + ups[:-1] + [finite_top]               # e.g. [0, 30, 60, finite_top]
        listed = ListedColormap(colours)
        norm = BoundaryNorm(bounds, listed.N, clip=True)        # >top → clipped into the last (red) band
        return plot_channel_array(positions, ch_names, kohm, ax=ax, title=title,
                                  label="impedance (kΩ)", diverging=False, cmap=listed, norm=norm,
                                  cbar_ticks=ups[:-1])           # tick the real thresholds (30, 60)
    if vmax is None:
        finite = np.asarray(kohm, float)
        finite = finite[np.isfinite(finite)]
        top = float(np.nanmax(finite)) if finite.size else 100.0
        vmax = max(100.0, float(np.ceil(top / 100.0) * 100.0))
    return plot_channel_array(positions, ch_names, kohm, ax=ax, title=title, vlim=(0.0, vmax),
                              label="impedance (kΩ)", diverging=False, cmap=cmap)

plot_window_sweep 

plot_window_sweep(windows, series, ylabel, chance=None, ax=None, ypct=False, title=None, ylim=None)

series: dict label -> (mean_array, sd_array) over windows.

ylim fixes the y-axis to (lo, hi) so plots are comparable across reports; None keeps matplotlib's autoscale. An auto-scaled axis makes every run's peak sit at the same height even when the runs differ by tens of bits/min, so the reader can't eyeball which run was better — the report layer passes a fixed range for exactly that reason.

Source code in src/ssvep/classification/viz.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
def plot_window_sweep(windows, series, ylabel, chance=None, ax=None, ypct=False, title=None,
                      ylim=None):
    """series: dict ``label -> (mean_array, sd_array)`` over ``windows``.

    ``ylim`` fixes the y-axis to ``(lo, hi)`` so plots are comparable across reports; ``None`` keeps
    matplotlib's autoscale. An auto-scaled axis makes every run's peak sit at the same height even
    when the runs differ by tens of bits/min, so the reader can't eyeball which run was better — the
    report layer passes a fixed range for exactly that reason.
    """
    if ax is None:
        _, ax = plt.subplots(figsize=(6.5, 4.3))
    for lab, (mean, sd) in series.items():
        mean, sd = np.asarray(mean, float), np.asarray(sd, float)
        if ypct:
            mean, sd = mean * 100, sd * 100
        line, = ax.plot(windows, mean, "-o", label=lab, zorder=3)
        ax.fill_between(windows, mean - sd, mean + sd, color=line.get_color(), alpha=0.2)
    if chance is not None:
        ax.axhline(chance * 100 if ypct else chance, color="grey", ls="--", lw=1,
                   label=f"chance ({chance*100:.1f}%)")
    ax.set_xlabel("Stimulation-window length (s)")
    ax.set_ylabel(ylabel + (" (%)" if ypct else ""))
    if ylim is not None:
        ax.set_ylim(*ylim)
    if title:
        ax.set_title(title, fontsize=10)
    ax.legend(fontsize=8); ax.grid(alpha=0.3)
    return ax.figure

ssvep.stim — stimulus design

Flicker math and design-time validation

ssvep.stim.flicker

Flicker math + design-time validation (pure; no pyglet, no GL).

Everything timing-critical about an SSVEP stimulus reduces to: what luminance should this target show on frame N of a display refreshing at R Hz? Keeping that here — as plain, deterministic functions of a frame counter — means the renderer is a thin shell and the important behaviour is unit-testable without a display (see tests: an FFT of frame_luminance_sequence must peak at the rendered frequency).

Conventions
  • Luminance is normalised to [0, 1] (0 = black, 1 = max). The renderer maps this to pixel values with :func:gamma_encode.
  • Phase is in cycles (0..1), matching the manifest. phase=0 starts a sine at its mean rising through zero; a trial resets the frame counter so phase is reproducible.
  • Time is derived from the frame counter, t = frame / refresh_hz — never wall-clock — so the waveform is deterministic and vsync-locked.

frames_per_cycle 

frames_per_cycle(freq_hz, refresh_hz)

Display frames per flicker cycle, refresh / freq (non-integer allowed for sine).

Source code in src/ssvep/stim/flicker.py 
41
42
43
44
45
def frames_per_cycle(freq_hz: float, refresh_hz: float) -> float:
    """Display frames per flicker cycle, ``refresh / freq`` (non-integer allowed for sine)."""
    if freq_hz <= 0:
        raise ValueError("freq_hz must be > 0")
    return refresh_hz / freq_hz

rendered_frequency 

rendered_frequency(freq_hz, refresh_hz, kind)

Actual displayed frequency given the render method.

  • sine — continuous phase → exact nominal frequency.
  • square — integer frames per (half-)cycle → quantises to refresh / round(refresh/f) and drifts from nominal at high freq / low frames-per-cycle.
Source code in src/ssvep/stim/flicker.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
def rendered_frequency(freq_hz: float, refresh_hz: float, kind: str) -> float:
    """Actual displayed frequency given the render method.

    * ``sine`` — continuous phase → exact nominal frequency.
    * ``square`` — integer frames per (half-)cycle → quantises to ``refresh / round(refresh/f)``
      and drifts from nominal at high freq / low frames-per-cycle.
    """
    if kind == "sine":
        return float(freq_hz)
    if kind == "square":
        fpc = max(round(refresh_hz / freq_hz), 2)
        return float(refresh_hz / fpc)
    raise ValueError(f"unknown flicker kind: {kind!r}")

is_renderable 

is_renderable(freq_hz, refresh_hz)

A frequency is renderable only below Nyquist of the display (freq < refresh/2).

Source code in src/ssvep/stim/flicker.py 
63
64
65
def is_renderable(freq_hz: float, refresh_hz: float) -> bool:
    """A frequency is renderable only below Nyquist of the display (``freq < refresh/2``)."""
    return 0.0 < freq_hz < refresh_hz / 2.0

photosensitivity_risk 

photosensitivity_risk(freq_hz, *, area_fraction=None, high_contrast=True)

Classify the seizure-provocation risk of a flicker frequency.

Frequency drives the classification; large area (>~25% of the field) and high contrast raise it. This is a design-time warning aid, not a medical guarantee.

Source code in src/ssvep/stim/flicker.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def photosensitivity_risk(freq_hz: float, *, area_fraction: float | None = None,
                          high_contrast: bool = True) -> PhotosensitivityRisk:
    """Classify the seizure-provocation risk of a flicker frequency.

    Frequency drives the classification; large area (>~25% of the field) and high contrast
    raise it. This is a design-time *warning* aid, not a medical guarantee.
    """
    big = area_fraction is not None and area_fraction >= 0.25
    if freq_hz < PROVOCATIVE_LO_HZ or freq_hz >= PROVOCATIVE_HI_HZ:
        # >= 30 Hz is the study's intended safe band; < 3 Hz is not a flicker-seizure risk.
        return PhotosensitivityRisk(
            "low", f"{freq_hz:g} Hz is outside the ~3–30 Hz provocative band (study-safe high band)."
        )
    # In the provocative band.
    peak = PEAK_LO_HZ <= freq_hz <= PEAK_HI_HZ
    if peak and (high_contrast or big):
        return PhotosensitivityRisk(
            "high",
            f"{freq_hz:g} Hz is in the peak photosensitive band ({PEAK_LO_HZ:g}–{PEAK_HI_HZ:g} Hz)"
            f"{' with large area' if big else ''}{' and high contrast' if high_contrast else ''}. "
            "Avoid for this REB protocol (use 30–50 Hz).",
        )
    return PhotosensitivityRisk(
        "elevated",
        f"{freq_hz:g} Hz is in the ~3–30 Hz provocative range (below the study's 30–50 Hz band). "
        "Prefer high-frequency stimuli; reduce contrast/size if it must be used.",
    )

sine_luminance 

sine_luminance(t, freq_hz, phase=0.0, depth=1.0, mean=0.5)

Sinusoidal luminance in [0,1]: mean + (depth/2)·sin(2π(f·t + phase)). Vectorised.

Source code in src/ssvep/stim/flicker.py 
106
107
108
109
def sine_luminance(t, freq_hz, phase=0.0, depth=1.0, mean=0.5):
    """Sinusoidal luminance in [0,1]: ``mean + (depth/2)·sin(2π(f·t + phase))``. Vectorised."""
    t = np.asarray(t, dtype=float)
    return mean + (depth / 2.0) * np.sin(2.0 * np.pi * (freq_hz * t + phase))

square_luminance 

square_luminance(t, freq_hz, phase=0.0, duty=0.5, low=0.0, high=1.0)

Two-level square luminance: high while within the duty fraction of the cycle.

Source code in src/ssvep/stim/flicker.py 
112
113
114
115
116
def square_luminance(t, freq_hz, phase=0.0, duty=0.5, low=0.0, high=1.0):
    """Two-level square luminance: ``high`` while within the ``duty`` fraction of the cycle."""
    t = np.asarray(t, dtype=float)
    cycle_pos = (freq_hz * t + phase) % 1.0
    return np.where(cycle_pos < duty, high, low).astype(float)

gamma_encode 

gamma_encode(luminance, gamma=2.2)

Map linear luminance [0,1] → display pixel value [0,1] using the inverse gamma.

Displays are ~gamma-2.2; to make perceived/linear luminance sinusoidal we raise to 1/gamma. Set gamma=1.0 if the monitor/LUT is already linearised.

Source code in src/ssvep/stim/flicker.py 
119
120
121
122
123
124
125
126
def gamma_encode(luminance, gamma=2.2):
    """Map linear luminance [0,1] → display pixel value [0,1] using the inverse gamma.

    Displays are ~gamma-2.2; to make *perceived/linear* luminance sinusoidal we raise to
    ``1/gamma``. Set ``gamma=1.0`` if the monitor/LUT is already linearised.
    """
    lum = np.clip(np.asarray(luminance, dtype=float), 0.0, 1.0)
    return lum ** (1.0 / gamma)

frame_luminance_sequence 

frame_luminance_sequence(freq_hz, refresh_hz, n_frames, *, kind='sine', phase=0.0, depth=1.0, mean=0.5, duty=0.5, gamma=1.0)

Per-frame luminance the renderer will show for n_frames (t = frame / refresh).

Returns a (n_frames,) array. This is exactly what the pyglet renderer samples, so a spectral test on it verifies the timing-critical output. gamma defaults to 1.0 (raw linear luminance) so tests see the pure waveform; the renderer passes the display gamma.

Source code in src/ssvep/stim/flicker.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def frame_luminance_sequence(freq_hz, refresh_hz, n_frames, *, kind="sine", phase=0.0,
                             depth=1.0, mean=0.5, duty=0.5, gamma=1.0):
    """Per-frame luminance the renderer will show for ``n_frames`` (t = frame / refresh).

    Returns a ``(n_frames,)`` array. This is exactly what the pyglet renderer samples, so a
    spectral test on it verifies the timing-critical output. ``gamma`` defaults to 1.0 (raw
    linear luminance) so tests see the pure waveform; the renderer passes the display gamma.
    """
    frames = np.arange(n_frames)
    t = frames / float(refresh_hz)
    if kind == "sine":
        lum = sine_luminance(t, freq_hz, phase, depth, mean)
    elif kind == "square":
        lum = square_luminance(t, freq_hz, phase, duty)
    else:
        raise ValueError(f"unknown flicker kind: {kind!r}")
    return gamma_encode(lum, gamma) if gamma != 1.0 else lum

luminance_at_frame 

luminance_at_frame(flicker_spec, frame, refresh_hz)

Scalar luminance in [0,1] for a manifest flicker block at a given frame.

Honours type (sine/square), freq_hz, phase, contrast (=modulation depth about mean 0.5) and duty (square only). This is the single per-frame luminance the renderer samples for a solid target — kept here so the contrast/duty maths is unit-tested without a display. t = frame / refresh (frame-counted, never wall-clock).

Source code in src/ssvep/stim/flicker.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def luminance_at_frame(flicker_spec: dict, frame: int, refresh_hz: float) -> float:
    """Scalar luminance in [0,1] for a manifest ``flicker`` block at a given frame.

    Honours ``type`` (sine/square), ``freq_hz``, ``phase``, ``contrast`` (=modulation depth
    about mean 0.5) and ``duty`` (square only). This is the single per-frame luminance the
    renderer samples for a solid target — kept here so the contrast/duty maths is unit-tested
    without a display. ``t = frame / refresh`` (frame-counted, never wall-clock).
    """
    t = frame / float(refresh_hz)
    kind = flicker_spec.get("type", "sine")
    phase = float(flicker_spec.get("phase", 0.0))
    depth = float(flicker_spec.get("contrast", 1.0))
    freq = float(flicker_spec["freq_hz"])
    if kind == "sine":
        return float(sine_luminance(t, freq, phase, depth=depth))
    if kind == "square":
        duty = float(flicker_spec.get("duty", 0.5))
        high, low = 0.5 + depth / 2.0, 0.5 - depth / 2.0
        return float(square_luminance(t, freq, phase, duty=duty, low=low, high=high))
    raise ValueError(f"unknown flicker kind: {kind!r}")

target_rgb 

target_rgb(luminance, color=(1.0, 1.0, 1.0), gamma=2.2, background_rgb=(0.0, 0.0, 0.0))

Combine a scalar luminance with a base colour → an 8-bit gamma-encoded (r,g,b).

The flicker modulates each channel linearly (in linear-light) between the background (luminance=0) and the target color (luminance=1), then gamma-encodes to pixels::

rgb_linear = background + luminance · (color − background)
pixel      = round(gamma_encode(rgb_linear) · 255)

So a white target on a black background reproduces the historic grayscale flicker, while a coloured target flickers between the background and that colour. color may be RGB or RGBA (alpha ignored). Pure + unit-tested; the renderer is a thin wrapper over this.

Source code in src/ssvep/stim/flicker.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def target_rgb(luminance: float, color=(1.0, 1.0, 1.0), gamma: float = 2.2,
               background_rgb=(0.0, 0.0, 0.0)) -> tuple[int, int, int]:
    """Combine a scalar luminance with a base colour → an 8-bit gamma-encoded ``(r,g,b)``.

    The flicker modulates each channel linearly (in linear-light) between the background
    (``luminance``=0) and the target ``color`` (``luminance``=1), then gamma-encodes to pixels::

        rgb_linear = background + luminance · (color − background)
        pixel      = round(gamma_encode(rgb_linear) · 255)

    So a white target on a black background reproduces the historic grayscale flicker, while a
    coloured target flickers between the background and that colour. ``color`` may be RGB or
    RGBA (alpha ignored). Pure + unit-tested; the renderer is a thin wrapper over this.
    """
    lum = float(np.clip(luminance, 0.0, 1.0))
    col = [float(c) for c in color[:3]]
    bg = [float(c) for c in background_rgb[:3]]
    out = []
    for c, b in zip(col, bg):
        lin = b + lum * (c - b)
        out.append(int(round(float(gamma_encode(lin, gamma)) * 255)))
    return out[0], out[1], out[2]

is_dropped_frame 

is_dropped_frame(dt_s, refresh_hz, tol=0.5)

True if an inter-frame interval overshot the expected period by more than tol.

dt beyond (1+tol)/refresh means at least one refresh was missed — the renderer logs these (an SSVEP killer if frequent).

Source code in src/ssvep/stim/flicker.py 
194
195
196
197
198
199
200
201
def is_dropped_frame(dt_s: float, refresh_hz: float, tol: float = 0.5) -> bool:
    """True if an inter-frame interval overshot the expected period by more than ``tol``.

    ``dt`` beyond ``(1+tol)/refresh`` means at least one refresh was missed — the renderer
    logs these (an SSVEP killer if frequent).
    """
    expected = 1.0 / refresh_hz
    return dt_s > expected * (1.0 + tol)

Run builder

ssvep.stim.builder

Run builder data-model: configure one SSVEP run and emit its manifest.

This is the UI-agnostic core of the stimulus builder (the PySide6 Design GUI will drive these objects). It models targets, timing, conditions, acquisition, and markers; validates the design at build time (renderability + photosensitivity — DESIGN_PRINCIPLES #4, COMPLIANCE R6); and serialises to a schema-valid run manifest — the single source of truth analysis reads. The central model is :class:RunSpec (one run). The session level — an ordered set of runs driven by a protocol — is :mod:ssvep.runtime.session (#43); a session-protocol references the run manifests a RunSpec emits.

Timing dataclass

Per-trial timing. stim_s is the usable (decodable) stimulation window.

onset_offset_s is a visual-latency lead-in that is added on top of stim_s: the target flickers for stim_s + onset_offset_s and analysis discards the lead-in, so the operator's "4 s stimulation" yields a full 4 s of usable steady-state data (flickering 4.14 s). See :data:ssvep.DEFAULT_ONSET_OFFSET_S for why.

Source code in src/ssvep/stim/builder.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
@dataclass
class Timing:
    """Per-trial timing. ``stim_s`` is the **usable** (decodable) stimulation window.

    ``onset_offset_s`` is a visual-latency lead-in that is **added on top** of ``stim_s``: the
    target flickers for ``stim_s + onset_offset_s`` and analysis discards the lead-in, so the
    operator's "4 s stimulation" yields a full 4 s of usable steady-state data (flickering 4.14 s).
    See :data:`ssvep.DEFAULT_ONSET_OFFSET_S` for why.
    """
    cue_s: float
    stim_s: float
    iti_s: float
    onset_offset_s: float = DEFAULT_ONSET_OFFSET_S

    @property
    def presented_stim_s(self) -> float:
        """What actually flickers per trial: usable window + the discarded visual-latency lead-in."""
        return float(self.stim_s) + float(self.onset_offset_s)

presented_stim_s property 

presented_stim_s

What actually flickers per trial: usable window + the discarded visual-latency lead-in.

MontageChannel dataclass

Source code in src/ssvep/stim/builder.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@dataclass
class MontageChannel:
    channel: str
    label: str
    position_2d: Optional[tuple[float, float]] = None
    impedance_kohm: Optional[float] = None
    position_pct: Optional[tuple[float, float]] = None
    """Head-relative 10-10 position ``(lateral_pct, height_pct)``, when the montage is *defined*
    that way rather than in centimetres (see :func:`occipital_ssvep_montage`).

    ``lateral_pct`` is the distance from the midline as a percentage of the **left→right
    preauricular arc**, signed left(-)/right(+); ``height_pct`` is the distance up from the
    **inion** as a percentage of the **nasion→inion arc**. Both are the 10-10 system's own units,
    so they are head-size independent — ``position_2d`` is the same point in cm on the
    :data:`REFERENCE_HEAD_CIRCUMFERENCE_CM` reference head, and is what the discrete channel-array
    plots use. Fixed-geometry carriers (the Unicorn, the actiCAPs) leave this ``None``: their
    electrodes are moulded or held in a cap, not measured onto the scalp."""

position_pct class-attribute instance-attribute 

position_pct = None

Head-relative 10-10 position (lateral_pct, height_pct), when the montage is defined that way rather than in centimetres (see :func:occipital_ssvep_montage).

lateral_pct is the distance from the midline as a percentage of the left→right preauricular arc, signed left(-)/right(+); height_pct is the distance up from the inion as a percentage of the nasion→inion arc. Both are the 10-10 system's own units, so they are head-size independent — position_2d is the same point in cm on the :data:REFERENCE_HEAD_CIRCUMFERENCE_CM reference head, and is what the discrete channel-array plots use. Fixed-geometry carriers (the Unicorn, the actiCAPs) leave this None: their electrodes are moulded or held in a cap, not measured onto the scalp.

Run dataclass

Run-identity block (manifest run, was experiment pre-1.3). Names one run + its BIDS task.

Source code in src/ssvep/stim/builder.py 
129
130
131
132
133
134
@dataclass
class Run:
    """Run-identity block (manifest ``run``, was ``experiment`` pre-1.3). Names one run + its BIDS task."""
    name: str
    task: str                       # BIDS-legal (alphanumeric)
    description: Optional[str] = None

RunSpec dataclass

One run's design — emits a run manifest via :meth:to_manifest (was Protocol pre-#43).

A run is one continuous period of data collection (CLAUDE.md §2.1). A session composes an ordered set of these by reference; see :mod:ssvep.runtime.session.

Source code in src/ssvep/stim/builder.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
@dataclass
class RunSpec:
    """One run's design — emits a run manifest via :meth:`to_manifest` (was ``Protocol`` pre-#43).

    A run is one continuous period of data collection (CLAUDE.md §2.1). A **session** composes an
    ordered set of these by reference; see :mod:`ssvep.runtime.session`.
    """
    run: Run
    display: Display
    stimuli: list[Target]
    conditions: list[Condition]
    timing: Timing
    # Optional since #147: hardware is no longer a *run design* fact. Build run emits a manifest with
    # no acquisition block at all; the protocol stamps one into every embedded run (Build Protocol →
    # "Set up acquisition hardware"), and the session may override it. A protocol whose runs still
    # have none cannot be locked — ssvep.ui.run.lock_blocker names it.
    acquisition: Optional[Acquisition] = None
    markers: Markers = field(default_factory=Markers)
    blocks_per_condition: Optional[int] = None
    trials_per_block: Optional[int] = None
    method: Optional[str] = None                   # static | factorial | staircase | bayesopt
    factors: Optional[dict] = None                 # sweep factors → level lists (design generation)
    paradigm: str = PARADIGM_SSVEP                 # ssvep (flickering targets) | resting (EO/EC)
    resting: Optional[RestingProtocol] = None      # block structure; required when paradigm=resting
    manifest_version: str = MANIFEST_VERSION

    # -- validation ---------------------------------------------------------
    @property
    def is_resting(self) -> bool:
        """True for the tone-cued eyes-open/eyes-closed resting paradigm (no flicker at all)."""
        return self.paradigm == PARADIGM_RESTING

    def effective_refresh_hz(self) -> float:
        """Refresh used for validation: the MEASURED rate if present, else the nominal one."""
        return self.display.measured_refresh_hz or self.display.refresh_hz

    def validate(self) -> ValidationReport:
        """Design-time checks: renderability (errors) + fidelity/photosensitivity (warnings).

        Judged against the *measured* refresh when available (see effective_refresh_hz). A resting
        protocol has no flicker, so the per-target renderability/photosensitivity checks don't apply
        and are replaced by block-design checks (see :meth:`_validate_resting`).
        """
        rep = ValidationReport()
        R = self.effective_refresh_hz()
        measured = self.display.measured_refresh_hz is not None
        if R <= 0:
            rep.errors.append("display.refresh_hz must be > 0")
            return rep
        # Resting renders a static cross at whatever rate the panel runs, so refresh gates neither
        # renderability nor fidelity — nagging to measure it would be pure noise.
        if not measured and not self.is_resting:
            rep.warnings.append(
                f"refresh {R:g} Hz is nominal (not measured) — measure the actual rate so "
                "renderability/fidelity feedback is trustworthy."
            )

        # Viewing geometry. Required to *render* deg; required to *report* visual angle in any unit.
        # Without it a recording cannot say how big its targets were — true of every run before
        # 2026-07-15 except one, which is why run-005's stimuli (2×2°) vs the rest (~6.8×6.4°) could
        # only be reconstructed by guessing the monitor.
        has_geom = bool(self.display.viewing_distance_cm and self.display.viewing_distance_cm > 0
                        and self.display.screen_diagonal_cm and self.display.screen_diagonal_cm > 0)
        if self.display.units == "deg":
            if not self.display.viewing_distance_cm or self.display.viewing_distance_cm <= 0:
                rep.errors.append("display.viewing_distance_cm must be > 0 when units=='deg'")
            if not self.display.screen_diagonal_cm or self.display.screen_diagonal_cm <= 0:
                rep.errors.append("display.screen_diagonal_cm must be > 0 when units=='deg'")
        elif not has_geom:
            rep.warnings.append(
                "display.viewing_distance_cm / screen_diagonal_cm are unset, so the stimulus size in "
                "degrees of visual angle cannot be recovered from this recording. Set them (or author "
                "in units='deg', the canonical unit) — target size drives SSVEP amplitude.")

        if self.is_resting:
            self._validate_resting(rep)
        for tgt in (self.stimuli if not self.is_resting else []):
            f = tgt.flicker.freq_hz
            if not flicker.is_renderable(f, R):
                rep.errors.append(
                    f"target {tgt.id}: {f:g} Hz is not renderable at {R:g} Hz "
                    f"(need 0 < f < Nyquist {R/2:g} Hz)"
                )
                continue
            fpc = flicker.frames_per_cycle(f, R)
            if tgt.flicker.kind == "sine" and fpc < flicker.MIN_FRAMES_PER_CYCLE_SINE:
                rep.warnings.append(
                    f"target {tgt.id}: sine at {f:g} Hz has only {fpc:.1f} frames/cycle "
                    f"(<{flicker.MIN_FRAMES_PER_CYCLE_SINE:g}); waveform will be coarsely sampled."
                )
            if tgt.flicker.kind == "square":
                rf = flicker.rendered_frequency(f, R, "square")
                if abs(rf - f) > 1e-6:
                    rep.warnings.append(
                        f"target {tgt.id}: square {f:g} Hz quantises to {rf:.3f} Hz at {R:g} Hz "
                        f"(square frequencies are refresh/(2k))."
                    )
            risk = flicker.photosensitivity_risk(f, area_fraction=_area_fraction(tgt, self.display))
            if risk.level in ("elevated", "high"):
                rep.warnings.append(f"target {tgt.id}: photosensitivity {risk.level} — {risk.message}")

        self._warn_offscreen(rep)
        self._validate_appearance(rep)

        # Schema conformance of the emitted manifest.
        try:
            manifest_io.validate_manifest(self.to_manifest())
        except Exception as exc:  # jsonschema.ValidationError or build error
            rep.errors.append(f"manifest schema: {getattr(exc, 'message', str(exc))}")
        return rep

    def _validate_resting(self, rep: ValidationReport) -> None:
        """Block-design checks for the resting paradigm (it has no flicker to check).

        The failure modes here are different in kind from SSVEP's: a resting run can't be
        *unrenderable*, but it can be **unanalysable** (nothing left after the settle window, no
        eyes-closed data to estimate PAF from) or **unrunnable** (cue tones the participant can't
        tell apart). Those are what's checked.
        """
        r = self.resting
        if r is None:
            rep.errors.append("paradigm 'resting' requires resting block parameters "
                              "(RunSpec.resting / manifest design.resting)")
            return
        if self.stimuli:
            rep.errors.append(
                f"paradigm 'resting' must define no stimuli (found {len(self.stimuli)}) — a resting "
                f"run has no flicker; that absence is exactly what makes it the SSVEP negative "
                f"control.")
        lo, hi = (float(r.band_hz[0]), float(r.band_hz[1]))
        if not (0 < lo < hi):
            rep.errors.append(f"resting.band_hz must satisfy 0 < lo < hi, got ({lo:g}, {hi:g})")
        if r.settle_s >= r.block_s:
            rep.errors.append(
                f"resting.settle_s ({r.settle_s:g} s) must be < block_s ({r.block_s:g} s) — analysis "
                f"discards settle_s from the start of every block, so no data would survive.")
        if r.n_blocks < 2:
            rep.warnings.append(
                "resting: fewer than 2 blocks — with only one eye-state there is no eyes-closed/"
                "eyes-open contrast, so alpha reactivity (the posterior-montage QC) can't be computed.")
        ec_s = sum(r.block_s for k in r.block_kinds() if k == EYES_CLOSED)
        usable_ec = max(0.0, ec_s - r.settle_s * max(1, sum(1 for k in r.block_kinds()
                                                            if k == EYES_CLOSED)))
        if usable_ec < 60.0:
            rep.warnings.append(
                f"resting: only {usable_ec:g} s of usable eyes-closed data (after discarding "
                f"settle_s per block) — a stable peak-alpha estimate wants >= 60 s (Klimesch 1999).")
        if abs(r.high_tone_hz - r.low_tone_hz) < 100.0:
            rep.warnings.append(
                f"resting: the HIGH ({r.high_tone_hz:g} Hz) and LOW ({r.low_tone_hz:g} Hz) cue tones "
                f"are < 100 Hz apart — participants confuse them, and the tone is the ONLY eye-state "
                f"cue (their eyes are shut half the run). Keep them clearly distinct.")

    def _validate_appearance(self, rep: ValidationReport) -> None:
        """Range-check the v1.1 appearance fields (contrast/duty/colour/pattern/gamma/background).

        Errors for out-of-range values that would render wrongly or fail the schema; warnings for
        combinations that are legal but likely a mistake (e.g. a pattern with no spatial frequency).
        """
        d = self.display
        if d.gamma is None or d.gamma <= 0:
            rep.errors.append("display.gamma must be > 0")
        bg = d.background_rgb
        if bg is not None:
            if len(bg) != 3 or any(not (0.0 <= float(c) <= 1.0) for c in bg):
                rep.errors.append("display.background_rgb must be 3 values in [0,1]")

        for t in self.stimuli:
            fl = t.flicker
            if not (0.0 <= fl.contrast <= 1.0):
                rep.errors.append(f"target {t.id}: flicker.contrast {fl.contrast:g} out of [0,1]")
            if not (0.0 < fl.duty < 1.0):
                rep.errors.append(f"target {t.id}: flicker.duty {fl.duty:g} out of (0,1)")
            if fl.contrast == 0.0:
                rep.warnings.append(f"target {t.id}: contrast 0 — target will not flicker (steady).")
            if t.color is not None:
                if len(t.color) not in (3, 4) or any(not (0.0 <= float(c) <= 1.0) for c in t.color):
                    rep.errors.append(f"target {t.id}: color must be 3–4 RGB(A) values in [0,1]")
            if t.cue_color is not None:
                if len(t.cue_color) != 3 or any(not (0.0 <= float(c) <= 1.0) for c in t.cue_color):
                    rep.errors.append(f"target {t.id}: cue_color must be 3 RGB values in [0,1]")
            if t.pattern not in PATTERNS:
                rep.errors.append(f"target {t.id}: pattern {t.pattern!r} not in {PATTERNS}")
            if t.spatial_freq_cpd is not None and t.spatial_freq_cpd <= 0:
                rep.errors.append(f"target {t.id}: spatial_freq_cpd must be > 0")
            if t.pattern in ("checkerboard", "grating") and t.spatial_freq_cpd is None:
                rep.warnings.append(
                    f"target {t.id}: pattern {t.pattern!r} without spatial_freq_cpd — "
                    "defaults to 1 cycle across the target.")
            if t.pattern == "solid" and t.pattern_reversal:
                rep.warnings.append(
                    f"target {t.id}: pattern_reversal has no effect on a solid target.")

    def _warn_offscreen(self, rep: ValidationReport) -> None:
        """Warn if any target would extend past the screen edges (so it renders black/clipped).

        - ``norm`` units are resolution-independent: the screen spans [-1, 1], so a target is
          off-screen when its centre ± half-size exceeds ±1 (checkable with no monitor info).
        - ``deg`` units need the render resolution + viewing geometry to map to pixels; skipped if
          those aren't known (the render uses the live window resolution, which may differ)."""
        d = self.display
        if d.units != "deg":
            for tgt in self.stimuli:
                w, h = tgt.size if tgt.size else (0.15, 0.15)
                x, y = tgt.position_xy
                if abs(x) + w / 2.0 > 1.0 or abs(y) + h / 2.0 > 1.0:
                    rep.warnings.append(
                        f"target {tgt.id}: extends off-screen (position/size exceed the ±1 screen "
                        f"edges) — it will be clipped or invisible.")
            return
        if not d.resolution_px:
            return
        if not (d.viewing_distance_cm and d.screen_diagonal_cm
                and d.viewing_distance_cm > 0 and d.screen_diagonal_cm > 0):
            return
        res_w, res_h = d.resolution_px
        for tgt in self.stimuli:
            cx, cy, w, h = geometry.target_center_size_px(
                tgt.position_xy, tgt.size, units="deg", resolution_px=d.resolution_px,
                viewing_distance_cm=d.viewing_distance_cm, screen_diagonal_cm=d.screen_diagonal_cm)
            if cx - w / 2 < 0 or cx + w / 2 > res_w or cy - h / 2 < 0 or cy + h / 2 > res_h:
                rep.warnings.append(
                    f"target {tgt.id}: extends off-screen at {d.viewing_distance_cm:g} cm viewing / "
                    f"{d.screen_diagonal_cm:g} cm diagonal — it will be clipped or invisible.")

    # -- serialisation ------------------------------------------------------
    def to_manifest(self) -> dict:
        """Serialise to a manifest dict conforming to run_manifest.schema.json."""
        m: dict = {
            "manifest_version": self.manifest_version,
            "paradigm": self.paradigm,
            "run": _drop_none({
                "name": self.run.name,
                "task": self.run.task,
                "description": self.run.description,
                "builder_version": BUILDER_VERSION,
            }),
            "display": _drop_none({
                "refresh_hz": self.display.refresh_hz,
                "resolution_px": list(self.display.resolution_px) if self.display.resolution_px else None,
                "measured_refresh_hz": self.display.measured_refresh_hz,
                "units": self.display.units,
                "viewing_distance_cm": self.display.viewing_distance_cm,
                "screen_diagonal_cm": self.display.screen_diagonal_cm,
                "gamma": self.display.gamma,
                "background_rgb": list(self.display.background_rgb) if self.display.background_rgb is not None else None,
                "luminance_cd_m2": self.display.luminance_cd_m2,
            }),
            "stimuli": [self._target_dict(t) for t in self.stimuli],
            "design": _drop_none({
                "conditions": [_drop_none({
                    "id": c.id, "label": c.label,
                    "stimulus_freqs_hz": list(c.stimulus_freqs_hz), "params": c.params,
                }) for c in self.conditions],
                "blocks_per_condition": self.blocks_per_condition,
                "trials_per_block": self.trials_per_block,
                "method": self.method,
                "factors": dict(self.factors) if self.factors else None,
                "timing": {"cue_s": self.timing.cue_s, "stim_s": self.timing.stim_s,
                           "iti_s": self.timing.iti_s,
                           "onset_offset_s": self.timing.onset_offset_s},
                "resting": _resting_dict(self.resting) if self.is_resting else None,
            }),
            "markers": _drop_none({
                "stream_name": self.markers.stream_name,
                "schema": self.markers.schema,
                "segments": list(self.markers.segments) if self.markers.segments else None,
            }),
        }
        # Omitted entirely when unset — a run with no acquisition block is a design that has not
        # chosen hardware, which is different from one that chose nothing (#147).
        if self.acquisition is not None:
            m["acquisition"] = acquisition_to_dict(self.acquisition)
        return m

    def _target_dict(self, t: Target) -> dict:
        fl = {"type": t.flicker.kind, "freq_hz": t.flicker.freq_hz, "phase": t.flicker.phase}
        if t.flicker.kind == "square":
            rf = flicker.rendered_frequency(t.flicker.freq_hz, self.display.refresh_hz, "square")
            if abs(rf - t.flicker.freq_hz) > 1e-6:
                fl["rendered_freq_hz"] = rf
            if abs(t.flicker.duty - 0.5) > 1e-9:
                fl["duty"] = t.flicker.duty
        if abs(t.flicker.contrast - 1.0) > 1e-9:
            fl["contrast"] = t.flicker.contrast
        return _drop_none({
            "id": t.id,
            "position_xy": list(t.position_xy),
            "size": list(t.size) if t.size else None,
            "shape": t.shape,
            "color": list(t.color) if t.color else None,
            "cue_color": list(t.cue_color) if t.cue_color else None,
            "pattern": t.pattern if t.pattern and t.pattern != "solid" else None,
            "pattern_reversal": t.pattern_reversal if t.pattern_reversal else None,
            "spatial_freq_cpd": t.spatial_freq_cpd,
            "flicker": fl,
        })

    def save(self, path) -> "str":
        """Validate (raise on any error) and write the manifest to ``path``."""
        rep = self.validate()
        if not rep.ok:
            raise ValueError(f"cannot save invalid protocol:\n{rep}")
        return str(manifest_io.save_manifest(self.to_manifest(), path))

is_resting property 

is_resting

True for the tone-cued eyes-open/eyes-closed resting paradigm (no flicker at all).

effective_refresh_hz 

effective_refresh_hz()

Refresh used for validation: the MEASURED rate if present, else the nominal one.

Source code in src/ssvep/stim/builder.py 
229
230
231
def effective_refresh_hz(self) -> float:
    """Refresh used for validation: the MEASURED rate if present, else the nominal one."""
    return self.display.measured_refresh_hz or self.display.refresh_hz

validate 

validate()

Design-time checks: renderability (errors) + fidelity/photosensitivity (warnings).

Judged against the measured refresh when available (see effective_refresh_hz). A resting protocol has no flicker, so the per-target renderability/photosensitivity checks don't apply and are replaced by block-design checks (see :meth:_validate_resting).

Source code in src/ssvep/stim/builder.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
def validate(self) -> ValidationReport:
    """Design-time checks: renderability (errors) + fidelity/photosensitivity (warnings).

    Judged against the *measured* refresh when available (see effective_refresh_hz). A resting
    protocol has no flicker, so the per-target renderability/photosensitivity checks don't apply
    and are replaced by block-design checks (see :meth:`_validate_resting`).
    """
    rep = ValidationReport()
    R = self.effective_refresh_hz()
    measured = self.display.measured_refresh_hz is not None
    if R <= 0:
        rep.errors.append("display.refresh_hz must be > 0")
        return rep
    # Resting renders a static cross at whatever rate the panel runs, so refresh gates neither
    # renderability nor fidelity — nagging to measure it would be pure noise.
    if not measured and not self.is_resting:
        rep.warnings.append(
            f"refresh {R:g} Hz is nominal (not measured) — measure the actual rate so "
            "renderability/fidelity feedback is trustworthy."
        )

    # Viewing geometry. Required to *render* deg; required to *report* visual angle in any unit.
    # Without it a recording cannot say how big its targets were — true of every run before
    # 2026-07-15 except one, which is why run-005's stimuli (2×2°) vs the rest (~6.8×6.4°) could
    # only be reconstructed by guessing the monitor.
    has_geom = bool(self.display.viewing_distance_cm and self.display.viewing_distance_cm > 0
                    and self.display.screen_diagonal_cm and self.display.screen_diagonal_cm > 0)
    if self.display.units == "deg":
        if not self.display.viewing_distance_cm or self.display.viewing_distance_cm <= 0:
            rep.errors.append("display.viewing_distance_cm must be > 0 when units=='deg'")
        if not self.display.screen_diagonal_cm or self.display.screen_diagonal_cm <= 0:
            rep.errors.append("display.screen_diagonal_cm must be > 0 when units=='deg'")
    elif not has_geom:
        rep.warnings.append(
            "display.viewing_distance_cm / screen_diagonal_cm are unset, so the stimulus size in "
            "degrees of visual angle cannot be recovered from this recording. Set them (or author "
            "in units='deg', the canonical unit) — target size drives SSVEP amplitude.")

    if self.is_resting:
        self._validate_resting(rep)
    for tgt in (self.stimuli if not self.is_resting else []):
        f = tgt.flicker.freq_hz
        if not flicker.is_renderable(f, R):
            rep.errors.append(
                f"target {tgt.id}: {f:g} Hz is not renderable at {R:g} Hz "
                f"(need 0 < f < Nyquist {R/2:g} Hz)"
            )
            continue
        fpc = flicker.frames_per_cycle(f, R)
        if tgt.flicker.kind == "sine" and fpc < flicker.MIN_FRAMES_PER_CYCLE_SINE:
            rep.warnings.append(
                f"target {tgt.id}: sine at {f:g} Hz has only {fpc:.1f} frames/cycle "
                f"(<{flicker.MIN_FRAMES_PER_CYCLE_SINE:g}); waveform will be coarsely sampled."
            )
        if tgt.flicker.kind == "square":
            rf = flicker.rendered_frequency(f, R, "square")
            if abs(rf - f) > 1e-6:
                rep.warnings.append(
                    f"target {tgt.id}: square {f:g} Hz quantises to {rf:.3f} Hz at {R:g} Hz "
                    f"(square frequencies are refresh/(2k))."
                )
        risk = flicker.photosensitivity_risk(f, area_fraction=_area_fraction(tgt, self.display))
        if risk.level in ("elevated", "high"):
            rep.warnings.append(f"target {tgt.id}: photosensitivity {risk.level} — {risk.message}")

    self._warn_offscreen(rep)
    self._validate_appearance(rep)

    # Schema conformance of the emitted manifest.
    try:
        manifest_io.validate_manifest(self.to_manifest())
    except Exception as exc:  # jsonschema.ValidationError or build error
        rep.errors.append(f"manifest schema: {getattr(exc, 'message', str(exc))}")
    return rep

to_manifest 

to_manifest()

Serialise to a manifest dict conforming to run_manifest.schema.json.

Source code in src/ssvep/stim/builder.py 
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
def to_manifest(self) -> dict:
    """Serialise to a manifest dict conforming to run_manifest.schema.json."""
    m: dict = {
        "manifest_version": self.manifest_version,
        "paradigm": self.paradigm,
        "run": _drop_none({
            "name": self.run.name,
            "task": self.run.task,
            "description": self.run.description,
            "builder_version": BUILDER_VERSION,
        }),
        "display": _drop_none({
            "refresh_hz": self.display.refresh_hz,
            "resolution_px": list(self.display.resolution_px) if self.display.resolution_px else None,
            "measured_refresh_hz": self.display.measured_refresh_hz,
            "units": self.display.units,
            "viewing_distance_cm": self.display.viewing_distance_cm,
            "screen_diagonal_cm": self.display.screen_diagonal_cm,
            "gamma": self.display.gamma,
            "background_rgb": list(self.display.background_rgb) if self.display.background_rgb is not None else None,
            "luminance_cd_m2": self.display.luminance_cd_m2,
        }),
        "stimuli": [self._target_dict(t) for t in self.stimuli],
        "design": _drop_none({
            "conditions": [_drop_none({
                "id": c.id, "label": c.label,
                "stimulus_freqs_hz": list(c.stimulus_freqs_hz), "params": c.params,
            }) for c in self.conditions],
            "blocks_per_condition": self.blocks_per_condition,
            "trials_per_block": self.trials_per_block,
            "method": self.method,
            "factors": dict(self.factors) if self.factors else None,
            "timing": {"cue_s": self.timing.cue_s, "stim_s": self.timing.stim_s,
                       "iti_s": self.timing.iti_s,
                       "onset_offset_s": self.timing.onset_offset_s},
            "resting": _resting_dict(self.resting) if self.is_resting else None,
        }),
        "markers": _drop_none({
            "stream_name": self.markers.stream_name,
            "schema": self.markers.schema,
            "segments": list(self.markers.segments) if self.markers.segments else None,
        }),
    }
    # Omitted entirely when unset — a run with no acquisition block is a design that has not
    # chosen hardware, which is different from one that chose nothing (#147).
    if self.acquisition is not None:
        m["acquisition"] = acquisition_to_dict(self.acquisition)
    return m

save 

save(path)

Validate (raise on any error) and write the manifest to path.

Source code in src/ssvep/stim/builder.py 
495
496
497
498
499
500
def save(self, path) -> "str":
    """Validate (raise on any error) and write the manifest to ``path``."""
    rep = self.validate()
    if not rep.ok:
        raise ValueError(f"cannot save invalid protocol:\n{rep}")
    return str(manifest_io.save_manifest(self.to_manifest(), path))

acquisition_to_dict 

acquisition_to_dict(acq)

Serialise an :class:Acquisition to the manifest acquisition block (schema-conformant).

Pulled out of :meth:RunSpec.to_manifest so it has exactly one implementation: a run built on Build run goes through it, and so does the Build Protocol "Set acquisition for all runs…" action (#121), which writes this same shape straight into every embedded run's manifest.

Source code in src/ssvep/stim/builder.py 
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
def acquisition_to_dict(acq: Acquisition) -> dict:
    """Serialise an :class:`Acquisition` to the manifest ``acquisition`` block (schema-conformant).

    Pulled out of :meth:`RunSpec.to_manifest` so it has exactly one implementation: a run built on
    Build run goes through it, and so does the Build Protocol "Set acquisition for all runs…" action
    (#121), which writes this same shape straight into every embedded run's manifest.
    """
    return _drop_none({
        "device": acq.device,
        "amplifier": acq.amplifier,
        "board": acq.board,
        "sfreq_hz": acq.sfreq_hz,
        "reference": acq.reference,
        "ground": acq.ground,
        "headset": acq.headset,
        "electrode_type": acq.electrode_type,
        "prep": acq.prep,
        "montage": [_drop_none({
            "channel": ch.channel, "label": ch.label,
            "position_2d": list(ch.position_2d) if ch.position_2d else None,
            "position_pct": list(ch.position_pct) if ch.position_pct else None,
            "impedance_kohm": ch.impedance_kohm,
        }) for ch in acq.montage],
    })

grid_positions 

grid_positions(n_cols, n_rows, spread=0.5)

Row-major normalised (x, y) centres for an n_cols × n_rows grid in [-spread, spread].

Source code in src/ssvep/stim/builder.py 
568
569
570
571
572
def grid_positions(n_cols: int, n_rows: int, spread: float = 0.5) -> list[tuple[float, float]]:
    """Row-major normalised (x, y) centres for an ``n_cols × n_rows`` grid in [-spread, spread]."""
    xs = [0.0] if n_cols == 1 else [(-spread + 2 * spread * i / (n_cols - 1)) for i in range(n_cols)]
    ys = [0.0] if n_rows == 1 else [(spread - 2 * spread * j / (n_rows - 1)) for j in range(n_rows)]
    return [(x, y) for y in ys for x in xs]

montage_pct_to_cm 

montage_pct_to_cm(lateral_pct, height_pct, nasion_inion_cm=REFERENCE_NASION_INION_ARC_CM, preauricular_cm=REFERENCE_PREAURICULAR_ARC_CM)

A 10-10 percentage position as tape-measure centimetres on a head of the given arc lengths.

Returns (x_cm, y_cm) — x left(-)/right(+) of the midline, y up from the inion — which is exactly what the operator measures: up the midline first, then out from that point. Defaults are the :data:REFERENCE_HEAD_CIRCUMFERENCE_CM head the montage was originally drawn on.

Source code in src/ssvep/stim/builder.py 
627
628
629
630
631
632
633
634
635
636
637
def montage_pct_to_cm(lateral_pct: float, height_pct: float,
                      nasion_inion_cm: float = REFERENCE_NASION_INION_ARC_CM,
                      preauricular_cm: float = REFERENCE_PREAURICULAR_ARC_CM) -> tuple[float, float]:
    """A 10-10 percentage position as tape-measure centimetres on a head of the given arc lengths.

    Returns ``(x_cm, y_cm)`` — ``x`` left(-)/right(+) of the midline, ``y`` up from the inion — which
    is exactly what the operator measures: up the midline first, then out from that point. Defaults
    are the :data:`REFERENCE_HEAD_CIRCUMFERENCE_CM` head the montage was originally drawn on.
    """
    return (round(lateral_pct / 100.0 * preauricular_cm, 2),
            round(height_pct / 100.0 * nasion_inion_cm, 2))

occipital_ssvep_montage 

occipital_ssvep_montage()

Free electrode occipital SSVEP montage — the 8-channel occipital array for the OpenBCI Cyton with free (unmounted) electrodes. The lab's dry headband carries the same layout.

The montage is defined in 10-10 units (:data:OCCIPITAL_SSVEP_MONTAGE_PCT) so it transfers across head sizes; position_pct carries that definition and position_2d is the same point in cm on the reference head — x left(-)/right(+) of the midline, y up from the inion, so the impedance map renders nose-up with the participant's left on the left.

E1..E8 are the Cyton's 8 EEG channels, labelled by their wire colour (:data:CYTON_CHANNEL_WIRE_COLOURS) because on a free-electrode array the colour is the only thing telling one electrode from another. Reference and ground are not EEG channels and so do not appear here: white/SRB goes to the left earlobe and black/BIAS to the right (:data:OCCIPITAL_SSVEP_REFERENCE, :data:OCCIPITAL_SSVEP_GROUND).

Source code in src/ssvep/stim/builder.py 
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
def occipital_ssvep_montage() -> list[MontageChannel]:
    """**Free electrode occipital SSVEP montage** — the 8-channel occipital array for the OpenBCI
    Cyton with free (unmounted) electrodes. The lab's dry headband carries the same layout.

    The montage is *defined* in 10-10 units (:data:`OCCIPITAL_SSVEP_MONTAGE_PCT`) so it transfers
    across head sizes; ``position_pct`` carries that definition and ``position_2d`` is the same point
    in **cm** on the reference head — ``x`` left(-)/right(+) of the midline, ``y`` up from the inion,
    so the impedance map renders nose-up with the participant's left on the left.

    ``E1..E8`` are the Cyton's 8 EEG channels, labelled by their **wire colour**
    (:data:`CYTON_CHANNEL_WIRE_COLOURS`) because on a free-electrode array the colour is the only
    thing telling one electrode from another. Reference and ground are not EEG channels and so do not
    appear here: white/SRB goes to the **left** earlobe and black/BIAS to the **right**
    (:data:`OCCIPITAL_SSVEP_REFERENCE`, :data:`OCCIPITAL_SSVEP_GROUND`).
    """
    return [MontageChannel(channel=f"E{i+1}", label=colour,
                           position_pct=pct, position_2d=montage_pct_to_cm(*pct))
            for i, (colour, pct) in enumerate(zip(CYTON_CHANNEL_WIRE_COLOURS,
                                                  OCCIPITAL_SSVEP_MONTAGE_PCT))]

occipital_1010_montage 

occipital_1010_montage()

Occipital 10-10 montage — P1 PO3 O1 POz Oz P2 PO4 O2 on an OpenBCI Cyton, in board order 1..8, with the same ear reference and ground as :func:occipital_ssvep_montage.

Unlike that montage, this one is defined by its site names: "PO3" already says exactly where the electrode goes, in a system every EEG lab and every cap already implements — so there is no position_pct to record, and nothing here needs a reference head. The gain is comparability (results line up with the SSVEP literature, and the channels mean something to a reader who has never seen this toolbox) and the ability to place it from a cap; the cost is less density right over the occipital pole, because the array has to land on sites the 10-10 grid happens to provide.

label is the 10-10 site — the electrode's identity here — but the wire colours are the same Cyton ribbon order (:data:CYTON_CHANNEL_WIRE_COLOURS), so channel 3 is still the blue lead. position_2d reuses :func:acticap_64ch_montage's coordinates for every site, so this montage and the 64-channel cap plot identically rather than drifting apart in two hand-typed tables.

Source code in src/ssvep/stim/builder.py 
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
def occipital_1010_montage() -> list[MontageChannel]:
    """**Occipital 10-10 montage** — ``P1 PO3 O1 POz Oz P2 PO4 O2`` on an OpenBCI Cyton, in board
    order 1..8, with the same ear reference and ground as :func:`occipital_ssvep_montage`.

    Unlike that montage, this one is *defined by its site names*: "PO3" already says exactly where the
    electrode goes, in a system every EEG lab and every cap already implements — so there is no
    ``position_pct`` to record, and nothing here needs a reference head. The gain is comparability
    (results line up with the SSVEP literature, and the channels mean something to a reader who has
    never seen this toolbox) and the ability to place it from a cap; the cost is less density right
    over the occipital pole, because the array has to land on sites the 10-10 grid happens to provide.

    ``label`` is the 10-10 site — the electrode's identity here — but the **wire colours are the same
    Cyton ribbon order** (:data:`CYTON_CHANNEL_WIRE_COLOURS`), so channel 3 is still the blue lead.
    ``position_2d`` reuses :func:`acticap_64ch_montage`'s coordinates for every site, so this montage
    and the 64-channel cap plot identically rather than drifting apart in two hand-typed tables.
    """
    pos = {ch.label: ch.position_2d for ch in acticap_64ch_montage()}
    return [MontageChannel(channel=f"E{i+1}", label=site, position_2d=pos[site])
            for i, site in enumerate(OCCIPITAL_1010_SITES)]

unicorn_hybrid_black_montage 

unicorn_hybrid_black_montage()

The g.tec Unicorn Hybrid Black's fixed 8-channel montage — Fz C3 Cz C4 Pz PO7 Oz PO8 (standard 10-20 positions) in the device's channel order 1..8 (see docs/UNICORN_INTEGRATION.md, validated on hardware 2026-07-07).

Unlike the OpenBCI carriers, the Unicorn's electrodes are moulded into the headset, so this montage is fixed (not editable). Positions are an approximate 2D scalp projection with the same convention as :func:occipital_ssvep_montage — x = left(-)/right(+) of the midline, y = anterior(+)/posterior(-), Cz at the origin — so the impedance map renders nose-up. label is the electrode's 10-20 name (the Unicorn labels channels by site, not by wire colour).

Source code in src/ssvep/stim/builder.py 
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
def unicorn_hybrid_black_montage() -> list[MontageChannel]:
    """The g.tec Unicorn Hybrid Black's fixed 8-channel montage — ``Fz C3 Cz C4 Pz PO7 Oz PO8``
    (standard 10-20 positions) in the device's channel order 1..8 (see ``docs/UNICORN_INTEGRATION.md``,
    validated on hardware 2026-07-07).

    Unlike the OpenBCI carriers, the Unicorn's electrodes are moulded into the headset, so this montage
    is fixed (not editable). Positions are an approximate 2D scalp projection with the same convention
    as :func:`occipital_ssvep_montage` — ``x`` = left(-)/right(+) of the midline, ``y`` =
    anterior(+)/posterior(-), Cz at the origin — so the impedance map renders nose-up. ``label`` is the
    electrode's 10-20 name (the Unicorn labels channels by site, not by wire colour).
    """
    # (10-20 site, (x_cm, y_cm)) in board order 1..8.
    sites = [("Fz",  ( 0.0,  4.6)),   # ch1
             ("C3",  (-4.6,  0.0)),   # ch2
             ("Cz",  ( 0.0,  0.0)),   # ch3
             ("C4",  ( 4.6,  0.0)),   # ch4
             ("Pz",  ( 0.0, -4.6)),   # ch5
             ("PO7", (-5.0, -7.1)),   # ch6
             ("Oz",  ( 0.0, -8.5)),   # ch7
             ("PO8", ( 5.0, -7.1))]   # ch8
    return [MontageChannel(channel=f"E{i+1}", label=lbl, position_2d=xy)
            for i, (lbl, xy) in enumerate(sites)]

acticap_64ch_montage 

acticap_64ch_montage()

Brain Products actiCAP 64Ch Standard-2 layout, in the actiCHamp's channel order 1..64.

Positions are an approximate 2D scalp projection in the same convention as the other montages — x = left(-)/right(+) of the midline, y = anterior(+)/posterior(-), Cz at the origin, in cm on a ~9.5 cm scalp disc — so the impedance map renders nose-up with the correct hemispheres. Labels are the 10-10 site names; channel is the amplifier channel index. Schematic (recognisable), not digitised positions — a proper digitiser layout can replace this later. (Ground/reference are the actiCHamp's dedicated GND/REF pins, not EEG channels, so they're not listed.)

Source code in src/ssvep/stim/builder.py 
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
def acticap_64ch_montage() -> list[MontageChannel]:
    """Brain Products **actiCAP 64Ch Standard-2** layout, in the actiCHamp's channel order 1..64.

    Positions are an approximate 2D scalp projection in the same convention as the other montages —
    ``x`` = left(-)/right(+) of the midline, ``y`` = anterior(+)/posterior(-), Cz at the origin, in cm
    on a ~9.5 cm scalp disc — so the impedance map renders nose-up with the correct hemispheres. Labels
    are the 10-10 site names; ``channel`` is the amplifier channel index. Schematic (recognisable), not
    digitised positions — a proper digitiser layout can replace this later. (Ground/reference are the
    actiCHamp's dedicated GND/REF pins, not EEG channels, so they're not listed.)
    """
    # (10-10 site, (x_cm, y_cm)) in actiCAP 64Ch Standard-2 channel order 1..64.
    sites = [
        ("Fp1", (-2.1,  9.0)), ("Fp2", ( 2.1,  9.0)), ("F7",  (-8.1,  5.2)), ("F3",  (-4.2,  5.2)),
        ("Fz",  ( 0.0,  5.2)), ("F4",  ( 4.2,  5.2)), ("F8",  ( 8.1,  5.2)), ("FC5", (-6.3,  2.9)),
        ("FC1", (-2.1,  2.9)), ("FC2", ( 2.1,  2.9)), ("FC6", ( 6.3,  2.9)), ("T7",  (-9.0,  0.0)),
        ("C3",  (-4.2,  0.0)), ("Cz",  ( 0.0,  0.0)), ("C4",  ( 4.2,  0.0)), ("T8",  ( 9.0,  0.0)),
        ("TP9", (-8.8, -3.5)), ("CP5", (-6.3, -2.9)), ("CP1", (-2.1, -2.9)), ("CP2", ( 2.1, -2.9)),
        ("CP6", ( 6.3, -2.9)), ("TP10",( 8.8, -3.5)), ("P7",  (-8.1, -5.2)), ("P3",  (-4.2, -5.2)),
        ("Pz",  ( 0.0, -5.2)), ("P4",  ( 4.2, -5.2)), ("P8",  ( 8.1, -5.2)), ("PO9", (-5.5, -8.0)),
        ("O1",  (-2.9, -8.8)), ("Oz",  ( 0.0, -9.3)), ("O2",  ( 2.9, -8.8)), ("PO10",( 5.5, -8.0)),
        ("AF7", (-7.1,  7.6)), ("AF3", (-3.3,  7.6)), ("AF4", ( 3.3,  7.6)), ("AF8", ( 7.1,  7.6)),
        ("F5",  (-6.2,  5.2)), ("F1",  (-2.1,  5.2)), ("F2",  ( 2.1,  5.2)), ("F6",  ( 6.2,  5.2)),
        ("FT9", (-9.2,  1.5)), ("FT7", (-8.6,  2.9)), ("FC3", (-4.2,  2.9)), ("FC4", ( 4.2,  2.9)),
        ("FT8", ( 8.6,  2.9)), ("FT10",( 9.2,  1.5)), ("C5",  (-6.3,  0.0)), ("C1",  (-2.1,  0.0)),
        ("C2",  ( 2.1,  0.0)), ("C6",  ( 6.3,  0.0)), ("TP7", (-8.6, -2.9)), ("CP3", (-4.2, -2.9)),
        ("CPz", ( 0.0, -2.9)), ("CP4", ( 4.2, -2.9)), ("TP8", ( 8.6, -2.9)), ("P5",  (-6.2, -5.2)),
        ("P1",  (-2.1, -5.2)), ("P2",  ( 2.1, -5.2)), ("P6",  ( 6.2, -5.2)), ("PO7", (-5.2, -7.1)),
        ("PO3", (-2.9, -7.1)), ("POz", ( 0.0, -7.1)), ("PO4", ( 2.9, -7.1)), ("PO8", ( 5.2, -7.1)),
    ]
    return [MontageChannel(channel=f"E{i+1}", label=lbl, position_2d=xy)
            for i, (lbl, xy) in enumerate(sites)]

acticap_32ch_posterior_montage 

acticap_32ch_posterior_montage()

NCIL's 32-channel posterior actiCAP layout for the actiCHamp — every 10-10 site at or behind the TP/CP row, plus FCz and Cz as midline anchors. Not a stock actiCAP montage: the electrodes are plugged into the posterior holders of the 64-holder cap, concentrating coverage over visual cortex.

Channel order is the amplifier's, as wired at the cap (confirmed with Aaron 2026-07-15): ch1 FCz, ch2 Cz, then complete rows sweeping left→right, rows running anterior→posterior. PO9/PO10 belong to the O row (they sit at occipital height), not the PO row. That fills 1..32 exactly — the 64Ch Standard-2 cap has precisely 30 holders from the TP/CP row back, and 30 + FCz + Cz = 32.

Positions reuse :func:acticap_64ch_montage's coordinates for every shared site, so the two montages plot identically; FCz is new (midline at the FC row's y). Same convention as the other montages — x = left(-)/right(+), y = anterior(+)/posterior(-), Cz at the origin, cm.

The actiCHamp is reference-free, so this montage decodes with a common-average reference and an occipital ROI automatically (32 > :data:ssvep.spatial.BIG_MONTAGE); it carries the whole validated occ-8 (O1 Oz O2 POz PO3 PO4 PO7 PO8), so no position-based fallback is needed. ⚠️ The amp still streams 64 channels with only module 1 populated (33-64 rail at +full-scale) — the LSL bridge truncates to this montage's length; see :func:ssvep.runtime.actichamp.stream_to_lsl.

Source code in src/ssvep/stim/builder.py 
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
def acticap_32ch_posterior_montage() -> list[MontageChannel]:
    """NCIL's **32-channel posterior** actiCAP layout for the actiCHamp — every 10-10 site at or behind
    the TP/CP row, plus FCz and Cz as midline anchors. Not a stock actiCAP montage: the electrodes are
    plugged into the posterior holders of the 64-holder cap, concentrating coverage over visual cortex.

    Channel order is the amplifier's, as wired at the cap (confirmed with Aaron 2026-07-15): ch1 FCz,
    ch2 Cz, then complete rows sweeping **left→right**, rows running **anterior→posterior**. PO9/PO10
    belong to the **O row** (they sit at occipital height), not the PO row. That fills 1..32 exactly —
    the 64Ch Standard-2 cap has precisely 30 holders from the TP/CP row back, and 30 + FCz + Cz = 32.

    Positions reuse :func:`acticap_64ch_montage`'s coordinates for every shared site, so the two
    montages plot identically; ``FCz`` is new (midline at the FC row's ``y``). Same convention as the
    other montages — ``x`` = left(-)/right(+), ``y`` = anterior(+)/posterior(-), Cz at the origin, cm.

    The actiCHamp is **reference-free**, so this montage decodes with a common-average reference and an
    occipital ROI automatically (32 > :data:`ssvep.spatial.BIG_MONTAGE`); it carries the whole validated
    occ-8 (O1 Oz O2 POz PO3 PO4 PO7 PO8), so no position-based fallback is needed. ⚠️ The amp still
    *streams* 64 channels with only module 1 populated (33-64 rail at +full-scale) — the LSL bridge
    truncates to this montage's length; see :func:`ssvep.runtime.actichamp.stream_to_lsl`.
    """
    # (10-10 site, (x_cm, y_cm)) in the cap's wiring order 1..32.
    sites = [
        ("FCz", ( 0.0,  2.9)), ("Cz",  ( 0.0,  0.0)),                                   # midline anchors
        ("TP9", (-8.8, -3.5)), ("TP7", (-8.6, -2.9)), ("CP5", (-6.3, -2.9)),            # TP/CP row
        ("CP3", (-4.2, -2.9)), ("CP1", (-2.1, -2.9)), ("CPz", ( 0.0, -2.9)),
        ("CP2", ( 2.1, -2.9)), ("CP4", ( 4.2, -2.9)), ("CP6", ( 6.3, -2.9)),
        ("TP8", ( 8.6, -2.9)), ("TP10",( 8.8, -3.5)),
        ("P7",  (-8.1, -5.2)), ("P5",  (-6.2, -5.2)), ("P3",  (-4.2, -5.2)),            # P row
        ("P1",  (-2.1, -5.2)), ("Pz",  ( 0.0, -5.2)), ("P2",  ( 2.1, -5.2)),
        ("P4",  ( 4.2, -5.2)), ("P6",  ( 6.2, -5.2)), ("P8",  ( 8.1, -5.2)),
        ("PO7", (-5.2, -7.1)), ("PO3", (-2.9, -7.1)), ("POz", ( 0.0, -7.1)),            # PO row
        ("PO4", ( 2.9, -7.1)), ("PO8", ( 5.2, -7.1)),
        ("PO9", (-5.5, -8.0)), ("O1",  (-2.9, -8.8)), ("Oz",  ( 0.0, -9.3)),            # O row
        ("O2",  ( 2.9, -8.8)), ("PO10",( 5.5, -8.0)),
    ]
    return [MontageChannel(channel=f"E{i+1}", label=lbl, position_2d=xy)
            for i, (lbl, xy) in enumerate(sites)]

device_for_board 

device_for_board(board)

Reverse-lookup a device display name from a BrainFlow board id (for loading manifests).

Source code in src/ssvep/stim/builder.py 
836
837
838
839
840
841
def device_for_board(board: str) -> Optional[str]:
    """Reverse-lookup a device display name from a BrainFlow board id (for loading manifests)."""
    for name, d in DEVICES.items():
        if d["board"] == board:
            return name
    return None

device_electrode_types 

device_electrode_types(name)

Electrode types this device can use, or None if it is not constrained.

A device whose electrodes are fixed in hardware (the actiCHamp's actiCAP, the Unicorn's moulded-in dry contacts) implies its electrode type; a flexible OpenBCI board does not.

Source code in src/ssvep/stim/builder.py 
844
845
846
847
848
849
850
def device_electrode_types(name: str) -> Optional[list[str]]:
    """Electrode types this device can use, or None if it is not constrained.

    A device whose electrodes are fixed in hardware (the actiCHamp's actiCAP, the Unicorn's moulded-in
    dry contacts) implies its electrode type; a flexible OpenBCI board does not.
    """
    return list(DEVICES.get(name, {}).get("electrode_types") or []) or None

device_preps 

device_preps(name)

Skin/electrode preps this device allows, or None if it is not constrained.

Source code in src/ssvep/stim/builder.py 
853
854
855
def device_preps(name: str) -> Optional[list[str]]:
    """Skin/electrode preps this device allows, or None if it is not constrained."""
    return list(DEVICES.get(name, {}).get("preps") or []) or None

device_headsets 

device_headsets(name)

The electrode carriers this device can wear, or None if it accepts any of them.

None = the OpenBCI boards, which take any of the interchangeable OpenBCI carriers. A list = the device constrains the choice: one entry means the montage is fixed in hardware (the Unicorn's moulded-in electrodes) and the Design tab locks it; several means the operator must pick which cap is on the participant (the actiCHamp's 64-ch vs 32-ch posterior actiCAP).

Source code in src/ssvep/stim/builder.py 
858
859
860
861
862
863
864
865
866
867
def device_headsets(name: str) -> Optional[list[str]]:
    """The electrode carriers this device can wear, or ``None`` if it accepts any of them.

    ``None`` = the OpenBCI boards, which take any of the interchangeable OpenBCI carriers. A list =
    the device constrains the choice: one entry means the montage is fixed in hardware (the Unicorn's
    moulded-in electrodes) and the Design tab locks it; several means the operator must pick which cap
    is on the participant (the actiCHamp's 64-ch vs 32-ch posterior actiCAP).
    """
    hs = DEVICES.get(name, {}).get("headsets")
    return list(hs) if hs else None

device_fixed_headset 

device_fixed_headset(name)

The carrier for a device whose montage is fixed in hardware, else None.

Only devices with exactly one allowed carrier are "fixed" — a device offering a choice (the actiCHamp) returns None here. Use :func:device_headsets to restrict a pick-list.

Source code in src/ssvep/stim/builder.py 
870
871
872
873
874
875
876
877
def device_fixed_headset(name: str) -> Optional[str]:
    """The carrier for a device whose montage is fixed in hardware, else ``None``.

    Only devices with exactly one allowed carrier are "fixed" — a device offering a choice (the
    actiCHamp) returns ``None`` here. Use :func:`device_headsets` to restrict a pick-list.
    """
    hs = device_headsets(name)
    return hs[0] if hs and len(hs) == 1 else None

canonical_headset 

canonical_headset(name)

The current name for a carrier, mapping any pre-rename alias forward. Unknown names pass through unchanged — an unrecognised carrier is data to preserve, not an error to raise.

Source code in src/ssvep/stim/builder.py 
918
919
920
921
922
923
def canonical_headset(name: Optional[str]) -> Optional[str]:
    """The current name for a carrier, mapping any pre-rename alias forward. Unknown names pass
    through unchanged — an unrecognised carrier is data to preserve, not an error to raise."""
    if name is None:
        return None
    return HEADSET_ALIASES.get(name, name)

headset_electrode_type 

headset_electrode_type(name)

The electrode kind implied by a carrier, or None for a carrier we don't recognise.

The single place the derivation lives, so the Design tab, the protocol-level acquisition dialog and the per-session override all record the same string for the same headset (#147).

Source code in src/ssvep/stim/builder.py 
947
948
949
950
951
952
953
def headset_electrode_type(name: str) -> Optional[str]:
    """The electrode kind implied by a carrier, or ``None`` for a carrier we don't recognise.

    The single place the derivation lives, so the Design tab, the protocol-level acquisition dialog
    and the per-session override all record the same string for the same headset (#147).
    """
    return HEADSET_ELECTRODE_TYPES.get(canonical_headset(name))

headset_reference_ground 

headset_reference_ground(name)

(reference, ground) implied by a carrier, or (None, None) when it implies neither.

Source code in src/ssvep/stim/builder.py 
970
971
972
def headset_reference_ground(name: str) -> tuple[Optional[str], Optional[str]]:
    """``(reference, ground)`` implied by a carrier, or ``(None, None)`` when it implies neither."""
    return HEADSET_REFERENCE_GROUND.get(canonical_headset(name), (None, None))

example_protocol 

example_protocol()

The worked 9-target, 36–44 Hz protocol (à la schemas/manifest_example.json), on the go-forward OpenBCI Cyton 8-ch occipital montage. Mirrors the pilot's operating band.

Source code in src/ssvep/stim/builder.py 
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
def example_protocol() -> RunSpec:
    """The worked 9-target, 36–44 Hz protocol (à la schemas/manifest_example.json), on the
    go-forward OpenBCI Cyton 8-ch occipital montage. Mirrors the pilot's operating band."""
    freqs = [36.0, 37.0, 38.0, 39.0, 40.0, 41.0, 42.0, 43.0, 44.0]
    pos = grid_positions(3, 3, spread=GRID_SPREAD_DEG)
    stimuli = [
        Target(id=f"S{i}", position_xy=pos[i], size=TARGET_SIZE_DEG, shape="rect",
               flicker=Flicker("sine", f))
        for i, f in enumerate(freqs)
    ]
    montage = headset_montage(DEFAULT_HEADSET)
    return RunSpec(
        run=Run(name="ssvep-highfreq-9class", task="ssvep",
                description="9-class high-frequency SSVEP, 36–44 Hz operating band"),
        display=Display(refresh_hz=240),
        stimuli=stimuli,
        conditions=[Condition(id="mid", label="mid 36–44 Hz (operating)",
                              stimulus_freqs_hz=freqs,
                              params={"band": "mid", "seizure_safety": "high-freq, lower risk"})],
        timing=Timing(cue_s=1.5, stim_s=4.0, iti_s=0.75),
        acquisition=Acquisition(device=DEFAULT_DEVICE, amplifier="OpenBCI Cyton", board="cyton",
                                sfreq_hz=250, reference=OCCIPITAL_SSVEP_REFERENCE, ground=OCCIPITAL_SSVEP_GROUND,
                                headset=DEFAULT_HEADSET, montage=montage),
        blocks_per_condition=3, trials_per_block=9,
    )

jfpm_phase_cycles 

jfpm_phase_cycles(n, step_rad=0.35 * math.pi)

JFPM phase schedule for n targets, returned in cycles (the manifest unit).

Joint frequency–phase modulation (Chen et al. 2015) assigns target k the phase k · step radians (canonical step 0.35π). The manifest stores phase in cycles, so this converts: phase_cycles = (k · step_rad) / (2π) (wrapped to [0,1)).

Source code in src/ssvep/stim/builder.py 
1013
1014
1015
1016
1017
1018
1019
1020
def jfpm_phase_cycles(n: int, step_rad: float = 0.35 * math.pi) -> list[float]:
    """JFPM phase schedule for ``n`` targets, returned in **cycles** (the manifest unit).

    Joint frequency–phase modulation (Chen et al. 2015) assigns target ``k`` the phase
    ``k · step`` radians (canonical step ``0.35π``). The manifest stores phase in cycles, so this
    converts: ``phase_cycles = (k · step_rad) / (2π)`` (wrapped to [0,1)).
    """
    return [((k * step_rad) / (2.0 * math.pi)) % 1.0 for k in range(n)]

preset_high_freq_9class 

preset_high_freq_9class()

Preset: the worked high-frequency 9-class protocol (alias of :func:example_protocol).

Source code in src/ssvep/stim/builder.py 
1023
1024
1025
def preset_high_freq_9class() -> RunSpec:
    """Preset: the worked high-frequency 9-class protocol (alias of :func:`example_protocol`)."""
    return example_protocol()

preset_alpha_9class 

preset_alpha_9class()

Preset: the 9-target grid of :func:example_protocol, but flickering in the alpha band (8–12 Hz, 0.5 Hz steps) instead of 36–44 Hz.

Purpose: a fair test for the dry/prototype headsets. The high-frequency (36–44 Hz) response is tiny, and on 2026-07-14 both dry systems (Zeist, Unicorn) decoded that band at chance; alpha-band SSVEP is an order of magnitude larger, so it separates "electrode can't reach the cortex" from "the response is just small up here". Same layout/timing/montage as the high-freq preset so the only changed factor is stimulation frequency.

⚠️ Two caveats, both surfaced by validate() / worth stating to the operator: * Photosensitivity (COMPLIANCE R6). 8–12 Hz sits in the provocative ~3–30 Hz band, so validate() raises elevated/high-risk warnings. This is a template to adapt under an appropriate REB amendment (docs/REB_AMENDMENTS.md), not to run as-is. * Endogenous-alpha confound. These frequencies overlap the participant's own ~10 Hz alpha rhythm (strongest eyes-open, occipital), which can inflate or contaminate the SSVEP — the opposite trade-off from the high-freq band. Interpret decode accuracy with that in mind.

Source code in src/ssvep/stim/builder.py 
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
def preset_alpha_9class() -> RunSpec:
    """Preset: the 9-target grid of :func:`example_protocol`, but flickering in the **alpha band**
    (8–12 Hz, 0.5 Hz steps) instead of 36–44 Hz.

    Purpose: a *fair test for the dry/prototype headsets*. The high-frequency (36–44 Hz) response is
    tiny, and on 2026-07-14 both dry systems (Zeist, Unicorn) decoded that band at chance; alpha-band
    SSVEP is an order of magnitude larger, so it separates "electrode can't reach the cortex" from
    "the response is just small up here". Same layout/timing/montage as the high-freq preset so the
    only changed factor is stimulation frequency.

    ⚠️ Two caveats, both surfaced by ``validate()`` / worth stating to the operator:
    * **Photosensitivity (COMPLIANCE R6).** 8–12 Hz sits in the provocative ~3–30 Hz band, so
      ``validate()`` raises elevated/high-risk warnings. This is a template to adapt under an
      appropriate REB amendment (``docs/REB_AMENDMENTS.md``), **not** to run as-is.
    * **Endogenous-alpha confound.** These frequencies overlap the participant's own ~10 Hz alpha
      rhythm (strongest eyes-open, occipital), which can inflate or contaminate the SSVEP — the
      opposite trade-off from the high-freq band. Interpret decode accuracy with that in mind.
    """
    freqs = [8.0, 8.5, 9.0, 9.5, 10.0, 10.5, 11.0, 11.5, 12.0]
    pos = grid_positions(3, 3, spread=GRID_SPREAD_DEG)
    stimuli = [
        Target(id=f"S{i}", position_xy=pos[i], size=TARGET_SIZE_DEG, shape="rect",
               flicker=Flicker("sine", f))
        for i, f in enumerate(freqs)
    ]
    return RunSpec(
        run=Run(name="ssvep-alpha-9class", task="ssvep",
                description="9-class alpha-band SSVEP, 8–12 Hz (dry-headset test)"),
        display=Display(refresh_hz=240),
        stimuli=stimuli,
        conditions=[Condition(id="alpha", label="alpha 8–12 Hz",
                              stimulus_freqs_hz=freqs,
                              params={"band": "alpha",
                                      "seizure_safety": "LOW — 8–12 Hz provocative band; validate() warns",
                                      "note": "endogenous-alpha confound possible"})],
        timing=Timing(cue_s=1.5, stim_s=4.0, iti_s=0.75),
        acquisition=Acquisition(device=DEFAULT_DEVICE, amplifier="OpenBCI Cyton", board="cyton",
                                sfreq_hz=250, reference=OCCIPITAL_SSVEP_REFERENCE,
                                ground=OCCIPITAL_SSVEP_GROUND, headset=DEFAULT_HEADSET,
                                montage=headset_montage(DEFAULT_HEADSET)),
        blocks_per_condition=3, trials_per_block=9,
    )

preset_jfpm_speller 

preset_jfpm_speller(n_cols=5, n_rows=8, f0=8.0, df=0.2, phase_step_rad=0.35 * math.pi)

Preset: a Chen et al. 2015-style JFPM speller (default 40 targets, 8–15.8 Hz, 0.2 Hz steps).

Frequencies step by df from f0; phases follow the JFPM schedule. NOTE: these low frequencies fall in the ~3–30 Hz provocative band, so validate() will warn — for this REB prefer the high-frequency band. Provided as a canonical template to adapt, not to run as-is.

Source code in src/ssvep/stim/builder.py 
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
def preset_jfpm_speller(n_cols: int = 5, n_rows: int = 8, f0: float = 8.0, df: float = 0.2,
                        phase_step_rad: float = 0.35 * math.pi) -> RunSpec:
    """Preset: a Chen et al. 2015-style JFPM speller (default 40 targets, 8–15.8 Hz, 0.2 Hz steps).

    Frequencies step by ``df`` from ``f0``; phases follow the JFPM schedule. NOTE: these low
    frequencies fall in the ~3–30 Hz provocative band, so ``validate()`` will warn — for this REB
    prefer the high-frequency band. Provided as a canonical template to adapt, not to run as-is.
    """
    n = n_cols * n_rows
    freqs = [round(f0 + i * df, 3) for i in range(n)]
    phases = jfpm_phase_cycles(n, phase_step_rad)
    # A speller packs more targets, so it needs a wider field than the 3x3 presets to stay separated.
    pos = grid_positions(n_cols, n_rows, spread=1.5 * GRID_SPREAD_DEG)
    stimuli = [
        Target(id=f"S{i}", position_xy=pos[i], size=SPELLER_SIZE_DEG, shape="rect",
               flicker=Flicker("sine", freqs[i], phase=phases[i]))
        for i in range(n)
    ]
    return RunSpec(
        run=Run(name="ssvep-jfpm-speller", task="ssvep",
                description=f"{n}-target JFPM speller ({freqs[0]:g}–{freqs[-1]:g} Hz)"),
        display=Display(refresh_hz=240),
        stimuli=stimuli,
        conditions=[Condition(id="jfpm", label=f"JFPM {n}-class",
                              stimulus_freqs_hz=freqs, params={"paradigm": "jfpm"})],
        timing=Timing(cue_s=1.0, stim_s=4.0, iti_s=0.5),
        acquisition=Acquisition(device="OpenBCI Cyton (8-ch)", amplifier="OpenBCI Cyton",
                                sfreq_hz=250, reference=OCCIPITAL_SSVEP_REFERENCE,
                                ground=OCCIPITAL_SSVEP_GROUND, headset=DEFAULT_HEADSET,
                                montage=headset_montage(DEFAULT_HEADSET)),
        blocks_per_condition=3, trials_per_block=n,
    )

preset_checkerboard_reversal 

preset_checkerboard_reversal()

Preset: a 4-target contrast-reversing checkerboard example in the study-safe high band.

Pattern-reversal (the two phases swap at the flicker frequency) with no net luminance change — the classic pattern-reversal SSVEP/VEP stimulus. Uses degrees-of-visual-angle geometry.

Source code in src/ssvep/stim/builder.py 
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
def preset_checkerboard_reversal() -> RunSpec:
    """Preset: a 4-target contrast-reversing checkerboard example in the study-safe high band.

    Pattern-reversal (the two phases swap at the flicker frequency) with no net luminance change —
    the classic pattern-reversal SSVEP/VEP stimulus. Uses degrees-of-visual-angle geometry."""
    freqs = [30.0, 32.0, 34.0, 36.0]
    pos = grid_positions(2, 2, spread=5.0)          # ±5° grid
    stimuli = [
        Target(id=f"S{i}", position_xy=pos[i], size=(3.0, 3.0), shape="rect",
               pattern="checkerboard", pattern_reversal=True, spatial_freq_cpd=0.5,
               flicker=Flicker("square", freqs[i]))
        for i in range(4)
    ]
    return RunSpec(
        run=Run(name="ssvep-checkerboard-reversal", task="ssvep",
                description="4-target contrast-reversing checkerboard, 30–36 Hz"),
        display=Display(refresh_hz=240),
        stimuli=stimuli,
        conditions=[Condition(id="checker", label="checkerboard reversal 30–36 Hz",
                              stimulus_freqs_hz=freqs, params={"pattern": "checkerboard-reversal"})],
        timing=Timing(cue_s=1.5, stim_s=4.0, iti_s=0.75),
        acquisition=Acquisition(device="OpenBCI Cyton (8-ch)", amplifier="OpenBCI Cyton",
                                sfreq_hz=250, reference=OCCIPITAL_SSVEP_REFERENCE,
                                ground=OCCIPITAL_SSVEP_GROUND, headset=DEFAULT_HEADSET,
                                montage=headset_montage(DEFAULT_HEADSET)),
        blocks_per_condition=3, trials_per_block=4,
    )

preset_resting_state 

preset_resting_state()

Preset: the tone-cued eyes-open / eyes-closed resting run — the odd duck with no flicker.

Resting is a paradigm, not a stimulus layout, so it carries paradigm='resting' and an empty stimuli list; its structure lives in design.resting (the manifest's authoritative record of it). Everything else about a protocol still applies unchanged — the device/montage/sampling rate chosen on the Design tab, the operator/consent/subject captured at Setup, the impedance check — which is the whole reason it's expressed as a manifest rather than a special-case dialog: it locks, records, and saves through exactly the same path as an SSVEP run.

Two jobs (see ssvep.stim.resting / ssvep.analysis.resting_paf): * PAF calibration — the participant's individual peak alpha frequency, so SSVEP designs can steer flicker away from PAF and its harmonics. * Negative control + posterior QC — a well-behaved decoder must sit at chance on data with no flicker in it, and eyes-closed alpha should exceed eyes-open (reactivity ratio > 1). A ratio near 1 condemns the posterior montage before an SSVEP run is wasted on it (the sub-903/904 lesson).

Literature defaults (Klimesch 1999; Barry et al. 2007): 6 × 60 s alternating from eyes-open → 3 min eyes-closed. Grey background + central fixation cross during eyes-open blocks.

⚠️ Not in the approved REB protocol (V6 covers visual flicker only — no resting recording and no auditory stimulation). Collection needs amendment A10; see docs/REB_AMENDMENTS.md.

Source code in src/ssvep/stim/builder.py 
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
def preset_resting_state() -> RunSpec:
    """Preset: the tone-cued **eyes-open / eyes-closed resting** run — the odd duck with no flicker.

    Resting is a *paradigm*, not a stimulus layout, so it carries ``paradigm='resting'`` and an empty
    ``stimuli`` list; its structure lives in ``design.resting`` (the manifest's authoritative record of
    it). Everything else about a protocol still applies unchanged — the device/montage/sampling rate
    chosen on the Design tab, the operator/consent/subject captured at Setup, the impedance check —
    which is the whole reason it's expressed as a manifest rather than a special-case dialog: it locks,
    records, and saves through exactly the same path as an SSVEP run.

    Two jobs (see ``ssvep.stim.resting`` / ``ssvep.analysis.resting_paf``):
      * **PAF calibration** — the participant's individual peak alpha frequency, so SSVEP designs can
        steer flicker away from PAF and its harmonics.
      * **Negative control + posterior QC** — a well-behaved decoder must sit at chance on data with no
        flicker in it, and eyes-closed alpha should exceed eyes-open (reactivity ratio > 1). A ratio
        near 1 condemns the posterior montage *before* an SSVEP run is wasted on it (the sub-903/904
        lesson).

    Literature defaults (Klimesch 1999; Barry et al. 2007): 6 × 60 s alternating from eyes-open →
    3 min eyes-closed. Grey background + central fixation cross during eyes-open blocks.

    ⚠️ **Not in the approved REB protocol** (V6 covers visual flicker only — no resting recording and
    no auditory stimulation). Collection needs amendment A10; see ``docs/REB_AMENDMENTS.md``.
    """
    rest = RestingProtocol()                      # literature defaults: 6 × 60 s, start eyes-open
    return RunSpec(
        paradigm=PARADIGM_RESTING,
        resting=rest,
        run=Run(name="resting-eyes-open-closed", task=rest.task,
                description="Tone-cued eyes-open/eyes-closed resting state "
                            "(peak alpha frequency + alpha reactivity; no flicker)"),
        # Neutral grey, not black: the eyes-open blocks are a fixation task, and a black field makes
        # the eyes-open/eyes-closed luminance step (the thing alpha reacts to) needlessly extreme.
        display=Display(refresh_hz=240, background_rgb=(0.5, 0.5, 0.5)),
        stimuli=[],                               # no flicker — the defining property of the paradigm
        conditions=[Condition(id=EYES_OPEN, label="eyes open", stimulus_freqs_hz=[],
                              params={"cue_tone_hz": rest.high_tone_hz, "cue": "HIGH tone"}),
                    Condition(id=EYES_CLOSED, label="eyes closed", stimulus_freqs_hz=[],
                              params={"cue_tone_hz": rest.low_tone_hz, "cue": "LOW tone"})],
        # Schema requires per-trial timing; for resting a "trial" IS a block, and there is no cue
        # lead-in or ITI. The authoritative structure is design.resting — this just keeps the
        # contract satisfied without inventing a second source of truth.
        timing=Timing(cue_s=0.0, stim_s=rest.block_s, iti_s=0.0, onset_offset_s=0.0),
        acquisition=Acquisition(device=DEFAULT_DEVICE, amplifier="OpenBCI Cyton", board="cyton",
                                sfreq_hz=250, reference=OCCIPITAL_SSVEP_REFERENCE,
                                ground=OCCIPITAL_SSVEP_GROUND, headset=DEFAULT_HEADSET,
                                montage=headset_montage(DEFAULT_HEADSET)),
        markers=Markers(schema="rest/{condition}/tone/{target_freq_hz}/block/{block}/{segment}",
                        segments=["block_start", "cue", "block_end"]),
    )

ssvep.runtime — sessions and acquisition

Run plan

ssvep.runtime.run_plan

Run plan: expand a run manifest into a deterministic, ordered list of trials + markers.

The runtime runs ONLY built, saved, versioned protocols. Given a manifest, this derives the condition × block × trial structure, assigns a cued target (frequency) per trial, and produces the marker sequence — all reproducibly from a seed (recorded in provenance). The runner attaches real LSL timestamps when each marker fires.

RunPlan dataclass

Source code in src/ssvep/runtime/run_plan.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
@dataclass
class RunPlan:
    trials: list[Trial]
    timing: dict               # {cue_s, stim_s, iti_s, onset_offset_s}
    marker_template: str
    seed: int | None
    resting: dict | None = None   # design.resting when this is a resting run; None for SSVEP

    @classmethod
    def from_manifest(cls, manifest: dict, seed: int | None = 0) -> "RunPlan":
        design = manifest["design"]
        timing = design["timing"]
        template = manifest["markers"].get(
            "schema", "cond/{condition}/freq/{target_freq_hz}/block/{block}/trial/{trial}/{segment}")
        # Resting is a block paradigm with no targets and no cued frequencies: its "trials" are the
        # eyes-open/eyes-closed blocks, and the real schedule is expanded frame-wise by the renderer
        # from design.resting (ssvep.stim.resting.build_resting_schedule). Planning it as SSVEP would
        # divide by an empty frequency set.
        if manifest_io.is_resting(manifest):
            return cls(trials=_resting_trials(design["resting"]), timing=timing,
                       marker_template=template, seed=seed, resting=dict(design["resting"]))

        freq_to_id = {round(float(s["flicker"]["freq_hz"]), 6): s["id"] for s in manifest["stimuli"]}
        rng = random.Random(seed) if seed is not None else None

        trials: list[Trial] = []
        for cond in design["conditions"]:
            freqs = [float(f) for f in cond["stimulus_freqs_hz"]]
            n_blocks = design.get("blocks_per_condition", 1)
            per_block = design.get("trials_per_block", len(freqs))
            for b in range(1, n_blocks + 1):
                order = _block_order(freqs, per_block, rng)
                for i, f in enumerate(order, start=1):
                    tid = freq_to_id.get(round(f, 6), _nearest_id(freq_to_id, f))
                    trials.append(Trial(cond["id"], b, i, tid, f))
        return cls(trials=trials, timing=timing, marker_template=template, seed=seed)

    # -- convenience --------------------------------------------------------
    @property
    def is_resting(self) -> bool:
        return self.resting is not None

    @property
    def n_trials(self) -> int:
        return len(self.trials)

    def estimated_duration_s(self) -> float:
        if self.is_resting:      # contiguous blocks, no cue/ITI, plus the quiet lead-in
            return (float(self.resting.get("lead_in_s", 0.0))
                    + float(self.resting["n_blocks"]) * float(self.resting["block_s"]))
        per = (self.timing["cue_s"] + self.timing["stim_s"]
               + float(self.timing.get("onset_offset_s", 0.0) or 0.0)   # extra flicker presented
               + self.timing["iti_s"])
        return per * self.n_trials

    def markers_for_trial(self, trial: Trial) -> list[Marker]:
        """cue → stim_on → stim_off markers for one trial (block_start/end added by the runner)."""
        common = dict(condition=trial.condition, trial=trial.trial, block=trial.block,
                      target_id=trial.target_id, target_freq_hz=trial.target_freq_hz)
        return [
            Marker(segment="cue", **common),
            Marker(segment="stim_on",
                   trial_dur_s=self.timing["stim_s"] + float(self.timing.get("onset_offset_s", 0.0) or 0.0),
                   **common),
            Marker(segment="stim_off", **common),
        ]

markers_for_trial 

markers_for_trial(trial)

cue → stim_on → stim_off markers for one trial (block_start/end added by the runner).

Source code in src/ssvep/runtime/run_plan.py 
81
82
83
84
85
86
87
88
89
90
91
def markers_for_trial(self, trial: Trial) -> list[Marker]:
    """cue → stim_on → stim_off markers for one trial (block_start/end added by the runner)."""
    common = dict(condition=trial.condition, trial=trial.trial, block=trial.block,
                  target_id=trial.target_id, target_freq_hz=trial.target_freq_hz)
    return [
        Marker(segment="cue", **common),
        Marker(segment="stim_on",
               trial_dur_s=self.timing["stim_s"] + float(self.timing.get("onset_offset_s", 0.0) or 0.0),
               **common),
        Marker(segment="stim_off", **common),
    ]

Session protocol and resolver

ssvep.runtime.session

The session level: a protocol (template) → a concrete, ordered, frozen set of runs (#43).

Terminology (CLAUDE.md §2.1): a run is one continuous recording (one run manifest, planned by :mod:ssvep.runtime.run_plan); a session is an ordered set of runs; a protocol is the instructions for running a session on a participant. This module is the protocol layer that #42 unblocked — it reuses the module name freed when the old session.py (a run planner) became run_plan.py.

A :class:SessionProtocol is a template: authored once, instantiated per participant. It contains its run manifests and groups them; each group declares how its runs are ordered for a given participant. :func:resolve_session turns (protocol, sub-XXX, ses-YYY) into an ordered list of :class:ResolvedRun — each carrying the frozen run manifest, a derived run_index, and the position block stamped into its recording's sidecar. The order is derived and recorded, never inferred from files on disk (the manifest-spine rule, one level up). A protocol never redefines a run's design: to run the same design on different hardware, add two run manifests — there are no per-run overrides.

Protocol schema v1.2 (#115) embeds each run manifest in the protocol; before it, a protocol held a path + pinned fingerprint and the resolver loaded the file, hard-stopping on drift. Embedding makes a protocol one self-contained artifact — copy it to an acquisition PC and the whole design goes with it — and means a saved protocol always runs, where before it could be silently broken by anyone editing a run manifest it pointed at. What was lost with the resolve-time drift stop is regained by :meth:SessionProtocol.source_report, which asks the same question at design time and can be acted on. Pre-1.2 protocols still resolve exactly as they did, drift stop included.

The core (protocol model + ordering + resolver) is pure and unit-tested: it scans no sourcedata, and for an embedded protocol it touches no filesystem at all. The helpers :func:next_session / :func:session_exists are separate, so the resolver stays pure.

ProtocolDriftError

Bases: RuntimeError

A referenced run manifest's on-disk fingerprint no longer matches the protocol's pin.

Raised by :func:resolve_session (unless allow_drift): the design a participant would run has silently diverged from the design the protocol was authored against. A design change must be a deliberate new protocol version, never a side effect of editing a referenced run manifest — so this is a hard stop, not a warning (CLAUDE.md §2: declared, never inferred).

Source code in src/ssvep/runtime/session.py 
51
52
53
54
55
56
57
58
class ProtocolDriftError(RuntimeError):
    """A referenced run manifest's on-disk fingerprint no longer matches the protocol's pin.

    Raised by :func:`resolve_session` (unless ``allow_drift``): the design a participant would run has
    silently diverged from the design the protocol was authored against. A design change must be a
    deliberate new protocol version, never a side effect of editing a referenced run manifest — so
    this is a hard stop, not a warning (CLAUDE.md §2: declared, never inferred).
    """

RunRef dataclass

One run of a group: the run manifest itself, plus where it was copied from.

Since protocol schema v1.2 (#115) the manifest is embedded and authoritative — the resolver runs :attr:manifest and never reads the store. That makes a protocol a single self-contained file (copy it to an acquisition PC and the whole design travels with it) and makes a saved protocol un-breakable by someone editing a run manifest it happens to point at.

:attr:run_manifest / :attr:fingerprint survive as the source pointer: where this copy came from and what it fingerprinted at embed time. Nothing at record time reads them. They exist so :meth:source_status can answer "has the file I copied this from moved on?" at design time, which is where a design question belongs — the previous scheme could only answer it at resolve time, i.e. by refusing to start a session with the participant already in the chair.

A pre-1.2 protocol has no :attr:manifest. Those still load, and :func:resolve_session still resolves them the old way — store lookup plus a fingerprint check that hard-stops on drift.

Source code in src/ssvep/runtime/session.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
@dataclass
class RunRef:
    """One run of a group: the run manifest itself, plus where it was copied from.

    Since protocol schema v1.2 (#115) the manifest is **embedded and authoritative** — the resolver
    runs :attr:`manifest` and never reads the store. That makes a protocol a single self-contained
    file (copy it to an acquisition PC and the whole design travels with it) and makes a saved
    protocol un-breakable by someone editing a run manifest it happens to point at.

    :attr:`run_manifest` / :attr:`fingerprint` survive as the **source pointer**: where this copy came
    from and what it fingerprinted at embed time. Nothing at record time reads them. They exist so
    :meth:`source_status` can answer "has the file I copied this from moved on?" at *design* time,
    which is where a design question belongs — the previous scheme could only answer it at resolve
    time, i.e. by refusing to start a session with the participant already in the chair.

    A pre-1.2 protocol has no :attr:`manifest`. Those still load, and :func:`resolve_session` still
    resolves them the old way — store lookup plus a fingerprint check that hard-stops on drift.
    """
    run_manifest: str = ""             # source path, relative to the run-manifest store
    fingerprint: str = ""              # the manifest's fingerprint when it was embedded/pinned
    label: str | None = None           # optional per-position label; falls back to run.name
    manifest: dict | None = None       # the embedded run manifest (v1.2+); None ⇒ legacy reference

    @property
    def embedded(self) -> bool:
        return self.manifest is not None

    def to_dict(self) -> dict:
        if not self.embedded:                              # legacy reference — round-trip unchanged
            d = {"run_manifest": self.run_manifest, "fingerprint": self.fingerprint}
            if self.label:
                d["label"] = self.label
            return d
        d: dict = {"manifest": self.manifest}
        if self.run_manifest:
            d["source"] = {"run_manifest": self.run_manifest, "fingerprint": self.fingerprint}
        if self.label:
            d["label"] = self.label
        return d

    @classmethod
    def from_dict(cls, d: dict) -> "RunRef":
        src = d.get("source") or {}
        return cls(
            # Read the source block first (v1.2), then the legacy top-level keys — a v1.2 run keeps
            # its provenance in 'source', a v1.0/v1.1 run has it at the top level and no manifest.
            run_manifest=src.get("run_manifest") or d.get("run_manifest") or "",
            fingerprint=src.get("fingerprint") or d.get("fingerprint") or "",
            label=d.get("label"),
            manifest=(manifest_io.canonicalize_manifest(d["manifest"])
                      if d.get("manifest") is not None else None))

    @classmethod
    def pin(cls, run_manifest: str, store_dir, label: str | None = None) -> "RunRef":
        """Embed the run manifest at ``store_dir / run_manifest``, recording where it came from.

        The manifest is loaded through :func:`ssvep.io.manifest.load_manifest`, so it is canonicalized
        to v1.3 shape and schema-validated here — an unloadable or invalid run manifest fails while
        someone is authoring, not while someone is recording.
        """
        m = manifest_io.load_manifest(Path(store_dir) / run_manifest)
        return cls(run_manifest=run_manifest, fingerprint=manifest_io.manifest_fingerprint(m),
                   label=label, manifest=m)

    def resolve_manifest(self, store_dir, *, allow_drift: bool = False,
                         protocol_name: str = "") -> tuple[dict, str]:
        """This run's manifest and its fingerprint — embedded copy, or the legacy store lookup.

        Embedded (v1.2+): returned as-is. There is nothing to verify, because there is nothing else
        it could have come from; the protocol *is* the design.

        Legacy (v1.0/v1.1): loaded from ``store_dir`` and re-fingerprinted, raising
        :class:`ProtocolDriftError` on a mismatch unless ``allow_drift``.
        """
        if self.embedded:
            return self.manifest, (self.fingerprint or manifest_io.manifest_fingerprint(self.manifest))
        if store_dir is None:
            raise ValueError(
                f"run {self.run_manifest!r} is a pre-1.2 reference, so resolving it needs the "
                f"run-manifest store it points into — pass store_dir.")
        manifest = manifest_io.load_manifest(Path(store_dir) / self.run_manifest)
        actual = manifest_io.manifest_fingerprint(manifest)
        if actual != self.fingerprint and not allow_drift:
            raise ProtocolDriftError(
                f"run manifest {self.run_manifest!r} has drifted from protocol "
                f"{protocol_name!r}: pinned {self.fingerprint}, on disk {actual}. "
                f"Re-pin deliberately (a design change is a new protocol version), or pass "
                f"allow_drift=True to override.")
        return manifest, actual

    def source_status(self, store_dir) -> str:
        """Whether the file this run was embedded from still matches it — a *design-time* question.

        One of :data:`SOURCE_CURRENT` / :data:`SOURCE_DRIFTED` / :data:`SOURCE_MISSING` /
        :data:`SOURCE_UNKNOWN`. Never raises: an unreadable or invalid source file is ``missing``,
        because the answer this feeds is a report, and a report that throws is a report nobody sees.
        """
        if not self.run_manifest or store_dir is None:
            return SOURCE_UNKNOWN
        try:
            current = manifest_io.manifest_fingerprint(
                manifest_io.load_manifest(Path(store_dir) / self.run_manifest))
        except Exception:
            return SOURCE_MISSING
        mine = self.fingerprint or (manifest_io.manifest_fingerprint(self.manifest)
                                    if self.embedded else "")
        return SOURCE_CURRENT if current == mine else SOURCE_DRIFTED

pin classmethod 

pin(run_manifest, store_dir, label=None)

Embed the run manifest at store_dir / run_manifest, recording where it came from.

The manifest is loaded through :func:ssvep.io.manifest.load_manifest, so it is canonicalized to v1.3 shape and schema-validated here — an unloadable or invalid run manifest fails while someone is authoring, not while someone is recording.

Source code in src/ssvep/runtime/session.py 
189
190
191
192
193
194
195
196
197
198
199
@classmethod
def pin(cls, run_manifest: str, store_dir, label: str | None = None) -> "RunRef":
    """Embed the run manifest at ``store_dir / run_manifest``, recording where it came from.

    The manifest is loaded through :func:`ssvep.io.manifest.load_manifest`, so it is canonicalized
    to v1.3 shape and schema-validated here — an unloadable or invalid run manifest fails while
    someone is authoring, not while someone is recording.
    """
    m = manifest_io.load_manifest(Path(store_dir) / run_manifest)
    return cls(run_manifest=run_manifest, fingerprint=manifest_io.manifest_fingerprint(m),
               label=label, manifest=m)

resolve_manifest 

resolve_manifest(store_dir, *, allow_drift=False, protocol_name='')

This run's manifest and its fingerprint — embedded copy, or the legacy store lookup.

Embedded (v1.2+): returned as-is. There is nothing to verify, because there is nothing else it could have come from; the protocol is the design.

Legacy (v1.0/v1.1): loaded from store_dir and re-fingerprinted, raising :class:ProtocolDriftError on a mismatch unless allow_drift.

Source code in src/ssvep/runtime/session.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def resolve_manifest(self, store_dir, *, allow_drift: bool = False,
                     protocol_name: str = "") -> tuple[dict, str]:
    """This run's manifest and its fingerprint — embedded copy, or the legacy store lookup.

    Embedded (v1.2+): returned as-is. There is nothing to verify, because there is nothing else
    it could have come from; the protocol *is* the design.

    Legacy (v1.0/v1.1): loaded from ``store_dir`` and re-fingerprinted, raising
    :class:`ProtocolDriftError` on a mismatch unless ``allow_drift``.
    """
    if self.embedded:
        return self.manifest, (self.fingerprint or manifest_io.manifest_fingerprint(self.manifest))
    if store_dir is None:
        raise ValueError(
            f"run {self.run_manifest!r} is a pre-1.2 reference, so resolving it needs the "
            f"run-manifest store it points into — pass store_dir.")
    manifest = manifest_io.load_manifest(Path(store_dir) / self.run_manifest)
    actual = manifest_io.manifest_fingerprint(manifest)
    if actual != self.fingerprint and not allow_drift:
        raise ProtocolDriftError(
            f"run manifest {self.run_manifest!r} has drifted from protocol "
            f"{protocol_name!r}: pinned {self.fingerprint}, on disk {actual}. "
            f"Re-pin deliberately (a design change is a new protocol version), or pass "
            f"allow_drift=True to override.")
    return manifest, actual

source_status 

source_status(store_dir)

Whether the file this run was embedded from still matches it — a design-time question.

One of :data:SOURCE_CURRENT / :data:SOURCE_DRIFTED / :data:SOURCE_MISSING / :data:SOURCE_UNKNOWN. Never raises: an unreadable or invalid source file is missing, because the answer this feeds is a report, and a report that throws is a report nobody sees.

Source code in src/ssvep/runtime/session.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
def source_status(self, store_dir) -> str:
    """Whether the file this run was embedded from still matches it — a *design-time* question.

    One of :data:`SOURCE_CURRENT` / :data:`SOURCE_DRIFTED` / :data:`SOURCE_MISSING` /
    :data:`SOURCE_UNKNOWN`. Never raises: an unreadable or invalid source file is ``missing``,
    because the answer this feeds is a report, and a report that throws is a report nobody sees.
    """
    if not self.run_manifest or store_dir is None:
        return SOURCE_UNKNOWN
    try:
        current = manifest_io.manifest_fingerprint(
            manifest_io.load_manifest(Path(store_dir) / self.run_manifest))
    except Exception:
        return SOURCE_MISSING
    mine = self.fingerprint or (manifest_io.manifest_fingerprint(self.manifest)
                                if self.embedded else "")
    return SOURCE_CURRENT if current == mine else SOURCE_DRIFTED

SessionProtocol dataclass

Source code in src/ssvep/runtime/session.py 
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
@dataclass
class SessionProtocol:
    name: str
    groups: list[Group] = field(default_factory=list)
    version: str | None = None         # author's design version (bump deliberately on a design change)
    description: str | None = None
    seed: int = 0                      # base seed for randomized groups
    protocol_version: str = protocol_io.PROTOCOL_VERSION   # schema version
    end_questionnaire: Q.Questionnaire | None = None       # administered once, at the very end (#20)

    def has_questionnaires(self) -> bool:
        return self.end_questionnaire is not None or any(g.questionnaire for g in self.groups)

    def has_embedded_runs(self) -> bool:
        """True once any run carries its manifest inline — i.e. this file needs a 1.2 reader (#115)."""
        return any(r.embedded for g in self.groups for r in g.runs)

    def source_report(self, store_dir) -> list[tuple[Group, RunRef, str]]:
        """Every run paired with whether its source file still matches it (:meth:`RunRef.source_status`).

        The design-time replacement for the old resolve-time drift stop: authoring can ask "what have
        I fallen behind on?" and act on it, instead of finding out when a session refuses to start.
        """
        return [(g, r, r.source_status(store_dir)) for g in self.groups for r in g.runs]

    def duplicate_run_labels(self, store_dir=None) -> list[dict]:
        """Runs in this protocol that share a ``run.name`` or ``run.description`` with another run.

        A protocol's runs are meant to be distinct designs, so two sharing either field is either a
        genuine duplicate (the same run embedded twice) or a copy-paste that was never updated for
        the run it was pasted into — exactly what happened in the 2026-08-28 pilot, where six runs
        at three different frequency bands (31-39 Hz, 46-54 Hz) all carried the stale description
        "9-class high-frequency SSVEP, 36-44 Hz operating band" from the run they were cloned from.
        Neither is something Save should wave through unremarked.

        Returns one entry per colliding ``(field, value)`` — ``{"field": "name"|"description",
        "value": ..., "runs": [(group_label, run_label), ...]}`` — worst (most runs sharing a
        value) first. A run whose manifest can't be resolved (a legacy reference with no store, or
        a source file that has moved) is skipped rather than raising: this is a design-time nicety
        layered on top of :meth:`source_report`, not a replacement for it.
        """
        by_field: dict[str, dict[str, list[tuple[str, str]]]] = {"name": {}, "description": {}}
        for g in self.groups:
            for ref in g.runs:
                try:
                    manifest, _ = ref.resolve_manifest(store_dir)
                except Exception:
                    continue
                run = manifest_io.run_meta(manifest)
                label = ref.label or run.get("name") or ref.run_manifest or "(unnamed)"
                for field_name in ("name", "description"):
                    value = run.get(field_name)
                    if value:
                        by_field[field_name].setdefault(value, []).append((g.label, label))
        out = [{"field": field_name, "value": value, "runs": runs}
               for field_name, table in by_field.items()
               for value, runs in table.items() if len(runs) > 1]
        out.sort(key=lambda d: -len(d["runs"]))
        return out

    def schema_version(self) -> str:
        """The schema version this protocol **is**, not the one it was loaded as.

        Derived, not stored: a protocol carrying questionnaire blocks uses 1.1 features and says so,
        while one without them round-trips byte-identically as whatever it already was. The stamp
        then means "this file needs a 1.1 reader", never "a 1.1 build opened it" — so loading a pilot
        protocol to look at it does not rewrite its version.

        Everything that reports a schema version must come through here. It did not, once: the file
        said 1.1 and every recording made from it said 1.0, because ``to_dict`` derived the value and
        the sidecar stamp read the stale ``protocol_version`` attribute off an in-memory object that
        had been loaded as 1.0 and had questionnaires added since (caught on the sub-000 hardware
        check, 2026-08-01). One protocol, two answers, depending on which artifact you read.

        Highest feature in use wins: embedded runs (1.2) outrank questionnaires (1.1), since a 1.2
        reader necessarily understands 1.1.
        """
        if self.has_embedded_runs():
            return protocol_io.PROTOCOL_VERSION_EMBEDDED_RUNS
        if self.has_questionnaires():
            return protocol_io.PROTOCOL_VERSION_QUESTIONNAIRES
        return self.protocol_version

    def to_dict(self) -> dict:
        d: dict = {"protocol_version": self.schema_version(), "name": self.name}
        if self.version:
            d["version"] = self.version
        if self.description:
            d["description"] = self.description
        d["seed"] = self.seed
        if self.end_questionnaire is not None:
            d["end_questionnaire"] = self.end_questionnaire.to_dict()
        d["groups"] = [g.to_dict() for g in self.groups]
        return d

    @classmethod
    def from_dict(cls, d: dict) -> "SessionProtocol":
        return cls(
            name=d["name"],
            groups=[Group.from_dict(g) for g in d["groups"]],
            version=d.get("version"),
            description=d.get("description"),
            seed=int(d.get("seed", 0)),
            protocol_version=d.get("protocol_version", protocol_io.PROTOCOL_VERSION),
            end_questionnaire=(Q.Questionnaire.from_dict(d["end_questionnaire"])
                               if d.get("end_questionnaire") else None),
        )

    def validate(self) -> None:
        protocol_io.validate_protocol(self.to_dict())

    def save(self, path) -> Path:
        return protocol_io.save_protocol(self.to_dict(), path)

    @classmethod
    def load(cls, path) -> "SessionProtocol":
        return cls.from_dict(protocol_io.load_protocol(path))

    def fingerprint(self) -> str:
        """Semantic fingerprint of the protocol itself — its identity in each recording's sidecar."""
        return manifest_io.manifest_fingerprint(self.to_dict())

has_embedded_runs 

has_embedded_runs()

True once any run carries its manifest inline — i.e. this file needs a 1.2 reader (#115).

Source code in src/ssvep/runtime/session.py 
280
281
282
def has_embedded_runs(self) -> bool:
    """True once any run carries its manifest inline — i.e. this file needs a 1.2 reader (#115)."""
    return any(r.embedded for g in self.groups for r in g.runs)

source_report 

source_report(store_dir)

Every run paired with whether its source file still matches it (:meth:RunRef.source_status).

The design-time replacement for the old resolve-time drift stop: authoring can ask "what have I fallen behind on?" and act on it, instead of finding out when a session refuses to start.

Source code in src/ssvep/runtime/session.py 
284
285
286
287
288
289
290
def source_report(self, store_dir) -> list[tuple[Group, RunRef, str]]:
    """Every run paired with whether its source file still matches it (:meth:`RunRef.source_status`).

    The design-time replacement for the old resolve-time drift stop: authoring can ask "what have
    I fallen behind on?" and act on it, instead of finding out when a session refuses to start.
    """
    return [(g, r, r.source_status(store_dir)) for g in self.groups for r in g.runs]

duplicate_run_labels 

duplicate_run_labels(store_dir=None)

Runs in this protocol that share a run.name or run.description with another run.

A protocol's runs are meant to be distinct designs, so two sharing either field is either a genuine duplicate (the same run embedded twice) or a copy-paste that was never updated for the run it was pasted into — exactly what happened in the 2026-08-28 pilot, where six runs at three different frequency bands (31-39 Hz, 46-54 Hz) all carried the stale description "9-class high-frequency SSVEP, 36-44 Hz operating band" from the run they were cloned from. Neither is something Save should wave through unremarked.

Returns one entry per colliding (field, value) — {"field": "name"|"description", "value": ..., "runs": [(group_label, run_label), ...]} — worst (most runs sharing a value) first. A run whose manifest can't be resolved (a legacy reference with no store, or a source file that has moved) is skipped rather than raising: this is a design-time nicety layered on top of :meth:source_report, not a replacement for it.

Source code in src/ssvep/runtime/session.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
def duplicate_run_labels(self, store_dir=None) -> list[dict]:
    """Runs in this protocol that share a ``run.name`` or ``run.description`` with another run.

    A protocol's runs are meant to be distinct designs, so two sharing either field is either a
    genuine duplicate (the same run embedded twice) or a copy-paste that was never updated for
    the run it was pasted into — exactly what happened in the 2026-08-28 pilot, where six runs
    at three different frequency bands (31-39 Hz, 46-54 Hz) all carried the stale description
    "9-class high-frequency SSVEP, 36-44 Hz operating band" from the run they were cloned from.
    Neither is something Save should wave through unremarked.

    Returns one entry per colliding ``(field, value)`` — ``{"field": "name"|"description",
    "value": ..., "runs": [(group_label, run_label), ...]}`` — worst (most runs sharing a
    value) first. A run whose manifest can't be resolved (a legacy reference with no store, or
    a source file that has moved) is skipped rather than raising: this is a design-time nicety
    layered on top of :meth:`source_report`, not a replacement for it.
    """
    by_field: dict[str, dict[str, list[tuple[str, str]]]] = {"name": {}, "description": {}}
    for g in self.groups:
        for ref in g.runs:
            try:
                manifest, _ = ref.resolve_manifest(store_dir)
            except Exception:
                continue
            run = manifest_io.run_meta(manifest)
            label = ref.label or run.get("name") or ref.run_manifest or "(unnamed)"
            for field_name in ("name", "description"):
                value = run.get(field_name)
                if value:
                    by_field[field_name].setdefault(value, []).append((g.label, label))
    out = [{"field": field_name, "value": value, "runs": runs}
           for field_name, table in by_field.items()
           for value, runs in table.items() if len(runs) > 1]
    out.sort(key=lambda d: -len(d["runs"]))
    return out

schema_version 

schema_version()

The schema version this protocol is, not the one it was loaded as.

Derived, not stored: a protocol carrying questionnaire blocks uses 1.1 features and says so, while one without them round-trips byte-identically as whatever it already was. The stamp then means "this file needs a 1.1 reader", never "a 1.1 build opened it" — so loading a pilot protocol to look at it does not rewrite its version.

Everything that reports a schema version must come through here. It did not, once: the file said 1.1 and every recording made from it said 1.0, because to_dict derived the value and the sidecar stamp read the stale protocol_version attribute off an in-memory object that had been loaded as 1.0 and had questionnaires added since (caught on the sub-000 hardware check, 2026-08-01). One protocol, two answers, depending on which artifact you read.

Highest feature in use wins: embedded runs (1.2) outrank questionnaires (1.1), since a 1.2 reader necessarily understands 1.1.

Source code in src/ssvep/runtime/session.py 
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
def schema_version(self) -> str:
    """The schema version this protocol **is**, not the one it was loaded as.

    Derived, not stored: a protocol carrying questionnaire blocks uses 1.1 features and says so,
    while one without them round-trips byte-identically as whatever it already was. The stamp
    then means "this file needs a 1.1 reader", never "a 1.1 build opened it" — so loading a pilot
    protocol to look at it does not rewrite its version.

    Everything that reports a schema version must come through here. It did not, once: the file
    said 1.1 and every recording made from it said 1.0, because ``to_dict`` derived the value and
    the sidecar stamp read the stale ``protocol_version`` attribute off an in-memory object that
    had been loaded as 1.0 and had questionnaires added since (caught on the sub-000 hardware
    check, 2026-08-01). One protocol, two answers, depending on which artifact you read.

    Highest feature in use wins: embedded runs (1.2) outrank questionnaires (1.1), since a 1.2
    reader necessarily understands 1.1.
    """
    if self.has_embedded_runs():
        return protocol_io.PROTOCOL_VERSION_EMBEDDED_RUNS
    if self.has_questionnaires():
        return protocol_io.PROTOCOL_VERSION_QUESTIONNAIRES
    return self.protocol_version

fingerprint 

fingerprint()

Semantic fingerprint of the protocol itself — its identity in each recording's sidecar.

Source code in src/ssvep/runtime/session.py 
385
386
387
def fingerprint(self) -> str:
    """Semantic fingerprint of the protocol itself — its identity in each recording's sidecar."""
    return manifest_io.manifest_fingerprint(self.to_dict())

ResolvedCheckpoint dataclass

One questionnaire administration in a resolved session (#20).

A checkpoint sits between runs — after a group's last run, or after the whole session — so nothing is streaming while it happens and there are no markers to write, only a timestamp. It is resolved rather than improvised for the same reason a run's run_index is: where in the session a participant was asked is part of the record, not something to reconstruct from file times.

Source code in src/ssvep/runtime/session.py 
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
@dataclass
class ResolvedCheckpoint:
    """One questionnaire administration in a resolved session (#20).

    A checkpoint sits **between** runs — after a group's last run, or after the whole session — so
    nothing is streaming while it happens and there are no markers to write, only a timestamp. It is
    resolved rather than improvised for the same reason a run's ``run_index`` is: where in the session
    a participant was asked is part of the record, not something to reconstruct from file times.
    """
    questionnaire: Q.Questionnaire
    group: str | None                  # the group it follows; None for the end-of-session instrument
    after_run_index: int               # the run_index it comes after (0 ⇒ before any run)
    index: int | None                  # administration ordinal → BIDS run-NNN; None if once per session
    position: dict                     # stamped into the response sidecar, like a run's

    @property
    def task(self) -> str:
        return self.questionnaire.task

    @property
    def label(self) -> str:
        return self.questionnaire.id

ResolvedSession dataclass

Source code in src/ssvep/runtime/session.py 
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
@dataclass
class ResolvedSession:
    protocol_name: str
    protocol_version: str | None
    protocol_fingerprint: str
    subject: str
    session: str
    runs: list[ResolvedRun] = field(default_factory=list)
    checkpoints: list[ResolvedCheckpoint] = field(default_factory=list)

    def __len__(self) -> int:
        return len(self.runs)

    def __iter__(self):
        return iter(self.runs)

    def sequence(self) -> list[ResolvedRun | ResolvedCheckpoint]:
        """Runs and checkpoints interleaved into the literal running order of the session.

        A checkpoint lands immediately after the run whose index it follows, so this is what the
        operator's checklist shows and what "the next thing to do" means.
        """
        out: list[ResolvedRun | ResolvedCheckpoint] = []
        for r in self.runs:
            out.append(r)
            out.extend(c for c in self.checkpoints if c.after_run_index == r.run_index)
        # A checkpoint on an empty group (or a protocol with no runs at all) has nothing to follow;
        # keep it rather than dropping it silently — a missing item is how a checkpoint gets skipped.
        placed = {id(c) for c in out if isinstance(c, ResolvedCheckpoint)}
        out.extend(c for c in self.checkpoints if id(c) not in placed)
        return out

sequence 

sequence()

Runs and checkpoints interleaved into the literal running order of the session.

A checkpoint lands immediately after the run whose index it follows, so this is what the operator's checklist shows and what "the next thing to do" means.

Source code in src/ssvep/runtime/session.py 
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
def sequence(self) -> list[ResolvedRun | ResolvedCheckpoint]:
    """Runs and checkpoints interleaved into the literal running order of the session.

    A checkpoint lands immediately after the run whose index it follows, so this is what the
    operator's checklist shows and what "the next thing to do" means.
    """
    out: list[ResolvedRun | ResolvedCheckpoint] = []
    for r in self.runs:
        out.append(r)
        out.extend(c for c in self.checkpoints if c.after_run_index == r.run_index)
    # A checkpoint on an empty group (or a protocol with no runs at all) has nothing to follow;
    # keep it rather than dropping it silently — a missing item is how a checkpoint gets skipped.
    placed = {id(c) for c in out if isinstance(c, ResolvedCheckpoint)}
    out.extend(c for c in self.checkpoints if id(c) not in placed)
    return out

RunStatus dataclass

Source code in src/ssvep/runtime/session.py 
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
@dataclass
class RunStatus:
    run_index: int
    label: str
    group: str
    state: str                         # pending | partial | done (see session_progress)
    sidecar: Path
    trials_recorded: int | None = None  # from the sidecar; None if it predates the field
    trials_planned: int | None = None

    @property
    def done(self) -> bool:
        """Recorded **and complete**. A partial run is not done — re-recording it is still owed."""
        return self.state == STATE_DONE

    @property
    def recorded(self) -> bool:
        """Something is on disk for this run (complete or not) — i.e. re-recording would overwrite."""
        return self.state in (STATE_DONE, STATE_PARTIAL)

done property 

done

Recorded and complete. A partial run is not done — re-recording it is still owed.

recorded property 

recorded

Something is on disk for this run (complete or not) — i.e. re-recording would overwrite.

ItemStatus dataclass

One row of the session checklist — a run or a checkpoint, with its state derived from disk.

Source code in src/ssvep/runtime/session.py 
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
@dataclass
class ItemStatus:
    """One row of the session checklist — a run or a checkpoint, with its state derived from disk."""
    kind: str                          # ITEM_RUN | ITEM_CHECKPOINT
    label: str
    group: str
    state: str                         # pending | partial | done | skipped
    path: Path                         # the file whose existence decides the state
    run_index: int | None = None       # runs only
    trials_recorded: int | None = None
    trials_planned: int | None = None
    run: ResolvedRun | None = None
    checkpoint: ResolvedCheckpoint | None = None

    @property
    def done(self) -> bool:
        return self.state == STATE_DONE

williams_sequences 

williams_sequences(n)

Balanced Latin square (Williams design) over n items, as index orderings of range(n).

Each item is immediately preceded by every other item equally often (first-order carryover balanced) — the right within-subject counterbalance where fatigue/adaptation carry over between SSVEP runs. Even n needs n sequences; odd n needs 2n (the square plus its row reverses). n<=1 ⇒ the single trivial ordering.

Source code in src/ssvep/runtime/session.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def williams_sequences(n: int) -> list[list[int]]:
    """Balanced Latin square (Williams design) over ``n`` items, as index orderings of ``range(n)``.

    Each item is immediately preceded by every other item equally often (first-order carryover
    balanced) — the right within-subject counterbalance where fatigue/adaptation carry over between
    SSVEP runs. Even ``n`` needs ``n`` sequences; odd ``n`` needs ``2n`` (the square plus its row
    reverses). ``n<=1`` ⇒ the single trivial ordering.
    """
    if n <= 1:
        return [list(range(max(n, 0)))]
    # First row: 0, 1, n-1, 2, n-2, 3, … (the standard Williams generator).
    first = []
    for j in range(n):
        if j == 0:
            first.append(0)
        elif j % 2 == 1:
            first.append((j + 1) // 2)
        else:
            first.append(n - j // 2)
    rows = [[(first[j] + i) % n for j in range(n)] for i in range(n)]
    if n % 2 == 1:                                   # odd n: add reverses to balance carryover
        rows = rows + [list(reversed(r)) for r in rows]
    return rows

order_indices 

order_indices(order, n, *, subject, seed, label)

This participant's ordering of n runs in a group, plus the counterbalance arm (or None).

Pure and deterministic in (order, n, subject, seed, label). fixed keeps authored order; counterbalanced picks a Williams sequence by subject_number mod #sequences; randomized shuffles with a per-group seed.

Source code in src/ssvep/runtime/session.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def order_indices(order: str, n: int, *, subject: str, seed: int, label: str) -> tuple[list[int], int | None]:
    """This participant's ordering of ``n`` runs in a group, plus the counterbalance arm (or ``None``).

    Pure and deterministic in ``(order, n, subject, seed, label)``. ``fixed`` keeps authored order;
    ``counterbalanced`` picks a Williams sequence by ``subject_number mod #sequences``; ``randomized``
    shuffles with a per-group seed.
    """
    if order == ORDER_FIXED:
        return list(range(n)), None
    if order == ORDER_COUNTERBALANCED:
        seqs = williams_sequences(n)
        arm = _subject_number(subject) % len(seqs)
        return list(seqs[arm]), arm
    if order == ORDER_RANDOMIZED:
        idx = list(range(n))
        _rng(seed, subject, label).shuffle(idx)
        return idx, None
    raise ValueError(f"unknown order {order!r}; expected one of {ORDERS}")

resolve_session 

resolve_session(protocol, subject, session, store_dir=None, *, allow_drift=False)

Instantiate protocol for one participant's session into an ordered, frozen run list.

Inputs are the template, the participant sub-XXX, and the visit ses-YYY. Groups run in authored order; within each group the runs are reordered per its scheme (derived from the participant), then flattened and stamped with run_index 1..N.

store_dir is needed only for a pre-1.2 protocol, whose runs are paths relative to the run-manifest store: those are loaded and re-fingerprinted, raising :class:ProtocolDriftError on a mismatch unless allow_drift. A v1.2 protocol carries its manifests, so store_dir is unused and may be omitted — the whole resolution is then pure in-memory, which is the point: no file outside the protocol can change what a participant runs.

The resolver never scans sourcedata and never assigns ses-YYY — that is the caller's, kept out so this stays a pure function of its inputs.

Source code in src/ssvep/runtime/session.py 
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
def resolve_session(protocol: SessionProtocol, subject: str, session: str, store_dir=None, *,
                    allow_drift: bool = False) -> ResolvedSession:
    """Instantiate ``protocol`` for one participant's session into an ordered, frozen run list.

    Inputs are the template, the participant ``sub-XXX``, and the visit ``ses-YYY``. Groups run in
    authored order; within each group the runs are reordered per its scheme (derived from the
    participant), then flattened and stamped with ``run_index`` 1..N.

    ``store_dir`` is needed **only** for a pre-1.2 protocol, whose runs are paths relative to the
    run-manifest store: those are loaded and re-fingerprinted, raising :class:`ProtocolDriftError` on
    a mismatch unless ``allow_drift``. A v1.2 protocol carries its manifests, so ``store_dir`` is
    unused and may be omitted — the whole resolution is then pure in-memory, which is the point: no
    file outside the protocol can change what a participant runs.

    The resolver never scans ``sourcedata`` and never assigns ``ses-YYY`` — that is the caller's, kept
    out so this stays a pure function of its inputs.
    """
    if not _SUBJECT_RE.match(subject or ""):
        raise ValueError(f"subject must be a de-identified code like 'sub-905', got {subject!r}")
    store = Path(store_dir) if store_dir is not None else None
    pfp = protocol.fingerprint()
    runs: list[ResolvedRun] = []
    checkpoints: list[ResolvedCheckpoint] = []
    run_index = 0
    for group in protocol.groups:
        n = len(group.runs)
        idx_order, arm = order_indices(group.order, n, subject=subject, seed=protocol.seed,
                                       label=group.label)
        for pos in idx_order:
            ref = group.runs[pos]
            manifest, actual = ref.resolve_manifest(store, allow_drift=allow_drift,
                                                    protocol_name=protocol.name)
            run_index += 1
            label = ref.label or manifest_io.run_meta(manifest).get("name", ref.run_manifest)
            position = {
                "protocol": protocol.name,
                "protocol_version": protocol.version,
                "protocol_schema_version": protocol.schema_version(),
                "protocol_fingerprint": pfp,
                "subject": subject,
                "session": session,
                "group": group.label,
                "order": group.order,
                "run_index": run_index,
                "arm": arm,
                "seed": protocol.seed,
                "label": label,
            }
            runs.append(ResolvedRun(
                manifest=manifest, run_index=run_index, group=group.label, order=group.order,
                run_manifest=ref.run_manifest, fingerprint=actual, label=label, arm=arm,
                position=position))
        if group.questionnaire is not None:
            checkpoints.append(_checkpoint(
                group.questionnaire, protocol, subject, session, pfp,
                group=group.label, after_run_index=run_index,
                # BIDS run-NNN for a repeated instrument = which administration this is. Counted over
                # administrations of THIS task, so a protocol mixing instruments numbers each cleanly.
                index=1 + sum(c.task == group.questionnaire.task for c in checkpoints)))
    if protocol.end_questionnaire is not None:
        # Once per session ⇒ no BIDS run entity. Unless a group happens to use the same task, in
        # which case it needs one or the two would write the same filename.
        same_task = sum(c.task == protocol.end_questionnaire.task for c in checkpoints)
        checkpoints.append(_checkpoint(
            protocol.end_questionnaire, protocol, subject, session, pfp,
            group=None, after_run_index=run_index,
            index=(same_task + 1) if same_task else None))
    return ResolvedSession(
        protocol_name=protocol.name, protocol_version=protocol.version, protocol_fingerprint=pfp,
        subject=subject, session=session, runs=runs, checkpoints=checkpoints)

session_progress 

session_progress(resolved, out_dir)

Which runs of a resolved session are already recorded — derived from disk, no progress file.

Three states, because two were not enough. A run is pending when its *.session.json sidecar is absent, so a crash before saving is correctly offered again. It is done when the sidecar shows every planned trial was recorded. In between it is partial: the sidecar exists, but the run was aborted or lost its stream part-way.

That middle state is the sub-002 lesson. An 11-second abort still saves (deliberately — partial data is data), and when "the sidecar exists" was the whole test, the checklist called that run recorded and moved on. The operator, who could see perfectly well that it had not been, worked around the checklist by hand-numbering the next run — which took it out of the protocol entirely and cost it its fingerprint pin. Showing partial is what removes the reason to improvise.

The recordings remain the state; there is still no second artifact to desync (the ses-002 lesson, §CLAUDE.md) — completeness is read out of the sidecars themselves.

(An observer-only session persists nothing, so every run reads pending here — the record UI runs those forward-only rather than resuming, since there is nothing on disk to resume from.)

Source code in src/ssvep/runtime/session.py 
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
def session_progress(resolved: ResolvedSession, out_dir) -> list[RunStatus]:
    """Which runs of a resolved session are already recorded — **derived from disk**, no progress file.

    Three states, because two were not enough. A run is **pending** when its ``*.session.json``
    sidecar is absent, so a crash *before* saving is correctly offered again. It is **done** when the
    sidecar shows every planned trial was recorded. In between it is **partial**: the sidecar exists,
    but the run was aborted or lost its stream part-way.

    That middle state is the sub-002 lesson. An 11-second abort still saves (deliberately — partial
    data is data), and when "the sidecar exists" was the whole test, the checklist called that run
    recorded and moved on. The operator, who could see perfectly well that it had not been, worked
    around the checklist by hand-numbering the next run — which took it out of the protocol entirely
    and cost it its fingerprint pin. Showing partial is what removes the reason to improvise.

    The recordings remain the state; there is still no second artifact to desync (the ses-002 lesson,
    §CLAUDE.md) — completeness is read out of the sidecars themselves.

    (An observer-only session persists nothing, so every run reads pending here — the record UI runs
    those forward-only rather than resuming, since there is nothing on disk to resume from.)
    """
    from .runner import run_output_base
    out: list[RunStatus] = []
    for r in resolved.runs:
        task = manifest_io.run_meta(r.manifest).get("task", "ssvep")
        base = run_output_base(out_dir, resolved.subject, resolved.session, task, r.run_index)
        sidecar = base.parent / (base.name + ".session.json")
        state, got, planned = _run_state(sidecar)
        out.append(RunStatus(run_index=r.run_index, label=r.label, group=r.group,
                             state=state, sidecar=sidecar,
                             trials_recorded=got, trials_planned=planned))
    return out

next_pending 

next_pending(progress)

The first unrecorded run (lowest run_index with no sidecar), or None if none is left.

Deliberately skips partial runs rather than re-offering them: something is on disk for those, and re-recording would overwrite it. They are shown as partial in the checklist and re-run through the confirmed path, so replacing collected data stays a decision rather than a default.

Source code in src/ssvep/runtime/session.py 
640
641
642
643
644
645
646
647
def next_pending(progress: list[RunStatus]) -> RunStatus | None:
    """The first **unrecorded** run (lowest run_index with no sidecar), or None if none is left.

    Deliberately skips *partial* runs rather than re-offering them: something is on disk for those,
    and re-recording would overwrite it. They are shown as partial in the checklist and re-run through
    the confirmed path, so replacing collected data stays a decision rather than a default.
    """
    return next((s for s in progress if s.state == STATE_PENDING), None)

questionnaire_mode 

questionnaire_mode(protocol, *, enabled)

Which stamp this session's runs carry. Declared-and-off is not the same as never-declared.

Absence of a response file has to mean one thing for the checklist to resume correctly ("not asked yet"), so the reason there will never be one gets written down instead of inferred. A rig test with no participant and a session whose responses were lost must not read the same way later — that ambiguity is the ses-002 lesson, one artifact over.

Source code in src/ssvep/runtime/session.py 
656
657
658
659
660
661
662
663
664
665
666
def questionnaire_mode(protocol: SessionProtocol | None, *, enabled: bool) -> str:
    """Which stamp this session's runs carry. Declared-and-off is *not* the same as never-declared.

    Absence of a response file has to mean one thing for the checklist to resume correctly ("not
    asked yet"), so the reason there will never be one gets written down instead of inferred. A rig
    test with no participant and a session whose responses were lost must not read the same way later
    — that ambiguity is the ses-002 lesson, one artifact over.
    """
    if protocol is None or not protocol.has_questionnaires():
        return Q_NONE
    return Q_ADMINISTERED if enabled else Q_SKIPPED

recorder_for 

recorder_for(run, resolved, *, operator, consent, out_dir, device=None, notes=None, seed=0, questionnaires=None, acquisition_override=None)

Build a :class:RunRecorder for one resolved run — the only place resolver meets recorder.

run_index, session, and the position block all come from the resolved run/session, so they are never hand-set at record time (the whole point of #43). Consent is passed per session by the caller (a visit-level property). Returns an un-run recorder; the record path drives it.

questionnaires (see :func:questionnaire_mode) rides in the position block rather than in the protocol, because whether they were administered is a fact about this session, decided at the bench — not part of the design. The resolver stays pure and unaware of it.

acquisition_override is the same kind of fact: what headset / electrode type / skin prep physically ran, which the protocol cannot know because electrodes get swapped at the bench. It was threaded through the ad-hoc record path only, so until #115 a run recorded from a protocol — the path the lab actually uses — silently dropped it, which is the sub-902 wet/dry metadata gap reappearing one level up. Removing the ad-hoc path made fixing this compulsory.

Source code in src/ssvep/runtime/session.py 
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
def recorder_for(run: ResolvedRun, resolved: ResolvedSession, *, operator, consent, out_dir,
                 device: dict | None = None, notes: str | None = None, seed: int | None = 0,
                 questionnaires: str | None = None, acquisition_override: dict | None = None):
    """Build a :class:`RunRecorder` for one resolved run — the only place resolver meets recorder.

    ``run_index``, ``session``, and the ``position`` block all come from the resolved run/session, so
    they are never hand-set at record time (the whole point of #43). Consent is passed per session by
    the caller (a visit-level property). Returns an un-run recorder; the record path drives it.

    ``questionnaires`` (see :func:`questionnaire_mode`) rides in the position block rather than in the
    protocol, because whether they were administered is a fact about *this session*, decided at the
    bench — not part of the design. The resolver stays pure and unaware of it.

    ``acquisition_override`` is the same kind of fact: what headset / electrode type / skin prep
    **physically** ran, which the protocol cannot know because electrodes get swapped at the bench.
    It was threaded through the ad-hoc record path only, so until #115 a run recorded *from a
    protocol* — the path the lab actually uses — silently dropped it, which is the sub-902 wet/dry
    metadata gap reappearing one level up. Removing the ad-hoc path made fixing this compulsory.
    """
    from .runner import RunRecorder
    position = dict(run.position)
    if questionnaires:
        position["questionnaires"] = questionnaires
    return RunRecorder(
        manifest=run.manifest, operator=operator, consent=consent, out_dir=str(out_dir),
        subject=resolved.subject, session=resolved.session, run_index=run.run_index,
        seed=seed, device=device, notes=notes, position=position,
        acquisition_override=acquisition_override)

session_checklist 

session_checklist(resolved, out_dir, *, questionnaires=True)

The whole session as an ordered to-do list — runs and checkpoints interleaved.

Completion is derived from disk for both kinds, by the same rule and for the same reason: a run is done when its *.session.json exists (and reports every planned trial), a checkpoint when its _beh.json exists. There is still no separate progress file to desync from the data.

questionnaires=False (the operator's "don't run questionnaires" switch — a rig test with no participant) marks checkpoints skipped rather than hiding them: they stay visible, so nobody has to wonder later whether the protocol had any.

Source code in src/ssvep/runtime/session.py 
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
def session_checklist(resolved: ResolvedSession, out_dir, *, questionnaires: bool = True
                      ) -> list[ItemStatus]:
    """The whole session as an ordered to-do list — runs and checkpoints interleaved.

    Completion is derived from disk for both kinds, by the same rule and for the same reason: a run
    is done when its ``*.session.json`` exists (and reports every planned trial), a checkpoint when
    its ``_beh.json`` exists. There is still no separate progress file to desync from the data.

    ``questionnaires=False`` (the operator's "don't run questionnaires" switch — a rig test with no
    participant) marks checkpoints **skipped** rather than hiding them: they stay visible, so nobody
    has to wonder later whether the protocol had any.
    """
    runs = {p.run_index: p for p in session_progress(resolved, out_dir)}
    out: list[ItemStatus] = []
    for item in resolved.sequence():
        if isinstance(item, ResolvedRun):
            p = runs[item.run_index]
            out.append(ItemStatus(kind=ITEM_RUN, label=p.label, group=p.group, state=p.state,
                                  path=p.sidecar, run_index=p.run_index,
                                  trials_recorded=p.trials_recorded,
                                  trials_planned=p.trials_planned, run=item))
        else:
            path = Q.response_path(out_dir, resolved.subject, resolved.session,
                                   item.task, item.index)
            if not questionnaires:
                state = STATE_SKIPPED
            else:
                state = STATE_DONE if path.is_file() else STATE_PENDING
            out.append(ItemStatus(kind=ITEM_CHECKPOINT, label=item.questionnaire.id,
                                  group=item.group or "—", state=state, path=path,
                                  checkpoint=item))
    return out

next_pending_item 

next_pending_item(checklist)

The next thing to do: the first item that is neither done, partial, nor skipped.

Partial runs are stepped over for the same reason :func:next_pending steps over them — something is on disk, so replacing it stays a deliberate re-run rather than the default.

Source code in src/ssvep/runtime/session.py 
760
761
762
763
764
765
766
def next_pending_item(checklist: list[ItemStatus]) -> ItemStatus | None:
    """The next thing to do: the first item that is neither done, partial, nor skipped.

    Partial runs are stepped over for the same reason :func:`next_pending` steps over them — something
    is on disk, so replacing it stays a deliberate re-run rather than the default.
    """
    return next((i for i in checklist if i.state == STATE_PENDING), None)

session_exists 

session_exists(out_dir, subject, session)

True if a recording session dir already exists — the GUI turns this into an overwrite warning.

Source code in src/ssvep/runtime/session.py 
779
780
781
def session_exists(out_dir, subject: str, session: str) -> bool:
    """True if a recording session dir already exists — the GUI turns this into an overwrite warning."""
    return _session_dir(out_dir, subject, session).is_dir()

next_session 

next_session(out_dir, subject)

Suggest the next ses-NNN for a participant by scanning existing session dirs (ses-001 if none).

A convenience for prefilling the operator's session field — deliberately separate from the pure resolver, which takes ses-YYY as an input rather than deriving it from disk.

Source code in src/ssvep/runtime/session.py 
784
785
786
787
788
789
790
791
792
793
def next_session(out_dir, subject: str) -> str:
    """Suggest the next ``ses-NNN`` for a participant by scanning existing session dirs (``ses-001`` if none).

    A convenience for prefilling the operator's session field — deliberately separate from the pure
    resolver, which takes ``ses-YYY`` as an input rather than deriving it from disk.
    """
    subj_dir = Path(out_dir) / "sourcedata" / subject
    nums = [int(m.group(1)) for p in (subj_dir.iterdir() if subj_dir.is_dir() else [])
            if (m := _SES_RE.match(p.name)) and p.is_dir()]
    return f"ses-{(max(nums) + 1) if nums else 1:03d}"

ssvep.io — manifests and data

Run manifest I/O

ssvep.io.manifest

Run-manifest I/O and schema validation.

The manifest (schemas/run_manifest.schema.json) is the toolbox's single source of truth: the builder emits it, acquisition stamps it into each recording, analysis reads it. This module loads the JSON Schemas and validates manifests/markers against them, so every producer (builder now; runtime later) can guarantee a conformant contract.

A manifest describes exactly one run (CLAUDE.md §2.1). Manifest v1.3 renamed the run-identity block experiment → run; :func:canonicalize_manifest maps the old key on read so every pre-1.3 protocol and recorded sidecar still loads, and :func:run_meta is the tolerant accessor for that block regardless of which key a given manifest carries.

canonicalize_manifest 

canonicalize_manifest(manifest)

Return manifest in the v1.3 shape, mapping a legacy experiment block to run.

Idempotent and non-mutating: if the manifest already has run (or has neither key) it is returned unchanged; otherwise a shallow copy is made with experiment moved to run. Every sidecar ever written embeds the manifest it recorded under (RunRecorder.sidecar), and the five saved protocols predate the rename, so this is the single point that keeps all of them loadable.

Source code in src/ssvep/io/manifest.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def canonicalize_manifest(manifest: dict) -> dict:
    """Return ``manifest`` in the v1.3 shape, mapping a legacy ``experiment`` block to ``run``.

    Idempotent and non-mutating: if the manifest already has ``run`` (or has neither key) it is
    returned unchanged; otherwise a shallow copy is made with ``experiment`` moved to ``run``. Every
    sidecar ever written embeds the manifest it recorded under (``RunRecorder.sidecar``), and the five
    saved protocols predate the rename, so this is the single point that keeps all of them loadable.
    """
    if not isinstance(manifest, dict):
        return manifest
    if _RUN_KEY in manifest or _LEGACY_RUN_KEY not in manifest:
        return manifest
    m = dict(manifest)
    m[_RUN_KEY] = m.pop(_LEGACY_RUN_KEY)
    return m

run_meta 

run_meta(manifest)

The run-identity block (name/task/description), tolerant of pre-1.3 manifests.

Reads run (v1.3+) or experiment (legacy). Use this everywhere a manifest — including one lifted out of an on-disk sidecar, which may carry either key — is read for its name or task.

Source code in src/ssvep/io/manifest.py 
56
57
58
59
60
61
62
63
def run_meta(manifest: dict) -> dict:
    """The run-identity block (``name``/``task``/``description``), tolerant of pre-1.3 manifests.

    Reads ``run`` (v1.3+) or ``experiment`` (legacy). Use this everywhere a manifest — including one
    lifted out of an on-disk sidecar, which may carry either key — is read for its name or task.
    """
    m = manifest or {}
    return m.get(_RUN_KEY) or m.get(_LEGACY_RUN_KEY) or {}

manifest_fingerprint 

manifest_fingerprint(manifest)

Short stable hash of a manifest's semantic content — the identity a session protocol pins.

Canonicalized first (§canonicalize_manifest), so the pre-1.3 experiment → run rename does NOT change a run manifest's fingerprint: a protocol authored against the legacy shape still resolves against the same file re-saved in v1.3 shape. Key order is irrelevant (sort_keys). Ties a recording / QC report / protocol pin to the exact design it was built against.

Source code in src/ssvep/io/manifest.py 
66
67
68
69
70
71
72
73
74
75
76
def manifest_fingerprint(manifest: dict) -> str:
    """Short stable hash of a manifest's *semantic* content — the identity a session protocol pins.

    Canonicalized first (§canonicalize_manifest), so the pre-1.3 ``experiment`` → ``run`` rename does
    NOT change a run manifest's fingerprint: a protocol authored against the legacy shape still
    resolves against the same file re-saved in v1.3 shape. Key order is irrelevant (``sort_keys``).
    Ties a recording / QC report / protocol pin to the exact design it was built against.
    """
    blob = json.dumps(canonicalize_manifest(manifest), sort_keys=True,
                      separators=(",", ":")).encode("utf-8")
    return hashlib.sha256(blob).hexdigest()[:12]

paradigm 

paradigm(manifest)

The manifest's paradigm: 'ssvep' (flickering targets) or 'resting' (eyes-open/closed).

Absent ⇒ 'ssvep', so every pre-1.2 manifest keeps reading correctly.

Source code in src/ssvep/io/manifest.py 
79
80
81
82
83
84
def paradigm(manifest: dict) -> str:
    """The manifest's paradigm: ``'ssvep'`` (flickering targets) or ``'resting'`` (eyes-open/closed).

    Absent ⇒ ``'ssvep'``, so every pre-1.2 manifest keeps reading correctly.
    """
    return (manifest or {}).get("paradigm") or SSVEP

is_resting 

is_resting(manifest)

True if manifest describes a tone-cued eyes-open/eyes-closed resting run.

Source code in src/ssvep/io/manifest.py 
87
88
89
def is_resting(manifest: dict) -> bool:
    """True if ``manifest`` describes a tone-cued eyes-open/eyes-closed resting run."""
    return paradigm(manifest) == RESTING

resting_params 

resting_params(manifest)

design.resting for a resting manifest (the authoritative block structure); {} if absent.

Source code in src/ssvep/io/manifest.py 
92
93
94
def resting_params(manifest: dict) -> dict:
    """``design.resting`` for a resting manifest (the authoritative block structure); ``{}`` if absent."""
    return ((manifest or {}).get("design", {}) or {}).get("resting", {}) or {}

validate_manifest 

validate_manifest(manifest)

Raise jsonschema.ValidationError if manifest violates the schema; else return.

Canonicalized first, so a legacy experiment-keyed manifest validates against the v1.3 schema.

Source code in src/ssvep/io/manifest.py 
111
112
113
114
115
116
def validate_manifest(manifest: dict) -> None:
    """Raise ``jsonschema.ValidationError`` if ``manifest`` violates the schema; else return.

    Canonicalized first, so a legacy ``experiment``-keyed manifest validates against the v1.3 schema.
    """
    jsonschema.validate(instance=canonicalize_manifest(manifest), schema=manifest_schema())

load_manifest 

load_manifest(path)

Load, canonicalize, and validate a manifest JSON file.

Returns the v1.3 shape (run key), so a legacy experiment-keyed file on disk — a saved protocol or a recorded sidecar's embedded manifest — is transparently upgraded for every caller.

Source code in src/ssvep/io/manifest.py 
127
128
129
130
131
132
133
134
135
136
def load_manifest(path) -> dict:
    """Load, canonicalize, and validate a manifest JSON file.

    Returns the v1.3 shape (``run`` key), so a legacy ``experiment``-keyed file on disk — a saved
    protocol or a recorded sidecar's embedded manifest — is transparently upgraded for every caller.
    """
    with open(path, encoding="utf-8") as f:
        manifest = canonicalize_manifest(json.load(f))
    validate_manifest(manifest)
    return manifest

save_manifest 

save_manifest(manifest, path)

Validate then write a manifest to path (pretty-printed). Returns the path.

Source code in src/ssvep/io/manifest.py 
139
140
141
142
143
144
145
146
147
def save_manifest(manifest: dict, path) -> Path:
    """Validate then write a manifest to ``path`` (pretty-printed). Returns the path."""
    validate_manifest(manifest)
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        json.dump(manifest, f, indent=2)
        f.write("\n")
    return path

BIDS conversion

ssvep.io.bids

Convert acquired XDF recordings (sourcedata/) into a BIDS-EEG dataset.

The runtime records each run as an XDF (durable, LabRecorder-compatible) plus a *.session.json that embeds the full run manifest, provenance, consent and impedance (see :mod:ssvep.io.xdf and the runtime). That XDF is the raw truth and belongs in sourcedata/. This module derives a BIDS-compliant view of it: BrainVision (.vhdr/.vmrk/.eeg) signals at the subject level, with *_eeg.json sidecars, *_channels.tsv, *_events.tsv (every structured marker, sample-accurate), plus the dataset-level dataset_description.json, README, CHANGES, participants.tsv/.json and a task-<task>_events.json describing the event columns.

Design rules honoured here: * The manifest is the source of truth. Channel names, sampling rate, montage, target frequencies, and trial structure all come from the embedded manifest/markers — never inferred from the signal. * Sample-accurate events. Each marker is mapped to its nearest EEG sample via the recorded LSL timestamps (not time × nominal_rate), so events stay aligned even when the effective rate drifts a few tenths of a percent from nominal. * Compliance. Only sub-XXX codes; no demographics are invented (unknown → n/a). Everything written lives under the git-ignored BIDS/ tree.

Units: BrainFlow (Cyton / Unicorn) delivers EEG in microvolts; MNE works in volts, so we scale µV → V on the way into :class:mne.io.RawArray and BrainVision stores true volts.

Conversion is not optional and not a separate errand. Analysis reads the XDF directly, so nothing downstream needs the BIDS view — which is exactly how it came to be skipped: the GUI's Analyze tab wrote derivatives/ for a dataset that had no BIDS-standard raw data to be a derivative of. :func:ensure_converted (with :func:is_converted as the "is it already there?" check) is that step, callable per selection, so every path that analyses data also derives the dataset — 01_xdf_to_bids.py remains the way to (re)derive the whole tree at once.

ConversionReport dataclass

What :func:ensure_converted did — one entry per run, for a log line or a GUI status.

Source code in src/ssvep/io/bids.py 
530
531
532
533
534
535
536
537
538
539
540
@dataclass
class ConversionReport:
    """What :func:`ensure_converted` did — one entry per run, for a log line or a GUI status."""
    converted: list[str] = field(default_factory=list)   # stems newly written into the BIDS tree
    present: list[str] = field(default_factory=list)     # stems already up to date
    failed: list[tuple[str, str]] = field(default_factory=list)   # (stem, error); XDF still analysed
    eventless: list[str] = field(default_factory=list)   # converted, but not one marker in it

    @property
    def n_total(self) -> int:
        return len(self.converted) + len(self.present) + len(self.failed)

load_session 

load_session(session_json_path)

Load a *.session.json (manifest + provenance + consent + plan + impedance).

Source code in src/ssvep/io/bids.py 
63
64
65
66
def load_session(session_json_path) -> dict:
    """Load a ``*.session.json`` (manifest + provenance + consent + plan + impedance)."""
    with open(session_json_path, encoding="utf-8") as f:
        return json.load(f)

markers_to_events 

markers_to_events(marker_vals, marker_ts, eeg_ts, sfreq)

Build a rich BIDS events table from the structured JSON markers.

Every marker becomes a row. Each is snapped to its nearest EEG sample (via the recorded timestamps), so onset = sample / sfreq lands on that exact sample. stim_on rows get a duration running to the matching stim_off; the extra columns carry the decoding label (target_freq_hz) and trial structure.

Source code in src/ssvep/io/bids.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def markers_to_events(marker_vals, marker_ts, eeg_ts, sfreq) -> pd.DataFrame:
    """Build a rich BIDS ``events`` table from the structured JSON markers.

    Every marker becomes a row. Each is snapped to its nearest EEG **sample** (via the recorded
    timestamps), so ``onset = sample / sfreq`` lands on that exact sample. ``stim_on`` rows get a
    ``duration`` running to the matching ``stim_off``; the extra columns carry the decoding label
    (``target_freq_hz``) and trial structure.
    """
    if len(eeg_ts) == 0:
        return pd.DataFrame(columns=_EVENT_COLUMNS)
    t0 = float(eeg_ts[0])
    n_samp = len(eeg_ts)

    parsed = []
    for v, t in zip(marker_vals, marker_ts):
        try:
            d = json.loads(v)
        except Exception:
            d = {"segment": str(v)}
        idx = int(np.searchsorted(eeg_ts, float(t)))
        idx = max(0, min(idx, n_samp - 1))
        parsed.append((idx, d))

    # Pair stim_on → stim_off (by condition/block/trial/freq) to fill stim_on durations.
    def key(d):
        return (d.get("condition"), d.get("block"), d.get("trial"), d.get("target_freq_hz"))

    off_idx = {}
    for idx, d in parsed:
        if d.get("segment") == _STIM_OFF:
            off_idx[key(d)] = idx

    rows = []
    for idx, d in parsed:
        seg = d.get("segment", NA)
        freq = d.get("target_freq_hz")
        dur = NA
        if seg == _STIM_ON:
            end = off_idx.get(key(d))
            if end is not None and end >= idx:
                dur = round((end - idx) / sfreq, 6)
        rows.append({
            "onset": round(idx / sfreq, 6),
            "duration": dur,
            "trial_type": seg,
            "value": freq if freq is not None else NA,
            "sample": idx,
            "target_freq_hz": freq if freq is not None else NA,
            "condition": d.get("condition", NA),
            "block": d.get("block", NA),
            "trial": d.get("trial", NA),
            "target_id": d.get("target_id", NA),
        })
    df = pd.DataFrame(rows, columns=_EVENT_COLUMNS)
    return df.sort_values(["sample"], kind="stable").reset_index(drop=True)

build_raw 

build_raw(session, eeg_uv, eeg_ts, sfreq, xdf_labels, marker_vals, marker_ts, *, profile=None)

Assemble an :class:mne.io.RawArray (volts) with montage, meas date, line freq, annotations.

Returns (raw, events_df). The events table is the authoritative one we later write to *_events.tsv; annotations on the raw exist so MNE-based tools see the same events.

The mains frequency comes from the site profile, not from a constant: 60 Hz is a fact about North America, and this toolbox is not only run there (#165). Unconfigured ⇒ None, which BIDS records as "n/a" — unknown said out loud rather than a plausible wrong number.

Source code in src/ssvep/io/bids.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def build_raw(session: dict, eeg_uv, eeg_ts, sfreq, xdf_labels, marker_vals, marker_ts, *,
              profile=None):
    """Assemble an :class:`mne.io.RawArray` (volts) with montage, meas date, line freq, annotations.

    Returns ``(raw, events_df)``. The events table is the authoritative one we later write to
    ``*_events.tsv``; annotations on the raw exist so MNE-based tools see the same events.

    The mains frequency comes from the **site profile**, not from a constant: 60 Hz is a fact about
    North America, and this toolbox is not only run there (#165). Unconfigured ⇒ ``None``, which
    BIDS records as ``"n/a"`` — unknown said out loud rather than a plausible wrong number.
    """
    labels = _montage_labels(session, xdf_labels)
    data_v = (np.asarray(eeg_uv, dtype=float) / 1e6).T          # (n_ch, n_samples) volts
    info = mne.create_info(ch_names=labels, sfreq=float(sfreq), ch_types="eeg")
    raw = mne.io.RawArray(data_v, info, verbose="ERROR")
    raw.info["line_freq"] = _site.resolve(profile).line_frequency_hz

    # Best-effort standard-1020 montage (Unicorn = real 10-20 sites; Zeist colour labels won't match).
    try:
        montage = mne.channels.make_standard_montage("standard_1020")
        placed = set(montage.ch_names)
        if labels and all(lbl in placed for lbl in labels):
            raw.set_montage(montage, match_case=False, on_missing="ignore", verbose="ERROR")
    except Exception:
        pass

    # Measurement date from provenance (naive local wall-clock → tag UTC so MNE accepts it).
    started = (session.get("provenance") or {}).get("started_at")
    if started:
        try:
            dt = datetime.fromisoformat(started)
            if dt.tzinfo is None:
                dt = dt.replace(tzinfo=timezone.utc)
            raw.set_meas_date(dt.astimezone(timezone.utc))
        except Exception:
            pass

    events_df = markers_to_events(marker_vals, marker_ts, eeg_ts, sfreq)
    if len(events_df):
        onsets = events_df["sample"].to_numpy(dtype=float) / float(sfreq)
        durations = [0.0 if d == NA else float(d) for d in events_df["duration"]]
        descs = events_df["trial_type"].astype(str).tolist()
        raw.set_annotations(mne.Annotations(onset=onsets, duration=durations, description=descs),
                            verbose="ERROR")
    return raw, events_df

task_of 

task_of(session, default=DEFAULT_TASK)

The BIDS task label for a run — from its manifest, never from a caller-wide default.

The run manifest's run.task (legacy experiment.task) is the design's own statement of what ran ('ssvep', 'rest', …). A single task= argument applied to every run in a dataset silently relabels any run that isn't the majority paradigm: sub-902/ses-002's resting run landed in BIDS as task-ssvep despite its manifest saying rest, which is exactly the "analysis infers design" failure the manifest spine exists to prevent (CLAUDE.md §2).

Source code in src/ssvep/io/bids.py 
431
432
433
434
435
436
437
438
439
440
441
def task_of(session: dict, default: str = DEFAULT_TASK) -> str:
    """The BIDS task label for a run — from its **manifest**, never from a caller-wide default.

    The run manifest's ``run.task`` (legacy ``experiment.task``) is the design's own statement of what
    ran ('ssvep', 'rest', …). A single ``task=`` argument applied to every run in a dataset silently
    relabels any run that isn't the majority paradigm: sub-902/ses-002's resting run landed in BIDS as
    ``task-ssvep`` despite its manifest saying ``rest``, which is exactly the "analysis infers design"
    failure the manifest spine exists to prevent (CLAUDE.md §2).
    """
    t = run_meta((session or {}).get("manifest", {})).get("task")
    return str(t).strip() if t and str(t).strip() else default

bids_path_for 

bids_path_for(ref, bids_root, *, session=None, task=None)

Where this run's BIDS view lives — the same path :func:convert_run would write.

Split out so a caller can ask whether a run has been converted without converting it (:func:is_converted); the entities are derived in exactly one place.

Source code in src/ssvep/io/bids.py 
444
445
446
447
448
449
450
451
452
453
def bids_path_for(ref: RunRef, bids_root, *, session: dict | None = None,
                  task: str | None = None) -> BIDSPath:
    """Where this run's BIDS view lives — the same path :func:`convert_run` would write.

    Split out so a caller can ask *whether* a run has been converted without converting it
    (:func:`is_converted`); the entities are derived in exactly one place.
    """
    session = load_session(ref.session_json) if session is None else session
    return BIDSPath(subject=ref.sub, session=ref.ses, task=task or task_of(session), run=ref.run,
                    acquisition=_acq_label(session), datatype="eeg", root=str(bids_root))

is_converted 

is_converted(ref, bids_root, *, session=None)

Is this run's BrainVision view already on disk, no older than its XDF, and produced by the converter code currently running?

Existence alone is not enough: the BIDS tree is derived, so a source recording that is newer than the copy derived from it means the copy describes data that has since changed. Re-deriving is idempotent and cheap relative to being quietly wrong (CLAUDE.md §2.2).

The mtime check alone misses the other direction of that same rule: a converter code change (a bug fix in the conversion logic) doesn't touch the source XDF at all, so a run converted before the fix landed would report "up to date" forever even though re-running the fixed converter would produce different output (#113) — a fix that never reaches what already happened. So conversion also stamps a *.converter.json marker (:func:_stamp_converter_version) with a fingerprint of this module's source (:func:_converter_fingerprint), and this check compares it against the fingerprint of the code running now. A run with no marker at all — converted before this check existed — is treated as needing reconversion too, which is the one-time cost of adopting it.

Source code in src/ssvep/io/bids.py 
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
def is_converted(ref: RunRef, bids_root, *, session: dict | None = None) -> bool:
    """Is this run's BrainVision view already on disk, no older than its XDF, **and produced by the
    converter code currently running**?

    Existence alone is not enough: the BIDS tree is *derived*, so a source recording that is newer
    than the copy derived from it means the copy describes data that has since changed. Re-deriving
    is idempotent and cheap relative to being quietly wrong (CLAUDE.md §2.2).

    The mtime check alone misses the other direction of that same rule: a **converter code change**
    (a bug fix in the conversion logic) doesn't touch the source XDF at all, so a run converted before
    the fix landed would report "up to date" forever even though re-running the fixed converter would
    produce different output (#113) — a fix that never reaches what already happened. So conversion
    also stamps a ``*.converter.json`` marker (:func:`_stamp_converter_version`) with a fingerprint of
    this module's source (:func:`_converter_fingerprint`), and this check compares it against the
    fingerprint of the code running *now*. A run with no marker at all — converted before this check
    existed — is treated as needing reconversion too, which is the one-time cost of adopting it.
    """
    try:
        bp = bids_path_for(ref, bids_root, session=session)
        header = Path(bp.copy().update(suffix="eeg", extension=".vhdr").fpath)
        signals = header.with_suffix(".eeg")
        if not (header.exists() and signals.exists()):
            return False
        if header.stat().st_mtime < Path(ref.xdf).stat().st_mtime:
            return False
        marker = _converter_marker_path(header)
        if not marker.exists():
            return False      # converted before the version marker existed - reconvert once
        stamped = json.loads(marker.read_text(encoding="utf-8")).get("converter_fingerprint")
        return stamped == _converter_fingerprint()
    except OSError:
        return True          # the header is there; a stat we can't take is not a reason to redo it
    except Exception:
        return False

ensure_converted 

ensure_converted(refs, bids_root, *, progress=None)

Derive the BIDS view of refs for any run that does not already have an up-to-date one.

This is the 01_xdf_to_bids.py step, scoped to the runs a caller cares about. It exists because analysis reads the XDF directly, so a pipeline run would otherwise fill derivatives/ while leaving the dataset without the BIDS-standard raw data it is a derivative of — a tree that looks analysed but isn't a BIDS dataset.

Failure to convert one run is recorded, not raised: the XDF is the source of truth and the analysis can still run from it. progress(i, n, stem) is called before each conversion.

Source code in src/ssvep/io/bids.py 
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
def ensure_converted(refs, bids_root, *, progress=None) -> ConversionReport:
    """Derive the BIDS view of ``refs`` for any run that does not already have an up-to-date one.

    This is the ``01_xdf_to_bids.py`` step, scoped to the runs a caller cares about. It exists
    because analysis reads the **XDF** directly, so a pipeline run would otherwise fill
    ``derivatives/`` while leaving the dataset without the BIDS-standard raw data it is a derivative
    *of* — a tree that looks analysed but isn't a BIDS dataset.

    Failure to convert one run is recorded, not raised: the XDF is the source of truth and the
    analysis can still run from it. ``progress(i, n, stem)`` is called before each conversion.
    """
    bids_root = Path(bids_root)
    refs = list(refs)
    report = ConversionReport()
    todo = []
    for ref in refs:
        try:
            session = load_session(ref.session_json)
        except Exception as exc:
            report.failed.append((_stem(ref), f"unreadable session.json: {exc}"))
            continue
        if is_converted(ref, bids_root, session=session):
            report.present.append(_stem(ref))
            _note_if_eventless(ref, bids_root, session, report)
        else:
            todo.append(ref)

    for i, ref in enumerate(todo, start=1):
        if progress is not None:
            progress(i, len(todo), _stem(ref))
        try:
            convert_run(ref, bids_root)
            report.converted.append(_stem(ref))
            _note_if_eventless(ref, bids_root, session=load_session(ref.session_json), report=report)
        except Exception as exc:
            report.failed.append((_stem(ref), str(exc)))

    if report.converted:
        # Dataset-level metadata describes the whole tree, so it is rebuilt from every run on disk —
        # not just this selection, which would drop the task labels of the runs we didn't touch.
        all_refs = discover_runs(bids_root / "sourcedata") or refs
        tasks = set()
        for ref in all_refs:
            try:
                tasks.add(task_of(load_session(ref.session_json)))
            except Exception:
                pass
        write_dataset_metadata(bids_root, tasks=sorted(tasks) or None)
        enrich_participants(bids_root)
    return report

convert_run 

convert_run(ref, bids_root, task=None, *, overwrite=True)

Convert one XDF run into the BIDS tree and return its EEG :class:BIDSPath.

task overrides the manifest's own label; leave it None (the normal path) to use task_of.

Source code in src/ssvep/io/bids.py 
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
def convert_run(ref: RunRef, bids_root, task: str | None = None, *, overwrite: bool = True) -> BIDSPath:
    """Convert one XDF run into the BIDS tree and return its EEG :class:`BIDSPath`.

    ``task`` overrides the manifest's own label; leave it None (the normal path) to use ``task_of``.
    """
    session = load_session(ref.session_json)
    task = task or task_of(session)
    eeg_uv, eeg_ts, sfreq, xdf_labels, mvals, mts = _read_xdf(ref.xdf)
    raw, events_df = build_raw(session, eeg_uv, eeg_ts, sfreq, xdf_labels, mvals, mts)

    bids_path = BIDSPath(subject=ref.sub, session=ref.ses, task=task, run=ref.run,
                         acquisition=_acq_label(session), datatype="eeg", root=str(bids_root))

    # Let mne-bids lay down BrainVision + channels.tsv + base events.tsv + sidecar + scans +
    # participants/dataset stubs. We then overwrite events.tsv with the rich, sample-accurate
    # table and enrich the sidecar with device/SSVEP fields.
    write_raw_bids(raw, bids_path, format="BrainVision", allow_preload=True,
                   overwrite=overwrite, verbose="ERROR")

    if len(events_df):
        events_tsv = bids_path.copy().update(suffix="events", extension=".tsv").fpath
        events_df.to_csv(events_tsv, sep="\t", index=False, na_rep=NA)

    _rewrite_run_level_events_sidecar(bids_path, session)

    sidecar = bids_path.copy().update(suffix="eeg", extension=".json").fpath
    _merge_json(Path(sidecar), _eeg_sidecar_extra(session))

    _augment_channels_impedance(bids_path, session)
    _stamp_converter_version(bids_path)
    return bids_path

discover_runs 

discover_runs(sourcedata_root)

Find every *_eeg.xdf under sourcedata/ and pair it with its *.session.json.

Source code in src/ssvep/io/bids.py 
718
719
720
721
722
723
724
725
726
727
728
729
def discover_runs(sourcedata_root: Path) -> list[RunRef]:
    """Find every ``*_eeg.xdf`` under ``sourcedata/`` and pair it with its ``*.session.json``."""
    refs: list[RunRef] = []
    for xdf in sorted(Path(sourcedata_root).glob("sub-*/ses-*/eeg/*_eeg.xdf")):
        sub, ses, run = _parse_entities(xdf)
        session_json = xdf.with_name(xdf.name.replace("_eeg.xdf", "_eeg.session.json"))
        if not session_json.exists():
            alt = xdf.with_suffix(".session.json")
            session_json = alt if alt.exists() else session_json
        if sub and ses and run and session_json.exists():
            refs.append(RunRef(xdf, session_json, sub, ses, run))
    return refs

migrate_toplevel_to_sourcedata 

migrate_toplevel_to_sourcedata(bids_root)

Move raw *.xdf / *.session.json / *_impedance.json sitting at the subject level (the pre-BIDS layout) down into sourcedata/. Idempotent — returns the files moved.

Source code in src/ssvep/io/bids.py 
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
def migrate_toplevel_to_sourcedata(bids_root: Path) -> list[Path]:
    """Move raw ``*.xdf`` / ``*.session.json`` / ``*_impedance.json`` sitting at the subject level
    (the pre-BIDS layout) down into ``sourcedata/``. Idempotent — returns the files moved.
    """
    bids_root = Path(bids_root)
    sourcedata = bids_root / "sourcedata"
    moved = []
    for eeg_dir in sorted(bids_root.glob("sub-*/ses-*/eeg")):
        if sourcedata in eeg_dir.parents:
            continue
        rel = eeg_dir.relative_to(bids_root)
        dest = sourcedata / rel
        for f in sorted(eeg_dir.iterdir()):
            if f.suffix == ".xdf" or f.name.endswith(".session.json") or f.name.endswith("_impedance.json"):
                dest.mkdir(parents=True, exist_ok=True)
                target = dest / f.name
                shutil.move(str(f), str(target))
                moved.append(target)
    return moved

write_dataset_metadata 

write_dataset_metadata(bids_root, task=None, *, tasks=None, profile=None)

(Re)write the dataset-level files with our richer, compliance-aware content.

tasks is every task label present in the dataset — each gets its own task-<x>_events.json, because a dataset holds more than one paradigm (ssvep + rest). task is the single-label legacy form.

Who made the dataset and under what approval comes from the site profile (#165). An unconfigured deployment produces a dataset that names no lab and cites no ethics file — and HowToAcknowledge is then omitted rather than invented, while License still defaults to "not licensed for redistribution", because the restrictive reading is the safe one when nobody has said (COMPLIANCE R5's principle applied to redistribution).

Source code in src/ssvep/io/bids.py 
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
def write_dataset_metadata(bids_root: Path, task: str | None = None, *, tasks=None,
                           profile=None) -> None:
    """(Re)write the dataset-level files with our richer, compliance-aware content.

    ``tasks`` is every task label present in the dataset — each gets its own ``task-<x>_events.json``,
    because a dataset holds more than one paradigm (ssvep + rest). ``task`` is the single-label legacy
    form.

    Who made the dataset and under what approval comes from the **site profile** (#165). An
    unconfigured deployment produces a dataset that names no lab and cites no ethics file — and
    ``HowToAcknowledge`` is then omitted rather than invented, while ``License`` still defaults to
    "not licensed for redistribution", because the restrictive reading is the safe one when nobody
    has said (COMPLIANCE R5's principle applied to redistribution).
    """
    bids_root = Path(bids_root)
    labels = list(tasks) if tasks else [task or DEFAULT_TASK]
    profile = _site.resolve(profile)

    dataset_description = {
        "Name": profile.dataset.name,
        "BIDSVersion": BIDS_VERSION,
        "DatasetType": "raw",
        "Authors": list(profile.dataset.authors),
        "Acknowledgements": profile.dataset.acknowledgements,
        **({"HowToAcknowledge": profile.governance.acknowledgement()}
           if profile.governance.acknowledgement() else {}),
        "License": profile.governance.data_license,
        "GeneratedBy": [{
            "Name": "ssvep-toolbox bids converter (ssvep.io.bids)",
            "Version": SSVEP_VERSION,
            "Description": "Derives BIDS-EEG (BrainVision) from sourcedata XDF + embedded manifest.",
        }],
        "SourceDatasets": [{"Description": "Raw XDF recordings under sourcedata/ (LabRecorder-compatible)."}],
    }
    (bids_root / "dataset_description.json").write_text(
        json.dumps(dataset_description, indent=2) + "\n", encoding="utf-8")

    (bids_root / "README").write_text(_readme_text(profile), encoding="utf-8")
    (bids_root / "CHANGES").write_text(_CHANGES.strip() + "\n", encoding="utf-8")
    (bids_root / ".bidsignore").write_text(_BIDSIGNORE.strip() + "\n", encoding="utf-8")

    (bids_root / "participants.json").write_text(
        json.dumps(_PARTICIPANTS_JSON, indent=2) + "\n", encoding="utf-8")

    for t in labels:
        (bids_root / f"task-{t}_events.json").write_text(
            json.dumps(_EVENTS_JSON, indent=2) + "\n", encoding="utf-8")

    # Root-level (inherited) channels.json documents the nonstandard impedance column.
    (bids_root / "channels.json").write_text(
        json.dumps(_CHANNELS_JSON, indent=2) + "\n", encoding="utf-8")

enrich_participants 

enrich_participants(bids_root)

Add species / group columns to the mne-bids-generated participants.tsv.

No demographics are invented — unknown fields stay n/a (COMPLIANCE: coarse-only, no linkage). group is inferred solely from the reserved sub-000 test code.

Source code in src/ssvep/io/bids.py 
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
def enrich_participants(bids_root: Path) -> None:
    """Add ``species`` / ``group`` columns to the mne-bids-generated ``participants.tsv``.

    No demographics are invented — unknown fields stay ``n/a`` (COMPLIANCE: coarse-only, no linkage).
    ``group`` is inferred solely from the reserved sub-000 test code.
    """
    bids_root = Path(bids_root)
    ptsv = bids_root / "participants.tsv"
    if not ptsv.exists():
        return
    df = pd.read_csv(ptsv, sep="\t", dtype=str).fillna(NA)
    df["species"] = "homo sapiens"
    df["group"] = df["participant_id"].map(lambda p: "test" if p == "sub-000" else "pilot")
    front = ["participant_id"]
    cols = front + [c for c in df.columns if c not in front]
    df[cols].to_csv(ptsv, sep="\t", index=False, na_rep=NA)

convert_dataset 

convert_dataset(bids_root, task=None, *, migrate=True)

End-to-end: migrate legacy layout → convert every sourcedata run → write dataset metadata.

Each run is labelled from its own manifest (:func:task_of); task forces one label onto every run and exists only for tests/legacy callers. A dataset can hold more than one paradigm, so a task-<x>_events.json is written for every task actually present.

Source code in src/ssvep/io/bids.py 
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
def convert_dataset(bids_root, task: str | None = None, *, migrate: bool = True) -> list[BIDSPath]:
    """End-to-end: migrate legacy layout → convert every sourcedata run → write dataset metadata.

    Each run is labelled from its **own** manifest (:func:`task_of`); ``task`` forces one label onto
    every run and exists only for tests/legacy callers. A dataset can hold more than one paradigm, so
    a ``task-<x>_events.json`` is written for every task actually present.
    """
    bids_root = Path(bids_root)
    if migrate:
        migrate_toplevel_to_sourcedata(bids_root)
    refs = discover_runs(bids_root / "sourcedata")
    written = [convert_run(ref, bids_root, task=task) for ref in refs]
    tasks = sorted({bp.task for bp in written}) or [task or DEFAULT_TASK]
    write_dataset_metadata(bids_root, tasks=tasks)
    enrich_participants(bids_root)
    return written

ssvep.analysis — offline pipelines

Batch pipeline

ssvep.analysis.batch

Run the offline pipeline over a selected part of a BIDS tree (#67).

This is the one place the offline batch lives. BIDS/code/02_run_offline_pipeline.py is a thin argparse wrapper over :func:run_batch, and the GUI's Analyze tab calls the same function on a worker thread — so a session analysed from the command line and one analysed from the GUI are the same code path, the same output location, and the same run log.

Selector. :func:select_runs filters what :func:ssvep.io.bids.discover_runs found, rather than narrowing the glob: discovery keeps one definition of what a run is. sub/ses accept either the bare number or the full entity (902 / sub-902) because operators type both. A selector that matches nothing raises :class:NoRunsSelected listing what is available — a silent "0 analysed" reads as success and is exactly how a typo'd ID becomes "my data is fine".

The BIDS view is derived first. Analysis reads sourcedata/ XDF, so nothing downstream needs the BrainVision copy — and so it was skipped: analysing from the GUI filled derivatives/ for a dataset that had no BIDS-standard raw data under it. :func:run_batch now runs the 01_xdf_to_bids.py step over its selection first (ensure_bids=True), converting only runs that lack an up-to-date BIDS view. A run that fails to convert is logged and still analysed.

Two paradigms, one runner. Each recording is dispatched on its manifest paradigm field (schema v1.2), never on the data: ssvep runs epoch → calibration-free decode → metrics → report, while resting gets peak-alpha-frequency + eyes-closed/eyes-open reactivity via :mod:ssvep.analysis.resting_paf, reported with the same *_report.html + *_metrics.json shape.

Spatial policy. Per run, the decode reference + channel set come from the manifest via :func:ssvep.spatial.occipital_decode_plan: a large, reference-free cap (the 64-ch actiCHamp recorded ground-only) decodes from the occipital ROI with a common-average reference; small/already-referenced montages are left as recorded. Without this the 64-ch cap decodes at chance (DESIGN_PRINCIPLES #8).

Run log. Every invocation writes a timestamped log under derivatives/ssvep-analysis/logs/ with the toolbox + dependency versions, the git commit, the selector used, and the per-run spatial policy + results — so a log is never ambiguous about whether it covered the dataset or one session.

Output layout + index. Reports land in derivatives/ssvep-analysis/sub-XXX/ses-YYY/ and the batch finishes by rebuilding the persistent index.html over the whole tree (:mod:ssvep.analysis.derivatives, #80/#18). It no longer writes a summary_<stamp>.html per invocation: that page described one batch, so a re-analysis left another one beside the reports and none of them was the entry point. The index is derived from disk, so it covers every run analysed so far — not just the ones this batch touched — and a crashed batch still leaves it truthful.

NoRunsSelected

Bases: Exception

The selector matched no recordings. Carries what is on disk so the caller can say so.

Source code in src/ssvep/analysis/batch.py 
61
62
63
64
65
66
class NoRunsSelected(Exception):
    """The selector matched no recordings. Carries what *is* on disk so the caller can say so."""

    def __init__(self, message: str, *, available: dict[str, list[str]] | None = None):
        super().__init__(message)
        self.available = available or {}

RunResult dataclass

One recording's outcome — enough for a GUI row without re-reading the metrics.

Source code in src/ssvep/analysis/batch.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
@dataclass
class RunResult:
    """One recording's outcome — enough for a GUI row without re-reading the metrics."""
    sub: str
    ses: str
    run: str
    stem: str
    paradigm: str = "ssvep"
    ok: bool = False
    report_html: Path | None = None
    metrics_json: Path | None = None
    summary: str = ""
    error: str | None = None

normalize_entity 

normalize_entity(value, prefix)

'902'/'sub-902'/902 → '902'. None/blank → None.

Only the label is returned (no prefix), because that is what :class:~ssvep.io.bids.RunRef carries. Operators type both forms and neither should be a miss.

Source code in src/ssvep/analysis/batch.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def normalize_entity(value, prefix: str) -> str | None:
    """``'902'``/``'sub-902'``/``902`` → ``'902'``. ``None``/blank → ``None``.

    Only the *label* is returned (no prefix), because that is what :class:`~ssvep.io.bids.RunRef`
    carries. Operators type both forms and neither should be a miss.
    """
    if value is None:
        return None
    text = str(value).strip()
    if not text:
        return None
    low = text.lower()
    if low.startswith(f"{prefix}-"):
        text = text[len(prefix) + 1:]
    return text.strip() or None

available_entities 

available_entities(refs)

{'902': ['001', '002'], …} — what the tree actually holds, for error messages and the GUI.

Source code in src/ssvep/analysis/batch.py 
122
123
124
125
126
127
128
129
def available_entities(refs) -> dict[str, list[str]]:
    """``{'902': ['001', '002'], …}`` — what the tree actually holds, for error messages and the GUI."""
    out: dict[str, list[str]] = {}
    for ref in refs:
        out.setdefault(ref.sub, [])
        if ref.ses not in out[ref.sub]:
            out[ref.sub].append(ref.ses)
    return {k: sorted(v) for k, v in sorted(out.items())}

selector_label 

selector_label(sub=None, ses=None, runs=None)

Human-readable description of a selection — recorded verbatim in the run log.

Source code in src/ssvep/analysis/batch.py 
132
133
134
135
136
137
138
139
140
def selector_label(sub=None, ses=None, runs=None) -> str:
    """Human-readable description of a selection — recorded verbatim in the run log."""
    if runs:
        return f"runs={','.join(sorted(runs))}"
    if sub and ses:
        return f"sub-{sub} ses-{ses}"
    if sub:
        return f"sub-{sub} (all sessions)"
    return "whole dataset"

select_runs 

select_runs(refs, *, sub=None, ses=None)

Filter discovered runs to a subject/session. Raises :class:NoRunsSelected on an empty match.

ses without sub is an error, not a cross-subject sweep: "session 002" is only meaningful within a participant. No selector at all keeps the whole tree.

Source code in src/ssvep/analysis/batch.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def select_runs(refs, *, sub=None, ses=None):
    """Filter discovered runs to a subject/session. Raises :class:`NoRunsSelected` on an empty match.

    ``ses`` without ``sub`` is an error, not a cross-subject sweep: "session 002" is only meaningful
    within a participant. No selector at all keeps the whole tree.
    """
    sub = normalize_entity(sub, "sub")
    ses = normalize_entity(ses, "ses")
    if ses and not sub:
        raise ValueError("--ses requires --sub: a session number is only meaningful within a subject.")
    picked = [r for r in refs
              if (sub is None or r.sub == sub) and (ses is None or r.ses == ses)]
    if not picked:
        avail = available_entities(refs)
        listing = "; ".join(f"sub-{s}: {', '.join('ses-' + x for x in v)}" for s, v in avail.items())
        raise NoRunsSelected(
            f"No recordings match {selector_label(sub, ses)}. Available — {listing or '(none)'}",
            available=avail)
    return picked

analyse_run 

analyse_run(ref, deriv, *, method='fbcca', auto_spatial=True)

Analyse one discovered run, dispatching on the manifest's paradigm. Never raises.

deriv is the derivatives root; the outputs go to this run's sub-XXX/ses-YYY/ under it (:func:ssvep.analysis.derivatives.run_dir), mirroring the raw data's relative path.

Source code in src/ssvep/analysis/batch.py 
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
def analyse_run(ref, deriv: Path, *, method: str = "fbcca", auto_spatial: bool = True) -> RunResult:
    """Analyse one discovered run, dispatching on the manifest's paradigm. Never raises.

    ``deriv`` is the derivatives **root**; the outputs go to this run's ``sub-XXX/ses-YYY/`` under it
    (:func:`ssvep.analysis.derivatives.run_dir`), mirroring the raw data's relative path.
    """
    stem = ref.xdf.stem.replace("_eeg", "")
    out_dir = derivatives.run_dir(deriv, ref.sub, ref.ses)
    try:
        manifest = bids.load_session(ref.session_json).get("manifest")
        # Resting recordings have no flicker and no trials to decode: they take the PAF /
        # alpha-reactivity path instead. Dispatch on the manifest's paradigm, never on the data
        # (or on the empty stimulus list) — DESIGN_PRINCIPLES: the manifest is the truth.
        if manifest_io.is_resting(manifest):
            return _analyse_resting(ref, stem, manifest, out_dir)
        return _analyse_ssvep(ref, stem, manifest, out_dir, method=method, auto_spatial=auto_spatial)
    except Exception as exc:                        # keep going; a bad run shouldn't abort the batch
        log.exception("%s: FAILED (%s)", stem, exc)
        return RunResult(ref.sub, ref.ses, ref.run, stem, ok=False, error=str(exc))

discover 

discover(bids_root)

Every run under <bids_root>/sourcedata (the GUI populates its tree from this).

Source code in src/ssvep/analysis/batch.py 
286
287
288
def discover(bids_root) -> list:
    """Every run under ``<bids_root>/sourcedata`` (the GUI populates its tree from this)."""
    return bids.discover_runs(Path(bids_root) / "sourcedata")

run_batch 

run_batch(bids_root, *, sub=None, ses=None, method='fbcca', auto_spatial=True, refs=None, progress=None, to_stdout=True, ensure_bids=True, on_convert=None)

Analyse the selected recordings under bids_root, writing reports + a run log.

refs lets a caller (the GUI) pass an explicit run list it already discovered and let the operator tick — the sub/ses selector is then not applied. progress is called as progress(i, n, RunResult|None): before each run with None, and after it with the result, so a GUI can show where it is without this module knowing about Qt.

ensure_bids (default on) derives the BIDS view of the selected runs first, for any that lack an up-to-date one — the 01_xdf_to_bids.py step, scoped to the selection. Analysis reads the XDF, so skipping it produced the failure this default exists to prevent: a derivatives/ tree over a dataset with no BIDS-standard raw data. on_convert(i, n, stem) reports its progress (it is the slow part on a first analysis). A run that fails to convert is logged and still analysed — the XDF is the source of truth.

Source code in src/ssvep/analysis/batch.py 
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
def run_batch(bids_root, *, sub=None, ses=None, method: str = "fbcca", auto_spatial: bool = True,
              refs=None, progress=None, to_stdout: bool = True, ensure_bids: bool = True,
              on_convert=None) -> BatchResult:
    """Analyse the selected recordings under ``bids_root``, writing reports + a run log.

    ``refs`` lets a caller (the GUI) pass an explicit run list it already discovered and let the
    operator tick — the sub/ses selector is then not applied. ``progress`` is called as
    ``progress(i, n, RunResult|None)``: before each run with ``None``, and after it with the result,
    so a GUI can show where it is without this module knowing about Qt.

    ``ensure_bids`` (default on) derives the **BIDS view** of the selected runs first, for any that
    lack an up-to-date one — the ``01_xdf_to_bids.py`` step, scoped to the selection. Analysis reads
    the XDF, so skipping it produced the failure this default exists to prevent: a ``derivatives/``
    tree over a dataset with no BIDS-standard raw data. ``on_convert(i, n, stem)`` reports its
    progress (it is the slow part on a first analysis). A run that fails to convert is logged and
    still analysed — the XDF is the source of truth.
    """
    bids_root = Path(bids_root)
    deriv = bids_root / DERIV_NAME
    deriv.mkdir(parents=True, exist_ok=True)
    logpath = _setup_logging(deriv / "logs", to_stdout=to_stdout)

    # Pre-#80 trees put every report in one flat directory beside a pile of summary_<stamp>.html.
    # Move them into sub-XXX/ses-YYY/ so the index can cover work analysed before this version —
    # re-running analysis to get a navigable tree would be a pointless hour per session.
    migrated = derivatives.migrate_flat_layout(deriv)
    if migrated:
        log.info("layout: moved %d file(s) into sub-XXX/ses-YYY/, removed %d obsolete summary page(s)",
                 len(migrated.moved), len(migrated.removed))

    if refs is None:
        found = discover(bids_root)
        if not found:
            raise NoRunsSelected(f"No recordings under {bids_root / 'sourcedata'}. "
                                 f"Run 01_xdf_to_bids.py first.")
        picked = select_runs(found, sub=sub, ses=ses)
        selector = selector_label(normalize_entity(sub, "sub"), normalize_entity(ses, "ses"))
    else:
        picked = list(refs)
        if not picked:
            raise NoRunsSelected("No recordings selected.")
        selector = selector_label(runs=[r.xdf.stem.replace("_eeg", "") for r in picked])

    log.info("=== SSVEP offline pipeline ===")
    log.info("bids_root=%s  method=%s  auto_spatial=%s", bids_root, method, auto_spatial)
    log.info("selector: %s", selector)          # never ambiguous: whole dataset, or one session?
    log.info("versions: %s", _versions())
    log.info("git: %s", _git_sha())
    log.info("selected %d run(s)", len(picked))

    out = BatchResult(selector=selector, log_path=logpath, deriv_dir=deriv)

    if ensure_bids:
        out.conversion = _ensure_bids(bids_root, picked, on_convert)

    for i, ref in enumerate(picked, start=1):
        if progress is not None:
            progress(i, len(picked), None)
        res = analyse_run(ref, deriv, method=method, auto_spatial=auto_spatial)
        out.runs.append(res)
        if progress is not None:
            progress(i, len(picked), res)

    # Rebuilt from disk, so it covers every run ever analysed under this root — not just this
    # batch's selection — and stays truthful if the batch above died part-way through.
    out.index_html = derivatives.build_index(deriv, bids_root=bids_root, git_sha=_git_sha())
    log.info("done [%s]: %d analysed, %d skipped/failed. reports+metrics -> %s ; log -> %s",
             selector, out.n_ok, out.n_failed, deriv, logpath)
    log.info("index -> %s", out.index_html)
    return out