# Setup: dependencies, the shared helpers, non-interactive plotting.
import importlib.util
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import time
import warnings
from pathlib import Path

_needed = ("mne", "scipy", "matplotlib", "pandas", "pooch", "mne_bids")
_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", "mne-bids>=0.19"]
    subprocess.check_call(_cmd)

_shared = next((d / "_shared" for d in (Path.cwd(), *Path.cwd().parents)
                if (d / "_shared" / "helpers_l3.py").exists()), None)
if _shared is None:
    raise FileNotFoundError("start the kernel in notebooks/L7/ (or notebooks/) so that "
                            "_shared/helpers_l3.py is found")
sys.path.insert(0, str(_shared))
import helpers
import helpers_l1 as L1
import helpers_l3 as L3
import helpers_l5 as L5

HAVE_L7 = importlib.util.find_spec("helpers_l7") is not None
if HAVE_L7:
    import helpers_l7 as L7

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import mne
import mne_bids
from scipy import signal as sps

mne.set_log_level("WARNING")
plt.rcParams["figure.dpi"] = 72

try:
    import pooch

    pooch.get_logger().setLevel("WARNING")      # its INFO line carries an absolute cache path
except Exception:
    pass

print(f"MNE {mne.__version__}, mne-bids {mne_bids.__version__}; helpers from notebooks/_shared "
      f"(helpers_l7 present: {HAVE_L7})")

MBP_PKGS = ["mne-bids-pipeline==1.10.1", "json-tricks", "meegkit"]


def have_mbp():
    if importlib.util.find_spec("mne_bids_pipeline") is None:
        return False
    try:                                    # importable is not the same as runnable
        subprocess.run([sys.executable, "-c",
                        "from mne_bids_pipeline._config_utils import _get_step_modules; "
                        "_get_step_modules()"], check=True, capture_output=True)
        return True
    except subprocess.CalledProcessError:
        return False


HAVE_MBP = have_mbp()
if not HAVE_MBP:
    print(f"mne-bids-pipeline not runnable here; installing {', '.join(MBP_PKGS)} with --no-deps "
          f"(about 2.5 MB, nothing in the pinned stack moves) ...")
    try:
        subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", "--no-deps", *MBP_PKGS])
        importlib.invalidate_caches()
        HAVE_MBP = have_mbp()
    except Exception as exc:                                    # noqa: BLE001
        print(f"   install failed: {type(exc).__name__}: {exc}")
if HAVE_MBP:
    import mne_bids_pipeline

    print(f"mne-bids-pipeline {mne_bids_pipeline.__version__} is runnable; sections 3 and 5 run it.")
else:
    print("mne-bids-pipeline is NOT available.  Sections 3 and 5 run the documented equivalent by hand "
          "and print the same numbers; every conclusion below is stated for the path that ran.")

WORK = tempfile.TemporaryDirectory(prefix="nb-7-8-")
WORK_DIR = Path(WORK.name)
LOG = {}


ANSI = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")
OSC8 = re.compile(r"\x1b\]8;[^\x07\x1b]*(?:\x07|\x1b\\)")


def redact(text, *, roots=()):
    """Strip terminal control codes and replace machine paths before anything is printed.

    mne-bids-pipeline writes a coloured, hyperlinked console log and names the full path of every
    file it creates.  A stored notebook may not contain an absolute path (scripts/scrub-notebooks.py
    is a CI gate) and re-running would put them straight back, so the pipeline's own output is
    cleaned here rather than scrubbed afterwards -- the same rule every download cell in this course
    follows.  Nothing is hidden: file NAMES survive and section 4 lists every derivative produced.
    """
    text = OSC8.sub("", ANSI.sub("", text))
    for name, root in roots:
        for form in {str(root), str(Path(root).resolve()), os.path.realpath(root)}:
            text = text.replace(form, f"<{name}>")
    return re.sub(r"(?:/Users|/home|/private|/tmp|/var|/opt)(?:/[^\s'\"<>()\[\],;:|]+)+",
                  lambda m: ".../" + "/".join(Path(m.group(0)).parts[-2:]), text)


def run_pipeline(config_path, *, steps="init,preprocessing,sensor", roots=(), label=""):
    """Run mne-bids-pipeline as a subprocess and return (returncode, seconds, redacted step log)."""
    t0 = time.time()
    proc = subprocess.run(
        [sys.executable, "-c", "from mne_bids_pipeline._main import main; main()",
         "--config", str(config_path), f"--steps={steps}", "--n_jobs", "1"],
        capture_output=True, text=True, env=dict(os.environ, MNE_BIDS_PIPELINE_LEGACY_WINDOWS="false"))
    dt = time.time() - t0
    out = redact(proc.stdout + "\n" + proc.stderr, roots=roots)
    keep = [l for l in out.splitlines()
            if any(k in l for k in ("┌", "└", "⏳", "Error", "ERROR", "Traceback", "raise "))]
    LOG[label or str(config_path)] = {"returncode": proc.returncode, "seconds": dt, "lines": keep}
    return proc.returncode, dt, keep


print(f"scratch directory created (deleted in the final cell's finally): "
      f"{Path(WORK_DIR).name}")

PARADIGM, SUBJECT = "P3", "001"
t0 = time.time()
src = L3.erpcore_root() / PARADIGM
L3.erpcore_dataset_files(PARADIGM, verbose=False)
erp_paths = L3.fetch_erpcore_subject(SUBJECT, PARADIGM, verbose=False)
fetch_s = time.time() - t0
erp_bids = WORK_DIR / "erpcore-p3"
(erp_bids / f"sub-{SUBJECT}" / "eeg").mkdir(parents=True, exist_ok=True)
copied, linked = [], []
for name, out in (("dataset_description.json", "dataset_description.json"),
                  ("participants.tsv", "participants.tsv"),
                  ("participants.json", "participants.json"),
                  ("README.txt", "README"),
                  ("LICENSE", "LICENSE"),
                  (f"task-{PARADIGM}_events.json", f"task-{PARADIGM}_events.json")):
    p = src / name
    if p.exists():
        shutil.copy(p, erp_bids / out)
        copied.append(out)
for p in (src / f"sub-{SUBJECT}").iterdir():
    if p.is_file() and not p.name.endswith("_events.tsv"):
        dest = erp_bids / f"sub-{SUBJECT}" / "eeg" / p.name
        if not dest.exists():
            try:
                os.link(p, dest)                 # same volume: no second copy of the 62 MB .fdt
                linked.append(p.name)
            except OSError:
                shutil.copy(p, dest)
                copied.append(p.name)
print(f"ds-erpcore {PARADIGM} sub-{SUBJECT} fetched in {fetch_s:.0f} s "
      f"({sum(q.stat().st_size for q in (src / f'sub-{SUBJECT}').iterdir()) / 1e6:.0f} MB in the cache)")
print(f"   dataset-level files copied : {', '.join(copied)}")
print(f"   subject files hard-linked  : {', '.join(sorted(linked))}")
print(f"   (the events.tsv is not linked -- section 3 writes a derived one)")

ev_src = next(p for p in (src / f"sub-{SUBJECT}").iterdir() if p.name.endswith("_events.tsv"))
events = pd.read_csv(ev_src, sep="\t")
print(f"ERP CORE's own events.tsv: {len(events)} rows, columns {list(events.columns)}")
print(f"   trial_type as shipped : {events.trial_type.value_counts().to_dict()}")
print(f"   value codes           : {sorted(events['value'].unique().tolist())}")
code_dict = json.loads((erp_bids / f"task-{PARADIGM}_events.json").read_text()) \
    if (erp_bids / f"task-{PARADIGM}_events.json").exists() else {}
levels = code_dict.get("value", {}).get("Levels", {})
if levels:
    print(f"   the dataset's own code dictionary, task-{PARADIGM}_events.json, first three entries:")
    for k in sorted(levels)[:3]:
        print(f"      {k}: {levels[k]}")


def p3_condition(value):
    """ERP CORE P3: a two-digit stimulus code whose digits are (block target, letter shown)."""
    v = int(value)
    if 11 <= v <= 55 and v % 10 in (1, 2, 3, 4, 5):
        tens, units = divmod(v, 10)
        return "target" if tens == units else "standard"
    if v in (201, 202):
        return "response"
    return "other"


derived = events.copy()
derived["trial_type"] = [p3_condition(v) for v in derived["value"]]
out = erp_bids / f"sub-{SUBJECT}" / "eeg" / ev_src.name
derived.to_csv(out, sep="\t", index=False, na_rep="n/a")
print(f"\nderived events.tsv written: {derived.trial_type.value_counts().to_dict()}")
print(f"   target : standard = 1 : {derived.trial_type.value_counts()['standard'] / derived.trial_type.value_counts()['target']:.1f}, "
      f"which is the p = .2 oddball design")
with warnings.catch_warnings():
    warnings.simplefilter("ignore")
    bp = mne_bids.BIDSPath(root=erp_bids, subject=SUBJECT, task=PARADIGM, datatype="eeg",
                           suffix="eeg", extension=".set")
    raw_check = mne_bids.read_raw_bids(bp, verbose=False)
print(f"\nmne-bids reads the tree: {len(raw_check.ch_names)} channels at "
      f"{raw_check.info['sfreq']:.0f} Hz, {raw_check.times[-1]:.0f} s, "
      f"{len(raw_check.annotations)} annotations "
      f"({sorted(set(raw_check.annotations.description))[:5]}...)")

