Added test for operational intent submission

config/bern/flight_declaration_via_operational_intent.json

@@ -0,0 +1,6 @@
+{
+  "minx": 7.4719589491516558,
+  "miny": 46.9799127188803993,
+  "maxx": 7.4870457729811619,
+  "maxy": 46.9865389634242945
+}

config/default.yaml

@@ -34,6 +34,7 @@ air_traffic_simulator_settings:
 data_files:
   trajectory: "config/bern/trajectory_f1.json" # Path to flight declarations JSON file
   flight_declaration: "config/bern/flight_declaration.json" # Path to flight declarations JSON file
+  flight_declaration_via_operational_intent: "config/bern/flight_declaration_via_operational_intent.json" # Path to flight declaration via operational intent JSON file
   # geo_fence: "config/geo_fences.json" # Path to geo-fences
 
 # List of test scenario IDs to execute

pyproject.toml

@@ -24,8 +24,7 @@ dependencies = [
     "jwcrypto==1.5.6",
     "pyproj==3.7.1",
     "shapely==2.1.0",
-    "implicitdict==3.0.0",
-    "uas-standards==3.5.0",
+    "implicitdict==4.0.1",
     "geojson==3.2.0",
     "folium>=0.20.0",
     "faker==9.3.1",
@@ -45,6 +44,7 @@ dependencies = [
     "pydantic-settings>=2.10.1",
     "websocket-client==1.9.0",
     "markdown>=3.10",
+    "uas-standards==4.2.0",
 ]
 
 [project.scripts]

src/openutm_verification/assets/simulator_templates/flight_declaration_via_operational_intent_template.json

@@ -0,0 +1,20 @@
+{
+    "exchange_type": "flight_declaration",
+    "aircraft_id": "a5dd8899-bc19-c8c4-2dd7-57f786d1379d",
+    "flight_id": "5a7f3377-b991-4cc8-af2d-379d57f786d1",
+    "plan_id": "a5b5484c-a23c-4e83-8bb8-a6a5c294e45b",
+    "flight_state": 2,
+    "flight_approved": 0,
+    "sequence_number": 0,
+    "start_datetime": "2023-06-12T16:35:08.842Z",
+    "end_datetime": "2023-06-12T16:40:08.842Z",
+    "version": "1.0.0",
+    "purpose": "Delivery",
+    "expect_telemetry": true,
+    "originating_party": "Medicine Delivery Company",
+    "contact_url": "https://utm.originatingparty.com/contact?5a7f3377-b991-4cc8-af2d-379d57f786d1",
+    "type_of_operation": 1,
+    "vehicle_id": "157de9bb-6b49-496b-bf3f-0b768ce6a3b6",
+    "operator_id": "4a725cb5-02d2-4f78-888f-b93088d324be",
+    "operational_intent_volume4ds": []
+}

src/openutm_verification/core/clients/flight_blender/flight_blender_client.py

@@ -40,7 +40,9 @@
     RIDOperatorDetails,
     UAClassificationEU,
 )
-from openutm_verification.simulator.models.flight_data_types import FlightObservationSchema
+from openutm_verification.simulator.models.flight_data_types import (
+    FlightObservationSchema,
+)
 
 
 def _create_rid_operator_details(operation_id: str) -> RIDOperatorDetails:
@@ -251,6 +253,65 @@ async def upload_flight_declaration(self, declaration: str | BaseModel) -> dict[
 
         return response_json
 
+    @scenario_step("Upload Flight Declaration Via Operational Intent")
+    async def upload_flight_declaration_via_operational_intent(self, declaration: str | BaseModel) -> dict[str, Any]:
+        """Upload a flight declaration to the Flight Blender API.
+
+        Accepts either a filename (str) containing JSON declaration data, or a
+        FlightDeclaration model instance. Adjusts datetimes to current time + offsets,
+        and posts it. Raises an error if the declaration is not approved.
+
+        Args:
+            declaration: Either a path to the JSON flight declaration file (str),
+                or a FlightDeclaration model instance.
+
+        Returns:
+            The JSON response from the API.
+
+        Raises:
+            FlightBlenderError: If the declaration is not approved or the request fails.
+            json.JSONDecodeError: If the file content is invalid JSON (when using filename).
+        """
+        endpoint = "/flight_declaration_ops/set_operational_intent"
+
+        # Handle different input types
+        if isinstance(declaration, str):
+            # Load from file
+            logger.debug(f"Uploading flight declaration from {declaration}")
+            with open(declaration, "r", encoding="utf-8") as flight_declaration_file:
+                f_d = flight_declaration_file.read()
+            flight_declaration = json.loads(f_d)
+        else:
+            # Assume it's a model with model_dump method
+            logger.debug("Uploading flight declaration from model")
+            flight_declaration = declaration.model_dump(mode="json")
+
+        # Adjust datetimes to current time + offsets
+        now = arrow.now()
+        few_seconds_from_now = now.shift(seconds=5)
+        four_minutes_from_now = now.shift(minutes=4)
+
+        flight_declaration["start_datetime"] = few_seconds_from_now.isoformat()
+        flight_declaration["end_datetime"] = four_minutes_from_now.isoformat()
+
+        response = await self.post(endpoint, json=flight_declaration)
+        logger.info(f"Flight declaration upload response: {response.status_code}")
+
+        response_json = response.json()
+
+        if not response_json.get("is_approved"):
+            logger.error(f"Flight declaration not approved. State: {OperationState(response_json.get('state')).name}")
+            raise FlightBlenderError(f"Flight declaration not approved. State: {OperationState(response_json.get('state')).name}")
+        # Store latest declaration id for later use
+        try:
+            self.latest_flight_declaration_id = response_json.get("id")
+            logger.info(f"Flight declaration uploaded and approved, ID: {self.latest_flight_declaration_id}")
+        except AttributeError:
+            self.latest_flight_declaration_id = None
+            logger.warning("Failed to extract flight declaration ID from response")
+
+        return response_json
+
     @scenario_step("Wait for User Input")
     async def wait_for_user_input(self, prompt: str = "Press Enter to continue...") -> str:
         """Wait for user input to proceed.
@@ -775,9 +836,37 @@ async def teardown_flight_declaration(self):
         logger.info("Tearing down flight declaration...")
         await self.delete_flight_declaration()
 
+    @scenario_step("Setup Flight Declaration via Operational Intent")
+    async def setup_flight_declaration_via_operational_intent(
+        self,
+        flight_declaration_via_operational_intent_path: str,
+        trajectory_path: str,
+    ) -> None:
+        """Generates data and uploads flight declaration via Operational Intent."""
+        from openutm_verification.scenarios.common import (
+            generate_flight_declaration_via_operational_intent,
+            generate_telemetry,
+        )
+        flight_declaration = generate_flight_declaration_via_operational_intent(flight_declaration_via_operational_intent_path)
+
+        telemetry_states = generate_telemetry(trajectory_path)
+
+        self.telemetry_states = telemetry_states
+
+        # Store data in ScenarioContext for reporting
+        ScenarioContext.set_flight_declaration_via_operational_intent_data(flight_declaration)
+        ScenarioContext.set_telemetry_data(telemetry_states)
+
+        upload_result = await self.upload_flight_declaration_via_operational_intent(flight_declaration)
+
+        if upload_result.status == Status.FAIL:
+            logger.error(f"Flight declaration upload failed: {upload_result}")
+            raise FlightBlenderError("Failed to upload flight declaration during setup_flight_declaration_via_operational_intent")
+
     @scenario_step("Setup Flight Declaration")
     async def setup_flight_declaration(self, flight_declaration_path: str, trajectory_path: str) -> None:
         """Generates data and uploads flight declaration."""
+
         from openutm_verification.scenarios.common import (
             generate_flight_declaration,
             generate_telemetry,
@@ -808,3 +897,19 @@ async def create_flight_declaration(self, data_files: DataFiles):
             yield
         finally:
             logger.info("All test steps complete..")
+
+    @asynccontextmanager
+    async def create_flight_declaration_via_operational_intent(self, data_files: DataFiles):
+        """Context manager to setup and teardown a flight operation based on scenario config."""
+        assert data_files.flight_declaration_via_operational_intent is not None, (
+            "Flight declaration via operational intent file path must be provided"
+        )
+        assert data_files.trajectory is not None, "Trajectory file path must be provided"
+        await self.setup_flight_declaration_via_operational_intent(
+            flight_declaration_via_operational_intent_path=data_files.flight_declaration_via_operational_intent,
+            trajectory_path=data_files.trajectory,
+        )
+        try:
+            yield
+        finally:
+            logger.info("All test steps complete..")

src/openutm_verification/core/execution/config_models.py

@@ -68,8 +68,14 @@ class DataFiles(StrictBaseModel):
     trajectory: str | None = None
     flight_declaration: str | None = None
     geo_fence: str | None = None
-
-    @field_validator("trajectory", "flight_declaration", "geo_fence")
+    flight_declaration_via_operational_intent: str | None = None
+
+    @field_validator(
+        "trajectory",
+        "flight_declaration",
+        "flight_declaration_via_operational_intent",
+        "geo_fence",
+    )
     @classmethod
     def validate_path(cls, v: str | None) -> str | None:
         """Validate that path is a non-empty string if provided."""
@@ -99,6 +105,11 @@ def resolve_and_validate_path(path_str: str, field_name: str) -> str:
             self.trajectory = resolve_and_validate_path(self.trajectory, "Trajectory")
         if self.flight_declaration:
             self.flight_declaration = resolve_and_validate_path(self.flight_declaration, "Flight declaration")
+        if self.flight_declaration_via_operational_intent:
+            self.flight_declaration_via_operational_intent = resolve_and_validate_path(
+                self.flight_declaration_via_operational_intent,
+                "Flight declaration via operational intent",
+            )
         if self.geo_fence:
             self.geo_fence = resolve_and_validate_path(self.geo_fence, "Geo-fence")
 

src/openutm_verification/core/execution/dependencies.py

@@ -1,15 +1,40 @@
-from typing import Any, AsyncGenerator, Callable, Coroutine, Generator, Iterable, TypeVar, cast
+from typing import (
+    Any,
+    AsyncGenerator,
+    Callable,
+    Coroutine,
+    Generator,
+    Iterable,
+    TypeVar,
+    cast,
+)
 
 from loguru import logger
 
 from openutm_verification.auth.providers import get_auth_provider
-from openutm_verification.core.clients.air_traffic.air_traffic_client import AirTrafficClient
-from openutm_verification.core.clients.air_traffic.base_client import create_air_traffic_settings
-from openutm_verification.core.clients.flight_blender.flight_blender_client import FlightBlenderClient
-from openutm_verification.core.clients.opensky.base_client import create_opensky_settings
+from openutm_verification.core.clients.air_traffic.air_traffic_client import (
+    AirTrafficClient,
+)
+from openutm_verification.core.clients.air_traffic.base_client import (
+    create_air_traffic_settings,
+)
+from openutm_verification.core.clients.flight_blender.flight_blender_client import (
+    FlightBlenderClient,
+)
+from openutm_verification.core.clients.opensky.base_client import (
+    create_opensky_settings,
+)
 from openutm_verification.core.clients.opensky.opensky_client import OpenSkyClient
-from openutm_verification.core.execution.config_models import AppConfig, DataFiles, ScenarioId, get_settings
-from openutm_verification.core.execution.dependency_resolution import CONTEXT, dependency
+from openutm_verification.core.execution.config_models import (
+    AppConfig,
+    DataFiles,
+    ScenarioId,
+    get_settings,
+)
+from openutm_verification.core.execution.dependency_resolution import (
+    CONTEXT,
+    dependency,
+)
 from openutm_verification.core.reporting.reporting_models import ScenarioResult
 from openutm_verification.scenarios.registry import SCENARIO_REGISTRY
 
@@ -66,7 +91,14 @@ def scenarios() -> Iterable[tuple[str, Callable[..., Coroutine[Any, Any, Scenari
             scenario_func = SCENARIO_REGISTRY[scenario_id].get("func")
             docs_content = get_scenario_docs(scenario_id)
 
-            CONTEXT.set({"scenario_id": scenario_id, "suite_scenario": suite_scenario, "suite_name": suite_name, "docs": docs_content})
+            CONTEXT.set(
+                {
+                    "scenario_id": scenario_id,
+                    "suite_scenario": suite_scenario,
+                    "suite_name": suite_name,
+                    "docs": docs_content,
+                }
+            )
             yield scenario_id, scenario_func
         else:
             logger.warning(f"Scenario {scenario_id} not found in registry.")
@@ -99,6 +131,9 @@ def data_files(scenario_id: ScenarioId) -> Generator[DataFiles, None, None]:
         # Merge suite overrides with base config
         trajectory = suite_scenario.trajectory or config.data_files.trajectory
         flight_declaration = suite_scenario.flight_declaration or config.data_files.flight_declaration
+        flight_declaration_via_operational_intent = (
+            suite_scenario.flight_declaration_via_operational_intent or config.data_files.flight_declaration_via_operational_intent
+        )
         geo_fence = suite_scenario.geo_fence or config.data_files.geo_fence
     else:
         # Use base config
@@ -109,6 +144,7 @@ def data_files(scenario_id: ScenarioId) -> Generator[DataFiles, None, None]:
     data = DataFiles(
         trajectory=trajectory,
         flight_declaration=flight_declaration,
+        flight_declaration_via_operational_intent=flight_declaration_via_operational_intent,
         geo_fence=geo_fence,
     )
     yield data
@@ -125,7 +161,9 @@ def app_config() -> Generator[AppConfig, None, None]:
 
 
 @dependency(FlightBlenderClient)
-async def flight_blender_client(config: AppConfig) -> AsyncGenerator[FlightBlenderClient, None]:
+async def flight_blender_client(
+    config: AppConfig,
+) -> AsyncGenerator[FlightBlenderClient, None]:
     """Provides a FlightBlenderClient instance for dependency injection.
 
     Args:
@@ -151,7 +189,9 @@ async def opensky_client(config: AppConfig) -> AsyncGenerator[OpenSkyClient, Non
 
 
 @dependency(AirTrafficClient)
-async def air_traffic_client(config: AppConfig) -> AsyncGenerator[AirTrafficClient, None]:
+async def air_traffic_client(
+    config: AppConfig,
+) -> AsyncGenerator[AirTrafficClient, None]:
     """Provides an AirTrafficClient instance for dependency injection."""
     settings = create_air_traffic_settings()
     async with AirTrafficClient(settings) as air_traffic_client: