|
1 | 1 | """ Validator utility functions """ |
2 | 2 |
|
3 | 3 | import logging |
| 4 | +from datetime import date, datetime |
4 | 5 | from enum import Enum |
5 | 6 | from pathlib import Path |
6 | | -from typing import Any, List, Optional |
| 7 | +from typing import Any, List, Optional, Union |
7 | 8 |
|
8 | 9 | from aind_data_schema.components.wrappers import AssetPath |
9 | 10 |
|
10 | 11 | # Fields that should have the same length as the coordinate system axes |
11 | 12 | AXIS_TYPES = ["Translation", "Rotation", "Scale"] |
12 | 13 |
|
13 | 14 |
|
| 15 | +class TimeValidation(Enum): |
| 16 | + """Enum for time validation types.""" |
| 17 | + |
| 18 | + BETWEEN = "between" |
| 19 | + """Time should be between start and end.""" |
| 20 | + AFTER = "after" |
| 21 | + """Time should be after the start time.""" |
| 22 | + BEFORE = "before" |
| 23 | + """Time should be before the end time.""" |
| 24 | + |
| 25 | + |
14 | 26 | class CoordinateSystemException(Exception): |
15 | 27 | """Raised when a coordinate system is missing.""" |
16 | 28 |
|
@@ -44,6 +56,83 @@ def subject_specimen_id_compatibility(subject_id: str, specimen_id: str) -> bool |
44 | 56 | return subject_id in specimen_id |
45 | 57 |
|
46 | 58 |
|
| 59 | +def recursive_time_validation_check(data, acquisition_start_time=None, acquisition_end_time=None): |
| 60 | + """Recursively check fields for TimeValidation annotations and validate against acquisition times. |
| 61 | +
|
| 62 | + Parameters |
| 63 | + ---------- |
| 64 | + data : Any |
| 65 | + The data structure to check recursively |
| 66 | + acquisition_start_time : Optional[datetime] |
| 67 | + The acquisition start time to validate against |
| 68 | + acquisition_end_time : Optional[datetime] |
| 69 | + The acquisition end time to validate against |
| 70 | + """ |
| 71 | + if not data: |
| 72 | + return |
| 73 | + |
| 74 | + # Check if this object has fields with TimeValidation annotations |
| 75 | + if hasattr(data, "__annotations__") and hasattr(data, "__dict__"): |
| 76 | + for field_name, field_value in data.__dict__.items(): |
| 77 | + if field_name in getattr(data, "__annotations__", {}): |
| 78 | + # Check if the field has TimeValidation annotation |
| 79 | + annotation = data.__annotations__[field_name] |
| 80 | + if hasattr(annotation, "__metadata__"): |
| 81 | + for metadata in annotation.__metadata__: |
| 82 | + if isinstance(metadata, TimeValidation): |
| 83 | + # Validate the field value against the time constraint |
| 84 | + if field_value and acquisition_start_time and acquisition_end_time: |
| 85 | + _validate_time_constraint( |
| 86 | + field_value, metadata, acquisition_start_time, acquisition_end_time, field_name |
| 87 | + ) |
| 88 | + |
| 89 | + # Recursively check nested structures |
| 90 | + _time_validation_recurse_helper(data, acquisition_start_time, acquisition_end_time) |
| 91 | + |
| 92 | + |
| 93 | +def _validate_time_constraint(field_value, time_validation, start_time, end_time, field_name): |
| 94 | + """Validate a single time field against the specified constraint.""" |
| 95 | + |
| 96 | + # Handle conversion between date and datetime objects for comparison |
| 97 | + def convert_to_comparable(value, reference_datetime): |
| 98 | + """Convert date to datetime using the timezone from reference, or return as-is if already datetime""" |
| 99 | + if isinstance(value, date) and not isinstance(value, datetime): |
| 100 | + # Convert date to datetime at midnight with same timezone as reference |
| 101 | + return datetime.combine(value, datetime.min.time()).replace(tzinfo=reference_datetime.tzinfo) |
| 102 | + return value |
| 103 | + |
| 104 | + # Convert field_value to be comparable with start_time and end_time |
| 105 | + comparable_field_value = convert_to_comparable(field_value, start_time) |
| 106 | + |
| 107 | + if time_validation == TimeValidation.BETWEEN: |
| 108 | + if not (start_time <= comparable_field_value <= end_time): |
| 109 | + raise ValueError( |
| 110 | + f"Field '{field_name}' with value {field_value} must be between {start_time} and {end_time}" |
| 111 | + ) |
| 112 | + elif time_validation == TimeValidation.AFTER: |
| 113 | + if comparable_field_value <= start_time: |
| 114 | + raise ValueError(f"Field '{field_name}' with value {field_value} must be after {start_time}") |
| 115 | + elif time_validation == TimeValidation.BEFORE: |
| 116 | + if comparable_field_value >= end_time: |
| 117 | + raise ValueError(f"Field '{field_name}' with value {field_value} must be before {end_time}") |
| 118 | + |
| 119 | + |
| 120 | +def _time_validation_recurse_helper(data, acquisition_start_time, acquisition_end_time): |
| 121 | + """Helper function for recursive_time_validation_check: recurse calls for lists and objects only""" |
| 122 | + if isinstance(data, list): |
| 123 | + for item in data: |
| 124 | + recursive_time_validation_check(item, acquisition_start_time, acquisition_end_time) |
| 125 | + return |
| 126 | + elif hasattr(data, "__dict__"): |
| 127 | + for attr_name, attr_value in data.__dict__.items(): |
| 128 | + if attr_name == "object_type": |
| 129 | + continue # skip object_type |
| 130 | + if callable(attr_value): |
| 131 | + continue # skip methods |
| 132 | + |
| 133 | + recursive_time_validation_check(attr_value, acquisition_start_time, acquisition_end_time) |
| 134 | + |
| 135 | + |
47 | 136 | def _recurse_helper(data, **kwargs): |
48 | 137 | """Helper function for recursive_axis_order_check: recurse calls for lists and objects only""" |
49 | 138 | if isinstance(data, list): |
@@ -156,3 +245,46 @@ def recursive_check_paths(obj: Any, directory: Optional[Path] = None): |
156 | 245 | elif hasattr(obj, "__dict__"): |
157 | 246 | for value in vars(obj).values(): |
158 | 247 | recursive_check_paths(value, directory) |
| 248 | + |
| 249 | + |
| 250 | +def validate_creation_time_after_midnight( |
| 251 | + creation_time: Optional[Union[datetime, date]], reference_time: Optional[datetime] |
| 252 | +) -> None: |
| 253 | + """Validate that creation_time is on or after midnight of the reference_time's day. |
| 254 | +
|
| 255 | + Parameters |
| 256 | + ---------- |
| 257 | + creation_time : Optional[datetime] |
| 258 | + The creation time to validate (datetime or date objects are supported) |
| 259 | + reference_time : Optional[datetime] |
| 260 | + The reference time to compare against (typically acquisition_end_time) |
| 261 | +
|
| 262 | + Raises |
| 263 | + ------ |
| 264 | + ValueError |
| 265 | + If creation_time is before midnight of the reference_time's day |
| 266 | + """ |
| 267 | + if not creation_time or not reference_time: |
| 268 | + return |
| 269 | + |
| 270 | + # Convert date to datetime if needed |
| 271 | + if isinstance(creation_time, date) and not isinstance(creation_time, datetime): |
| 272 | + creation_time = datetime.combine(creation_time, datetime.min.time()) |
| 273 | + |
| 274 | + # If creation_time is timezone-naive (local time), |
| 275 | + # add the same timezone as reference_time |
| 276 | + if isinstance(creation_time, datetime) and creation_time.tzinfo is None and reference_time.tzinfo is not None: |
| 277 | + creation_time = creation_time.replace(tzinfo=reference_time.tzinfo) |
| 278 | + |
| 279 | + # Get midnight of the reference time day |
| 280 | + reference_date = reference_time.date() |
| 281 | + midnight_of_reference_day = datetime.combine(reference_date, datetime.min.time()).replace( |
| 282 | + tzinfo=reference_time.tzinfo |
| 283 | + ) |
| 284 | + |
| 285 | + # Validate that creation_time is on or after midnight of the reference day |
| 286 | + if isinstance(creation_time, datetime) and creation_time < midnight_of_reference_day: |
| 287 | + raise ValueError( |
| 288 | + f"Creation time ({creation_time}) " |
| 289 | + f"must be on or after midnight of the reference day ({midnight_of_reference_day})" |
| 290 | + ) |
0 commit comments