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/cli/__init__.py

@@ -65,6 +65,9 @@ def main():
     session_manager = SessionManager(config_path=str(config_path))
     session_manager.config = config
     session_manager.config_path = config_path
+    session_manager.current_output_dir = output_dir
+    session_manager.current_timestamp_str = timestamp_str
+    session_manager.current_start_time = run_timestamp
 
     # Run verification scenarios
     failed = asyncio.run(run_verification_scenarios(config, config_path, session_manager=session_manager))

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

@@ -59,7 +59,7 @@ async def generate_simulated_air_traffic_data(
         self,
         config_path: str | None = None,
         duration: int | None = None,
-    ) -> list[list[FlightObservationSchema]]:
+    ) -> list[FlightObservationSchema]:
         """Generate simulated air traffic data from GeoJSON configuration.
 
         Loads GeoJSON data from the specified config path and uses it to generate
@@ -110,23 +110,21 @@ async def generate_simulated_air_traffic_data_with_latency(
         self,
         config_path: str | None = None,
         duration: int | None = None,
-    ) -> list[list[FlightObservationSchema]]:
+    ) -> 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
 
         modified_flight_observations = []
-        for track_observations in flight_observations:
-            modified_track_observations = []
-            for obs in track_observations:
-                if random.random() < LATENCY_PROBABILITY:
-                    # Simulate latency by removing some observations
-                    if random.random() < 0.5:  # 50% chance to remove observation
-                        continue
-                    # Simulate timestamp shift
-                    shift_seconds = random.uniform(*TIMESTAMP_SHIFT_RANGE_SECONDS)
-                    obs.timestamp += int(shift_seconds * 1000)  # Convert seconds to milliseconds
-                modified_track_observations.append(obs)
-            modified_flight_observations.append(modified_track_observations)
+        for obs in flight_observations:
+            if random.random() < LATENCY_PROBABILITY:
+                # Simulate latency by removing some observations
+                if random.random() < 0.5:  # 50% chance to remove observation
+                    continue
+                # Simulate timestamp shift
+                shift_seconds = random.uniform(*TIMESTAMP_SHIFT_RANGE_SECONDS)
+                obs = obs.model_copy(update={"timestamp": obs.timestamp + int(shift_seconds * 1000)})
+            modified_flight_observations.append(obs)
         return modified_flight_observations

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

@@ -70,19 +70,19 @@ async def generate_bayesian_sim_air_traffic_data(
         self,
         config_path: str | None = None,
         duration: int | None = None,
-    ) -> list[list[FlightObservationSchema]]:
+    ) -> list[FlightObservationSchema]:
         """Run Bayesian scenario and sample aircraft state every second.
 
         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.
+            list[FlightObservationSchema]: flat list of observations across all aircraft,
+            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
@@ -124,7 +124,7 @@ async def generate_bayesian_sim_air_traffic_data(
         logger.info(f"Successfully generated {len(tracks)} tracks.")
 
         base_timestamp = int(arrow.utcnow().timestamp())
-        all_observations: list[list[FlightObservationSchema]] = []
+        all_observations: list[FlightObservationSchema] = []
 
         for track_idx, track in enumerate(tracks):
             icao_address = random_icao()
@@ -136,12 +136,10 @@ async def generate_bayesian_sim_air_traffic_data(
                 sensor_ids=sensor_ids,
                 use_multiple_sensors=use_multiple_sensors,
             )
-            all_observations.append(observations)
+            all_observations.extend(observations)
             logger.info(f"Track {track_idx} ({icao_address}): {len(observations)} observations")
-        logger.info(
-            f"Generated observations for {len(all_observations)} tracks, with {sum(len(obs) for obs in all_observations)} total observations."
-        )
-        logger.info(f"First observation altitude: {all_observations[0][0].icao_address}")
+        logger.info(f"Generated observations for {len(tracks)} tracks, with {len(all_observations)} total observations.")
+        logger.info(f"First observation altitude: {all_observations[0].icao_address}")
 
         return all_observations
 
@@ -213,29 +211,27 @@ async def generate_bayesian_sim_air_traffic_data_with_sensor_latency_issues(
         self,
         config_path: str | None = None,
         duration: int | None = None,
-    ) -> list[list[FlightObservationSchema]]:
+    ) -> list[FlightObservationSchema]:
         """
         This method modifies the retrieved simulation data by changing the timestamp and adding latency to the observed dataset.
         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
 
         modified_flight_observations = []
-        for track_observations in flight_observations:
-            modified_track_observations = []
-            for obs in track_observations:
-                if random.random() < LATENCY_PROBABILITY:
-                    # Simulate latency by removing some observations
-                    if random.random() < 0.5:  # 50% chance to remove observation
-                        continue
-                    # Simulate timestamp shift
-                    shift_seconds = random.uniform(*TIMESTAMP_SHIFT_RANGE_SECONDS)
-                    obs.timestamp += int(shift_seconds * 1000)  # Convert seconds to milliseconds
-                modified_track_observations.append(obs)
-            modified_flight_observations.append(modified_track_observations)
+        for obs in flight_observations:
+            if random.random() < LATENCY_PROBABILITY:
+                # Simulate latency by removing some observations
+                if random.random() < 0.5:  # 50% chance to remove observation
+                    continue
+                # Simulate timestamp shift
+                shift_seconds = random.uniform(*TIMESTAMP_SHIFT_RANGE_SECONDS)
+                obs = obs.model_copy(update={"timestamp": obs.timestamp + int(shift_seconds * 1000)})
+            modified_flight_observations.append(obs)
 
         return modified_flight_observations