nb-c2-clean-pipeline · Capstone C2 — Clean pipeline¶
Capstone C2 · Level 2 · Status draft — for expert review; uncertain points carry TODO(confirm).
Brief (spec §6, C2). A script that takes raw BIDS → cleaned continuous data + a QC HTML report per subject for one full spine dataset, driven entirely by a configuration file, running unattended over every subject, and producing for each: a bad-channel table, the ICA components removed with reasons, the percentage of data rejected per condition, the filter settings with their justification, and a run log with package versions and the seed. Rubric: runs unattended on all subjects; no condition-biased rejection; the QC report is readable by a stranger.
Dataset choice, and why. ds-erpcore P3 rather than ds-eegbci. It has two conditions, so the per-condition rejection table means something and the rubric's central check is testable; it has three EOG channels, so blink handling can be compared across methods (L2.6, L2.7); and it is the paradigm the rest of Level 2 and all of Level 3 use, so the pipeline that comes out of here is the one C3 reuses unchanged. ds-eegbci is the cheaper choice — smaller files, no download barrier — and its BIDS mirror (OpenNeuro ds004362, CC0) is the entry point nb-2-1-bids uses; pipelines/configs/eegbci.yaml runs the identical pipeline on it if you would rather.
Subset rule and runtime (§11). The documented subset is sub-001 … sub-010 — ten of the paradigm's forty participants, listed by ID below, about 560 MB of downloads on a machine with an empty cache. Setting FULL_COHORT = True in the configuration cell runs the identical code over all forty (~2.2 GB of downloads; local runs only). The per-subject function does not change; only the number of rows in the group table does.
What it produces. Per subject: a cleaned continuous recording, the surviving epochs, a machine-readable run log and a self-contained QC HTML page. Across subjects: a cohort index, a group summary table, a condition-bias check, and a methods paragraph. Nothing is written into the repository; the output directory is a temporary folder by default and EEG_COURSE_C2_OUT points it elsewhere.
TODO(confirm): the author mirrors the ERP CORE entry into the catalog registry and signs off the dataset page (§10.11 item 8). The licence is contested at source and three real statements disagree: the LICENSE file shipped with the data says CC BY-SA 4.0, dataset_description.json says CC0, and the OSF node record says CC BY 4.0. Spec §10.7 makes the most restrictive reading govern, so the site records CC BY-SA 4.0 (§13 item 15, answered 2026-09-18). Share-alike binds anything you derive from these data, including whatever this capstone produces.
# Setup: dependencies, the shared helpers, non-interactive plotting.
import importlib.util
import subprocess
import sys
import warnings
from pathlib import Path
# 1. Dependencies are pinned in notebooks/requirements.txt. Nothing is installed
# when the pinned stack is already present (local runs, CI); a fresh Colab or
# Binder kernel installs it once. On Colab, run from a clone of the repository
# so that notebooks/_shared/ is available (repository URL: TODO(confirm), spec
# section 13 item 3).
_needed = ('mne', 'scipy', 'matplotlib', 'pooch', 'pyprep', 'autoreject', 'mne_bids', 'yaml')
_missing = [p for p in _needed if importlib.util.find_spec(p) is None]
if _missing:
_req = next((d / "requirements.txt" for d in (Path.cwd(), *Path.cwd().parents)
if (d / "requirements.txt").exists()), None)
_cmd = [sys.executable, "-m", "pip", "install", "-q"]
_cmd += ["-r", str(_req)] if _req else ["mne==1.10.2", "pooch>=1.8"]
if "pyprep" in _missing:
_cmd += ["pyprep>=0.9"]
if "autoreject" in _missing:
_cmd += ["autoreject>=0.5"]
if "mne_bids" in _missing:
_cmd += ["mne-bids>=0.16"]
if "yaml" in _missing:
_cmd += ["pyyaml>=6"]
subprocess.check_call(_cmd)
# 2. Shared helpers (notebooks/_shared/helpers.py and helpers_l2.py), located
# relative to the working directory -- notebooks/<level>/ or notebooks/ --
# never through an absolute path.
_shared = next((d / "_shared" for d in (Path.cwd(), *Path.cwd().parents)
if (d / "_shared" / "helpers_l2.py").exists()), None)
if _shared is None:
raise FileNotFoundError("start the kernel in notebooks/L2/ (or notebooks/) so that _shared/helpers_l2.py is found")
sys.path.insert(0, str(_shared))
import helpers
import helpers_l2 as l2
# 3. Plotting: Jupyter's default inline backend renders static PNGs through Agg
# (no windows, nothing blocks); outside Jupyter the helpers select Agg. Every
# MNE figure is requested with show=False, and plt.show() renders each cell's
# figures in place.
import matplotlib.pyplot as plt
import numpy as np
import mne
# Warnings are worth reading, so they are not silenced -- but their default format prints the
# absolute path of the file that raised them, which is nobody else's business and would put this
# machine's directory layout into the saved outputs. Only the class and the message are shown.
warnings.formatwarning = lambda message, category, *a, **k: f"{category.__name__}: {message}\n"
mne.viz.set_browser_backend("matplotlib", verbose=False)
mne.set_log_level("WARNING")
plt.rcParams["figure.dpi"] = 72
print(f"MNE {mne.__version__}; helpers imported from notebooks/_shared")
print("ERP CORE downloads (~56 MB per subject) are cached under data/downloads/erpcore/ in a repository "
"clone, otherwise under MNE's data directory; nothing is re-fetched.")
1. The pipeline package, and the configuration that drives it¶
Every parameter of the pipeline — the subject list, the exclusions with their reasons, the montage, the filter cutoffs and their written justification, the detection thresholds, the reference, the ICA method and rank policy, the rejection criteria, the seed — lives in one YAML file. Nothing is typed at a prompt; nothing is commented out to switch behaviour. The two overrides below are arguments, not edits: where the output goes, and (because mne-icalabel is installed here) that ICLabel should give its second opinion on every component.
FULL_COHORT is the §11 switch: it changes the subject list and nothing else.
import json
import os
import shutil
import tempfile
import time
from IPython.display import HTML, display
FULL_COHORT = False # True: all 40 participants (~2.2 GB of downloads; local runs only)
_pipelines = next((d / "pipelines" for d in (Path.cwd(), *Path.cwd().parents)
if (d / "pipelines" / "eegpipe" / "__init__.py").exists()), None)
EEGPIPE = None
if _pipelines is not None:
sys.path.insert(0, str(_pipelines))
try:
import eegpipe as EEGPIPE
except Exception as exc:
print(f"pipelines/eegpipe found but not importable: {type(exc).__name__}: {exc}")
_out = tempfile.TemporaryDirectory(prefix="nb-c2-")
OUTPUT_DIR = Path(os.environ.get("EEG_COURSE_C2_OUT", _out.name))
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
available = l2.available_subjects("P3")
SUBSET = [f"sub-{i:03d}" for i in range(1, 11)]
SUBJECTS = available if FULL_COHORT else [s for s in SUBSET if s in available]
if EEGPIPE is not None:
CONFIG_PATH = _pipelines / "configs" / "erpcore-p3.yaml"
OVERRIDES = {"output.dir": str(OUTPUT_DIR), "subjects.include": SUBJECTS,
"steps.ica.iclabel.enabled": True}
config = EEGPIPE.load_config(CONFIG_PATH, overrides=OVERRIDES)
print(f"pipeline: pipelines/eegpipe {EEGPIPE.__version__}")
print(f" canonical order: {' -> '.join(EEGPIPE.CANONICAL_ORDER)}")
print(f" configuration: {CONFIG_PATH.relative_to(_pipelines.parent)} "
f"({len(CONFIG_PATH.read_text().splitlines())} lines)")
print(f" overrides: {{'output.dir': <temporary>, 'subjects.include': {len(SUBJECTS)} subjects, "
f"'steps.ica.iclabel.enabled': True}}")
print(f" name {config.name!r}, dataset {config.dataset.id} ({config.dataset.paradigm}), "
f"seed {config.seed}")
print(f" exclusions declared in the file: {config.subjects.exclude or 'none -- ds-erpcore documents no defective subjects'}")
else:
config = l2.PipelineConfig(subjects=tuple(SUBJECTS), seed=l2.SEED)
print("pipeline: helpers_l2.run_subject -- the interface site/CONTRACTS.md documents for "
"pipelines/eegpipe, used because that package is not importable here")
print()
print(f"cohort: {'FULL (all participants)' if FULL_COHORT else 'documented subset sub-001..sub-010'} -> "
f"{len(SUBJECTS)} subjects")
print(f" {', '.join(SUBJECTS)}")
print(f" the paradigm's OSF component lists {len(available)} subjects; already on this machine: "
f"{len(l2.cached_subjects('P3'))}")
print(f" output: a temporary directory (set EEG_COURSE_C2_OUT to keep the cleaned data and the reports)")
print(f" free disk: {helpers.free_disk_mb(OUTPUT_DIR) / 1000:.1f} GB")
if EEGPIPE is not None:
text = CONFIG_PATH.read_text(encoding="utf-8")
body = [l for l in text.splitlines() if not l.lstrip().startswith("#")]
print("\n".join(body)[:3200].rstrip())
print(" ... (the rest of the file is the reject step and the autoreject alternative)")
else:
print(config.to_yaml())
2. The entry point is raw BIDS¶
ERP CORE ships each paradigm as a per-subject BIDS-compatible folder: sub-XXX_task-P3_eeg.set/.fdt with _eeg.json, _channels.tsv, _events.tsv, _electrodes.tsv and _coordsystem.json beside it, plus dataset_description.json and participants.tsv at the top. What it does not have is the sub-XXX/eeg/ datatype directory that BIDS requires, so read_raw_bids cannot open it as it stands.
The cell below arranges the cached files into a valid BIDS tree using hard links, so not a byte of EEG is copied, and reads one subject back with read_raw_bids — the L2.1 skill applied to this capstone's own input. That read is the proof that the entry point is BIDS: the channel types and the line frequency come from the sidecars rather than from anybody's memory. The pipeline itself then reads the same files through its own loader.
from mne_bids import BIDSPath, read_raw_bids
BIDS_ROOT = OUTPUT_DIR / "bids"
src_root = l2.erpcore_cache() / "P3"
def link_or_copy(src: Path, dst: Path):
dst.parent.mkdir(parents=True, exist_ok=True)
if dst.exists():
return
try:
os.link(src, dst) # no bytes copied
except OSError:
shutil.copyfile(src, dst)
t0 = time.time()
for sid in SUBJECTS: # cached files are never re-fetched
l2.fetch_erpcore_subject("P3", sid)
print(f"{len(SUBJECTS)} subjects present in the cache ({time.time() - t0:.0f} s; "
f"{helpers.free_disk_mb(l2.erpcore_cache()) / 1000:.1f} GB free)")
for name in ("dataset_description.json", "participants.tsv", "participants.json", "README.txt",
"CHANGES", "task-P3_events.json"):
src = src_root / name
if src.exists():
link_or_copy(src, BIDS_ROOT / ("README" if name == "README.txt" else name))
linked = []
for sid in SUBJECTS:
d = src_root / sid
if d.is_dir():
for f in sorted(d.iterdir()):
if f.is_file():
link_or_copy(f, BIDS_ROOT / sid / "eeg" / f.name)
linked.append(sid)
print(f"BIDS tree built for {len(linked)} subjects with hard links -- no EEG data copied "
f"({sum(f.stat().st_size for f in BIDS_ROOT.rglob('*') if f.is_file()) / 1e6:.0f} MB linked, "
f"{helpers.free_disk_mb(OUTPUT_DIR) / 1000:.1f} GB still free)")
for line in sorted(str(p.relative_to(BIDS_ROOT)) for p in BIDS_ROOT.rglob("*") if p.is_file())[:10]:
print(f" {line}")
print(" ...")
BIDS_OK, bids_note = False, ""
try:
bp = BIDSPath(subject=SUBJECTS[0].replace("sub-", ""), task="P3", datatype="eeg", root=BIDS_ROOT,
suffix="eeg", extension=".set")
with warnings.catch_warnings():
warnings.simplefilter("ignore")
raw_bids = read_raw_bids(bp, verbose=False)
BIDS_OK = True
types = {t: raw_bids.get_channel_types().count(t) for t in sorted(set(raw_bids.get_channel_types()))}
bids_note = (f"read_raw_bids opened {SUBJECTS[0]}: {types} at {raw_bids.info['sfreq']:g} Hz, "
f"{len(raw_bids.annotations)} annotations, PowerLineFrequency "
f"{raw_bids.info.get('line_freq')} Hz -- every one of those read from a sidecar, not assumed")
except Exception as e:
bids_note = (f"read_raw_bids could not open the tree ({type(e).__name__}: {str(e)[:140]}); the pipeline "
"reads the EEGLAB files directly with the sidecars alongside")
print()
print(bids_note)
3. Run the cohort unattended¶
One call. A failure is a row in the report, never an exception that stops the loop — the rubric's first line. Each subject leaves four files behind: the cleaned continuous recording, the surviving epochs, the machine-readable run log, and the QC page.
def norm(res):
"""One shape for eegpipe.RunResult and for helpers_l2.run_subject's dict."""
if hasattr(res, "status"):
per = (res.rejection or {}).get("per_condition", {})
return {"subject": res.subject, "ok": res.status == "ok", "error": res.error,
"bads": {ch: v.get("criteria", []) for ch, v in (res.bad_channels or {}).items()},
"interpolated": list(res.interpolated or []), "rank": res.rank,
"rank_why": res.ica.get("rank_why", ""), "filter": res.filter or {},
"removed": [{"component": e["index"], "class": e["label"],
"probability": e.get("score", float("nan")), "evidence": e["reason"]}
for e in res.ica.get("excluded", [])],
"rank_after": res.ica.get("rank_after"),
"per_condition": {c: v["percent_rejected"] for c, v in per.items()},
"kept": {c: v["n_kept"] for c, v in per.items()},
"percent_rejected": (res.rejection or {}).get("percent_rejected", float("nan")),
"imbalance_pp": (res.rejection or {}).get("condition_imbalance_pp", float("nan")),
"criterion": (res.rejection or {}).get("criterion", {}),
"steps": [{"name": s["name"], "duration_s": s["duration_s"]} for s in res.steps],
"flags": list(res.flags or []), "outputs": dict(res.outputs or {}),
"config_hash": res.config_hash, "seed": res.seed, "versions": res.versions,
"duration_s": res.duration_s, "obj": res}
d = res
tab = {r["condition"]: r for r in d.get("rejection", {}).get("table", [])}
return {"subject": d["subject"], "ok": d["ok"], "error": d["error"],
"bads": d.get("bad_channels", {}).get("detection", {}).get("by_channel", {}),
"interpolated": d.get("bad_channels", {}).get("decision", {}).get("bads", []),
"rank": d.get("rank", {}).get("rank"), "rank_why": d.get("rank", {}).get("arithmetic", ""),
"filter": d.get("filter_resolved", {}), "removed": d.get("ica", {}).get("removed", []),
"rank_after": d.get("ica", {}).get("rank_after"),
"per_condition": {c: tab[c]["percent_rejected"] for c in ("target", "standard") if c in tab},
"kept": {c: tab[c]["n_kept"] for c in ("target", "standard") if c in tab},
"percent_rejected": tab.get("all", {}).get("percent_rejected", float("nan")),
"imbalance_pp": (tab["target"]["percent_rejected"] - tab["standard"]["percent_rejected"]
if "target" in tab and "standard" in tab else float("nan")),
"criterion": {"peak_to_peak_uv": d.get("rejection", {}).get("threshold_uv")},
"steps": d["log"].steps, "flags": [], "outputs": {}, "config_hash": d.get("config_hash"),
"seed": d["config"].get("seed"), "versions": d.get("versions", {}),
"duration_s": d.get("duration_s", float("nan")), "obj": d}
t_cohort = time.time()
if EEGPIPE is not None:
raw_results = EEGPIPE.run_cohort(config, SUBJECTS, progress=False)
else:
raw_results = [l2.run_subject(config, sid, verbose=False) for sid in SUBJECTS]
R = {}
for res in raw_results:
r = norm(res)
R[r["subject"]] = r
if not r["ok"]:
print(f" {r['subject']}: FAILED -- {str(r['error'])[:110]}")
continue
print(f" {r['subject']}: {len(r['interpolated'])} interpolated {r['interpolated'] or ''}, "
f"rank {r['rank']}, {len(r['removed'])} components removed, "
f"{r['percent_rejected']:.1f} % of epochs rejected "
f"(imbalance {r['imbalance_pp']:+.1f} pp), {r['duration_s']:.0f} s"
+ (f" flags: {r['flags']}" if r["flags"] else ""))
COHORT_S = time.time() - t_cohort
OK = [r for r in R.values() if r["ok"]]
files = [p for p in OUTPUT_DIR.rglob("*") if p.is_file() and "bids" not in p.parts]
print(f"\n{len(SUBJECTS)} subjects in {COHORT_S:.0f} s ({COHORT_S / max(len(SUBJECTS), 1):.0f} s each); "
f"{len(SUBJECTS) - len(OK)} failure(s)")
print(f"written: {len([f for f in files if f.name.endswith('_raw.fif')])} cleaned recordings, "
f"{len([f for f in files if f.name.endswith('_epo.fif')])} epoch files, "
f"{len([f for f in files if f.suffix == '.html'])} QC pages, "
f"{len([f for f in files if f.name == 'run-log.json'])} run logs -- "
f"{sum(f.stat().st_size for f in files) / 1e6:.0f} MB in total")
4. The group summary¶
# The subject-exclusion rule, fixed in the configuration file before any data were seen and applied here
# blind to any effect: a condition may not lose more than max_rejected_fraction of its trials, and a
# condition must keep at least MIN_TRIALS_PER_CONDITION.
MAX_REJECTED_PCT = (config.steps.reject.max_rejected_fraction * 100 if EEGPIPE is not None
else config.max_percent_rejected)
MIN_TRIALS_PER_CONDITION = 20
group = []
for sid in SUBJECTS:
r = R[sid]
worst = max([v for v in r["per_condition"].values()] or [float("nan")])
fewest = min([v for v in r["kept"].values()] or [0])
r["excluded"] = bool(not r["ok"] or worst > MAX_REJECTED_PCT or fewest < MIN_TRIALS_PER_CONDITION)
r["exclusion_reason"] = ("did not complete" if not r["ok"] else
f"{worst:.0f} % of a condition rejected (limit {MAX_REJECTED_PCT:.0f} %)"
if worst > MAX_REJECTED_PCT else
f"only {fewest} trials kept in a condition (minimum {MIN_TRIALS_PER_CONDITION})"
if fewest < MIN_TRIALS_PER_CONDITION else "")
group.append({"subject": sid, "ok": r["ok"],
"bads": r["interpolated"] or ["-"],
"rank after cleaning": r["rank"] if r["ok"] else float("nan"),
"ICA removed": len(r["removed"]) if r["ok"] else float("nan"),
"% rejected target": r["per_condition"].get("target", float("nan")),
"% rejected standard": r["per_condition"].get("standard", float("nan")),
"imbalance (pp)": r["imbalance_pp"],
"target kept": r["kept"].get("target", float("nan")),
"standard kept": r["kept"].get("standard", float("nan")),
"excluded": r["excluded"], "why": r["exclusion_reason"] or "-",
"flags": len(r["flags"]), "seconds": r["duration_s"]})
print(l2.fmt_table(group, list(group[0]), floatfmt="{:.2f}"))
print()
KEPT = [r for r in R.values() if r["ok"] and not r["excluded"]]
print(f"Subject-exclusion rule, fixed in the configuration before any data were seen and applied blind to any "
f"effect: a condition may lose at most {MAX_REJECTED_PCT:.0f} % of its trials and must keep at least "
f"{MIN_TRIALS_PER_CONDITION}. {len(KEPT)} of {len(SUBJECTS)} subjects pass.")
for r in R.values():
if r["excluded"]:
print(f" excluded: {r['subject']} -- {r['exclusion_reason']}")
print()
n_bads = [len(r["interpolated"]) for r in OK]
n_removed = [len(r["removed"]) for r in OK]
print(f"{len(OK)} of {len(SUBJECTS)} subjects completed.")
print(f"Channels interpolated: median {np.median(n_bads):.0f}, range {min(n_bads)}-{max(n_bads)} of 30.")
print(f"ICA components removed: median {np.median(n_removed):.0f}, range {min(n_removed)}-{max(n_removed)} "
f"of a rank of about {int(np.median([r['rank'] for r in OK]))} -- comparable across subjects is what "
"pf-overcleaning-ica asks for.")
print(f"Data rejected: target median {np.median([r['per_condition'].get('target', np.nan) for r in OK]):.1f} %, "
f"standard median {np.median([r['per_condition'].get('standard', np.nan) for r in OK]):.1f} %.")
print("Flags raised by the pipeline (recorded, never fatal):")
for r in OK:
if r["flags"]:
print(f" {r['subject']}: " + "; ".join(str(f) for f in r["flags"]))
n_conv = len([r for r in OK if any("converge" in str(f) for f in r["flags"])])
if n_conv:
print(f"\n{n_conv} of {len(OK)} subjects raise an ICA convergence flag at the configuration's iteration "
"limit. The decompositions are still usable -- the classifier labels them confidently and the "
"removed components are the expected ones -- but a run that reports this on most of its cohort is "
"telling you to raise the limit and re-run before publishing anything from it. That is what a flag "
"is for: it is data, not a crash, and it is in every subject's report.")
if EEGPIPE is not None:
print()
print("The package's own summary (capstone deliverable 4), written as summary.csv and summary.json:")
print(l2.fmt_table(EEGPIPE.summarize(raw_results), floatfmt="{:.2f}"))
5. The rubric's central check: was the rejection condition-biased?¶
The criterion is one number applied to every condition, fixed in the configuration file before any data were seen, so the condition played no part in setting it. That does not guarantee equal rejection — it guarantees only that the experimenter did not choose the imbalance. The check is to look.
gaps = np.array([r["imbalance_pp"] for r in OK], float)
print(f"per-condition imbalance (percentage points), {len(gaps)} subjects:")
print(f" median {np.nanmedian(gaps):+.1f} pp, mean {np.nanmean(gaps):+.1f} pp, "
f"range {np.nanmin(gaps):+.1f} to {np.nanmax(gaps):+.1f} pp")
for level in (20, 10):
who = [r["subject"] for r in OK if abs(r["imbalance_pp"]) >= level]
print(f" subjects with |imbalance| >= {level} pp: {who or 'none'}")
print(f" rejection criterion in force: {OK[0]['criterion']}")
print()
print("A difference of a few percentage points is life. A subject at 20 points is a finding about the "
"pipeline, not about the brain (L2.5), and it belongs in the report whether or not it changes the "
"conclusion.")
fig, axes = plt.subplots(1, 3, figsize=(15, 4.0))
labels = [r["subject"].replace("sub-", "") for r in OK]
axes[0].bar(range(len(OK)), [r["per_condition"].get("target", np.nan) for r in OK], width=0.4,
align="edge", color="tab:blue", label="target")
axes[0].bar([i + 0.4 for i in range(len(OK))], [r["per_condition"].get("standard", np.nan) for r in OK],
width=0.4, align="edge", color="tab:orange", label="standard")
axes[0].set(xticks=[i + 0.4 for i in range(len(OK))], ylabel="Trials rejected (%)",
title="Data rejected per condition, per subject (%)")
axes[0].set_xticklabels(labels, rotation=90, fontsize=7)
axes[0].legend(fontsize=8); axes[0].grid(alpha=0.3, axis="y")
axes[1].axhline(0, color="gray", lw=0.6)
for y in (20, -20):
axes[1].axhline(y, color="tab:red", lw=0.8, ls="--")
axes[1].bar(range(len(OK)), gaps, color=["tab:red" if abs(g) >= 20 else "tab:blue" for g in gaps])
axes[1].set(xticks=range(len(OK)), ylabel="target - standard (percentage points)",
title="Condition imbalance per subject (percentage points); dashed lines +/- 20 pp")
axes[1].set_xticklabels(labels, rotation=90, fontsize=7)
axes[1].grid(alpha=0.3, axis="y")
CH, WINDOW = l2.P3_CHANNEL, l2.P3_WINDOW
diffs, amps = [], []
for r in KEPT:
ep = getattr(r["obj"], "epochs", None) if hasattr(r["obj"], "status") else r["obj"].get("epochs")
if ep is None:
p = r["outputs"].get("epochs")
ep = mne.read_epochs(p, verbose=False) if p and Path(p).exists() else None
if ep is None or CH not in ep.ch_names:
continue
d = l2.difference_wave(ep)
diffs.append(d)
amps.append({"subject": r["subject"], f"{CH} mean (uV)": l2.mean_amplitude(d, CH, WINDOW),
"trials": int(d.nave)})
axes[2].plot(d.times * 1000, d.data[d.ch_names.index(CH)] * 1e6, lw=0.6, color="0.7")
for r in R.values(): # the excluded subjects, drawn but not averaged
if not r["excluded"] or not r["ok"]:
continue
ep = getattr(r["obj"], "epochs", None) if hasattr(r["obj"], "status") else r["obj"].get("epochs")
if ep is None:
p = r["outputs"].get("epochs")
ep = mne.read_epochs(p, verbose=False) if p and Path(p).exists() else None
if ep is not None and CH in ep.ch_names:
d = l2.difference_wave(ep)
axes[2].plot(d.times * 1000, d.data[d.ch_names.index(CH)] * 1e6, lw=0.7, color="tab:red", ls=":",
label=f"{r['subject']} (excluded)")
if diffs:
grand = mne.grand_average(diffs)
axes[2].plot(grand.times * 1000, grand.data[grand.ch_names.index(CH)] * 1e6, lw=2.0, color="tab:blue",
label=f"grand average, n = {len(diffs)}")
axes[2].legend(fontsize=8)
axes[2].axvspan(WINDOW[0] * 1000, WINDOW[1] * 1000, color="tab:orange", alpha=0.18)
axes[2].axhline(0, color="gray", lw=0.6); axes[2].axvline(0, color="gray", lw=0.6)
axes[2].set(xlabel="Time from stimulus (ms)", ylabel="Amplitude (uV)",
title=f"Target minus standard at {CH} (uV, positive up; dotted red = excluded by the rule)")
axes[2].grid(alpha=0.3)
fig.tight_layout()
plt.show() # render the static figure(s) of this cell inline
if amps:
print(l2.fmt_table(amps, list(amps[0]), floatfmt="{:+.2f}"))
v = np.array([a[f"{CH} mean (uV)"] for a in amps])
print(f"\ngrand mean over the {len(v)} retained subjects: {v.mean():+.2f} uV "
f"(SD {v.std(ddof=1):.2f}, SEM {v.std(ddof=1) / np.sqrt(len(v)):.2f}, "
f"positive in {int((v > 0).sum())}/{len(v)} subjects). The excluded subjects are drawn in the "
"figure and left out of the average, which is the whole point of fixing the rule in advance.")
6. One QC report, as a stranger would read it¶
SHOW = OK[0]["subject"]
qc_path = R[SHOW]["outputs"].get("qc")
if qc_path and Path(qc_path).exists():
html = Path(qc_path).read_text(encoding="utf-8")
print(f"{SHOW}: {Path(qc_path).name}, {len(html) / 1000:.0f} kB of self-contained HTML; the same page "
f"exists for every subject, and index.html links them all")
else:
html = l2.qc_report_html(R[SHOW]["obj"], title=f"QC report -- {SHOW} (ds-erpcore P3)")
print(f"{SHOW}: rendered by helpers_l2.qc_report_html, {len(html) / 1000:.0f} kB")
display(HTML(html))
7. The methods paragraph¶
A reader should be able to follow this and reproduce the pipeline. Every number in it comes from the run logs above, not from memory.
r0 = R[SHOW]
fa = (r0["filter"].get("analysis") or r0["filter"])
ic = r0["filter"].get("ica_copy") or {}
versions = {k: v for k, v in (r0["versions"] or {}).items() if v not in (None, "not installed")}
just = r0["filter"].get("justification") or {}
METHODS = f"""\
Data were the P3 (active visual oddball) paradigm of ERP CORE, {len(SUBJECTS)} of the 40 participants \
({SUBJECTS[0]}-{SUBJECTS[-1]}); 30 EEG and 3 EOG channels recorded with a Biosemi ActiveTwo against the CMS \
arrangement at 1024 Hz with no software filters and 60 Hz mains. The per-subject folders are \
BIDS-compatible and were read as BIDS ({'confirmed with read_raw_bids' if BIDS_OK else 'sidecars read alongside the EEGLAB files'}). \
Processing was scripted with a single configuration file (hash {r0['config_hash']}), executed identically \
for every participant with one random seed ({r0['seed']}) passed into every stochastic step, in the \
canonical order load, montage, bad-channel detection, filter, interpolate, re-reference, ICA, epoch, reject.
Channel names were mapped onto the {'standard_1005' if EEGPIPE is None else config.steps.montage.name} \
template montage (the file spells the frontal pair FP1/FP2, the montage Fp1/Fp2) and the three EOG channels \
were typed as EOG so that they could score ICA components without joining the average reference. Bad \
channels were detected from flatness, robust amplitude deviation, neighbour correlation and \
high-frequency noise on data that had not yet been low-passed, because high-frequency noise is one of the \
criteria; the rule interpolated a median of {np.median(n_bads):.0f} channels \
(range {min(n_bads)}-{max(n_bads)} of 30) by spherical splines.
The continuous data were filtered {fa.get('l_freq_hz')}-{fa.get('h_freq_hz')} Hz \
({str(fa.get('method', 'fir')).upper()}, {fa.get('phase', 'zero')}-phase, \
{fa.get('filter_length_samples', fa.get('fir_length_samples', '?'))} taps = \
{fa.get('filter_length_s', fa.get('fir_length_s', float('nan'))):.1f} s) before epoching, so that the \
filter's edge artifacts lie at the ends of the recording rather than inside every epoch, and resampled to \
{r0['filter'].get('resampled_to_hz', 256)} Hz after the low-pass. \
{' '.join(str(just.get('highpass', '')).split())} {' '.join(str(just.get('lowpass', '')).split())} \
A separate copy high-passed at {ic.get('l_freq_hz', 1.0)} Hz was kept for the ICA fit alone. \
{' '.join(str(just.get('ica_highpass', '')).split())} \
Data were then re-referenced to the average of the 30 EEG channels, with the rank tracked through \
interpolation and referencing ({r0['rank_why']}) and passed to ICA as its component count.
ICA used extended Infomax with the tracked rank and the run's seed. Components were removed when the \
recording's own EOG channels identified them, or when ICLabel classified them into one of the artifact \
classes above the configured probability, subject to a cap; a median of {np.median(n_removed):.0f} \
components was removed (range {min(n_removed)}-{max(n_removed)}), and every removed component is listed in \
the subject's QC report with its class, its score and the sentence explaining why. Labels are algorithmic \
and no person has reviewed them.
Epochs ran from -0.2 to 0.8 s around each stimulus with a -0.2 to 0 s baseline; targets and standards were \
identified from the dataset's own event dictionary, in which a stimulus code with equal digits is a target. \
Epochs were rejected on a peak-to-peak criterion identical for both conditions and fixed before the data \
were seen ({r0['criterion']}), which removed a median of \
{np.median([r['per_condition'].get('target', np.nan) for r in OK]):.1f}% of target and \
{np.median([r['per_condition'].get('standard', np.nan) for r in OK]):.1f}% of standard trials, a median \
per-condition imbalance of {np.nanmedian(gaps):+.1f} percentage points. Subjects were excluded when a \
condition lost more than {MAX_REJECTED_PCT:.0f}% of its trials or kept fewer than \
{MIN_TRIALS_PER_CONDITION}: {', '.join(r['subject'] for r in R.values() if r['excluded']) or 'no subject'} \
met that rule, leaving {len(KEPT)} of {len(SUBJECTS)}.
The P3 was measured as the mean amplitude of the target-minus-standard difference wave at {CH} over \
{WINDOW[0] * 1000:.0f}-{WINDOW[1] * 1000:.0f} ms, a window fixed before any waveform was inspected. \
Package versions are recorded in every run log: {', '.join(f'{k} {v}' for k, v in versions.items())}."""
print(METHODS)
8. The numbers, and the rubric¶
print("nb-c2-clean-pipeline -- Capstone C2 deliverables (draft; TODO(confirm) at author review)")
print(f"Implementation: " + (f"pipelines/eegpipe {EEGPIPE.__version__}, configuration "
f"{CONFIG_PATH.relative_to(_pipelines.parent)}" if EEGPIPE is not None
else "helpers_l2.run_subject (the documented interface)"))
print(f"Dataset: ds-erpcore P3 (CC BY-SA 4.0 per data/directory.yaml; contested at source -- the LICENSE file "
f"says CC BY-SA 4.0, dataset_description.json says CC0, the OSF node record says CC BY 4.0; spec 10.7 "
f"makes the most restrictive reading govern, so share-alike binds what you derive here).")
print(f"Cohort: {'FULL (all participants)' if FULL_COHORT else 'documented subset sub-001..sub-010'} -> "
f"{len(SUBJECTS)} subjects; FULL_COHORT switch at the top of section 1.")
print(f"Configuration hash {R[SHOW]['config_hash']}, seed {R[SHOW]['seed']}.")
print(f"BIDS entry point: {bids_note}")
print(f"Wall clock: {COHORT_S:.0f} s for {len(SUBJECTS)} subjects "
f"({COHORT_S / max(len(SUBJECTS), 1):.0f} s each).")
print()
print("Group summary table (deliverable 4):")
print(l2.fmt_table(group, list(group[0]), floatfmt="{:.2f}"))
print()
print("Per-subject step durations (s):")
print(l2.fmt_table([dict(subject=r["subject"], **{s["name"]: s["duration_s"] for s in r["steps"]},
total=r["duration_s"]) for r in OK], floatfmt="{:.1f}"))
print()
print("ICA components removed, per subject, with the reason recorded for each (deliverable 3):")
for r in OK:
print(f" {r['subject']}: rank {r['rank']} -> {r['rank_after']} after cleaning")
for e in r["removed"]:
print(f" IC{e['component']:<3} {str(e['class']):16s} score {e['probability']:.3f} "
f"{str(e['evidence'])[:92]}")
if not r["removed"]:
print(" nothing met the policy")
print()
print("Rubric, with the measured state of each item:")
RUBRIC = [
("Runs unattended on all subjects -- one call, no prompts, a failure does not stop the cohort",
f"{len(OK)}/{len(SUBJECTS)} completed in one call to run_cohort; "
f"{len(SUBJECTS) - len(OK)} failure(s) recorded as rows; nb-2-8-pipeline section 6 shows the failure "
"path on a subject that does not exist"),
("No condition-biased rejection -- criterion set condition-blind, per-condition table for every subject",
f"one peak-to-peak criterion for both conditions, fixed in the configuration file; imbalance median "
f"{np.nanmedian(gaps):+.1f} pp, range {np.nanmin(gaps):+.1f} to {np.nanmax(gaps):+.1f} pp; "
f"{len([r for r in OK if abs(r['imbalance_pp']) >= 20])} subject(s) at or above 20 pp"),
("QC report readable by a stranger",
f"{len([f for f in files if f.suffix == '.html'])} self-contained HTML pages plus a cohort index; "
"one is rendered in section 6"),
("Driven entirely by a configuration file, seed and versions in the run log",
f"hash {R[SHOW]['config_hash']}, seed {R[SHOW]['seed']}, versions recorded per subject; the two "
"overrides applied here are arguments (output directory, ICLabel on), not edits"),
("Canonical order implemented and justified",
("load -> montage -> bad-channel detection -> filter -> interpolate -> re-reference -> ICA -> epoch -> "
"reject") + "; justified step by step in the methods paragraph, and the filter's justification is "
"carried in the configuration file itself"),
("Rank tracked through interpolation and re-referencing and passed to ICA",
f"e.g. {SHOW}: {R[SHOW]['rank_why']}; ICA component count = that number"),
("Every removed ICA component has a class and evidence; the number removed is comparable across subjects",
f"median {np.median(n_removed):.0f}, range {min(n_removed)}-{max(n_removed)}; each carries a detector, "
"a score and a sentence"),
("The subject-exclusion rule was fixed in advance and applied blind to the effect",
f"a condition may lose at most {MAX_REJECTED_PCT:.0f} % of its trials (the configuration's "
f"max_rejected_fraction) and must keep at least {MIN_TRIALS_PER_CONDITION}: "
+ (", ".join(f"{r['subject']} ({r['exclusion_reason']})" for r in R.values() if r['excluded'])
or "no subject crossed it") + f"; {len(KEPT)} of {len(SUBJECTS)} retained, and the rule was applied "
"before any grand average was computed"),
("Dataset-specific documented defects handled explicitly",
"ds-erpcore documents no defective subjects; the configuration's subjects.exclude field is where "
"ds-eegbci's S088/S089/S092/S100 and S038/S104 go, each with its reason (pipelines/configs/eegbci.yaml)"),
]
for item, state in RUBRIC:
print(f" [x] {item}\n {state}")
print()
print(f"Deliverables written: {len([f for f in files if f.name.endswith('_raw.fif')])} cleaned continuous "
f"recordings, {len([f for f in files if f.name.endswith('_epo.fif')])} epoch files, "
f"{len([f for f in files if f.suffix == '.html'])} QC pages and "
f"{len([f for f in files if f.name == 'run-log.json'])} run logs, "
f"{sum(f.stat().st_size for f in files) / 1e6:.0f} MB. Set EEG_COURSE_C2_OUT to keep them.")
if amps:
v = np.array([a[f"{CH} mean (uV)"] for a in amps])
print()
print(f"Grand-average P3 across the {len(v)} retained subjects: {CH} mean amplitude {v.mean():+.2f} uV "
f"(SD {v.std(ddof=1):.2f}, SEM {v.std(ddof=1) / np.sqrt(len(v)):.2f}). "
"C3 reuses this pipeline unchanged.")
_out.cleanup()
print("\ntemporary output directory removed; the ERP CORE downloads stay in the cache.")