[monitorlib] Add generic flight_planning client

monitoring/mock_uss/scdsc/flight_planning.py

@@ -9,6 +9,7 @@
 from monitoring.monitorlib.geotemporal import Volume4DCollection
 from monitoring.monitorlib.locality import Locality
 from monitoring.monitorlib.uspace import problems_with_flight_authorisation
+from uas_standards.interuss.automated_testing.scd.v1.api import OperationalIntentState
 
 
 class PlanningError(Exception):
@@ -47,6 +48,7 @@ def validate_request(req_body: scd_api.InjectFlightRequest, locality: Locality)
     # Validate max planning horizon for creation
     start_time = Volume4DCollection.from_interuss_scd_api(
         req_body.operational_intent.volumes
+        + req_body.operational_intent.off_nominal_volumes
     ).time_start.datetime
     time_delta = start_time - datetime.now(tz=start_time.tzinfo)
     if (
@@ -87,6 +89,13 @@ def check_for_disallowed_conflicts(
     if log is None:
         log = lambda msg: None
 
+    if req_body.operational_intent.state not in (
+        OperationalIntentState.Accepted,
+        OperationalIntentState.Activated,
+    ):
+        # No conflicts are disallowed if the flight is not nominal
+        return
+
     v1 = Volume4DCollection.from_interuss_scd_api(req_body.operational_intent.volumes)
 
     for op_intent in op_intents:

monitoring/mock_uss/scdsc/routes_injection.py

@@ -244,9 +244,11 @@ def log(msg: str):
         # Check for operational intents in the DSS
         step_name = "querying for operational intents"
         log("Obtaining latest operational intent information")
-        vol4 = Volume4DCollection.from_interuss_scd_api(
+        v1 = Volume4DCollection.from_interuss_scd_api(
             req_body.operational_intent.volumes
-        ).bounding_volume.to_f3548v21()
+            + req_body.operational_intent.off_nominal_volumes
+        )
+        vol4 = v1.bounding_volume.to_f3548v21()
         op_intents = query_operational_intents(vol4)
 
         # Check for intersections

monitoring/monitorlib/clients/flight_planning/client.py

@@ -0,0 +1,94 @@
+from abc import ABC, abstractmethod
+from typing import List, Optional, Union
+
+from monitoring.monitorlib.clients.flight_planning.test_preparation import (
+    TestPreparationActivityResponse,
+)
+
+from monitoring.monitorlib.clients.flight_planning.flight_info import (
+    FlightInfo,
+    FlightID,
+    ExecutionStyle,
+)
+from monitoring.monitorlib.clients.flight_planning.planning import (
+    PlanningActivityResponse,
+)
+from monitoring.monitorlib.fetch import Query
+from monitoring.monitorlib.geotemporal import Volume4D
+
+
+class PlanningActivityError(Exception):
+    queries: List[Query]
+
+    def __init__(
+        self, message: str, queries: Optional[Union[Query, List[Query]]] = None
+    ):
+        super(PlanningActivityError, self).__init__(message)
+        if queries is None:
+            self.queries = []
+        elif isinstance(queries, Query):
+            self.queries = [queries]
+        else:
+            self.queries = queries
+
+
+class FlightPlannerClient(ABC):
+    """Client to interact with a USS as a user performing flight planning activities and as the test director preparing for tests involving flight planning activities."""
+
+    # ===== Emulation of user actions =====
+
+    @abstractmethod
+    def try_plan_flight(
+        self, flight_info: FlightInfo, execution_style: ExecutionStyle
+    ) -> PlanningActivityResponse:
+        """Instruct the USS to emulate a normal user trying to plan the described flight.
+
+        Raises:
+            * PlanningActivityError
+        """
+        raise NotImplementedError()
+
+    @abstractmethod
+    def try_update_flight(
+        self,
+        flight_id: FlightID,
+        updated_flight_info: FlightInfo,
+        execution_style: ExecutionStyle,
+    ) -> PlanningActivityResponse:
+        """Instruct the USS to emulate a normal user trying to update the specified flight as described.
+
+        Raises:
+            * PlanningActivityError
+        """
+        raise NotImplementedError()
+
+    @abstractmethod
+    def try_end_flight(
+        self, flight_id: FlightID, execution_style: ExecutionStyle
+    ) -> PlanningActivityResponse:
+        """Instruct the USS to emulate a normal user trying to end the specified flight.
+
+        Raises:
+            * PlanningActivityError
+        """
+        raise NotImplementedError()
+
+    # ===== Test preparation activities =====
+
+    @abstractmethod
+    def report_readiness(self) -> TestPreparationActivityResponse:
+        """Acting as test director, ask the USS about its readiness to use its flight planning interface for automated testing.
+
+        Raises:
+            * PlanningActivityError
+        """
+        raise NotImplementedError()
+
+    @abstractmethod
+    def clear_area(self, area: Volume4D) -> TestPreparationActivityResponse:
+        """Acting as test director, instruct the USS to close/end/remove all flights it manages within the specified area.
+
+        Raises:
+            * PlanningActivityError
+        """
+        raise NotImplementedError()