Update acquisition_type logic

pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
 name = "aind-metadata-mapper"
 description = "Package to manage mapping of source data into aind-data-schema metadata files."
 license = {text = "MIT"}
-requires-python = ">=3.10"
+requires-python = ">=3.10,<3.14"
 authors = [
     {name = "Allen Institute for Neural Dynamics"}
 ]
@@ -21,7 +21,7 @@ dependencies = [
     "pyyaml",
     "pydantic-settings>=2.8.0",
     "jsonschema",
-    "aind-data-schema>=2.9.0,<3",
+    "aind-data-schema>=2.9.1,<3",
     "aind-metadata-extractor==0.3.11",
 ]
 

src/aind_metadata_mapper/fip/constants.py

@@ -38,14 +38,6 @@
 DEFAULT_LED_POWER = 1.0
 
 
-# ==============================================================================
-# Acquisition Type
-# ==============================================================================
-# Hardcoded acquisition type for VrForaging-FIP experiments
-# NOTE: Ideally this would come from the extracted metadata, but we don't have it yet.
-ACQUISITION_TYPE_AIND_VR_FORAGING = "AindVrForaging"
-
-
 # ==============================================================================
 # Device Name Transformations
 # ==============================================================================

src/aind_metadata_mapper/fip/mapper.py

@@ -34,7 +34,6 @@
 
 from aind_metadata_mapper.base import MapperJob, MapperJobSettings
 from aind_metadata_mapper.fip.constants import (
-    ACQUISITION_TYPE_AIND_VR_FORAGING,
     AIND_PHYSIOLOGY_FIP_PACKAGE_NAME,
     AIND_PHYSIOLOGY_FIP_REPO_URL,
     CAMERA_EXPOSURE_TIME_MICROSECONDS_PER_MILLISECOND,
@@ -307,6 +306,7 @@ def transform(
         intended_measurements: Optional[Dict[str, Dict[str, Optional[str]]]] = None,
         implanted_fibers: Optional[List[int]] = None,
         ethics_review_id: Optional[List[str]] = None,
+        acquisition_type: Optional[str] = None,
     ) -> Acquisition:
         """Transforms intermediate metadata into a complete Acquisition model.
 
@@ -329,6 +329,10 @@ def transform(
             protocol is found, in which case it is filled by the acquisition merge
             downstream. If provided, the value is verified against LabTracks and a
             ValueError is raised on mismatch.
+        acquisition_type : Optional[str], optional
+            Descriptive acquisition type. Passed straight through when provided. If None,
+            an empty string is used rather than None, since aind-data-schema requires the
+            field; an empty value is left to be filled by the acquisition merge downstream.
 
         Returns
         -------
@@ -430,7 +434,10 @@ def transform(
             experimenters=session.get("experimenter", []),
             ethics_review_id=ethics_review_id,
             instrument_id=instrument_id,
-            acquisition_type=ACQUISITION_TYPE_AIND_VR_FORAGING,
+            # Empty string rather than None when unset: acquisition_type is required by
+            # aind-data-schema, and an empty value is filled in by the acquisition merge
+            # downstream (see aind-data-schema#1876).
+            acquisition_type=acquisition_type or "",
             notes=session.get("notes"),
             data_streams=[data_stream],
             stimulus_epochs=[],

src/aind_metadata_mapper/gather_metadata.py

@@ -364,7 +364,12 @@ def get_acquisition(self) -> Optional[dict]:
                 directory=self.settings.metadata_dir, file_name_prefix="acquisition"
             )
             if files:
-                return self._merge_models(Acquisition, files)
+                merged_acquisition = self._merge_models(Acquisition, files)
+
+                if not merged_acquisition.get("acquisition_type") and self.settings.raise_if_invalid:
+                    raise ValueError("Acquisition.acquisition_type is empty in the merged acquisition metadata.")
+
+                return merged_acquisition
             else:
                 logging.debug("No acquisition metadata file found.")
                 return None

tests/fip/test_mapper.py

@@ -19,7 +19,6 @@
 
 from aind_metadata_mapper.fip import mapper as mapper_mod
 from aind_metadata_mapper.fip.constants import (
-    ACQUISITION_TYPE_AIND_VR_FORAGING,
     AIND_PHYSIOLOGY_FIP_PACKAGE_NAME,
     AIND_PHYSIOLOGY_FIP_REPO_URL,
 )
@@ -78,7 +77,7 @@ def test_transform_basic(self):
 
         self.assertEqual(acquisition.subject_id, "test")
         self.assertEqual(acquisition.instrument_id, "test_rig")
-        self.assertEqual(acquisition.acquisition_type, ACQUISITION_TYPE_AIND_VR_FORAGING)
+        self.assertEqual(acquisition.acquisition_type, "")
         self.assertEqual(len(acquisition.experimenters), 2)
         self.assertEqual(acquisition.experimenters[0], "Foo")
         self.assertEqual(acquisition.notes, "test session")
@@ -281,6 +280,38 @@ def test_code_field_without_rig_version(self):
         data_stream = acquisition.data_streams[0]
         self.assertIsNone(data_stream.code)
 
+    def test_acquisition_type_passed_through(self):
+        """Test that a provided acquisition_type is used verbatim."""
+        acquisition = self.mapper.transform(
+            self.example_intermediate_data,
+            skip_validation=True,
+            intended_measurements=self.test_intended_measurements,
+            implanted_fibers=self.test_implanted_fibers,
+            ethics_review_id=self.test_ethics_review_id,
+            acquisition_type="AindVrForaging",
+        )
+        self.assertEqual(acquisition.acquisition_type, "AindVrForaging")
+
+    def test_acquisition_type_defaults_to_empty_string(self):
+        """Test that an unset acquisition_type becomes an empty string, not None.
+
+        acquisition_type is required by aind-data-schema, so None is not a valid
+        value. An empty string is left to be filled by the acquisition merge
+        downstream, the same way ethics_review_id is.
+        """
+        for value in (None, ""):
+            with self.subTest(acquisition_type=value):
+                acquisition = self.mapper.transform(
+                    self.example_intermediate_data,
+                    skip_validation=True,
+                    intended_measurements=self.test_intended_measurements,
+                    implanted_fibers=self.test_implanted_fibers,
+                    ethics_review_id=self.test_ethics_review_id,
+                    acquisition_type=value,
+                )
+                self.assertEqual(acquisition.acquisition_type, "")
+                self.assertIsNotNone(acquisition.acquisition_type)
+
     def test_active_devices(self):
         """Test that active devices list is populated with all FIP system components.
 

tests/test_gather_metadata.py

@@ -768,6 +768,24 @@ def test_get_acquisition_no_file(self, mock_file_exists):
 
         self.assertIsNone(result)
 
+    def test_get_acquisition_empty_type_raises_when_invalid(self):
+        """Test get_acquisition rejects merged metadata with an empty acquisition_type."""
+        settings = self.test_settings.model_copy(update={"raise_if_invalid": True})
+        with patch("os.makedirs"):
+            job = GatherMetadataJob(settings=settings)
+
+        with (
+            patch.object(job, "_does_file_exist_in_user_defined_dir", return_value=False),
+            patch.object(job, "_run_mappers_for_acquisition"),
+            patch.object(job, "_get_prefixed_files_from_directory", return_value=[{}]),
+            patch.object(job, "_merge_models", return_value={"acquisition_type": ""}),
+        ):
+            with self.assertRaisesRegex(
+                ValueError,
+                "Acquisition.acquisition_type is empty in the merged acquisition metadata.",
+            ):
+                job.get_acquisition()
+
     @patch("os.makedirs")
     def test_run_mappers_for_acquisition_no_metadata_dir(self, mock_makedirs):
         """Test _run_mappers_for_acquisition when metadata_dir is None"""
@@ -923,7 +941,9 @@ def test_merge_models_acquisition(self):
             base_acquisition = json.load(f)
 
         acquisition1 = base_acquisition.copy()
+        acquisition1["acquisition_type"] = ""
         acquisition2 = base_acquisition.copy()
+        acquisition2["acquisition_type"] = "test"
 
         if "subject_details" in acquisition2:
             del acquisition2["subject_details"]
@@ -932,6 +952,7 @@ def test_merge_models_acquisition(self):
 
         self.assertIsInstance(result, dict)
         self.assertIn("acquisition_start_time", result)
+        self.assertEqual(result["acquisition_type"], "test")
 
     def test_merge_models_datetime_serialization(self):
         """Test that merged models can be JSON serialized (datetime objects converted to strings)"""