Air Traffic Client Unification phase 1

config/pull_request.yaml

@@ -32,6 +32,14 @@ blue_sky_air_traffic_simulator_settings:
   single_or_multiple_sensors: "multiple" # this setting specifiies if the traffic data is submitted from a single sensor or multiple sensors
   sensor_ids: ["562e6297036a4adebb4848afcd1ede90"] # List of sensor IDs to use when 'multiple' is selected
 
+# Bayesian Air traffic data configuration
+bayesian_air_traffic_simulator_settings:
+  number_of_aircraft: 3
+  simulation_duration_seconds: 30
+  single_or_multiple_sensors: "multiple" # this setting specifies if the traffic data is submitted from a single sensor or multiple sensors
+  sensor_ids: ["562e6297036a4adebb4848afcd1ede90"] # List of sensor IDs to use when 'multiple' is selected
+  session_ids: ["ee9405e564ea4373823e37d950858e6a"] # List of session IDs to use when 'multiple' is selected, a session id is needed in Flight Blender to depict a period of time these observations were made (this assumes the observations may not be continuous); if empty, random UUIDs will be generated
+
 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

pyproject.toml

@@ -49,6 +49,7 @@ dependencies = [
     "uvicorn[standard]>=0.38.0",  # includes watchfiles for efficient reload
     "bluesky-simulator==1.1.0",
     "rtree==1.4.1",
+    "cam-track-gen @ git+https://github.com/openutm-labs/Canadian-Airspace-Models.git"
 ]
 
 [project.scripts]
@@ -70,9 +71,6 @@ dev = [
     "types-requests",
     "pytest-cov>=7.0.0",
 ]
-bayesian-track-generation = [
-    "cam-track-gen @ git+https://github.com/openutm-labs/Canadian-Airspace-Models.git",
-]
 
 [build-system]
 requires = ["hatchling"]
@@ -85,6 +83,9 @@ packages = ["src/openutm_verification"]
 [tool.hatch.build.targets.wheel.force-include]
 "docs/scenarios" = "openutm_verification/docs/scenarios"
 
+[tool.hatch.metadata]
+allow-direct-references = true
+
 [tool.pytest.ini_options]
 pythonpath = [".", "src/openutm_verification"]
 testpaths = ["tests"]

scenarios/stream_air_traffic_example.yaml

@@ -0,0 +1,53 @@
+name: stream_air_traffic_example
+version: "1.0"
+description: |
+  Example scenario demonstrating the unified "Stream Air Traffic" step.
+
+  This step replaces the provider-specific steps:
+  - "Generate Simulated Air Traffic Data" (geojson)
+  - "Generate BlueSky Simulation Air Traffic Data" (bluesky)
+  - "Generate Bayesian Simulation Air Traffic Data" (bayesian)
+  - "Fetch OpenSky Data" (opensky)
+
+  All providers can now be used with a single consistent interface.
+
+steps:
+  # Example 1: GeoJSON provider with data generation only (no delivery)
+  - step: Stream Air Traffic
+    id: geojson_only
+    arguments:
+      provider: geojson
+      duration: 10
+      target: none  # Don't send anywhere, just return data
+      config_path: config/bern/trajectory_f1.json
+      number_of_aircraft: 2
+
+  # Example 2: GeoJSON with delivery to Flight Blender
+  - step: Stream Air Traffic
+    id: geojson_stream
+    arguments:
+      provider: geojson
+      duration: 30
+      target: flight_blender
+      config_path: config/bern/trajectory_f1.json
+      number_of_aircraft: 2
+      session_ids:
+        - "550e8400-e29b-41d4-a716-446655440001"
+
+  # Example 3: BlueSky simulator (requires bluesky-simulator package)
+  # - step: Stream Air Traffic
+  #   id: bluesky_stream
+  #   arguments:
+  #     provider: bluesky
+  #     duration: 30
+  #     target: flight_blender
+  #     config_path: config/bern/blue_sky_sim_bern.scn
+
+  # Example 4: Live OpenSky data for Switzerland region
+  # - step: Stream Air Traffic
+  #   id: opensky_live
+  #   arguments:
+  #     provider: opensky
+  #     duration: 30
+  #     target: flight_blender
+  #     viewport: [45.8389, 47.8229, 5.9962, 10.5226]

src/openutm_verification/core/execution/dependencies.py

@@ -12,7 +12,11 @@
 from openutm_verification.core.clients.air_traffic.air_traffic_client import (
     AirTrafficClient,
 )
-from openutm_verification.core.clients.air_traffic.base_client import AirTrafficSettings, BayesianAirTrafficSettings, BlueSkyAirTrafficSettings
+from openutm_verification.core.clients.air_traffic.base_client import (
+    AirTrafficSettings,
+    BayesianAirTrafficSettings,
+    BlueSkyAirTrafficSettings,
+)
 from openutm_verification.core.clients.air_traffic.bayesian_air_traffic_client import (
     BayesianTrafficClient,
 )
@@ -41,6 +45,7 @@
     CONTEXT,
     dependency,
 )
+from openutm_verification.core.steps.air_traffic_step import AirTrafficStepClient
 from openutm_verification.server.runner import SessionManager
 from openutm_verification.utils.paths import get_docs_directory
 
@@ -226,3 +231,10 @@ async def amqp_client(config: AppConfig) -> AsyncGenerator[AMQPClient, None]:
     settings = AMQPSettings.from_config(config.amqp) if config.amqp else AMQPSettings()
     async with AMQPClient(settings) as client:
         yield client
+
+
+@dependency(AirTrafficStepClient)
+async def air_traffic_step_client() -> AsyncGenerator[AirTrafficStepClient, None]:
+    """Provides an AirTrafficStepClient instance for the unified Stream Air Traffic step."""
+    async with AirTrafficStepClient() as client:
+        yield client

src/openutm_verification/core/providers/__init__.py

@@ -0,0 +1,15 @@
+"""Air traffic providers module.
+
+Providers generate or fetch air traffic observation data from various sources.
+"""
+
+from .factory import ProviderType, create_provider
+from .opensky_provider import DEFAULT_SWITZERLAND_VIEWPORT
+from .protocol import AirTrafficProvider
+
+__all__ = [
+    "AirTrafficProvider",
+    "DEFAULT_SWITZERLAND_VIEWPORT",
+    "ProviderType",
+    "create_provider",
+]

src/openutm_verification/core/providers/bayesian_provider.py

@@ -0,0 +1,98 @@
+"""Bayesian air traffic provider - wraps BayesianTrafficClient."""
+
+from __future__ import annotations
+
+from openutm_verification.core.clients.air_traffic.base_client import (
+    BayesianAirTrafficSettings,
+)
+from openutm_verification.core.clients.air_traffic.bayesian_air_traffic_client import (
+    BayesianTrafficClient,
+)
+from openutm_verification.simulator.models.flight_data_types import (
+    FlightObservationSchema,
+)
+
+
+class BayesianProvider:
+    """Provider that generates air traffic using Bayesian track generation.
+
+    Wraps the existing BayesianTrafficClient to provide a consistent interface.
+    Note: Requires the cam-track-gen package to be installed.
+    """
+
+    def __init__(
+        self,
+        config_path: str | None = None,
+        number_of_aircraft: int | None = None,
+        duration: int | None = None,
+        sensor_ids: list[str] | None = None,
+        session_ids: list[str] | None = None,
+    ):
+        """Initialize the Bayesian provider.
+
+        Args:
+            config_path: Path to config (currently unused by Bayesian client).
+            number_of_aircraft: Number of aircraft to simulate.
+            duration: Simulation duration in seconds.
+            sensor_ids: List of sensor UUID strings.
+            session_ids: List of session UUID strings.
+        """
+        self._config_path = config_path or ""
+        self._number_of_aircraft = number_of_aircraft or 2
+        self._duration = duration or 30
+        self._sensor_ids = sensor_ids or []
+        self._session_ids = session_ids or []
+
+    @property
+    def name(self) -> str:
+        """Provider identifier."""
+        return "bayesian"
+
+    @classmethod
+    def from_kwargs(
+        cls,
+        config_path: str | None = None,
+        number_of_aircraft: int | None = None,
+        duration: int | None = None,
+        sensor_ids: list[str] | None = None,
+        session_ids: list[str] | None = None,
+        **_kwargs,  # Ignore unknown kwargs for flexibility
+    ) -> "BayesianProvider":
+        """Factory method to create provider from keyword arguments."""
+        return cls(
+            config_path=config_path,
+            number_of_aircraft=number_of_aircraft,
+            duration=duration,
+            sensor_ids=sensor_ids,
+            session_ids=session_ids,
+        )
+
+    async def get_observations(
+        self,
+        duration: int | None = None,
+    ) -> list[list[FlightObservationSchema]]:
+        """Generate observations using the underlying BayesianTrafficClient.
+
+        Args:
+            duration: Override duration in seconds.
+
+        Returns:
+            List of observation lists per aircraft.
+        """
+        effective_duration = duration or self._duration
+
+        settings = BayesianAirTrafficSettings(
+            simulation_config_path=self._config_path,
+            simulation_duration_seconds=effective_duration,
+            number_of_aircraft=self._number_of_aircraft,
+            sensor_ids=self._sensor_ids,
+            session_ids=self._session_ids,
+        )
+
+        async with BayesianTrafficClient(settings) as client:
+            result = await client.generate_bayesian_sim_air_traffic_data(
+                config_path=self._config_path,
+                duration=effective_duration,
+            )
+            # Handle case where Bayesian client returns None or empty
+            return result if result else []