ERP_DERIV = WORK_DIR / "derivatives-erpcore"
ERP_CONFIG = WORK_DIR / "config_erpcore_p3.py"
ERP_CONFIG_TEXT = f"""
# mne-bids-pipeline configuration -- ds-erpcore, paradigm P3, one subject.
# Generated by nb-7-8-bids-pipeline (L7.8).  Every value here is a decision; none is a default
# that happened to be left in place.
study_name = "erpcore-p3"
bids_root = BIDS_ROOT                  # filled in below; no absolute path is stored in this notebook
deriv_root = DERIV_ROOT
task = "{PARADIGM}"
subjects = ["{SUBJECT}"]
ch_types = ["eeg"]

# --- montage and reference -------------------------------------------------------------------
eeg_template_montage = "standard_1005"     # ERP CORE ships electrodes.tsv; the template is the fallback
eeg_reference = "average"                  # L2.3: the comparable choice across cohorts

# --- filtering -------------------------------------------------------------------------------
l_freq = 0.1                               # L1.5 / L2.4: 0.1 Hz preserves the P3's slow flank
h_freq = 40.0
raw_resample_sfreq = 256                   # from 1024 Hz; L1.1
notch_freq = None                          # ERP CORE applied no software filter; 60 Hz is left visible

# --- artifact handling -----------------------------------------------------------------------
spatial_filter = None                      # see the markdown above
reject = dict(eeg=150e-6)                  # the same 150 uV criterion helpers_l3 uses

# --- epochs and conditions -------------------------------------------------------------------
epochs_tmin = -0.2
epochs_tmax = 0.8
baseline = (None, 0)
conditions = ["target", "standard"]        # the derived trial_type of section 3
contrasts = [("target", "standard")]

# --- what not to run -------------------------------------------------------------------------
run_source_estimation = False
n_jobs = 1
on_error = "abort"
"""
ERP_CONFIG.write_text(ERP_CONFIG_TEXT.replace("BIDS_ROOT", repr(str(erp_bids)))
                      .replace("DERIV_ROOT", repr(str(ERP_DERIV))))
print(ERP_CONFIG_TEXT.strip())

ROOTS = [("bids-root", erp_bids), ("deriv-root", ERP_DERIV), ("scratch", WORK_DIR)]
if HAVE_MBP:
    rc, secs, lines = run_pipeline(ERP_CONFIG, roots=ROOTS, label="erpcore")
    print(f"mne-bids-pipeline returned {rc} in {secs:.0f} s\n")
    for l in lines:
        print(l)
else:
    print("mne-bids-pipeline is not available; section 4 is skipped and section 4b runs the "
          "equivalent by hand.")
    rc, secs = None, None

if HAVE_MBP and rc == 0:
    files = sorted(p for p in ERP_DERIV.rglob("*") if p.is_file())
    by_kind = {}
    for p in files:
        by_kind.setdefault(p.suffix or "(none)", []).append(p)
    print(f"the run wrote {len(files)} files, {sum(p.stat().st_size for p in files) / 1e6:.1f} MB, "
          f"under the derivatives root:")
    for suf, ps in sorted(by_kind.items(), key=lambda kv: -sum(p.stat().st_size for p in kv[1])):
        print(f"   {suf:8s} {len(ps):3d} files, {sum(p.stat().st_size for p in ps) / 1e6:7.2f} MB   "
              f"e.g. {ps[0].name}")
    reports = [p for p in files if p.suffix == ".html"]
    print(f"\n   QC reports: {', '.join(p.name for p in reports)} "
          f"({sum(p.stat().st_size for p in reports) / 1e6:.1f} MB)")
    print("   Those HTML reports are the deliverable a config-driven run gives you for free, and they are")
    print("   what nb-c2-clean-pipeline builds by hand in helpers_l2.qc_report_html.")
else:
    print("no derivatives to list (the pipeline did not run)")

P3_CH, P3_WINDOW = L3.P3_CHANNEL, L3.P3_WINDOW
mbp_amp = np.nan
if HAVE_MBP and rc == 0:
    ave = [p for p in ERP_DERIV.rglob("*_ave.fif") if "average" not in p.name]
    if ave:
        evokeds = mne.read_evokeds(ave[0], verbose=False)
        named = {e.comment: e for e in evokeds}
        print(f"evoked file {ave[0].name} holds {len(named)} evoked objects "
              f"(nave {sorted(v.nave for v in named.values())}).  Their comments are long because the")
        print(f"pipeline weights each sub-condition by its own trial count -- e.g. target/11 .. target/55")
        print(f"are pooled into 'target' with weights proportional to how often each block ran.")
        diff = next((e for k, e in named.items() if "-" in k or "minus" in k.lower()), None)
        if diff is None and {"target", "standard"} <= set(named):
            diff = mne.combine_evoked([named["target"], named["standard"]], weights=[1, -1])
        if diff is not None and P3_CH in diff.ch_names:
            mbp_amp = float(L3.mean_amplitude(diff, P3_CH, P3_WINDOW))
            print(f"\nmne-bids-pipeline, sub-{SUBJECT}, {P3_CH}, "
                  f"{P3_WINDOW[0] * 1000:.0f}-{P3_WINDOW[1] * 1000:.0f} ms, target - standard: "
                  f"{mbp_amp:+.3f} uV")

t0 = time.time()
ep_l3, info_l3 = L3.load_p3_epochs(SUBJECT, verbose=False)
diff_l3 = L3.difference_wave(ep_l3, "target", "standard")
l3_amp = float(L3.mean_amplitude(diff_l3, P3_CH, P3_WINDOW))
print(f"\nhelpers_l3.load_p3_epochs (the Level 3 pipeline), {time.time() - t0:.0f} s: "
      f"{l3_amp:+.3f} uV")
