🐛 Reseparate longitudinal expectedOutputs from sessions specific

Author: shnizzedy

CPAC/pipeline/cpac_pipeline.py

@@ -1,4 +1,4 @@
-# Copyright (C) 2012-2024  C-PAC Developers
+# Copyright (C) 2012-2025  C-PAC Developers
 
 # This file is part of C-PAC.
 
@@ -271,8 +271,7 @@ def run_workflow(
 
     subject_id, p_name, log_dir = set_subject(sub_dict, c)
     c["subject_id"] = subject_id
-
-    init_loggers(subject_id, c, log_dir, mock=True, longitudinal=False)
+    init_loggers(subject_id, c, log_dir, mock=True)
 
     # Start timing here
     pipeline_start_time = time.time()
@@ -1208,7 +1207,6 @@ def build_workflow(subject_id, sub_dict, cfg, pipeline_name=None):
     wf = initialize_nipype_wf(
         cfg, sub_dict["subject_id"], sub_dict.get("unique_id", None), name=pipeline_name
     )
-
     # Extract credentials path if it exists
     try:
         creds_path = sub_dict["creds_path"]

CPAC/pipeline/cpac_runner.py

@@ -253,7 +253,7 @@ def run_T1w_longitudinal(sublist, cfg: Configuration, dry_run: bool = False):
             log_dir: str
             _, _, log_dir = set_subject(sub_list[0], cfg)
             log_dir = str(Path(log_dir).parent / f"{subject_id}_longitudinal")
-            init_loggers(subject_id, cfg, log_dir, mock=True, longitudinal=True)
+            init_loggers(subject_id, cfg, log_dir, mock=True)
             anat_longitudinal_wf(subject_id, sub_list, cfg, dry_run=dry_run)
         elif len(sub_list) == 1:
             warnings.warn(

CPAC/pipeline/engine.py

@@ -50,6 +50,7 @@
 from CPAC.utils.bids_utils import res_in_filename
 from CPAC.utils.configuration import Configuration
 from CPAC.utils.datasource import (
+    bidsier_prefix,
     create_anat_datasource,
     create_func_datasource,
     create_general_datasource,
@@ -64,6 +65,7 @@
     WARNING_FREESURFER_OFF_WITH_DATA,
     WFLOGGER,
 )
+from CPAC.utils.monitoring.custom_logging import MockLogger
 from CPAC.utils.outputs import Outputs
 from CPAC.utils.utils import (
     check_prov_for_regtool,
@@ -1133,10 +1135,20 @@ def post_process(self, wf, label, connection, json_info, pipe_idx, pipe_x, outs)
     def gather_pipes(self, wf, cfg, all=False, add_incl=None, add_excl=None):
         excl = []
         substring_excl = []
-        outputs_logger = getLogger(
-            f'{cfg.get("subject_id", getattr(wf, "name", ""))}_expectedOutputs'
-        )
-        expected_outputs = ExpectedOutputs()
+        try:
+            unique_id = re.match(r"(.*_)(sub-.*)", wf.name).group(2)  # pyright: ignore[reportOptionalMemberAccess]
+        except (AttributeError, IndexError):
+            unique_id = cfg.get("subject_id", getattr(wf, "name", ""))
+        unique_id = bidsier_prefix(unique_id)
+        outputs_logger = getLogger(f"{unique_id}_expectedOutputs")
+        expected = {}
+        if isinstance(outputs_logger, MockLogger):
+            try:
+                # load already-expected outputs
+                expected = outputs_logger.yaml_contents()
+            except (FileNotFoundError, TypeError):
+                pass
+        expected_outputs = ExpectedOutputs(expected)
 
         if add_excl:
             excl += add_excl

CPAC/utils/monitoring/custom_logging.py

@@ -1,4 +1,4 @@
-# Copyright (C) 2022-2024  C-PAC Developers
+# Copyright (C) 2022-2025  C-PAC Developers
 
 # This file is part of C-PAC.
 
@@ -24,6 +24,7 @@
 from traceback import print_exception
 from typing import Literal, Optional, Sequence, TYPE_CHECKING, TypeAlias
 
+import yaml
 from nipype import config as nipype_config, logging as nipype_logging
 
 from CPAC.utils.docs import docstring_parameter
@@ -163,6 +164,22 @@ def log_subprocess(cmd, *args, raise_error=True, **kwargs):
     return output, 0
 
 
+class ListToSetYamlLoader(yaml.Loader):
+    """Custom YAML loader to convert lists to sets."""
+
+    def construct_sequence(  # pyright: ignore[reportIncompatibleMethodOverride]
+        self, node, deep=False
+    ) -> set[str]:
+        """Convert YAML sequence to a set."""
+        return set(super().construct_sequence(node, deep))
+
+
+ListToSetYamlLoader.add_constructor(
+    yaml.resolver.BaseResolver.DEFAULT_SEQUENCE_TAG,
+    ListToSetYamlLoader.construct_sequence,
+)
+
+
 # pylint: disable=too-few-public-methods
 class MockHandler:
     """Handler for MockLogger."""
@@ -233,6 +250,24 @@ def _get_first_file_handler(
                 return handler
         return None
 
+    def yaml_contents(self) -> dict:
+        """If the logger's first handler is a YAML file, return the contents and delete them from the logger."""
+        file = self._get_first_file_handler(self.handlers)
+        if hasattr(file, "baseFilename"):
+            file = Path(getattr(file, "baseFilename"))
+            if file.suffix == ".yml":
+                with file.open("r", encoding="utf-8") as f:
+                    contents = yaml.load(f.read(), Loader=ListToSetYamlLoader)
+                with file.open("w", encoding="utf-8") as f:
+                    f.write("")
+                return contents
+            error = TypeError
+            msg = f"Could not load YAML contents from {file}"
+        else:
+            error = FileNotFoundError
+            msg = f"Could not find file handler for {self.name}"
+        raise error(msg)
+
 
 def _lazy_sub(message, *items):
     """Given lazy-logging syntax, return string with substitutions.
@@ -268,7 +303,6 @@ def set_up_logger(
     level: Optional[LogLevel] = None,
     log_dir: Optional[Path | str] = None,
     mock: bool = False,
-    overwrite_existing: bool = False,
 ) -> logging.Logger | MockLogger:
     r"""Initialize a logger.
 
@@ -328,9 +362,6 @@ def set_up_logger(
         level = logging.NOTSET
     log_dir = Path(log_dir) if log_dir else Path.cwd()
     filepath = log_dir / filename
-    if overwrite_existing and filepath.exists():
-        with filepath.open("w", encoding="utf-8") as log_file:
-            log_file.write("")
     if not filepath.exists():
         filepath.parent.mkdir(parents=True, exist_ok=True)
     if mock:
@@ -356,16 +387,12 @@ def init_loggers(
 
     if "subject_id" not in cpac_config:
         cpac_config["subject_id"] = subject_id
-
     set_up_logger(
-        f"{subject_id}_expectedOutputs",
+        f"{cpac_config['subject_id']}_expectedOutputs",
         filename=f"{bidsier_prefix(cpac_config['subject_id'])}_expectedOutputs.yml",
         level="info",
         log_dir=log_dir,
         mock=mock,
-        overwrite_existing=(  # don't overwrite if we have a longitudinal template
-            longitudinal or not cpac_config["longitudinal_template_generation", "run"]
-        ),
     )
 
     if cpac_config["pipeline_setup", "Debugging", "verbose"]: