diff --git a/docs/source/user_guide/agent_cli.rst b/docs/source/user_guide/agent_cli.rst index 680ba43f..590ae927 100644 --- a/docs/source/user_guide/agent_cli.rst +++ b/docs/source/user_guide/agent_cli.rst @@ -141,6 +141,26 @@ high-pass cleaning criteria off unless the YAML config sets them explicitly. Set ``highpass`` as a two-value transition band, for example ``highpass: [0.25, 0.75]``. +Manifest Portability And Migration +================================== + +The ``eegprep.manifest.v2`` schema stores the ``path`` fields in +``input_files`` and ``output_files`` relative to the manifest file. Stored +relative paths always use forward slashes. Runtime manifests created by +``build_manifest()`` use absolute paths, and ``read_manifest()`` restores v2 +relative paths to absolute paths for the current machine. + +This preserves artifact relationships when the manifest and files move +together. References using ``..`` only remain valid when the referenced files +move with the same directory layout. A path on another Windows drive cannot be +made relative and remains absolute, so it is not portable to another machine. + +Version 1 and unknown schema versions keep their original path strings when +read or written; EEGPrep cannot infer what their relative paths were relative +to. Consumers that need resolved v2 paths should use ``read_manifest()`` +instead of loading the JSON directly. Path serialization does not alter or +verify the recorded SHA-256 values. + QC And Reports ============== diff --git a/src/eegprep/cli/commands/transforms.py b/src/eegprep/cli/commands/transforms.py index 368a6ae4..c0ba9fb2 100644 --- a/src/eegprep/cli/commands/transforms.py +++ b/src/eegprep/cli/commands/transforms.py @@ -20,6 +20,7 @@ from eegprep.cli.core import ( EEGPrepCLIError, + MANIFEST_SCHEMA_VERSION as _MANIFEST_SCHEMA_VERSION, build_manifest, file_sha256, utc_now, @@ -38,7 +39,7 @@ logger = logging.getLogger(__name__) RESULT_SCHEMA_VERSION = "eegprep.transform_result.v1" -MANIFEST_SCHEMA_VERSION = "eegprep.manifest.v1" +MANIFEST_SCHEMA_VERSION = _MANIFEST_SCHEMA_VERSION DATASET_OUTPUT_TYPE = "eeglab_set" diff --git a/src/eegprep/cli/core.py b/src/eegprep/cli/core.py index e5378438..64912cdf 100644 --- a/src/eegprep/cli/core.py +++ b/src/eegprep/cli/core.py @@ -2,14 +2,16 @@ from __future__ import annotations +import copy import hashlib import json import math +import os import platform import sys from dataclasses import dataclass from datetime import datetime, timezone -from pathlib import Path +from pathlib import Path, PurePosixPath, PureWindowsPath from typing import Any import numpy as np @@ -18,6 +20,7 @@ COMMAND_RESULT_SCHEMA_VERSION = "eegprep.cli.result.v1" +MANIFEST_SCHEMA_VERSION = "eegprep.manifest.v2" class EEGPrepCLIError(Exception): @@ -238,11 +241,12 @@ def build_manifest( warnings: list[Any] | None = None, ) -> dict[str, Any]: stamp = runtime_stamp(started_at) if finished_at is None else RuntimeStamp(started_at, finished_at) + input_records = [_input_file_record(path) for path in input_files] manifest: dict[str, Any] = { - "schema_version": "eegprep.manifest.v1", + "schema_version": MANIFEST_SCHEMA_VERSION, "command": command, - "input_files": [_input_file_record(path) for path in input_files], - "output_files": output_files, + "input_files": _make_manifest_paths_absolute(input_records), + "output_files": _make_manifest_paths_absolute(output_files), "parameters": json_safe(parameters), "history": history, "software": software_info(), @@ -254,12 +258,53 @@ def build_manifest( return manifest +def _make_manifest_relative(manifest: dict[str, Any], manifest_path: Path) -> dict[str, Any]: + """Return the on-disk representation of a v2 manifest.""" + prepared = copy.deepcopy(json_safe(manifest)) + if prepared.get("schema_version") != MANIFEST_SCHEMA_VERSION: + return prepared + + base_dir = manifest_path.resolve().parent + for item in _manifest_file_records(prepared): + value = item.get("path") + if not isinstance(value, str) or not value: + continue + native_path = Path(value).expanduser() + if not native_path.is_absolute(): + continue + try: + item["path"] = Path(os.path.relpath(native_path, start=base_dir)).as_posix() + except ValueError: + # Windows cannot express a path on another drive as a relative path. + continue + return prepared + + +def read_manifest(path: str | Path) -> dict[str, Any]: + """Read a manifest, resolving v2 artifact paths against its directory.""" + resolved = Path(path).expanduser().resolve() + with resolved.open("r", encoding="utf-8") as stream: + manifest = json.load(stream) + + if manifest.get("schema_version") != MANIFEST_SCHEMA_VERSION: + return manifest + + for item in _manifest_file_records(manifest): + value = item.get("path") + if not isinstance(value, str) or not value or _is_absolute_on_any_platform(value): + continue + relative_path = Path(*PurePosixPath(value).parts) + item["path"] = str((resolved.parent / relative_path).resolve()) + return manifest + + def write_manifest(path: str | Path | None, manifest: dict[str, Any]) -> Path | None: if path is None: return None resolved = Path(path).expanduser() resolved.parent.mkdir(parents=True, exist_ok=True) - resolved.write_text(json.dumps(json_safe(manifest), indent=2, sort_keys=True) + "\n", encoding="utf-8") + rel_manifest = _make_manifest_relative(manifest, resolved) + resolved.write_text(json.dumps(json_safe(rel_manifest), indent=2, sort_keys=True) + "\n", encoding="utf-8") return resolved @@ -284,7 +329,39 @@ def write_json_file( def write_manifest_file(path: str | Path, manifest: dict[str, Any], *, overwrite: bool = False) -> dict[str, Any]: - return write_json_file(path, manifest, overwrite=overwrite, output_type="manifest") + target = Path(path) + rel_manifest = _make_manifest_relative(manifest, target) + return write_json_file(target, rel_manifest, overwrite=overwrite, output_type="manifest") + + +def _make_manifest_paths_absolute(records: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Normalize runtime artifact paths before they acquire manifest-relative semantics.""" + prepared = copy.deepcopy(json_safe(records)) + for item in prepared: + value = item.get("path") + if not isinstance(value, str) or not value or _is_foreign_absolute_path(value): + continue + item["path"] = str(Path(value).expanduser().resolve()) + return prepared + + +def _manifest_file_records(manifest: dict[str, Any]) -> list[dict[str, Any]]: + records: list[dict[str, Any]] = [] + for key in ("input_files", "output_files"): + value = manifest.get(key) + if isinstance(value, list): + records.extend(item for item in value if isinstance(item, dict)) + return records + + +def _is_absolute_on_any_platform(value: str) -> bool: + return Path(value).is_absolute() or PurePosixPath(value).is_absolute() or PureWindowsPath(value).is_absolute() + + +def _is_foreign_absolute_path(value: str) -> bool: + return not Path(value).is_absolute() and ( + PurePosixPath(value).is_absolute() or PureWindowsPath(value).is_absolute() + ) def _input_file_record(path: Path | dict[str, Any]) -> dict[str, Any]: diff --git a/tests/test_cli_pipeline_qc_report.py b/tests/test_cli_pipeline_qc_report.py index 2c2f3d14..a4b7f9fb 100644 --- a/tests/test_cli_pipeline_qc_report.py +++ b/tests/test_cli_pipeline_qc_report.py @@ -4,6 +4,7 @@ import numpy as np import yaml +from eegprep.cli.core import read_manifest from eegprep.cli.commands import transforms as transforms_cli from eegprep.cli.commands.pipeline import ( _channel_indices, @@ -56,10 +57,13 @@ def test_pipeline_run_writes_qc_report_and_manifest(tmp_path): assert (tmp_path / "out" / "report.html").is_file() manifest_path = tmp_path / "out" / "eegprep_manifest.json" assert manifest_path.is_file() - manifest = json.loads(manifest_path.read_text(encoding="utf-8")) - assert manifest["schema_version"] == "eegprep.manifest.v1" + stored_manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + assert {item["path"] for item in stored_manifest["output_files"]} == {"qc.json", "report.html"} + manifest = read_manifest(manifest_path) + assert manifest["schema_version"] == "eegprep.manifest.v2" assert manifest["command"] == "pipeline run" assert {item["type"] for item in manifest["output_files"]} == {"json", "html_report"} + assert all(Path(item["path"]).is_absolute() for item in manifest["input_files"] + manifest["output_files"]) assert isinstance(result["qc"]["recommendations"], list) @@ -73,7 +77,7 @@ def test_pipeline_run_resample_writes_dataset_manifest_and_history(tmp_path): assert result["status"] == "ok" assert (tmp_path / "out" / "eeglab_data_eegprep.set").is_file() - manifest = json.loads((tmp_path / "out" / "eegprep_manifest.json").read_text(encoding="utf-8")) + manifest = read_manifest(tmp_path / "out" / "eegprep_manifest.json") assert "pop_resample" in manifest["history"] assert any(item["type"] == "eeglab_set" for item in manifest["output_files"]) assert any("pop_resample" in item for item in result["history"]) @@ -301,8 +305,8 @@ def test_report_and_qc_report_write_html_and_manifests(tmp_path): assert qc_result["status"] == "ok" assert "EEGPrep Report" in (tmp_path / "dataset_report.html").read_text(encoding="utf-8") assert "EEGPrep QC Report" in (tmp_path / "qc_report.html").read_text(encoding="utf-8") - assert json.loads((tmp_path / "dataset_report.manifest.json").read_text(encoding="utf-8"))["command"] == "report" - assert json.loads((tmp_path / "qc_report.manifest.json").read_text(encoding="utf-8"))["command"] == "qc report" + assert read_manifest(tmp_path / "dataset_report.manifest.json")["command"] == "report" + assert read_manifest(tmp_path / "qc_report.manifest.json")["command"] == "qc report" def _write_pipeline_config(tmp_path, *, steps, input_path=SAMPLE_SET): diff --git a/tests/test_cli_transforms.py b/tests/test_cli_transforms.py index 51de08e7..019caf0b 100644 --- a/tests/test_cli_transforms.py +++ b/tests/test_cli_transforms.py @@ -7,6 +7,7 @@ import subprocess import sys +from eegprep.cli.core import read_manifest from eegprep.cli.commands import transforms from eegprep.functions.popfunc.pop_loadset import pop_loadset from eegprep.functions.popfunc.pop_saveset import pop_saveset @@ -79,11 +80,14 @@ def test_resample_writes_dataset_manifest_and_clean_json_stdout(tmp_path): manifest_path = output.with_suffix(output.suffix + ".manifest.json") assert payload["manifest"]["path"] == str(manifest_path) - manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + stored_manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + assert {record["path"] for record in stored_manifest["output_files"]} == {"resampled.set"} + manifest = read_manifest(manifest_path) assert manifest["schema_version"] == transforms.MANIFEST_SCHEMA_VERSION assert manifest["parameters"]["freq"] == 128 assert "pop_resample" in manifest["history"] assert any(record["path"].endswith("eeglab_data.fdt") for record in manifest["input_files"]) + assert all(Path(record["path"]).is_absolute() for record in manifest["input_files"] + manifest["output_files"]) loaded = pop_loadset(str(output)) assert loaded["srate"] == 128 diff --git a/tests/test_manifest_paths.py b/tests/test_manifest_paths.py new file mode 100644 index 00000000..6e15cd9b --- /dev/null +++ b/tests/test_manifest_paths.py @@ -0,0 +1,150 @@ +import json +import os +from pathlib import Path + +import pytest + +from eegprep.cli.core import ( + MANIFEST_SCHEMA_VERSION, + build_manifest, + read_manifest, + write_manifest, + write_manifest_file, +) + + +def _build_manifest(input_path: Path, output_paths: list[Path | str]) -> dict: + return build_manifest( + command="test", + input_files=[input_path], + output_files=[{"path": str(path), "type": "artifact"} for path in output_paths], + parameters={}, + started_at="2026-01-01T00:00:00Z", + finished_at="2026-01-01T00:00:01Z", + ) + + +def test_manifest_v2_round_trip_uses_manifest_relative_posix_paths(tmp_path): + project = tmp_path / "project" + input_path = project / "raw" / "sub-01.set" + output_path = project / "derivatives" / "sub-01-clean.set" + sidecar_path = project / "derivatives" / "sub-01-clean.fdt" + manifest_path = project / "derivatives" / "manifests" / "sub-01.json" + input_path.parent.mkdir(parents=True) + input_path.write_bytes(b"input") + + manifest = _build_manifest(input_path, [output_path, sidecar_path]) + original_paths = [item["path"] for item in manifest["input_files"] + manifest["output_files"]] + + entry = write_manifest_file(manifest_path, manifest) + + assert entry["path"] == str(manifest_path) + stored = json.loads(manifest_path.read_text(encoding="utf-8")) + assert stored["schema_version"] == MANIFEST_SCHEMA_VERSION + assert stored["input_files"][0]["path"] == "../../raw/sub-01.set" + assert [item["path"] for item in stored["output_files"]] == [ + "../sub-01-clean.set", + "../sub-01-clean.fdt", + ] + assert [item["path"] for item in manifest["input_files"] + manifest["output_files"]] == original_paths + + loaded = read_manifest(manifest_path) + assert [item["path"] for item in loaded["input_files"] + loaded["output_files"]] == original_paths + + +def test_build_manifest_normalizes_runtime_relative_paths(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + input_path = Path("raw/input.set") + output_path = Path("derivatives/output.set") + input_path.parent.mkdir() + input_path.write_bytes(b"input") + + manifest = build_manifest( + command="test", + input_files=[input_path, {"path": str(input_path), "sha256": "recorded"}], + output_files=[{"path": str(output_path), "type": "artifact"}], + parameters={}, + started_at="2026-01-01T00:00:00Z", + finished_at="2026-01-01T00:00:01Z", + ) + + assert manifest["input_files"][0]["path"] == str((tmp_path / input_path).resolve()) + assert manifest["input_files"][1]["path"] == str((tmp_path / input_path).resolve()) + assert manifest["output_files"][0]["path"] == str((tmp_path / output_path).resolve()) + + +def test_write_manifest_preserves_relative_return_path(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + input_path = tmp_path / "input.set" + input_path.write_bytes(b"input") + manifest = _build_manifest(input_path, [tmp_path / "output.set"]) + manifest_path = Path("records/manifest.json") + + result = write_manifest(manifest_path, manifest) + file_manifest_path = Path("records/manifest-file.json") + entry = write_manifest_file(file_manifest_path, manifest) + + assert result == manifest_path + assert (tmp_path / manifest_path).is_file() + assert entry["path"] == str(file_manifest_path) + assert (tmp_path / file_manifest_path).is_file() + + +@pytest.mark.parametrize("schema_version", ["eegprep.manifest.v1", "eegprep.manifest.v3"]) +def test_read_manifest_does_not_reinterpret_other_schema_paths(tmp_path, schema_version): + manifest_path = tmp_path / "manifest.json" + payload = { + "schema_version": schema_version, + "input_files": [{"path": "../legacy/input.set"}], + "output_files": [{"path": "output.set"}], + } + manifest_path.write_text(json.dumps(payload), encoding="utf-8") + + assert read_manifest(manifest_path) == payload + + +def test_write_manifest_does_not_rewrite_v1_paths(tmp_path): + manifest_path = tmp_path / "manifest.json" + absolute_path = str((tmp_path / "input.set").resolve()) + payload = { + "schema_version": "eegprep.manifest.v1", + "input_files": [{"path": absolute_path}], + "output_files": [], + } + + write_manifest(manifest_path, payload) + + assert json.loads(manifest_path.read_text(encoding="utf-8")) == payload + + +def test_read_manifest_preserves_foreign_absolute_paths(tmp_path): + manifest_path = tmp_path / "manifest.json" + foreign_path = "/volume/input.set" if os.name == "nt" else r"C:\data\input.set" + payload = { + "schema_version": MANIFEST_SCHEMA_VERSION, + "input_files": [{"path": foreign_path}], + "output_files": [], + } + manifest_path.write_text(json.dumps(payload), encoding="utf-8") + + assert read_manifest(manifest_path)["input_files"][0]["path"] == foreign_path + + +@pytest.mark.skipif(os.name != "nt", reason="Windows drive semantics") +def test_manifest_keeps_paths_on_another_windows_drive_absolute(tmp_path): + manifest_path = tmp_path / "manifest.json" + manifest_drive = manifest_path.drive.casefold() + other_drive = next( + f"{letter}:" for letter in "CDEFGHIJKLMNOPQRSTUVWXYZ" if f"{letter}:".casefold() != manifest_drive + ) + foreign_path = rf"{other_drive}\data\input.set" + payload = { + "schema_version": MANIFEST_SCHEMA_VERSION, + "input_files": [{"path": foreign_path}], + "output_files": [], + } + + write_manifest(manifest_path, payload) + + stored = json.loads(manifest_path.read_text(encoding="utf-8")) + assert stored["input_files"][0]["path"] == foreign_path