src/openutm_verification/core/providers/bluesky_provider.py

@@ -0,0 +1,96 @@
+"""BlueSky simulation air traffic provider - wraps BlueSkyClient."""
+
+from __future__ import annotations
+
+from openutm_verification.core.clients.air_traffic.base_client import (
+    BlueSkyAirTrafficSettings,
+)
+from openutm_verification.core.clients.air_traffic.blue_sky_client import (
+    BlueSkyClient,
+)
+from openutm_verification.simulator.models.flight_data_types import (
+    FlightObservationSchema,
+)
+
+
+class BlueSkyProvider:
+    """Provider that generates air traffic from BlueSky simulator scenarios.
+
+    Wraps the existing BlueSkyClient to provide a consistent interface.
+    Note: Requires the bluesky-simulator package to be installed.
+    """
+
+    def __init__(
+        self,
+        config_path: str | None = None,
+        number_of_aircraft: int | None = None,
+        duration: int | None = None,
+        sensor_ids: list[str] | None = None,
+        session_ids: list[str] | None = None,
+    ):
+        """Initialize the BlueSky provider.
+
+        Args:
+            config_path: Path to the BlueSky .scn scenario file.
+            number_of_aircraft: Number of aircraft to simulate.
+            duration: Simulation duration in seconds.
+            sensor_ids: List of sensor UUID strings.
+            session_ids: List of session UUID strings.
+        """
+        self._config_path = config_path or ""
+        self._number_of_aircraft = number_of_aircraft or 2
+        self._duration = duration or 30
+        self._sensor_ids = sensor_ids or []
+        self._session_ids = session_ids or []
+
+    @property
+    def name(self) -> str:
+        """Provider identifier."""
+        return "bluesky"
+
+    @classmethod
+    def from_kwargs(
+        cls,
+        config_path: str | None = None,
+        number_of_aircraft: int | None = None,
+        duration: int | None = None,
+        sensor_ids: list[str] | None = None,
+        session_ids: list[str] | None = None,
+        **_kwargs,  # Ignore unknown kwargs for flexibility
+    ) -> "BlueSkyProvider":
+        """Factory method to create provider from keyword arguments."""
+        return cls(
+            config_path=config_path,
+            number_of_aircraft=number_of_aircraft,
+            duration=duration,
+            sensor_ids=sensor_ids,
+            session_ids=session_ids,
+        )
+
+    async def get_observations(
+        self,
+        duration: int | None = None,
+    ) -> list[list[FlightObservationSchema]]:
+        """Generate observations using the underlying BlueSkyClient.
+
+        Args:
+            duration: Override duration in seconds.
+
+        Returns:
+            List of observation lists per aircraft.
+        """
+        effective_duration = duration or self._duration
+
+        settings = BlueSkyAirTrafficSettings(
+            simulation_config_path=self._config_path,
+            simulation_duration_seconds=effective_duration,
+            number_of_aircraft=self._number_of_aircraft,
+            sensor_ids=self._sensor_ids,
+            session_ids=self._session_ids,
+        )
+
+        async with BlueSkyClient(settings) as client:
+            return await client.generate_bluesky_sim_air_traffic_data(
+                config_path=self._config_path,
+                duration=effective_duration,
+            )