nb-2-8-pipeline · Pipeline order and reproducibility (L2.8)¶
Lesson L2.8 · Level 2 · Status draft — for expert review; uncertain points carry TODO(confirm).
Everything in Level 2 becomes one function here: run_subject(config, subject) executing the canonical order
load → montage → bad-channel detection → filter → interpolate → re-reference → ICA → epoch → reject
driven entirely by a configuration file, logging every step, and emitting a per-subject QC HTML report.
- Which implementation is running — the repository's
pipelines/eegpipepackage if it is there, and the same documented interface innotebooks/_shared/helpers_l2.pyif it is not. - The configuration: one YAML file, validated, with a hash.
- One subject end to end, with the run log: what each step did, how long it took, and the resolved parameter values rather than the requested ones.
- The QC report, rendered inline.
- Two runs differ only by their configuration — one parameter changed, and the consequences read off the two logs.
- A failure is a row, not a stopped cohort.
Where the code lives. The Phase 2 build contract reserves pipelines/eegpipe for this package — config.py, steps/ (one module per step), run.py with run_subject(config, subject) -> RunResult, qc.py (the HTML report), cli.py and tests/. The package is imported without being installed, by walking up from the working directory to the repository's pipelines/ folder — the same trick that finds _shared/, and no absolute path anywhere.
Data. ds-erpcore P3 (CC BY-SA 4.0 per data/directory.yaml, contested at source; open; per-subject downloadable): sub-001 for the worked run (~56 MB on an empty cache; already-cached subjects are re-used). TODO(confirm): the author mirrors the ERP CORE entry into the catalog registry and signs off the dataset page (§10.11 item 8).
# 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', '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 "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. Which implementation is running¶
import tempfile
import time
from IPython.display import HTML, display
_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}")
if EEGPIPE is not None:
print(f"implementation: pipelines/eegpipe {EEGPIPE.__version__} (the package the Phase 2 contract reserves)")
print(f" canonical order: {' -> '.join(EEGPIPE.CANONICAL_ORDER)}")
print(f" step modules: {', '.join(EEGPIPE.STEPS)}")
run_one = EEGPIPE.run_subject
else:
print("implementation: helpers_l2.run_subject -- the same documented interface "
"(site/CONTRACTS.md, Phase 2 addendum), used because pipelines/eegpipe is not importable here")
run_one = l2.run_subject
print()
print("Documented interface (site/CONTRACTS.md, Phase 2 addendum):")
print(" config.py dataclass schema + YAML load/validate")
print(" steps/ load, montage, bad_channels, filter, interpolate, reference, ica, epoch, reject")
print(" run.py run_subject(config, subject) -> RunResult")
print(" qc.py HTML report: bad-channel table, ICA components removed with reasons, percent data")
print(" rejected per condition, filter settings, run log with versions and seed")
print(" cli.py python -m eegpipe run --config ...")
print(" tests/ pytest, including a fast synthetic test that downloads nothing")
# Both implementations are read through one small adapter, so every table below is written once.
def norm(res):
"""One shape for eegpipe.RunResult and for helpers_l2.run_subject's dict."""
if hasattr(res, "status"): # eegpipe.RunResult
bads = {ch: v.get("criteria", []) for ch, v in (res.bad_channels or {}).items()}
removed = [{"component": e["index"], "class": e["label"],
"probability": e.get("score", float("nan")), "evidence": e["reason"]}
for e in res.ica.get("excluded", [])]
per = (res.rejection or {}).get("per_condition", {})
table = [{"condition": c, "n_presented": v["n_before"], "n_rejected": v["n_rejected"],
"n_kept": v["n_kept"], "percent_rejected": v["percent_rejected"]} for c, v in per.items()]
table.append({"condition": "all", "n_presented": res.rejection.get("n_epochs_before", 0),
"n_rejected": res.rejection.get("n_epochs_before", 0) - res.rejection.get("n_epochs_after", 0),
"n_kept": res.rejection.get("n_epochs_after", 0),
"percent_rejected": res.rejection.get("percent_rejected", float("nan"))})
return {"subject": res.subject, "ok": res.status == "ok", "error": res.error,
"steps": [{"name": s["name"], "duration_s": s["duration_s"], "params": s.get("params", {}),
"notes": s.get("notes", "")} for s in res.steps],
"bads": bads, "interpolated": list(res.interpolated or []), "rank": res.rank,
"rank_why": res.ica.get("rank_why", ""), "filter": res.filter or {},
"ica": {"n_components": res.ica.get("n_components"), "method": res.ica.get("method"),
"seed": res.ica.get("seed"), "fitted_on": res.ica.get("fitted_on"),
"rank_after": res.ica.get("rank_after"), "removed": removed,
"kept_by_cap": res.ica.get("kept_by_max_remove", []),
"converged": res.ica.get("converged")},
"rejection": {"criterion": res.rejection.get("criterion", {}), "table": table,
"imbalance_pp": res.rejection.get("condition_imbalance_pp", float("nan"))},
"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}
d = res # helpers_l2.run_subject
det = d.get("bad_channels", {}).get("detection", {})
dec = d.get("bad_channels", {}).get("decision", {})
return {"subject": d["subject"], "ok": d["ok"], "error": d["error"],
"steps": d["log"].steps, "bads": det.get("by_channel", {}),
"interpolated": dec.get("bads", []), "rank": d.get("rank", {}).get("rank"),
"rank_why": d.get("rank", {}).get("arithmetic", ""), "filter": d.get("filter_resolved", {}),
"ica": {"n_components": d.get("ica", {}).get("log", {}).get("n_components"),
"method": d.get("ica", {}).get("log", {}).get("method"),
"seed": d.get("ica", {}).get("log", {}).get("seed"),
"fitted_on": d.get("ica", {}).get("log", {}).get("fit_copy"),
"rank_after": d.get("ica", {}).get("rank_after"),
"removed": d.get("ica", {}).get("removed", []), "kept_by_cap": [], "converged": None},
"rejection": {"criterion": {"peak_to_peak_uv": d.get("rejection", {}).get("threshold_uv")},
"table": d.get("rejection", {}).get("table", []),
"imbalance_pp": float("nan")},
"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"))}
2. The configuration¶
One file holds every parameter: the subject list and the exclusions with their reasons, the montage, the filter cutoffs and their justification, the detection thresholds, the reference, the ICA method and rank policy, the rejection criteria, the seed. Nothing is typed at a prompt; nothing is commented out to switch behaviour; no parameter appears twice.
The practical test the lesson gives is: can you reproduce last month's result by checking out the configuration file? The hash below is what a run log records so that question has an answer.
Two overrides are applied here, both of them arguments rather than edits: the output goes to a temporary directory, and ICLabel is switched on, because mne-icalabel is installed in this environment and the lesson wants the classifier's second opinion (L2.6).
_out = tempfile.TemporaryDirectory(prefix="nb-2-8-")
OUT = Path(_out.name)
if EEGPIPE is not None:
CONFIG_PATH = _pipelines / "configs" / "erpcore-p3.yaml"
OVERRIDES = {"output.dir": str(OUT), "steps.ica.iclabel.enabled": True,
"subjects.include": ["sub-001"]}
SHOWN_OVERRIDES = {**OVERRIDES, "output.dir": "<a temporary directory>"}
config = EEGPIPE.load_config(CONFIG_PATH, overrides=OVERRIDES)
text = CONFIG_PATH.read_text(encoding="utf-8")
print(f"configuration file: {CONFIG_PATH.relative_to(_pipelines.parent)} "
f"({len(text.splitlines())} lines, {len(text) / 1000:.1f} kB)")
print(f"overrides applied as arguments, not edits: {SHOWN_OVERRIDES}")
print(f"name: {config.name!r}; dataset {config.dataset.id} ({config.dataset.paradigm}); "
f"seed {config.seed}")
print()
print("".join(l for l in text.splitlines(keepends=True)
if not l.startswith("#"))[:2600].rstrip() + "\n ...")
else:
config = l2.PipelineConfig(subjects=("sub-001",), seed=l2.SEED)
print(config.to_yaml())
print()
print("The configuration's hash is recorded by every run (printed with the run log in section 3).")
3. One subject, end to end¶
The run log is not a debug convenience, it is the evidence. Each step records its name, its parameters as used (after defaults are resolved), its duration, and what it did.
t0 = time.time()
raw_result = run_one(config, "sub-001") if EEGPIPE is None else run_one(config, "sub-001", progress=False)
R = norm(raw_result)
print(f"\nreturned in {time.time() - t0:.0f} s; ok = {R['ok']}; configuration hash {R['config_hash']}, "
f"seed {R['seed']}")
print()
print("Run log:")
print(l2.fmt_table([{"step": s["name"], "duration (s)": s["duration_s"],
"notes": str(s.get("notes", ""))[:78]} for s in R["steps"]],
["step", "duration (s)", "notes"], floatfmt="{:.2f}"))
print(f" total {R['duration_s']:.1f} s")
if R["flags"]:
print(f"\nflags raised by the run (recorded, not fatal): {R['flags']}")
print(f"Bad channels ({len(R['bads'])} flagged, {len(R['interpolated'])} interpolated):")
print(l2.fmt_table([{"channel": ch, "criteria": crit,
"decision": "interpolated" if ch in R["interpolated"] else "kept"}
for ch, crit in R["bads"].items()]
or [{"channel": "(none flagged)", "criteria": "-", "decision": "-"}],
["channel", "criteria", "decision"]))
print()
print(f"Rank: {R['rank']} ({R['rank_why']})")
print()
print("Filter as resolved -- the values actually used, not the values requested:")
f = R["filter"]
flat = {k: v for k, v in f.items() if not isinstance(v, dict)}
print(l2.fmt_table([{"field": k, "value": str(v)[:70]} for k, v in flat.items()], ["field", "value"]))
for sub in ("analysis", "ica_copy"):
if isinstance(f.get(sub), dict):
print(f" {sub}: " + ", ".join(f"{k} {v}" for k, v in f[sub].items() if k != "why"))
if f[sub].get("why"):
print(f" why: {f[sub]['why']}")
for k, v in (f.get("justification") or {}).items():
print(f" justification, {k}: {' '.join(str(v).split())}")
ica = R["ica"]
print(f"ICA: {ica['method']}, {ica['n_components']} components, seed {ica['seed']}, fitted on "
f"{ica['fitted_on']}; converged: {ica['converged']}")
print(f" components removed: {len(ica['removed'])}; rank after cleaning {ica['rank_after']}")
print(l2.fmt_table(ica["removed"] or [{"component": "-", "class": "-", "probability": float("nan"),
"evidence": "nothing met the policy"}],
["component", "class", "probability", "evidence"], floatfmt="{:.3f}"))
if ica["kept_by_cap"]:
print(f" kept only because the cap was reached (logged, not silently dropped): {ica['kept_by_cap']}")
print()
print(f"Rejection criterion (one criterion for every condition, fixed before the data were looked at): "
f"{R['rejection']['criterion']}")
print(l2.fmt_table(R["rejection"]["table"],
["condition", "n_presented", "n_rejected", "n_kept", "percent_rejected"], floatfmt="{:.2f}"))
print(f" per-condition imbalance: {R['rejection']['imbalance_pp']:.2f} percentage points")
print()
print(f"Files written: " + ", ".join(f"{k} -> {Path(v).name}" for k, v in R["outputs"].items())
if R["outputs"] else "Files written: none (this implementation returns objects rather than files)")
4. The QC report¶
One HTML page per subject, self-contained, readable by someone who did not write the pipeline: the numbers that would make you distrust the subject at the top, then the bad-channel table, the removed components with their evidence, the per-condition rejection, the filter settings with a sentence each, and the run log with versions and the seed.
qc_path = R["outputs"].get("qc")
if qc_path and Path(qc_path).exists():
html = Path(qc_path).read_text(encoding="utf-8")
print(f"{Path(qc_path).name}: {len(html) / 1000:.0f} kB of self-contained HTML "
"(every figure embedded as a base64 PNG)")
else:
html = l2.qc_report_html(raw_result, title="QC report -- sub-001 (ds-erpcore P3)")
print(f"rendered by helpers_l2.qc_report_html: {len(html) / 1000:.0f} kB")
display(HTML(html))
5. Two runs differ only by their configuration¶
One parameter is changed — the analysis high-pass moves from 0.1 Hz to 1 Hz, which L2.4 says costs a P3 a third of its amplitude — and nothing else. The two configuration hashes and the two run logs are the whole record of the difference. No file was edited to produce the second run.
if EEGPIPE is not None:
config_b = EEGPIPE.load_config(CONFIG_PATH, overrides={**OVERRIDES, "steps.filter.l_freq": 1.0,
"output.dir": str(OUT / "run-b")})
result_b = norm(EEGPIPE.run_subject(config_b, "sub-001"))
else:
import dataclasses
config_b = dataclasses.replace(config, l_freq=1.0)
result_b = norm(l2.run_subject(config_b, "sub-001", verbose=False))
rows = []
for label, res in (("A: analysis high-pass 0.1 Hz", R), ("B: analysis high-pass 1.0 Hz", result_b)):
fa = (res["filter"].get("analysis") or res["filter"])
rows.append({"run": label, "config hash": res["config_hash"],
"high-pass (Hz)": fa.get("l_freq_hz", float("nan")),
"FIR length (s)": fa.get("filter_length_s", res["filter"].get("fir_length_s", float("nan"))),
"bads interpolated": res["interpolated"] or ["-"], "rank": res["rank"],
"ICA removed": len(res["ica"]["removed"]),
"% rejected (all)": [r for r in res["rejection"]["table"] if r["condition"] == "all"][0]["percent_rejected"],
"imbalance (pp)": res["rejection"]["imbalance_pp"],
"duration (s)": res["duration_s"]})
print(l2.fmt_table(rows, list(rows[0]), floatfmt="{:.2f}"))
print()
print("Everything a reader needs in order to know why the two runs differ is the pair of configuration "
"hashes. That is what 'two runs differ only by their configuration' buys, and it is the test L2.8 "
"gives: could you reproduce last month's result by checking out the configuration file?")
# The measurement itself, so that the difference is a number rather than a claim.
CH, WINDOW = l2.P3_CHANNEL, l2.P3_WINDOW
measure_rows = []
def epochs_of(res_obj):
"""The cleaned epochs, whichever implementation produced them."""
if hasattr(res_obj, "status"):
if getattr(res_obj, "epochs", None) is not None:
return res_obj.epochs
p = (res_obj.outputs or {}).get("epochs")
return mne.read_epochs(p, verbose=False) if p and Path(p).exists() else None
return res_obj.get("epochs")
ep_a = epochs_of(raw_result)
ep_b_src = (EEGPIPE.run_subject(config_b, "sub-001") if EEGPIPE is not None
else l2.run_subject(config_b, "sub-001", verbose=False))
ep_b = epochs_of(ep_b_src)
fig, ax = plt.subplots(figsize=(9, 4.2))
for ep, label, color in ((ep_a, "A: high-pass 0.1 Hz", "tab:blue"), (ep_b, "B: high-pass 1.0 Hz", "tab:red")):
if ep is None:
continue
d = l2.difference_wave(ep)
amp = l2.mean_amplitude(d, CH, WINDOW)
measure_rows.append({"run": label, f"{CH} mean (uV)": amp, "trials": int(d.nave)})
ax.plot(d.times * 1000, d.data[d.ch_names.index(CH)] * 1e6, lw=1.6, color=color,
label=f"{label}: {amp:+.2f} uV")
ax.axvspan(WINDOW[0] * 1000, WINDOW[1] * 1000, color="tab:orange", alpha=0.18, label="a-priori window")
ax.axhline(0, color="gray", lw=0.6); ax.axvline(0, color="gray", lw=0.6)
ax.set(xlabel="Time from stimulus (ms)", ylabel="Amplitude (uV)",
title=f"sub-001: target minus standard at {CH} under the two configurations (uV, positive up)")
ax.legend(fontsize=8); ax.grid(alpha=0.3)
fig.tight_layout()
plt.show() # render the static figure(s) of this cell inline
print(l2.fmt_table(measure_rows, list(measure_rows[0]) if measure_rows else [], floatfmt="{:+.2f}"))
6. A failure is a row, not a stopped cohort¶
A subject whose file is missing, whose events cannot be read, or whose ICA does not converge must appear in the report as exactly that. It must not raise out of the loop and end the run at subject 7 of 40.
bad = run_one(config, "sub-999") if EEGPIPE is None else run_one(config, "sub-999", progress=False)
B = norm(bad)
print(f"ok = {B['ok']}")
print(f"error recorded: {str(B['error'])[:200]}")
print(f"the run log still exists, with {len(B['steps'])} step(s):")
print(l2.fmt_table([{"step": s["name"], "duration (s)": s["duration_s"],
"notes": str(s.get("notes", ""))[:70]} for s in B["steps"]]
or [{"step": "(none completed)", "duration (s)": 0.0, "notes": ""}],
["step", "duration (s)", "notes"], floatfmt="{:.2f}"))
qc_bad = B["outputs"].get("qc")
if qc_bad and Path(qc_bad).exists():
print("\nand the QC report renders, saying what happened:")
display(HTML(Path(qc_bad).read_text(encoding="utf-8")))
else:
print("\nand the QC report renders, saying what happened:")
display(HTML(l2.qc_report_html(bad, title="QC report -- sub-999 (failed)")
if EEGPIPE is None else f"<p><b>sub-999 failed:</b> {B['error']}</p>"))
7. The numbers¶
print("nb-2-8-pipeline -- L2.8 numbers (draft; TODO(confirm) at author review)")
print(f"Implementation: " + (f"pipelines/eegpipe {EEGPIPE.__version__}" if EEGPIPE is not None
else "helpers_l2.run_subject (the documented interface; pipelines/eegpipe absent)"))
print(f"Configuration: " + (str(CONFIG_PATH.relative_to(_pipelines.parent)) if EEGPIPE is not None
else "helpers_l2.PipelineConfig defaults")
+ f", hash {R['config_hash']}, seed {R['seed']}")
print(f"Data: ds-erpcore P3 sub-001 (CC BY-SA 4.0 per data/directory.yaml, contested at source, open, per-subject downloadable).")
print(f"Canonical order executed: "
+ (" -> ".join(EEGPIPE.CANONICAL_ORDER) if EEGPIPE is not None
else "load -> montage -> bad-channel detection -> filter -> interpolate -> re-reference -> ICA -> "
"epoch -> reject"))
print()
print("Run log (step, duration, notes):")
for s in R["steps"]:
print(f" {s['name']:<16s} {s['duration_s']:6.2f} s {str(s.get('notes', ''))[:96]}")
print(f" {'TOTAL':<16s} {R['duration_s']:6.2f} s")
print()
print(f"Bad channels: {R['interpolated'] or 'none'} interpolated of {len(R['bads'])} flagged "
f"({', '.join(f'{ch}: {c}' for ch, c in R['bads'].items()) or 'nothing flagged'})")
print(f"Rank: {R['rank']} -- {R['rank_why']}; after ICA cleaning {R['ica']['rank_after']}")
fa = R["filter"].get("analysis") or R["filter"]
print(f"Filter as resolved (analysis branch): {fa.get('l_freq_hz')}-{fa.get('h_freq_hz')} Hz, "
f"{fa.get('method', '?')}, phase {fa.get('phase', '?')}, "
f"{fa.get('filter_length_samples', fa.get('fir_length_samples', '?'))} taps = "
f"{fa.get('filter_length_s', fa.get('fir_length_s', float('nan'))):.1f} s")
if isinstance(R["filter"].get("ica_copy"), dict):
ic = R["filter"]["ica_copy"]
print(f"Filter as resolved (ICA branch): {ic.get('l_freq_hz')}-{ic.get('h_freq_hz')} Hz, "
f"{ic.get('filter_length_samples', '?')} taps = {ic.get('filter_length_s', float('nan')):.1f} s")
print(f"ICA: {R['ica']['method']}, {R['ica']['n_components']} components, seed {R['ica']['seed']}, "
f"fitted on {R['ica']['fitted_on']}; {len(R['ica']['removed'])} removed:")
for r in R["ica"]["removed"]:
print(f" IC{r['component']:<3} {str(r['class']):16s} score {r['probability']:.3f} {r['evidence'][:96]}")
print("Rejection:")
for r in R["rejection"]["table"]:
print(f" {str(r['condition']):9s} presented {r['n_presented']:3d} rejected {r['n_rejected']:3d} "
f"kept {r['n_kept']:3d} ({r['percent_rejected']:.2f} %)")
print(f" per-condition imbalance {R['rejection']['imbalance_pp']:.2f} pp; criterion "
f"{R['rejection']['criterion']}")
if measure_rows:
print()
print("Measured P3 under the two configurations (the only difference is the analysis high-pass):")
for m in measure_rows:
print(f" {m['run']:26s} {CH} mean amplitude {m[f'{CH} mean (uV)']:+5.2f} uV "
f"({m['trials']} trials)")
print()
print(f"Versions recorded in the run log: {R['versions']}")
print()
print("L2.8's exercises are an ordering exercise (ex-2-8-pipeline-order; key: load, montage, bad-channel "
"detection, filter, interpolate, re-reference, ICA, epoch, reject) and a free response; this notebook "
"produces no numeric key. The QC report in section 4 is the C2 deliverable in miniature.")
_out.cleanup()
print("\ntemporary output directory removed; the ERP CORE downloads stay in the cache.")