feat: validate datetime objects in the schema (except QC and model) (#1491)

* feat: validate times

* fix: reduce requirement for data_description.name to just the same day

* chore: lint

* chore: lint

* tests: fix variable passing issue

* fix: remove validators in procedures

* chore: lint

* fix: add the additional time validators for other core files

* fix: revert validation on prep_date field

* tests: add full coverage for metadata new time validators

* chore: lint

* fix: remove unnecessary try/except

* chore: lint

* refactor: simplify time validator by pulling function into a util

* chore: lint

Author: dbirman

src/aind_data_schema/components/measurements.py

@@ -1,14 +1,15 @@
 """Calibration data models"""
 
+from typing import Annotated, List, Literal, Optional
 from enum import Enum
-from typing import List, Literal, Optional
 
 from aind_data_schema_models.units import UNITS, PowerUnit, TimeUnit, VolumeUnit, VoltageUnit
 from pydantic import model_validator
 
 from aind_data_schema.base import AwareDatetimeWithDefault, DataModel, Discriminated, Field, GenericModel
 from aind_data_schema.components.configs import DeviceConfig
 from aind_data_schema.components.reagent import Reagent
+from aind_data_schema.utils.validators import TimeValidation
 
 
 class FitType(Enum):
@@ -50,7 +51,9 @@ def validate_fit_type(cls, values):
 class Calibration(DeviceConfig):
     """Generic calibration class"""
 
-    calibration_date: AwareDatetimeWithDefault = Field(..., title="Date and time of calibration")
+    calibration_date: Annotated[AwareDatetimeWithDefault, TimeValidation.BEFORE] = Field(
+        ..., title="Date and time of calibration"
+    )
     description: str = Field(..., title="Description", description="Brief description of what is being calibrated")
     input: List[float | str] = Field(..., description="Calibration input", title="Inputs")
     input_unit: UNITS = Field(..., title="Input unit")
@@ -111,7 +114,9 @@ class PowerCalibration(Calibration):
 class Maintenance(DeviceConfig):
     """Generic maintenance class"""
 
-    maintenance_date: AwareDatetimeWithDefault = Field(..., title="Date and time of maintenance")
+    maintenance_date: Annotated[AwareDatetimeWithDefault, TimeValidation.BEFORE] = Field(
+        ..., title="Date and time of maintenance"
+    )
     description: str = Field(..., title="Description", description="Description on maintenance procedure")
     protocol_id: Optional[str] = Field(default=None, title="Protocol ID")
 

src/aind_data_schema/components/subjects.py

@@ -3,7 +3,7 @@
 from datetime import date as date_type
 from datetime import time
 from enum import Enum
-from typing import List, Optional
+from typing import Annotated, List, Optional
 
 from aind_data_schema_models.organizations import Organization
 from aind_data_schema_models.pid_names import PIDName
@@ -12,6 +12,7 @@
 from pydantic_core.core_schema import ValidationInfo
 
 from aind_data_schema.base import DataModel
+from aind_data_schema.utils.validators import TimeValidation
 
 
 class Sex(str, Enum):
@@ -81,7 +82,7 @@ class MouseSubject(DataModel):
     """Description of a mouse subject"""
 
     sex: Sex = Field(..., title="Sex")
-    date_of_birth: date_type = Field(..., title="Date of birth")
+    date_of_birth: Annotated[date_type, TimeValidation.BEFORE] = Field(..., title="Date of birth")
     strain: Strain.ONE_OF = Field(..., title="Strain")
     species: Species.ONE_OF = Field(..., title="Species")
     alleles: List[PIDName] = Field(default=[], title="Alleles", description="Allele names and persistent IDs")

src/aind_data_schema/core/acquisition.py

@@ -1,11 +1,12 @@
 """Schema describing data acquisition metadata and configurations"""
 
 from decimal import Decimal
-from typing import List, Literal, Optional
+from typing import Annotated, List, Literal, Optional
 
 from aind_data_schema_models.modalities import Modality
 from aind_data_schema_models.stimulus_modality import StimulusModality
 from aind_data_schema_models.units import MassUnit, VolumeUnit
+from aind_data_schema.utils.validators import TimeValidation
 from pydantic import Field, SkipValidation, model_validator
 
 from aind_data_schema.base import AwareDatetimeWithDefault, DataCoreModel, DataModel, DiscriminatedList, GenericModel
@@ -87,8 +88,16 @@ class DataStream(DataModel):
     same time.
     """
 
-    stream_start_time: AwareDatetimeWithDefault = Field(..., title="Stream start time")
-    stream_end_time: AwareDatetimeWithDefault = Field(..., title="Stream stop time")
+    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"
     )
@@ -165,12 +174,12 @@ class StimulusEpoch(DataModel):
     same time. Not all acquisitions have StimulusEpochs.
     """
 
-    stimulus_start_time: AwareDatetimeWithDefault = Field(
+    stimulus_start_time: Annotated[AwareDatetimeWithDefault, TimeValidation.BETWEEN] = Field(
         ...,
         title="Stimulus start time",
         description="When a specific stimulus begins. This might be the same as the acquisition start time.",
     )
-    stimulus_end_time: AwareDatetimeWithDefault = Field(
+    stimulus_end_time: Annotated[AwareDatetimeWithDefault, TimeValidation.BETWEEN] = Field(
         ...,
         title="Stimulus end time",
         description="When a specific stimulus ends. This might be the same as the acquisition end time.",

src/aind_data_schema/core/metadata.py

@@ -30,6 +30,7 @@
 from aind_data_schema.core.quality_control import QualityControl
 from aind_data_schema.core.subject import Subject
 from aind_data_schema.utils.compatibility_check import InstrumentAcquisitionCompatibility
+from aind_data_schema.utils.validators import recursive_time_validation_check, validate_creation_time_after_midnight
 
 CORE_FILES = [
     "subject",
@@ -269,6 +270,79 @@ def validate_training_protocol_references(self):
 
         return self
 
+    @model_validator(mode="after")
+    def validate_time_constraints(self):
+        """Validate that all fields with TimeValidation annotations respect acquisition time bounds
+        (if acquisition is present)"""
+        if self.acquisition:
+            acquisition_start_time = None
+            acquisition_end_time = None
+            if hasattr(self.acquisition, "acquisition_start_time") and hasattr(
+                self.acquisition, "acquisition_end_time"
+            ):
+                acquisition_start_time = self.acquisition.acquisition_start_time
+                acquisition_end_time = self.acquisition.acquisition_end_time
+
+            recursive_time_validation_check(
+                self.acquisition,
+                acquisition_start_time=acquisition_start_time,
+                acquisition_end_time=acquisition_end_time,
+            )
+
+            if self.processing:
+                recursive_time_validation_check(
+                    self.processing,
+                    acquisition_start_time=acquisition_start_time,
+                    acquisition_end_time=acquisition_end_time,
+                )
+            if self.subject:
+                recursive_time_validation_check(
+                    self.subject,
+                    acquisition_start_time=acquisition_start_time,
+                    acquisition_end_time=acquisition_end_time,
+                )
+            if self.instrument:
+                recursive_time_validation_check(
+                    self.instrument,
+                    acquisition_start_time=acquisition_start_time,
+                    acquisition_end_time=acquisition_end_time,
+                )
+            if self.procedures:
+                recursive_time_validation_check(
+                    self.procedures,
+                    acquisition_start_time=acquisition_start_time,
+                    acquisition_end_time=acquisition_end_time,
+                )
+
+        return self
+
+    @model_validator(mode="after")
+    def validate_data_description_name_time_consistency(self):
+        """Validate that the creation_time from data_description.name is on or after midnight
+        on the same day as acquisition.acquisition_end_time"""
+        if self.data_description and self.acquisition:
+            if (
+                self.data_description.name
+                and hasattr(self.acquisition, "acquisition_end_time")
+                and self.acquisition.acquisition_end_time is not None
+            ):
+                # Parse the name to extract creation_time
+                parsed_name = DataDescription.parse_name(self.data_description.name, self.data_description.data_level)
+                name_creation_time = parsed_name.get("creation_time")
+
+                if name_creation_time:
+                    try:
+                        validate_creation_time_after_midnight(name_creation_time, self.acquisition.acquisition_end_time)
+                    except ValueError as e:
+                        # Re-raise with more specific context for data_description.name
+                        raise ValueError(
+                            f"Creation time from data_description.name ({name_creation_time}) "
+                            f"must be on or after midnight of the acquisition day "
+                            f"({self.acquisition.acquisition_end_time.date()})"
+                        ) from e
+
+        return self
+
 
 def create_metadata_json(
     name: str,

src/aind_data_schema/core/processing.py

@@ -3,7 +3,7 @@
 import re
 import warnings
 from enum import Enum
-from typing import Dict, List, Literal, Optional
+from typing import Annotated, Dict, List, Literal, Optional
 
 from aind_data_schema_models.process_names import ProcessName
 from aind_data_schema_models.units import MemoryUnit, UnitlessUnit
@@ -13,6 +13,7 @@
 from aind_data_schema.components.identifiers import Code, Person
 from aind_data_schema.components.wrappers import AssetPath
 from aind_data_schema.utils.merge import merge_notes, merge_optional_list
+from aind_data_schema.utils.validators import TimeValidation
 
 
 class ProcessStage(str, Enum):
@@ -63,8 +64,8 @@ class DataProcess(DataModel):
     pipeline_name: Optional[str] = Field(
         default=None, title="Pipeline name", description="Pipeline names must exist in Processing.pipelines"
     )
-    start_date_time: AwareDatetimeWithDefault = Field(..., title="Start date time")
-    end_date_time: AwareDatetimeWithDefault = Field(..., title="End date time")
+    start_date_time: Annotated[AwareDatetimeWithDefault, TimeValidation.AFTER] = Field(..., title="Start date time")
+    end_date_time: Annotated[AwareDatetimeWithDefault, TimeValidation.AFTER] = Field(..., title="End date time")
     output_path: Optional[AssetPath] = Field(
         default=None, title="Output path", description="Path to processing outputs, if stored."
     )

src/aind_data_schema/utils/validators.py

@@ -1,16 +1,28 @@
 """ Validator utility functions """
 
 import logging
+from datetime import date, datetime
 from enum import Enum
 from pathlib import Path
-from typing import Any, List, Optional
+from typing import Any, List, Optional, Union
 
 from aind_data_schema.components.wrappers import AssetPath
 
 # Fields that should have the same length as the coordinate system axes
 AXIS_TYPES = ["Translation", "Rotation", "Scale"]
 
 
+class TimeValidation(Enum):
+    """Enum for time validation types."""
+
+    BETWEEN = "between"
+    """Time should be between start and end."""
+    AFTER = "after"
+    """Time should be after the start time."""
+    BEFORE = "before"
+    """Time should be before the end time."""
+
+
 class CoordinateSystemException(Exception):
     """Raised when a coordinate system is missing."""
 
@@ -44,6 +56,83 @@ def subject_specimen_id_compatibility(subject_id: str, specimen_id: str) -> bool
     return subject_id in specimen_id
 
 
+def recursive_time_validation_check(data, acquisition_start_time=None, acquisition_end_time=None):
+    """Recursively check fields for TimeValidation annotations and validate against acquisition times.
+
+    Parameters
+    ----------
+    data : Any
+        The data structure to check recursively
+    acquisition_start_time : Optional[datetime]
+        The acquisition start time to validate against
+    acquisition_end_time : Optional[datetime]
+        The acquisition end time to validate against
+    """
+    if not data:
+        return
+
+    # Check if this object has fields with TimeValidation annotations
+    if hasattr(data, "__annotations__") and hasattr(data, "__dict__"):
+        for field_name, field_value in data.__dict__.items():
+            if field_name in getattr(data, "__annotations__", {}):
+                # Check if the field has TimeValidation annotation
+                annotation = data.__annotations__[field_name]
+                if hasattr(annotation, "__metadata__"):
+                    for metadata in annotation.__metadata__:
+                        if isinstance(metadata, TimeValidation):
+                            # Validate the field value against the time constraint
+                            if field_value and acquisition_start_time and acquisition_end_time:
+                                _validate_time_constraint(
+                                    field_value, metadata, acquisition_start_time, acquisition_end_time, field_name
+                                )
+
+    # Recursively check nested structures
+    _time_validation_recurse_helper(data, acquisition_start_time, acquisition_end_time)
+
+
+def _validate_time_constraint(field_value, time_validation, start_time, end_time, field_name):
+    """Validate a single time field against the specified constraint."""
+
+    # Handle conversion between date and datetime objects for comparison
+    def convert_to_comparable(value, reference_datetime):
+        """Convert date to datetime using the timezone from reference, or return as-is if already datetime"""
+        if isinstance(value, date) and not isinstance(value, datetime):
+            # Convert date to datetime at midnight with same timezone as reference
+            return datetime.combine(value, datetime.min.time()).replace(tzinfo=reference_datetime.tzinfo)
+        return value
+
+    # Convert field_value to be comparable with start_time and end_time
+    comparable_field_value = convert_to_comparable(field_value, start_time)
+
+    if time_validation == TimeValidation.BETWEEN:
+        if not (start_time <= comparable_field_value <= end_time):
+            raise ValueError(
+                f"Field '{field_name}' with value {field_value} must be between {start_time} and {end_time}"
+            )
+    elif time_validation == TimeValidation.AFTER:
+        if comparable_field_value <= start_time:
+            raise ValueError(f"Field '{field_name}' with value {field_value} must be after {start_time}")
+    elif time_validation == TimeValidation.BEFORE:
+        if comparable_field_value >= end_time:
+            raise ValueError(f"Field '{field_name}' with value {field_value} must be before {end_time}")
+
+
+def _time_validation_recurse_helper(data, acquisition_start_time, acquisition_end_time):
+    """Helper function for recursive_time_validation_check: recurse calls for lists and objects only"""
+    if isinstance(data, list):
+        for item in data:
+            recursive_time_validation_check(item, acquisition_start_time, acquisition_end_time)
+        return
+    elif hasattr(data, "__dict__"):
+        for attr_name, attr_value in data.__dict__.items():
+            if attr_name == "object_type":
+                continue  # skip object_type
+            if callable(attr_value):
+                continue  # skip methods
+
+            recursive_time_validation_check(attr_value, acquisition_start_time, acquisition_end_time)
+
+
 def _recurse_helper(data, **kwargs):
     """Helper function for recursive_axis_order_check: recurse calls for lists and objects only"""
     if isinstance(data, list):
@@ -156,3 +245,46 @@ def recursive_check_paths(obj: Any, directory: Optional[Path] = None):
     elif hasattr(obj, "__dict__"):
         for value in vars(obj).values():
             recursive_check_paths(value, directory)
+
+
+def validate_creation_time_after_midnight(
+    creation_time: Optional[Union[datetime, date]], reference_time: Optional[datetime]
+) -> None:
+    """Validate that creation_time is on or after midnight of the reference_time's day.
+
+    Parameters
+    ----------
+    creation_time : Optional[datetime]
+        The creation time to validate (datetime or date objects are supported)
+    reference_time : Optional[datetime]
+        The reference time to compare against (typically acquisition_end_time)
+
+    Raises
+    ------
+    ValueError
+        If creation_time is before midnight of the reference_time's day
+    """
+    if not creation_time or not reference_time:
+        return
+
+    # Convert date to datetime if needed
+    if isinstance(creation_time, date) and not isinstance(creation_time, datetime):
+        creation_time = datetime.combine(creation_time, datetime.min.time())
+
+    # If creation_time is timezone-naive (local time),
+    # add the same timezone as reference_time
+    if isinstance(creation_time, datetime) and creation_time.tzinfo is None and reference_time.tzinfo is not None:
+        creation_time = creation_time.replace(tzinfo=reference_time.tzinfo)
+
+    # Get midnight of the reference time day
+    reference_date = reference_time.date()
+    midnight_of_reference_day = datetime.combine(reference_date, datetime.min.time()).replace(
+        tzinfo=reference_time.tzinfo
+    )
+
+    # Validate that creation_time is on or after midnight of the reference day
+    if isinstance(creation_time, datetime) and creation_time < midnight_of_reference_day:
+        raise ValueError(
+            f"Creation time ({creation_time}) "
+            f"must be on or after midnight of the reference day ({midnight_of_reference_day})"
+        )