ssvep.io — manifests and data

Part of the API reference, generated from the docstrings in src/ssvep.

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 (AGENTS.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. Manifest v1.4 adds image targets and retires the stimulus shape: "image" that never drew one; :func:canonicalize_manifest maps that too.

canonicalize_manifest

canonicalize_manifest(manifest)

Return manifest in the current shape, mapping the two legacy spellings a saved one can have.

• experiment → run (v1.3, #42). Every sidecar ever written embeds the manifest it recorded under (RunRecorder.sidecar), and the older saved protocols predate the rename.
• Stimulus shape: "image" → "rect" (v1.4, #256). The value was in the schema's enum from v1.0 and nothing ever drew an image for it — the renderer fell through to a rectangle, in whatever pattern the target had. So "rect" is what such a manifest showed, and mapping it there keeps the manifest describing what was on screen. Only when the stimulus also carries an image block — a picture that no renderer has yet drawn, but that the manifest does declare — does it become pattern: "image", since an image is a fill, not an outline. Mapping it to pattern: "image" unconditionally would declare a picture that was never there, and would stop the manifest validating: v1.4 requires a picture for an image fill.

Idempotent and non-mutating: a manifest needing neither mapping is returned as the same object; otherwise a shallow copy is made (and each rewritten stimulus is itself a copy). This is the single point that keeps every saved manifest loadable, and it runs inside :func:manifest_fingerprint, so a legacy file and its re-saved current-shape copy fingerprint identically. The shape mapping does move the fingerprint of a file that still says "image" relative to builds before it; no saved manifest, protocol or sidecar in this repository says it, and a pre-1.2 protocol pinning one would report drift — the hard stop those have always had.

Source code in src/ssvep/io/manifest.py

def canonicalize_manifest(manifest: dict) -> dict:
    """Return ``manifest`` in the current shape, mapping the two legacy spellings a saved one can have.

    * **``experiment`` → ``run``** (v1.3, #42). Every sidecar ever written embeds the manifest it
      recorded under (``RunRecorder.sidecar``), and the older saved protocols predate the rename.
    * **Stimulus ``shape: "image"`` → ``"rect"``** (v1.4, #256). The value was in the schema's enum
      from v1.0 and nothing ever drew an image for it — the renderer fell through to a rectangle, in
      whatever ``pattern`` the target had. So ``"rect"`` is what such a manifest *showed*, and
      mapping it there keeps the manifest describing what was on screen. Only when the stimulus also
      carries an ``image`` block — a picture that no renderer has yet drawn, but that the manifest
      does declare — does it become ``pattern: "image"``, since an image is a fill, not an outline.
      Mapping it to ``pattern: "image"`` unconditionally would declare a picture that was never
      there, and would stop the manifest validating: v1.4 requires a picture for an image fill.

    Idempotent and non-mutating: a manifest needing neither mapping is returned **as the same
    object**; otherwise a shallow copy is made (and each rewritten stimulus is itself a copy). This
    is the single point that keeps every saved manifest loadable, and it runs inside
    :func:`manifest_fingerprint`, so a legacy file and its re-saved current-shape copy fingerprint
    identically. The shape mapping does move the fingerprint of a file that still says ``"image"``
    relative to builds before it; no saved manifest, protocol or sidecar in this repository says it,
    and a pre-1.2 protocol pinning one would report drift — the hard stop those have always had.
    """
    if not isinstance(manifest, dict):
        return manifest
    m = manifest
    if _RUN_KEY not in m and _LEGACY_RUN_KEY in m:
        m = dict(m)
        m[_RUN_KEY] = m.pop(_LEGACY_RUN_KEY)
    stimuli = m.get("stimuli")
    if isinstance(stimuli, list) and any(isinstance(s, dict) and s.get("shape") == _LEGACY_IMAGE_SHAPE
                                         for s in stimuli):
        if m is manifest:
            m = dict(m)
        m["stimuli"] = [_canonical_stimulus(s) for s in stimuli]
    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

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

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.
    """
    canon = canonicalize_manifest(manifest)
    acq = canon.get("acquisition") if isinstance(canon, dict) else None
    if isinstance(acq, dict) and "carrier_hash" in acq:
        # Derived from the block's own wiring (#410), so it adds no identity: re-saving a design that
        # predates it must not move its fingerprint.
        canon = {**canon, "acquisition": {k: v for k, v in acq.items() if k != "carrier_hash"}}
    blob = json.dumps(canon, sort_keys=True, separators=(",", ":")).encode("utf-8")
    return hashlib.sha256(blob).hexdigest()[:12]

design_fingerprint

design_fingerprint(manifest)

:func:manifest_fingerprint with provenance-only and prose fields ignored.

"Is this the same design?" for the name check (#467): run.description is prose, and run.builder_version / manifest_version record which build wrote the file, so none of them make a different design. Without this an identical copy would stop being identical after a package version bump. Nor does acquisition: hardware is the session's, not the run's (#484), so two copies of one design that an older build stamped with different hardware are one design. Never stored or pinned anywhere, so it cannot move any recorded fingerprint; :func:manifest_fingerprint itself is untouched.

Source code in src/ssvep/io/manifest.py

def design_fingerprint(manifest: dict) -> str:
    """:func:`manifest_fingerprint` with provenance-only and prose fields ignored.

    "Is this the same design?" for the name check (#467): ``run.description`` is prose, and
    ``run.builder_version`` / ``manifest_version`` record which build wrote the file, so none of them
    make a different design. Without this an identical copy would stop being identical after a
    package version bump. Nor does ``acquisition``: hardware is the session's, not the run's (#484),
    so two copies of one design that an older build stamped with different hardware are one design.
    Never stored or pinned anywhere, so it cannot move any recorded fingerprint;
    :func:`manifest_fingerprint` itself is untouched.
    """
    canon = canonicalize_manifest(manifest)
    if isinstance(canon, dict):
        canon = {k: v for k, v in canon.items() if k not in ("manifest_version", "acquisition")}
        run = canon.get(_RUN_KEY)
        if isinstance(run, dict):
            canon[_RUN_KEY] = {k: v for k, v in run.items()
                               if k not in ("description", "builder_version")}
    return manifest_fingerprint(canon)

name_clashes

name_clashes(manifest, folders, exclude=None)

Saved run manifests in folders that carry this manifest's run.name as a different design.

The save-time duplicate-name check (#467), asked only when a collision would be created (see DesignWindow._resolve_name_clash). Same name + same design (:func:design_fingerprint) is not a clash, so re-saving a run, or saving an identical copy, is allowed. exclude is the path being saved to (a file never clashes with itself). Unreadable or non-manifest *.json files are skipped: this is a nicety, never a reason a save fails. Folders are scanned non-recursively and de-duplicated; a missing folder is ignored.

Source code in src/ssvep/io/manifest.py

def name_clashes(manifest: dict, folders, exclude=None) -> list[Path]:
    """Saved run manifests in ``folders`` that carry this manifest's ``run.name`` as a different design.

    The save-time duplicate-name check (#467), asked only when a collision would be *created*
    (see ``DesignWindow._resolve_name_clash``). Same name + same design (:func:`design_fingerprint`)
    is not a clash, so re-saving a run, or saving an identical copy, is allowed. ``exclude`` is the
    path being saved to (a file never clashes with itself). Unreadable or non-manifest ``*.json``
    files are skipped: this is a nicety, never a reason a save fails. Folders are scanned
    non-recursively and de-duplicated; a missing folder is ignored.
    """
    name = run_meta(manifest).get("name")
    if not name:
        return []
    mine = design_fingerprint(manifest)
    skip = os.path.normcase(os.path.abspath(str(exclude))) if exclude else None
    seen: set[str] = set()
    out: list[Path] = []
    for folder in folders:
        d = Path(folder)
        if not d.is_dir():
            continue
        for f in sorted(d.glob("*.json")):
            key = os.path.normcase(os.path.abspath(str(f)))
            if key in seen or key == skip:
                continue
            seen.add(key)
            try:
                other = load_manifest(f)
            except Exception:
                continue
            if run_meta(other).get("name") == name and design_fingerprint(other) != mine:
                out.append(f)
    return out

saved_run_names

saved_run_names(folders)

Every run.name among the readable run manifests in folders (see :func:name_clashes).

Source code in src/ssvep/io/manifest.py

def saved_run_names(folders) -> set[str]:
    """Every ``run.name`` among the readable run manifests in ``folders`` (see :func:`name_clashes`)."""
    out: set[str] = set()
    for folder in folders:
        d = Path(folder)
        for f in (sorted(d.glob("*.json")) if d.is_dir() else ()):
            try:
                n = run_meta(load_manifest(f)).get("name")
            except Exception:
                continue
            if n:
                out.add(n)
    return out

suggest_free_name

suggest_free_name(name, taken)

name with the smallest " (2)", " (3)"... suffix not in taken.

Source code in src/ssvep/io/manifest.py

def suggest_free_name(name: str, taken) -> str:
    """``name`` with the smallest ``" (2)"``, ``" (3)"``... suffix not in ``taken``."""
    used = set(taken)
    n = 2
    while f"{name} ({n})" in used:
        n += 1
    return f"{name} ({n})"

paradigm

paradigm(manifest)

The manifest's paradigm: 'ssvep' (flickering targets), 'resting' (eyes-open/closed) or 'fpvs' (fast periodic visual stimulation, #494).

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

Source code in src/ssvep/io/manifest.py

def paradigm(manifest: dict) -> str:
    """The manifest's paradigm: ``'ssvep'`` (flickering targets), ``'resting'`` (eyes-open/closed)
    or ``'fpvs'`` (fast periodic visual stimulation, #494).

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

feedback_config

feedback_config(manifest)

{mode, wait_s, show_s} when the run declares participant feedback, else None.

The manifest is authoritative (AGENTS.md section 2.2): the schedule, the renderer, the live answer's route and the analysis all read this one function and never infer feedback from the data. Absent block, or mode: "none", is None - a run that showed nothing between trials. A resting or FPVS run has no decoder to answer from, so it never has feedback.

Source code in src/ssvep/io/manifest.py

def feedback_config(manifest: dict) -> dict | None:
    """``{mode, wait_s, show_s}`` when the run declares participant feedback, else ``None``.

    The manifest is authoritative (AGENTS.md section 2.2): the schedule, the renderer, the live
    answer's route and the analysis all read this one function and never infer feedback from the
    data. Absent block, or ``mode: "none"``, is ``None`` - a run that showed nothing between trials.
    A resting or FPVS run has no decoder to answer from, so it never has feedback.
    """
    if not has_decoder(manifest):
        return None
    fb = ((manifest or {}).get("design") or {}).get("feedback") or {}
    mode = fb.get("mode") or FEEDBACK_NONE
    if mode == FEEDBACK_NONE:
        return None
    return {"mode": mode,
            "wait_s": float(fb.get("wait_s", FEEDBACK_WAIT_S)),
            "show_s": float(fb.get("show_s", FEEDBACK_SHOW_S))}

cue_mode

cue_mode(manifest)

'cued' or 'free': who picked each trial's target, as the manifest declares it.

Every consumer of a trial's label asks this one function (never "is target_freq_hz there?"): a free-choice run has no label by design, and inferring that from a missing field is how a free run would silently be treated as cued. A resting run has no targets to choose and reads cued.

Source code in src/ssvep/io/manifest.py

def cue_mode(manifest: dict) -> str:
    """``'cued'`` or ``'free'``: who picked each trial's target, as the manifest declares it.

    Every consumer of a trial's label asks this one function (never "is ``target_freq_hz`` there?"):
    a free-choice run has no label by design, and inferring that from a missing field is how a free
    run would silently be treated as cued. A resting run has no targets to choose and reads ``cued``.
    """
    if not has_decoder(manifest):
        return CUE_CUED
    return ((manifest or {}).get("design") or {}).get("cue_mode") or CUE_CUED

is_cued

is_cued(manifest)

False only for a free-choice run (design.cue_mode == 'free'): it carries no label.

Source code in src/ssvep/io/manifest.py

def is_cued(manifest: dict) -> bool:
    """False only for a free-choice run (``design.cue_mode == 'free'``): it carries no label."""
    return cue_mode(manifest) != CUE_FREE

feedback_mode

feedback_mode(manifest)

The declared feedback mode ('none' when the run showed none).

Source code in src/ssvep/io/manifest.py

def feedback_mode(manifest: dict) -> str:
    """The declared feedback mode (``'none'`` when the run showed none)."""
    cfg = feedback_config(manifest)
    return cfg["mode"] if cfg else FEEDBACK_NONE

early_stop_config

early_stop_config(manifest)

The run's live early-stopping rule (design.early_stop, v1.8), or None for a run whose trials all last stim_s (an absent block: every earlier manifest).

Returns {statistic, method, min_window_s, max_window_s, step_s, confidence_margin, calibration}. max_window_s is not a manifest field: it is design.timing.stim_s (one fact, one place), read here so no consumer re-derives it. Every consumer (the runtime, the renderer, the analysis) asks this one function and never infers early stopping from short trials (AGENTS.md section 2.2). A resting or FPVS run has no decoder and reads None.

Source code in src/ssvep/io/manifest.py

def early_stop_config(manifest: dict) -> dict | None:
    """The run's live early-stopping rule (``design.early_stop``, v1.8), or ``None`` for a run whose
    trials all last ``stim_s`` (an absent block: every earlier manifest).

    Returns ``{statistic, method, min_window_s, max_window_s, step_s, confidence_margin,
    calibration}``. ``max_window_s`` is not a manifest field: it **is** ``design.timing.stim_s``
    (one fact, one place), read here so no consumer re-derives it. Every consumer (the runtime, the
    renderer, the analysis) asks this one function and never infers early stopping from short
    trials (AGENTS.md section 2.2). A resting or FPVS run has no decoder and reads ``None``.
    """
    if not has_decoder(manifest):
        return None
    design = (manifest or {}).get("design") or {}
    es = design.get("early_stop")
    if not es:
        return None
    timing = design.get("timing") or {}
    stim_s = timing.get("stim_s")
    missing = [k for k in ("min_window_s", "step_s", "confidence_margin") if es.get(k) is None]
    if missing:
        raise jsonschema.ValidationError(
            f"design.early_stop is missing {', '.join(missing)}: a live early-stopping rule needs "
            f"its first look, its step and its threshold")
    try:
        nums = {k: float(es[k]) for k in ("min_window_s", "step_s", "confidence_margin")}
        max_s = float(stim_s) if stim_s is not None else None
    except (TypeError, ValueError):
        raise jsonschema.ValidationError(
            "design.early_stop: min_window_s, step_s and confidence_margin (and "
            "design.timing.stim_s) must be numbers") from None
    return {"statistic": es.get("statistic"), "method": es.get("method"),
            "min_window_s": nums["min_window_s"], "max_window_s": max_s,
            "step_s": nums["step_s"], "confidence_margin": nums["confidence_margin"],
            "calibration": dict(es.get("calibration") or {})}

early_stop_schedule_problem

early_stop_schedule_problem(min_window_s, max_window_s, step_s, confidence_margin)

Why a live early-stopping schedule cannot mean anything, in one wording for the builder and the loader (the maximum window is design.timing.stim_s); None when it is sound. Beyond :func:~ssvep.classification.decoding.check_decision_schedule (which the offline replay also uses, and which allows a single look at the maximum), a live declaration needs its first look before the end of the stimulation: with only the last look it can never stop early, and is the fixed rule under the early-stop stamp.

Source code in src/ssvep/io/manifest.py

def early_stop_schedule_problem(min_window_s, max_window_s, step_s, confidence_margin) -> str | None:
    """Why a live early-stopping schedule cannot mean anything, in one wording for the builder and
    the loader (the maximum window is ``design.timing.stim_s``); ``None`` when it is sound. Beyond
    :func:`~ssvep.classification.decoding.check_decision_schedule` (which the offline replay also
    uses, and which allows a single look at the maximum), a live declaration needs its first look
    *before* the end of the stimulation: with only the last look it can never stop early, and is
    the fixed rule under the early-stop stamp."""
    from ..classification import decoding
    try:
        decoding.check_decision_schedule(min_window_s, max_window_s, step_s, confidence_margin)
    except (ValueError, TypeError) as exc:
        return f"{exc} (the maximum window is the stimulation duration, design.timing.stim_s)"
    if max_window_s is not None and min_window_s >= max_window_s:
        return (f"early stopping's first look (min_window_s {min_window_s:g} s) must come before "
                f"the end of the stimulation ({max_window_s:g} s): with only the last look a trial "
                f"can never stop early")
    return None

early_stop_source_problem

early_stop_source_problem(method, confidence_margin, source)

Why a noise_default source cannot carry this threshold, or None (#487). For a design being authored or used for a new recording, never for loading a recorded manifest. Only that source is judged: study may carry any threshold the schedule check accepts.

Source code in src/ssvep/io/manifest.py

def early_stop_source_problem(method, confidence_margin, source) -> str | None:
    """Why a ``noise_default`` source cannot carry this threshold, or ``None`` (#487). For a design
    being authored or used for a new recording, never for loading a recorded manifest. Only that
    source is judged: ``study`` may carry any threshold the schedule check accepts."""
    if source != EARLY_STOP_SOURCE_NOISE:
        return None
    from ..classification import decoding
    default = decoding.DEFAULT_CONFIDENCE_MARGINS.get(method)
    if default is None or abs(float(confidence_margin) - default) <= 1e-9:
        return None
    return (f"calibration.source 'noise_default' means the shipped {method} threshold "
            f"({default:g}), but confidence_margin is {confidence_margin:g}. Set source to "
            f"'study' and give its 'basis' (how the threshold was chosen)")

early_stop_offset_problem

early_stop_offset_problem(onset_offset_s)

Why design.timing.onset_offset_s cannot serve a live early-stopping run, or None.

The live window origin is stim_on plus the lead-in the schedule presents, so the field must be present and finite: without it the live decoder would wait for a lead-in that is never presented and could never decide at the maximum window.

Source code in src/ssvep/io/manifest.py

def early_stop_offset_problem(onset_offset_s) -> str | None:
    """Why ``design.timing.onset_offset_s`` cannot serve a live early-stopping run, or ``None``.

    The live window origin is ``stim_on`` plus the lead-in the schedule presents, so the field must
    be present and finite: without it the live decoder would wait for a lead-in that is never
    presented and could never decide at the maximum window.
    """
    ok = (isinstance(onset_offset_s, (int, float)) and not isinstance(onset_offset_s, bool)
          and math.isfinite(float(onset_offset_s)) and onset_offset_s >= 0)
    if ok:
        return None
    return ("early stopping needs design.timing.onset_offset_s, a finite lead-in of 0 s or more "
            f"(got {onset_offset_s!r}): the live window origin needs the lead-in the schedule "
            "presents")

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

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

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 {}

is_fpvs

is_fpvs(manifest)

True if manifest describes a fast periodic visual stimulation run (#494).

Source code in src/ssvep/io/manifest.py

def is_fpvs(manifest: dict) -> bool:
    """True if ``manifest`` describes a fast periodic visual stimulation run (#494)."""
    return paradigm(manifest) == FPVS

fpvs_params

fpvs_params(manifest)

design.fpvs for an FPVS manifest (the authoritative stream design); {} if absent.

Source code in src/ssvep/io/manifest.py

def fpvs_params(manifest: dict) -> dict:
    """``design.fpvs`` for an FPVS manifest (the authoritative stream design); ``{}`` if absent."""
    return ((manifest or {}).get("design", {}) or {}).get("fpvs", {}) or {}

has_decoder

has_decoder(manifest)

True only for the paradigm a live SSVEP decoder can answer (ssvep).

A resting run has nothing to decode, and an FPVS run's response is read offline from its spectrum (ssvep.analysis.fpvs), never by classifying windows: neither can have participant feedback, free choice or live early stopping, and neither is an SSVEP decoding contrast.

Source code in src/ssvep/io/manifest.py

def has_decoder(manifest: dict) -> bool:
    """True only for the paradigm a live SSVEP decoder can answer (``ssvep``).

    A resting run has nothing to decode, and an FPVS run's response is read offline from its
    spectrum (``ssvep.analysis.fpvs``), never by classifying windows: neither can have participant
    feedback, free choice or live early stopping, and neither is an SSVEP decoding contrast.
    """
    return paradigm(manifest) == SSVEP

carrier_hash

carrier_hash(acq)

Short content hash (12 hex) of the wiring a manifest's acquisition block records, or None when there is no montage to hash.

The wiring is what distinguishes two carriers that share a name: the amplifier channel and the 10-10 site of each montage entry, in order, plus the electrode type. Positions and measured impedance are deliberately left out (positions follow from the site; impedance changes every session). Computed from the block itself, never from a carrier file, so it can be re-checked anywhere the manifest travels (#410).

Source code in src/ssvep/io/manifest.py

def carrier_hash(acq: dict) -> str | None:
    """Short content hash (12 hex) of the wiring a manifest's ``acquisition`` block records, or
    ``None`` when there is no montage to hash.

    The wiring is what distinguishes two carriers that share a name: the amplifier channel and the
    10-10 site of each montage entry, in order, plus the electrode type. Positions and measured
    impedance are deliberately left out (positions follow from the site; impedance changes every
    session). Computed from the block itself, never from a carrier file, so it can be re-checked
    anywhere the manifest travels (#410).
    """
    acq = acq or {}
    montage = acq.get("montage") or []
    if not montage:
        return None
    wiring = {"electrode_type": acq.get("electrode_type") or None,
              "channels": [[m.get("channel"), m.get("label")] for m in montage]}
    blob = json.dumps(wiring, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:12]

carrier_hash_problem

carrier_hash_problem(acq)

A sentence when acq carries a carrier_hash its own wiring does not reproduce (the copy was edited by hand after it was written), else None. A block without the field (every stock carrier, and every manifest from before it existed) is never a problem.

Source code in src/ssvep/io/manifest.py

def carrier_hash_problem(acq: dict) -> str | None:
    """A sentence when ``acq`` carries a ``carrier_hash`` its own wiring does not reproduce (the
    copy was edited by hand after it was written), else ``None``. A block without the field
    (every stock carrier, and every manifest from before it existed) is never a problem."""
    stated = (acq or {}).get("carrier_hash")
    if stated is None:
        return None
    actual = carrier_hash(acq)
    if stated == actual:
        return None
    return (f"acquisition.carrier_hash is {stated!r} but the carrier copy in this block (headset "
            f"{acq.get('headset')!r}) hashes to {actual!r}: its wiring or electrode type was changed "
            f"after it was written. Re-pick the carrier instead of editing the copy by hand.")

unknown_carrier_problem

unknown_carrier_problem(acq)

A sentence when acq names a carrier that is neither built in nor copied into the block.

A user-defined carrier is copied whole into acquisition.montage when it is picked, so the copy is what makes a non-built-in name usable. A name with no copy (a retired built-in, a typo, a hand-written protocol) has no wiring anywhere this manifest goes, and is refused out loud rather than set up on a substitute montage. A block without a headset is never a problem.

Source code in src/ssvep/io/manifest.py

def unknown_carrier_problem(acq: dict) -> str | None:
    """A sentence when ``acq`` names a carrier that is neither built in nor copied into the block.

    A user-defined carrier is copied whole into ``acquisition.montage`` when it is picked, so the
    copy is what makes a non-built-in name usable. A name with no copy (a retired built-in, a typo,
    a hand-written protocol) has no wiring anywhere this manifest goes, and is refused out loud
    rather than set up on a substitute montage. A block without a headset is never a problem."""
    acq = acq or {}
    name = acq.get("headset")
    if not name or acq.get("montage"):
        return None
    from ..stim import builder              # here, not at import: the BIDS side imports this module
    if builder.canonical_headset(name) in builder.HEADSETS:
        return None
    return builder.unknown_headset_message(name) + "."

carrier_label

carrier_label(acq)

The headset name, followed by its wiring hash in brackets when the block carries one.

Source code in src/ssvep/io/manifest.py

def carrier_label(acq: dict) -> str:
    """The headset name, followed by its wiring hash in brackets when the block carries one."""
    acq = acq or {}
    name = acq.get("headset") or "-"
    return f"{name} [{acq['carrier_hash']}]" if acq.get("carrier_hash") else name

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. A picture on a non-image fill gets a message that names the problem (:func:_explain_image_fill_conflicts) rather than the schema's own.

Source code in src/ssvep/io/manifest.py

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.
    A picture on a non-image fill gets a message that names the problem
    (:func:`_explain_image_fill_conflicts`) rather than the schema's own.
    """
    _explain_image_fill_conflicts(manifest)
    jsonschema.validate(instance=canonicalize_manifest(manifest), schema=manifest_schema())
    # A hand-edited carrier copy is refused, not warned about: the hash exists so that a wiring
    # change cannot pass unnoticed, and a recording made on an altered copy cannot be un-made.
    problem = carrier_hash_problem((canonicalize_manifest(manifest) or {}).get("acquisition"))
    if problem:
        raise jsonschema.ValidationError(problem)
    problem = unknown_carrier_problem((canonicalize_manifest(manifest) or {}).get("acquisition"))
    if problem:
        raise jsonschema.ValidationError(problem)
    # Early stopping (v1.8): a schedule the schema cannot judge alone (its maximum is stim_s).
    problem = _early_stop_problem(canonicalize_manifest(manifest))
    if problem:
        raise jsonschema.ValidationError(problem)
    # FPVS (v1.9): frame arithmetic at the declared refresh and the realised sequence's rules.
    problem = _fpvs_problem(canonicalize_manifest(manifest))
    if problem:
        raise jsonschema.ValidationError(problem)

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

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

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

@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
    # (stem, picture path) for each picture an image-target run showed that is not in the BIDS
    # view's stimuli/ — its sourcedata copy is missing or not the pinned file (#258)
    stimuli_missing: list[tuple[str, str]] = field(default_factory=list)
    # what declaring the dataset's pilot / development sessions decided (#161): a hand-set group kept
    # over an all-pilot participant, a converter-set 'pilot' cleared, a session with mixed runs
    pilot_notes: list[str] = field(default_factory=list)
    # (stem, why) for each run its data-use stamp does not let this machine convert (#286 P3d): no
    # stamp, a malformed one, or only the older consent record. Never converted, never analysed.
    refused: list[tuple[str, str]] = field(default_factory=list)
    # (stem, why) for each run saved with no EEG at all (#482): markers and sidecar only, kept so the
    # run is on record, but there is no signal to put in a BIDS EEG file. Skipped on purpose, not a
    # failure, and named with the sidecar's own reason.
    no_eeg: list[tuple[str, str]] = field(default_factory=list)

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

ConversionRefused

Bases: ValueError

A run its data-use stamp does not let this machine convert. The message names why.

Source code in src/ssvep/io/bids.py

class ConversionRefused(ValueError):
    """A run its data-use stamp does not let this machine convert. The message names why."""

NoEEGRun

Bases: ValueError

A run saved with no EEG samples (#482): there is nothing to write as BIDS EEG.

Source code in src/ssvep/io/bids.py

class NoEEGRun(ValueError):
    """A run saved with no EEG samples (#482): there is nothing to write as BIDS EEG."""

NotStudy dataclass

Why a session (or one recording of it) is not study data: :data:NOT_STUDY_PILOT, :data:NOT_STUDY_DECLARED (group is the declared group) or :data:NOT_STUDY_UNREADABLE.

Source code in src/ssvep/io/bids.py

@dataclass(frozen=True)
class NotStudy:
    """Why a session (or one recording of it) is not study data: :data:`NOT_STUDY_PILOT`,
    :data:`NOT_STUDY_DECLARED` (``group`` is the declared group) or :data:`NOT_STUDY_UNREADABLE`."""
    kind: str
    detail: str
    group: str | None = None

NotStudySessions dataclass

Every session that is not study data, and why: the one rule export, the demographics fill, session_type and every contrast read (#400), built by :func:not_study_sessions.

locked: {sub: {ses}}, sessions any of whose sidecars was locked as a pilot / development session (:func:is_pilot_session). declared: {sub: {ses: group}}, the group pilot / test a session's declarations give it after it was recorded (#384). unreadable: {sub: {ses: why}}, a session whose declarations cannot be read (they may say pilot). dataset: why no declaration could be read at all, which holds back every session. seen: every (sub, ses) with a recording or a session folder.

Source code in src/ssvep/io/bids.py

@dataclass
class NotStudySessions:
    """Every session that is not study data, and why: **the one rule** export, the demographics
    fill, ``session_type`` and every contrast read (#400), built by :func:`not_study_sessions`.

    ``locked``: ``{sub: {ses}}``, sessions any of whose sidecars was locked as a pilot / development
    session (:func:`is_pilot_session`). ``declared``: ``{sub: {ses: group}}``, the group
    ``pilot`` / ``test`` a session's declarations give it after it was recorded (#384).
    ``unreadable``: ``{sub: {ses: why}}``, a session whose declarations cannot be read (they may say
    pilot). ``dataset``: why no declaration could be read at all, which holds back every session.
    ``seen``: every ``(sub, ses)`` with a recording or a session folder.
    """
    locked: dict[str, set[str]] = field(default_factory=dict)
    declared: dict[str, dict[str, str]] = field(default_factory=dict)
    unreadable: dict[str, dict[str, str]] = field(default_factory=dict)
    dataset: str | None = None
    seen: set[tuple[str, str]] = field(default_factory=set)

    def why(self, sub: str, ses: str, sidecar=_SESSION) -> NotStudy | None:
        """Why ``sub``/``ses`` is not study data, or ``None`` when it is.

        Without ``sidecar``, the session: locked as pilot when **any** of its sidecars says so
        (export, the demographics fill and ``session_type`` ask this). With ``sidecar``, one
        recording of it: locked as pilot only when that sidecar says so (a contrast asks this, so a
        session locked more than once keeps its study recordings, #161). A declared group or an
        unreadable declarations file is about the whole session either way.
        """
        locked = (ses in self.locked.get(sub, ()) if sidecar is _SESSION
                  else is_pilot_session(sidecar))
        if locked:
            return NotStudy(NOT_STUDY_PILOT, "locked as a pilot / development session")
        group = self.declared.get(sub, {}).get(ses)
        if group:
            return NotStudy(NOT_STUDY_DECLARED, f"declared group {group} after it was recorded",
                            group=group)
        why = self.unreadable.get(sub, {}).get(ses) or self.dataset
        if why:
            return NotStudy(NOT_STUDY_UNREADABLE, why)
        return None

    def sessions(self) -> dict[str, set[str]]:
        """``{sub: {ses}}`` for every session held back, for any of the three reasons."""
        out: dict[str, set[str]] = {}
        for sub, sess in self.locked.items():
            out.setdefault(sub, set()).update(sess)
        for table in (self.declared, self.unreadable):
            for sub, by_ses in table.items():
                out.setdefault(sub, set()).update(by_ses)
        if self.dataset:
            for sub, ses in self.seen:
                out.setdefault(sub, set()).add(ses)
        return out

    def keys(self) -> dict[tuple[str, str], NotStudy]:
        """``{(sub, ses): why}`` for every session held back."""
        return {(sub, ses): self.why(sub, ses)
                for sub, sess in self.sessions().items() for ses in sess}

why

why(sub, ses, sidecar=_SESSION)

Why sub/ses is not study data, or None when it is.

Without sidecar, the session: locked as pilot when any of its sidecars says so (export, the demographics fill and session_type ask this). With sidecar, one recording of it: locked as pilot only when that sidecar says so (a contrast asks this, so a session locked more than once keeps its study recordings, #161). A declared group or an unreadable declarations file is about the whole session either way.

Source code in src/ssvep/io/bids.py

def why(self, sub: str, ses: str, sidecar=_SESSION) -> NotStudy | None:
    """Why ``sub``/``ses`` is not study data, or ``None`` when it is.

    Without ``sidecar``, the session: locked as pilot when **any** of its sidecars says so
    (export, the demographics fill and ``session_type`` ask this). With ``sidecar``, one
    recording of it: locked as pilot only when that sidecar says so (a contrast asks this, so a
    session locked more than once keeps its study recordings, #161). A declared group or an
    unreadable declarations file is about the whole session either way.
    """
    locked = (ses in self.locked.get(sub, ()) if sidecar is _SESSION
              else is_pilot_session(sidecar))
    if locked:
        return NotStudy(NOT_STUDY_PILOT, "locked as a pilot / development session")
    group = self.declared.get(sub, {}).get(ses)
    if group:
        return NotStudy(NOT_STUDY_DECLARED, f"declared group {group} after it was recorded",
                        group=group)
    why = self.unreadable.get(sub, {}).get(ses) or self.dataset
    if why:
        return NotStudy(NOT_STUDY_UNREADABLE, why)
    return None

sessions

sessions()

{sub: {ses}} for every session held back, for any of the three reasons.

Source code in src/ssvep/io/bids.py

def sessions(self) -> dict[str, set[str]]:
    """``{sub: {ses}}`` for every session held back, for any of the three reasons."""
    out: dict[str, set[str]] = {}
    for sub, sess in self.locked.items():
        out.setdefault(sub, set()).update(sess)
    for table in (self.declared, self.unreadable):
        for sub, by_ses in table.items():
            out.setdefault(sub, set()).update(by_ses)
    if self.dataset:
        for sub, ses in self.seen:
            out.setdefault(sub, set()).add(ses)
    return out

keys

keys()

{(sub, ses): why} for every session held back.

Source code in src/ssvep/io/bids.py

def keys(self) -> dict[tuple[str, str], NotStudy]:
    """``{(sub, ses): why}`` for every session held back."""
    return {(sub, ses): self.why(sub, ses)
            for sub, sess in self.sessions().items() for ses in sess}

load_session

load_session(session_json_path)

Load a *.session.json (manifest + provenance + data-use stamp + plan + impedance).

Source code in src/ssvep/io/bids.py

def load_session(session_json_path) -> dict:
    """Load a ``*.session.json`` (manifest + provenance + data-use stamp + plan + impedance)."""
    with open(session_json_path, encoding="utf-8-sig") 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

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 = []
    has_feedback = any(d.get("segment") == _FEEDBACK for _i, d in parsed)
    has_free = any(d.get("cued") is False for _i, d in parsed)
    # Read from the markers' own fields, which only an early-stopping run's stimulus window writes.
    has_stop = any(d.get("segment") == _STIM_OFF and d.get("stopped") is not None
                   for _i, d in parsed)
    # An FPVS run's oddball / item markers name the item shown (#494); the column exists only then.
    has_item = any(d.get("item") is not None for _i, d in parsed)
    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),
            **({"shown": d.get("shown", NA), "latency_s": d.get("latency_s", NA),
                "shown_target_id": d.get("shown_target_id", NA)}
               if has_feedback else {}),
            **({"cued": "false" if d.get("cued") is False else NA} if has_free else {}),
            **({k: (d[k] if seg == _STIM_OFF and d.get(k) is not None else NA)
                for k in _EARLY_STOP_COLUMNS} if has_stop else {}),
            **({"item": d.get("item", NA)} if has_item else {}),
        })
    # A run that showed participant feedback (#268) carries what was shown, per feedback row; the
    # columns exist only then, so every other run's events table is exactly what it was.
    df = pd.DataFrame(rows, columns=_EVENT_COLUMNS + (_FEEDBACK_COLUMNS if has_feedback else [])
                      + (_FREE_COLUMNS if has_free else [])
                      + (_EARLY_STOP_COLUMNS if has_stop else [])
                      + (_FPVS_COLUMNS if has_item else []))
    return df.sort_values(["sample"], kind="stable").reset_index(drop=True)

measurement_date

measurement_date(session)

The run's BIDS meas_date (and so scans.tsv acq_time): its date-shifted start, in UTC — or None, which BIDS records as n/a.

None in three cases, all on purpose (#286):

• Not shifted. A sidecar written before date-shifting carries the real date. The converter can't shift it (the offset it would need was never kept, by design) and won't copy it, so the BIDS view simply has no date. It never writes a real one.
• No timezone. Those same old sidecars stored local wall-clock time with no offset, and this used to tag it UTC without converting it — the time was then off by the site's UTC offset. A naive time names no instant, so it is left out rather than guessed at. Everything the toolbox writes now is shifted and in UTC.
• Before MNE's floor. MNE stores meas_date as signed 32-bit seconds, so a date before 1901-12-13T20:45:52Z would convert and then break every later raw.save(). The toolbox's own shift window starts in 1902, so only a hand-edited sidecar gets here.

Source code in src/ssvep/io/bids.py

def measurement_date(session: dict):
    """The run's BIDS ``meas_date`` (and so ``scans.tsv`` ``acq_time``): its date-shifted start, in
    UTC — or ``None``, which BIDS records as ``n/a``.

    ``None`` in three cases, all on purpose (#286):

    * **Not shifted.** A sidecar written before date-shifting carries the real date. The converter
      can't shift it (the offset it would need was never kept, by design) and won't copy it, so the
      BIDS view simply has no date. It never writes a real one.
    * **No timezone.** Those same old sidecars stored local wall-clock time with no offset, and this
      used to tag it UTC without converting it — the time was then off by the site's UTC offset. A
      naive time names no instant, so it is left out rather than guessed at. Everything the toolbox
      writes now is shifted and in UTC.
    * **Before MNE's floor.** MNE stores ``meas_date`` as signed 32-bit seconds, so a date before
      1901-12-13T20:45:52Z would convert and then break every later ``raw.save()``. The toolbox's
      own shift window starts in 1902, so only a hand-edited sidecar gets here.
    """
    started = (session.get("provenance") or {}).get("started_at")
    if not _dates.is_shifted(started):
        return None
    meas = _dates.to_utc(started)
    if meas is None or meas < MNE_MEAS_DATE_MIN:
        return None
    return meas

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

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; wire-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

    meas_date = measurement_date(session)
    if meas_date is not None:
        raw.set_meas_date(meas_date)

    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: a resting run once 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 (AGENTS.md §2).

Source code in src/ssvep/io/bids.py

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: a resting run once 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 (AGENTS.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

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 (AGENTS.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

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 (AGENTS.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

no_eeg_reason

no_eeg_reason(session)

Why this recording has no EEG to convert or analyse, or None (#482).

Read from the sidecar the recorder wrote (provenance.eeg_samples == 0, with its incomplete_reason), never inferred from the XDF: the manifest spine says what the run is. A run that is merely short still has EEG, and still converts.

Source code in src/ssvep/io/bids.py

def no_eeg_reason(session) -> str | None:
    """Why this recording has no EEG to convert or analyse, or ``None`` (#482).

    Read from the sidecar the recorder wrote (``provenance.eeg_samples == 0``, with its
    ``incomplete_reason``), never inferred from the XDF: the manifest spine says what the run is. A
    run that is merely short still has EEG, and still converts.
    """
    prov = (session or {}).get("provenance") or {}
    if prov.get("eeg_samples") != 0:
        return None
    why = prov.get("incomplete_reason") or "no EEG samples were received"
    return f"markers only, no EEG ({why})"

active_kind

active_kind(profile=None)

This machine's site-profile kind, for :func:ssvep.runtime.data_use.data_use_of.

It decides only whether an unstamped recording counts as teaching (#286 Q1) and which purpose a contrast pools. A profile that cannot be read falls back to generic: never teaching, so the fallback can only refuse more, never infer a purpose nobody stamped.

Source code in src/ssvep/io/bids.py

def active_kind(profile=None) -> str:
    """This machine's site-profile kind, for :func:`ssvep.runtime.data_use.data_use_of`.

    It decides only whether an unstamped recording counts as teaching (#286 Q1) and which purpose a
    contrast pools. A profile that cannot be read falls back to ``generic``: never ``teaching``, so
    the fallback can only refuse more, never infer a purpose nobody stamped.
    """
    try:
        return _site.resolve(profile).kind
    except _site.SiteProfileError:
        return "generic"

conversion_refusal

conversion_refusal(session, *, kind=None)

Why this recording (its *.session.json) may not be converted here, or None.

The one matrix's CONVERT column (:data:ssvep.runtime.data_use.MATRIX), read through the fail-closed reader: every readable stamp with something saved may be converted (research from tier 1, pilot, teaching, bench), and tier 0 never has anything to convert, since nothing was written. A recording with no stamp, a malformed one, or only the older three-flag consent record is refused: there is no migration (#286 Q2). The one exception is #286 Q1: on a teaching machine an unstamped recording that does not say it was made on a lab machine counts as teaching, which may be converted and analysed locally (and is never pushed or exported). kind is the machine's kind (:func:active_kind when None).

Source code in src/ssvep/io/bids.py

def conversion_refusal(session, *, kind: str | None = None) -> str | None:
    """Why this recording (its ``*.session.json``) may not be converted here, or ``None``.

    The one matrix's CONVERT column (:data:`ssvep.runtime.data_use.MATRIX`), read through the
    fail-closed reader: every readable stamp with something saved may be converted (research from
    tier 1, pilot, teaching, bench), and tier 0 never has anything to convert, since nothing was
    written. A recording with no stamp, a malformed one, or only the older three-flag consent record
    is refused: there is no migration (#286 Q2). The one exception is #286 Q1: on a teaching machine
    an unstamped recording that does not say it was made on a lab machine counts as teaching, which
    may be converted and analysed locally (and is never pushed or exported). ``kind`` is the
    machine's kind (:func:`active_kind` when ``None``).
    """
    kind = active_kind() if kind is None else kind
    return _du.refusal(_du.Action.CONVERT, _du.data_use_of(session, kind))

recording_purpose

recording_purpose(session, *, kind=None)

The purpose a recording's sidecar declares, for labelling (session_type, SSVEPPurposes), never for granting anything: pilot when it is declared a pilot / development session (:func:is_pilot_session, which also reads the older session_position.pilot), else its stamp's purpose as the fail-closed reader reads it (including the #286 Q1 teaching inference), else :data:PURPOSE_UNSTAMPED.

Source code in src/ssvep/io/bids.py

def recording_purpose(session, *, kind: str | None = None) -> str:
    """The purpose a recording's sidecar declares, for labelling (``session_type``,
    ``SSVEPPurposes``), never for granting anything: ``pilot`` when it is declared a pilot /
    development session (:func:`is_pilot_session`, which also reads the older
    ``session_position.pilot``), else its stamp's purpose as the fail-closed reader reads it
    (including the #286 Q1 teaching inference), else :data:`PURPOSE_UNSTAMPED`."""
    if is_pilot_session(session):
        return _du.Purpose.PILOT.value
    reading = _du.data_use_of(session, active_kind() if kind is None else kind)
    return reading.record.purpose if reading.record is not None else PURPOSE_UNSTAMPED

stim_file_column

stim_file_column(events_df, stim)

events.tsv:stim_file for each row, from {stimulus id: stim_file} (:func:_bids_stimuli).

A row whose cued target is an image target names its picture. An FPVS oddball/item row (#494) names the item it marks: that item's picture is the image stimulus "item <id>" (:func:ssvep.stim.fpvs.item_image_stimuli). Everything else is n/a.

Source code in src/ssvep/io/bids.py

def stim_file_column(events_df, stim: dict) -> list:
    """``events.tsv:stim_file`` for each row, from ``{stimulus id: stim_file}`` (:func:`_bids_stimuli`).

    A row whose cued target is an image target names its picture. An FPVS ``oddball``/``item`` row
    (#494) names the item it marks: that item's picture is the image stimulus ``"item <id>"``
    (:func:`ssvep.stim.fpvs.item_image_stimuli`). Everything else is ``n/a``.
    """
    items = events_df["item"] if "item" in events_df.columns else [NA] * len(events_df)
    out = []
    for t, i in zip(events_df["target_id"], items):
        if isinstance(i, str) and i not in ("", NA):
            out.append(stim.get(f"item {i}", NA))
        else:
            out.append(stim.get(str(t), NA) if t != NA else NA)
    return out

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.

A run its data-use stamp does not let this machine convert (:func:conversion_refusal) is recorded in :attr:ConversionReport.refused with the reason, and is neither converted nor counted as present, whatever is already on disk.

Source code in src/ssvep/io/bids.py

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.

    A run its data-use stamp does not let this machine convert (:func:`conversion_refusal`) is
    recorded in :attr:`ConversionReport.refused` with the reason, and is neither converted nor
    counted as present, whatever is already on disk.
    """
    bids_root = Path(bids_root)
    refs = list(refs)
    report = ConversionReport()
    todo = []
    kind = active_kind()
    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
        why = conversion_refusal(session, kind=kind)
        if why is not None:
            report.refused.append((_stem(ref), why + _stale_view_note(ref, bids_root, session)))
            continue
        why = no_eeg_reason(session)
        if why is not None:
            report.no_eeg.append((_stem(ref), why))
            continue
        if is_converted(ref, bids_root, session=session):
            report.present.append(_stem(ref))
            _note_if_eventless(ref, bids_root, session, report)
            _note_missing_stimuli(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))
            converted = load_session(ref.session_json)
            _note_if_eventless(ref, bids_root, session=converted, report=report)
            _note_missing_stimuli(ref, bids_root, converted, 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)
        report.pilot_notes = enrich_participants(bids_root)
    return report

profile_mismatch

profile_mismatch(session, profile=None)

Why this run must not be converted under the active profile, or None (#298).

A run recorded under a configured deployment (its sidecar's site_profile.kind is not generic) and converted by a process whose active profile is generic would be stamped with no institution, no governance and no licence — silently, and undetectably afterwards. That is what happens when the converter runs in a process that could not see SSVEP_SITE_PROFILE (another user, a scheduled task, a shell opened before it was set). Sidecars from before the stamp existed carry no site_profile and are not second-guessed.

Source code in src/ssvep/io/bids.py

def profile_mismatch(session: dict, profile=None) -> str | None:
    """Why this run must not be converted under the active profile, or ``None`` (#298).

    A run recorded under a configured deployment (its sidecar's ``site_profile.kind`` is not
    ``generic``) and converted by a process whose active profile *is* generic would be stamped with
    no institution, no governance and no licence — silently, and undetectably afterwards. That is
    what happens when the converter runs in a process that could not see ``SSVEP_SITE_PROFILE``
    (another user, a scheduled task, a shell opened before it was set). Sidecars from before the
    stamp existed carry no ``site_profile`` and are not second-guessed.
    """
    recorded = (session or {}).get("site_profile") or {}
    kind = recorded.get("kind") or ""
    if not kind or kind == "generic":
        return None
    active = _site.resolve(profile)
    if not active.is_generic:
        return None
    label = recorded.get("label") or "(no label)"
    return (f"recorded under the {kind} profile '{label}', but this conversion is running under "
            f"the generic profile ({active.source or 'no site profile visible to this process'}) and "
            f"would stamp it with no institution or governance. Run it where that deployment's "
            f"profile is visible — set {_site.PROFILE_ENV} for this process (docs/SITE_PROFILE.md "
            f"§10) — and convert again.")

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. A run recorded under a configured deployment is refused under the generic profile (:func:profile_mismatch), and a run its data-use stamp does not let this machine convert raises :class:ConversionRefused (:func:conversion_refusal) before anything is written.

Source code in src/ssvep/io/bids.py

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``.
    A run recorded under a configured deployment is refused under the generic profile
    (:func:`profile_mismatch`), and a run its data-use stamp does not let this machine convert
    raises :class:`ConversionRefused` (:func:`conversion_refusal`) before anything is written.
    """
    session = load_session(ref.session_json)
    why = profile_mismatch(session)
    if why:
        raise _site.SiteProfileError(f"{Path(ref.xdf).name}: {why}")
    why = conversion_refusal(session)
    if why:
        raise ConversionRefused(f"{Path(ref.xdf).name}: not converted: {why}")
    why = no_eeg_reason(session)
    if why:
        raise NoEEGRun(f"{Path(ref.xdf).name}: not converted: {why}")
    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")

    # Image targets (#258): the pictures go to the dataset-root stimuli/, and each row whose cued
    # target is an image target names its picture. A run without them gets no stim_file column.
    stim = _bids_stimuli(ref, bids_root, session)
    if stim is not None and len(events_df):
        events_df["stim_file"] = stim_file_column(events_df, stim)

    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,
                                      stim_file="stim_file" in events_df.columns,
                                      early_stop="stopped" in events_df.columns)

    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.

sub/ses come from the filename's entities, not the folder it sits in, and the two are not cross-checked: a recording filed in another participant's folder is that filename's. Readers keyed by folder (the beh/ demographics fill) agree with it on every correctly filed dataset only (#372 review).

Source code in src/ssvep/io/bids.py

def discover_runs(sourcedata_root: Path) -> list[RunRef]:
    """Find every ``*_eeg.xdf`` under ``sourcedata/`` and pair it with its ``*.session.json``.

    ``sub``/``ses`` come from the **filename's** entities, not the folder it sits in, and the two
    are not cross-checked: a recording filed in another participant's folder is that filename's.
    Readers keyed by folder (the ``beh/`` demographics fill) agree with it on every correctly filed
    dataset only (#372 review)."""
    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

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.

SSVEPPurposes (:data:PURPOSES_KEY) lists the purposes the dataset's recordings were made for (:func:dataset_purposes), so a reader of the dataset alone can tell it is not all research. SSVEPStudy (:data:STUDY_KEY) names its study, when it has one (#396).

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

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.

    ``SSVEPPurposes`` (:data:`PURPOSES_KEY`) lists the purposes the dataset's recordings were made
    for (:func:`dataset_purposes`), so a reader of the dataset alone can tell it is not all research.
    ``SSVEPStudy`` (:data:`STUDY_KEY`) names its study, when it has one (#396).

    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)
    carried = {GROUP_RULE_KEY: GROUP_RULE_VERSION} if _group_rule_applied(bids_root) else {}
    ours = _participants.converter_pilot_groups(bids_root)
    if ours:          # which 'pilot' groups the converter set itself, so it can clear its own (#161)
        carried[_participants.PILOT_GROUP_KEY] = sorted(ours)
    if _is_synthetic(bids_root):
        # The synthetic dataset's marker (#269) is set once, by the generator; re-rendering must
        # keep it, or every contrast on it stops pooling its bench recordings (#286 P3d). Carried
        # only when already truthy: a re-render never makes a dataset synthetic.
        carried[SYNTHETIC_KEY] = True

    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)."}],
        # Which purposes the recordings were made for (#286 P3d): sorted, no dates, no identifiers.
        PURPOSES_KEY: dataset_purposes(bids_root, kind=active_kind(profile)),
        # Which study the dataset is (#396), beside its purposes: absent for the implicit study.
        **dataset_study(bids_root, profile),
        **carried,     # re-rendering must not undo the one-time group rewrite (enrich_participants)
    }
    (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")

    # Merge, never overwrite: a hand-documented custom column (e.g. a between-subject contrast
    # factor's Description/Levels) must survive every re-conversion — see
    # ssvep.io.participants.render_participants_json for why the merge happens here, from disk.
    rendered_participants = _participants.render_participants_json(bids_root, _PARTICIPANTS_JSON)
    (bids_root / "participants.json").write_text(
        json.dumps(rendered_participants, 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)

Fill species/group and age_range/gender in the mne-bids-generated participants.tsv, and declare the dataset's pilot / development sessions. Returns notes worth reporting.

No demographics are invented — unknown fields stay n/a (COMPLIANCE: coarse-only, no linkage). group is a declaration, never overwritten; see :mod:ssvep.io.participants for its rule, including the one-time rewrite of the pilot stamp this function wrote on every conversion before #269. A between-subject factor declared with the recordings (sourcedata/sub-XXX/ses-YYY/declarations/, #276) is rendered into an undeclared cell; a disagreement is kept, reported, and excluded from contrasts on that factor (:func:ssvep.io.participants.apply_declarations). A group declared there binds only its own session and is never rendered. age_range/gender (#290) are derived from the background questionnaire's beh/ answers, when any exist — :func:~ssvep.io.participants.resolved_demographics has the multi-session rule.

This is the one function every conversion path calls — :func:convert_dataset, :func:ensure_converted, and Analyze's batch (ssvep.analysis.batch) — so they declare pilot sessions identically (#161). It reads every recording under sourcedata/, not only the ones just converted: whether every recording of a participant is pilot is a fact about all of them. Each recording declared pilot (its data-use stamp's purpose, or the legacy session_position.pilot, read only as a restriction) becomes session_type = pilot in sub-XXX_sessions.tsv (:func:write_session_types), and group becomes pilot only when every recording of the participant is pilot and no group is declared (:func:ssvep.io.participants.apply_pilot_groups). Neither is what keeps pilot data out of a contrast: the contrast reads each recording's own sidecar.

Every session's session_type is its recordings' purpose (#286 P3d), combined with a group its declarations give it afterwards (#400, :func:session_types). The demographics fill skips every session :func:not_study_sessions holds back — locked as pilot, declared pilot / test afterwards, or with declarations that cannot be read — the same set export holds back.

Source code in src/ssvep/io/bids.py

def enrich_participants(bids_root: Path) -> list[str]:
    """Fill ``species``/``group`` and ``age_range``/``gender`` in the mne-bids-generated
    ``participants.tsv``, and declare the dataset's pilot / development sessions. Returns notes worth
    reporting.

    No demographics are invented — unknown fields stay ``n/a`` (COMPLIANCE: coarse-only, no
    linkage). ``group`` is a **declaration**, never overwritten; see :mod:`ssvep.io.participants` for
    its rule, including the one-time rewrite of the ``pilot`` stamp this function wrote on every
    conversion before #269. A between-subject factor declared **with the recordings**
    (``sourcedata/sub-XXX/ses-YYY/declarations/``, #276) is rendered into an undeclared cell; a
    disagreement is kept, reported, and excluded from contrasts on that factor
    (:func:`ssvep.io.participants.apply_declarations`). A ``group`` declared there binds only its
    own session and is never rendered. ``age_range``/``gender`` (#290) are derived from the background
    questionnaire's ``beh/`` answers, when any exist —
    :func:`~ssvep.io.participants.resolved_demographics` has the multi-session rule.

    This is the **one** function every conversion path calls — :func:`convert_dataset`,
    :func:`ensure_converted`, and Analyze's batch (``ssvep.analysis.batch``) — so they declare pilot
    sessions identically (#161). It reads every recording under ``sourcedata/``, not only the ones
    just converted: whether *every* recording of a participant is pilot is a fact about all of them.
    Each recording declared pilot (its data-use stamp's purpose, or the legacy
    ``session_position.pilot``, read only as a restriction) becomes ``session_type = pilot`` in
    ``sub-XXX_sessions.tsv`` (:func:`write_session_types`), and ``group`` becomes ``pilot`` only
    when every recording of the participant is pilot and no group is declared
    (:func:`ssvep.io.participants.apply_pilot_groups`). Neither is what keeps pilot data out of a
    contrast: the contrast reads each recording's own sidecar.

    Every session's ``session_type`` is its recordings' purpose (#286 P3d), combined with a group
    its declarations give it afterwards (#400, :func:`session_types`). The demographics fill skips
    every session :func:`not_study_sessions` holds back — locked as pilot, declared pilot / test
    afterwards, or with declarations that cannot be read — the same set export holds back.
    """
    bids_root = Path(bids_root)
    marked = _participants.apply_group_rule(bids_root)
    # Factor levels declared with the recordings (#276) are rendered next (a declared group binds
    # only its session and is never rendered). Like the pilot summary below, only once the dataset
    # is marked: the pre-#269 stamp must already have been rewritten. Notes are logged at the end.
    decl_notes = _participants.apply_declarations(bids_root) if marked else []
    refs = discover_runs(bids_root / "sourcedata")
    runs = pilot_runs(refs)
    held = not_study_sessions(bids_root, refs=refs)
    # Demographics need only participants.tsv, not the group-rule marker, so they run either way.
    # They are given the one "not study data" rule export, session_type and every contrast read, so
    # an answer from a pilot session, locked or declared, never labels a participant's study data
    # (#290 follow-up, #400).
    _participants.apply_demographics(bids_root, pilot_sessions=held.sessions())
    notes = decl_notes + _mixed_session_demographics_notes(bids_root, runs)
    if marked:
        notes += _participants.apply_pilot_groups(
            bids_root, {sub: [flag for flags in by_ses.values() for flag in flags]
                        for sub, by_ses in runs.items()})
        notes += write_session_types(bids_root, runs, purposes=purpose_runs(refs), held=held)
    notes += fpvs_order_notes(refs)
    for note in notes:
        log.warning("BIDS: %s", note)
    return notes

fpvs_order_notes

fpvs_order_notes(refs)

One note per FPVS recording whose design asks for a new item order each session (design.fpvs.realise: "per_session") but which has no per-session realisation record: a toolbox that could not draw the order recorded it, so the design's sample order was played, the same as in every other such recording (#530). Reported by every conversion, never passed off as a per-session order. A sidecar that cannot be read is left to the converter to report.

Source code in src/ssvep/io/bids.py

def fpvs_order_notes(refs) -> list[str]:
    """One note per FPVS recording whose design asks for a new item order each session
    (``design.fpvs.realise: "per_session"``) but which has no per-session ``realisation`` record:
    a toolbox that could not draw the order recorded it, so the design's sample order was played,
    the same as in every other such recording (#530). Reported by every conversion, never passed off
    as a per-session order. A sidecar that cannot be read is left to the converter to report."""
    from ..stim import fpvs as _fpvs
    notes = []
    for ref in refs:
        try:
            with open(ref.session_json, encoding="utf-8") as fh:
                manifest = (json.load(fh) or {}).get("manifest") or {}
        except (OSError, ValueError):
            continue
        if not isinstance(manifest, dict) or _paradigm(manifest) != "fpvs":
            continue
        order = _fpvs.order_source((manifest.get("design") or {}).get("fpvs") or {})
        if order["source"] == _fpvs.ORDER_DESIGN_FALLBACK:
            notes.append(f"{Path(ref.session_json).name}: {order['text']}.")
    return notes

is_pilot_session

is_pilot_session(session)

Whether a recording's sidecar (a run's *.session.json or an answer's _beh.json) declares it a pilot / development session. Declared at lock by the operator, never inferred.

Two declarations count, and either is enough, because this is read only to restrict (a contrast excludes the recording, the sessions table marks it, the demographics fill skips it, an export holds it back): its data-use stamp's purpose is pilot (#286 P3, the one source of truth since), or the older session_position.pilot is true (#161, no longer written). Not validated beyond that: a malformed stamp that says pilot still excludes, and every gate that grants something reads the stamp strictly (:func:ssvep.runtime.data_use.data_use_of). Strict about types: only the string "pilot" and the JSON true count, never "true" or 1.

Source code in src/ssvep/io/bids.py

def is_pilot_session(session: dict) -> bool:
    """Whether a recording's sidecar (a run's ``*.session.json`` or an answer's ``_beh.json``)
    declares it a pilot / development session. Declared at lock by the operator, never inferred.

    Two declarations count, and either is enough, because this is read only to **restrict** (a
    contrast excludes the recording, the sessions table marks it, the demographics fill skips it,
    an export holds it back): its data-use stamp's purpose is ``pilot`` (#286 P3, the one source of
    truth since), or the older ``session_position.pilot`` is ``true`` (#161, no longer written). Not
    validated beyond that: a malformed stamp that says ``pilot`` still excludes, and every gate that
    *grants* something reads the stamp strictly (:func:`ssvep.runtime.data_use.data_use_of`). Strict
    about types: only the string ``"pilot"`` and the JSON ``true`` count, never ``"true"`` or ``1``."""
    session = session if isinstance(session, dict) else {}
    stamp = session.get("data_use")
    if isinstance(stamp, dict) and stamp.get("purpose") == "pilot":
        return True
    pos = session.get("session_position")
    return isinstance(pos, dict) and pos.get("pilot") is True

pilot_runs

pilot_runs(refs)

{"sub-XXX": {"ses-YYY": [whether each run is a pilot / development session]}}.

A sidecar that cannot be read is left out: it declares nothing (and cannot be converted either).

Source code in src/ssvep/io/bids.py

def pilot_runs(refs) -> dict[str, dict[str, list[bool]]]:
    """``{"sub-XXX": {"ses-YYY": [whether each run is a pilot / development session]}}``.

    A sidecar that cannot be read is left out: it declares nothing (and cannot be converted either).
    """
    out: dict[str, dict[str, list[bool]]] = {}
    for ref in refs:
        try:
            flag = is_pilot_session(load_session(ref.session_json))
        except Exception:        # noqa: BLE001 — unreadable: no declaration to mirror
            continue
        out.setdefault(_entity(ref.sub, "sub"), {}).setdefault(_entity(ref.ses, "ses"), []).append(flag)
    return out

pilot_sessions

pilot_sessions(runs)

{"sub-XXX": {"ses-YYY", ...}}: the sessions :func:pilot_runs says are pilot / development sessions (#161) — a session is pilot when any of its recordings declares pilot (:func:is_pilot_session: the stamp's purpose, or the legacy session_position.pilot), the same test :func:write_session_types uses for session_type.

The one place that turns per-recording declarations into per-session ones, so the sessions table and the demographics fill (:func:ssvep.io.participants.resolved_demographics, #290 follow-up) cannot disagree about which sessions are pilot. A session with no readable recording declares nothing, so it is not in the result.

Source code in src/ssvep/io/bids.py

def pilot_sessions(runs: dict[str, dict[str, list[bool]]]) -> dict[str, set[str]]:
    """``{"sub-XXX": {"ses-YYY", ...}}``: the sessions :func:`pilot_runs` says are pilot /
    development sessions (#161) — a session is pilot when **any** of its recordings declares pilot
    (:func:`is_pilot_session`: the stamp's purpose, or the legacy ``session_position.pilot``), the same test :func:`write_session_types` uses for ``session_type``.

    The one place that turns per-recording declarations into per-session ones, so the sessions table
    and the demographics fill (:func:`ssvep.io.participants.resolved_demographics`, #290 follow-up)
    cannot disagree about which sessions are pilot. A session with no readable recording declares
    nothing, so it is not in the result.
    """
    return {sub: {ses for ses, flags in by_ses.items() if any(flags)} for sub, by_ses in runs.items()}

purpose_runs

purpose_runs(refs, *, kind=None)

{"sub-XXX": {"ses-YYY": [each run's purpose]}} (:func:recording_purpose), keyed like :func:pilot_runs. A sidecar that cannot be read is left out: it declares nothing.

Source code in src/ssvep/io/bids.py

def purpose_runs(refs, *, kind: str | None = None) -> dict[str, dict[str, list[str]]]:
    """``{"sub-XXX": {"ses-YYY": [each run's purpose]}}`` (:func:`recording_purpose`), keyed like
    :func:`pilot_runs`. A sidecar that cannot be read is left out: it declares nothing."""
    kind = active_kind() if kind is None else kind
    out: dict[str, dict[str, list[str]]] = {}
    for ref in refs:
        try:
            purpose = recording_purpose(load_session(ref.session_json), kind=kind)
        except Exception:        # noqa: BLE001 — unreadable: no declaration to mirror
            continue
        out.setdefault(_entity(ref.sub, "sub"), {}).setdefault(_entity(ref.ses, "ses"), []).append(purpose)
    return out

dataset_study

dataset_study(bids_root, profile=None)

{STUDY_KEY: id} for a study's root (:func:ssvep.runtime.study.root_study), {} for the implicit study — so a dataset with no study is rendered byte for byte as before. A reader of the dataset alone, without its marker or the site profile, can then tell which study it is.

Source code in src/ssvep/io/bids.py

def dataset_study(bids_root, profile=None) -> dict:
    """``{STUDY_KEY: id}`` for a study's root (:func:`ssvep.runtime.study.root_study`), ``{}`` for
    the implicit study — so a dataset with no study is rendered byte for byte as before. A reader
    of the dataset alone, without its marker or the site profile, can then tell which study it is.
    """
    from ..runtime import study as _study

    if bids_root is None:
        return {}
    sid = _study.root_study(bids_root, profile)
    return {STUDY_KEY: sid} if sid else {}

dataset_purposes

dataset_purposes(bids_root, *, refs=None, kind=None)

Sorted purposes of every recording under sourcedata/ (:func:recording_purpose), with :data:PURPOSE_UNSTAMPED among them when any recording has no readable stamp: what :data:PURPOSES_KEY says. Purpose words only: no dates, no codes, no counts.

Source code in src/ssvep/io/bids.py

def dataset_purposes(bids_root, *, refs=None, kind: str | None = None) -> list[str]:
    """Sorted purposes of every recording under ``sourcedata/`` (:func:`recording_purpose`), with
    :data:`PURPOSE_UNSTAMPED` among them when any recording has no readable stamp: what
    :data:`PURPOSES_KEY` says. Purpose words only: no dates, no codes, no counts."""
    refs = discover_runs(Path(bids_root) / "sourcedata") if refs is None else refs
    return sorted({p for by_ses in purpose_runs(refs, kind=kind).values()
                   for ps in by_ses.values() for p in ps})

not_study_sessions

not_study_sessions(bids_root, *, refs=None, resolved=None)

Every session under bids_root that is not study data, and why (:class:NotStudySessions).

The one "pilot / excluded session" rule (#400): export holds these sessions back under both purposes, the demographics fill skips their answers, session_type labels them, and a contrast leaves their recordings out. A session is held back when:

• it was locked as pilot: any of its recordings declares pilot (:func:is_pilot_session: the data-use stamp's purpose, or the older session_position.pilot, read only as a restriction). Read two ways, and either counts: each XDF's sidecar, keyed by the recording's file name (:func:pilot_runs), and every sidecar in the session's folder, keyed by the folder: each *.session.json (a rescued one, or one whose XDF is gone, included) and each questionnaire answer's *_beh.json, which carries the same stamp;
• it was declared pilot or test afterwards: its declarations file gives it a group (:mod:ssvep.io.declarations, #384);
• its declarations cannot be read, so they may say pilot (fail-closed), including when no declaration in the dataset can be read at all.

refs (from :func:discover_runs) and resolved (from :func:ssvep.io.declarations.resolve) save reading them again when a caller already has them.

Source code in src/ssvep/io/bids.py

def not_study_sessions(bids_root, *, refs=None, resolved=None) -> NotStudySessions:
    """Every session under ``bids_root`` that is not study data, and why (:class:`NotStudySessions`).

    **The one "pilot / excluded session" rule** (#400): export holds these sessions back under
    both purposes, the demographics fill skips their answers, ``session_type`` labels them, and a
    contrast leaves their recordings out. A session is held back when:

    * **it was locked as pilot**: any of its recordings declares pilot (:func:`is_pilot_session`:
      the data-use stamp's purpose, or the older ``session_position.pilot``, read only as a
      restriction). Read two ways, and either counts: each XDF's sidecar, keyed by the recording's
      file name (:func:`pilot_runs`), and every sidecar in the session's folder, keyed by the
      folder: each ``*.session.json`` (a rescued one, or one whose XDF is gone, included) and each
      questionnaire answer's ``*_beh.json``, which carries the same stamp;
    * **it was declared pilot or test afterwards**: its declarations file gives it a ``group``
      (:mod:`ssvep.io.declarations`, #384);
    * **its declarations cannot be read**, so they may say pilot (fail-closed), including when no
      declaration in the dataset can be read at all.

    ``refs`` (from :func:`discover_runs`) and ``resolved`` (from
    :func:`ssvep.io.declarations.resolve`) save reading them again when a caller already has them.
    """
    root = Path(bids_root)
    refs = discover_runs(root / "sourcedata") if refs is None else refs
    out = NotStudySessions()
    for ref in refs:
        out.seen.add((_entity(ref.sub, "sub"), _entity(ref.ses, "ses")))
    for sub, sess in pilot_sessions(pilot_runs(refs)).items():
        if sess:
            out.locked.setdefault(sub, set()).update(sess)
    for d in _session_folders(root):
        sub, ses = d.parent.name, d.name
        out.seen.add((sub, ses))
        for s in sorted([*(d / "eeg").glob("*.session.json"), *(d / "beh").glob("*_beh.json")]):
            if _declarations._is_link(s):
                continue
            try:
                side = json.loads(s.read_text(encoding="utf-8-sig"))
            except Exception:                     # noqa: BLE001 — unreadable: declares nothing
                continue
            if isinstance(side, dict) and is_pilot_session(side):
                out.locked.setdefault(sub, set()).add(ses)
    resolved = _declarations.resolve(root) if resolved is None else resolved
    out.declared = {sub: dict(by_ses) for sub, by_ses in resolved.session_groups.items()}
    out.unreadable = {sub: dict(by_ses) for sub, by_ses in resolved.session_problems.items()}
    for sub, problems in resolved.problems.items():
        if None not in problems:
            continue                              # a factor disagreement: not about pilot
        if sub == _declarations.DATASET:
            out.dataset = problems[None]
            continue
        for s, ses in sorted(out.seen):           # none of this participant's files can be read
            if s == sub:
                out.unreadable.setdefault(sub, {}).setdefault(ses, problems[None])
    return out

session_types

session_types(purposes, held=None)

{sub: {ses: session_type}} for every session in purposes (:func:purpose_runs) and every session held holds back, recording or not.

The recordings' purpose (any pilot recording makes the session pilot, as it does for export and the demographics fill; several other purposes are joined by +; a session with no known purpose is n/a), with _declared_<group> appended when the session's declarations give it a group afterwards (#400): a declared pilot is labelled, and the purpose it was locked with is kept beside the label. A held-back session with no recording (a pilot questionnaire answer saved before run 1, or only a declaration) is pilot or declared_<group>.

Source code in src/ssvep/io/bids.py

def session_types(purposes: dict[str, dict[str, list[str]]],
                  held: NotStudySessions | None = None) -> dict[str, dict[str, str]]:
    """``{sub: {ses: session_type}}`` for every session in ``purposes`` (:func:`purpose_runs`) and
    every session ``held`` holds back, recording or not.

    The recordings' purpose (any pilot recording makes the session ``pilot``, as it does for
    export and the demographics fill; several other purposes are joined by ``+``; a session with no
    known purpose is ``n/a``), with ``_declared_<group>`` appended when the session's declarations
    give it a group afterwards (#400): a declared pilot is labelled, and the purpose it was locked
    with is kept beside the label. A held-back session with no recording (a pilot questionnaire
    answer saved before run 1, or only a declaration) is ``pilot`` or ``declared_<group>``.
    """
    held_sessions = held.sessions() if held is not None else {}
    out: dict[str, dict[str, str]] = {}
    for sub in sorted(set(purposes) | set(held_sessions)):
        by_ses = purposes.get(sub, {})
        for ses in sorted(set(by_ses) | held_sessions.get(sub, set())):
            kinds = set(by_ses.get(ses, ()))
            locked = SESSION_TYPE_PILOT in kinds or (held is not None
                                                     and ses in held.locked.get(sub, ()))
            base = SESSION_TYPE_PILOT if locked else ("+".join(sorted(kinds)) or NA)
            group = held.declared.get(sub, {}).get(ses) if held is not None else None
            if group:
                base = (f"{base}{SESSION_TYPE_DECLARED}{group}" if base != NA
                        else f"{SESSION_TYPE_DECLARED.lstrip('_')}{group}")
            out.setdefault(sub, {})[ses] = base
    return out

write_session_types

write_session_types(bids_root, runs, *, purposes=None, held=None)

Write session_type into each converted participant's sub-XXX_sessions.tsv (#161,

286 P3d, #400).

session_type is derived on every conversion from the recordings and their declarations (:func:session_types): their stamped purpose, pilot for a session with a pilot / development recording, and _declared_<group> appended for a session declared pilot / test afterwards. purposes is :func:purpose_runs (None: only runs' pilot flags are known, so a non-pilot session is n/a) and held is :func:not_study_sessions. Written for every participant whose sub-XXX/ exists (converted). Every other column and row is kept; a session missing from the file gets a row. A different value already in the cell is replaced, and reported, except by n/a or unstamped: neither says anything was derived (an unstamped recording is not even converted here), so a hand-set value stays. Returns notes.

Source code in src/ssvep/io/bids.py

def write_session_types(bids_root, runs: dict[str, dict[str, list[bool]]], *,
                        purposes: dict[str, dict[str, list[str]]] | None = None,
                        held: NotStudySessions | None = None) -> list[str]:
    """Write ``session_type`` into each converted participant's ``sub-XXX_sessions.tsv`` (#161,
    #286 P3d, #400).

    ``session_type`` is derived on every conversion from the recordings and their declarations
    (:func:`session_types`): their stamped purpose, ``pilot`` for a session with a pilot /
    development recording, and ``_declared_<group>`` appended for a session declared pilot / test
    afterwards. ``purposes`` is :func:`purpose_runs` (``None``: only ``runs``' pilot flags are
    known, so a non-pilot session is ``n/a``) and ``held`` is :func:`not_study_sessions`. Written
    for every participant whose ``sub-XXX/`` exists (converted). Every other column and row is
    kept; a session missing from the file gets a row. A different value already in the cell is
    replaced, and reported, except by ``n/a`` or ``unstamped``: neither says anything was derived
    (an unstamped recording is not even converted here), so a hand-set value stays. Returns notes.
    """
    bids_root = Path(bids_root)
    notes: list[str] = []
    declared = pilot_sessions(runs)
    if purposes is None:
        purposes = {sub: {ses: ([SESSION_TYPE_PILOT] if any(flags) else []) for ses, flags in by_ses.items()}
                    for sub, by_ses in runs.items()}
    derived = session_types(purposes, held)
    for sub in sorted(set(runs) | set(derived)):
        by_ses = runs.get(sub, {})
        sub_dir = bids_root / sub
        tsv = sub_dir / f"{sub}_sessions.tsv"
        types = derived.get(sub, {})
        if not sub_dir.is_dir() or sub_dir.is_symlink():
            continue
        if all(t == NA for t in types.values()) and not tsv.is_file():
            continue                      # nothing known to say, and no table to keep in step
        for ses in sorted(declared.get(sub, ())):
            if not all(by_ses[ses]):
                notes.append(f"{sub} {ses}: some of its recordings are pilot / development sessions "
                             f"and some are not (it was locked more than once), so session_type is "
                             f"'pilot'. Every contrast still excludes only the pilot recordings.")
        if held is not None:
            for ses in sorted(set(by_ses) | set(types)):
                why = held.why(sub, ses)
                if why is not None and why.kind == NOT_STUDY_UNREADABLE:
                    notes.append(f"{sub} {ses}: its declarations cannot be read, so session_type "
                                 f"shows only its recordings' purpose; it is held back from export, "
                                 f"the demographics fill and every contrast ({why.detail})")
        header, rows = ["session_id"], []
        if tsv.is_file():
            try:
                with tsv.open(newline="", encoding="utf-8") as fh:
                    reader = csv.DictReader(fh, delimiter="\t")
                    rows = [dict(r) for r in reader]
                    header = list(reader.fieldnames or ["session_id"])
            except (OSError, UnicodeDecodeError, csv.Error) as exc:
                notes.append(f"{tsv.name} could not be read ({exc}), so session_type was not written")
                continue
            if "session_id" not in header:
                notes.append(f"{tsv.name} has no session_id column, so session_type was not written")
                continue
        if "session_type" not in header:
            header.append("session_type")
        have = {(r.get("session_id") or "").strip() for r in rows}
        rows += [{"session_id": ses} for ses in sorted(set(by_ses) | set(types)) if ses not in have]
        for r in rows:
            sid = (r.get("session_id") or "").strip()
            cur = (r.get("session_type") or "").strip() or NA
            new = types.get(sid, NA)
            if new == NA or (new == PURPOSE_UNSTAMPED and cur != NA):
                r["session_type"] = cur           # nothing derived for it: a hand-set value stays
                continue
            if cur not in (NA, new):
                notes.append(f"{sub} {sid}: session_type was '{cur}', but its recordings and "
                             f"declarations say '{new}', so it is now '{new}'")
            r["session_type"] = new
        header = ["session_id"] + [c for c in header if c != "session_id"]
        with tsv.open("w", newline="", encoding="utf-8") as fh:
            w = csv.DictWriter(fh, fieldnames=header, delimiter="\t", restval=NA,
                               extrasaction="ignore", lineterminator="\n")
            w.writeheader()
            w.writerows(rows)
        side = tsv.with_suffix(".json")
        try:
            existing = json.loads(side.read_text(encoding="utf-8-sig")) if side.is_file() else {}
        except (OSError, ValueError):
            existing = {}
        merged = dict(_SESSIONS_JSON)
        for key, value in (existing.items() if isinstance(existing, dict) else ()):
            if key != "session_type":
                merged.setdefault(key, value)   # a hand-documented column survives
        side.write_text(json.dumps(merged, indent=2) + "\n", encoding="utf-8")
    return notes

convert_dataset

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

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.

A run its data-use stamp does not let this machine convert (:func:conversion_refusal) is skipped, logged with its reason, and named in report.refused when a report is passed (the conversion scripts print it); the others are converted. Only a profile mismatch refuses the whole dataset (#298), because that one would stamp every run wrongly.

Source code in src/ssvep/io/bids.py

def convert_dataset(bids_root, task: str | None = None, *, migrate: bool = True,
                    report: ConversionReport | None = None) -> 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.

    A run its data-use stamp does not let this machine convert (:func:`conversion_refusal`) is
    skipped, logged with its reason, and named in ``report.refused`` when a ``report`` is passed
    (the conversion scripts print it); the others are converted. Only a profile mismatch refuses
    the whole dataset (#298), because that one would stamp every run wrongly.
    """
    bids_root = Path(bids_root)
    _site.check_data_root(bids_root)          # a declared-but-missing data root is an error (#298)
    if migrate:
        migrate_toplevel_to_sourcedata(bids_root)
    refs = discover_runs(bids_root / "sourcedata")
    # Every run is checked BEFORE any is converted (#298): refusing halfway would leave a tree
    # half-stamped by the right deployment and half by nobody.
    sessions = [load_session(ref.session_json) for ref in refs]
    refused = [(ref, why) for ref, session in zip(refs, sessions) if (why := profile_mismatch(session))]
    if refused:
        listing = "\n".join(f"  {Path(str(ref.xdf)).name}" for ref, _ in refused)
        raise _site.SiteProfileError(
            f"Not converting {bids_root}: {len(refused)} run(s) were {refused[0][1]}\n{listing}")
    report = ConversionReport() if report is None else report
    kind = active_kind()
    allowed = []
    for ref, session in zip(refs, sessions):
        why = conversion_refusal(session, kind=kind)
        if why is not None:
            report.refused.append((_stem(ref), why + _stale_view_note(ref, bids_root, session)))
            continue
        why = no_eeg_reason(session)
        if why is not None:
            report.no_eeg.append((_stem(ref), why))
            continue
        allowed.append(ref)
    written = [convert_run(ref, bids_root, task=task) for ref in allowed]
    report.converted += [_stem(ref) for ref in allowed]
    tasks = sorted({bp.task for bp in written}) or [task or DEFAULT_TASK]
    write_dataset_metadata(bids_root, tasks=tasks)
    report.pilot_notes = enrich_participants(bids_root)
    return written