Barseq blackbox acquisition (#1781)

* First pass at minimal blackbox acquisition. Has file that generates acquisition for two subjects, plus json outputs

* Added reference to output files

* Added specimen IDs to barseq acquisitions

* Removed readme file

* Updating examples/barseq_acquisition.py docstring

* Cleaned up comments in examples/barseq_acquisition.py

* Set timestamps to pacific with offset

* Changed experimenters to 'Barseq team'

* Removed references to local paths, removed duplicate jsons

* Added ExternalDataStream, made instrument optional, added test, updated example barseq

* Updated example json outputs for barseq subjects

* update docs [skip actions]

* Linting fixes

* Used DiscriminatedList

* Clean up stream adding logic

* Simplify barseq acquisition example

* Removed jsons

* clean up test and import properly

* Add ExternalDataStream support to Metadata validators and test that no instrument warning is raised for external acquisitions.

* Wording updates as suggested by Dan

---------

Co-authored-by: github-actions <41898282+github-actions[bot]@users.noreply.github.com>

Author: dougollerenshaw

docs/source/acquisition.md

@@ -85,13 +85,13 @@ while the StimulusEpoch represents all stimuli being presented.
 | `experimenters` | `List[str]` | experimenter(s)  |
 | `protocol_id` | `Optional[List[str]]` | Protocol ID (DOI for protocols.io) |
 | `ethics_review_id` | `Optional[List[str]]` | Ethics review ID  |
-| `instrument_id` | `str` | Instrument ID (Should match the Instrument.instrument_id) |
+| `instrument_id` | `Optional[str]` | Instrument ID (Should match the Instrument.instrument_id. Required when instrument metadata is available.) |
 | `acquisition_type` | `str` | Acquisition type (Descriptive string detailing the type of acquisition, should be consistent across similar acquisitions for the same experiment.) |
 | `notes` | `Optional[str]` | Notes  |
 | `coordinate_system` | Optional[[CoordinateSystem](components/coordinates.md#coordinatesystem)] | Coordinate system (Origin and axis definitions for determining the configured position of devices during acquisition. Required when coordinates are provided within the Acquisition) |
 | `calibrations` | List[[Calibration](components/measurements.md#calibration) or [VolumeCalibration](components/measurements.md#volumecalibration) or [PowerCalibration](components/measurements.md#powercalibration)] | Calibrations (List of calibration measurements taken prior to acquisition.) |
 | `maintenance` | List[[Maintenance](components/measurements.md#maintenance)] | Maintenance (List of maintenance on instrument prior to acquisition.) |
-| `data_streams` | List[[DataStream](acquisition.md#datastream)] | Data streams (A data stream is a collection of devices that are acquiring data simultaneously. Each acquisition can include multiple streams. Streams should be split when configurations are changed.) |
+| `data_streams` | List[[DataStream](acquisition.md#datastream) or [ExternalDataStream](acquisition.md#externaldatastream)] | Data streams (A data stream is a collection of devices that are acquiring data simultaneously. Each acquisition can include multiple streams. Streams should be split when configurations are changed. Use ExternalDataStream for acquisitions where instrument metadata is unavailable.) |
 | `stimulus_epochs` | List[[StimulusEpoch](acquisition.md#stimulusepoch)] | Stimulus (A stimulus epoch captures all stimuli being presented during an acquisition. Epochs should be split when the purpose of the stimulus changes.) |
 | `manipulations` | List[[Manipulation](acquisition.md#manipulation)] | Manipulations (Procedures performed during the acquisition.) |
 | `subject_details` | Optional[[AcquisitionSubjectDetails](acquisition.md#acquisitionsubjectdetails)] | Subject details (Required for in vivo acquisitions.) |
@@ -131,6 +131,18 @@ same time.
 | `connections` | List[[Connection](components/connections.md#connection)] | Connections (Connections are links between devices that are specific to this acquisition (i.e. not already defined in the Instrument)) |
 
 
+### ExternalDataStream
+
+A simplified data stream for acquisitions where instrument metadata is unavailable.
+
+| Field | Type | Title (Description) |
+|-------|------|-------------|
+| `stream_start_time` | `datetime (timezone-aware)` | Stream start time  |
+| `stream_end_time` | `datetime (timezone-aware)` | Stream stop time  |
+| `modalities` | List[[Modality](aind_data_schema_models/modalities.md#modality)] | Modalities (Modalities that are acquired in this stream) |
+| `notes` | `Optional[str]` | Notes  |
+
+
 ### Manipulation
 
 Description of procedures performed during an acquisition.

examples/barseq_acquisition.py

@@ -0,0 +1,40 @@
+"""Example of a BARseq acquisition using ExternalDataStream.
+
+BARseq is acquired by the AIBS Molecular Anatomy team using an instrument
+and workflows not yet fully documented in the AIND schema. ExternalDataStream
+is used here to record acquisition metadata without requiring full instrument
+configuration details, treating the data as if it were acquired externally.
+"""
+
+import argparse
+from datetime import datetime
+from zoneinfo import ZoneInfo
+
+from aind_data_schema_models.modalities import Modality
+
+from aind_data_schema.core.acquisition import Acquisition, ExternalDataStream
+
+acquisition = Acquisition(
+    subject_id="123456",
+    specimen_id=["123456_bar001", "123456_bar002"],
+    acquisition_start_time=datetime(2025, 1, 1, 9, 0, 0, tzinfo=ZoneInfo("America/Los_Angeles")),
+    acquisition_end_time=datetime(2025, 1, 31, 17, 0, 0, tzinfo=ZoneInfo("America/Los_Angeles")),
+    acquisition_type="BarcodeSequencing",
+    data_streams=[
+        ExternalDataStream(
+            stream_start_time=datetime(2025, 1, 1, 9, 0, 0, tzinfo=ZoneInfo("America/Los_Angeles")),
+            stream_end_time=datetime(2025, 1, 31, 17, 0, 0, tzinfo=ZoneInfo("America/Los_Angeles")),
+            modalities=[Modality.BARSEQ],
+            notes="Acquired externally.",
+        )
+    ],
+)
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--output-dir", default=None, help="Output directory for generated JSON file")
+    args = parser.parse_args()
+
+    serialized = acquisition.model_dump_json()
+    deserialized = Acquisition.model_validate_json(serialized)
+    deserialized.write_standard_file(prefix="barseq", output_directory=args.output_dir)

src/aind_data_schema/core/acquisition.py

@@ -108,6 +108,7 @@ class DataStream(DataModel):
     same time.
     """
 
+    object_type: Literal["DataStream"] = "DataStream"
     stream_start_time: Annotated[
         AwareDatetimeWithDefault,
         Field(..., title="Stream start time"),
@@ -250,6 +251,26 @@ def __add__(self, other: "DataStream", overlap_s: int = 120) -> "DataStream":
         )
 
 
+class ExternalDataStream(DataModel):
+    """A simplified data stream for acquisitions where instrument metadata is unavailable."""
+
+    object_type: Literal["ExternalDataStream"] = "ExternalDataStream"
+    stream_start_time: Annotated[
+        AwareDatetimeWithDefault,
+        Field(..., title="Stream start time"),
+        TimeValidation.BETWEEN,
+    ]
+    stream_end_time: Annotated[
+        AwareDatetimeWithDefault,
+        Field(..., title="Stream stop time"),
+        TimeValidation.BETWEEN,
+    ]
+    modalities: List[Modality.ONE_OF] = Field(
+        ..., title="Modalities", description="Modalities that are acquired in this stream"
+    )
+    notes: Optional[str] = Field(default=None, title="Notes")
+
+
 class StimulusEpoch(DataModel):
     """All stimuli being presented to the subject. starting and stopping at approximately the
     same time. Not all acquisitions have StimulusEpochs.
@@ -374,7 +395,11 @@ def coerce_fixed_offset_tz_string(cls, v):
     )
     protocol_id: Optional[List[str]] = Field(default=None, title="Protocol ID", description="DOI for protocols.io")
     ethics_review_id: Optional[List[str]] = Field(default=None, title="Ethics review ID")
-    instrument_id: str = Field(..., title="Instrument ID", description="Should match the Instrument.instrument_id")
+    instrument_id: Optional[str] = Field(
+        default=None,
+        title="Instrument ID",
+        description="Should match the Instrument.instrument_id. Required when instrument metadata is available.",
+    )
     acquisition_type: str = Field(
         ...,
         title="Acquisition type",
@@ -406,12 +431,13 @@ def coerce_fixed_offset_tz_string(cls, v):
     )
 
     # Acquisition data
-    data_streams: List[DataStream] = Field(
+    data_streams: DiscriminatedList[DataStream | ExternalDataStream] = Field(
         ...,
         title="Data streams",
         description=(
             "A data stream is a collection of devices that are acquiring data simultaneously. Each acquisition can "
-            "include multiple streams. Streams should be split when configurations are changed."
+            "include multiple streams. Streams should be split when configurations are changed. "
+            "Use ExternalDataStream for acquisitions where instrument metadata is unavailable."
         ),
     )
     stimulus_epochs: List[StimulusEpoch] = Field(
@@ -463,6 +489,16 @@ def check_subject_specimen_id(self):
 
         return self
 
+    @model_validator(mode="after")
+    def instrument_id_required_for_data_streams(self):
+        """Require instrument_id when any standard DataStream is present"""
+        if not hasattr(self, "data_streams"):
+            return self
+        if any(isinstance(stream, DataStream) for stream in self.data_streams):
+            if not self.instrument_id:
+                raise ValueError("instrument_id is required when data_streams contains a DataStream")
+        return self
+
     @model_validator(mode="after")
     def specimen_required(self):
         """Check if specimen ID is required for in vitro imaging modalities"""
@@ -546,7 +582,10 @@ def __add__(self, other: "Acquisition") -> "Acquisition":
         ethics_review_id = merge_optional_list(self.ethics_review_id, other.ethics_review_id)
         calibrations = self.calibrations + other.calibrations
         maintenance = self.maintenance + other.maintenance
-        data_streams = Acquisition._merge_data_streams(self.data_streams + other.data_streams)
+        all_streams = self.data_streams + other.data_streams
+        external_streams = [s for s in all_streams if isinstance(s, ExternalDataStream)]
+        regular_streams = [s for s in all_streams if isinstance(s, DataStream)]
+        data_streams = Acquisition._merge_data_streams(regular_streams) + external_streams
         stimulus_epochs = self.stimulus_epochs + other.stimulus_epochs
 
         # Remove duplicates

src/aind_data_schema/core/metadata.py

@@ -21,7 +21,7 @@
 from aind_data_schema.base import DataCoreModel
 from aind_data_schema.components.identifiers import DatabaseIdentifiers
 from aind_data_schema.components.subjects import CalibrationObject
-from aind_data_schema.core.acquisition import Acquisition
+from aind_data_schema.core.acquisition import Acquisition, DataStream, ExternalDataStream
 from aind_data_schema.core.data_description import DataDescription
 from aind_data_schema.core.instrument import Instrument
 from aind_data_schema.core.model import Model
@@ -149,9 +149,16 @@ def validate_expected_files_by_modality(self):
 
         for file in REQUIRED_FILE_SETS.keys():
             if getattr(self, file):
-                for file in REQUIRED_FILE_SETS[file]:
-                    if not getattr(self, file):
-                        warnings.warn(f"Metadata missing required file: {file}")
+                for required_file in REQUIRED_FILE_SETS[file]:
+                    if not getattr(self, required_file):
+                        # Skip instrument warning when acquisition only has ExternalDataStream
+                        if (
+                            required_file == "instrument"
+                            and self.acquisition
+                            and all(isinstance(s, ExternalDataStream) for s in self.acquisition.data_streams)
+                        ):
+                            continue
+                        warnings.warn(f"Metadata missing required file: {required_file}")
 
         return self
 
@@ -218,7 +225,8 @@ def validate_acquisition_active_devices(self):
 
         if self.acquisition:
             for data_stream in self.acquisition.data_streams:
-                active_devices.extend(data_stream.active_devices)
+                if isinstance(data_stream, DataStream):
+                    active_devices.extend(data_stream.active_devices)
 
         device_names = []
 
@@ -253,6 +261,8 @@ def validate_acquisition_connections(self):
         if self.acquisition:
             data_streams = self.acquisition.data_streams
             for data_stream in data_streams:
+                if not isinstance(data_stream, DataStream):
+                    continue
                 for connection in data_stream.connections:
                     # Check both source and target devices exist
                     missing_devices = []

tests/test_acquisition.py

@@ -22,8 +22,14 @@
     SampleChamberConfig,
 )
 from aind_data_schema.components.coordinates import CoordinateSystemLibrary, Translation
-from aind_data_schema.core.acquisition import Acquisition, AcquisitionSubjectDetails, DataStream, StimulusEpoch
+from aind_data_schema.core.acquisition import (
+    Acquisition,
+    AcquisitionSubjectDetails,
+    DataStream,
+    StimulusEpoch,
+)
 from aind_data_schema.components.connections import Connection
+from examples.barseq_acquisition import acquisition as barseq_acquisition
 from examples.ephys_acquisition import acquisition as ephys_acquisition
 from examples.exaspim_acquisition import acq as exaspim_acquisition
 from examples.mri_acquisition import acquisition as mri_acquisition, scan1
@@ -53,6 +59,33 @@ def test_constructors(self):
             scan1_dict["notes"] = ""
             MRIScan.model_validate(scan1_dict)
 
+    def test_external_data_stream(self):
+        """Test ExternalDataStream: valid without instrument_id, and DataStream requires instrument_id"""
+        # Happy path: BARseq example uses ExternalDataStream and has no instrument_id
+        self.assertIsNotNone(barseq_acquisition)
+        self.assertIsNone(barseq_acquisition.instrument_id)
+
+        # Guard: DataStream without instrument_id should fail
+        start = datetime(2025, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
+        end = datetime(2025, 1, 2, 0, 0, 0, tzinfo=timezone.utc)
+        with self.assertRaises(ValidationError) as context:
+            Acquisition(
+                subject_id="123456",
+                acquisition_start_time=start,
+                acquisition_end_time=end,
+                acquisition_type="Test",
+                data_streams=[
+                    DataStream(
+                        stream_start_time=start,
+                        stream_end_time=end,
+                        modalities=[Modality.BARSEQ],
+                        active_devices=[],
+                        configurations=[],
+                    )
+                ],
+            )
+        self.assertIn("instrument_id is required", str(context.exception))
+
     def test_check_subject_specimen_id(self):
         """Test that subject and specimen IDs match"""
         with self.assertRaises(ValueError) as context:

tests/test_metadata.py

@@ -29,6 +29,7 @@
 from aind_data_schema.components.subject_procedures import TrainingProtocol
 from aind_data_schema.core.acquisition import StimulusEpoch
 
+from examples.barseq_acquisition import acquisition as barseq_acquisition
 from examples.data_description import d as data_description
 from examples.subject import s as subject
 
@@ -334,6 +335,19 @@ def test_validate_expected_files_by_modality(self):
             str(context.exception),
         )
 
+    def test_external_data_stream_no_instrument_warning(self):
+        """Test that ExternalDataStream-only acquisitions do not warn about missing instrument"""
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
+            Metadata(
+                name="655019_2023-04-03T181709",
+                location="bucket",
+                subject=subject,
+                acquisition=barseq_acquisition,
+            )
+        instrument_warnings = [str(warning.message) for warning in w if "instrument" in str(warning.message)]
+        self.assertEqual([], instrument_warnings)
+
     def test_validate_acquisition_connections(self):
         """Tests that acquisition connections are validated correctly."""
         # Case where all connection devices are present in instrument components