[uss_qualifier] Use FlightInfoTemplate for flight intent storage

monitoring/monitorlib/clients/flight_planning/flight_info_template.py

@@ -13,6 +13,7 @@
     FlightInfo,
 )
 from monitoring.monitorlib.geotemporal import Volume4DTemplate, resolve_volume4d
+from uas_standards.interuss.automated_testing.scd.v1 import api as scd_api
 
 
 class BasicFlightPlanInformationTemplate(ImplicitDict):
@@ -51,3 +52,58 @@ def resolve(self, start_of_test: datetime) -> FlightInfo:
         kwargs = {k: v for k, v in self.items()}
         kwargs["basic_information"] = self.basic_information.resolve(start_of_test)
         return ImplicitDict.parse(kwargs, FlightInfo)
+
+    def scd_inject_request(
+        self, start_of_test: datetime
+    ) -> scd_api.InjectFlightRequest:
+        """Render a legacy SCD injection API request object from this object."""
+
+        info = self.resolve(start_of_test)
+        if "astm_f3548_21" not in info or not info.astm_f3548_21:
+            raise ValueError(
+                f"Legacy SCD injection API requires astm_f3548_21 operational intent priority to be specified in FlightInfo"
+            )
+        if (
+            "uspace_flight_authorisation" not in info
+            or not info.uspace_flight_authorisation
+        ):
+            raise ValueError(
+                f"Legacy SCD injection API requires uspace_flight_authorisation to be specified in FlightInfo"
+            )
+        volumes = [v.to_interuss_scd_api() for v in info.basic_information.area]
+        if info.basic_information.usage_state == AirspaceUsageState.Planned:
+            state = scd_api.OperationalIntentState.Accepted
+            off_nominal_volumes = []
+        elif info.basic_information.usage_state == AirspaceUsageState.InUse:
+            if info.basic_information.uas_state == UasState.Nominal:
+                state = scd_api.OperationalIntentState.Activated
+                off_nominal_volumes = []
+            elif info.basic_information.uas_state == UasState.OffNominal:
+                state = scd_api.OperationalIntentState.Nonconforming
+                off_nominal_volumes = volumes
+                volumes = []
+            elif info.basic_information.uas_state == UasState.Contingent:
+                state = scd_api.OperationalIntentState.Contingent
+                off_nominal_volumes = volumes
+                volumes = []
+            else:
+                raise ValueError(
+                    f"Unrecongized uas_state '{info.basic_information.uas_state}'"
+                )
+        else:
+            raise ValueError(
+                f"Unrecognized usage_state '{info.basic_information.usage_state}'"
+            )
+        operational_intent = scd_api.OperationalIntentTestInjection(
+            state=state,
+            priority=scd_api.Priority(info.astm_f3548_21.priority),
+            volumes=volumes,
+            off_nominal_volumes=off_nominal_volumes,
+        )
+        flight_authorisation = ImplicitDict.parse(
+            info.uspace_flight_authorisation, scd_api.FlightAuthorisationData
+        )
+        return scd_api.InjectFlightRequest(
+            operational_intent=operational_intent,
+            flight_authorisation=flight_authorisation,
+        )

monitoring/monitorlib/geotemporal.py

@@ -10,12 +10,11 @@
 from implicitdict import ImplicitDict, StringBasedTimeDelta, StringBasedDateTime
 from pvlib.solarposition import get_solarposition
 import s2sphere as s2sphere
-from uas_standards.astm.f3411.v22a.api import Polygon
 from uas_standards.astm.f3548.v21 import api as f3548v21
 from uas_standards.interuss.automated_testing.scd.v1 import api as interuss_scd_api
 
 from monitoring.monitorlib import geo
-from monitoring.monitorlib.geo import LatLngPoint, Circle, Altitude, Volume3D
+from monitoring.monitorlib.geo import LatLngPoint, Circle, Altitude, Volume3D, Polygon
 
 
 class OffsetTime(ImplicitDict):

monitoring/monitorlib/uspace.py

@@ -1,24 +1,28 @@
 from typing import List
 from urllib.parse import urlparse
 
