Fix NIDM append discarding all but one stats file; align with PyNIDM 4.5.0 + per-subject NIDM input

Dockerfile

@@ -32,7 +32,7 @@ RUN python3 -m pip install --upgrade pip setuptools wheel && \
     python3 -m pip install -e . && \
     cd /opt && \
     python3 -m pip install -r requirements.txt && \
-    python3 -m pip install --no-cache-dir --upgrade 'rdflib>=7.0.0' && \
+    python3 -m pip install --no-cache-dir --upgrade 'rdflib>=7.0.0,<8' && \
     python3 -m pip install -e .
 
 # =======================================

Singularity

@@ -32,7 +32,7 @@ From: vnmd/freesurfer_8.0.0
 
     cd /opt
     python3 -m pip install -r requirements.txt
-    python3 -m pip install --no-cache-dir --upgrade 'rdflib>=7.0.0'
+    python3 -m pip install --no-cache-dir --upgrade 'rdflib>=7.0.0,<8'
     python3 -m pip install -e .
 
 %environment

requirements.txt

@@ -3,4 +3,10 @@ click>=8.0.0
 pybids>=0.15.1
 pytest>=7.0.0
 rdflib~=6.3.2
-pynidm==4.2.4
+pynidm==4.5.0
+# PyNIDM 4.5.0 adds an oxigraph-backed rdflib store; pinned explicitly so the
+# runtime dep is visible. Requires rdflib<8 (see build files' rdflib upgrade).
+oxrdflib~=0.5.0
+# prov 3.0.0 moved NetworkX graph interop to the optional "graph" extra; pynidm's
+# nidm.experiment modules import prov.graph, so the extra (networkx) is required.
+prov[graph]

src/run.py