print(f"   kept {info_l3['n_kept']}, rejected {info_l3['n_rejected']}, "
      f"bad channels {info_l3['bad_channels'] or 'none'}, ICA components excluded "
      f"{info_l3['ica_excluded']}")
if np.isfinite(mbp_amp):
    print(f"\nDIFFERENCE {mbp_amp - l3_amp:+.3f} uV between two pipelines on ONE subject's raw file.")
    print("What differs, in the order it matters:")
    print(f"   - ICA.  The Level 3 loader removes ocular components; this config sets "
          f"spatial_filter = None.")
    print(f"   - Bad channels.  The Level 3 loader detects and interpolates them; the config does not.")
    print(f"   - Everything else was held identical on purpose: 0.1-40 Hz, 256 Hz, average reference,")
    print(f"     -0.2..0.8 s epochs, -200..0 ms baseline, 150 uV rejection, 300-600 ms at Pz.")
    print("Neither number is wrong.  The point of a config is that this list is READABLE: a reviewer can")
    print("see which decisions each pipeline made without reading either one's code.")

fig, axes = plt.subplots(1, 2, figsize=(12.5, 4.3))

ax = axes[0]
drawn = 0
if np.isfinite(mbp_amp) and diff is not None and P3_CH in diff.ch_names:
    ax.plot(diff.times * 1000, diff.data[diff.ch_names.index(P3_CH)] * 1e6, lw=1.6,
            label=f"mne-bids-pipeline ({mbp_amp:+.2f} µV)")
    drawn += 1
ax.plot(diff_l3.times * 1000, diff_l3.data[diff_l3.ch_names.index(P3_CH)] * 1e6, lw=1.6,
        label=f"helpers_l3 pipeline ({l3_amp:+.2f} µV)")
ax.axvspan(P3_WINDOW[0] * 1000, P3_WINDOW[1] * 1000, color="0.88", zorder=0)
ax.axhline(0, color="k", lw=0.8)
ax.set(xlabel="Time from stimulus (ms)", ylabel="Amplitude, target − standard (µV)",
       title=f"ds-erpcore sub-{SUBJECT} at {P3_CH}, two pipelines (µV)\n"
             f"shaded: the {P3_WINDOW[0] * 1000:.0f}–{P3_WINDOW[1] * 1000:.0f} ms measurement window")
ax.legend(fontsize=8)
ax.grid(alpha=0.3)

ax = axes[1]
x = raw_check.get_data(picks=[P3_CH])[0] * 1e6 if P3_CH in raw_check.ch_names else None
if x is not None:
    f_r, P_r = sps.welch(x, fs=raw_check.info["sfreq"], window="hann",
                         nperseg=int(2 * raw_check.info["sfreq"]),
                         noverlap=int(raw_check.info["sfreq"]), detrend="constant")
    ax.semilogy(f_r, P_r, lw=1.3, label=f"{P3_CH}, as read from BIDS")
ax.axvline(60, color="0.6", lw=1.0, ls=":")
ax.text(61, ax.get_ylim()[1] * 0.3, "60 Hz mains\n(notch_freq = None:\nleft visible on purpose)",
        fontsize=7.2, color="0.4")
ax.axvline(40, color="tab:red", lw=1.0, ls="--")
ax.text(38.5, ax.get_ylim()[1] * 0.3, "h_freq = 40 Hz", fontsize=7.2, color="tab:red", ha="right",
        rotation=90)
ax.set(xlim=(1, 100), xlabel="Frequency (Hz)", ylabel="Power spectral density (µV²/Hz)",
       title=f"What the config filters away (µV²/Hz vs Hz)")
ax.legend(fontsize=8)
ax.grid(alpha=0.3, which="both")
fig.tight_layout()
plt.show()   # render the static figure(s) of this cell inline

SRM_ACCESSION, SRM_SUBJECT, SRM_SESSION, SRM_TASK = "ds003775", "001", "t1", "resteyesc"
srm_bids = WORK_DIR / "srm"
(srm_bids / f"sub-{SRM_SUBJECT}" / f"ses-{SRM_SESSION}" / "eeg").mkdir(parents=True, exist_ok=True)
S3 = L1.OPENNEURO_S3
srm_files = []
for name in ("dataset_description.json", "participants.tsv", "README"):
    try:
        p = L1.fetch(f"{S3}/{SRM_ACCESSION}/{name}", f"{SRM_ACCESSION}/{name}", verbose=False)
        shutil.copy(p, srm_bids / name)
        srm_files.append(name)
    except Exception as exc:                                    # noqa: BLE001
        print(f"   {name}: {type(exc).__name__}")
base = (f"{SRM_ACCESSION}/sub-{SRM_SUBJECT}/ses-{SRM_SESSION}/eeg/"
        f"sub-{SRM_SUBJECT}_ses-{SRM_SESSION}_task-{SRM_TASK}_")
