Air Traffic unification Part 2

config/default.yaml

@@ -35,15 +35,15 @@ air_traffic_simulator_settings:
 # Bluesky Air traffic data configuration
 blue_sky_air_traffic_simulator_settings:
   number_of_aircraft: 3
-  simulation_duration_seconds: 30
+  simulation_duration: 30
   single_or_multiple_sensors: "single" # 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
 
 # Bayesian Air traffic data configuration
 bayesian_air_traffic_simulator_settings:
   number_of_aircraft: 3
-  simulation_duration_seconds: 30
+  simulation_duration: 30
   single_or_multiple_sensors: "single" # 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

config/pull_request.yaml

@@ -28,14 +28,14 @@ air_traffic_simulator_settings:
 # Bluesky Air traffic data configuration
 blue_sky_air_traffic_simulator_settings:
   number_of_aircraft: 3
-  simulation_duration_seconds: 30
+  simulation_duration: 30
   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
+  simulation_duration: 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

scenarios/bayesian_sim_air_traffic_data.yaml

@@ -1,10 +1,8 @@
 name: bayesian_sim_air_traffic_data
 description: Bayesian simulation air traffic data test
 steps:
-- id: generate_bayesian_simulation_air_traffic_data
-  step: Generate Bayesian Simulation Air Traffic Data
-- step: Fetch Session IDs
-- step: Submit Simulated Air Traffic
+- step: Stream Air Traffic
+  id: stream_air_traffic
   arguments:
-    observations: ${{ steps.generate_bayesian_simulation_air_traffic_data.result }}
-    session_ids: ${{ steps.Fetch Session IDs.result }}
+    provider: bayesian
+    target: flight_blender

scenarios/bayesian_sim_air_traffic_data_varying_refresh_rates.yaml

@@ -1,10 +1,9 @@
 name: bayesian_sim_air_traffic_data_varying_refresh_rates
 description: Bayesian simulation air traffic data test with varying refresh rates
 steps:
-- id: generate_bayesian_simulation_air_traffic_data
-  step: Generate Bayesian Simulation Air Traffic Data
-- step: Fetch Session IDs
-- step: Submit Simulated Air Traffic at varying refresh rates
+- step: Stream Air Traffic
+  id: stream_air_traffic
   arguments:
-    observations: ${{ steps.generate_bayesian_simulation_air_traffic_data.result }}
-    session_ids: ${{ steps.Fetch Session IDs.result }}
+    provider: bayesian
+    target: flight_blender
+    refresh_mode: varying

scenarios/bluesky_sim_air_traffic_data.yaml

@@ -1,10 +1,8 @@
 name: bluesky_sim_air_traffic_data
 description: Blue Sky test
 steps:
-- id: generate_bluesky_simulation_air_traffic_data
-  step: Generate BlueSky Simulation Air Traffic Data
-- step: Fetch Session IDs
-- step: Submit Simulated Air Traffic
+- step: Stream Air Traffic
+  id: stream_air_traffic
   arguments:
-    observations: ${{ steps.generate_bluesky_simulation_air_traffic_data.result }}
-    session_ids: ${{ steps.Fetch Session IDs.result }}
+    provider: bluesky
+    target: flight_blender

scenarios/bluesky_sim_air_traffic_data_latency_issues.yaml

@@ -1,10 +1,9 @@
 name: bluesky_sim_air_traffic_data_latency_issues
 description: F3623 Blue Sky dataset with latency issues test
 steps:
-- id: generate_bluesky_sim_air_traffic_data_with_sensor_latency_issues
-  step: Generate BlueSky Simulation Air Traffic Data with latency issues
-- step: Fetch Session IDs
-- step: Submit Simulated Air Traffic
+- step: Stream Air Traffic
+  id: stream_air_traffic
   arguments:
-    observations: ${{ steps.generate_bluesky_sim_air_traffic_data_with_sensor_latency_issues.result }}
-    session_ids: ${{ steps.Fetch Session IDs.result }}
+    provider: bluesky
+    target: flight_blender
+    data_quality: latency

scenarios/opensky_live_data.yaml

@@ -1,24 +1,11 @@
 name: opensky_live_data
 description: Fetch live flight data from OpenSky and submit to Flight Blender.
 
-groups:
-  fetch_and_submit_opensky:
-    description: Fetches OpenSky data and submits it to Flight Blender
-    steps:
-      - id: fetch
-        step: Fetch OpenSky Data
-
-      - id: submit
-        step: Submit Air Traffic
-        arguments:
-          observations: ${{ steps.fetch.result }}
-
-      - id: wait
-        step: Wait X seconds
-        arguments:
-          duration: 3
-
 steps:
-  - step: fetch_and_submit_opensky
+  - step: Stream Air Traffic
+    id: stream_opensky
+    arguments:
+      provider: opensky
+      target: flight_blender
     loop:
       count: 5

scenarios/openutm_sim_air_traffic_data.yaml

@@ -1,10 +1,9 @@
 name: openutm_sim_air_traffic_data
-description: Generate simulated air traffic data using OpenSky client and submit to Flight Blender.
+description: Generate simulated air traffic data using GeoJSON provider and submit to Flight Blender.
 
 steps:
-  - step: Generate Simulated Air Traffic Data
-  - step: Fetch Session IDs
-  - step: Submit Simulated Air Traffic
+  - step: Stream Air Traffic
+    id: stream_air_traffic
     arguments:
-      observations: ${{ steps.Generate Simulated Air Traffic Data.result }}
-      session_ids: ${{ steps.Fetch Session IDs.result }}
+      provider: geojson
+      target: flight_blender

scenarios/sdsp_track.yaml

@@ -7,10 +7,11 @@ steps:
   arguments:
     session_id: ${{ steps.Generate UUID.result }}
     action: START
-- step: Generate Simulated Air Traffic Data
-- step: Submit Simulated Air Traffic
+- id: stream_air_traffic
+  step: Stream Air Traffic
   arguments:
-    observations: ${{ steps.Generate Simulated Air Traffic Data.result }}
+    provider: geojson
+    target: flight_blender
   background: true
 - id: wait_2_seconds
   step: Wait X seconds
@@ -31,4 +32,4 @@ steps:
     session_id: ${{ steps.Generate UUID.result }}
     action: STOP
   needs:
-  - Submit Simulated Air Traffic
+  - stream_air_traffic

scenarios/verify_sdsp_metrics.yaml

@@ -1,14 +1,12 @@
 name: verify_sdsp_metrics
 description: Verify that the SDSP is correctly reporting metrics based on submitted air traffic data.
 steps:
-- id: generate_bayesian_simulation_air_traffic_data
-  step: Generate Bayesian Simulation Air Traffic Data
-- step: Fetch Session IDs for Bayesian Simulation
-- step: Submit Simulated Air Traffic
-  description: The observations will be generated and submitted using a session_id to the air traffic endpoint, this session_id is different from the one use to create the SDSP session.
+- id: stream_air_traffic
+  step: Stream Air Traffic
+  description: Generate and submit Bayesian air traffic data in the background.
   arguments:
-    observations: ${{ steps.generate_bayesian_simulation_air_traffic_data.result }}
-    session_ids: ${{ steps.Fetch Session IDs for Bayesian Simulation.result }}
+    provider: bayesian
+    target: flight_blender
   background: true
 - id: wait_1_seconds
   step: Wait X seconds
@@ -27,13 +25,15 @@ steps:
   arguments:
     duration: 10
 - step: Verify Reported Metrics in Flight Blender
+  needs:
+    - stream_air_traffic
   arguments:
-    observations: ${{ steps.generate_bayesian_simulation_air_traffic_data.result }}
+    observations: ${{ steps.stream_air_traffic.result.observations }}
     # session_id: ${{ steps.generated_sdsp_session_id.result }}
 - id: stop_sdsp_session
   step: Start / Stop SDSP Session
   arguments:
     session_id: ${{ steps.generated_sdsp_session_id.result }}
     action: STOP
   needs:
-  - Submit Simulated Air Traffic
+  - stream_air_traffic

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

@@ -112,7 +112,8 @@ async def generate_simulated_air_traffic_data_with_latency(
         duration: int | None = None,
     ) -> list[list[FlightObservationSchema]]:
         """This method, simulates a adding latency to the flight observations list"""
-        flight_observations = self.generate_simulated_air_traffic_data(config_path=config_path, duration=duration)
+        step_result = await self.generate_simulated_air_traffic_data(config_path=config_path, duration=duration)
+        flight_observations = step_result.result
         LATENCY_PROBABILITY = 0.1  # 10% chance to have latency issues
         TIMESTAMP_SHIFT_RANGE_SECONDS = (-1, 2.5)  # Shift timestamps by -5 to +5 seconds
 

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

