fix: checks for existing processing, merges models (#436)

* fix: checks for existing processing, merges models

* raise error when merge fails, load of existing processing fails

* linter fix

* linter fix (in tests)

Author: mekhlakapoor

src/aind_metadata_mapper/gather_metadata.py

@@ -313,7 +313,7 @@ def _run_mappers_for_acquisition(self):
                 job_settings = MapperJobSettings(
                     input_filepath=Path(input_path),
                     output_directory=Path(input_dir),
-                    output_filename_suffix=mapper_name
+                    output_filename_suffix=mapper_name,
                 )
                 try:
                     mapper = mapper_cls()

src/aind_metadata_mapper/gather_processing_job.py

@@ -1,6 +1,7 @@
 """Entrypoint to create a processing.json file"""
 
 import json
+import logging
 import sys
 from pathlib import Path
 
@@ -30,12 +31,42 @@ def __init__(self, settings: JobSettings):
         """Class constructor"""
         self.settings = settings
 
+    def load_existing_processing(self) -> Processing | None:
+        """Load an existing processing.json file from the output directory, if it exists."""
+        file_name = Processing.default_filename()
+        existing_path = Path(self.settings.output_directory) / file_name
+        if existing_path.exists():
+            try:
+                with open(existing_path, "r") as f:
+                    existing_data = json.load(f)
+                return Processing(**existing_data)
+            except Exception as e:
+                error_msg = "Failed to load existing processing.json: {e}. "
+                logging.error(error_msg)
+                raise e
+        return None
+
     def run_job(self):
         """Write the processing object to a directory."""
 
+        existing_processing = self.load_existing_processing()
+        if existing_processing is not None:
+            try:
+                merged_processing = existing_processing + self.settings.processing
+            except Exception as e:
+                error_msg = (
+                    f"Failed to merge existing processing.json with new processing data: {e}. "
+                    "Cannot proceed without risking data loss. Please verify the existing processing.json "
+                    "file is valid and compatible with the new processing data."
+                )
+                logging.error(error_msg)
+                raise e
+        else:
+            merged_processing = self.settings.processing
+
         file_name = Processing.default_filename()
         output_path = Path(self.settings.output_directory) / file_name
-        json_contents = self.settings.processing.model_dump(mode="json", exclude_none=True)
+        json_contents = merged_processing.model_dump(mode="json", exclude_none=True)
         with open(output_path, "w") as f:
             json.dump(json_contents, f, indent=3, ensure_ascii=False, sort_keys=True)
 

tests/test_base.py

@@ -26,9 +26,7 @@ def test_jobsettings_fields(self):
         output_dir = Path("/tmp")
         output_suffix = "_test"
         settings = MapperJobSettings(
-            input_filepath=input_fp,
-            output_directory=output_dir,
-            output_filename_suffix=output_suffix
+            input_filepath=input_fp, output_directory=output_dir, output_filename_suffix=output_suffix
         )
         self.assertEqual(settings.input_filepath, input_fp)
         self.assertEqual(settings.output_directory, output_dir)
@@ -37,9 +35,7 @@ def test_jobsettings_fields(self):
     def test_mapperjob_not_implemented(self):
         """Test that MapperJob raises NotImplementedError"""
         settings = MapperJobSettings(
-            input_filepath=Path("/tmp/in.json"),
-            output_directory=Path("/tmp"),
-            output_filename_suffix="_test"
+            input_filepath=Path("/tmp/in.json"), output_directory=Path("/tmp"), output_filename_suffix="_test"
         )
         mapper = MapperJob()
         with self.assertRaises(NotImplementedError):
@@ -51,9 +47,7 @@ def test_mapperjob_subclass(self):
         output_dir = Path("/tmp")
         output_suffix = "_test"
         settings = MapperJobSettings(
-            input_filepath=input_fp,
-            output_directory=output_dir,
-            output_filename_suffix=output_suffix
+            input_filepath=input_fp, output_directory=output_dir, output_filename_suffix=output_suffix
         )
         mapper = DummyMapper()
         mapper.was_run = False

tests/test_gather_processing_job.py

@@ -23,10 +23,78 @@
 class TestGatherProcessingJob(unittest.TestCase):
     """Tests methods in GatherProcessingJob class."""
 
+    @patch("pathlib.Path.exists")
+    @patch("builtins.open", new_callable=mock_open, read_data='{"data_processes": []}')
+    @patch("json.load")
+    def test_load_existing_processing_exists_and_valid(
+        self, mock_json_load: MagicMock, mock_file_open: MagicMock, mock_exists: MagicMock
+    ):
+        """Tests load_existing_processing when file exists and is valid."""
+        mock_exists.return_value = True
+        existing_processing_dict = {
+            "data_processes": [
+                {
+                    "start_date_time": "2024-09-09T01:02:03",
+                    "end_date_time": "2024-09-10T01:02:03",
+                    "process_type": "Image atlas alignment",
+                    "experimenters": ["Existing Experimenter"],
+                    "stage": "Processing",
+                    "code": {"url": "www.example.com/existing"},
+                }
+            ]
+        }
+        mock_json_load.return_value = existing_processing_dict
+
+        example_processing = Processing(data_processes=[])
+        example_settings = JobSettings(output_directory="example", processing=example_processing)
+        job = GatherProcessingJob(settings=example_settings)
+
+        result = job.load_existing_processing()
+
+        self.assertIsNotNone(result)
+        self.assertIsInstance(result, Processing)
+        self.assertEqual(len(result.data_processes), 1)
+        mock_exists.assert_called_once()
+        mock_json_load.assert_called_once()
+
+    @patch("pathlib.Path.exists")
+    def test_load_existing_processing_does_not_exist(self, mock_exists: MagicMock):
+        """Tests load_existing_processing when file does not exist."""
+        mock_exists.return_value = False
+
+        example_processing = Processing(data_processes=[])
+        example_settings = JobSettings(output_directory="example", processing=example_processing)
+        job = GatherProcessingJob(settings=example_settings)
+
+        result = job.load_existing_processing()
+
+        self.assertIsNone(result)
+        mock_exists.assert_called_once()
+
+    @patch("pathlib.Path.exists")
+    @patch("builtins.open", new_callable=mock_open)
+    @patch("json.load")
+    @patch("logging.error")
+    def test_load_existing_processing_invalid_schema(
+        self, mock_log_error: MagicMock, mock_json_load: MagicMock, mock_file_open: MagicMock, mock_exists: MagicMock
+    ):
+        """Tests load_existing_processing when file exists but doesn't match Processing schema."""
+        mock_exists.return_value = True
+        mock_json_load.return_value = {"invalid_field": "value"}
+
+        example_processing = Processing(data_processes=[])
+        example_settings = JobSettings(output_directory="example", processing=example_processing)
+        job = GatherProcessingJob(settings=example_settings)
+
+        with self.assertRaises(Exception):
+            job.load_existing_processing()
+
+        mock_log_error.assert_called_once()
+
     @patch("builtins.open", new_callable=mock_open)
     @patch("json.dump")
     def test_run_job(self, mock_json_dump: MagicMock, mock_file_open: MagicMock):
-        """Tests run_job method."""
+        """Tests run_job method with example processing data."""
         # noinspection PyArgumentList
         example_processing = Processing(
             data_processes=[
@@ -58,6 +126,55 @@ def test_run_job(self, mock_json_dump: MagicMock, mock_file_open: MagicMock):
         mock_file_open.assert_has_calls(calls=[call(Path("example") / "processing.json", "w")])
         mock_json_dump.assert_called_once()
 
+    @patch("aind_metadata_mapper.gather_processing_job.GatherProcessingJob.load_existing_processing")
+    @patch("builtins.open", new_callable=mock_open)
+    @patch("json.dump")
+    @patch("logging.error")
+    def test_run_job_merge_fails_raises_error(
+        self,
+        mock_log_error: MagicMock,
+        mock_json_dump: MagicMock,
+        mock_file_open: MagicMock,
+        mock_load_existing: MagicMock,
+    ):
+        """Tests run_job raises exception when merge fails."""
+        existing_processing = Processing(
+            data_processes=[
+                DataProcess(
+                    start_date_time=datetime(2024, 9, 9, 1, 2, 3),
+                    end_date_time=datetime(2024, 9, 10, 1, 2, 3),
+                    process_type=ProcessName.IMAGE_ATLAS_ALIGNMENT,
+                    experimenters=["Existing Experimenter"],
+                    stage=ProcessStage.PROCESSING,
+                    code=Code(url="www.example.com/existing_process"),
+                )
+            ]
+        )
+        mock_load_existing.return_value = existing_processing
+        with patch.object(Processing, "__add__", side_effect=Exception("Merge failed")):
+            new_processing = Processing(
+                data_processes=[
+                    DataProcess(
+                        start_date_time=datetime(2024, 10, 10, 1, 2, 3),
+                        end_date_time=datetime(2024, 10, 11, 1, 2, 3),
+                        process_type=ProcessName.COMPRESSION,
+                        experimenters=["AIND Scientific Computing"],
+                        stage=ProcessStage.PROCESSING,
+                        code=Code(url="www.example.com/ephys_compression"),
+                    )
+                ]
+            )
+            example_settings = JobSettings(output_directory="example", processing=new_processing)
+            job = GatherProcessingJob(settings=example_settings)
+
+            with self.assertRaises(Exception) as context:
+                job.run_job()
+
+            self.assertIn("Merge failed", str(context.exception))
+            mock_log_error.assert_called_once()
+            self.assertIn("Failed to merge existing processing.json", str(mock_log_error.call_args))
+            mock_json_dump.assert_not_called()
+
 
 if __name__ == "__main__":
     unittest.main()