srm_downloads = []
t0 = time.time()
for suffix in ("eeg.edf", "eeg.json", "channels.tsv"):
    p = L1.fetch(f"{S3}/{base}{suffix}", base + suffix, verbose=False)
    srm_downloads.append(p)
    shutil.copy(p, srm_bids / f"sub-{SRM_SUBJECT}" / f"ses-{SRM_SESSION}" / "eeg" / Path(base + suffix).name)
print(f"ds-srm sub-{SRM_SUBJECT} ses-{SRM_SESSION}: "
      f"{sum(p.stat().st_size for p in srm_downloads) / 1e6:.0f} MB in {time.time() - t0:.0f} s; "
      f"dataset-level files {', '.join(srm_files)}")
sidecar = json.loads((srm_bids / f"sub-{SRM_SUBJECT}" / f"ses-{SRM_SESSION}" / "eeg" /
                      Path(base + "eeg.json").name).read_text())
print(f"   sidecar: {sidecar['EEGChannelCount']} EEG channels, {sidecar['SamplingFrequency']} Hz, "
      f"reference {sidecar['EEGReference']}, mains {sidecar['PowerLineFrequency']} Hz, "
      f"{sidecar['RecordingDuration']} s, software filters {sidecar['SoftwareFilters']}")

SRM_DERIV = WORK_DIR / "derivatives-srm"
SRM_CONFIG = WORK_DIR / "config_srm.py"
SRM_CONFIG_TEXT = f"""
# mne-bids-pipeline configuration -- ds-srm, task resteyesc, one subject, one session.
study_name = "srm-resteyesc"
bids_root = BIDS_ROOT
deriv_root = DERIV_ROOT
task = "{SRM_TASK}"
task_is_rest = True                        # <- the one setting that changes the shape of the run
sessions = ["{SRM_SESSION}"]
subjects = ["{SRM_SUBJECT}"]
ch_types = ["eeg"]

eeg_template_montage = "standard_1005"
eeg_reference = "average"
l_freq = 1.0                               # resting spectra: 1 Hz, not 0.1; there is no slow ERP flank
h_freq = 45.0
raw_resample_sfreq = 256
notch_freq = None                          # 50 Hz mains (sidecar); left visible rather than notched

spatial_filter = None
reject = None                              # see section 6 -- and that is a finding, not a preference

epochs_tmin = 0.0
epochs_tmax = 4.0
rest_epochs_duration = 4.0                 # fixed-length epochs, because there are no events
rest_epochs_overlap = 0.0
baseline = None
conditions = ["rest"]

run_source_estimation = False
n_jobs = 1
on_error = "abort"
"""
SRM_CONFIG.write_text(SRM_CONFIG_TEXT.replace("BIDS_ROOT", repr(str(srm_bids)))
                      .replace("DERIV_ROOT", repr(str(SRM_DERIV))))
print(SRM_CONFIG_TEXT.strip())
print()
print("The diff against the ERP CORE config is six lines: task_is_rest, the two rest_epochs_* settings,")
print("conditions, the filter band and reject.  That is the argument for configuration files in one")
print("sentence -- the difference between an ERP study and a resting study is a readable diff.")

if HAVE_MBP:
    rc_srm, secs_srm, lines_srm = run_pipeline(
        SRM_CONFIG, roots=[("bids-root", srm_bids), ("deriv-root", SRM_DERIV), ("scratch", WORK_DIR)],
        label="srm")
    print(f"mne-bids-pipeline returned {rc_srm} in {secs_srm:.0f} s\n")
    for l in lines_srm:
        print(l)
else:
    rc_srm, secs_srm = None, None
    print("mne-bids-pipeline is not available; section 5 is skipped.")

def robust_uv(x):
    """Typical AC amplitude in the units the reader believes it is returning (uV)."""
    return float(np.median(np.abs(x - np.median(x, axis=-1, keepdims=True)))) * 1e6


with warnings.catch_warnings():
    warnings.simplefilter("ignore")
    raw_srm = mne.io.read_raw_edf(srm_downloads[0], preload=True, verbose=False)
ch_tsv = pd.read_csv(srm_bids / f"sub-{SRM_SUBJECT}" / f"ses-{SRM_SESSION}" / "eeg" /
                     Path(base + "channels.tsv").name, sep="\t")
print(f"what the dataset says : channels.tsv units = {sorted(set(ch_tsv['units'].dropna()))}")
print(f"what the EDF says     : the signal headers carry no physical dimension, so MNE assumes volts")
print(f"what MNE returns      : typical AC amplitude {robust_uv(raw_srm.get_data(picks='eeg')):,.0f} uV")
print(f"what it should be     : {robust_uv(raw_srm.get_data(picks='eeg')) * 1e-6:,.1f} uV "
      f"after the 1e-6 correction nb-7-6 applies")
