def validate_manifest(
manifest_csv: Path,
*,
default_label_column: str | None = None,
default_group_column: str | None = None,
default_n_splits: int = 5,
) -> list[ManifestValidation]:
"""Validate staged files and metadata referenced by a benchmark manifest."""
if default_n_splits < 2:
raise ValueError("default_n_splits must be at least 2")
manifest = pd.read_csv(manifest_csv)
required = {"subject", "epochs"}
missing_columns = sorted(required.difference(manifest.columns))
if missing_columns:
raise ValueError(f"Manifest is missing required columns: {missing_columns}")
base_dir = manifest_csv.parent
validations: list[ManifestValidation] = []
for _, row in manifest.iterrows():
subject = _value(row, "subject", "<missing>") or "<missing>"
messages: list[str] = []
epochs_path = _resolve(_value(row, "epochs"), base_dir)
label_column = _value(row, "label_column", default_label_column)
group_column = _value(row, "group_column", default_group_column)
try:
n_splits = _int_value(row, "n_splits", default_n_splits)
except ValueError as exc:
messages.append(str(exc))
n_splits = default_n_splits
if label_column is None:
messages.append("label_column is missing")
if epochs_path is None:
messages.append("epochs path is missing")
elif not epochs_path.exists():
messages.append(f"epochs file does not exist: {epochs_path}")
metadata, metadata_messages = _load_metadata_for_row(row, base_dir)
messages.extend(metadata_messages)
epochs_metadata: pd.DataFrame | None = None
if epochs_path is not None and epochs_path.exists():
try:
epochs = mne.read_epochs(epochs_path, preload=False, verbose="error")
epochs_metadata = epochs.metadata.copy() if epochs.metadata is not None else None
n_epochs = len(epochs)
except Exception as exc: # pragma: no cover - MNE raises several concrete IO errors.
messages.append(f"could not read epochs file: {exc}")
n_epochs = None
else:
if metadata is not None and len(metadata) != n_epochs:
messages.append(f"metadata rows ({len(metadata)}) do not match epochs ({n_epochs})")
else:
n_epochs = None
effective_metadata = metadata if metadata is not None else epochs_metadata
if effective_metadata is None:
if not any("metadata_csv" in message or "events_csv" in message for message in messages):
messages.append("no metadata source available; provide metadata_csv, events_csv, or epochs metadata")
elif label_column is not None:
messages.extend(_validate_class_balance(effective_metadata, label_column, n_splits))
if group_column is not None:
if group_column not in effective_metadata.columns:
messages.append(f"group column '{group_column}' is missing")
elif effective_metadata[group_column].dropna().nunique() < n_splits:
messages.append(f"group column '{group_column}' has fewer than n_splits={n_splits} unique groups")
validations.append(ManifestValidation(subject=subject, ok=not messages, messages=messages))
return validations