@@ -51,7 +51,7 @@ class BlueSkyAirTrafficSettings(BaseModel):
     """Settings for BlueSky Air Traffic API."""
 
     simulation_config_path: str = ""
-    simulation_duration_seconds: int = 30
+    simulation_duration: int = 30
     number_of_aircraft: int = 2
     single_or_multiple_sensors: Literal["single", "multiple"] = SENSOR_MODE_SINGLE
     sensor_ids: list[str] = []
@@ -62,7 +62,7 @@ def from_config(cls, sim_config: "BlueSkySimConfig", simulation_path: str | None
         """Create settings from config."""
         return cls(
             simulation_config_path=simulation_path or "",
-            simulation_duration_seconds=sim_config.simulation_duration_seconds,
+            simulation_duration=sim_config.simulation_duration,
             number_of_aircraft=sim_config.number_of_aircraft,
             single_or_multiple_sensors=sim_config.single_or_multiple_sensors,
             sensor_ids=sim_config.sensor_ids,
@@ -74,7 +74,7 @@ class BayesianAirTrafficSettings(BaseModel):
     """Settings for Bayesian Air Traffic API."""
 
     simulation_config_path: str = ""
-    simulation_duration_seconds: int = 30
+    simulation_duration: int = 30
     number_of_aircraft: int = 2
 
     single_or_multiple_sensors: Literal["single", "multiple"] = SENSOR_MODE_SINGLE
@@ -86,7 +86,7 @@ def from_config(cls, sim_config: "BayesianSimConfig", simulation_path: str | Non
         """Create settings from config."""
         return cls(
             simulation_config_path=simulation_path or "",
-            simulation_duration_seconds=sim_config.simulation_duration_seconds,
+            simulation_duration=sim_config.simulation_duration,
             number_of_aircraft=sim_config.number_of_aircraft,
             single_or_multiple_sensors=sim_config.single_or_multiple_sensors,
             sensor_ids=sim_config.sensor_ids,

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

@@ -75,14 +75,14 @@ async def generate_bayesian_sim_air_traffic_data(
 
         Args:
             config_path: Path to .scn scenario file. Defaults to settings.simulation_config_path.
-            duration: Simulation duration in seconds. Defaults to settings.simulation_duration_seconds (expected 30).
+            duration: Simulation duration in seconds. Defaults to settings.simulation_duration (expected 30).
         Returns:
             list[list[FlightObservationSchema]]: outer list per aircraft (icao_address),
             inner list is time-series sampled at 1 Hz.
         """
 
         # scn_path = config_path or self.settings.simulation_config_path
-        duration_in_seconds = int(duration or self.settings.simulation_duration_seconds or 30)
+        duration_in_seconds = int(duration or self.settings.simulation_duration or 30)
         number_of_aircraft = self.settings.number_of_aircraft or 3
         sensor_ids = self.settings.sensor_ids
         use_multiple_sensors = self.settings.single_or_multiple_sensors == SENSOR_MODE_MULTIPLE
@@ -219,7 +219,8 @@ async def generate_bayesian_sim_air_traffic_data_with_sensor_latency_issues(
         Latency is simulated by randomly removing some observations and randomly shifting the timestamps of some observations
         to be earlier or later than the actual timestamp, mimicking real-world sensor latency issues.
         """
-        flight_observations = self.generate_bayesian_sim_air_traffic_data(config_path=config_path, duration=duration)
+        step_result = await self.generate_bayesian_sim_air_traffic_data(config_path=config_path, duration=duration)
+        flight_observations = step_result.result
 
         LATENCY_PROBABILITY = 0.1  # 10% chance to have latency issues
         TIMESTAMP_SHIFT_RANGE_SECONDS = (-1, 2.5)  # Shift timestamps by -5 to +5 seconds

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

@@ -46,15 +46,15 @@ async def generate_bluesky_sim_air_traffic_data(
 
         Args:
             config_path: Path to .scn scenario file. Defaults to settings.simulation_config_path.
-            duration: Simulation duration in seconds. Defaults to settings.simulation_duration_seconds (expected 30).
+            duration: Simulation duration in seconds. Defaults to settings.simulation_duration (expected 30).
 
         Returns:
             list[list[FlightObservationSchema]]: outer list per aircraft (icao_address),
             inner list is time-series sampled at 1 Hz.
         """
 
         scn_path = config_path or self.settings.simulation_config_path
-        duration_s = int(duration or self.settings.simulation_duration_seconds or 30)
+        duration_s = int(duration or self.settings.simulation_duration or 30)
 
         sensor_ids = self.settings.sensor_ids
         use_multiple_sensors = self.settings.single_or_multiple_sensors == SENSOR_MODE_MULTIPLE
@@ -147,7 +147,7 @@ async def generate_bluesky_sim_air_traffic_data_with_sensor_latency_issues(
         """This method generates"""
 
         scn_path = config_path or self.settings.simulation_config_path
-        duration_s = int(duration or self.settings.simulation_duration_seconds or 30)
+        duration_s = int(duration or self.settings.simulation_duration or 30)
 
         sensor_ids = self.settings.sensor_ids
         use_multiple_sensors = self.settings.single_or_multiple_sensors == SENSOR_MODE_MULTIPLE

src/openutm_verification/core/clients/amqp/amqp_client.py

@@ -373,7 +373,8 @@ async def wait_for_messages(
         start_time = time.time()
 
         while (time.time() - start_time) < timeout:
-            messages = await self.get_received_messages(routing_key_filter=routing_key_filter)
+            step_result = await self.get_received_messages(routing_key_filter=routing_key_filter)
+            messages = step_result.result
             if len(messages) >= count:
                 return {
                     "success": True,
@@ -384,7 +385,8 @@ async def wait_for_messages(
             await asyncio.sleep(0.5)
 
         # Timeout reached
-        messages = await self.get_received_messages(routing_key_filter=routing_key_filter)
+        step_result = await self.get_received_messages(routing_key_filter=routing_key_filter)
+        messages = step_result.result
         return {
             "success": False,
             "message_count": len(messages),

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

@@ -843,7 +843,7 @@ async def submit_simulated_air_traffic(
             "observations_submitted": observations_submitted,
             "submission_errors": submission_errors,
             "duration_seconds": round(duration_seconds, 2),
-            "simulation_duration_seconds": (simulation_end - simulation_start).total_seconds(),
+            "simulation_duration": (simulation_end - simulation_start).total_seconds(),
         }
 
     @scenario_step("Submit Simulated Air Traffic at varying refresh rates")
@@ -1041,7 +1041,7 @@ async def verify_reported_metrics_in_flight_blender(self, observations: list[lis
                 error_message="No valid start/end times found in observations",
             )
 
-        simulation_start, simulation_end, duration_seconds = time_window
+        simulation_start, simulation_end, simulation_duration = time_window
         session_param = f"session_id={session_id}&" if session_id is not None else ""
         metrics_endpoint = f"/surveillance_monitoring_ops/service_metrics?{session_param}start_time={simulation_start}&end_time={simulation_end}"
         logger.info(f"Querying metrics from Flight Blender: {metrics_endpoint}")
@@ -1051,7 +1051,7 @@ async def verify_reported_metrics_in_flight_blender(self, observations: list[lis
             return StepResult(
                 name="Verify Reported Metrics in Flight Blender",
                 status=Status.FAIL,
-                duration=round(duration_seconds, 2),
+                duration=round(simulation_duration, 2),
                 error_message=f"Metrics endpoint returned HTTP {metrics_response.status_code}",
             )
 
@@ -1061,21 +1061,21 @@ async def verify_reported_metrics_in_flight_blender(self, observations: list[lis
             return StepResult(
                 name="Verify Reported Metrics in Flight Blender",
                 status=Status.FAIL,
-                duration=round(duration_seconds, 2),
+                duration=round(simulation_duration, 2),
                 error_message=f"Invalid metrics response structure: {e}",
             )
 
         num_aircraft = sum(1 for a in observations if a)
         total_observations = sum(len(a) for a in observations if a)
         # Track update probability and heartbeat rate share the same formula
-        rate = total_observations / (num_aircraft * duration_seconds)
+        rate = total_observations / (num_aircraft * simulation_duration)
 
         errors = self._validate_reported_metrics(metrics, expected_track_probability=rate, expected_heartbeat_rate=rate)
 
         return StepResult(
             name="Verify Reported Metrics in Flight Blender",
             status=Status.PASS if not errors else Status.FAIL,
-            duration=round(duration_seconds, 2),
+            duration=round(simulation_duration, 2),
             error_message=None if not errors else "; ".join(errors),
         )