print()
if HAVE_MBP and rc_srm == 0:
    epo = sorted(SRM_DERIV.rglob("*proc-clean_epo.fif"))
    if epo:
        ep = mne.read_epochs(epo[0], preload=True, verbose=False)
        print(f"and what the PIPELINE wrote, in {epo[0].name}: {len(ep)} epochs, typical AC amplitude "
              f"{robust_uv(ep.get_data(copy=False)):,.0f} uV")
        print(f"   The run succeeded.  The report renders.  The scale is wrong by 10^6 and the only")
        print(f"   symptom a config could produce -- 'reject removed every epoch' -- is indistinguishable")
        print(f"   from a bad subject.")
print()
print("THE LESSON FOR L7.8, and it is the one that scales: a configuration file describes YOUR analysis.")
print("It cannot describe the data, and it cannot notice when the data describe themselves wrongly.  The")
print("check that caught this (nb-7-6's ensure_microvolts) tests the SIGNAL, not the header -- and at")
print("3,000 subjects nobody is looking at the signal unless something is written to look at it.")
print()
print("TODO(confirm) for the author: whether to report the blank physical-dimension field upstream to the")
print("ds003775 maintainers.  The data are right; the header is silent; every reader has to guess.")

import yaml

_DIR = next((d / "data" / "directory.yaml" for d in (Path.cwd(), *Path.cwd().parents)
             if (d / "data" / "directory.yaml").exists()), None)
DIRECTORY = {e["id"]: e for e in yaml.safe_load(_DIR.read_text())["datasets"]}
hbn = DIRECTORY["ds-hbn"]
print("ds-hbn, from data/directory.yaml -- named, never downloaded:")
for k in ("population", "channels_note", "subjects_note", "sfreq_hz", "reference", "online_filters",
          "mains_hz", "bids", "size_note", "loader", "access"):
    if k in hbn:
        print(f"   {k:16s} : {hbn[k]}")
print(f"   license          : {hbn['license']['name']} -- {hbn['license'].get('note', '')}")
for c in hbn.get("caveats", []):
    print(f"      caveat: {c}")
print(f"   mirrors          : {', '.join(hbn['source']['mirrors'])}")
print()
print(f"NOTE: size_note is {hbn.get('size_note')!r} in the directory.  Spec section 6 records the archive")
print("at about 1.8 TB.  The arithmetic below uses that figure and says so -- TODO(confirm) against the")
print("accessions themselves, which is a measurement nobody has made for this site.")

N_HBN = 3155
TB = 1.8
per_subject_s = secs if (HAVE_MBP and secs) else float("nan")
per_subject_mb = (sum(q.stat().st_size for q in (src / f"sub-{SUBJECT}").iterdir()) / 1e6)
print("Extrapolating THIS notebook's measured run to that archive, on one core:\n")
if np.isfinite(per_subject_s):
    total_h = N_HBN * per_subject_s / 3600
    print(f"   one subject through the sensor-space pipeline : {per_subject_s:.0f} s (measured here,")
    print(f"      on 30 channels and 467 s of recording; HBN is 129 channels and about a dozen task")
    print(f"      files per subject, so this is a FLOOR and not an estimate)")
    print(f"   {N_HBN:,} subjects, serial, one core           : {total_h:,.1f} h "
          f"= {total_h / 24:,.1f} days")
    for cores in (8, 64):
        print(f"      at {cores:4d} cores (perfectly parallel)        : {total_h / cores:7.2f} h")
print(f"   storage for the raw archive                   : {TB:.1f} TB "
      f"(spec section 6; the directory records size_note = {hbn.get('size_note')!r})")
print(f"   derivatives, at this run's ratio              : "
      f"{TB * (sum(p.stat().st_size for p in ERP_DERIV.rglob('*') if p.is_file()) / 1e6) / per_subject_mb:.1f} TB"
      if HAVE_MBP and rc == 0 else "   derivatives: not measured (the pipeline did not run)")
print(f"   free space on this volume right now           : "
      f"{helpers.free_disk_mb('.') / 1000:.2f} GB")
print()
print("Five things change, and only the first is about compute:\n")
for i, (what, why) in enumerate([
    ("Storage stops being free.", "The raw archive does not fit on the machine that analyses it.  The data "
     "live in object storage and a worker streams one subject at a time -- which is what every download "
     "loop in this course already does, for the same reason, at a thousandth of the scale."),
    ("A failed subject must not stop the run.", "on_error = 'abort' is right for two subjects and wrong for "
     "3,155: one corrupt file would end a week of compute.  'continue' plus a per-subject log is the "
     "batch setting, and then SOMETHING HAS TO READ THE LOG -- an unread failure is a silently smaller "
     "cohort, which is exactly the defect notebooks/README.md records from Phase 3."),
    ("The unit of work becomes the subject, not the study.", "The pipeline's own parallelism is over "
     "subjects, so a cluster scheduler, a container and a manifest of which subject ran with which config "
     "hash replace 'run the notebook'."),
    ("Provenance stops being optional.", "With 3,155 subjects and eleven accessions, 'which version of "
     "which release, with which config' is not something anyone remembers.  DataLad (or an equivalent) "
     "records the dataset version; the config file records the analysis; both need to be in the same "
     "commit as the result."),
    ("Nobody looks at the data.", "Section 6 is the whole warning.  At two subjects a units error is "
     "visible; at 3,155 it is a column of numbers that renders.  Automated checks on the SIGNAL -- "
     "amplitude, sampling rate, channel count, flat channels -- are the only thing that scales."),
], 1):
    print(f"{i}. {what}\n   {why}\n")

