Save scenario related generated data in reports.

config/pull_request.yaml

@@ -33,14 +33,14 @@ data_files:
 suites:
   pull_requests:
     scenarios:
+    - name: add_flight_declaration
     - name: F1_happy_path
       trajectory: "config/bern/trajectory_f1.json"
     - name: F2_contingent_path
       trajectory: "config/bern/trajectory_f2.json"
     - name: F3_non_conforming_path
       trajectory: "config/bern/trajectory_f3.json"
     - name: opensky_live_data
-    - name: add_flight_declaration
     - name: geo_fence_upload
     - name: openutm_sim_air_traffic_data
 

src/openutm_verification/auth/dev_auth.py

@@ -1,7 +1,6 @@
 import datetime
 import uuid
 from abc import ABC, abstractmethod
-from typing import List
 
 import jwcrypto.jwk
 import jwcrypto.jwt
@@ -23,7 +22,7 @@ class AuthAdapter(ABC):
     """Abstract base class for an adapter that adds JWTs to requests."""
 
     @abstractmethod
-    def issue_token(self, intended_audience: str, scopes: List[str]) -> str:
+    def issue_token(self, intended_audience: str, scopes: list[str]) -> str:
         """Subclasses must return a bearer token for the given audience."""
         pass
 
@@ -60,7 +59,7 @@ def __init__(self, sub: str = "uss_noauth"):
         self.sub = sub
 
     # Overrides method in AuthAdapter
-    def issue_token(self, intended_audience: str, scopes: List[str]) -> str:
+    def issue_token(self, intended_audience: str, scopes: list[str]) -> str:
         timestamp = int((datetime.datetime.now(datetime.timezone.utc) - EPOCH).total_seconds())
         jwt = jwcrypto.jwt.JWT(
             header={"typ": "JWT", "alg": "RS256"},

src/openutm_verification/auth/noauth.py

@@ -2,15 +2,13 @@
 No-authentication credentials provider for development/testing.
 """
 
-from typing import List
-
 from openutm_verification.auth.dev_auth import NoAuth
 
 
 class NoAuthCredentialsGetter:
     """Credentials getter that uses dummy authentication for development."""
 
-    def get_cached_credentials(self, audience: str, scopes: List[str]):
+    def get_cached_credentials(self, audience: str, scopes: list[str]):
         """Get cached credentials using dummy authentication."""
         if not audience:
             return {"error": "An audience parameter must be provided"}

src/openutm_verification/auth/oauth2.py

@@ -1,5 +1,4 @@
 import time
-from typing import Optional
 
 import httpx
 from loguru import logger
@@ -16,7 +15,7 @@ class OAuth2Token(BaseModel):
     access_token: str
     token_type: str = "Bearer"
     expires_in: int
-    expires_at: Optional[float] = None
+    expires_at: float | None = None
 
     def is_expired(self, buffer_seconds: int = 60) -> bool:
         """Check if token is expired with buffer time."""
@@ -39,7 +38,7 @@ def __init__(
         self.client_id = client_id
         self.client_secret = client_secret
         self.client = httpx.AsyncClient(timeout=timeout)
-        self._token: Optional[OAuth2Token] = None
+        self._token: OAuth2Token | None = None
 
     async def get_access_token(self) -> str:
         """Get valid access token, acquiring or refreshing as needed."""

src/openutm_verification/auth/passport.py

@@ -3,7 +3,6 @@
 """
 
 from os import environ as env
-from typing import List, Optional
 
 import requests
 from loguru import logger
@@ -14,19 +13,19 @@ class PassportCredentialsGetter:
 
     def __init__(
         self,
-        client_id: Optional[str] = None,
-        client_secret: Optional[str] = None,
-        audience: Optional[str] = None,
-        token_endpoint: Optional[str] = None,
-        passport_base_url: Optional[str] = None,
+        client_id: str | None = None,
+        client_secret: str | None = None,
+        audience: str | None = None,
+        token_endpoint: str | None = None,
+        passport_base_url: str | None = None,
     ):
         self.client_id = client_id or env.get("BLENDER_WRITE_CLIENT_ID")
         self.client_secret = client_secret or env.get("BLENDER_WRITE_CLIENT_SECRET")
         self.audience = audience
         self.token_endpoint = token_endpoint or env.get("PASSPORT_TOKEN_URL")
         self.base_url = passport_base_url or env.get("PASSPORT_URL")
 
-    def get_cached_credentials(self, audience: Optional[str] = None, scopes: Optional[List[str]] = None):
+    def get_cached_credentials(self, audience: str | None = None, scopes: list[str] | None = None):
         """Get cached credentials with token refresh logic."""
 
         if not audience:

src/openutm_verification/cli/__init__.py

@@ -42,11 +42,17 @@ def main():
     ConfigProxy.initialize(config)
 
     # Setup logging
-    output_dir = Path(config.reporting.output_dir)
+    run_timestamp = datetime.now(timezone.utc)
+    timestamp_str = run_timestamp.strftime("%Y-%m-%dT%H-%M-%SZ")
+
+    base_output_dir = Path(config.reporting.output_dir)
+    output_dir = base_output_dir / f"run_{timestamp_str}"
     output_dir.mkdir(parents=True, exist_ok=True)
 
-    run_timestamp = datetime.now(timezone.utc)
-    base_filename = f"report_{run_timestamp.strftime('%Y-%m-%dT%H-%M-%SZ')}"
+    # Update config so downstream components use the new directory
+    config.reporting.output_dir = str(output_dir)
+
+    base_filename = "report"
     log_file = setup_logging(output_dir, base_filename, config.reporting.formats, args.debug)
 
     # Run verification scenarios

src/openutm_verification/core/clients/air_traffic/air_traffic_client.py

@@ -1,6 +1,5 @@
 import json
 import uuid
-from typing import Optional
 from uuid import UUID
 
 from loguru import logger
@@ -18,6 +17,7 @@
 )
 from openutm_verification.simulator.models.flight_data_types import (
     AirTrafficGeneratorConfiguration,
+    FlightObservationSchema,
 )
 
 
