[uss_qualifier] Establish and use TestTimeContext (#1265)

Author: BenjaminPelletier

.basedpyright/baseline.json

@@ -17344,14 +17344,6 @@
                     "endColumn": 98,
                     "lineCount": 1
                 }
-            },
-            {
-                "code": "reportAttributeAccessIssue",
-                "range": {
-                    "startColumn": 104,
-                    "endColumn": 110,
-                    "lineCount": 1
-                }
             }
         ],
         "./monitoring/uss_qualifier/scenarios/astm/utm/dss/report.py": [

monitoring/monitorlib/clients/flight_planning/flight_info_template.py

@@ -14,7 +14,7 @@
     Volume4DCollection,
     Volume4DTemplateCollection,
 )
-from monitoring.monitorlib.temporal import Time, TimeDuringTest
+from monitoring.monitorlib.temporal import TestTimeContext
 from monitoring.monitorlib.transformations import Transformation
 
 
@@ -30,9 +30,9 @@ class BasicFlightPlanInformationTemplate(ImplicitDict):
     area: Volume4DTemplateCollection
     """User intends to or may fly anywhere in this entire area."""
 
-    def resolve(self, times: dict[TimeDuringTest, Time]) -> BasicFlightPlanInformation:
+    def resolve(self, context: TestTimeContext) -> BasicFlightPlanInformation:
         kwargs = {k: v for k, v in self.items()}
-        kwargs["area"] = Volume4DCollection([t.resolve(times) for t in self.area])
+        kwargs["area"] = Volume4DCollection([t.resolve(context) for t in self.area])
         return ImplicitDict.parse(kwargs, BasicFlightPlanInformation)
 
 
@@ -53,21 +53,21 @@ class FlightInfoTemplate(ImplicitDict):
     transformations: list[Transformation] | None
     """If specified, transform this flight according to these transformations in order (after all templates are resolved)."""
 
-    def resolve(self, times: dict[TimeDuringTest, Time]) -> FlightInfo:
+    def resolve(self, context: TestTimeContext) -> FlightInfo:
         kwargs = {k: v for k, v in self.items() if k not in {"transformations"}}
-        basic_info = self.basic_information.resolve(times)
+        basic_info = self.basic_information.resolve(context)
         if "transformations" in self and self.transformations:
             for xform in self.transformations:
                 basic_info.area = [v.transform(xform) for v in basic_info.area]
         kwargs["basic_information"] = basic_info
         return ImplicitDict.parse(kwargs, FlightInfo)
 
     def to_scd_inject_request(
-        self, times: dict[TimeDuringTest, Time]
+        self, context: TestTimeContext
     ) -> scd_api.InjectFlightRequest:
         """Render a legacy SCD injection API request object from this object."""
 
-        info = self.resolve(times)
+        info = self.resolve(context)
         if "astm_f3548_21" not in info or not info.astm_f3548_21:
             raise ValueError(
                 "Legacy SCD injection API requires astm_f3548_21 operational intent priority to be specified in FlightInfo"

monitoring/monitorlib/geotemporal.py

@@ -14,7 +14,11 @@
 
 from monitoring.monitorlib import geo
 from monitoring.monitorlib.geo import Altitude, Circle, LatLngPoint, Polygon, Volume3D
-from monitoring.monitorlib.temporal import TestTime, Time, TimeDuringTest
+from monitoring.monitorlib.temporal import (
+    TestTime,
+    TestTimeContext,
+    Time,
+)
 from monitoring.monitorlib.transformations import Transformation
 
 
@@ -64,16 +68,16 @@ def resolve_3d(self) -> Volume3D:
         return result
 
     def resolve_times(
-        self, times: dict[TimeDuringTest, Time]
+        self, context: TestTimeContext
     ) -> tuple[Time | None, Time | None]:
         """Resolve Volume4DTemplate into concrete temporal bounds (start, end)"""
         if self.start_time is not None:
-            time_start = self.start_time.resolve(times)
+            time_start = self.start_time.resolve(context)
         else:
             time_start = None
 
         if self.end_time is not None:
-            time_end = self.end_time.resolve(times)
+            time_end = self.end_time.resolve(context)
         else:
             time_end = None
 
@@ -93,14 +97,14 @@ def resolve_times(
 
         return time_start, time_end
 
-    def resolve(self, times: dict[TimeDuringTest, Time]) -> Volume4D:
+    def resolve(self, context: TestTimeContext) -> Volume4D:
         """Resolve Volume4DTemplate into concrete Volume4D."""
         volume = self.resolve_3d()
 
         # Make 4D volume
         kwargs = {"volume": volume}
 
-        time_start, time_end = self.resolve_times(times)
+        time_start, time_end = self.resolve_times(context)
 
         if time_start is not None:
             kwargs["time_start"] = time_start

monitoring/monitorlib/temporal.py

@@ -62,16 +62,71 @@ class NextDay(ImplicitDict):
     """Acceptable days of the week.  Omit to indicate that any day of the week is acceptable."""
 
 
-class TimeDuringTest(str, Enum):
-    StartOfTestRun = "StartOfTestRun"
+class TimeDuringTest(str):
+    """A particular time during the test, as identified/named by this value.
+
+    Names listed below are provided by the framework when appropriate:
+        * StartOfTestRun: The time at which the test run started.
+        * StartOfScenario: The time at which the current scenario started.
+        * TimeOfEvaluation: The time at which a TestTime was resolved to an absolute time; generally close to 'now'.
+    """
+
+    StartOfTestRun: TimeDuringTest
     """The time at which the test run started."""
 
-    StartOfScenario = "StartOfScenario"
+    StartOfScenario: TimeDuringTest
     """The time at which the current scenario started."""
 
-    TimeOfEvaluation = "TimeOfEvaluation"
+    TimeOfEvaluation: TimeDuringTest
     """The time at which a TestTime was resolved to an absolute time; generally close to 'now'."""
 
+    ProvidedByFramework: list[TimeDuringTest]
+    """The TimeDuringTests provided by the framework when appropriate"""
+
+
+TimeDuringTest.StartOfTestRun = TimeDuringTest("StartOfTestRun")
+TimeDuringTest.StartOfScenario = TimeDuringTest("StartOfScenario")
+TimeDuringTest.TimeOfEvaluation = TimeDuringTest("TimeOfEvaluation")
+TimeDuringTest.ProvidedByFramework = [
+    TimeDuringTest.StartOfTestRun,
+    TimeDuringTest.StartOfScenario,
+    TimeDuringTest.TimeOfEvaluation,
+]
+
+
+class Time(StringBasedDateTime):
+    def offset(self, dt: timedelta) -> Time:
+        return Time(self.datetime + dt)
+
+    def to_f3548v21(self) -> f3548v21.Time:
+        return f3548v21.Time(value=self)
+
+
+class TestTimeContext(dict[TimeDuringTest, Time]):
+    """Context in which TestTimes are evaluated.
+
+    Stores definitions for TimeDuringTests."""
+
+    def evaluate_now(self) -> TestTimeContext:
+        """Set TimeOfEvaluation in this context to the current time.
+
+        This should be performed once before resolving a TestTime."""
+        self[TimeDuringTest.TimeOfEvaluation] = Time(arrow.utcnow().datetime)
+        return self
+
+    @staticmethod
+    def all_times_are(t: Time) -> TestTimeContext:
+        """Returns a TestTimeContext where all framework-provided times are the provided time.
+
+        For the purpose of testing/validation."""
+        return TestTimeContext(
+            {
+                TimeDuringTest.StartOfTestRun: t,
+                TimeDuringTest.StartOfScenario: t,
+                TimeDuringTest.TimeOfEvaluation: t,
+            }
+        )
+
 
 class TestTime(ImplicitDict):
     """Exactly one of the time option fields of this object must be specified."""
@@ -85,6 +140,9 @@ class TestTime(ImplicitDict):
     time_during_test: TimeDuringTest | None = None
     """Time option field to, if specified, use a timestamp relating to the current test run."""
 
+    name: TimeDuringTest | None
+    """If specified, update the TestTimeContext with the time computed for this TestTime as this name, which may then later be referenced by a different TestTime via time_during_test."""
+
     next_day: NextDay | None = None
     """Time option field to use a timestamp equal to midnight beginning the next occurrence of any matching day following the specified reference timestamp."""
 
@@ -101,20 +159,20 @@ class TestTime(ImplicitDict):
       * "-08:00" (ISO time zone)
       * "US/Pacific" (IANA time zone)"""
 
-    def resolve(self, times: dict[TimeDuringTest, Time]) -> Time:
+    def resolve(self, context: TestTimeContext) -> Time:
         """Resolve TestTime into specific Time."""
         result = None
         if self.absolute_time is not None:
             result = self.absolute_time.datetime
         elif self.time_during_test is not None:
-            if self.time_during_test not in times:
+            if self.time_during_test not in context:
                 raise ValueError(
-                    f"Specified {self.time_during_test} time during test was not provided when resolving TestTime"
+                    f"Specified '{self.time_during_test}' time during test was not provided when resolving TestTime"
                 )
-            result = times[self.time_during_test].datetime
+            result = context[self.time_during_test].datetime
         elif self.next_day is not None:
             t0 = (
-                arrow.get(self.next_day.starting_from.resolve(times).datetime)
+                arrow.get(self.next_day.starting_from.resolve(context).datetime)
                 .to(self.next_day.time_zone)
                 .datetime
             )
@@ -130,11 +188,11 @@ def resolve(self, times: dict[TimeDuringTest, Time]) -> Time:
             result = t
         elif self.offset_from is not None:
             result = (
-                self.offset_from.starting_from.resolve(times).datetime
+                self.offset_from.starting_from.resolve(context).datetime
                 + self.offset_from.offset.timedelta
             )
         elif self.next_sun_position is not None:
-            t0 = self.next_sun_position.starting_from.resolve(times).datetime
+            t0 = self.next_sun_position.starting_from.resolve(context).datetime
 
             dt = timedelta(minutes=5)
             lat = self.next_sun_position.observed_from.lat
@@ -186,7 +244,12 @@ def resolve(self, times: dict[TimeDuringTest, Time]) -> Time:
         if self.use_timezone:
             result = arrow.get(result).to(self.use_timezone).datetime
 
-        return Time(result)
+        t_result = Time(result)
+
+        if "name" in self and self.name:
+            context[self.name] = t_result
+
+        return t_result
 
 
 _weekdays = [
@@ -212,11 +275,3 @@ def _sun_elevation(t: datetime, lat_deg: float, lng_deg: float) -> float:
     Returns: Degrees above the horizon of the center of the sun.
     """
     return get_solarposition(t, lat_deg, lng_deg).elevation.values[0]  # pyright:ignore[reportAttributeAccessIssue]
-
-
-class Time(StringBasedDateTime):
-    def offset(self, dt: timedelta) -> Time:
-        return Time(self.datetime + dt)
-
-    def to_f3548v21(self) -> f3548v21.Time:
-        return f3548v21.Time(value=self)

monitoring/uss_qualifier/resources/flight_planning/flight_intent_validation.py

@@ -14,7 +14,7 @@
     FlightInfoTemplate,
 )
 from monitoring.monitorlib.geotemporal import Volume4D, Volume4DCollection
-from monitoring.monitorlib.temporal import Time, TimeDuringTest
+from monitoring.monitorlib.temporal import TestTimeContext, Time, TimeDuringTest
 from monitoring.monitorlib.uspace import problems_with_flight_authorisation
 from monitoring.uss_qualifier.resources.flight_planning.flight_intent import (
     FlightIntentID,
@@ -53,23 +53,16 @@ def validate_flight_intent_templates(
     extents = Volume4DCollection([])
 
     now = Time(arrow.utcnow().datetime)
-    times = {
-        TimeDuringTest.StartOfTestRun: now,
-        TimeDuringTest.StartOfScenario: now,
-        TimeDuringTest.TimeOfEvaluation: now,
-    }
-    flight_intents = {k: v.resolve(times) for k, v in templates.items()}
+    context = TestTimeContext.all_times_are(now)
+    flight_intents = {k: v.resolve(context) for k, v in templates.items()}
     for flight_intent in flight_intents.values():
         extents.extend(flight_intent.basic_information.area)
     validate_flight_intents(flight_intents, expected_intents, now)
 
     later = Time(now.datetime + MAX_TEST_RUN_DURATION)
-    times = {
-        TimeDuringTest.StartOfTestRun: now,
-        TimeDuringTest.StartOfScenario: later,
-        TimeDuringTest.TimeOfEvaluation: later,
-    }
-    flight_intents = {k: v.resolve(times) for k, v in templates.items()}
+    context = TestTimeContext.all_times_are(later)
+    context[TimeDuringTest.StartOfTestRun] = now
+    flight_intents = {k: v.resolve(context) for k, v in templates.items()}
     for flight_intent in flight_intents.values():
         extents.extend(flight_intent.basic_information.area)
     validate_flight_intents(flight_intents, expected_intents, later)

monitoring/uss_qualifier/resources/netrid/service_area.py

@@ -4,8 +4,7 @@
 import arrow
 from implicitdict import ImplicitDict
 
-from monitoring.monitorlib.geotemporal import Volume4D
-from monitoring.monitorlib.temporal import Time, TimeDuringTest
+from monitoring.monitorlib.temporal import TestTimeContext, Time
 from monitoring.uss_qualifier.resources import VolumeResource
 from monitoring.uss_qualifier.resources.resource import Resource
 
@@ -35,11 +34,7 @@ def __init__(
 
         now = Time(arrow.utcnow().datetime)
         resolved_for_tests = self._volume.specification.template.resolve(
-            {
-                TimeDuringTest.StartOfTestRun: now,
-                TimeDuringTest.StartOfScenario: now,
-                TimeDuringTest.TimeOfEvaluation: now,
-            }
+            TestTimeContext.all_times_are(now)
         )
 
         if (
@@ -55,10 +50,6 @@ def __init__(
                 f"In order to be usable for a ServiceAreaResource, the provided VolumeResource must declare time bounds. The volume template was obtained from: {resource_origin}"
             )
 
-    def resolved_volume4d(self, times: dict[TimeDuringTest, Time]) -> Volume4D:
-        times[TimeDuringTest.TimeOfEvaluation] = Time(arrow.utcnow().datetime)
-        return self._volume.specification.template.resolve(times)
-
     def s2_vertices(self):
         return self._volume.specification.s2_vertices()
 
@@ -85,9 +76,11 @@ def altitude_max(self) -> float:
         return v3d.altitude_upper.to_w84_m()
 
     def resolved_time_bounds(
-        self, times: dict[TimeDuringTest, Time]
+        self, context: TestTimeContext
     ) -> tuple[datetime.datetime, datetime.datetime]:
-        time_start, time_end = self._volume.specification.template.resolve_times(times)
+        time_start, time_end = self._volume.specification.template.resolve_times(
+            context
+        )
         if time_start is None or time_end is None:
             # Note this should not happen as we check at construction time that these bounds exist
             raise ValueError("The underlying volume does not have time bounds")

monitoring/uss_qualifier/resources/planning_area.py

@@ -14,7 +14,7 @@
 from monitoring.monitorlib.geo import make_latlng_rect
 from monitoring.monitorlib.geotemporal import Volume4D
 from monitoring.monitorlib.subscription_params import SubscriptionParams
-from monitoring.monitorlib.temporal import Time, TimeDuringTest
+from monitoring.monitorlib.temporal import TestTimeContext, Time
 from monitoring.monitorlib.testing import make_fake_url
 from monitoring.uss_qualifier.resources.resource import Resource
 from monitoring.uss_qualifier.resources.volume import VolumeResource
@@ -56,8 +56,8 @@ def __init__(
         self.specification = specification
         self._volume = volume
 
-    def resolved_volume4d(self, times: dict[TimeDuringTest, Time]) -> Volume4D:
-        return self._volume.specification.template.resolve(times)
+    def resolved_volume4d(self, context: TestTimeContext) -> Volume4D:
+        return self._volume.specification.template.resolve(context)
 
     def resolved_volume4d_with_times(
         self, time_start: datetime.datetime | None, time_end: datetime.datetime | None