From 73a7be84864516322954cb9831f98100b1ef04fb Mon Sep 17 00:00:00 2001 From: Juniper Bevensee Date: Mon, 22 Jun 2026 16:35:43 +1200 Subject: [PATCH] feat(study): MEG data-sample validation study (M100, memory-bounded, resumable) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a meg_validation study type — the first heavy / real-data study built on the stateful study pipeline. Validates the auditory M100 evoked-response claim (magnetic peak ~80-120 ms) against a bounded sample of an open MEG dataset, as five resumable checkpointed steps: fetch_sample -> preprocess -> epoch -> evoked -> validate_finding. - engine/meg_study.py: the five steps + a memory-bounded real mne backend (_MneIO) and an injected-I/O contract (MegIO). Heavy deps (mne/numpy) are imported ONLY inside backend methods — the module and the whole plugin import fine with neither installed. The real backend never full-preloads: preload= False, single run, gradiometer subset, crop, downsample; each step reloads the prior step's on-disk artifact, so an OOM/restart resumes from the last completed step instead of re-fetching. validate_finding emits supported (peak in window) / refuted (clearly outside) / inconclusive (degenerate sample), with measured latency + amplitude as evidence. - tools.py: wire meg_validation into matilde_study_create (default plan + params: dataset_id, expected_window_ms, bounds) and matilde_study_run dispatch, mirroring the bibliography dispatch; handlers stay thin. - tests: stdlib + fakes (no mne/numpy/network/data) — step state transitions, verdict boundaries (100ms supported / 300ms refuted / empty inconclusive), custom window, and resumability proven with call counters (fail at evoked, resume, earlier steps not re-run). Plus an opt-in live test (MATILDE_LIVE=1 + mne) on bst_auditory, skipped otherwise. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/meg-validation-study.md | 66 ++++++ matilde_plugin/engine/meg_study.py | 338 +++++++++++++++++++++++++++++ matilde_plugin/tools.py | 33 ++- tests/test_meg_study.py | 243 +++++++++++++++++++++ tests/test_meg_study_tools.py | 83 +++++++ 5 files changed, 761 insertions(+), 2 deletions(-) create mode 100644 docs/meg-validation-study.md create mode 100644 matilde_plugin/engine/meg_study.py create mode 100644 tests/test_meg_study.py create mode 100644 tests/test_meg_study_tools.py diff --git a/docs/meg-validation-study.md b/docs/meg-validation-study.md new file mode 100644 index 0000000..5d10cb7 --- /dev/null +++ b/docs/meg-validation-study.md @@ -0,0 +1,66 @@ +# MEG data-sample validation study (design) + +Builds on the stateful study pipeline (`docs/stateful-study-pipeline.md`). First +**heavy / real-data** study type: validate a reported neuroscience finding against +a **bounded sample** of an open MEG dataset, as resumable checkpointed steps. + +## Why "data sample", not full-dataset + +Full-preloading a multi-GB MEG recording into the agent process exhausts container +memory. This study deliberately validates against a **bounded sample** — +enough signal to test the claimed effect, never the whole recording. Combined +with the pipeline's per-step checkpointing, a memory kill or container restart +**resumes from the last completed step** instead of restarting the analysis. + +## Worked claim (first target) + +The auditory **M100 / N100m** evoked response: a magnetic field peak ~80–120 ms +after an auditory stimulus. Open data: the Brainstorm auditory sample +(`mne.datasets.brainstorm.bst_auditory`, the ds000246 tutorial data) — standard +tones, a well-established M100. The study measures the peak latency of the evoked +response to standard tones and validates it falls within the expected window. + +## Steps (resumable; each checkpoints an artifact + may emit a finding) + +1. `fetch_sample` — obtain a bounded slice: download/locate the dataset, but load + **memory-bounded** — `preload=False`, pick one run, a subset of gradiometer + channels, crop to a stimulus-locked window, and downsample. Persist a small + intermediate (e.g. epochs/evoked `.npy` or `.fif`) as an artifact; never hold + the full recording. +2. `preprocess` — band-pass filter (e.g. 1–40 Hz) the bounded sample; checkpoint. +3. `epoch` — epoch around standard-tone events (the analysis must read the events + from the data, not assume counts); checkpoint epoch metadata. +4. `evoked` — average to the evoked response; checkpoint the evoked array artifact. +5. `validate_finding` — measure peak latency in the M100 window; emit a finding: + `supported` if within the expected window (default 80–120 ms), `refuted` if + clearly outside, `inconclusive` if SNR too low / sample too small. Record the + measured latency + amplitude as evidence. + +## Implementation rules (match the repo) + +- **Lazy heavy deps**: `mne`/`numpy` imported **inside** the step functions, never + at module import — the plugin must load (and the rest of the engine stay + stdlib-only) without mne installed. The agent container already has mne available + at runtime. +- **Injected I/O for tests**: the step functions take injected loader/analysis + handles so the whole study is testable with **fakes** (no mne, no network, no + data) — deterministic, fast, stdlib-only. The real mne path is exercised only at + runtime and by an **opt-in** live test (`MATILDE_LIVE=1`, skipped in CI). +- **Wire into the framework**: add a `meg_validation` study type the + `matilde_study_create` plan can request (params: dataset id, expected-window ms, + channel/sample bounds), runnable via `matilde_study_run` and resumable like any + study. + +## Tests + +- Unit (stdlib, fakes): each step's logic; `validate_finding` verdict boundaries + (within window → supported; far outside → refuted; degenerate sample → + inconclusive); resumability (fail at `evoked`, resume, earlier steps not re-run). +- Opt-in live (`MATILDE_LIVE=1`): run the real bounded mne pipeline on + `bst_auditory`, assert a peak is found in a plausible window and the study + completes `done` within a bounded memory footprint. Skipped without the flag. + +## Out of scope (later) + +Multi-subject/meta-analysis, source localization, statcheck/GRIM re-checking — +separate study types once this single-finding validation is proven live. diff --git a/matilde_plugin/engine/meg_study.py b/matilde_plugin/engine/meg_study.py new file mode 100644 index 0000000..cf1fd50 --- /dev/null +++ b/matilde_plugin/engine/meg_study.py @@ -0,0 +1,338 @@ +"""MEG data-sample validation study — the first heavy / real-data study type. + +Validates a reported neuroscience finding against a **bounded sample** of an open +MEG dataset, as resumable checkpointed steps. The worked claim is the auditory +**M100 / N100m** evoked response: a magnetic field peak ~80-120 ms after an +auditory stimulus. + +Five resumable steps, each reading the **artifact the previous step checkpointed** +and writing its own — so a memory kill or container rebuild resumes from the last +completed step by reloading that on-disk intermediate, never re-running the whole +chain: + + 1. ``fetch_sample`` — locate the dataset and load a **memory-bounded** slice, + persist a small raw-sample artifact. + 2. ``preprocess`` — load that, band-pass filter, persist a filtered artifact. + 3. ``epoch`` — load that, epoch around stimulus events (read from the + data), persist an epochs artifact. + 4. ``evoked`` — load that, average to the evoked response, persist it. + 5. ``validate_finding`` — load the evoked artifact, measure the peak latency, + emit a finding. + +Two hard design constraints (mirroring the rest of ``matilde_plugin/engine``): + +* **Lazy heavy deps.** ``mne`` / ``numpy`` are imported ONLY inside the real I/O + backend's methods (``_MneIO``), never at module import. This module — and the + whole plugin — imports fine with neither installed. The agent container has mne + available at runtime. + +* **Injected I/O.** Every step calls through an injected :class:`MegIO` handle + (``io=...``) that does exactly one load-from-prior-artifact + one transform + + one persist. The whole study is unit-testable with FAKES — no mne, no numpy, no + network, no data files, deterministic. Production passes no ``io`` and gets the + real, memory-bounded :class:`_MneIO` backend. + +Because each step's only input is the prior step's *artifact path* (threaded +through the pipeline's ``ctx.results``, which is plain JSON), a resumed step +reloads from disk — there is no live object held across the run and no re-fetch +of earlier stages. This is exactly the durability the bounded-sample design needs. + +**Memory-bounded by design.** The real backend never full-preloads a recording: +it reads with ``preload=False``, picks a single run, restricts to a subset of +gradiometer channels, crops to a stimulus-locked window, and downsamples — then +persists small intermediates as artifacts. This is what keeps a multi-GB MEG +recording from exhausting the container's memory budget; combined with per-step +checkpointing, an OOM resumes from the last completed step instead of restarting. +""" +from __future__ import annotations + +import abc +from typing import Any, Dict, List, Optional, Tuple + +from .pipeline import Step, StepContext, StepResult + +# Default M100 window (ms) and the bounded-sample defaults. These keep the real +# backend's footprint small; tests override them via build_steps args. +DEFAULT_WINDOW_MS: Tuple[float, float] = (80.0, 120.0) +DEFAULT_BOUNDS: Dict[str, Any] = { + "run": 1, # one run only + "max_channels": 60, # subset of gradiometers, not the full array + "crop_tmax": 30.0, # seconds — a stimulus-locked slice, never the whole recording + "resample_hz": 200, # downsample to shrink the in-memory array + "l_freq": 1.0, # band-pass low edge (Hz) + "h_freq": 40.0, # band-pass high edge (Hz) + "tmin": -0.1, # epoch window start (s) relative to event + "tmax": 0.3, # epoch window end (s) relative to event +} + +# How far outside the expected window a measured peak must fall before the finding +# is called ``refuted`` (rather than merely off-window). Inside the window -> +# supported; clearly outside (beyond this margin) -> refuted; just off it -> +# inconclusive (not clear-cut either way). +_REFUTE_MARGIN_MS = 50.0 + + +class MegIO(abc.ABC): + """Structural contract for the MEG I/O backend the steps call through. + + Any object with these five methods satisfies it (duck-typed via + ``__subclasshook__``, like ``collections.abc``) — the real :class:`_MneIO` or + a test fake. Each method takes the prior step's artifact handle (or dataset + id) and returns a new handle; the heavy work (and heavy imports) live behind + this boundary. + """ + + _REQUIRED = ("fetch_sample", "preprocess", "epoch", "evoked", "measure_peak") + + @classmethod + def __subclasshook__(cls, other: type): + if cls is MegIO: + return all(any(m in B.__dict__ for B in other.__mro__) + for m in cls._REQUIRED) + return NotImplemented + + +# A module-level test hook: when set, the tools-layer meg_validation dispatch uses +# it instead of the real lazy-mne backend. Lets the offline tool test stay +# stdlib-only. Production never sets this. +_TEST_IO: Optional[Any] = None + + +# --------------------------------------------------------------------------- +# Real, memory-bounded mne backend. ALL heavy imports are inside the methods. +# Each method persists its output to disk and returns a small JSON-able handle +# (path + scalar metadata) so the pipeline can checkpoint it and a later/ resumed +# step can reload from that path alone. +# --------------------------------------------------------------------------- + +class _MneIO: + """The real I/O backend. Memory-bounded: never full-preloads a recording. + + Imports ``mne`` / ``numpy`` lazily inside each method, so importing this + module costs nothing and works without the scientific stack installed. + """ + + def __init__(self, work_dir: Optional[str] = None): + import tempfile + self.work_dir = work_dir or tempfile.mkdtemp(prefix="meg_study_") + + def _path(self, name: str) -> str: + import os + return os.path.join(self.work_dir, name) + + def fetch_sample(self, *, dataset_id: str, bounds: dict) -> dict: + import mne # lazy + + if dataset_id != "bst_auditory": + raise ValueError( + f"meg_validation currently supports dataset 'bst_auditory', " + f"not {dataset_id!r}.") + data_path = mne.datasets.brainstorm.bst_auditory.data_path() + run = bounds.get("run", DEFAULT_BOUNDS["run"]) + raw_fname = (f"{data_path}/MEG/bst_auditory/" + f"S01_AEF_20131218_0{run}.ds") + # preload=False: header/metadata only — the samples are NOT pulled into + # memory here. We crop + pick + resample BEFORE materializing. + raw = mne.io.read_raw_ctf(raw_fname, preload=False, verbose="ERROR") + crop_tmax = float(bounds.get("crop_tmax", DEFAULT_BOUNDS["crop_tmax"])) + crop_tmax = min(crop_tmax, raw.times[-1]) + raw.crop(tmax=crop_tmax) + # Keep a subset of gradiometer channels (+ stim) before loading samples. + picks = mne.pick_types(raw.info, meg="grad", eeg=False, stim=True, + exclude="bads") + max_ch = int(bounds.get("max_channels", DEFAULT_BOUNDS["max_channels"])) + keep = [raw.info["ch_names"][p] for p in picks[:max_ch]] + raw.pick(keep) + raw.load_data(verbose="ERROR") # now small: cropped + channel-subset + resample_hz = bounds.get("resample_hz", DEFAULT_BOUNDS["resample_hz"]) + if resample_hz: + raw.resample(float(resample_hz), verbose="ERROR") + path = self._path("raw_sample.fif") + raw.save(path, overwrite=True, verbose="ERROR") + return {"path": path, "dataset_id": dataset_id, "bounds": dict(bounds)} + + def preprocess(self, raw_handle: dict, *, l_freq: float, h_freq: float) -> dict: + import mne # lazy + + raw = mne.io.read_raw_fif(raw_handle["path"], preload=True, + verbose="ERROR") + raw.filter(l_freq=l_freq, h_freq=h_freq, verbose="ERROR") + path = self._path("filtered.fif") + raw.save(path, overwrite=True, verbose="ERROR") + return {"path": path, "filtered": [l_freq, h_freq]} + + def epoch(self, filtered_handle: dict, *, tmin: float, tmax: float) -> dict: + import mne # lazy + + raw = mne.io.read_raw_fif(filtered_handle["path"], preload=True, + verbose="ERROR") + # Read events FROM the data rather than assuming counts. + events = mne.find_events(raw, verbose="ERROR") + epochs = mne.Epochs(raw, events, tmin=tmin, tmax=tmax, + baseline=(None, 0), preload=True, verbose="ERROR") + path = self._path("epochs-epo.fif") + epochs.save(path, overwrite=True, verbose="ERROR") + return {"path": path, "n_epochs": len(epochs), "window": [tmin, tmax]} + + def evoked(self, epochs_handle: dict) -> dict: + import mne # lazy + + epochs = mne.read_epochs(epochs_handle["path"], preload=True, + verbose="ERROR") + ev = epochs.average() + path = self._path("evoked-ave.fif") + ev.save(path, overwrite=True, verbose="ERROR") + return {"path": path, "n_epochs": epochs_handle.get("n_epochs", + len(epochs))} + + def measure_peak(self, evoked_handle: dict, *, + window_ms: Tuple[float, float]) -> dict: + import mne # lazy + + n_epochs = int(evoked_handle.get("n_epochs", 0)) + ev = mne.read_evokeds(evoked_handle["path"], verbose="ERROR")[0] + if n_epochs <= 0 or len(ev.times) == 0: + return {"latency_ms": None, "amplitude": None, "n_epochs": n_epochs} + tmin_s, tmax_s = window_ms[0] / 1000.0, window_ms[1] / 1000.0 + # Search a slightly widened window so an out-of-window peak is still + # measurable (and can be reported as refuted). + lo = max(ev.times[0], tmin_s - 0.1) + hi = min(ev.times[-1], tmax_s + 0.1) + ch, latency_s, amp = ev.get_peak(tmin=lo, tmax=hi, mode="abs", + return_amplitude=True) + return {"latency_ms": float(latency_s) * 1000.0, + "amplitude": float(amp), "n_epochs": n_epochs, "channel": ch} + + +# --------------------------------------------------------------------------- +# Steps. Each reads the prior step's artifact handle from ctx.results, does ONE +# io transform, and persists its own. Built closing over the injected ``io``. +# --------------------------------------------------------------------------- + +def _classify(latency_ms: Optional[float], + window_ms: Tuple[float, float]) -> str: + """Verdict for a measured peak latency against the expected window.""" + lo, hi = window_ms + if latency_ms is None: + return "inconclusive" # degenerate sample / no measurable peak + if lo <= latency_ms <= hi: + return "supported" # within the expected window + if latency_ms < lo - _REFUTE_MARGIN_MS or latency_ms > hi + _REFUTE_MARGIN_MS: + return "refuted" # clearly outside + return "inconclusive" # just off the window — not clear-cut + + +def _fetch_sample_step(io: Any, dataset_id: str, bounds: dict) -> Step: + def fn(ctx: StepContext) -> StepResult: + handle = io.fetch_sample(dataset_id=dataset_id, bounds=bounds) + return StepResult( + data={"dataset_id": dataset_id, "bounds": dict(bounds), + "path": handle.get("path")}, + artifacts=[{"path": handle.get("path", ""), "kind": "raw_sample", + "meta": {"dataset_id": dataset_id, "bounds": bounds}}], + ) + return Step(name="fetch_sample", fn=fn) + + +def _preprocess_step(io: Any, bounds: dict) -> Step: + def fn(ctx: StepContext) -> StepResult: + prior = ctx.results["fetch_sample"] + out = io.preprocess( + {"path": prior.get("path")}, + l_freq=bounds.get("l_freq", DEFAULT_BOUNDS["l_freq"]), + h_freq=bounds.get("h_freq", DEFAULT_BOUNDS["h_freq"])) + return StepResult( + data={"filtered": out.get("filtered"), "path": out.get("path")}, + artifacts=[{"path": out.get("path", ""), "kind": "filtered_sample", + "meta": {"filtered": out.get("filtered")}}], + ) + return Step(name="preprocess", fn=fn) + + +def _epoch_step(io: Any, bounds: dict) -> Step: + def fn(ctx: StepContext) -> StepResult: + prior = ctx.results["preprocess"] + epochs = io.epoch( + {"path": prior.get("path")}, + tmin=bounds.get("tmin", DEFAULT_BOUNDS["tmin"]), + tmax=bounds.get("tmax", DEFAULT_BOUNDS["tmax"])) + return StepResult( + data={"n_epochs": epochs.get("n_epochs"), + "window": epochs.get("window"), "path": epochs.get("path")}, + artifacts=[{"path": epochs.get("path", ""), "kind": "epochs", + "meta": {"n_epochs": epochs.get("n_epochs")}}], + ) + return Step(name="epoch", fn=fn) + + +def _evoked_step(io: Any) -> Step: + def fn(ctx: StepContext) -> StepResult: + prior = ctx.results["epoch"] + ev = io.evoked({"path": prior.get("path"), + "n_epochs": prior.get("n_epochs")}) + # may raise (e.g. OOM) -> step fails, study resumable + return StepResult( + data={"n_epochs": ev.get("n_epochs"), "path": ev.get("path")}, + artifacts=[{"path": ev.get("path", ""), "kind": "evoked", + "meta": {"n_epochs": ev.get("n_epochs")}}], + ) + return Step(name="evoked", fn=fn) + + +def _validate_finding_step(io: Any, dataset_id: str, + window_ms: Tuple[float, float]) -> Step: + def fn(ctx: StepContext) -> StepResult: + prior = ctx.results["evoked"] + peak = io.measure_peak( + {"path": prior.get("path"), "n_epochs": prior.get("n_epochs")}, + window_ms=window_ms) + latency = peak.get("latency_ms") + amplitude = peak.get("amplitude") + verdict = _classify(latency, window_ms) + claim = (f"auditory M100 peak for {dataset_id} falls within " + f"{window_ms[0]:.0f}-{window_ms[1]:.0f} ms") + finding = { + "claim": claim, + "verdict": verdict, + "score": None, + "evidence": { + "latency_ms": latency, + "amplitude": amplitude, + "expected_window_ms": list(window_ms), + "n_epochs": peak.get("n_epochs"), + }, + } + return StepResult(data={"verdict": verdict, "latency_ms": latency, + "amplitude": amplitude}, + findings=[finding]) + return Step(name="validate_finding", fn=fn) + + +# --------------------------------------------------------------------------- +# Public builder — mirrors bibliography_study.build_steps. +# --------------------------------------------------------------------------- + +def build_steps(*, dataset_id: str = "bst_auditory", + io: Optional[Any] = None, + expected_window_ms: Tuple[float, float] = DEFAULT_WINDOW_MS, + bounds: Optional[dict] = None) -> List[Step]: + """Build the five meg_validation steps for *dataset_id*. + + Pass *io* to inject the MEG I/O backend (tests pass a fake). With ``io=None`` + the real, memory-bounded :class:`_MneIO` backend is used (lazy mne). *bounds* + overrides the memory-bounding defaults (run/channels/crop/resample/filter/ + epoch window); *expected_window_ms* is the M100 window the finding is judged + against. The returned steps feed ``pipeline.run`` / ``resume``. + """ + io = io if io is not None else _MneIO() + merged_bounds = dict(DEFAULT_BOUNDS) + if bounds: + merged_bounds.update(bounds) + window = (float(expected_window_ms[0]), float(expected_window_ms[1])) + return [ + _fetch_sample_step(io, dataset_id, merged_bounds), + _preprocess_step(io, merged_bounds), + _epoch_step(io, merged_bounds), + _evoked_step(io), + _validate_finding_step(io, dataset_id, window), + ] diff --git a/matilde_plugin/tools.py b/matilde_plugin/tools.py index 30c7a44..7778412 100644 --- a/matilde_plugin/tools.py +++ b/matilde_plugin/tools.py @@ -479,8 +479,11 @@ def _coerce_plan(plan: Any) -> list: "title": {"type": "string", "description": "Human-readable title."}, "plan": {"type": "array", "items": {"type": "string"}, "description": "Ordered step names (e.g. ['parse_refs','verify_each','summarize'])."}, - "kind": {"type": "string", "description": "Optional study kind. 'bibliography' wires the built-in reference-validation steps."}, + "kind": {"type": "string", "description": "Optional study kind. 'bibliography' wires the built-in reference-validation steps; 'meg_validation' wires the memory-bounded MEG evoked-peak (M100) validation steps."}, "bibtex": {"type": "string", "description": "For kind='bibliography': the BibTeX to validate (stored on the study)."}, + "dataset_id": {"type": "string", "description": "For kind='meg_validation': the open MEG dataset to validate against (e.g. 'bst_auditory')."}, + "expected_window_ms": {"type": "array", "items": {"type": "number"}, "description": "For kind='meg_validation': the [low, high] expected peak-latency window in ms (default [80, 120] for the auditory M100)."}, + "bounds": {"type": "object", "description": "For kind='meg_validation': sample-bounding overrides (run, max_channels, crop_tmax, resample_hz, l_freq, h_freq, tmin, tmax) keeping the loaded slice small."}, }, "required": ["slug"], }, @@ -498,11 +501,27 @@ def _handle_study_create(args: dict, **kwargs: Any) -> str: bibtex = args.get("bibtex") if kind == "bibliography" and not plan: plan = ["parse_refs", "verify_each", "summarize"] + if kind == "meg_validation" and not plan: + plan = ["fetch_sample", "preprocess", "epoch", "evoked", + "validate_finding"] meta: dict = {} if kind: meta["kind"] = kind if isinstance(bibtex, str) and bibtex.strip(): meta["bibtex"] = bibtex + if kind == "meg_validation": + dsid = str(args.get("dataset_id", "")).strip() or "bst_auditory" + meta["dataset_id"] = dsid + window = args.get("expected_window_ms") + if isinstance(window, (list, tuple)) and len(window) == 2: + try: + meta["expected_window_ms"] = [float(window[0]), + float(window[1])] + except (TypeError, ValueError): + pass + bounds = args.get("bounds") + if isinstance(bounds, dict): + meta["bounds"] = bounds store = _open_store() sid = store.create_study(slug=slug, title=title, plan=plan, meta=meta) store.add_steps(sid, plan) @@ -551,11 +570,21 @@ def _handle_study_run(args: dict, **kwargs: Any) -> str: if kind == "bibliography": from .engine.bibliography_study import build_steps steps = build_steps(bibtex=meta.get("bibtex", "")) + elif kind == "meg_validation": + from .engine import meg_study + window = meta.get("expected_window_ms") or list( + meg_study.DEFAULT_WINDOW_MS) + # _TEST_IO is a stdlib test hook (None in production -> real lazy-mne). + steps = meg_study.build_steps( + dataset_id=meta.get("dataset_id", "bst_auditory"), + io=getattr(meg_study, "_TEST_IO", None), + expected_window_ms=(window[0], window[1]), + bounds=meta.get("bounds")) else: return _tool_error( f"Study {sid} has no runnable step implementations " f"(kind={kind!r}). Built-in runner currently supports " - f"kind='bibliography'.", study_id=sid) + f"kind='bibliography' and kind='meg_validation'.", study_id=sid) summary = run(store, sid, steps) return _tool_result( study_id=sid, status=summary["status"], steps=summary["steps"], diff --git a/tests/test_meg_study.py b/tests/test_meg_study.py new file mode 100644 index 0000000..bd7fc58 --- /dev/null +++ b/tests/test_meg_study.py @@ -0,0 +1,243 @@ +"""Tests for the MEG data-sample validation study (the M100 worked claim). + +The whole study is exercised with FAKES — no mne, no numpy, no network, no data +files. Each step takes injected loader/analysis callables, so the unit tests are +deterministic and stdlib-only. Covered here: + + - each step transforms the study state correctly (fetch -> preprocess -> epoch + -> evoked -> validate_finding), reading prior-step data and persisting an + artifact per heavy step; + - the ``validate_finding`` verdict boundaries — peak at 100 ms -> supported, + peak at 300 ms -> refuted, empty/degenerate sample -> inconclusive; + - resumability — force a failure at the ``evoked`` step, assert the study is + blocked with the earlier steps done, then resume with a working analyzer and + assert fetch/preprocess/epoch are NOT re-run (call counters). + +An opt-in live test (``MATILDE_LIVE=1`` and mne importable) runs the real bounded +pipeline on ``bst_auditory`` — it SKIPS without the flag / without mne. +""" +from __future__ import annotations + +import os +import sys + +import pytest + +sys.path.insert(0, os.path.normpath(os.path.join(os.path.dirname(__file__), ".."))) + +from matilde_plugin.engine.meg_study import ( # noqa: E402 + MegIO, + build_steps, +) +from matilde_plugin.engine.pipeline import resume, run # noqa: E402 +from matilde_plugin.engine.store import StudyStore # noqa: E402 + + +# --------------------------------------------------------------------------- +# A fake MEG I/O backend: pure dicts, deterministic, counts each call so the +# resumability test can prove done steps are not re-run. No mne / numpy / fs. +# --------------------------------------------------------------------------- + +class FakeMegIO: + """Stand-in for the real (lazy mne) I/O. Each method bumps a counter and + returns a small dict 'handle' threaded through the steps.""" + + def __init__(self, *, peak_latency_ms=100.0, peak_amp=5.0, n_epochs=40, + fail_evoked=False): + self.peak_latency_ms = peak_latency_ms + self.peak_amp = peak_amp + self.n_epochs = n_epochs + self.fail_evoked = fail_evoked + self.calls = {"fetch": 0, "preprocess": 0, "epoch": 0, "evoked": 0, + "measure_peak": 0} + + def fetch_sample(self, *, dataset_id, bounds): + self.calls["fetch"] += 1 + return {"dataset_id": dataset_id, "bounds": dict(bounds), + "path": "/tmp/fake_raw.fif"} + + def preprocess(self, raw, *, l_freq, h_freq): + self.calls["preprocess"] += 1 + return {**raw, "filtered": [l_freq, h_freq], "path": "/tmp/fake_filt.fif"} + + def epoch(self, filtered, *, tmin, tmax): + self.calls["epoch"] += 1 + return {**filtered, "n_epochs": self.n_epochs, + "window": [tmin, tmax], "path": "/tmp/fake_epo.fif"} + + def evoked(self, epochs): + self.calls["evoked"] += 1 + if self.fail_evoked: + raise RuntimeError("OOM while averaging evoked response") + return {**epochs, "averaged": True, "path": "/tmp/fake_ave.fif"} + + def measure_peak(self, evoked, *, window_ms): + self.calls["measure_peak"] += 1 + if self.n_epochs <= 0: # degenerate sample -> no measurable peak + return {"latency_ms": None, "amplitude": None, "n_epochs": 0} + return {"latency_ms": self.peak_latency_ms, "amplitude": self.peak_amp, + "n_epochs": self.n_epochs} + + +@pytest.fixture() +def store(tmp_path): + return StudyStore(str(tmp_path / "studies.db")) + + +def _plan(): + return ["fetch_sample", "preprocess", "epoch", "evoked", "validate_finding"] + + +# --------------------------------------------------------------------------- +# MegIO is a Protocol-ish contract; the fake should satisfy it (sanity). +# --------------------------------------------------------------------------- + +def test_fakeio_satisfies_contract(): + io = FakeMegIO() + assert isinstance(io, MegIO) + + +# --------------------------------------------------------------------------- +# Happy path — peak at 100 ms is within the default 80-120 window -> supported. +# --------------------------------------------------------------------------- + +def test_happy_path_peak_within_window_is_supported(store): + sid = store.create_study(slug="m100", title="M100", plan=_plan()) + io = FakeMegIO(peak_latency_ms=100.0) + steps = build_steps(dataset_id="bst_auditory", io=io) + summary = run(store, sid, steps) + + assert summary["status"] == "done" + assert store.get_study(sid)["status"] == "done" + + findings = store.get_findings(sid) + assert len(findings) == 1 + f = findings[0] + assert f["verdict"] == "supported" + assert f["evidence"]["latency_ms"] == 100.0 + assert f["evidence"]["amplitude"] == 5.0 + + # each heavy step checkpointed an artifact (resume-after-OOM target) + arts = {a["step_name"] for a in store.get_artifacts(sid)} + assert {"fetch_sample", "preprocess", "epoch", "evoked"} <= arts + + +# --------------------------------------------------------------------------- +# Verdict boundaries. +# --------------------------------------------------------------------------- + +def test_peak_far_outside_window_is_refuted(store): + sid = store.create_study(slug="r", title="R", plan=_plan()) + io = FakeMegIO(peak_latency_ms=300.0) + run(store, sid, build_steps(dataset_id="bst_auditory", io=io)) + f = store.get_findings(sid)[0] + assert f["verdict"] == "refuted" + assert f["evidence"]["latency_ms"] == 300.0 + + +def test_degenerate_sample_is_inconclusive(store): + sid = store.create_study(slug="i", title="I", plan=_plan()) + io = FakeMegIO(n_epochs=0) # no epochs -> no measurable peak + run(store, sid, build_steps(dataset_id="bst_auditory", io=io)) + f = store.get_findings(sid)[0] + assert f["verdict"] == "inconclusive" + assert f["evidence"]["latency_ms"] is None + + +def test_custom_window_changes_verdict(store): + """A peak at 150 ms is refuted under default (80-120) but supported under a + wider custom window — proves the expected-window param is honored.""" + sid = store.create_study(slug="w", title="W", plan=_plan()) + io = FakeMegIO(peak_latency_ms=150.0) + run(store, sid, build_steps(dataset_id="bst_auditory", io=io, + expected_window_ms=(130.0, 170.0))) + assert store.get_findings(sid)[0]["verdict"] == "supported" + + +# --------------------------------------------------------------------------- +# Step state transitions — each step's data threads into the next. +# --------------------------------------------------------------------------- + +def test_steps_thread_state_forward(store): + sid = store.create_study(slug="t", title="T", plan=_plan()) + io = FakeMegIO() + run(store, sid, build_steps(dataset_id="bst_auditory", io=io, + bounds={"crop_tmax": 9.0})) + + # bounds flowed into the fetch handle and the artifact meta. + fetch_res = store.get_step(sid, "fetch_sample")["result"] + assert fetch_res["dataset_id"] == "bst_auditory" + assert fetch_res["bounds"]["crop_tmax"] == 9.0 + + epoch_res = store.get_step(sid, "epoch")["result"] + assert epoch_res["n_epochs"] == 40 + + evoked_res = store.get_step(sid, "evoked")["result"] + assert evoked_res["n_epochs"] == 40 + + +# --------------------------------------------------------------------------- +# Resumability — fail at evoked, resume, earlier steps not re-run. +# --------------------------------------------------------------------------- + +def test_fail_at_evoked_then_resume_does_not_rerun_earlier_steps(store): + sid = store.create_study(slug="resume", title="Resume", plan=_plan()) + + # First pass: evoked raises (simulated OOM mid-average). + failing_io = FakeMegIO(fail_evoked=True) + run(store, sid, build_steps(dataset_id="bst_auditory", io=failing_io)) + + assert store.get_step(sid, "fetch_sample")["status"] == "done" + assert store.get_step(sid, "preprocess")["status"] == "done" + assert store.get_step(sid, "epoch")["status"] == "done" + assert store.get_step(sid, "evoked")["status"] == "failed" + assert store.get_step(sid, "validate_finding")["status"] == "pending" + assert store.get_study(sid)["status"] in ("blocked", "failed") + assert store.get_findings(sid) == [] # no finding emitted yet + + assert failing_io.calls["fetch"] == 1 + assert failing_io.calls["preprocess"] == 1 + assert failing_io.calls["epoch"] == 1 + assert failing_io.calls["evoked"] == 1 # attempted, raised + + # Second pass: working io, resume. + good_io = FakeMegIO(fail_evoked=False, peak_latency_ms=100.0) + resume(store, sid, build_steps(dataset_id="bst_auditory", io=good_io)) + + assert store.get_study(sid)["status"] == "done" + # Earlier (done) steps were NOT re-run on the working io. + assert good_io.calls["fetch"] == 0 + assert good_io.calls["preprocess"] == 0 + assert good_io.calls["epoch"] == 0 + # Only the previously-failed step + the remaining step ran. + assert good_io.calls["evoked"] == 1 + assert good_io.calls["measure_peak"] == 1 + + f = store.get_findings(sid)[0] + assert f["verdict"] == "supported" + + +# --------------------------------------------------------------------------- +# Opt-in LIVE test — real bounded mne pipeline. SKIPS without flag / mne. +# --------------------------------------------------------------------------- + +@pytest.mark.skipif(os.environ.get("MATILDE_LIVE") != "1", + reason="live MEG test is opt-in (set MATILDE_LIVE=1)") +def test_live_bst_auditory_m100_bounded(): + pytest.importorskip("mne") + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + store = StudyStore(os.path.join(tmp, "studies.db")) + sid = store.create_study(slug="live-m100", title="Live M100", + plan=_plan()) + # Default io=None -> the real, memory-bounded mne backend. + steps = build_steps(dataset_id="bst_auditory") + summary = run(store, sid, steps) + + assert summary["status"] == "done", summary + f = store.get_findings(sid)[0] + # A plausible auditory peak should be found and measured. + assert f["evidence"]["latency_ms"] is not None + assert 50.0 <= f["evidence"]["latency_ms"] <= 200.0 + assert f["verdict"] in ("supported", "refuted") diff --git a/tests/test_meg_study_tools.py b/tests/test_meg_study_tools.py new file mode 100644 index 0000000..46dc1bd --- /dev/null +++ b/tests/test_meg_study_tools.py @@ -0,0 +1,83 @@ +"""Offline tests for wiring the meg_validation study type into the agent tools. + +Mirrors test_study_tools.py: loads the plugin, creates a kind='meg_validation' +study via matilde_study_create with params (dataset id, expected window, sample +bounds), and asserts the study runs through the meg_validation dispatch in +matilde_study_run. The real run path is lazy-mne; this test injects a FAKE io via +the module-level test hook so it stays stdlib-only (no mne / network / data). +""" +from __future__ import annotations + +import importlib.util +import json +import os +import sys + +ROOT = os.path.normpath(os.path.join(os.path.dirname(__file__), "..")) +sys.path.insert(0, ROOT) + +# Import as a real package first so matilde_plugin is registered in sys.modules +# (mirrors test_meg_study.py); _load_plugin then re-execs the same package name. +import matilde_plugin # noqa: E402,F401 + + +def _load_plugin(): + path = os.path.join(ROOT, "matilde_plugin", "__init__.py") + spec = importlib.util.spec_from_file_location("matilde_plugin", path) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +class _FakeIO: + """Minimal fake satisfying the MegIO contract for the tools dispatch test.""" + + def fetch_sample(self, *, dataset_id, bounds): + return {"path": "/tmp/fake_raw.fif", "dataset_id": dataset_id, + "bounds": dict(bounds)} + + def preprocess(self, raw, *, l_freq, h_freq): + return {"path": "/tmp/fake_filt.fif", "filtered": [l_freq, h_freq]} + + def epoch(self, filtered, *, tmin, tmax): + return {"path": "/tmp/fake_epo.fif", "n_epochs": 30, + "window": [tmin, tmax]} + + def evoked(self, epochs): + return {"path": "/tmp/fake_ave.fif", "n_epochs": 30} + + def measure_peak(self, evoked, *, window_ms): + return {"latency_ms": 100.0, "amplitude": 4.0, "n_epochs": 30} + + +def test_meg_study_create_and_run(monkeypatch, tmp_path): + plugin = _load_plugin() + monkeypatch.setenv("MATILDE_STUDY_DB", str(tmp_path / "s.db")) + + # Inject the fake io for the meg_validation dispatch (the test hook keeps the + # tools layer stdlib-only — production uses the real lazy-mne backend). + from matilde_plugin.engine import meg_study + monkeypatch.setattr(meg_study, "_TEST_IO", _FakeIO(), raising=False) + + created = json.loads(plugin._handle_study_create({ + "slug": "m100-study", + "title": "M100 validation", + "kind": "meg_validation", + "dataset_id": "bst_auditory", + "expected_window_ms": [80, 120], + "bounds": {"crop_tmax": 9.0}, + })) + assert created["success"] is True + sid = created["study_id"] + # A default plan was filled in for meg_validation. + assert created["plan"] == ["fetch_sample", "preprocess", "epoch", + "evoked", "validate_finding"] + + out = json.loads(plugin._handle_study_run({"study_id": str(sid)})) + assert out["success"] is True, out + assert out["status"] == "done" + + status = json.loads(plugin._handle_study_status({"study_id": sid})) + assert status["finding_count"] == 1 + assert status["findings"][0]["verdict"] == "supported" + assert status["findings"][0]["evidence"]["latency_ms"] == 100.0