[mock_uss] Change scdsc database to use FlightInfo (#277)

Change mock_uss scdsc database to use FlightInfo

monitoring/mock_uss/scdsc/database.py

@@ -1,6 +1,7 @@
 import json
 from typing import Dict, Optional
 
+from monitoring.monitorlib.clients.flight_planning.flight_info import FlightInfo
 from monitoring.monitorlib.multiprocessing import SynchronizedValue
 from uas_standards.interuss.automated_testing.scd.v1 import api as scd_injection_api
 from implicitdict import ImplicitDict
@@ -13,9 +14,9 @@
 class FlightRecord(ImplicitDict):
     """Representation of a flight in a USS"""
 
-    op_intent_injection: scd_injection_api.OperationalIntentTestInjection
-    flight_authorisation: scd_injection_api.FlightAuthorisationData
-    op_intent_reference: OperationalIntentReference
+    flight_info: FlightInfo
+    op_intent: OperationalIntent
+
     locked: bool = False
 
 

monitoring/mock_uss/scdsc/flight_planning.py

@@ -115,7 +115,7 @@ def check_for_disallowed_conflicts(
     for op_intent in op_intents:
         if (
             existing_flight
-            and existing_flight.op_intent_reference.id == op_intent.reference.id
+            and existing_flight.op_intent.reference.id == op_intent.reference.id
         ):
             log(
                 f"intersection with {op_intent.reference.id} not considered: intersection with a past version of this flight"
@@ -143,14 +143,14 @@ def check_for_disallowed_conflicts(
 
         modifying_activated = (
             existing_flight
-            and existing_flight.op_intent_reference.state
+            and existing_flight.op_intent.reference.state
             == scd_api.OperationalIntentState.Activated
             and req_body.operational_intent.state
             == scd_api.OperationalIntentState.Activated
         )
         if modifying_activated:
             preexisting_conflict = Volume4DCollection.from_interuss_scd_api(
-                existing_flight.op_intent_injection.volumes
+                existing_flight.op_intent.details.volumes
             ).intersects_vol4s(v2)
             if preexisting_conflict:
                 log(

monitoring/mock_uss/scdsc/routes_injection.py

@@ -10,12 +10,14 @@
 from loguru import logger
 import requests.exceptions
 
+from monitoring.monitorlib.clients.flight_planning.flight_info import FlightInfo
 from uas_standards.astm.f3548.v21 import api
 from uas_standards.astm.f3548.v21.api import (
     OperationalIntent,
     PutOperationalIntentDetailsParameters,
     ImplicitSubscriptionParameters,
     PutOperationalIntentReferenceParameters,
+    OperationalIntentDetails,
 )
 from uas_standards.interuss.automated_testing.scd.v1.api import (
     InjectFlightRequest,
@@ -79,7 +81,7 @@ def query_operational_intents(
     )
     tx = db.value
     get_details_for = []
-    own_flights = {f.op_intent_reference.id: f for f in tx.flights.values() if f}
+    own_flights = {f.op_intent.reference.id: f for f in tx.flights.values() if f}
     result = []
     for op_intent_ref in op_intent_refs:
         if op_intent_ref.id in own_flights:
@@ -214,7 +216,7 @@ def log(msg: str):
     try:
         # Check the transition is valid
         state_transition_from = (
-            OperationalIntentState(existing_flight.op_intent_reference.state)
+            OperationalIntentState(existing_flight.op_intent.reference.state)
             if existing_flight
             else None
         )
@@ -280,13 +282,13 @@ def log(msg: str):
             new_subscription=ImplicitSubscriptionParameters(uss_base_url=base_url),
         )
         if existing_flight:
-            id = existing_flight.op_intent_reference.id
+            id = existing_flight.op_intent.reference.id
             step_name = f"updating existing operational intent {id} in DSS"
             log(step_name)
             result = scd_client.update_operational_intent_reference(
                 utm_client,
                 id,
-                existing_flight.op_intent_reference.ovn,
+                existing_flight.op_intent.reference.ovn,
                 req,
             )
         else:
@@ -320,10 +322,17 @@ def log(msg: str):
         # Store flight in database
         step_name = "storing flight in database"
         log("Storing flight in database")
+        op_intent = OperationalIntent(
+            reference=result.operational_intent_reference,
+            details=OperationalIntentDetails(
+                volumes=req_body.operational_intent.volumes,
+                off_nominal_volumes=req_body.operational_intent.off_nominal_volumes,
+                priority=req_body.operational_intent.priority,
+            ),
+        )
         record = database.FlightRecord(
-            op_intent_reference=result.operational_intent_reference,
-            op_intent_injection=req_body.operational_intent,
-            flight_authorisation=req_body.flight_authorisation,
+            op_intent=op_intent,
+            flight_info=FlightInfo.from_scd_inject_flight_request(req_body),
         )
         with db as tx:
             tx.flights[flight_id] = record
@@ -422,12 +431,12 @@ def delete_flight(flight_id) -> Tuple[dict, int]:
     # Delete operational intent from DSS
     step_name = "performing unknown operation"
     try:
-        step_name = f"deleting operational intent {flight.op_intent_reference.id} with OVN {flight.op_intent_reference.ovn} from DSS"
+        step_name = f"deleting operational intent {flight.op_intent.reference.id} with OVN {flight.op_intent.reference.ovn} from DSS"
         logger.debug(f"[delete_flight/{pid}:{flight_id}] {step_name}")
         result = scd_client.delete_operational_intent_reference(
             utm_client,
-            flight.op_intent_reference.id,
-            flight.op_intent_reference.ovn,
+            flight.op_intent.reference.id,
+            flight.op_intent.reference.ovn,
         )
 
         step_name = "notifying subscribers"
@@ -558,7 +567,7 @@ def make_result(success: bool, msg: str) -> ClearAreaResponse:
                     if record is None or record.locked:
                         pending_flights.add(flight_id)
                         continue
-                    if record.op_intent_reference.id in deleted:
+                    if record.op_intent.reference.id in deleted:
                         flights_to_delete.append(flight_id)
                 for flight_id in flights_to_delete:
                     del tx.flights[flight_id]

monitoring/mock_uss/scdsc/routes_scdsc.py

@@ -21,7 +21,7 @@ def scdsc_get_operational_intent_details(entityid: str):
     tx = db.value
     flight = None
     for f in tx.flights.values():
-        if f.op_intent_reference.id == entityid:
+        if f.op_intent.reference.id == entityid:
             flight = f
             break
 
@@ -47,11 +47,11 @@ def scdsc_get_operational_intent_details(entityid: str):
 
 def op_intent_from_flightrecord(flight: FlightRecord) -> OperationalIntent:
     return OperationalIntent(
-        reference=flight.op_intent_reference,
+        reference=flight.op_intent.reference,
         details=OperationalIntentDetails(
-            volumes=flight.op_intent_injection.volumes,
-            off_nominal_volumes=flight.op_intent_injection.off_nominal_volumes,
-            priority=flight.op_intent_injection.priority,
+            volumes=flight.op_intent.details.volumes,
+            off_nominal_volumes=flight.op_intent.details.off_nominal_volumes,
+            priority=flight.op_intent.details.priority,
         ),
     )
 

monitoring/monitorlib/clients/flight_planning/flight_info.py

@@ -1,12 +1,13 @@
+from __future__ import annotations
 from enum import Enum
 from typing import Optional, List
 
 from implicitdict import ImplicitDict
 from uas_standards.ansi_cta_2063_a import SerialNumber
 from uas_standards.en4709_02 import OperatorRegistrationNumber
+from uas_standards.interuss.automated_testing.scd.v1 import api as scd_api
 
-from monitoring.monitorlib.geotemporal import Volume4D
-
+from monitoring.monitorlib.geotemporal import Volume4D, Volume4DCollection
 
 # ===== ASTM F3548-21 =====
 
@@ -224,6 +225,57 @@ class FlightInfo(ImplicitDict):
     additional_information: Optional[dict]
     """Any information relevant to a particular jurisdiction or use case not described in the standard schema. The keys and values must be agreed upon between the test designers and USSs under test."""
 
+    @staticmethod
+    def from_scd_inject_flight_request(
+        request: scd_api.InjectFlightRequest,
+    ) -> FlightInfo:
+        usage_states = {
+            scd_api.OperationalIntentState.Accepted: AirspaceUsageState.Planned,
+            scd_api.OperationalIntentState.Activated: AirspaceUsageState.InUse,
+            scd_api.OperationalIntentState.Nonconforming: AirspaceUsageState.InUse,
+            scd_api.OperationalIntentState.Contingent: AirspaceUsageState.InUse,
+        }
+        uas_states = {
+            scd_api.OperationalIntentState.Accepted: UasState.Nominal,
+            scd_api.OperationalIntentState.Activated: UasState.Nominal,
+            scd_api.OperationalIntentState.Nonconforming: UasState.OffNominal,
+            scd_api.OperationalIntentState.Contingent: UasState.Contingent,
+        }
+        if (
+            request.operational_intent.state
+            in (
+                scd_api.OperationalIntentState.Accepted,
+                scd_api.OperationalIntentState.Activated,
+            )
+            and request.operational_intent.off_nominal_volumes
+        ):
+            # This invalid request can no longer be represented with a standard flight planning request
+            raise ValueError(
+                f"Request for nominal {request.operational_intent.state} operational intent is invalid because it contains off-nominal volumes"
+            )
+        v4c = Volume4DCollection.from_interuss_scd_api(
+            request.operational_intent.volumes
+        ) + Volume4DCollection.from_interuss_scd_api(
+            request.operational_intent.off_nominal_volumes
+        )
+        basic_information = BasicFlightPlanInformation(
+            usage_state=usage_states[request.operational_intent.state],
+            uas_state=uas_states[request.operational_intent.state],
+            area=v4c.volumes,
+        )
+        astm_f3548v21 = ASTMF354821OpIntentInformation(
+            priority=request.operational_intent.priority
+        )
+        uspace_flight_authorisation = ImplicitDict.parse(
+            request.flight_authorisation, FlightAuthorisationData
+        )
+        flight_info = FlightInfo(
+            basic_information=basic_information,
+            astm_f3548_21=astm_f3548v21,
+            uspace_flight_authorisation=uspace_flight_authorisation,
+        )
+        return flight_info
+
 
 class ExecutionStyle(str, Enum):
     Hypothetical = "Hypothetical"