@@ -146,11 +146,20 @@ def nidm_conversion(
 
     existing_nidm_file = None
     if nidm_input_dir and nidm_input_dir.exists():
-        # Prefer top-level nidm.ttl then fall back to any .ttl/.jsonld file present
-        primary_candidate = nidm_input_dir / "nidm.ttl"
-        if primary_candidate.exists():
-            existing_nidm_file = primary_candidate
-        else:
+        # Look for an existing NIDM file to append to, most specific first.
+        # Supports both the legacy single-file layout (top-level nidm.ttl) and
+        # the per-subject layout used by newer NIDM datasets (e.g. nidm_4.5.0:
+        # sub-<id>/nidm.ttl, optionally nested under a ses-<id> subdir).
+        subject_dirname = f"sub-{participant_label}"
+        candidates = []
+        if bids_session:
+            candidates.append(nidm_input_dir / subject_dirname / f"ses-{bids_session}" / "nidm.ttl")
+            candidates.append(nidm_input_dir / f"{subject_dirname}_ses-{bids_session}" / "nidm.ttl")
+        candidates.append(nidm_input_dir / subject_dirname / "nidm.ttl")
+        candidates.append(nidm_input_dir / "nidm.ttl")
+        existing_nidm_file = next((c for c in candidates if c.exists()), None)
+        if existing_nidm_file is None:
+            # Legacy fallback: any top-level serialized NIDM file.
             for pattern in ("*.ttl", "*.jsonld", "*.json-ld"):
                 try:
                     existing_nidm_file = next(nidm_input_dir.glob(pattern))
@@ -182,7 +191,13 @@ def nidm_conversion(
     # Build command: use -n for existing NIDM file, -o for new output
     # Note: fs_to_nidm does not allow both -n and -o at the same time
     if copied_nidm:
-        cmd = [sys.executable, "-m", module_name, "-s", subject_dir, "-n", str(copied_nidm), "-j", "--forcenidm"]
+        # NOTE: do NOT pass -j here. In fs_to_nidm's append branch the per-stats-file
+        # loop accumulates by re-reading the -n file each iteration and writing back to
+        # it. With -j the result is written to "<nidm_file>.json" instead, so the next
+        # iteration re-reads the *unmodified* -n file and every stats file except the
+        # last-globbed one is silently discarded (135 vs 5188 measurements on ABIDE
+        # sub-0051456, and nondeterministic because glob order is filesystem-dependent).
+        cmd = [sys.executable, "-m", module_name, "-s", subject_dir, "-n", str(copied_nidm), "--forcenidm"]
         if verbose:
             logger.info(f"Found existing NIDM file: {existing_nidm_file}")
             logger.info(f"Adding data to existing NIDM file: {copied_nidm}")
@@ -216,8 +231,10 @@ def nidm_conversion(
             logger.error(f"Command output: {result.stdout}")
         sys.exit(1)
 
-    # Log what files exist in output directory after conversion
-    logger.info(f"Files in NIDM output directory after conversion:")
+    # Log what files exist in output directory after conversion. These are the
+    # intermediate converter outputs; the aggregated per-subject TTL is written
+    # further below and logged separately.
+    logger.info(f"Intermediate NIDM converter outputs:")
     for file in Path(nidm_dir).glob("*"):
         logger.info(f"  - {file.name} ({file.stat().st_size} bytes)")
 
@@ -233,7 +250,23 @@ def nidm_conversion(
     elif existing_nidm_file and existing_nidm_file.exists():
         aggregation_sources.append(existing_nidm_file)
 
+    # The FreeSurfer CDE vocabulary (fs_cde.ttl) is a static, deterministic lookup
+    # table -- byte-identical for every subject and every run. It is required to
+    # interpret the fs_* measurement predicates, but it is already written to this
+    # same nidm/ directory as its own file, so folding it into each per-subject TTL
+    # is pure duplication (~2.0 MB of a 2.3 MB output, per subject). Excluding it
+    # keeps the per-subject TTL at ~250 KB of genuinely subject-specific triples and
+    # lets git-annex store the shared vocabulary once. This matches the sibling
+    # fsl-nidm BIDSapp, which likewise ships fsl_cde.ttl alongside rather than inside
+    # its merged nidm.ttl. Consumers must load nidm/fs_cde.ttl to resolve fs_* terms.
+    CDE_FILENAMES = {"fs_cde.ttl"}
     for candidate in new_outputs:
+        if candidate.name in CDE_FILENAMES:
+            logger.info(
+                f"Not merging shared CDE vocabulary into per-subject TTL: {candidate.name} "
+                f"(shipped alongside in {Path(nidm_dir).name}/)"
+            )
+            continue
         if candidate.suffix.lower() in {".ttl", ".json", ".jsonld", ".json-ld"}:
             aggregation_sources.append(candidate)
 
@@ -265,6 +298,11 @@ def _guess_rdf_format(path: Path) -> str:
 
     try:
         aggregated_graph.serialize(destination=str(target_ttl), format="turtle")
+        logger.info(
+            f"Wrote aggregated NIDM output: {target_ttl.name} "
+            f"({target_ttl.stat().st_size} bytes, {len(aggregated_graph)} triples) "
+            f"from {len(aggregation_sources)} source(s)"
+        )
     except Exception as serialize_error:  # pragma: no cover
         logger.warning(f"Failed to write aggregated TTL output {target_ttl}: {serialize_error}")
 

tests/test_nidm_handling.py

@@ -250,6 +250,118 @@ def test_skip_nidm_flag(self, tmp_path, bids_single_session):
             assert len(nidm_calls) == 0
 
 
+class TestNIDMAppendCommand:
+    """Regression tests for the fs_to_nidm append (-n) command construction."""
+
+    def _nidm_cmd(self, mock_subprocess):
+        """Return the argv list of the fs_to_nidm invocation, or None."""
+        for c in mock_subprocess.call_args_list:
+            argv = c.args[0] if c.args else c.kwargs.get("args")
+            if isinstance(argv, list) and any("fs_to_nidm" in str(a) for a in argv):
+                return argv
+        return None
+
+    def test_append_command_does_not_pass_jsonld(self, tmp_path, bids_single_session):
+        """-j must NOT be passed alongside -n.
+
+        fs_to_nidm's append branch accumulates across stats files by re-reading the
+        -n file each loop iteration and writing back to it. With -j the output goes
+        to '<nidm_file>.json' instead, so every stats file except the last-globbed
+        one is silently discarded. See src/run.py for the full explanation.
+        """
+        nidm_dir = bids_single_session.parent / "NIDM"
+        nidm_dir.mkdir()
+        (nidm_dir / "nidm.ttl").write_text("@prefix : <http://example.org/> .")
+
+        output_dir = tmp_path / "output"
+        output_dir.mkdir()
+
+        runner = CliRunner()
+        with patch('src.run.FreeSurferWrapper') as mock_wrapper, \
+             patch('src.run.subprocess.run') as mock_subprocess:
+            mock_wrapper.return_value.process_subject.return_value = True
+            mock_subprocess.return_value = MagicMock(returncode=0, stdout="", stderr="")
+
+            runner.invoke(cli, [
+                str(bids_single_session),
+                str(output_dir),
+                'participant',
+                '--participant-label', '001',
+                '--skip-bids-validation'
+            ])
+
+            cmd = self._nidm_cmd(mock_subprocess)
+            assert cmd is not None, "fs_to_nidm was never invoked"
+            assert "-n" in cmd, f"expected append mode (-n) in: {cmd}"
+            assert "-j" not in cmd, (
+                "-j must not be combined with -n; it breaks multi-stats-file "
+                f"accumulation. Got: {cmd}"
+            )
+            assert "--forcenidm" in cmd, f"expected --forcenidm in: {cmd}"
+
+    def test_shared_cde_not_merged_into_per_subject_ttl(self, tmp_path, bids_single_session):
+        """fs_cde.ttl must ship alongside, not be folded into each subject TTL.
+
+        The CDE vocabulary is a static lookup table, byte-identical for every
+        subject, and is already written to the same nidm/ directory. Embedding it
+        makes every per-subject TTL unique and ~2 MB larger for no information gain.
+        """
+        nidm_in = bids_single_session.parent / "NIDM"
+        nidm_in.mkdir()
+        (nidm_in / "nidm.ttl").write_text(
+            '@prefix ex: <http://example.org/> .\nex:subject a ex:Person .\n'
+        )
+
+        output_dir = tmp_path / "output"
+        output_dir.mkdir()
+
+        # Match on the local name, not the full URI: rdflib re-serializes
+        # <http://example.org/CDE_ONLY_TERM> in prefixed form (ex:CDE_ONLY_TERM),
+        # so asserting on the full URI would pass even when the CDE was merged.
+        cde_marker = "CDE_ONLY_TERM"
+
+        def fake_convert(cmd, *args, **kwargs):
+            """Emulate fs_to_nidm: append to the -n file and drop a fs_cde.ttl."""
+            nidm_file = Path(cmd[cmd.index("-n") + 1])
+            nidm_file.write_text(
+                '@prefix ex: <http://example.org/> .\n'
+                'ex:subject a ex:Person .\n'
+                'ex:collection ex:measure "42" .\n'
+            )
+            (nidm_file.parent / "fs_cde.ttl").write_text(
+                '@prefix ex: <http://example.org/> .\n'
+                f'ex:{cde_marker} a ex:DataElement .\n'
+            )
+            return MagicMock(returncode=0, stdout="", stderr="")
+
+        runner = CliRunner()
+        with patch('src.run.FreeSurferWrapper') as mock_wrapper, \
+             patch('src.run.subprocess.run', side_effect=fake_convert):
+            mock_wrapper.return_value.process_subject.return_value = True
+
+            runner.invoke(cli, [
+                str(bids_single_session),
+                str(output_dir),
+                'participant',
+                '--participant-label', '001',
+                '--skip-bids-validation'
+            ])
+
+        nidm_out = output_dir / "freesurfer-nidm_bidsapp" / "nidm"
+        aggregated = nidm_out / "sub-001.ttl"
+        assert aggregated.exists(), f"aggregated TTL missing; have: {list(nidm_out.glob('*'))}"
+
+        # The shared vocabulary must still be shipped...
+        assert (nidm_out / "fs_cde.ttl").exists(), "fs_cde.ttl must ship alongside"
+        # ...but must NOT have been merged into the per-subject TTL.
+        assert cde_marker not in aggregated.read_text(), (
+            "fs_cde.ttl content was merged into the per-subject TTL; it should be "
+            "shipped alongside instead"
+        )
+        # Subject-specific content must still be there.
+        assert "collection" in aggregated.read_text()
+
+
 class TestNIDMDatasetDescription:
     """Test NIDM dataset_description.json creation."""