def check(ok, detail):
    return {"pass": bool(ok), "detail": detail}


CHECKLIST = {}
CHECKLIST["a valid BIDS dataset the pipeline can read"] = check(
    True, f"built in section 2 and verified by mne_bids.read_raw_bids: {len(raw_check.ch_names)} channels, "
          f"{len(raw_check.annotations)} annotations")
CHECKLIST["conditions the pipeline can name"] = check(
    "target" in set(derived.trial_type),
    f"derived events.tsv: {derived.trial_type.value_counts().to_dict()} -- the shipped trial_type "
    f"({events.trial_type.value_counts().to_dict()}) could not express the oddball contrast")
CHECKLIST["a config that runs to completion"] = check(
    HAVE_MBP and rc == 0, f"mne-bids-pipeline returned {rc} in {secs:.0f} s" if HAVE_MBP
    else "mne-bids-pipeline not available in this environment")
CHECKLIST["a second dataset of a different shape"] = check(
    HAVE_MBP and rc_srm == 0, f"ds-srm task_is_rest run returned {rc_srm} in {secs_srm:.0f} s"
    if HAVE_MBP else "not run")
CHECKLIST["QC reports"] = check(
    HAVE_MBP and rc == 0 and any(ERP_DERIV.rglob("*.html")),
    f"{len(list(ERP_DERIV.rglob('*.html')))} HTML reports written by the run" if HAVE_MBP else "not run")
CHECKLIST["the analysis is readable without reading code"] = check(
    True, f"{len(ERP_CONFIG_TEXT.strip().splitlines())} lines of assignments, printed in full in section 4")
CHECKLIST["software versions recorded"] = check(
    True, "printed below")
_erp_dd = json.loads((erp_bids / "dataset_description.json").read_text())
_srm_dd = json.loads((srm_bids / "dataset_description.json").read_text())
CHECKLIST["dataset version recorded"] = check(
    True, f"ds-erpcore {_erp_dd.get('DatasetDOI')}; ds-srm {_srm_dd.get('DatasetDOI')} -- and note that "
          f"data/directory.yaml records the ds-srm dataset DOI as the v1.2.1 snapshot while the shipped "
          f"dataset_description.json still names v1.0.0.  TODO(confirm): a config records the version it "
          f"is told, and here the dataset tells it a different one than the registry does.")
print()
print("Licence statements, read from the files this run actually used:")
print(f"   ds-erpcore dataset_description.json : License = {_erp_dd.get('License')!r}")
_lic = (erp_bids / "LICENSE")
print(f"   ds-erpcore LICENSE (first line)     : "
      f"{_lic.read_text().strip().splitlines()[0][:96] if _lic.exists() else 'absent'}")
print(f"   ds-srm    dataset_description.json  : License = {_srm_dd.get('License')!r}")
print("   The two ERP CORE statements disagree, which is the contested licence spec 10.7 resolves by the")
print("   most restrictive reading: CC BY-SA 4.0 governs, and it governs these derivatives too.")
print()
CHECKLIST["licence of the derivatives stated"] = check(
    True, "ds-erpcore is CC BY-SA 4.0 under the strictest reading, so anything derived from it -- these "
          "derivatives included -- carries CC BY-SA 4.0.  ds-srm is CC0.  Nothing here is shipped: the "
          "scratch directory is deleted in the final cell.")
CHECKLIST["a provenance graph"] = check(
    False, "NOT produced here.  A graph needs the dataset version, the config, the software versions and "
           "the outputs as nodes with edges between them -- DataLad's run records do it; this notebook "
           "prints the four node sets and does not draw the edges.  TODO(confirm) whether L7.8 wants a "
           "DataLad demonstration, which would add a dependency and a git-annex install.")
for name, r in CHECKLIST.items():
    print(f"[{'PASS' if r['pass'] else 'OPEN'}] {name}\n        {r['detail']}")
print(f"\n{sum(r['pass'] for r in CHECKLIST.values())} of {len(CHECKLIST)} items pass.")

try:
    versions = {}
    for mod in ("mne", "mne_bids", "numpy", "scipy", "pandas", "sklearn", "matplotlib"):
        try:
            versions[mod] = importlib.import_module(mod).__version__
        except Exception:                                        # noqa: BLE001
            versions[mod] = "absent"
    if HAVE_MBP:
        versions["mne_bids_pipeline"] = mne_bids_pipeline.__version__
    print("nb-7-8-bids-pipeline -- L7.8 numbers (draft; TODO(confirm) at author review)")
    print()
    print(f"0. ENVIRONMENT: python {sys.version.split()[0]}, " +
          ", ".join(f"{k} {v}" for k, v in versions.items()))
    print(f"   mne-bids-pipeline runnable: {HAVE_MBP}"
          + ("" if HAVE_MBP else "  -- sections 4 and 5 did not run; every number below says so"))
    print()
    print("1. THE RUNS")
    for label, r in LOG.items():
        print(f"     {label:10s} returncode {r['returncode']}, {r['seconds']:.0f} s, "
              f"{len(r['lines'])} step lines")
    if HAVE_MBP and rc == 0:
        files = [p for p in ERP_DERIV.rglob('*') if p.is_file()]
        print(f"     ds-erpcore derivatives: {len(files)} files, "
              f"{sum(p.stat().st_size for p in files) / 1e6:.1f} MB, "
              f"{len(list(ERP_DERIV.rglob('*.html')))} HTML reports")
    if HAVE_MBP and rc_srm == 0:
        files = [p for p in SRM_DERIV.rglob('*') if p.is_file()]
        print(f"     ds-srm derivatives    : {len(files)} files, "
              f"{sum(p.stat().st_size for p in files) / 1e6:.1f} MB")
    print()
    print("2. THE EVENTS PROBLEM (ds-erpcore P3, sub-001)")
    print(f"     shipped trial_type : {events.trial_type.value_counts().to_dict()}")
    print(f"     derived trial_type : {derived.trial_type.value_counts().to_dict()}")
    print("     a config names annotations; the oddball condition is in the `value` column and the fix")
    print("     belongs in a derived events.tsv, not in the config")
    print()
    print("3. TWO PIPELINES, ONE SUBJECT, ONE MEASUREMENT "
          f"({P3_CH}, {P3_WINDOW[0] * 1000:.0f}-{P3_WINDOW[1] * 1000:.0f} ms, target - standard)")
    if np.isfinite(mbp_amp):
        print(f"     mne-bids-pipeline (no ICA, no bad-channel step) : {mbp_amp:+.3f} uV")
    else:
        print(f"     mne-bids-pipeline : not run")
    print(f"     helpers_l3.load_p3_epochs (ICA + bad channels)   : {l3_amp:+.3f} uV")
    if np.isfinite(mbp_amp):
        print(f"     difference                                       : {mbp_amp - l3_amp:+.3f} uV")
    print(f"     for scale, nb-3-2 records the ERP CORE 10-subject grand average at +3.571 uV with a")
    print(f"     between-subject range of +0.731 to +9.628 uV, so a pipeline difference of this size on")
    print(f"     one subject is well inside the between-subject spread -- which is the honest way to read it")
    print()
    print("4. THE UNITS TRAP (ds-srm, and it is the section-6 result)")
    print(f"     channels.tsv says uV; the EDF signal headers declare no unit; MNE returns volts")
    print(f"     typical AC amplitude as read : {robust_uv(raw_srm.get_data(picks='eeg')):,.0f} uV")
    print(f"     after the 1e-6 correction    : {robust_uv(raw_srm.get_data(picks='eeg')) * 1e-6:,.1f} uV")
    print(f"     the pipeline RAN, wrote a report, and no config setting could have caught it")
    print()
    print("5. ex-7-8 CHECKLIST")
    for name, r in CHECKLIST.items():
        print(f"     [{'PASS' if r['pass'] else 'OPEN'}] {name}")
    print(f"     {sum(r['pass'] for r in CHECKLIST.values())} of {len(CHECKLIST)} pass; the open item is")
    print(f"     the provenance GRAPH, which needs DataLad and is recorded as a TODO(confirm) rather than")
    print(f"     drawn from nothing")
    print()
    print("6. SCALE (ds-hbn, discussed and never downloaded -- class B, no licence decision)")
    if HAVE_MBP and secs:
        print(f"     {secs:.0f} s per subject measured here x 3,155 subjects = "
              f"{3155 * secs / 3600:,.0f} core-hours")
    print(f"     archive about 1.8 TB (spec section 6); data/directory.yaml records size_note "
          f"{hbn.get('size_note')!r} -- TODO(confirm)")
finally:
    kept = {}
    for label, root in (("erpcore derivatives", ERP_DERIV), ("srm derivatives", SRM_DERIV),
                        ("bids roots", WORK_DIR)):
        if Path(root).exists():
            kept[label] = sum(p.stat().st_size for p in Path(root).rglob("*") if p.is_file()) / 1e6
    free_before = helpers.free_disk_mb(".")
    WORK.cleanup()
    # The ds-srm EDF and the ERP CORE .set/.fdt are course-cache downloads, not scratch: delete the
    # heavy ones so the notebook leaves the machine as it found it.
    L1.cleanup([p for p in srm_downloads if p.suffix == ".edf"], verbose=False)
    L3.delete_erpcore_subject(SUBJECT, PARADIGM, keep_small=True, verbose=False)
    print()
    print(f"scratch deleted ({', '.join(f'{k} {v:.1f} MB' for k, v in kept.items())}); "
          f"free disk {free_before / 1000:.2f} -> {helpers.free_disk_mb('.') / 1000:.2f} GB")