-from uas_standards.interuss.automated_testing.scd.v1 import api as scd_injection_api
+from monitoring.monitorlib.clients.flight_planning.flight_info import (
+    FlightAuthorisationData,
+    UASClass,
+    FlightAuthorisationDataOperationCategory,
+)
 from uas_standards.ansi_cta_2063_a import SerialNumber
 from uas_standards.en4709_02 import OperatorRegistrationNumber
 
 
 def problems_with_flight_authorisation(
-    flight_auth: scd_injection_api.FlightAuthorisationData,
+    flight_auth: FlightAuthorisationData,
 ) -> List[str]:
     problems: List[str] = []
     if not SerialNumber(flight_auth.uas_serial_number).valid:
         problems.append("Invalid serial number")
     if not OperatorRegistrationNumber(flight_auth.operator_id).valid:
         problems.append("Invalid operator ID")
-    if flight_auth.uas_class == scd_injection_api.UASClass.Other:
+    if flight_auth.uas_class == UASClass.Other:
         problems.append("Invalid UAS class")
     if (
         flight_auth.operation_category
-        == scd_injection_api.FlightAuthorisationDataOperationCategory.Unknown
+        == FlightAuthorisationDataOperationCategory.Unknown
     ):
         problems.append("Invalid operation category")
     if (

monitoring/uss_qualifier/configurations/dev/f3548_self_contained.yaml

@@ -53,7 +53,7 @@ v1:
           specification:
             planning_time: '0:05:00'
             file:
-              path: file://./test_data/che/flight_intents/conflicting_flights.json
+              path: file://./test_data/che/flight_intents/conflicting_flights.yaml
 
         # Details of priority-preemption flights (used in nominal planning priority scenario)
         priority_preemption_flights:

monitoring/uss_qualifier/configurations/dev/library/resources.yaml

@@ -117,7 +117,7 @@ che_invalid_flight_auth_flights:
   specification:
     planning_time: '0:05:00'
     file:
-      path: file://./test_data/che/flight_intents/invalid_flight_auths.json
+      path: file://./test_data/che/flight_intents/invalid_flight_auths.yaml
 
 che_conflicting_flights:
   # Includes flight intents for both equal-priority-not-permitted and higher-priority
@@ -126,9 +126,9 @@ che_conflicting_flights:
   specification:
     planning_time: '0:05:00'
     file:
-      path: file://./test_data/che/flight_intents/conflicting_flights.json
+      path: file://./test_data/che/flight_intents/conflicting_flights.yaml
       # Note that this hash_sha512 field can be safely deleted if the content changes
-      hash_sha512: 6467e88fb69702216dabe1f9723b68fe68a1a9d4b1a40e3af5c0b1ed233061a088c9c30be8a0ada447b18f2c6354db52d84aa12c1e7dfd13f99262d91ba173f4
+      hash_sha512: 86b7172de3a6029efdb5c39dff00f96578f81d25b65b4a0a9d731a42a1e260ef632a0891b744b6d1f62fd7bec61cdad3cb8e6b6811a16cbc17ec2dbd081cbbf6
 
 che_invalid_flight_intents:
   $content_schema: monitoring/uss_qualifier/resources/definitions/ResourceDeclaration.json
@@ -138,30 +138,6 @@ che_invalid_flight_intents:
     file:
       path: test_data.che.flight_intents.invalid_flight_intents
 
-kentland_conflicting_flights:
-  $content_schema: monitoring/uss_qualifier/resources/definitions/ResourceDeclaration.json
-  resource_type: resources.flight_planning.FlightIntentsResource
-  specification:
-    planning_time: '0:05:00'
-    file:
-      path: file://./test_data/usa/kentland/flight_intents/conflicting_flights.yaml
-
-kentland_priority_preemption_flights:
-  $content_schema: monitoring/uss_qualifier/resources/definitions/ResourceDeclaration.json
-  resource_type: resources.flight_planning.FlightIntentsResource
-  specification:
-    planning_time: '0:05:00'
-    file:
-      path: test_data.usa.kentland.flight_intents.priority_preemption
-
-kentland_invalid_flight_intents:
-  $content_schema: monitoring/uss_qualifier/resources/definitions/ResourceDeclaration.json
-  resource_type: resources.flight_planning.FlightIntentsResource
-  specification:
-    planning_time: '0:05:00'
-    file:
-      path: test_data.usa.kentland.flight_intents.invalid_flight_intents
-
 # ===== General flight authorization =====
 
 example_flight_check_table:

monitoring/uss_qualifier/resources/flight_planning/flight_intent.py

@@ -1,24 +1,40 @@
+from __future__ import annotations
+
+import json
 from typing import Optional, Dict
 
-from implicitdict import ImplicitDict, StringBasedDateTime, StringBasedTimeDelta
+import arrow
+
+from implicitdict import ImplicitDict, StringBasedDateTime
+from monitoring.monitorlib.clients.flight_planning.flight_info_template import (
+    FlightInfoTemplate,
+)
 
-from monitoring.uss_qualifier.fileio import FileReference
 from monitoring.uss_qualifier.resources.files import ExternalFile
+from monitoring.uss_qualifier.resources.overrides import apply_overrides
 from uas_standards.interuss.automated_testing.scd.v1.api import InjectFlightRequest
 
 
 class FlightIntent(ImplicitDict):
+    """DEPRECATED.  Use FlightInfoTemplate instead."""
+
     reference_time: StringBasedDateTime
     """The time that all other times in the FlightInjectionAttempt are relative to. If this FlightInjectionAttempt is initiated by uss_qualifier at t_test, then each t_volume_original timestamp within test_injection should be adjusted to t_volume_adjusted such that t_volume_adjusted = t_test + planning_time when t_volume_original = reference_time"""
 
     request: InjectFlightRequest
     """Definition of the flight the user wants to create."""
 
+    @staticmethod
+    def from_flight_info_template(info_template: FlightInfoTemplate) -> FlightIntent:
+        t = arrow.utcnow().datetime
+        request = info_template.scd_inject_request(t)
+        return FlightIntent(reference_time=StringBasedDateTime(t), request=request)
+
 
 FlightIntentID = str
-"""Identifier for a flight intent within a collection of flight intents.
+"""Identifier for a flight planning intent within a collection of flight planning intents.
 
-To be used only within uss_qualifier (not visible to participants under test) to select an appropriate flight intent from the collection."""
+To be used only within uss_qualifier (not visible to participants under test) to select an appropriate flight planning intent from the collection."""
 
 
 class DeltaFlightIntent(ImplicitDict):
@@ -28,29 +44,71 @@ class DeltaFlightIntent(ImplicitDict):
     """Base the flight intent for this element of a FlightIntentCollection on the element of the collection identified by this field."""
 
     mutation: Optional[dict]
-    """For each subfield specified in this object, override the value in the corresponding subfield of the flight intent for this element with the specified value."""
+    """For each leaf subfield specified in this object, override the value in the corresponding subfield of the flight intent for this element with the specified value.
+
+    Consider subfields prefixed with + as leaf subfields."""
 
 
 class FlightIntentCollectionElement(ImplicitDict):
     """Definition of a single flight intent within a FlightIntentCollection.  Exactly one field must be specified."""
 
-    full: Optional[FlightIntent]
-    """If specified, the full definition of the flight intent."""
+    full: Optional[FlightInfoTemplate]
+    """If specified, the full definition of the flight planning intent."""
 
     delta: Optional[DeltaFlightIntent]
-    """If specified, a flight intent based on another flight intent, but with some changes."""
+    """If specified, a flight planning intent based on another flight intent, but with some changes."""
 
 
 class FlightIntentCollection(ImplicitDict):
     """Specification for a collection of flight intents, each identified by a FlightIntentID."""
 
     intents: Dict[FlightIntentID, FlightIntentCollectionElement]
-    """Flights that users want to create."""
+    """Flight planning actions that users want to perform."""
+
+    def resolve(self) -> Dict[FlightIntentID, FlightInfoTemplate]:
+        """Resolve the underlying delta flight intents."""
+
+        # process intents in order of dependency to resolve deltas
+        processed_intents: Dict[FlightIntentID, FlightInfoTemplate] = {}
+        unprocessed_intent_ids = list(self.intents.keys())
+
+        while unprocessed_intent_ids:
+            nb_processed = 0
+            for intent_id in unprocessed_intent_ids:
+                unprocessed_intent = self.intents[intent_id]
+                processed_intent: FlightInfoTemplate
+
+                # copy intent and resolve delta
+                if unprocessed_intent.has_field_with_value("full"):
+                    processed_intent = ImplicitDict.parse(
+                        json.loads(json.dumps(unprocessed_intent.full)),
+                        FlightInfoTemplate,
+                    )
+                elif unprocessed_intent.has_field_with_value("delta"):
+                    if unprocessed_intent.delta.source not in processed_intents:
+                        # delta source has not been processed yet
+                        continue
+
+                    processed_intent = apply_overrides(
+                        processed_intents[unprocessed_intent.delta.source],
+                        unprocessed_intent.delta.mutation,
+                    )
+                else:
+                    raise ValueError(f"{intent_id} is invalid")
+
+                nb_processed += 1
+                processed_intents[intent_id] = processed_intent
+                unprocessed_intent_ids.remove(intent_id)
+
+            if nb_processed == 0 and unprocessed_intent_ids:
+                raise ValueError(
+                    "Unresolvable dependency detected between intents: "
+                    + ", ".join(i_id for i_id in unprocessed_intent_ids)
+                )
+
+        return processed_intents
 
 
 class FlightIntentsSpecification(ImplicitDict):
-    planning_time: StringBasedTimeDelta
-    """Time delta between the time uss_qualifier initiates this FlightInjectionAttempt and when a timestamp within the test_injection equal to reference_time occurs"""
-
     file: ExternalFile
-    """Location of file to load"""
+    """Location of file to load, containing a FlightIntentCollection"""

monitoring/uss_qualifier/resources/flight_planning/flight_intents_resource.py

@@ -1,85 +1,27 @@
-from datetime import timedelta
 import json
 from typing import Dict
 
-import arrow
-from implicitdict import ImplicitDict, StringBasedDateTime
+from implicitdict import ImplicitDict
+from monitoring.monitorlib.clients.flight_planning.flight_info_template import (
+    FlightInfoTemplate,
+)
 
 from monitoring.uss_qualifier.resources.files import load_dict
-from monitoring.uss_qualifier.resources.overrides import apply_overrides
 from monitoring.uss_qualifier.resources.resource import Resource
 from monitoring.uss_qualifier.resources.flight_planning.flight_intent import (
     FlightIntentCollection,
     FlightIntentsSpecification,
-    FlightIntent,
     FlightIntentID,
 )
 
 
 class FlightIntentsResource(Resource[FlightIntentsSpecification]):
-    _planning_time: timedelta
     _intent_collection: FlightIntentCollection
 
     def __init__(self, specification: FlightIntentsSpecification):
         self._intent_collection = ImplicitDict.parse(
             load_dict(specification.file), FlightIntentCollection
         )
-        self._planning_time = specification.planning_time.timedelta
-
-    def get_flight_intents(self) -> Dict[FlightIntentID, FlightIntent]:
-        """Resolve the underlying delta flight intents and shift appropriately times."""
-
-        # process intents in order of dependency to resolve deltas
-        processed_intents: Dict[FlightIntentID, FlightIntent] = {}
-        unprocessed_intent_ids = list(self._intent_collection.intents.keys())
-
-        while unprocessed_intent_ids:
-            nb_processed = 0
-            for intent_id in unprocessed_intent_ids:
-                unprocessed_intent = self._intent_collection.intents[intent_id]
-                processed_intent: FlightIntent
-
-                # copy intent and resolve delta
-                if unprocessed_intent.has_field_with_value("full"):
-                    processed_intent = ImplicitDict.parse(
-                        json.loads(json.dumps(unprocessed_intent.full)), FlightIntent
-                    )
-                elif unprocessed_intent.has_field_with_value("delta"):
-                    if unprocessed_intent.delta.source not in processed_intents:
-                        # delta source has not been processed yet
-                        continue
-
-                    processed_intent = apply_overrides(
-                        processed_intents[unprocessed_intent.delta.source],
-                        unprocessed_intent.delta.mutation,
-                    )
-                else:
-                    raise ValueError(f"{intent_id} is invalid")
-
-                nb_processed += 1
-                processed_intents[intent_id] = processed_intent
-                unprocessed_intent_ids.remove(intent_id)
-
-            if nb_processed == 0 and unprocessed_intent_ids:
-                raise ValueError(
-                    "Unresolvable dependency detected between intents: "
-                    + ", ".join(i_id for i_id in unprocessed_intent_ids)
-                )
-
-        # shift times
-        t0 = arrow.utcnow() + self._planning_time
-
-        for intent_id, intent in processed_intents.items():
-            dt = t0 - intent.reference_time.datetime
-            for volume in (
-                intent.request.operational_intent.volumes
-                + intent.request.operational_intent.off_nominal_volumes
-            ):
-                volume.time_start.value = StringBasedDateTime(
-                    volume.time_start.value.datetime + dt
-                )
-                volume.time_end.value = StringBasedDateTime(
-                    volume.time_end.value.datetime + dt
-                )
 
-        return processed_intents
+    def get_flight_intents(self) -> Dict[FlightIntentID, FlightInfoTemplate]:
+        return self._intent_collection.resolve()