diff --git a/monitoring/monitorlib/clients/flight_planning/__init__.py b/monitoring/monitorlib/clients/flight_planning/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/monitoring/monitorlib/clients/flight_planning/client.py b/monitoring/monitorlib/clients/flight_planning/client.py new file mode 100644 index 0000000000..e64a242efb --- /dev/null +++ b/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() diff --git a/monitoring/monitorlib/clients/flight_planning/client_scd.py b/monitoring/monitorlib/clients/flight_planning/client_scd.py new file mode 100644 index 0000000000..a62569acdd --- /dev/null +++ b/monitoring/monitorlib/clients/flight_planning/client_scd.py @@ -0,0 +1,271 @@ +import uuid +from typing import Dict + +from implicitdict import ImplicitDict +from monitoring.monitorlib.clients.flight_planning.client import ( + FlightPlannerClient, + PlanningActivityError, +) +from monitoring.monitorlib.clients.flight_planning.test_preparation import ( + TestPreparationActivityResponse, +) +from uas_standards.interuss.automated_testing.scd.v1 import api as scd_api +from uas_standards.interuss.automated_testing.scd.v1 import ( + constants as scd_api_constants, +) + +from monitoring.monitorlib.clients.flight_planning.flight_info import ( + FlightInfo, + FlightID, + AirspaceUsageState, + UasState, + ExecutionStyle, +) +from monitoring.monitorlib.clients.flight_planning.planning import ( + PlanningActivityResponse, + PlanningActivityResult, + FlightPlanStatus, +) +from monitoring.monitorlib.fetch import query_and_describe +from monitoring.monitorlib.geotemporal import Volume4D +from monitoring.monitorlib.infrastructure import UTMClientSession + + +class SCDFlightPlannerClient(FlightPlannerClient): + SCD_SCOPE = scd_api_constants.Scope.Inject + _session: UTMClientSession + _plan_statuses: Dict[FlightID, FlightPlanStatus] + + def __init__(self, session: UTMClientSession): + self._session = session + self._plan_statuses = {} + + def _inject( + self, + flight_id: FlightID, + flight_info: FlightInfo, + execution_style: ExecutionStyle, + ) -> PlanningActivityResponse: + if execution_style != ExecutionStyle.IfAllowed: + raise PlanningActivityError( + f"Legacy scd automated testing API only supports {ExecutionStyle.IfAllowed} actions; '{execution_style}' is not supported" + ) + usage_state = flight_info.basic_information.usage_state + uas_state = flight_info.basic_information.uas_state + if uas_state == UasState.Nominal: + if usage_state == AirspaceUsageState.Planned: + state = scd_api.OperationalIntentState.Accepted + elif usage_state == AirspaceUsageState.InUse: + state = scd_api.OperationalIntentState.Activated + else: + raise NotImplementedError( + f"Unsupported operator AirspaceUsageState '{usage_state}' with UasState '{uas_state}'" + ) + volumes = [ + v.to_scd_automated_testing_api() + for v in flight_info.basic_information.area + ] + off_nominal_volumes = [] + elif usage_state == AirspaceUsageState.InUse: + if uas_state == UasState.OffNominal: + state = scd_api.OperationalIntentState.Nonconforming + elif uas_state == UasState.Contingent: + state = scd_api.OperationalIntentState.Contingent + else: + raise NotImplementedError( + f"Unsupported operator UasState '{uas_state}' with AirspaceUsageState '{usage_state}'" + ) + volumes = [] + off_nominal_volumes = [ + v.to_scd_automated_testing_api() + for v in flight_info.basic_information.area + ] + else: + raise NotImplementedError( + f"Unsupported combination of operator AirspaceUsageState '{usage_state}' and UasState '{uas_state}'" + ) + + if "astm_f3548_21" in flight_info and flight_info.astm_f3548_21: + priority = flight_info.astm_f3548_21.priority + else: + priority = 0 + + operational_intent = scd_api.OperationalIntentTestInjection( + state=state, + priority=priority, + volumes=volumes, + off_nominal_volumes=off_nominal_volumes, + ) + + kwargs = {"operational_intent": operational_intent} + if ( + "uspace_flight_authorisation" in flight_info + and flight_info.uspace_flight_authorisation + ): + kwargs["flight_authorisation"] = ImplicitDict.parse( + flight_info.uspace_flight_authorisation, scd_api.FlightAuthorisationData + ) + req = scd_api.InjectFlightRequest(**kwargs) + + op = scd_api.OPERATIONS[scd_api.OperationID.InjectFlight] + url = op.path.format(flight_id=flight_id) + query = query_and_describe( + self._session, op.verb, url, json=req, scope=self.SCD_SCOPE + ) + if query.status_code != 200 and query.status_code != 201: + raise PlanningActivityError( + f"Attempt to plan flight returned status {query.status_code} rather than 200 as expected", + query, + ) + try: + resp: scd_api.InjectFlightResponse = ImplicitDict.parse( + query.response.json, scd_api.InjectFlightResponse + ) + except ValueError as e: + raise PlanningActivityError( + f"Response to plan flight could not be parsed: {str(e)}", query + ) + + old_state = ( + self._plan_statuses[flight_id] + if flight_id in self._plan_statuses + else FlightPlanStatus.NotPlanned + ) + response = PlanningActivityResponse( + flight_id=flight_id, + queries=[query], + activity_result={ + scd_api.InjectFlightResponseResult.Planned: PlanningActivityResult.Completed, + scd_api.InjectFlightResponseResult.ReadyToFly: PlanningActivityResult.Completed, + scd_api.InjectFlightResponseResult.ConflictWithFlight: PlanningActivityResult.Rejected, + scd_api.InjectFlightResponseResult.Rejected: PlanningActivityResult.Rejected, + scd_api.InjectFlightResponseResult.Failed: PlanningActivityResult.Failed, + scd_api.InjectFlightResponseResult.NotSupported: PlanningActivityResult.NotSupported, + }[resp.result], + flight_plan_status={ + scd_api.InjectFlightResponseResult.Planned: FlightPlanStatus.Planned, + scd_api.InjectFlightResponseResult.ReadyToFly: FlightPlanStatus.OkToFly, + scd_api.InjectFlightResponseResult.ConflictWithFlight: old_state, + scd_api.InjectFlightResponseResult.Rejected: old_state, + scd_api.InjectFlightResponseResult.Failed: old_state, + scd_api.InjectFlightResponseResult.NotSupported: old_state, + }[resp.result], + ) + self._plan_statuses[flight_id] = response.flight_plan_status + return response + + def try_plan_flight( + self, flight_info: FlightInfo, execution_style: ExecutionStyle + ) -> PlanningActivityResponse: + return self._inject(str(uuid.uuid4()), flight_info, execution_style) + + def try_update_flight( + self, + flight_id: FlightID, + updated_flight_info: FlightInfo, + execution_style: ExecutionStyle, + ) -> PlanningActivityResponse: + return self._inject(flight_id, updated_flight_info, execution_style) + + def try_end_flight( + self, flight_id: FlightID, execution_style: ExecutionStyle + ) -> PlanningActivityResponse: + op = scd_api.OPERATIONS[scd_api.OperationID.DeleteFlight] + url = op.path.format(flight_id=flight_id) + query = query_and_describe(self._session, op.verb, url, scope=self.SCD_SCOPE) + if query.status_code != 200: + raise PlanningActivityError( + f"Attempt to delete flight returned status {query.status_code} rather than 200 as expected", + query, + ) + try: + resp: scd_api.DeleteFlightResponse = ImplicitDict.parse( + query.response.json, scd_api.DeleteFlightResponse + ) + except ValueError as e: + raise PlanningActivityError( + f"Response to delete flight could not be parsed: {str(e)}", query + ) + + old_state = ( + self._plan_statuses[flight_id] + if flight_id in self._plan_statuses + else FlightPlanStatus.NotPlanned + ) + response = PlanningActivityResponse( + flight_id=flight_id, + queries=[query], + activity_result={ + scd_api.DeleteFlightResponseResult.Closed: PlanningActivityResult.Completed, + scd_api.DeleteFlightResponseResult.Failed: PlanningActivityResult.Failed, + }[resp.result], + flight_plan_status={ + scd_api.DeleteFlightResponseResult.Closed: FlightPlanStatus.Closed, + scd_api.DeleteFlightResponseResult.Failed: old_state, + }[resp.result], + ) + if resp.result == scd_api.DeleteFlightResponseResult.Closed: + del self._plan_statuses[flight_id] + else: + self._plan_statuses[flight_id] = response.flight_plan_status + return response + + def report_readiness(self) -> TestPreparationActivityResponse: + op = scd_api.OPERATIONS[scd_api.OperationID.GetStatus] + query = query_and_describe( + self._session, op.verb, op.path, scope=self.SCD_SCOPE + ) + if query.status_code != 200: + raise PlanningActivityError( + f"Attempt to get interface status returned status {query.status_code} rather than 200 as expected", + query, + ) + try: + resp: scd_api.StatusResponse = ImplicitDict.parse( + query.response.json, scd_api.StatusResponse + ) + except ValueError as e: + raise PlanningActivityError( + f"Response to get interface status could not be parsed: {str(e)}", query + ) + + if resp.status == scd_api.StatusResponseStatus.Ready: + errors = [] + elif resp.status == scd_api.StatusResponseStatus.Starting: + errors = ["SCD flight planning interface is still starting (not ready)"] + else: + errors = [f"Unrecognized status '{resp.status}'"] + + # Note that checking capabilities is not included because the SCD flight planning interface is deprecated and does not warrant full support + + return TestPreparationActivityResponse(errors=errors, queries=[query]) + + def clear_area(self, area: Volume4D) -> TestPreparationActivityResponse: + req = scd_api.ClearAreaRequest( + request_id=str(uuid.uuid4()), extent=area.to_scd_automated_testing_api() + ) + + op = scd_api.OPERATIONS[scd_api.OperationID.ClearArea] + query = query_and_describe( + self._session, op.verb, op.path, json=req, scope=self.SCD_SCOPE + ) + if query.status_code != 200: + raise PlanningActivityError( + f"Attempt to clear area returned status {query.status_code} rather than 200 as expected", + query, + ) + try: + resp: scd_api.ClearAreaResponse = ImplicitDict.parse( + query.response.json, scd_api.ClearAreaResponse + ) + except ValueError as e: + raise PlanningActivityError( + f"Response to clear area could not be parsed: {str(e)}", query + ) + + if resp.outcome.success: + errors = [] + else: + errors = [f"[{resp.outcome.timestamp}]: {resp.outcome.message}"] + + return TestPreparationActivityResponse(errors=errors, queries=[query]) diff --git a/monitoring/monitorlib/clients/flight_planning/flight_info.py b/monitoring/monitorlib/clients/flight_planning/flight_info.py new file mode 100644 index 0000000000..2dfb838bc0 --- /dev/null +++ b/monitoring/monitorlib/clients/flight_planning/flight_info.py @@ -0,0 +1,236 @@ +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 monitoring.monitorlib.geotemporal import Volume4D + + +# ===== ASTM F3548-21 ===== + + +Priority = int +"""Ordinal priority that the flight's operational intent should be assigned, as defined in ASTM F3548-21.""" + + +class ASTMF354821OpIntentInformation(ImplicitDict): + """Information provided about a flight plan that is necessary for ASTM F3548-21.""" + + priority: Optional[Priority] + + +# ===== U-space ===== + + +class FlightAuthorisationDataOperationCategory(str, Enum): + """Category of UAS operation (‘open’, ‘specific’, ‘certified’) as defined in COMMISSION DELEGATED REGULATION (EU) 2019/945. Required by ANNEX IV of COMMISSION IMPLEMENTING REGULATION (EU) 2021/664, paragraph 4.""" + + Unknown = "Unknown" + Open = "Open" + Specific = "Specific" + Certified = "Certified" + + +class OperationMode(str, Enum): + """Specify if the operation is a `VLOS` or `BVLOS` operation. Required by ANNEX IV of COMMISSION IMPLEMENTING REGULATION (EU) 2021/664, paragraph 2.""" + + Undeclared = "Undeclared" + Vlos = "Vlos" + Bvlos = "Bvlos" + + +class UASClass(str, Enum): + """Specify the class of the UAS to be flown, the specifition matches EASA class identification label categories. UAS aircraft class as defined in COMMISSION DELEGATED REGULATION (EU) 2019/945 (C0 to C4) and COMMISSION DELEGATED REGULATION (EU) 2020/1058 (C5 and C6). This field is required by ANNEX IV of COMMISSION IMPLEMENTING REGULATION (EU) 2021/664, paragraph 4.""" + + Other = "Other" + C0 = "C0" + C1 = "C1" + C2 = "C2" + C3 = "C3" + C4 = "C4" + C5 = "C5" + C6 = "C6" + + +class FlightAuthorisationData(ImplicitDict): + """The details of a UAS flight authorization request, as received from the user. + + Note that a full description of a flight authorisation must include mandatory information required by ANNEX IV of COMMISSION IMPLEMENTING REGULATION (EU) 2021/664 for an UAS flight authorisation request. Reference: https://eur-lex.europa.eu/legal-content/EN/TXT/HTML/?uri=CELEX:32021R0664&from=EN#d1e32-178-1 + """ + + uas_serial_number: SerialNumber + """Unique serial number of the unmanned aircraft or, if the unmanned aircraft is privately built, the unique serial number of the add-on. This is expressed in the ANSI/CTA-2063 Physical Serial Number format. Required by ANNEX IV of COMMISSION IMPLEMENTING REGULATION (EU) 2021/664, paragraph 1.""" + + operation_mode: OperationMode + + operation_category: FlightAuthorisationDataOperationCategory + """Category of UAS operation (‘open’, ‘specific’, ‘certified’) as defined in COMMISSION DELEGATED REGULATION (EU) 2019/945. Required by ANNEX IV of COMMISSION IMPLEMENTING REGULATION (EU) 2021/664, paragraph 4.""" + + uas_class: UASClass + + identification_technologies: List[str] + """Technology used to identify the UAS. Required by ANNEX IV of COMMISSION IMPLEMENTING REGULATION (EU) 2021/664, paragraph 6.""" + + uas_type_certificate: Optional[str] + """Provisional field. Not applicable as of September 2021. Required only if `uas_class` is set to `other` by ANNEX IV of COMMISSION IMPLEMENTING REGULATION (EU) 2021/664, paragraph 4.""" + + connectivity_methods: List[str] + """Connectivity methods. Required by ANNEX IV of COMMISSION IMPLEMENTING REGULATION (EU) 2021/664, paragraph 7.""" + + endurance_minutes: int + """Endurance of the UAS. This is expressed in minutes. Required by ANNEX IV of COMMISSION IMPLEMENTING REGULATION (EU) 2021/664, paragraph 8.""" + + emergency_procedure_url: str + """The URL at which the applicable emergency procedure in case of a loss of command and control link may be retrieved. Required by ANNEX IV of COMMISSION IMPLEMENTING REGULATION (EU) 2021/664, paragraph 9.""" + + operator_id: OperatorRegistrationNumber + """Registration number of the UAS operator. + The format is defined in EASA Easy Access Rules for Unmanned Aircraft Systems GM1 to AMC1 + Article 14(6) Registration of UAS operators and ‘certified’ UAS. + Required by ANNEX IV of COMMISSION IMPLEMENTING REGULATION (EU) 2021/664, paragraph 10. + """ + + uas_id: Optional[str] + """When applicable, the registration number of the unmanned aircraft. + This is expressed using the nationality and registration mark of the unmanned aircraft in + line with ICAO Annex 7. + Specified by ANNEX IV of COMMISSION IMPLEMENTING REGULATION (EU) 2021/664, paragraph 10. + """ + + +# ===== RPAS Operating Rules 2.6 ===== + + +class RPAS26FlightDetailsOperatorType(str, Enum): + """The type of operator.""" + + Recreational = "Recreational" + CommercialExcluded = "CommercialExcluded" + ReOC = "ReOC" + + +class RPAS26FlightDetailsAircraftType(str, Enum): + """Type of vehicle being used as per ASTM F3411-22a.""" + + NotDeclared = "NotDeclared" + Aeroplane = "Aeroplane" + Helicopter = "Helicopter" + Gyroplane = "Gyroplane" + HybridLift = "HybridLift" + Ornithopter = "Ornithopter" + Glider = "Glider" + Kite = "Kite" + FreeBalloon = "FreeBalloon" + CaptiveBalloon = "CaptiveBalloon" + Airship = "Airship" + FreeFallOrParachute = "FreeFallOrParachute" + Rocket = "Rocket" + TetheredPoweredAircraft = "TetheredPoweredAircraft" + GroundObstacle = "GroundObstacle" + Other = "Other" + + +class RPAS26FlightDetailsFlightProfile(str, Enum): + """Type of flight profile.""" + + AutomatedGrid = "AutomatedGrid" + AutomatedWaypoint = "AutomatedWaypoint" + Manual = "Manual" + + +class RPAS26FlightDetails(ImplicitDict): + """Information about a flight necessary to plan successfully using the RPAS Platform Operating Rules version 2.6.""" + + operator_type: Optional[RPAS26FlightDetailsOperatorType] + """The type of operator.""" + + uas_serial_numbers: Optional[List[str]] + """The list of UAS/drone serial numbers that will be operated during the operation.""" + + uas_registration_numbers: Optional[List[str]] + """The list of UAS/drone registration numbers that will be operated during the operation.""" + + aircraft_type: Optional[RPAS26FlightDetailsAircraftType] + """Type of vehicle being used as per ASTM F3411-22a.""" + + flight_profile: Optional[RPAS26FlightDetailsFlightProfile] + """Type of flight profile.""" + + pilot_license_number: Optional[str] + """License number for the pilot.""" + + pilot_phone_number: Optional[str] + """Contact phone number for the pilot.""" + + operator_number: Optional[str] + """Operator number.""" + + +# ===== General flight information ===== + + +FlightID = str + + +class AirspaceUsageState(str, Enum): + """User's current usage of the airspace defined in the flight plan.""" + + Planned = "Planned" + """The user intends to fly according to this flight plan, but is not currently using the defined area with an active UAS.""" + + InUse = "InUse" + """The user is currently using the defined area with an active UAS.""" + + +class UasState(str, Enum): + """State of the user's UAS associated with a flight plan.""" + + Nominal = "Nominal" + """The user or UAS reports or implies that it is performing nominally.""" + + OffNominal = "OffNominal" + """The user or UAS reports or implies that it is temporarily not performing nominally, but expects to be able to recover to normal operation.""" + + Contingent = "Contingent" + """The user or UAS reports or implies that it is not performing nominally and may be unable to recover to normal operation.""" + + +class BasicFlightPlanInformation(ImplicitDict): + """Basic information about a flight plan that an operator and/or UAS can be expected to provide in most flight planning scenarios.""" + + usage_state: AirspaceUsageState + """User's current usage of the airspace specified in the flight plan.""" + + uas_state: UasState + """State of the user's UAS associated with this flight plan.""" + + area: List[Volume4D] + """User intends to or may fly anywhere in this entire area.""" + + +class FlightInfo(ImplicitDict): + """Details of user's intent to create or modify a flight plan.""" + + basic_information: BasicFlightPlanInformation + + astm_f3548_21: Optional[ASTMF354821OpIntentInformation] + + uspace_flight_authorisation: Optional[FlightAuthorisationData] + + rpas_operating_rules_2_6: Optional[RPAS26FlightDetails] + + 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.""" + + +class ExecutionStyle(str, Enum): + Hypothetical = "Hypothetical" + """The user does not want the USS to actually perform any action regarding the actual flight plan. Instead, the user would like to know the likely outcome if the action were hypothetically attempted. The response to this request will not refer to an actual flight plan, or an actual state change in an existing flight plan, but rather a hypothetical flight plan or a hypothetical change to an existing flight plan.""" + + IfAllowed = "IfAllowed" + """The user would like to perform the requested action if it is allowed. If the requested action is allowed, the USS should actually perform the action (e.g., actually create a new ASTM F3548-21 operational intent). If the requested action is not allowed, the USS should indicate that the action is Rejected and not perform the action. The response to this request will refer to an actual flight plan when appropriate, and never refer to a hypothetical flight plan or status.""" + + InReality = "InReality" + """The user is communicating an actual state of reality. The USS should consider the user to be actually performing (or attempting to perform) this action, regardless of whether or not the action is allowed under relevant UTM rules.""" diff --git a/monitoring/monitorlib/clients/flight_planning/flight_info_template.py b/monitoring/monitorlib/clients/flight_planning/flight_info_template.py new file mode 100644 index 0000000000..eb8ffd4819 --- /dev/null +++ b/monitoring/monitorlib/clients/flight_planning/flight_info_template.py @@ -0,0 +1,53 @@ +from datetime import datetime +from typing import List, Optional + +from implicitdict import ImplicitDict + +from monitoring.monitorlib.clients.flight_planning.flight_info import ( + AirspaceUsageState, + UasState, + ASTMF354821OpIntentInformation, + FlightAuthorisationData, + RPAS26FlightDetails, + BasicFlightPlanInformation, + FlightInfo, +) +from monitoring.monitorlib.geotemporal import Volume4DTemplate, resolve_volume4d + + +class BasicFlightPlanInformationTemplate(ImplicitDict): + """Template to provide (at runtime) basic information about a flight plan that an operator and/or UAS can be expected to provide in most flight planning scenarios.""" + + usage_state: AirspaceUsageState + """User's current usage of the airspace specified in the flight plan.""" + + uas_state: UasState + """State of the user's UAS associated with this flight plan.""" + + area: List[Volume4DTemplate] + """User intends to or may fly anywhere in this entire area.""" + + def resolve(self, start_of_test: datetime) -> BasicFlightPlanInformation: + kwargs = {k: v for k, v in self.items()} + kwargs["area"] = [resolve_volume4d(t, start_of_test) for t in self.area] + return ImplicitDict.parse(kwargs, BasicFlightPlanInformation) + + +class FlightInfoTemplate(ImplicitDict): + """Template to provide (at runtime) details of user's intent to create or modify a flight plan.""" + + basic_information: BasicFlightPlanInformationTemplate + + astm_f3548_21: Optional[ASTMF354821OpIntentInformation] + + uspace_flight_authorisation: Optional[FlightAuthorisationData] + + rpas_operating_rules_2_6: Optional[RPAS26FlightDetails] + + 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.""" + + 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) diff --git a/monitoring/monitorlib/clients/flight_planning/planning.py b/monitoring/monitorlib/clients/flight_planning/planning.py new file mode 100644 index 0000000000..8f473f8575 --- /dev/null +++ b/monitoring/monitorlib/clients/flight_planning/planning.py @@ -0,0 +1,71 @@ +from enum import Enum +from typing import Optional, List + +from implicitdict import ImplicitDict + +from monitoring.monitorlib.clients.flight_planning.flight_info import FlightID +from monitoring.monitorlib.fetch import Query + + +class PlanningActivityResult(str, Enum): + """The result of the flight planning operation.""" + + Completed = "Completed" + """The user's flight plan has been updated according to the situation specified by the user.""" + + Rejected = "Rejected" + """The updates the user requested to their flight plan are not allowed according to the rules under which the flight plan is being managed. The reasons for rejection may include a disallowed conflict with another flight during preflight.""" + + Failed = "Failed" + """The USS was not able to successfully authorize or update the flight plan due to a problem with the USS or a downstream system.""" + + NotSupported = "NotSupported" + """The USS's implementation does not support the attempted interaction. For instance, if the request specified a high-priority flight and the USS does not support management of high-priority flights.""" + + +class FlightPlanStatus(str, Enum): + """Status of a user's flight plan.""" + + NotPlanned = "NotPlanned" + """The USS has not created an authorized flight plan for the user.""" + + Planned = "Planned" + """The USS has created an authorized flight plan for the user, but the user may not yet start flying (even if within the time bounds of the flight plan).""" + + OkToFly = "OkToFly" + """The flight plan is in a state such that it is ok for the user to nominally fly within the bounds (including time) of the flight plan.""" + + OffNominal = "OffNominal" + """The flight plan now reflects the operator's actions, but the flight plan is not in a nominal state (e.g., the USS has placed the ASTM F3548-21 operational intent into one of the Nonconforming or Contingent states).""" + + Closed = "Closed" + """The flight plan was closed successfully by the USS and is now out of the UTM system.""" + + +class AdvisoryInclusion(str, Enum): + """Indication of whether any advisories or conditions were provided to the user along with the result of a flight planning attempt.""" + + Unknown = "Unknown" + """It is unknown or irrelevant whether advisories or conditions were provided to the user.""" + + AtLeastOneAdvisoryOrCondition = "AtLeastOneAdvisoryOrCondition" + """At least one advisory or condition was provided to the user.""" + + NoAdvisoriesOrConditions = "NoAdvisoriesOrConditions" + """No advisories or conditions were provided to the user.""" + + +class PlanningActivityResponse(ImplicitDict): + flight_id: FlightID + """Identity of flight for which the planning activity was conducted.""" + + queries: List[Query] + """Queries used to accomplish this activity.""" + + activity_result: PlanningActivityResult + """The result of the flight planning activity.""" + + flight_plan_status: FlightPlanStatus + """Status of the flight plan following the flight planning activity.""" + + includes_advisories: Optional[AdvisoryInclusion] = AdvisoryInclusion.Unknown diff --git a/monitoring/monitorlib/clients/flight_planning/test_preparation.py b/monitoring/monitorlib/clients/flight_planning/test_preparation.py new file mode 100644 index 0000000000..e88bf6679f --- /dev/null +++ b/monitoring/monitorlib/clients/flight_planning/test_preparation.py @@ -0,0 +1,25 @@ +from typing import Optional, List + +from implicitdict import ImplicitDict + +from monitoring.monitorlib.fetch import Query + + +class ClearAreaOutcome(ImplicitDict): + success: Optional[bool] = False + """True if, and only if, all flight plans in the specified area managed by the USS were canceled and removed.""" + + message: Optional[str] + """If the USS was unable to clear the entire area, this message can provide information on the problem encountered.""" + + +class ClearAreaResponse(ImplicitDict): + outcome: ClearAreaOutcome + + +class TestPreparationActivityResponse(ImplicitDict): + errors: Optional[List[str]] = None + """If any errors occurred during this activity, a list of those errors.""" + + queries: List[Query] + """Queries used to accomplish this activity."""