@@ -33,9 +33,9 @@ def __init__(self, settings: AirTrafficSettings):
     @scenario_step("Generate Simulated Air Traffic Data")
     async def generate_simulated_air_traffic_data(
         self,
-        config_path: Optional[str] = None,
-        duration: Optional[int] = None,
-    ) -> list[list[dict]]:
+        config_path: str | None = None,
+        duration: int | None = None,
+    ) -> list[list[FlightObservationSchema]]:
         """Generate simulated air traffic data from GeoJSON configuration.
 
         Loads GeoJSON data from the specified config path and uses it to generate

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

@@ -1,8 +1,6 @@
-from typing import Any
-
 import httpx
 from loguru import logger
-from websocket import create_connection
+from websocket import WebSocket, create_connection
 
 from openutm_verification.models import FlightBlenderError
 
@@ -65,7 +63,7 @@ async def __aenter__(self):
     async def __aexit__(self, exc_type, exc_val, exc_tb):
         await self.client.aclose()
 
-    def create_websocket_connection(self, endpoint) -> Any:
+    def create_websocket_connection(self, endpoint) -> WebSocket:
         """Create and return a WebSocket connection to the Flight Blender service.
 
         This method establishes a WebSocket connection using the configured
@@ -86,7 +84,7 @@ def create_websocket_connection(self, endpoint) -> Any:
             websocket_connection.send(self.client.headers["Authorization"])
         return websocket_connection
 
-    def close_websocket_connection(self, ws_connection: Any) -> None:
+    def close_websocket_connection(self, ws_connection: WebSocket) -> None:
         """Close the given WebSocket connection.
 
         Args: