diff --git a/config/default.yaml b/config/default.yaml index 2063c92d..861fe270 100644 --- a/config/default.yaml +++ b/config/default.yaml @@ -35,7 +35,7 @@ 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 @@ -43,7 +43,7 @@ blue_sky_air_traffic_simulator_settings: # 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 diff --git a/config/pull_request.yaml b/config/pull_request.yaml index ffea1578..ded32bb6 100644 --- a/config/pull_request.yaml +++ b/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 diff --git a/scenarios/bayesian_sim_air_traffic_data.yaml b/scenarios/bayesian_sim_air_traffic_data.yaml index 2afdfe2e..e226a2e0 100644 --- a/scenarios/bayesian_sim_air_traffic_data.yaml +++ b/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 diff --git a/scenarios/bayesian_sim_air_traffic_data_varying_refresh_rates.yaml b/scenarios/bayesian_sim_air_traffic_data_varying_refresh_rates.yaml index 8ab01517..4c24e25c 100644 --- a/scenarios/bayesian_sim_air_traffic_data_varying_refresh_rates.yaml +++ b/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 diff --git a/scenarios/bluesky_sim_air_traffic_data.yaml b/scenarios/bluesky_sim_air_traffic_data.yaml index 89b007e7..19f13f4e 100644 --- a/scenarios/bluesky_sim_air_traffic_data.yaml +++ b/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 diff --git a/scenarios/bluesky_sim_air_traffic_data_latency_issues.yaml b/scenarios/bluesky_sim_air_traffic_data_latency_issues.yaml index c01e373f..880f168e 100644 --- a/scenarios/bluesky_sim_air_traffic_data_latency_issues.yaml +++ b/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 diff --git a/scenarios/opensky_live_data.yaml b/scenarios/opensky_live_data.yaml index 0b4729e4..ed4fd709 100644 --- a/scenarios/opensky_live_data.yaml +++ b/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 diff --git a/scenarios/openutm_sim_air_traffic_data.yaml b/scenarios/openutm_sim_air_traffic_data.yaml index 6f79c4a4..f4c0376e 100644 --- a/scenarios/openutm_sim_air_traffic_data.yaml +++ b/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 diff --git a/scenarios/sdsp_track.yaml b/scenarios/sdsp_track.yaml index c3013d91..bec7867e 100644 --- a/scenarios/sdsp_track.yaml +++ b/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 diff --git a/scenarios/verify_sdsp_metrics.yaml b/scenarios/verify_sdsp_metrics.yaml index 8c5089a2..169b2b3c 100644 --- a/scenarios/verify_sdsp_metrics.yaml +++ b/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,8 +25,10 @@ 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 @@ -36,4 +36,4 @@ steps: session_id: ${{ steps.generated_sdsp_session_id.result }} action: STOP needs: - - Submit Simulated Air Traffic + - stream_air_traffic diff --git a/src/openutm_verification/cli/__init__.py b/src/openutm_verification/cli/__init__.py index 6bfcb3e2..ceddeacf 100644 --- a/src/openutm_verification/cli/__init__.py +++ b/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)) diff --git a/src/openutm_verification/core/clients/air_traffic/air_traffic_client.py b/src/openutm_verification/core/clients/air_traffic/air_traffic_client.py index 73a68b54..75ce3215 100644 --- a/src/openutm_verification/core/clients/air_traffic/air_traffic_client.py +++ b/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 diff --git a/src/openutm_verification/core/clients/air_traffic/base_client.py b/src/openutm_verification/core/clients/air_traffic/base_client.py index 32be5741..62679126 100644 --- a/src/openutm_verification/core/clients/air_traffic/base_client.py +++ b/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, diff --git a/src/openutm_verification/core/clients/air_traffic/bayesian_air_traffic_client.py b/src/openutm_verification/core/clients/air_traffic/bayesian_air_traffic_client.py index a9af8ff9..afe1ded6 100644 --- a/src/openutm_verification/core/clients/air_traffic/bayesian_air_traffic_client.py +++ b/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 diff --git a/src/openutm_verification/core/clients/air_traffic/blue_sky_client.py b/src/openutm_verification/core/clients/air_traffic/blue_sky_client.py index 34eb47aa..d63f0bbe 100644 --- a/src/openutm_verification/core/clients/air_traffic/blue_sky_client.py +++ b/src/openutm_verification/core/clients/air_traffic/blue_sky_client.py @@ -41,20 +41,20 @@ async def generate_bluesky_sim_air_traffic_data( self, config_path: str | None = None, duration: int | None = None, - ) -> list[list[FlightObservationSchema]]: + ) -> list[FlightObservationSchema]: """Run BlueSky 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_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 @@ -135,19 +135,22 @@ async def generate_bluesky_sim_air_traffic_data( logger.debug(f"{acid:>6} lat={lat:.6f} lon={lon:.6f} alt_mm={altitude_mm:.1f}") - # Convert dict -> list[list[FlightObservationSchema]] with stable ordering - return [results_by_acid[acid] for acid in sorted(results_by_acid.keys())] + # Flatten dict -> list[FlightObservationSchema] with stable ordering + all_obs: list[FlightObservationSchema] = [] + for acid in sorted(results_by_acid.keys()): + all_obs.extend(results_by_acid[acid]) + return all_obs @scenario_step("Generate BlueSky Simulation Air Traffic Data with latency issues") async def generate_bluesky_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 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 @@ -166,7 +169,7 @@ async def generate_bluesky_sim_air_traffic_data_with_sensor_latency_issues( # detached=True prevents UI/event loop from blocking. # Use a temporary directory for BlueSky working files to avoid polluting ~/bluesky # BlueSky's pathfinder.init() auto-creates required subdirs (scenario, plugins, output, cache) - flight_observations: list[list[FlightObservationSchema]] = [] + flight_observations: list[FlightObservationSchema] = [] with tempfile.TemporaryDirectory(prefix="openutm-bluesky-") as tmp_dir: cfg_path = os.path.join(tmp_dir, "settings.cfg") bs.init(mode="sim", detached=True, workdir=tmp_dir, configfile=cfg_path) @@ -229,26 +232,24 @@ async def generate_bluesky_sim_air_traffic_data_with_sensor_latency_issues( logger.debug(f"{acid:>6} lat={lat:.6f} lon={lon:.6f} alt_mm={altitude_mm:.1f}") - # Convert dict -> list[list[FlightObservationSchema]] with stable ordering - flight_observations = [results_by_acid[acid] for acid in sorted(results_by_acid.keys())] + # Flatten dict -> list[FlightObservationSchema] with stable ordering + for acid in sorted(results_by_acid.keys()): + flight_observations.extend(results_by_acid[acid]) # This method modifies the retrieved simulation data by changing the timestamp and adding latency to the observed dataset 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 diff --git a/src/openutm_verification/core/clients/amqp/amqp_client.py b/src/openutm_verification/core/clients/amqp/amqp_client.py index 8aa12e4d..6972682a 100644 --- a/src/openutm_verification/core/clients/amqp/amqp_client.py +++ b/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), diff --git a/src/openutm_verification/core/clients/flight_blender/flight_blender_client.py b/src/openutm_verification/core/clients/flight_blender/flight_blender_client.py index 4888223d..b14903ef 100644 --- a/src/openutm_verification/core/clients/flight_blender/flight_blender_client.py +++ b/src/openutm_verification/core/clients/flight_blender/flight_blender_client.py @@ -1,4 +1,5 @@ import asyncio +import bisect import json import math import random @@ -740,7 +741,7 @@ async def delete_flight_declaration(self, flight_declaration_id: str | None = No @scenario_step("Submit Simulated Air Traffic") async def submit_simulated_air_traffic( self, - observations: list[list[FlightObservationSchema]], + observations: list[FlightObservationSchema], session_ids: list[uuid.UUID] | None = None, single_or_multiple_sensors: str = "single", ) -> dict[str, Any]: @@ -749,13 +750,12 @@ async def submit_simulated_air_traffic( Plays back observations in real-time, submitting one observation per aircraft per second. Args: - observations: List of observation lists, one per aircraft. + observations: Flat list of observations across all aircraft. single_or_multiple_sensors: Whether to use single or multiple sensor IDs. Returns: Dictionary with submission statistics. """ - # Generate a random session ID for this submission session_ids = session_ids or [uuid.uuid4()] if not observations: logger.warning("No air traffic observations to submit.") @@ -766,74 +766,68 @@ async def submit_simulated_air_traffic( "duration_seconds": 0, } - number_of_aircraft = len(observations) + # Pre-convert all timestamps to Arrow objects once and group by aircraft. + # Sorting per-aircraft enables O(log N) bisect lookups in the inner loop. + all_arrow_timestamps: list[arrow.Arrow] = [] + aircraft_data: dict[str, list[tuple[arrow.Arrow, FlightObservationSchema]]] = {} + for obs in observations: + ts = arrow.get(obs.timestamp) + all_arrow_timestamps.append(ts) + aircraft_data.setdefault(obs.icao_address, []).append((ts, obs)) + + number_of_aircraft = len(aircraft_data) logger.debug(f"Submitting simulated air traffic for {number_of_aircraft} aircraft") - # TODO: When single_or_multiple_sensors is "single", we need to aggregate all observations under one sensor ID - # and when it's "multiple", we need to submit them separately as different session_ids, one for each track - session_id = str(session_ids[0]) - # get the start and end point of the simulation - start_times = [] - end_times = [] - for aircraft_obs in observations: - if not aircraft_obs: - continue - start_times.append(arrow.get(aircraft_obs[0].timestamp)) - end_times.append(arrow.get(aircraft_obs[-1].timestamp)) + simulation_start = min(all_arrow_timestamps) + simulation_end = max(all_arrow_timestamps) - if not start_times: - logger.warning("No valid start/end times found in observations.") - return { - "success": True, - "aircraft_count": number_of_aircraft, - "observations_submitted": 0, - "duration_seconds": 0, - "warning": "No valid start/end times found in observations", - } + # Sort per-aircraft observations and extract timestamp lists for bisect + for icao in aircraft_data: + aircraft_data[icao].sort(key=lambda pair: pair[0]) + aircraft_ts: dict[str, list[arrow.Arrow]] = {icao: [ts for ts, _ in pairs] for icao, pairs in aircraft_data.items()} - simulation_start = min(start_times) - simulation_end = max(end_times) + # Only iterate over seconds that actually contain observations + unique_time_slots = sorted({ts.floor("second") for ts in all_arrow_timestamps}) - now = arrow.now() - start_time = now + start_time = arrow.now() observations_submitted = 0 submission_errors = 0 + session_id = str(session_ids[0]) - current_simulation_time = simulation_start - # Loop through the simulation time from start to end, advancing by 1 second each iteration - while current_simulation_time < simulation_end: - # Calculate the corresponding real-world time for the current simulation time - target_real_time = start_time + (current_simulation_time - simulation_start) - # Wait until the current real time reaches the target time - while arrow.now() < target_real_time: - await asyncio.sleep(0.1) - # For each aircraft, find the observation closest to the current simulation time - filtered_observations: list[list[FlightObservationSchema]] = [] - for aircraft_obs in observations: - if not aircraft_obs: - continue - closest_obs = min( - aircraft_obs, - key=lambda obs: abs(arrow.get(obs.timestamp) - current_simulation_time), - ) - filtered_observations.append([closest_obs]) - # Submit the filtered observations for each aircraft to the API - logger.debug(f"Submitting {len(filtered_observations)} air traffic observations") - for filtered_observation in filtered_observations: - endpoint = f"/flight_stream/set_air_traffic/{session_id}" - payload = {"observations": [obs.model_dump(mode="json") for obs in filtered_observation]} - - ScenarioContext.add_air_traffic_data(filtered_observation) - try: - response = await self.post(endpoint, json=payload) - logger.debug(f"Air traffic submission response: {response.text}") - logger.info(f"Observations submitted for aircraft {filtered_observation[0].icao_address} at time {current_simulation_time}") - observations_submitted += 1 - except Exception as e: - logger.error(f"Failed to submit observation: {e}") - submission_errors += 1 - # Advance the simulation time by 1 second - current_simulation_time = current_simulation_time.shift(seconds=1) + for slot_index, current_simulation_time in enumerate(unique_time_slots): + # Pace at 1 second per slot with a single sleep instead of busy-wait + if slot_index > 0: + sleep_seconds = (start_time.shift(seconds=slot_index) - arrow.now()).total_seconds() + if sleep_seconds > 0: + await asyncio.sleep(sleep_seconds) + + # Collect closest observation per aircraft using binary search + slot_observations: list[FlightObservationSchema] = [] + for icao, pairs in aircraft_data.items(): + ts_list = aircraft_ts[icao] + idx = bisect.bisect_left(ts_list, current_simulation_time) + if idx == 0: + closest_obs = pairs[0][1] + elif idx >= len(pairs): + closest_obs = pairs[-1][1] + elif abs(ts_list[idx - 1] - current_simulation_time) <= abs(ts_list[idx] - current_simulation_time): + closest_obs = pairs[idx - 1][1] + else: + closest_obs = pairs[idx][1] + slot_observations.append(closest_obs) + + # Batch all aircraft for this time slot into a single request + endpoint = f"/flight_stream/set_air_traffic/{session_id}" + payload = {"observations": [obs.model_dump(mode="json") for obs in slot_observations]} + ScenarioContext.add_air_traffic_data(slot_observations) + try: + response = await self.post(endpoint, json=payload) + logger.debug(f"Air traffic submission response: {response.text}") + logger.info(f"Submitted {len(slot_observations)} observations at time {current_simulation_time}") + observations_submitted += len(slot_observations) + except Exception as e: + logger.error(f"Failed to submit observations at {current_simulation_time}: {e}") + submission_errors += len(slot_observations) duration_seconds = (arrow.now() - start_time).total_seconds() @@ -843,13 +837,13 @@ 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") async def submit_simulated_air_traffic_at_random_refresh_rates( self, - observations: list[list[FlightObservationSchema]], + observations: list[FlightObservationSchema], session_ids: list[uuid.UUID] | None = None, single_or_multiple_sensors: str = "single", ) -> StepResult: @@ -861,7 +855,7 @@ async def submit_simulated_air_traffic_at_random_refresh_rates( - Large forward jumps (timestamp shifted into the future) Args: - observations: List of observation lists, one per aircraft. + observations: Flat list of observations across all aircraft. session_ids: Optional list of session UUIDs. single_or_multiple_sensors: Whether to use single or multiple sensor IDs. @@ -878,40 +872,28 @@ async def submit_simulated_air_traffic_at_random_refresh_rates( error_message="No air traffic observations provided", ) - number_of_aircraft = len(observations) + # Pre-convert timestamps to Arrow once and group by aircraft + all_arrow_timestamps: list[arrow.Arrow] = [] + obs_by_aircraft: dict[str, list[FlightObservationSchema]] = {} + for obs in observations: + all_arrow_timestamps.append(arrow.get(obs.timestamp)) + obs_by_aircraft.setdefault(obs.icao_address, []).append(obs) + + number_of_aircraft = len(obs_by_aircraft) logger.debug(f"Submitting simulated air traffic (off-nominal timestamps) for {number_of_aircraft} aircraft") session_id = str(session_ids[0]) - # Determine simulation time range - start_times = [] - end_times = [] - for aircraft_obs in observations: - if not aircraft_obs: - continue - start_times.append(arrow.get(aircraft_obs[0].timestamp)) - end_times.append(arrow.get(aircraft_obs[-1].timestamp)) - - if not start_times: - logger.warning("No valid start/end times found in observations.") - return StepResult( - name="Submit Simulated Air Traffic at varying refresh rates", - status=Status.FAIL, - duration=0, - error_message="No valid start/end times found in observations", - ) + # Build unique time slots from original (uncorrupted) timestamps + unique_time_slots = sorted({ts.floor("second") for ts in all_arrow_timestamps}) - simulation_start = min(start_times) - simulation_end = max(end_times) - - now = arrow.now() - start_time = now + start_time = arrow.now() observations_submitted = 0 submission_errors = 0 - # Build a flat list of observations with corrupted timestamps per aircraft - corrupted_observations: list[list[FlightObservationSchema]] = [] - for aircraft_obs in observations: + # Build corrupted observations per aircraft + corrupted_by_aircraft: dict[str, list[FlightObservationSchema]] = {} + for icao, aircraft_obs in obs_by_aircraft.items(): corrupted_aircraft_obs: list[FlightObservationSchema] = [] last_used_timestamp: int | None = None for obs in aircraft_obs: @@ -919,60 +901,67 @@ async def submit_simulated_air_traffic_at_random_refresh_rates( anomaly_roll = random.random() if anomaly_roll < 0.3 and last_used_timestamp is not None: - # 30% chance: stale timestamp — repeat the previous timestamp new_timestamp = last_used_timestamp logger.debug(f"[off-nominal] Stale timestamp for {obs.icao_address}: kept {new_timestamp} instead of {original_timestamp}") elif anomaly_roll < 0.5: - # 20% chance: backward jump — shift timestamp 10-60s into the past offset = random.randint(10, 60) new_timestamp = original_timestamp - offset - logger.debug(f"[off-nominal] Backward jump for {obs.icao_address}: {original_timestamp} -> {new_timestamp} (−{offset}s)") + logger.debug(f"[off-nominal] Backward jump for {obs.icao_address}: {original_timestamp} -> {new_timestamp} (\u2212{offset}s)") elif anomaly_roll < 0.65: - # 15% chance: forward jump — shift timestamp 10-60s into the future offset = random.randint(10, 60) new_timestamp = original_timestamp + offset logger.debug(f"[off-nominal] Forward jump for {obs.icao_address}: {original_timestamp} -> {new_timestamp} (+{offset}s)") else: - # 35% chance: keep the original timestamp (normal) new_timestamp = original_timestamp corrupted_obs = obs.model_copy(update={"timestamp": new_timestamp}) corrupted_aircraft_obs.append(corrupted_obs) last_used_timestamp = new_timestamp - corrupted_observations.append(corrupted_aircraft_obs) - - # Play back observations in real-time using the original simulation timeline - current_simulation_time = simulation_start - while current_simulation_time < simulation_end: - target_real_time = start_time + (current_simulation_time - simulation_start) - while arrow.now() < target_real_time: - await asyncio.sleep(0.1) - - for aircraft_obs in corrupted_observations: - if not aircraft_obs: + corrupted_by_aircraft[icao] = corrupted_aircraft_obs + + # Pre-convert corrupted timestamps to Arrow and sort for bisect lookups + corrupted_data: dict[str, list[tuple[arrow.Arrow, FlightObservationSchema]]] = {} + for icao, corrupted_obs_list in corrupted_by_aircraft.items(): + pairs = [(arrow.get(obs.timestamp), obs) for obs in corrupted_obs_list] + pairs.sort(key=lambda p: p[0]) + corrupted_data[icao] = pairs + corrupted_ts: dict[str, list[arrow.Arrow]] = {icao: [ts for ts, _ in pairs] for icao, pairs in corrupted_data.items()} + + # Play back using unique time slots, pacing at 1 second per slot + for slot_index, current_simulation_time in enumerate(unique_time_slots): + if slot_index > 0: + sleep_seconds = (start_time.shift(seconds=slot_index) - arrow.now()).total_seconds() + if sleep_seconds > 0: + await asyncio.sleep(sleep_seconds) + + slot_observations: list[FlightObservationSchema] = [] + for icao, pairs in corrupted_data.items(): + if not pairs: continue - # Find observation closest to current simulation time using original positions - closest_obs = min( - aircraft_obs, - key=lambda obs: abs(arrow.get(obs.timestamp) - current_simulation_time), - ) - endpoint = f"/flight_stream/set_air_traffic/{session_id}" - payload = {"observations": [closest_obs.model_dump(mode="json")]} - ScenarioContext.add_air_traffic_data([closest_obs]) - try: - response = await self.post(endpoint, json=payload) - logger.debug(f"Air traffic submission response: {response.text}") - logger.info( - f"Off-nominal observation submitted for {closest_obs.icao_address} " - f"at sim time {current_simulation_time} with timestamp {closest_obs.timestamp}" - ) - observations_submitted += 1 - except Exception as e: - logger.error(f"Failed to submit off-nominal observation: {e}") - submission_errors += 1 - - current_simulation_time = current_simulation_time.shift(seconds=1) + ts_list = corrupted_ts[icao] + idx = bisect.bisect_left(ts_list, current_simulation_time) + if idx == 0: + closest_obs = pairs[0][1] + elif idx >= len(pairs): + closest_obs = pairs[-1][1] + elif abs(ts_list[idx - 1] - current_simulation_time) <= abs(ts_list[idx] - current_simulation_time): + closest_obs = pairs[idx - 1][1] + else: + closest_obs = pairs[idx][1] + slot_observations.append(closest_obs) + + endpoint = f"/flight_stream/set_air_traffic/{session_id}" + payload = {"observations": [obs.model_dump(mode="json") for obs in slot_observations]} + ScenarioContext.add_air_traffic_data(slot_observations) + try: + response = await self.post(endpoint, json=payload) + logger.debug(f"Air traffic submission response: {response.text}") + logger.info(f"Submitted {len(slot_observations)} off-nominal observations at sim time {current_simulation_time}") + observations_submitted += len(slot_observations) + except Exception as e: + logger.error(f"Failed to submit off-nominal observations at {current_simulation_time}: {e}") + submission_errors += len(slot_observations) duration_seconds = (arrow.now() - start_time).total_seconds() return StepResult( @@ -982,14 +971,13 @@ async def submit_simulated_air_traffic_at_random_refresh_rates( error_message=None if submission_errors == 0 else f"{submission_errors} submission errors occurred", ) - def _extract_simulation_time_window(self, observations: list[list[FlightObservationSchema]]) -> tuple[arrow.Arrow, arrow.Arrow, float] | None: + def _extract_simulation_time_window(self, observations: list[FlightObservationSchema]) -> tuple[arrow.Arrow, arrow.Arrow, float] | None: """Returns (simulation_start, simulation_end, duration_seconds) or None if no valid observations.""" - start_times = [arrow.get(a[0].timestamp) for a in observations if a] - end_times = [arrow.get(a[-1].timestamp) for a in observations if a] - if not start_times: + if not observations: return None - sim_start = min(start_times) - sim_end = max(end_times) + all_timestamps = [obs.timestamp for obs in observations] + sim_start = arrow.get(min(all_timestamps)) + sim_end = arrow.get(max(all_timestamps)) return sim_start, sim_end, (sim_end - sim_start).total_seconds() + 1.0 def _validate_reported_metrics( @@ -1023,7 +1011,7 @@ def _validate_reported_metrics( return errors @scenario_step("Verify Reported Metrics in Flight Blender") - async def verify_reported_metrics_in_flight_blender(self, observations: list[list[FlightObservationSchema]], session_id: uuid.UUID | None = None): + async def verify_reported_metrics_in_flight_blender(self, observations: list[FlightObservationSchema], session_id: uuid.UUID | None = None): """ Queries the SDSP metrics endpoint and verifies reported values against expected values derived from the Bayesian simulation observations. @@ -1041,9 +1029,9 @@ 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 - logger.info(f"Extracted simulation time window: start={simulation_start}, end={simulation_end}, duration_seconds={duration_seconds}") - logger.info(f"Start time: {simulation_start.isoformat()}, End time: {simulation_end.isoformat()}, Duration (s): {duration_seconds}") + simulation_start, simulation_end, simulation_duration = time_window + logger.info(f"Extracted simulation time window: start={simulation_start}, end={simulation_end}, duration_seconds={simulation_duration}") + logger.info(f"Start time: {simulation_start.isoformat()}, End time: {simulation_end.isoformat()}, Duration (s): {simulation_duration}") session_param = f"session_id={session_id}&" if session_id is not None else "" simulation_end_payload = simulation_end.format("YYYY-MM-DDTHH:mm:ssZ") simulation_start_payload = simulation_start.format("YYYY-MM-DDTHH:mm:ssZ") @@ -1057,7 +1045,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}", ) @@ -1067,13 +1055,14 @@ 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) - rate = total_observations / (num_aircraft * duration_seconds) + icao_addresses = set(obs.icao_address for obs in observations) + num_aircraft = len(icao_addresses) + total_observations = len(observations) + rate = total_observations / (num_aircraft * simulation_duration) expected_track_probability = min(1.0, rate) expected_heartbeat_rate = min(1.0, rate) @@ -1084,7 +1073,7 @@ async def verify_reported_metrics_in_flight_blender(self, observations: list[lis 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), ) diff --git a/src/openutm_verification/core/execution/config_models.py b/src/openutm_verification/core/execution/config_models.py index cdbe047d..d0fd0ad2 100644 --- a/src/openutm_verification/core/execution/config_models.py +++ b/src/openutm_verification/core/execution/config_models.py @@ -37,12 +37,12 @@ class FlightBlenderConfig(StrictBaseModel): class AirTrafficSimulatorSettings(StrictBaseModel): number_of_aircraft: int - simulation_duration: int | str + simulation_duration: int single_or_multiple_sensors: Literal["single", "multiple"] = "single" sensor_ids: list[str] = Field(default_factory=list) session_ids: list[str] = Field(default_factory=list) - @field_validator("simulation_duration") + @field_validator("simulation_duration", mode="before") @classmethod def validate_duration(cls, v: int | str) -> int: return int(parse_duration(v)) @@ -50,7 +50,7 @@ def validate_duration(cls, v: int | str) -> int: class BlueSkyAirTrafficSimulatorSettings(StrictBaseModel): number_of_aircraft: int - simulation_duration_seconds: int + simulation_duration: int single_or_multiple_sensors: Literal["single", "multiple"] = "single" sensor_ids: list[str] = Field(default_factory=list) session_ids: list[str] = Field(default_factory=list) @@ -58,7 +58,7 @@ class BlueSkyAirTrafficSimulatorSettings(StrictBaseModel): class BayesianAirTrafficSimulatorSettings(StrictBaseModel): number_of_aircraft: int - simulation_duration_seconds: int + simulation_duration: int single_or_multiple_sensors: Literal["single", "multiple"] = "single" sensor_ids: list[str] = Field(default_factory=list) session_ids: list[str] = Field(default_factory=list) diff --git a/src/openutm_verification/core/execution/dependencies.py b/src/openutm_verification/core/execution/dependencies.py index 76f9b777..a0b2a4af 100644 --- a/src/openutm_verification/core/execution/dependencies.py +++ b/src/openutm_verification/core/execution/dependencies.py @@ -9,20 +9,6 @@ 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 ( - AirTrafficSettings, - BayesianAirTrafficSettings, - BlueSkyAirTrafficSettings, -) -from openutm_verification.core.clients.air_traffic.bayesian_air_traffic_client import ( - BayesianTrafficClient, -) -from openutm_verification.core.clients.air_traffic.blue_sky_client import ( - BlueSkyClient, -) from openutm_verification.core.clients.amqp import ( AMQPClient, AMQPSettings, @@ -31,10 +17,6 @@ from openutm_verification.core.clients.flight_blender.flight_blender_client import ( FlightBlenderClient, ) -from openutm_verification.core.clients.opensky.base_client import ( - OpenSkySettings, -) -from openutm_verification.core.clients.opensky.opensky_client import OpenSkyClient from openutm_verification.core.execution.config_models import ( AppConfig, DataFiles, @@ -174,25 +156,6 @@ async def flight_blender_client(config: AppConfig, data_files: DataFiles) -> Asy yield fb_client -@dependency(OpenSkyClient) -async def opensky_client(config: AppConfig) -> AsyncGenerator[OpenSkyClient, None]: - """Provides an OpenSkyClient instance for dependency injection.""" - settings = OpenSkySettings.from_config(config.opensky) - async with OpenSkyClient(settings) as client: - yield client - - -@dependency(AirTrafficClient) -async def air_traffic_client(config: AppConfig, data_files: DataFiles) -> AsyncGenerator[AirTrafficClient, None]: - """Provides an AirTrafficClient instance for dependency injection.""" - settings = AirTrafficSettings.from_config( - config.air_traffic_simulator_settings, - trajectory_path=data_files.trajectory, - ) - async with AirTrafficClient(settings) as client: - yield client - - @dependency(SessionManager) async def session_manager() -> AsyncGenerator[SessionManager, None]: yield SessionManager() @@ -203,28 +166,6 @@ async def common_client() -> AsyncGenerator[CommonClient, None]: yield CommonClient() -@dependency(BlueSkyClient) -async def bluesky_client(config: AppConfig, data_files: DataFiles) -> AsyncGenerator[BlueSkyClient, None]: - """Provides a BlueSkyClient instance for dependency injection.""" - settings = BlueSkyAirTrafficSettings.from_config( - config.blue_sky_air_traffic_simulator_settings, - simulation_path=data_files.simulation, - ) - async with BlueSkyClient(settings) as client: - yield client - - -@dependency(BayesianTrafficClient) -async def bayesian_air_traffic_client(config: AppConfig, data_files: DataFiles) -> AsyncGenerator[BayesianTrafficClient, None]: - """Provides a BayesianTrafficClient instance for dependency injection.""" - settings = BayesianAirTrafficSettings.from_config( - config.bayesian_air_traffic_simulator_settings, - simulation_path=data_files.simulation, - ) - async with BayesianTrafficClient(settings) as client: - yield client - - @dependency(AMQPClient) async def amqp_client(config: AppConfig) -> AsyncGenerator[AMQPClient, None]: """Provides an AMQPClient instance for dependency injection.""" diff --git a/src/openutm_verification/core/execution/scenario_runner.py b/src/openutm_verification/core/execution/scenario_runner.py index 018bb812..40422f0c 100644 --- a/src/openutm_verification/core/execution/scenario_runner.py +++ b/src/openutm_verification/core/execution/scenario_runner.py @@ -55,7 +55,7 @@ class ScenarioState: flight_declaration_data: FlightDeclaration | None = None flight_declaration_via_operational_intent_data: FlightDeclarationViaOperationalIntent | None = None telemetry_data: list[RIDAircraftState] | None = None - air_traffic_data: list[list[FlightObservationSchema]] = field(default_factory=list) + air_traffic_data: list[FlightObservationSchema] = field(default_factory=list) added_results: Queue[StepResult[Any]] = field(default_factory=Queue) @property @@ -146,7 +146,7 @@ def set_telemetry_data(cls, data: list[RIDAircraftState]) -> None: def add_air_traffic_data(cls, data: list[FlightObservationSchema]) -> None: state = _scenario_state.get() if state and state.active: - state.air_traffic_data.append(data) + state.air_traffic_data.extend(data) @property def state(self) -> ScenarioState | None: @@ -183,7 +183,7 @@ def telemetry_data(self) -> list[RIDAircraftState] | None: return state.telemetry_data if state else None @property - def air_traffic_data(self) -> list[list[FlightObservationSchema]]: + def air_traffic_data(self) -> list[FlightObservationSchema]: if self._state: return self._state.air_traffic_data state = _scenario_state.get() diff --git a/src/openutm_verification/core/providers/__init__.py b/src/openutm_verification/core/providers/__init__.py index 6ce8c3e9..ae16b21a 100644 --- a/src/openutm_verification/core/providers/__init__.py +++ b/src/openutm_verification/core/providers/__init__.py @@ -4,12 +4,16 @@ """ from .factory import ProviderType, create_provider +from .latency import DataQualityType, drop_observations, shift_timestamps from .opensky_provider import DEFAULT_SWITZERLAND_VIEWPORT from .protocol import AirTrafficProvider __all__ = [ "AirTrafficProvider", "DEFAULT_SWITZERLAND_VIEWPORT", + "DataQualityType", "ProviderType", "create_provider", + "drop_observations", + "shift_timestamps", ] diff --git a/src/openutm_verification/core/providers/bayesian_provider.py b/src/openutm_verification/core/providers/bayesian_provider.py index 4facc680..11e31819 100644 --- a/src/openutm_verification/core/providers/bayesian_provider.py +++ b/src/openutm_verification/core/providers/bayesian_provider.py @@ -8,6 +8,7 @@ from openutm_verification.core.clients.air_traffic.bayesian_air_traffic_client import ( BayesianTrafficClient, ) +from openutm_verification.core.reporting.reporting_models import Status from openutm_verification.simulator.models.flight_data_types import ( FlightObservationSchema, ) @@ -70,29 +71,31 @@ def from_kwargs( async def get_observations( self, duration: int | None = None, - ) -> list[list[FlightObservationSchema]]: + ) -> list[FlightObservationSchema]: """Generate observations using the underlying BayesianTrafficClient. Args: duration: Override duration in seconds. Returns: - List of observation lists per aircraft. + Flat list of observations across all aircraft. """ effective_duration = duration or self._duration settings = BayesianAirTrafficSettings( simulation_config_path=self._config_path, - simulation_duration_seconds=effective_duration, + simulation_duration=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( + step_result = await client.generate_bayesian_sim_air_traffic_data( config_path=self._config_path, duration=effective_duration, ) + if step_result.status == Status.FAIL: + raise RuntimeError(step_result.error_message or "Bayesian generation failed") # Handle case where Bayesian client returns None or empty - return result if result else [] + return step_result.result if step_result.result else [] diff --git a/src/openutm_verification/core/providers/bluesky_provider.py b/src/openutm_verification/core/providers/bluesky_provider.py index a69ecd26..4f109bc4 100644 --- a/src/openutm_verification/core/providers/bluesky_provider.py +++ b/src/openutm_verification/core/providers/bluesky_provider.py @@ -8,6 +8,7 @@ from openutm_verification.core.clients.air_traffic.blue_sky_client import ( BlueSkyClient, ) +from openutm_verification.core.reporting.reporting_models import Status from openutm_verification.simulator.models.flight_data_types import ( FlightObservationSchema, ) @@ -70,27 +71,30 @@ def from_kwargs( async def get_observations( self, duration: int | None = None, - ) -> list[list[FlightObservationSchema]]: + ) -> list[FlightObservationSchema]: """Generate observations using the underlying BlueSkyClient. Args: duration: Override duration in seconds. Returns: - List of observation lists per aircraft. + Flat list of observations across all aircraft. """ effective_duration = duration or self._duration settings = BlueSkyAirTrafficSettings( simulation_config_path=self._config_path, - simulation_duration_seconds=effective_duration, + simulation_duration=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( + step_result = await client.generate_bluesky_sim_air_traffic_data( config_path=self._config_path, duration=effective_duration, ) + if step_result.status == Status.FAIL: + raise RuntimeError(step_result.error_message or "BlueSky generation failed") + return step_result.result diff --git a/src/openutm_verification/core/providers/factory.py b/src/openutm_verification/core/providers/factory.py index b494c72f..6125efc7 100644 --- a/src/openutm_verification/core/providers/factory.py +++ b/src/openutm_verification/core/providers/factory.py @@ -1,14 +1,28 @@ """Factory for creating air traffic providers.""" -from typing import Literal +from enum import StrEnum from .bayesian_provider import BayesianProvider from .bluesky_provider import BlueSkyProvider from .geojson_provider import GeoJSONProvider +from .latency import DataQualityType, LatencyProviderWrapper from .opensky_provider import OpenSkyProvider from .protocol import AirTrafficProvider -ProviderType = Literal["geojson", "bluesky", "bayesian", "opensky"] + +class ProviderType(StrEnum): + GEOJSON = "geojson" + BLUESKY = "bluesky" + BAYESIAN = "bayesian" + OPENSKY = "opensky" + + +# Registry mapping data quality types to their wrapper classes. +# Add new entries here to support additional quality degradation modes +# without modifying the create_provider function. +_QUALITY_WRAPPERS: dict[DataQualityType, type] = { + DataQualityType.LATENCY: LatencyProviderWrapper, +} def create_provider( @@ -20,6 +34,7 @@ def create_provider( sensor_ids: list[str] | None = None, session_ids: list[str] | None = None, viewport: tuple[float, float, float, float] | None = None, + data_quality: DataQualityType = DataQualityType.NOMINAL, **kwargs, ) -> AirTrafficProvider: """Factory function to create providers by name. @@ -32,25 +47,26 @@ def create_provider( sensor_ids: List of sensor UUID strings. session_ids: List of session UUID strings. viewport: Geographic bounds for OpenSky (lat_min, lat_max, lon_min, lon_max). + data_quality: Data quality mode - "nominal" or "latency". **kwargs: Additional provider-specific arguments. Returns: - An AirTrafficProvider instance. + An AirTrafficProvider instance, optionally wrapped with latency simulation. Raises: ValueError: If the provider name is not recognized. """ - providers: dict[str, type] = { - "geojson": GeoJSONProvider, - "bluesky": BlueSkyProvider, - "bayesian": BayesianProvider, - "opensky": OpenSkyProvider, + providers: dict[ProviderType, type] = { + ProviderType.GEOJSON: GeoJSONProvider, + ProviderType.BLUESKY: BlueSkyProvider, + ProviderType.BAYESIAN: BayesianProvider, + ProviderType.OPENSKY: OpenSkyProvider, } if name not in providers: raise ValueError(f"Unknown provider: {name}. Available: {list(providers.keys())}") - return providers[name].from_kwargs( + provider = providers[name].from_kwargs( config_path=config_path, number_of_aircraft=number_of_aircraft, duration=duration, @@ -59,3 +75,10 @@ def create_provider( viewport=viewport, **kwargs, ) + + # Apply quality wrapper if registered for this quality type + wrapper_cls = _QUALITY_WRAPPERS.get(data_quality) + if wrapper_cls: + return wrapper_cls(provider) + + return provider diff --git a/src/openutm_verification/core/providers/geojson_provider.py b/src/openutm_verification/core/providers/geojson_provider.py index 8b59e99d..9cb661aa 100644 --- a/src/openutm_verification/core/providers/geojson_provider.py +++ b/src/openutm_verification/core/providers/geojson_provider.py @@ -10,6 +10,7 @@ from openutm_verification.core.clients.air_traffic.base_client import ( AirTrafficSettings, ) +from openutm_verification.core.reporting.reporting_models import Status if TYPE_CHECKING: from openutm_verification.simulator.models.flight_data_types import ( @@ -73,14 +74,14 @@ def from_kwargs( async def get_observations( self, duration: int | None = None, - ) -> list[list["FlightObservationSchema"]]: + ) -> list["FlightObservationSchema"]: """Generate observations using the underlying AirTrafficClient. Args: duration: Override duration in seconds. Returns: - List of observation lists per aircraft. + Flat list of observations across all aircraft. """ effective_duration = duration or self._duration @@ -93,7 +94,10 @@ async def get_observations( ) async with AirTrafficClient(settings) as client: - return await client.generate_simulated_air_traffic_data( + step_result = await client.generate_simulated_air_traffic_data( config_path=self._config_path, duration=effective_duration, ) + if step_result.status == Status.FAIL: + raise RuntimeError(step_result.error_message or "GeoJSON generation failed") + return step_result.result diff --git a/src/openutm_verification/core/providers/latency.py b/src/openutm_verification/core/providers/latency.py new file mode 100644 index 00000000..bdfbf090 --- /dev/null +++ b/src/openutm_verification/core/providers/latency.py @@ -0,0 +1,157 @@ +"""Latency simulation utilities for air traffic providers. + +Applies realistic sensor latency effects to observation data: +- Random observation drops (simulating missed readings) +- Timestamp shifts (simulating delayed sensor data) + +These effects are consistent across all providers (GeoJSON, BlueSky, Bayesian) +and match the original per-client implementations. +""" + +from __future__ import annotations + +import random +from enum import StrEnum +from typing import TYPE_CHECKING + +from loguru import logger + +if TYPE_CHECKING: + from openutm_verification.core.providers.protocol import AirTrafficProvider + from openutm_verification.simulator.models.flight_data_types import ( + FlightObservationSchema, + ) + +# Default latency parameters — consistent with the original per-client implementations +LATENCY_PROBABILITY = 0.1 # 10% chance per observation +TIMESTAMP_SHIFT_RANGE_SECONDS = (-1, 2.5) # Shift range in seconds + + +class DataQualityType(StrEnum): + """Data quality modes for air traffic observations. + + Each quality type applies independent degradation effects to observation data. + StrEnum allows direct comparison with string values (e.g., ``quality == "latency"``). + """ + + NOMINAL = "nominal" + LATENCY = "latency" + + +def drop_observations( + observations: list[FlightObservationSchema], + probability: float = LATENCY_PROBABILITY, +) -> tuple[list[FlightObservationSchema], int]: + """Randomly drop observations to simulate missed sensor readings. + + Each observation has an independent probability of being removed. + + Args: + observations: Flat list of observations across all aircraft. + probability: Probability each observation is dropped (0.0-1.0). + + Returns: + Tuple of (modified observation list, total observations dropped). + """ + kept = [] + total_dropped = 0 + for obs in observations: + if random.random() < probability: + total_dropped += 1 + else: + kept.append(obs) + return kept, total_dropped + + +def shift_timestamps( + observations: list[FlightObservationSchema], + probability: float = LATENCY_PROBABILITY, + shift_range: tuple[float, float] = TIMESTAMP_SHIFT_RANGE_SECONDS, +) -> tuple[list[FlightObservationSchema], int]: + """Randomly shift observation timestamps to simulate delayed sensor data. + + Each observation has an independent probability of having its timestamp + shifted by a random amount within the given range (in seconds). + + Args: + observations: Flat list of observations across all aircraft. + probability: Probability each observation is shifted (0.0-1.0). + shift_range: Range (min, max) for timestamp shifts in seconds. + + Returns: + Tuple of (modified observation list, total observations shifted). + """ + new_observations = [] + total_shifted = 0 + for obs in observations: + if random.random() < probability: + shift_seconds = random.uniform(*shift_range) + obs = obs.model_copy(update={"timestamp": obs.timestamp + int(shift_seconds)}) + total_shifted += 1 + new_observations.append(obs) + return new_observations, total_shifted + + +def apply_latency( + observations: list[FlightObservationSchema], + *, + latency_probability: float = LATENCY_PROBABILITY, + timestamp_shift_range: tuple[float, float] = TIMESTAMP_SHIFT_RANGE_SECONDS, +) -> list[FlightObservationSchema]: + """Apply simulated sensor latency effects to observations. + + Composes independent quality degradation effects: + 1. Random observation drops (simulating missed readings) + 2. Random timestamp shifts (simulating delayed sensor data) + + Each effect is applied independently, making it straightforward to add + new quality degradation types in the future. + + Args: + observations: Flat list of observations across all aircraft. + latency_probability: Base probability for each effect (0.0-1.0). + Split equally between drops and shifts to maintain overall 50/50 ratio. + timestamp_shift_range: Range (min, max) for timestamp shifts in seconds. + + Returns: + Modified observation list with latency effects applied. + """ + drop_prob = latency_probability / 2 + shift_prob = latency_probability / 2 + + result, total_dropped = drop_observations(observations, probability=drop_prob) + result, total_shifted = shift_timestamps(result, probability=shift_prob, shift_range=timestamp_shift_range) + + logger.info(f"Latency simulation applied: {total_dropped} observations dropped, {total_shifted} timestamps shifted") + return result + + +class LatencyProviderWrapper: + """Wraps an air traffic provider to add latency simulation to its observations. + + This decorator pattern preserves the original provider's name while applying + latency post-processing to the generated observations. + """ + + def __init__(self, inner: AirTrafficProvider): + self._inner = inner + + @property + def name(self) -> str: + """Provider identifier (passes through to inner provider).""" + return self._inner.name + + async def get_observations( + self, + duration: int | None = None, + ) -> list[FlightObservationSchema]: + """Get observations with latency effects applied. + + Args: + duration: Override duration in seconds. + + Returns: + Observation list with simulated latency effects. + """ + observations = await self._inner.get_observations(duration=duration) + return apply_latency(observations) diff --git a/src/openutm_verification/core/providers/opensky_provider.py b/src/openutm_verification/core/providers/opensky_provider.py index 64ba48d7..84790a6e 100644 --- a/src/openutm_verification/core/providers/opensky_provider.py +++ b/src/openutm_verification/core/providers/opensky_provider.py @@ -60,14 +60,14 @@ def from_kwargs( async def get_observations( self, duration: int | None = None, - ) -> list[list["FlightObservationSchema"]]: + ) -> list["FlightObservationSchema"]: """Fetch observations from OpenSky Network. Args: duration: Override duration in seconds (currently single fetch). Returns: - List containing a single observation list (all aircraft in one batch). + Flat list of observations across all aircraft. Returns empty list if no data available. """ # Get OpenSky config from application settings @@ -80,9 +80,8 @@ async def get_observations( ) async with OpenSkyClient(settings) as client: - observations = await client.fetch_data() + observations = (await client.fetch_data()).result if observations is None: return [] - # Wrap flat list in outer list for interface consistency - # OpenSky returns all aircraft in a single list, not grouped by aircraft - return [observations] + # OpenSky returns all aircraft in a single flat list + return observations diff --git a/src/openutm_verification/core/providers/protocol.py b/src/openutm_verification/core/providers/protocol.py index 0a0f4a84..d380420e 100644 --- a/src/openutm_verification/core/providers/protocol.py +++ b/src/openutm_verification/core/providers/protocol.py @@ -31,7 +31,7 @@ def name(self) -> str: async def get_observations( self, duration: int | None = None, - ) -> list[list["FlightObservationSchema"]]: + ) -> list["FlightObservationSchema"]: """Get observation batches for the configured duration. Args: @@ -39,7 +39,6 @@ async def get_observations( If None, uses provider's default configuration. Returns: - List of observation lists - outer list is per aircraft/track, - inner list is the time series of observations. + Flat list of flight observations across all aircraft/tracks. """ ... diff --git a/src/openutm_verification/core/reporting/reporting_models.py b/src/openutm_verification/core/reporting/reporting_models.py index bb829b7f..42b29572 100644 --- a/src/openutm_verification/core/reporting/reporting_models.py +++ b/src/openutm_verification/core/reporting/reporting_models.py @@ -60,7 +60,7 @@ class ScenarioResult(BaseModel): flight_declaration_data: FlightDeclaration | None = None flight_declaration_via_operational_intent_data: FlightDeclarationViaOperationalIntent | None = None telemetry_data: list[RIDAircraftState] | None = None - air_traffic_data: list[list[FlightObservationSchema]] | None = None + air_traffic_data: list[FlightObservationSchema] | None = None visualization_2d_path: str | None = None visualization_3d_path: str | None = None docs: str | None = None diff --git a/src/openutm_verification/core/reporting/visualize_flight.py b/src/openutm_verification/core/reporting/visualize_flight.py index f549e0da..3307bf67 100644 --- a/src/openutm_verification/core/reporting/visualize_flight.py +++ b/src/openutm_verification/core/reporting/visualize_flight.py @@ -32,7 +32,7 @@ def visualize_flight_path_2d( telemetry_data: list[RIDAircraftState], declaration_data: dict, output_html_path: Path, - air_traffic_data: list[list[FlightObservationSchema]] | None = None, + air_traffic_data: list[FlightObservationSchema] | None = None, ): """ Creates an interactive 2D map from flight telemetry and declaration data. @@ -41,7 +41,7 @@ def visualize_flight_path_2d( telemetry_data (dict): The flight telemetry data as a dictionary. declaration_data (dict): The flight declaration data as a dictionary. output_html_path (Path): The full path where the output HTML map will be saved. - air_traffic_data (list[list[FlightObservationSchema]] | None): Optional air traffic data from simulators. + air_traffic_data (list[FlightObservationSchema] | None): Optional air traffic data from simulators. """ logger.info("Starting 2D flight path visualization") @@ -120,48 +120,45 @@ def visualize_flight_path_2d( def _reorganize_air_traffic_by_aircraft( - air_traffic_data: list[list[FlightObservationSchema]], + air_traffic_data: list[FlightObservationSchema], ) -> dict[str, list]: """ - Reorganizes air traffic data from timestamp-based to aircraft-based grouping. + Reorganizes air traffic data into aircraft-based grouping. - The input data may be organized as list of timestamps, where each timestamp - contains observations from multiple aircraft. This function reorganizes it - into a dict keyed by ICAO address with all observations for that aircraft. + Groups the flat observation list by ICAO address. Args: - air_traffic_data: Air traffic data (may be organized by timestamp or aircraft). + air_traffic_data: Flat list of air traffic observations. Returns: Dict mapping ICAO address to list of observations for that aircraft. """ aircraft_tracks: dict[str, list] = {} - for observations in air_traffic_data: - for obs in observations: - # Handle both dict and Pydantic model - if isinstance(obs, dict): - icao = obs.get("icao_address", "UNKNOWN") - else: - icao = obs.icao_address + for obs in air_traffic_data: + # Handle both dict and Pydantic model + if isinstance(obs, dict): + icao = obs.get("icao_address", "UNKNOWN") + else: + icao = obs.icao_address - if icao not in aircraft_tracks: - aircraft_tracks[icao] = [] - aircraft_tracks[icao].append(obs) + if icao not in aircraft_tracks: + aircraft_tracks[icao] = [] + aircraft_tracks[icao].append(obs) return aircraft_tracks def _add_air_traffic_to_2d_map( flight_map: folium.Map, - air_traffic_data: list[list[FlightObservationSchema]], + air_traffic_data: list[FlightObservationSchema], ) -> None: """ Adds airplane/air traffic paths to the 2D map with distinct colors. Args: flight_map: The Folium map to add paths to. - air_traffic_data: Air traffic data as list of aircraft, each with list of observations. + air_traffic_data: Flat list of air traffic observations. """ # Reorganize data by aircraft ICAO address aircraft_tracks = _reorganize_air_traffic_by_aircraft(air_traffic_data) @@ -386,7 +383,7 @@ def _create_airplane_path_group(projected_path: list[tuple[float, float, float]] def _add_air_traffic_to_3d_scene( scene: three.Scene, - air_traffic_data: list[list[FlightObservationSchema]], + air_traffic_data: list[FlightObservationSchema], project_fn, ) -> list[tuple[float, float, float]]: """ @@ -394,7 +391,7 @@ def _add_air_traffic_to_3d_scene( Args: scene: The pythreejs Scene to add paths to. - air_traffic_data: Air traffic data as list of aircraft, each with list of observations. + air_traffic_data: Flat list of air traffic observations. project_fn: Projection function (lon, lat, alt) -> (x, y, z). Returns: @@ -444,7 +441,7 @@ def visualize_flight_path_3d( telemetry_data: list[RIDAircraftState], declaration_data: dict, output_html_path: Path, - air_traffic_data: list[list[FlightObservationSchema]] | None = None, + air_traffic_data: list[FlightObservationSchema] | None = None, ): """Creates an interactive 3D visualization of the flight path and geofence. diff --git a/src/openutm_verification/core/steps/air_traffic_step.py b/src/openutm_verification/core/steps/air_traffic_step.py index 5b44882b..a7a47069 100644 --- a/src/openutm_verification/core/steps/air_traffic_step.py +++ b/src/openutm_verification/core/steps/air_traffic_step.py @@ -4,9 +4,44 @@ replacing the multiple provider-specific steps with a unified interface. """ +from __future__ import annotations + +from loguru import logger + +from openutm_verification.core.execution.config_models import get_settings +from openutm_verification.core.execution.dependency_resolution import CONTEXT from openutm_verification.core.execution.scenario_runner import scenario_step -from openutm_verification.core.providers import ProviderType, create_provider -from openutm_verification.core.streamers import StreamResult, TargetType, create_streamer +from openutm_verification.core.providers import DataQualityType, ProviderType, create_provider +from openutm_verification.core.streamers import RefreshModeType, StreamResult, TargetType, create_streamer + + +def _get_data_file_path(field_name: str) -> str | None: + """Get a data file path from the current suite context or global config. + + Checks the current CONTEXT for suite-specific overrides first, + then falls back to the global config data_files. + + Args: + field_name: Data file field name (e.g., "trajectory", "simulation"). + + Returns: + The resolved file path, or None if not configured. + """ + try: + context = CONTEXT.get() + suite_scenario = context.get("suite_scenario") if context else None + if suite_scenario and hasattr(suite_scenario, field_name): + value = getattr(suite_scenario, field_name, None) + if value: + return value + except (LookupError, AttributeError): + pass + + try: + config = get_settings() + return getattr(config.data_files, field_name, None) + except Exception: + return None class AirTrafficStepClient: @@ -14,7 +49,8 @@ class AirTrafficStepClient: This client wraps the provider/streamer architecture to expose a single scenario step that can handle all air traffic generation and streaming - operations. + operations. When step arguments are not provided, defaults are read + from the application configuration. """ async def __aenter__(self): @@ -23,19 +59,106 @@ async def __aenter__(self): async def __aexit__(self, exc_type, exc_val, exc_tb): pass + def _apply_config_defaults( + self, + provider: ProviderType, + *, + duration: int | None, + config_path: str | None, + number_of_aircraft: int | None, + sensor_ids: list[str] | None, + session_ids: list[str] | None, + ) -> tuple[int, str | None, int | None, list[str] | None, list[str] | None]: + """Fill in missing parameters from application configuration. + + Reads defaults from the provider-specific config section when + step arguments are not explicitly provided. + + Args: + provider: The provider type to read defaults for. + duration: Explicit duration or None. + config_path: Explicit config path or None. + number_of_aircraft: Explicit count or None. + sensor_ids: Explicit sensor IDs or None. + session_ids: Explicit session IDs or None. + + Returns: + Tuple of (duration, config_path, number_of_aircraft, sensor_ids, session_ids) + with defaults filled in. + """ + try: + app_config = get_settings() + except Exception: + logger.debug("Could not read application config for defaults, using step arguments only.") + resolved_duration = duration if duration is not None else 30 + return (resolved_duration, config_path, number_of_aircraft, sensor_ids, session_ids) + + try: + if provider == ProviderType.GEOJSON: + sim = app_config.air_traffic_simulator_settings + if duration is None: + duration = sim.simulation_duration + if number_of_aircraft is None: + number_of_aircraft = sim.number_of_aircraft + if sensor_ids is None and sim.sensor_ids: + sensor_ids = sim.sensor_ids + if session_ids is None and sim.session_ids: + session_ids = sim.session_ids + if config_path is None: + config_path = _get_data_file_path("trajectory") + + elif provider == ProviderType.BLUESKY: + sim = app_config.blue_sky_air_traffic_simulator_settings + if duration is None: + duration = sim.simulation_duration + if number_of_aircraft is None: + number_of_aircraft = sim.number_of_aircraft + if sensor_ids is None and sim.sensor_ids: + sensor_ids = sim.sensor_ids + if session_ids is None and sim.session_ids: + session_ids = sim.session_ids + if config_path is None: + config_path = _get_data_file_path("simulation") + + elif provider == ProviderType.BAYESIAN: + sim = app_config.bayesian_air_traffic_simulator_settings + if duration is None: + duration = sim.simulation_duration + if number_of_aircraft is None: + number_of_aircraft = sim.number_of_aircraft + if sensor_ids is None and sim.sensor_ids: + sensor_ids = sim.sensor_ids + if session_ids is None and sim.session_ids: + session_ids = sim.session_ids + + elif provider == ProviderType.OPENSKY: + pass # OpenSky reads its own config in the provider + except Exception: + logger.debug("Could not read application config for defaults, using step arguments only.") + + # Ensure duration has a value + if duration is None: + duration = 30 + + return (duration, config_path, number_of_aircraft, sensor_ids, session_ids) + @scenario_step("Stream Air Traffic") async def stream_air_traffic( self, provider: ProviderType, - duration: int, - target: TargetType = "flight_blender", + duration: int | None = None, + target: TargetType = TargetType.FLIGHT_BLENDER, *, - # Provider settings (optional overrides) + # Provider settings (optional overrides — defaults read from config) config_path: str | None = None, number_of_aircraft: int | None = None, sensor_ids: list[str] | None = None, session_ids: list[str] | None = None, viewport: tuple[float, float, float, float] | None = None, + # Data quality mode + data_quality: DataQualityType = DataQualityType.NOMINAL, + # Streamer settings + refresh_mode: RefreshModeType = RefreshModeType.NORMAL, ) -> StreamResult: """Stream air traffic data from a provider to a target system. @@ -43,15 +166,20 @@ async def stream_air_traffic( Supports synthetic data generation (GeoJSON, BlueSky, Bayesian) and live data fetching (OpenSky Network). + When arguments are not provided, defaults are read from the application + configuration (e.g., air_traffic_simulator_settings, data_files). + Args: provider: Data source - geojson, bluesky, bayesian, or opensky. - duration: Streaming duration in seconds. + duration: Streaming duration in seconds (defaults from config). target: Delivery target - flight_blender, amqp, or none (default: flight_blender). - config_path: Path to configuration file (provider-specific). - number_of_aircraft: Number of aircraft to simulate. - sensor_ids: Sensor UUIDs for observations. - session_ids: Session UUIDs for grouping. + config_path: Path to configuration file (provider-specific, defaults from data_files). + number_of_aircraft: Number of aircraft to simulate (defaults from config). + sensor_ids: Sensor UUIDs for observations (defaults from config). + session_ids: Session UUIDs for grouping (defaults from config). viewport: Geographic bounds for OpenSky (lat_min, lat_max, lon_min, lon_max). + data_quality: Data quality mode - "nominal" or "latency" for simulated sensor issues. + refresh_mode: Submission mode - "normal" or "varying" for corrupted timestamps. Returns: StreamResult with success status, counts, and optionally the observations. @@ -63,7 +191,28 @@ async def stream_air_traffic( duration: 30 target: flight_blender config_path: config/bern/trajectory.geojson + + # Minimal form (reads all defaults from config): + - step: Stream Air Traffic + arguments: + provider: bayesian + + # With latency simulation: + - step: Stream Air Traffic + arguments: + provider: bluesky + data_quality: latency """ + # Apply config defaults for any unset parameters + duration, config_path, number_of_aircraft, sensor_ids, session_ids = self._apply_config_defaults( + provider, + duration=duration, + config_path=config_path, + number_of_aircraft=number_of_aircraft, + sensor_ids=sensor_ids, + session_ids=session_ids, + ) + # Build provider from arguments provider_instance = create_provider( name=provider, @@ -73,16 +222,24 @@ async def stream_air_traffic( sensor_ids=sensor_ids, session_ids=session_ids, viewport=viewport, + data_quality=data_quality, ) # Build streamer (or null streamer for target=none) streamer_instance = create_streamer( name=target, session_ids=session_ids, + refresh_mode=refresh_mode, ) # Execute streaming - return await streamer_instance.stream_from_provider( + stream_result = await streamer_instance.stream_from_provider( provider=provider_instance, duration_seconds=duration, ) + + # If streaming failed, raise so the scenario step is marked as failed + if not stream_result.success: + raise RuntimeError(f"Air traffic streaming failed: {stream_result.errors}") + + return stream_result diff --git a/src/openutm_verification/core/streamers/__init__.py b/src/openutm_verification/core/streamers/__init__.py index beabeb31..c57ccce0 100644 --- a/src/openutm_verification/core/streamers/__init__.py +++ b/src/openutm_verification/core/streamers/__init__.py @@ -4,10 +4,12 @@ """ from .factory import TargetType, create_streamer +from .flight_blender_streamer import RefreshModeType from .protocol import AirTrafficStreamer, StreamResult __all__ = [ "AirTrafficStreamer", + "RefreshModeType", "StreamResult", "TargetType", "create_streamer", diff --git a/src/openutm_verification/core/streamers/amqp_streamer.py b/src/openutm_verification/core/streamers/amqp_streamer.py index 7bc684c7..7dee9dc8 100644 --- a/src/openutm_verification/core/streamers/amqp_streamer.py +++ b/src/openutm_verification/core/streamers/amqp_streamer.py @@ -56,7 +56,7 @@ async def stream_from_provider( # Get observations from provider observations = await provider.get_observations(duration=duration_seconds) - total_observations = sum(len(batch) for batch in observations) + total_observations = len(observations) return StreamResult( success=True, @@ -64,7 +64,7 @@ async def stream_from_provider( target=self.name, duration_seconds=duration_seconds, total_observations=total_observations, - total_batches=len(observations), + total_batches=1, errors=["AMQP streaming not fully implemented - data collected but not sent"], observations=observations, ) diff --git a/src/openutm_verification/core/streamers/factory.py b/src/openutm_verification/core/streamers/factory.py index a1b14a2d..5d06f47f 100644 --- a/src/openutm_verification/core/streamers/factory.py +++ b/src/openutm_verification/core/streamers/factory.py @@ -1,19 +1,24 @@ """Factory for creating air traffic streamers.""" -from typing import Literal +from enum import StrEnum from .amqp_streamer import AMQPStreamer -from .flight_blender_streamer import FlightBlenderStreamer +from .flight_blender_streamer import FlightBlenderStreamer, RefreshModeType from .null_streamer import NullStreamer from .protocol import AirTrafficStreamer -TargetType = Literal["flight_blender", "amqp", "none"] + +class TargetType(StrEnum): + FLIGHT_BLENDER = "flight_blender" + AMQP = "amqp" + NONE = "none" def create_streamer( name: TargetType, *, session_ids: list[str] | None = None, + refresh_mode: RefreshModeType = RefreshModeType.NORMAL, **kwargs, ) -> AirTrafficStreamer: """Factory function to create streamers by name. @@ -21,6 +26,7 @@ def create_streamer( Args: name: Target type - flight_blender, amqp, or none. session_ids: Optional list of session UUID strings (for flight_blender). + refresh_mode: Submission mode for flight_blender - "normal" or "varying". **kwargs: Additional streamer-specific arguments. Returns: @@ -29,13 +35,13 @@ def create_streamer( Raises: ValueError: If the streamer name is not recognized. """ - streamers: dict[str, type] = { - "flight_blender": FlightBlenderStreamer, - "amqp": AMQPStreamer, - "none": NullStreamer, + streamers: dict[TargetType, type] = { + TargetType.FLIGHT_BLENDER: FlightBlenderStreamer, + TargetType.AMQP: AMQPStreamer, + TargetType.NONE: NullStreamer, } if name not in streamers: raise ValueError(f"Unknown streamer: {name}. Available: {list(streamers.keys())}") - return streamers[name].from_kwargs(session_ids=session_ids, **kwargs) + return streamers[name].from_kwargs(session_ids=session_ids, refresh_mode=refresh_mode, **kwargs) diff --git a/src/openutm_verification/core/streamers/flight_blender_streamer.py b/src/openutm_verification/core/streamers/flight_blender_streamer.py index e435cf9a..e862f979 100644 --- a/src/openutm_verification/core/streamers/flight_blender_streamer.py +++ b/src/openutm_verification/core/streamers/flight_blender_streamer.py @@ -7,14 +7,17 @@ from __future__ import annotations import uuid +from enum import StrEnum from typing import TYPE_CHECKING from loguru import logger +from openutm_verification.auth.providers import get_auth_provider from openutm_verification.core.clients.flight_blender.flight_blender_client import ( FlightBlenderClient, ) from openutm_verification.core.execution.config_models import get_settings +from openutm_verification.core.reporting.reporting_models import Status, StepResult from .protocol import StreamResult @@ -22,20 +25,39 @@ from openutm_verification.core.providers.protocol import AirTrafficProvider +class RefreshModeType(StrEnum): + """Submission mode for Flight Blender streaming. + + Controls how observations are submitted to the Flight Blender API. + StrEnum allows direct comparison with string values. + """ + + NORMAL = "normal" + VARYING = "varying" + + class FlightBlenderStreamer: """Streamer that sends observations to Flight Blender via HTTP API. - Wraps the existing FlightBlenderClient's submit_simulated_air_traffic - method to provide the unified streaming interface. + Wraps the existing FlightBlenderClient's submit methods to provide + the unified streaming interface. Supports both normal and varying + refresh rate submission modes. """ - def __init__(self, session_ids: list[uuid.UUID] | None = None): + def __init__( + self, + session_ids: list[uuid.UUID] | None = None, + refresh_mode: RefreshModeType = "normal", + ): """Initialize the Flight Blender streamer. Args: session_ids: Optional list of session UUIDs for grouping observations. + refresh_mode: Submission mode - "normal" for standard real-time playback, + "varying" for corrupted timestamps simulating malfunctioning sensors. """ self._session_ids = session_ids + self._refresh_mode = refresh_mode @property def name(self) -> str: @@ -46,12 +68,14 @@ def name(self) -> str: def from_kwargs( cls, session_ids: list[str] | None = None, + refresh_mode: RefreshModeType = "normal", **_kwargs, ) -> "FlightBlenderStreamer": """Factory method to create streamer from configuration. Args: session_ids: Optional list of session UUID strings. + refresh_mode: Submission mode - "normal" or "varying". """ parsed_ids = None if session_ids: @@ -59,7 +83,7 @@ def from_kwargs( parsed_ids = [uuid.UUID(sid) for sid in session_ids] except ValueError: logger.warning("Invalid session ID format detected, will auto-generate. Ensure session IDs are valid UUIDs.") - return cls(session_ids=parsed_ids) + return cls(session_ids=parsed_ids, refresh_mode=refresh_mode) def _make_result( self, @@ -81,7 +105,7 @@ def _make_result( total_observations=total_observations, total_batches=total_batches, errors=errors or [], - observations=observations or [], + observations=observations if observations is not None else None, ) async def stream_from_provider( @@ -91,8 +115,10 @@ async def stream_from_provider( ) -> StreamResult: """Stream observations from provider to Flight Blender. - Gets observations from the provider, then submits them to Flight Blender - in real-time playback mode (one observation per second per aircraft). + Gets observations from the provider, then submits them to Flight Blender. + In "normal" mode, submits using standard real-time playback. + In "varying" mode, uses the client's varying-refresh submission method + which applies its own timestamp anomalies to simulate malfunctioning sensors. Args: provider: The air traffic provider to get observations from. @@ -125,45 +151,53 @@ async def stream_from_provider( observations=observations, ) - username = config.flight_blender.auth.username - password = config.flight_blender.auth.password - - if not username or not password: - error_msg = ( - "Flight Blender credentials are not configured. " - "Please set 'flight_blender.auth.username' and " - "'flight_blender.auth.password' in your configuration." - ) - logger.error(error_msg) - return self._make_result( - success=False, - provider_name=provider.name, - duration_seconds=duration_seconds, - errors=[error_msg], - observations=observations, - ) + auth_provider = get_auth_provider(config.flight_blender.auth) + credentials = auth_provider.get_cached_credentials( + audience=config.flight_blender.auth.audience, + scopes=config.flight_blender.auth.scopes, + ) try: async with FlightBlenderClient( base_url=config.flight_blender.url, - credentials={"username": username, "password": password}, + credentials=credentials, ) as client: - result = await client.submit_simulated_air_traffic( + # Choose submission method based on refresh mode. + # The varying-refresh client method applies its own timestamp + # anomalies, so we always pass the original observations. + if self._refresh_mode == "varying": + submit_fn = client.submit_simulated_air_traffic_at_random_refresh_rates + else: + submit_fn = client.submit_simulated_air_traffic + + step_result = await submit_fn( observations=observations, session_ids=self._session_ids, ) + # The @scenario_step decorator wraps the return in StepResult. + # Extract success from the inner result. + if isinstance(step_result, StepResult): + if step_result.status == Status.FAIL: + raise RuntimeError(step_result.error_message or "Submission failed") + raw = step_result.result + success = raw.get("success", False) if isinstance(raw, dict) else True + elif isinstance(step_result, dict): + success = step_result.get("success", False) + else: + success = True + return self._make_result( - success=result.get("success", False), + success=success, provider_name=provider.name, duration_seconds=duration_seconds, - total_observations=sum(len(batch) for batch in observations), - total_batches=len(observations), + total_observations=len(observations), + total_batches=1, observations=observations, ) except Exception as e: - logger.error(f"Flight Blender streaming failed: {e}") + logger.exception(f"Flight Blender streaming failed: {e}") return self._make_result( success=False, provider_name=provider.name, diff --git a/src/openutm_verification/core/streamers/null_streamer.py b/src/openutm_verification/core/streamers/null_streamer.py index ea7cc24a..1826292d 100644 --- a/src/openutm_verification/core/streamers/null_streamer.py +++ b/src/openutm_verification/core/streamers/null_streamer.py @@ -49,7 +49,7 @@ async def stream_from_provider( """ observations = await provider.get_observations(duration=duration_seconds) - total_observations = sum(len(batch) for batch in observations) + total_observations = len(observations) return StreamResult( success=True, @@ -57,7 +57,7 @@ async def stream_from_provider( target=self.name, duration_seconds=duration_seconds, total_observations=total_observations, - total_batches=len(observations), + total_batches=1, errors=[], observations=observations, ) diff --git a/src/openutm_verification/core/streamers/protocol.py b/src/openutm_verification/core/streamers/protocol.py index 21b1bd51..138c0d60 100644 --- a/src/openutm_verification/core/streamers/protocol.py +++ b/src/openutm_verification/core/streamers/protocol.py @@ -33,7 +33,7 @@ class StreamResult: errors: list[str] = field(default_factory=list) # For downstream steps - stores the observations that were streamed - observations: list[list["FlightObservationSchema"]] | None = None + observations: list["FlightObservationSchema"] | None = None @runtime_checkable diff --git a/src/openutm_verification/scenarios/common.py b/src/openutm_verification/scenarios/common.py index e667947c..5ba3f093 100644 --- a/src/openutm_verification/scenarios/common.py +++ b/src/openutm_verification/scenarios/common.py @@ -1,4 +1,5 @@ import json +import random import uuid from pathlib import Path @@ -69,3 +70,9 @@ def get_geo_fence_path(geo_fence_filename: str) -> str: async def generate_uuid() -> str: """Generates a random UUID.""" return str(uuid.uuid4()) + + +@scenario_step("Generate Random Number") +async def generate_random_number(min: int = 0, max: int = 5) -> int: + """Generates a random number.""" + return random.randint(min, max) diff --git a/src/openutm_verification/server/introspection.py b/src/openutm_verification/server/introspection.py index 0f89fe56..4d669807 100644 --- a/src/openutm_verification/server/introspection.py +++ b/src/openutm_verification/server/introspection.py @@ -1,7 +1,9 @@ import inspect import re +import types +import typing from enum import Enum -from typing import Any, Dict, Type +from typing import Any, Dict, Type, Union, get_type_hints from openutm_verification.core.execution.dependency_resolution import DEPENDENCIES @@ -13,6 +15,16 @@ def _get_type_info(annotation: Any) -> tuple[str, bool, list[dict[str, Any]] | N options = None if annotation != inspect.Parameter.empty: + # Check for union types (e.g. ProviderType | None) and extract the enum member + origin = typing.get_origin(annotation) + if origin is Union or isinstance(annotation, types.UnionType): + for arg in typing.get_args(annotation): + if inspect.isclass(arg) and issubclass(arg, Enum): + is_enum = True + type_str = arg.__name__ + options = [{"name": e.name, "value": e.value} for e in arg] + return type_str, is_enum, options + if inspect.isclass(annotation) and issubclass(annotation, Enum): is_enum = True type_str = annotation.__name__ @@ -72,13 +84,30 @@ def process_method(client_class: Type, method: Any) -> Dict[str, Any] | None: step_name = getattr(method, "_step_name") sig = inspect.signature(method) + + # Resolve string annotations from `from __future__ import annotations` + resolved_hints: dict[str, Any] = {} + try: + unwrapped = inspect.unwrap(method) + resolved_hints = get_type_hints(unwrapped) + except Exception: + pass + parameters = [] for param_name, param in sig.parameters.items(): + # Use resolved type hint if available (handles stringified annotations) + annotation = resolved_hints.get(param_name, param.annotation) # Skip dependencies that are automatically injected - if param.annotation in DEPENDENCIES: + if annotation in DEPENDENCIES: continue - param_info = process_parameter(param_name, param) + resolved_param = inspect.Parameter( + param_name, + param.kind, + default=param.default, + annotation=annotation, + ) + param_info = process_parameter(param_name, resolved_param) if param_info: parameters.append(param_info) diff --git a/src/openutm_verification/server/main.py b/src/openutm_verification/server/main.py index c1123a1e..ced267da 100644 --- a/src/openutm_verification/server/main.py +++ b/src/openutm_verification/server/main.py @@ -142,6 +142,9 @@ async def reset_session( logger.debug(f"Session reset request received: {config.model_dump()}") await runner.close_session() + runner.current_output_dir = None + runner.current_timestamp_str = None + runner.current_start_time = None # If configuration is provided from the frontend, apply it if config.flight_blender: diff --git a/src/openutm_verification/server/runner.py b/src/openutm_verification/server/runner.py index 92adfcb1..cc17e5ea 100644 --- a/src/openutm_verification/server/runner.py +++ b/src/openutm_verification/server/runner.py @@ -695,22 +695,25 @@ async def run_scenario(self, scenario: ScenarioDefinition) -> List[StepResult]: ) await self.initialize_session() - # Set up logging to file for this run - run_timestamp = datetime.now(timezone.utc) - self.current_start_time = run_timestamp - self.current_timestamp_str = get_run_timestamp_str(run_timestamp) - base_output_dir = Path(self.config.reporting.output_dir) - self.current_output_dir = base_output_dir / self.current_timestamp_str - self.current_output_dir.mkdir(parents=True, exist_ok=True) - - log_file = setup_logging( - self.current_output_dir, - "report", - self.config.reporting.formats, - debug=False, - ) - if log_file: - logger.info(f"Logging to file: {log_file}") + # Set up logging to file for this run (skip if already configured, + # e.g. when called from CLI mode which sets up a shared output dir) + if not self.current_start_time: + self.current_start_time = datetime.now(timezone.utc) + if not self.current_output_dir: + run_timestamp = datetime.now(timezone.utc) + self.current_timestamp_str = get_run_timestamp_str(run_timestamp) + base_output_dir = Path(self.config.reporting.output_dir) + self.current_output_dir = base_output_dir / self.current_timestamp_str + self.current_output_dir.mkdir(parents=True, exist_ok=True) + + log_file = setup_logging( + self.current_output_dir, + "report", + self.config.reporting.formats, + debug=False, + ) + if log_file: + logger.info(f"Logging to file: {log_file}") # Validate and prepare steps seen_ids = set() diff --git a/src/openutm_verification/simulator/geo_json_telemetry.py b/src/openutm_verification/simulator/geo_json_telemetry.py index 9a20a399..56b6d29e 100644 --- a/src/openutm_verification/simulator/geo_json_telemetry.py +++ b/src/openutm_verification/simulator/geo_json_telemetry.py @@ -81,7 +81,7 @@ def generate_air_traffic_data( sensor_ids: list[UUID], number_of_aircraft: int = 1, use_multiple_sensors: bool = False, - ) -> list[list[FlightObservationSchema]]: + ) -> list[FlightObservationSchema]: """Generate simulated air traffic observations for the specified duration. Creates random flight trajectories within the configured geographic bounds @@ -94,9 +94,8 @@ def generate_air_traffic_data( use_multiple_sensors: If True, randomly assign sensor IDs from the list. Returns: - A list of flights, where each flight is a list of ``FlightObservationSchema`` - instances containing position, altitude, and metadata for each generated - data point. + A flat list of ``FlightObservationSchema`` instances containing position, + altitude, and metadata for each generated data point across all aircraft. """ logger.info(f"Generating air traffic data for {duration} seconds with {'multiple' if use_multiple_sensors else 'single'} sensor(s)") all_trajectories = [] @@ -106,11 +105,10 @@ def generate_air_traffic_data( trajectory_geojson = generate_random_geojson("LineString", boundingBox=self.box.bounds, numberVertices=duration) all_trajectories.append(trajectory_geojson) - all_air_traffic: list[list[FlightObservationSchema]] = [] + all_air_traffic: list[FlightObservationSchema] = [] for trajectory_geojson in all_trajectories: # A GeoJSON LineString has 'coordinates' as a list of [lon, lat] pairs coordinates = trajectory_geojson["coordinates"] - airtraffic: list[FlightObservationSchema] = [] icao_address = "".join(random.choices("0123456789ABCDEF", k=6)) # Each coordinate corresponds to one second of flight time for i, point in enumerate(coordinates): @@ -122,7 +120,7 @@ def generate_air_traffic_data( metadata = {"sensor_id": str(selected_sensor_id)} if selected_sensor_id else {} # Convert altitude from meters to millimeters for altitude_mm field altitude_m = self.config.altitude_of_ground_level_wgs_84 - airtraffic.append( + all_air_traffic.append( FlightObservationSchema( lat_dd=point[1], lon_dd=point[0], @@ -134,8 +132,7 @@ def generate_air_traffic_data( metadata=metadata, ) ) - all_air_traffic.append(airtraffic) - logger.info(f"Generated observations for {len(all_air_traffic)} aircraft") + logger.info(f"Generated {len(all_air_traffic)} observations for {number_of_aircraft} aircraft") return all_air_traffic diff --git a/tests/test_altitude_units.py b/tests/test_altitude_units.py index 3a759111..75aee2eb 100644 --- a/tests/test_altitude_units.py +++ b/tests/test_altitude_units.py @@ -111,10 +111,9 @@ def test_air_traffic_altitude_is_in_millimeters(self, sample_geojson): number_of_aircraft=1, ) - # Get first observation from first aircraft + # Get first observation from flat list assert len(result) > 0 - assert len(result[0]) > 0 - first_obs = result[0][0] + first_obs = result[0] # altitude_mm should be altitude_meters * 1000 expected_mm = altitude_meters * 1000 @@ -141,7 +140,7 @@ def test_single_sensor_mode_uses_one_sensor_id(self, sample_geojson): use_multiple_sensors=False, ) - unique_sensor_ids = {obs.metadata["sensor_id"] for flight in result for obs in flight} + unique_sensor_ids = {obs.metadata["sensor_id"] for obs in result} assert len(unique_sensor_ids) == 1, f"Expected 1 sensor ID in single mode, got {len(unique_sensor_ids)}: {unique_sensor_ids}" assert unique_sensor_ids == {str(sensor_ids[0])} @@ -179,7 +178,7 @@ def test_multiple_sensor_mode_assigns_varied_sensor_ids(self, sample_geojson): use_multiple_sensors=True, ) - unique_sensor_ids = {obs.metadata["sensor_id"] for flight in result for obs in flight} + unique_sensor_ids = {obs.metadata["sensor_id"] for obs in result} assert len(unique_sensor_ids) > 1, ( f"Expected multiple distinct sensor IDs in multiple mode, got {len(unique_sensor_ids)}: {unique_sensor_ids}" ) @@ -288,7 +287,7 @@ def test_2d_map_converts_mm_to_meters_for_display(self): timestamp=1234567890, ) - air_traffic_data = [[obs]] + air_traffic_data = [obs] result = _reorganize_air_traffic_by_aircraft(air_traffic_data) assert "TEST01" in result diff --git a/tests/test_client_settings.py b/tests/test_client_settings.py index 74de4a28..15313d97 100644 --- a/tests/test_client_settings.py +++ b/tests/test_client_settings.py @@ -170,7 +170,7 @@ def test_from_config_all_fields(self): """All config fields are correctly mapped.""" sim_config = BlueSkyAirTrafficSimulatorSettings( number_of_aircraft=10, - simulation_duration_seconds=120, + simulation_duration=120, single_or_multiple_sensors="single", sensor_ids=["bluesky_sensor"], session_ids=["bluesky_session"], @@ -182,7 +182,7 @@ def test_from_config_all_fields(self): ) assert settings.simulation_config_path == "/path/to/simulation.scn" - assert settings.simulation_duration_seconds == 120 + assert settings.simulation_duration == 120 assert settings.number_of_aircraft == 10 assert settings.single_or_multiple_sensors == SENSOR_MODE_SINGLE assert settings.sensor_ids == ["bluesky_sensor"] @@ -192,7 +192,7 @@ def test_from_config_no_simulation(self): """Settings work without simulation path.""" sim_config = BlueSkyAirTrafficSimulatorSettings( number_of_aircraft=3, - simulation_duration_seconds=30, + simulation_duration=30, ) settings = BlueSkyAirTrafficSettings.from_config(sim_config, simulation_path=None) @@ -204,7 +204,7 @@ def test_from_config_defaults(self): """Default values work correctly.""" sim_config = BlueSkyAirTrafficSimulatorSettings( number_of_aircraft=2, - simulation_duration_seconds=30, + simulation_duration=30, ) settings = BlueSkyAirTrafficSettings.from_config(sim_config) diff --git a/tests/test_client_steps.py b/tests/test_client_steps.py index 7a2fb40e..61b154c1 100644 --- a/tests/test_client_steps.py +++ b/tests/test_client_steps.py @@ -31,7 +31,7 @@ def fb_client(): def at_client(): settings = MagicMock() settings.simulation_config_path = "test_config.json" - settings.simulation_duration_seconds = 60 + settings.simulation_duration = 60 settings.number_of_aircraft = 1 settings.sensor_ids = [] client = AirTrafficClient(settings) @@ -327,52 +327,48 @@ async def test_submit_simulated_air_traffic(fb_client): mock_response.text = "ok" fb_client.post.return_value = mock_response - # Create dummy observations + # Create dummy observations (flat list) obs = [ - [ - FlightObservationSchema( - lat_dd=0.0, - lon_dd=0.0, - altitude_mm=0.0, - traffic_source=0, - source_type=0, - timestamp=0, - metadata={"session_id": "sess1"}, - icao_address="A1", - ), - FlightObservationSchema( - lat_dd=0.0, - lon_dd=0.0, - altitude_mm=0.0, - traffic_source=0, - source_type=0, - timestamp=1, - metadata={"session_id": "sess1"}, - icao_address="A1", - ), - ], - [ - FlightObservationSchema( - lat_dd=0.0, - lon_dd=0.0, - altitude_mm=0.0, - traffic_source=0, - source_type=0, - timestamp=0, - metadata={"session_id": "sess2"}, - icao_address="A2", - ), - FlightObservationSchema( - lat_dd=0.0, - lon_dd=0.0, - altitude_mm=0.0, - traffic_source=0, - source_type=0, - timestamp=1, - metadata={"session_id": "sess2"}, - icao_address="A2", - ), - ], + FlightObservationSchema( + lat_dd=0.0, + lon_dd=0.0, + altitude_mm=0.0, + traffic_source=0, + source_type=0, + timestamp=0, + metadata={"session_id": "sess1"}, + icao_address="A1", + ), + FlightObservationSchema( + lat_dd=0.0, + lon_dd=0.0, + altitude_mm=0.0, + traffic_source=0, + source_type=0, + timestamp=1, + metadata={"session_id": "sess1"}, + icao_address="A1", + ), + FlightObservationSchema( + lat_dd=0.0, + lon_dd=0.0, + altitude_mm=0.0, + traffic_source=0, + source_type=0, + timestamp=0, + metadata={"session_id": "sess2"}, + icao_address="A2", + ), + FlightObservationSchema( + lat_dd=0.0, + lon_dd=0.0, + altitude_mm=0.0, + traffic_source=0, + source_type=0, + timestamp=1, + metadata={"session_id": "sess2"}, + icao_address="A2", + ), ] # Mock Arrow objects @@ -395,6 +391,17 @@ def __add__(self, other): def shift(self, seconds=0): return MockArrow(self.time_val + seconds) + def floor(self, _unit="second"): + return MockArrow(int(self.time_val)) + + def __hash__(self): + return hash(int(self.time_val)) + + def __eq__(self, other): + if isinstance(other, MockArrow): + return self.time_val == other.time_val + return NotImplemented + def __repr__(self): return f"MockArrow({self.time_val})" @@ -421,24 +428,15 @@ def side_effect_get(arg): mock_get.side_effect = side_effect_get # Mock arrow.now() to advance time - # Logic in code: - # start_time = now() (0) - # loop: current_sim_time (0) < sim_end (1) - # target_real_time = start_time + (current - start) = 0 + 0 = 0 - # while now() < target_real_time: sleep - # submit - # current_sim_time shift +1 -> 1 - # loop: current_sim_time (1) < sim_end (1) -> False (Wait, loop is while < end) - # Actually max(end_times) is 1. So loop runs for 0. - - # We need to ensure loop runs at least once. - # If start=0, end=1. Loop runs for 0. Next is 1. 1 < 1 is False. + # The code iterates over unique time slots [0, 1] with 1s pacing. + # slot 0: slot_index==0 → skip wait → batch submit 2 aircraft + # slot 1: slot_index==1 → asyncio.sleep(shift(1) - now()) → batch submit 2 aircraft + # After loop: now() for duration calculation mock_now.side_effect = [ MockArrow(0), # start_time - MockArrow(0), # check loop wait - MockArrow(2), # check loop wait (exit wait) - MockArrow(3), # next call + MockArrow(0), # slot 1: sleep_seconds calc → (shift(1) - 0).total_seconds() = 1 + MockArrow(3), # duration calculation ] result = await fb_client.submit_simulated_air_traffic(observations=obs) @@ -446,7 +444,7 @@ def side_effect_get(arg): # Result now returns detailed dict with submission stats assert isinstance(result.result, dict) assert result.result["aircraft_count"] == 2 - assert result.result["observations_submitted"] == 2 + assert result.result["observations_submitted"] == 4 assert fb_client.post.called @@ -523,11 +521,11 @@ async def test_generate_simulated_air_traffic_data(at_client): patch("openutm_verification.core.clients.air_traffic.air_traffic_client.GeoJSONAirtrafficSimulator") as MockSim, ): mock_sim_instance = MockSim.return_value - mock_sim_instance.generate_air_traffic_data.return_value = [[{"obs": 1}]] + mock_sim_instance.generate_air_traffic_data.return_value = [{"obs": 1}] result = await at_client.generate_simulated_air_traffic_data() - assert result.result == [[{"obs": 1}]] + assert result.result == [{"obs": 1}] mock_sim_instance.generate_air_traffic_data.assert_called_once() @@ -546,7 +544,7 @@ async def test_generate_simulated_air_traffic_data_single_sensor(): patch("openutm_verification.core.clients.air_traffic.air_traffic_client.GeoJSONAirtrafficSimulator") as MockSim, ): mock_sim_instance = MockSim.return_value - mock_sim_instance.generate_air_traffic_data.return_value = [[{"obs": 1}]] + mock_sim_instance.generate_air_traffic_data.return_value = [{"obs": 1}] await client.generate_simulated_air_traffic_data() @@ -569,7 +567,7 @@ async def test_generate_simulated_air_traffic_data_multiple_sensors(): patch("openutm_verification.core.clients.air_traffic.air_traffic_client.GeoJSONAirtrafficSimulator") as MockSim, ): mock_sim_instance = MockSim.return_value - mock_sim_instance.generate_air_traffic_data.return_value = [[{"obs": 1}]] + mock_sim_instance.generate_air_traffic_data.return_value = [{"obs": 1}] await client.generate_simulated_air_traffic_data() @@ -596,33 +594,33 @@ async def test_fetch_data(os_client): os_client.get.assert_called_once() -def _make_observations(num_aircraft: int = 2, duration_seconds: int = 30) -> list[list[FlightObservationSchema]]: +def _make_observations(num_aircraft: int = 2, duration_seconds: int = 30) -> list[FlightObservationSchema]: """Build a minimal observations list: num_aircraft aircraft, each with duration_seconds 1-Hz observations.""" base_ts = int(_time.time()) - duration_seconds result = [] for i in range(num_aircraft): - aircraft_obs = [ - FlightObservationSchema( - lat_dd=46.97 + i * 0.001, - lon_dd=7.47, - altitude_mm=100000.0, - traffic_source=0, - source_type=0, - icao_address=f"AABBCC{i:02d}", - timestamp=base_ts + t, - metadata={}, + for t in range(duration_seconds): + result.append( + FlightObservationSchema( + lat_dd=46.97 + i * 0.001, + lon_dd=7.47, + altitude_mm=100000.0, + traffic_source=0, + source_type=0, + icao_address=f"AABBCC{i:02d}", + timestamp=base_ts + t, + metadata={}, + ) ) - for t in range(duration_seconds) - ] - result.append(aircraft_obs) return result -def _valid_metrics_payload(observations: list[list[FlightObservationSchema]]) -> dict: - num_aircraft = sum(1 for a in observations if a) - total_obs = sum(len(a) for a in observations if a) - start_ts = min(a[0].timestamp for a in observations if a) - end_ts = max(a[-1].timestamp for a in observations if a) +def _valid_metrics_payload(observations: list[FlightObservationSchema]) -> dict: + icao_addresses = set(obs.icao_address for obs in observations) + num_aircraft = len(icao_addresses) + total_obs = len(observations) + start_ts = min(obs.timestamp for obs in observations) + end_ts = max(obs.timestamp for obs in observations) duration = end_ts - start_ts + 1 # +1 to include the last observation interval prob = min(1.0, total_obs / (num_aircraft * duration)) window_start = "2024-01-01T00:00:00Z" diff --git a/tests/test_group_execution.py b/tests/test_group_execution.py index e206f013..b408d36c 100644 --- a/tests/test_group_execution.py +++ b/tests/test_group_execution.py @@ -50,7 +50,10 @@ async def test_group_with_loop(): fetch_data: steps: - id: fetch - step: Fetch OpenSky Data + step: Stream Air Traffic + arguments: + provider: opensky + target: flight_blender - id: wait step: Wait X seconds arguments: @@ -85,12 +88,20 @@ async def test_group_references_within_group(): groups: process_data: steps: - - id: fetch - step: Fetch OpenSky Data - - id: submit - step: Submit Air Traffic + - id: stream + step: Stream Air Traffic + arguments: + provider: opensky + target: flight_blender + - id: generate + step: Generate Random Number + arguments: + min: 1 + max: 5 + - id: wait + step: Wait X seconds arguments: - observations: ${{ group.fetch.result }} + duration: ${{ group.generate.result }} steps: - step: process_data @@ -103,8 +114,8 @@ async def test_group_references_within_group(): assert "process_data" in scenario.groups group = scenario.groups["process_data"] - # Verify the submit step has a reference to fetch - assert group.steps[1].arguments["observations"] == "${{ group.fetch.result }}" + # Verify the submit step has correct arguments + assert group.steps[2].arguments["duration"] == "${{ group.generate.result }}" @pytest.mark.asyncio diff --git a/tests/test_reporting_output.py b/tests/test_reporting_output.py index a47143e3..61d40d7c 100644 --- a/tests/test_reporting_output.py +++ b/tests/test_reporting_output.py @@ -29,10 +29,10 @@ def test_report_outputs_use_result(tmp_path: Path): number_of_aircraft=1, simulation_duration=1, single_or_multiple_sensors="single", sensor_ids=[] ), blue_sky_air_traffic_simulator_settings=BlueSkyAirTrafficSimulatorSettings( - number_of_aircraft=1, simulation_duration_seconds=1, single_or_multiple_sensors="single", sensor_ids=[] + number_of_aircraft=1, simulation_duration=1, single_or_multiple_sensors="single", sensor_ids=[] ), bayesian_air_traffic_simulator_settings=BayesianAirTrafficSimulatorSettings( - number_of_aircraft=1, simulation_duration_seconds=1, single_or_multiple_sensors="single", sensor_ids=[] + number_of_aircraft=1, simulation_duration=1, single_or_multiple_sensors="single", sensor_ids=[] ), data_files=DataFiles(), suites={}, diff --git a/tests/test_step_status_updates.py b/tests/test_step_status_updates.py index e98c6b1b..8311d91d 100644 --- a/tests/test_step_status_updates.py +++ b/tests/test_step_status_updates.py @@ -43,7 +43,7 @@ async def test_step_status_fail_on_validation_error(self, session_manager): session_manager.session_context = mock_context session_manager.session_resolver = MagicMock() - step = StepDefinition(id="test_step", step="Submit Air Traffic", arguments={"observations": None}) + step = StepDefinition(id="test_step", step="Stream Air Traffic", arguments={"provider": "opensky", "target": "flight_blender"}) # Simulate a validation error by making _execute_step raise with patch.object(session_manager, "_execute_step") as mock_execute: @@ -117,8 +117,8 @@ async def test_remaining_group_steps_marked_skip_on_failure(self, session_manage "my_group": GroupDefinition( description="Test group", steps=[ - StepDefinition(id="step1", step="Fetch OpenSky Data"), - StepDefinition(id="step2", step="Submit Air Traffic", arguments={"observations": []}), + StepDefinition(id="step1", step="Stream Air Traffic", arguments={"provider": "opensky", "target": "flight_blender"}), + StepDefinition(id="step2", step="Wait X seconds", arguments={"duration": 1}), StepDefinition(id="step3", step="Wait X seconds", arguments={"duration": 1}), ], ) @@ -201,11 +201,11 @@ async def test_group_step_with_condition_skipped(self, session_manager): "my_group": GroupDefinition( description="Test group", steps=[ - StepDefinition(id="fetch", step="Fetch OpenSky Data"), + StepDefinition(id="fetch", step="Stream Air Traffic", arguments={"provider": "opensky", "target": "flight_blender"}), StepDefinition( id="submit", - step="Submit Air Traffic", - arguments={"observations": []}, + step="Wait X seconds", + arguments={"duration": 1}, if_condition="steps.fetch.result != None", ), StepDefinition(id="wait", step="Wait X seconds", arguments={"duration": 1}), @@ -290,11 +290,11 @@ async def test_group_context_takes_priority_over_state_step_results(self, sessio "my_group": GroupDefinition( description="Test group", steps=[ - StepDefinition(id="fetch", step="Fetch OpenSky Data"), + StepDefinition(id="fetch", step="Stream Air Traffic", arguments={"provider": "opensky", "target": "flight_blender"}), StepDefinition( id="submit", - step="Submit Air Traffic", - arguments={"observations": "${{ steps.fetch.result }}"}, + step="Wait X seconds", + arguments={"duration": 1}, ), ], ) diff --git a/tests/test_stream_air_traffic.py b/tests/test_stream_air_traffic.py index 84cfb9c7..ed9454bd 100644 --- a/tests/test_stream_air_traffic.py +++ b/tests/test_stream_air_traffic.py @@ -6,13 +6,20 @@ from openutm_verification.core.providers import ProviderType, create_provider from openutm_verification.core.providers.geojson_provider import GeoJSONProvider +from openutm_verification.core.providers.latency import shift_timestamps from openutm_verification.core.providers.opensky_provider import OpenSkyProvider +from openutm_verification.core.reporting.reporting_models import Status, StepResult from openutm_verification.core.steps import AirTrafficStepClient from openutm_verification.core.streamers import StreamResult, TargetType, create_streamer from openutm_verification.core.streamers.null_streamer import NullStreamer from openutm_verification.simulator.models.flight_data_types import FlightObservationSchema +def _wrap_step_result(result, step_name="test"): + """Wrap a raw result in a StepResult as the @scenario_step decorator would.""" + return StepResult(name=step_name, status=Status.PASS, duration=0.0, result=result) + + class TestProviderFactory: """Tests for the provider factory.""" @@ -43,6 +50,30 @@ def test_create_unknown_provider_raises(self): with pytest.raises(ValueError, match="Unknown provider"): create_provider(name="unknown") # type: ignore + def test_create_provider_with_latency_wraps_in_latency_provider(self): + """Test that data_quality='latency' wraps the provider with LatencyProviderWrapper.""" + from openutm_verification.core.providers.latency import LatencyProviderWrapper + + provider = create_provider( + name="geojson", + config_path="/some/path.geojson", + data_quality="latency", + ) + assert isinstance(provider, LatencyProviderWrapper) + assert provider.name == "geojson" + + def test_create_provider_nominal_does_not_wrap(self): + """Test that data_quality='nominal' returns unwrapped provider.""" + from openutm_verification.core.providers.latency import LatencyProviderWrapper + + provider = create_provider( + name="geojson", + config_path="/some/path.geojson", + data_quality="nominal", + ) + assert isinstance(provider, GeoJSONProvider) + assert not isinstance(provider, LatencyProviderWrapper) + class TestStreamerFactory: """Tests for the streamer factory.""" @@ -94,6 +125,55 @@ def test_stream_result_with_errors(self): assert len(result.errors) == 2 +class TestLatencySimulation: + """Tests for the apply_latency function and LatencyProviderWrapper.""" + + def test_apply_latency_does_not_mutate_originals(self): + """Test that apply_latency creates copies instead of mutating original observations.""" + + original_obs = FlightObservationSchema( + lat_dd=46.9, + lon_dd=7.4, + altitude_mm=1000000, + traffic_source=0, + source_type=0, + icao_address="TEST1", + timestamp=1000000, + ) + observations = [original_obs] + + # Force all observations to be shifted (100% probability, fixed shift) + shift_timestamps(observations, probability=1.0, shift_range=(1.0, 1.0)) + + # Original should be unchanged + assert original_obs.timestamp == 1000000 + + def test_apply_latency_shifts_in_seconds_not_milliseconds(self): + """Test that timestamp shifts are applied in seconds, not milliseconds.""" + obs = FlightObservationSchema( + lat_dd=46.9, + lon_dd=7.4, + altitude_mm=1000000, + traffic_source=0, + source_type=0, + icao_address="TEST1", + timestamp=1000000, + ) + observations = [obs] + + # Use 100% probability and fixed shift of exactly 2 seconds. + # Calling shift_timestamps directly guarantees the shift path + # is always taken (no random drop/shift branching). + result, total_shifted = shift_timestamps(observations, probability=1.0, shift_range=(2.0, 2.0)) + + assert total_shifted == 1 + assert len(result) == 1 + shifted_obs = result[0] + # Shift should be 2 seconds, not 2000 milliseconds + shift = abs(shifted_obs.timestamp - 1000000) + assert shift == 2, f"Timestamp shift {shift} is not 2 seconds" + + class TestAirTrafficStepClient: """Tests for the AirTrafficStepClient.""" @@ -166,18 +246,14 @@ def test_step_param_model_has_required_fields(self): def test_provider_type_literal(self): """Test that ProviderType includes expected values.""" - from typing import get_args - expected = {"geojson", "bluesky", "bayesian", "opensky"} - actual = set(get_args(ProviderType)) + actual = set(item.value for item in ProviderType) assert actual == expected def test_target_type_literal(self): """Test that TargetType includes expected values.""" - from typing import get_args - expected = {"flight_blender", "amqp", "none"} - actual = set(get_args(TargetType)) + actual = set(item.value for item in TargetType) assert actual == expected @@ -189,28 +265,24 @@ def test_target_type_literal(self): def _create_mock_observations(): """Helper to create mock flight observations.""" return [ - [ - FlightObservationSchema( - lat_dd=46.9, - lon_dd=7.4, - altitude_mm=1000000, - traffic_source=0, - source_type=0, - icao_address="ABC123", - timestamp=1234567890, - ), - ], - [ - FlightObservationSchema( - lat_dd=46.95, - lon_dd=7.45, - altitude_mm=1100000, - traffic_source=0, - source_type=0, - icao_address="DEF456", - timestamp=1234567891, - ), - ], + FlightObservationSchema( + lat_dd=46.9, + lon_dd=7.4, + altitude_mm=1000000, + traffic_source=0, + source_type=0, + icao_address="ABC123", + timestamp=1234567890, + ), + FlightObservationSchema( + lat_dd=46.95, + lon_dd=7.45, + altitude_mm=1100000, + traffic_source=0, + source_type=0, + icao_address="DEF456", + timestamp=1234567891, + ), ] @@ -225,7 +297,9 @@ async def test_geojson_provider_instantiates_client_with_correct_settings(self, # Setup mock client instance mock_client_instance = AsyncMock() - mock_client_instance.generate_simulated_air_traffic_data = AsyncMock(return_value=mock_observations) + mock_client_instance.generate_simulated_air_traffic_data = AsyncMock( + return_value=_wrap_step_result(mock_observations, "Generate Simulated Air Traffic Data") + ) mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client_instance) mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) @@ -268,7 +342,9 @@ async def test_geojson_provider_uses_default_duration_when_not_overridden(self, mock_observations = _create_mock_observations() mock_client_instance = AsyncMock() - mock_client_instance.generate_simulated_air_traffic_data = AsyncMock(return_value=mock_observations) + mock_client_instance.generate_simulated_air_traffic_data = AsyncMock( + return_value=_wrap_step_result(mock_observations, "Generate Simulated Air Traffic Data") + ) mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client_instance) mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) @@ -292,21 +368,21 @@ async def test_bluesky_provider_instantiates_client_with_correct_settings(self, from openutm_verification.core.providers.bluesky_provider import BlueSkyProvider mock_observations = [ - [ - FlightObservationSchema( - lat_dd=46.9, - lon_dd=7.4, - altitude_mm=5000000, - traffic_source=0, - source_type=0, - icao_address="BLUESKY1", - timestamp=1234567890, - ), - ], + FlightObservationSchema( + lat_dd=46.9, + lon_dd=7.4, + altitude_mm=5000000, + traffic_source=0, + source_type=0, + icao_address="BLUESKY1", + timestamp=1234567890, + ), ] mock_client_instance = AsyncMock() - mock_client_instance.generate_bluesky_sim_air_traffic_data = AsyncMock(return_value=mock_observations) + mock_client_instance.generate_bluesky_sim_air_traffic_data = AsyncMock( + return_value=_wrap_step_result(mock_observations, "Generate BlueSky Simulation Air Traffic Data") + ) mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client_instance) mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) @@ -325,7 +401,7 @@ async def test_bluesky_provider_instantiates_client_with_correct_settings(self, settings = mock_client_class.call_args[0][0] assert settings.simulation_config_path == "/path/to/scenario.scn" - assert settings.simulation_duration_seconds == 25 + assert settings.simulation_duration == 25 assert settings.number_of_aircraft == 2 assert settings.sensor_ids == ["sensor-1"] assert settings.session_ids == ["session-1"] @@ -349,21 +425,21 @@ async def test_bayesian_provider_instantiates_client_with_correct_settings(self, from openutm_verification.core.providers.bayesian_provider import BayesianProvider mock_observations = [ - [ - FlightObservationSchema( - lat_dd=47.0, - lon_dd=8.0, - altitude_mm=3000000, - traffic_source=0, - source_type=0, - icao_address="BAYES1", - timestamp=1234567890, - ), - ], + FlightObservationSchema( + lat_dd=47.0, + lon_dd=8.0, + altitude_mm=3000000, + traffic_source=0, + source_type=0, + icao_address="BAYES1", + timestamp=1234567890, + ), ] mock_client_instance = AsyncMock() - mock_client_instance.generate_bayesian_sim_air_traffic_data = AsyncMock(return_value=mock_observations) + mock_client_instance.generate_bayesian_sim_air_traffic_data = AsyncMock( + return_value=_wrap_step_result(mock_observations, "Generate Bayesian Air Traffic Data") + ) mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client_instance) mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) @@ -382,7 +458,7 @@ async def test_bayesian_provider_instantiates_client_with_correct_settings(self, settings = mock_client_class.call_args[0][0] assert settings.simulation_config_path == "/path/to/model.mat" - assert settings.simulation_duration_seconds == 80 + assert settings.simulation_duration == 80 assert settings.number_of_aircraft == 5 assert settings.sensor_ids == ["sensor-bayesian"] assert settings.session_ids == ["session-bayesian"] @@ -396,7 +472,9 @@ async def test_bayesian_provider_handles_none_result(self, mock_client_class): from openutm_verification.core.providers.bayesian_provider import BayesianProvider mock_client_instance = AsyncMock() - mock_client_instance.generate_bayesian_sim_air_traffic_data = AsyncMock(return_value=None) + mock_client_instance.generate_bayesian_sim_air_traffic_data = AsyncMock( + return_value=_wrap_step_result(None, "Generate Bayesian Air Traffic Data") + ) mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client_instance) mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) @@ -442,7 +520,7 @@ async def test_opensky_provider_instantiates_client_with_correct_viewport(self, mock_get_settings.return_value = mock_config mock_client_instance = AsyncMock() - mock_client_instance.fetch_data = AsyncMock(return_value=mock_observations) + mock_client_instance.fetch_data = AsyncMock(return_value=_wrap_step_result(mock_observations, "Fetch OpenSky Data")) mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client_instance) mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) @@ -462,8 +540,8 @@ async def test_opensky_provider_instantiates_client_with_correct_viewport(self, # Verify fetch_data was called mock_client_instance.fetch_data.assert_called_once() - # Result should be wrapped in outer list for consistency - assert result == [mock_observations] + # Result should be the flat list directly + assert result == mock_observations @pytest.mark.asyncio @patch("openutm_verification.core.providers.opensky_provider.OpenSkyClient") @@ -476,7 +554,7 @@ async def test_opensky_provider_returns_empty_list_when_no_data(self, mock_get_s mock_get_settings.return_value = mock_config mock_client_instance = AsyncMock() - mock_client_instance.fetch_data = AsyncMock(return_value=None) + mock_client_instance.fetch_data = AsyncMock(return_value=_wrap_step_result(None, "Fetch OpenSky Data")) mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client_instance) mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) @@ -491,8 +569,9 @@ class TestFlightBlenderStreamerIntegration: @pytest.mark.asyncio @patch("openutm_verification.core.streamers.flight_blender_streamer.FlightBlenderClient") + @patch("openutm_verification.core.streamers.flight_blender_streamer.get_auth_provider") @patch("openutm_verification.core.streamers.flight_blender_streamer.get_settings") - async def test_flight_blender_streamer_submits_to_client(self, mock_get_settings, mock_fb_class): + async def test_flight_blender_streamer_submits_to_client(self, mock_get_settings, mock_get_auth, mock_fb_class): """Test that FlightBlenderStreamer properly submits observations to FlightBlenderClient.""" from openutm_verification.core.streamers.flight_blender_streamer import FlightBlenderStreamer @@ -501,12 +580,19 @@ async def test_flight_blender_streamer_submits_to_client(self, mock_get_settings # Mock get_settings mock_config = MagicMock() mock_config.flight_blender.url = "http://test-flight-blender:8080" - mock_config.flight_blender.auth.username = "test-user" - mock_config.flight_blender.auth.password = "test-pass" + mock_config.flight_blender.auth.audience = "test-audience" + mock_config.flight_blender.auth.scopes = [] mock_get_settings.return_value = mock_config + # Mock auth provider + mock_auth = MagicMock() + mock_auth.get_cached_credentials.return_value = {"token": "test-token"} + mock_get_auth.return_value = mock_auth + mock_fb_client = AsyncMock() - mock_fb_client.submit_simulated_air_traffic = AsyncMock(return_value={"success": True, "observations_submitted": 1}) + mock_fb_client.submit_simulated_air_traffic = AsyncMock( + return_value=_wrap_step_result({"success": True, "observations_submitted": 1}, "Submit Simulated Air Traffic") + ) mock_fb_class.return_value.__aenter__ = AsyncMock(return_value=mock_fb_client) mock_fb_class.return_value.__aexit__ = AsyncMock(return_value=None) @@ -522,8 +608,7 @@ async def test_flight_blender_streamer_submits_to_client(self, mock_get_settings mock_fb_class.assert_called_once() call_kwargs = mock_fb_class.call_args[1] assert call_kwargs["base_url"] == "http://test-flight-blender:8080" - assert call_kwargs["credentials"]["username"] == "test-user" - assert call_kwargs["credentials"]["password"] == "test-pass" + assert call_kwargs["credentials"] == {"token": "test-token"} # Verify submit was called with observations mock_fb_client.submit_simulated_air_traffic.assert_called_once() @@ -534,7 +619,7 @@ async def test_flight_blender_streamer_submits_to_client(self, mock_get_settings assert result.success is True assert result.provider == "geojson" assert result.target == "flight_blender" - assert result.total_batches == 2 + assert result.total_batches == 1 @pytest.mark.asyncio async def test_flight_blender_streamer_handles_empty_observations(self): @@ -555,8 +640,9 @@ async def test_flight_blender_streamer_handles_empty_observations(self): @pytest.mark.asyncio @patch("openutm_verification.core.streamers.flight_blender_streamer.FlightBlenderClient") + @patch("openutm_verification.core.streamers.flight_blender_streamer.get_auth_provider") @patch("openutm_verification.core.streamers.flight_blender_streamer.get_settings") - async def test_flight_blender_streamer_handles_client_error(self, mock_get_settings, mock_fb_class): + async def test_flight_blender_streamer_handles_client_error(self, mock_get_settings, mock_get_auth, mock_fb_class): """Test that FlightBlenderStreamer handles client errors gracefully.""" from openutm_verification.core.streamers.flight_blender_streamer import FlightBlenderStreamer @@ -564,10 +650,14 @@ async def test_flight_blender_streamer_handles_client_error(self, mock_get_setti mock_config = MagicMock() mock_config.flight_blender.url = "http://test:8080" - mock_config.flight_blender.auth.username = "user" - mock_config.flight_blender.auth.password = "pass" + mock_config.flight_blender.auth.audience = "test-audience" + mock_config.flight_blender.auth.scopes = [] mock_get_settings.return_value = mock_config + mock_auth = MagicMock() + mock_auth.get_cached_credentials.return_value = {"token": "test-token"} + mock_get_auth.return_value = mock_auth + mock_fb_client = AsyncMock() mock_fb_client.submit_simulated_air_traffic = AsyncMock(side_effect=Exception("Connection refused")) mock_fb_class.return_value.__aenter__ = AsyncMock(return_value=mock_fb_client) @@ -580,7 +670,9 @@ async def test_flight_blender_streamer_handles_client_error(self, mock_get_setti streamer = FlightBlenderStreamer() result = await streamer.stream_from_provider(mock_provider, duration_seconds=30) + # Protocol-compliant: returns StreamResult with success=False instead of raising assert result.success is False + assert len(result.errors) == 1 assert "Connection refused" in result.errors[0] @@ -591,28 +683,24 @@ class TestNullStreamerIntegration: async def test_null_streamer_collects_observations_without_sending(self): """Test that NullStreamer collects all observations and returns them.""" mock_observations = [ - [ - FlightObservationSchema( - lat_dd=46.9, - lon_dd=7.4, - altitude_mm=1000000, - traffic_source=0, - source_type=0, - icao_address="NULL1", - timestamp=1234567890, - ), - ], - [ - FlightObservationSchema( - lat_dd=47.0, - lon_dd=7.5, - altitude_mm=1100000, - traffic_source=0, - source_type=0, - icao_address="NULL2", - timestamp=1234567891, - ), - ], + FlightObservationSchema( + lat_dd=46.9, + lon_dd=7.4, + altitude_mm=1000000, + traffic_source=0, + source_type=0, + icao_address="NULL1", + timestamp=1234567890, + ), + FlightObservationSchema( + lat_dd=47.0, + lon_dd=7.5, + altitude_mm=1100000, + traffic_source=0, + source_type=0, + icao_address="NULL2", + timestamp=1234567891, + ), ] mock_provider = AsyncMock() @@ -628,7 +716,7 @@ async def test_null_streamer_collects_observations_without_sending(self): # Verify result contains all observations assert result.success is True assert result.target == "none" - assert result.total_batches == 2 + assert result.total_batches == 1 assert result.total_observations == 2 assert result.observations == mock_observations @@ -643,22 +731,22 @@ async def test_stream_air_traffic_with_null_target(self, mock_client_class): from openutm_verification.core.reporting.reporting_models import Status mock_observations = [ - [ - FlightObservationSchema( - lat_dd=46.9, - lon_dd=7.4, - altitude_mm=1000000, - traffic_source=0, - source_type=0, - icao_address="E2E1", - timestamp=1234567890, - ), - ], + FlightObservationSchema( + lat_dd=46.9, + lon_dd=7.4, + altitude_mm=1000000, + traffic_source=0, + source_type=0, + icao_address="E2E1", + timestamp=1234567890, + ), ] # Mock the AirTrafficClient used by GeoJSONProvider mock_client_instance = AsyncMock() - mock_client_instance.generate_simulated_air_traffic_data = AsyncMock(return_value=mock_observations) + mock_client_instance.generate_simulated_air_traffic_data = AsyncMock( + return_value=_wrap_step_result(mock_observations, "Generate Simulated Air Traffic Data") + ) mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client_instance) mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) diff --git a/tests/test_suite_scenario_merge.py b/tests/test_suite_scenario_merge.py index c070aa86..47d34a7e 100644 --- a/tests/test_suite_scenario_merge.py +++ b/tests/test_suite_scenario_merge.py @@ -136,11 +136,11 @@ def test_config_loading_merges_defaults(self, tmp_path): }, "blue_sky_air_traffic_simulator_settings": { "number_of_aircraft": 3, - "simulation_duration_seconds": 30, + "simulation_duration": 30, }, "bayesian_air_traffic_simulator_settings": { "number_of_aircraft": 3, - "simulation_duration_seconds": 30, + "simulation_duration": 30, }, "data_files": { "trajectory": "default_traj.json", @@ -212,11 +212,11 @@ def test_daa_scenario_config_example(self, tmp_path): }, "blue_sky_air_traffic_simulator_settings": { "number_of_aircraft": 3, - "simulation_duration_seconds": 30, + "simulation_duration": 30, }, "bayesian_air_traffic_simulator_settings": { "number_of_aircraft": 3, - "simulation_duration_seconds": 30, + "simulation_duration": 30, }, "data_files": { "trajectory": "trajectory_f1.json", # DEFAULT diff --git a/tests/test_ui_groups.py b/tests/test_ui_groups.py index 64d26e4a..56eaf5da 100644 --- a/tests/test_ui_groups.py +++ b/tests/test_ui_groups.py @@ -13,8 +13,8 @@ def test_group_roundtrip(): "process_data": { "description": "Process some data", "steps": [ - {"id": "fetch", "step": "Fetch OpenSky Data"}, - {"id": "submit", "step": "Submit Air Traffic", "arguments": {"observations": "${{ group.fetch.result }}"}}, + {"id": "fetch", "step": "Stream Air Traffic", "arguments": {"provider": "opensky", "target": "flight_blender"}}, + {"id": "submit", "step": "Wait X seconds", "arguments": {"duration": 1}}, ], } }, @@ -29,7 +29,7 @@ def test_group_roundtrip(): assert scenario.groups["process_data"].description == "Process some data" assert len(scenario.groups["process_data"].steps) == 2 assert scenario.groups["process_data"].steps[0].id == "fetch" - assert scenario.groups["process_data"].steps[1].arguments["observations"] == "${{ group.fetch.result }}" + assert scenario.groups["process_data"].steps[1].arguments["duration"] == 1 # Verify the group is referenced in steps assert scenario.steps[0].step == "process_data" diff --git a/tests/test_yaml_scenarios.py b/tests/test_yaml_scenarios.py index f746100f..7b89309c 100644 --- a/tests/test_yaml_scenarios.py +++ b/tests/test_yaml_scenarios.py @@ -35,9 +35,8 @@ def mock_clients(): fb_client.start_stop_sdsp_session.return_value = "Session Started" # Mock methods that return objects with attributes accessed in YAML - # e.g. ${{ steps.Generate Simulated Air Traffic Data.result.result }} - # But wait, we changed it to just .result in the previous turn. - # Let's check if any other steps return complex objects. + # e.g. ${{ steps.stream_air_traffic.result.observations }} + # The unified Stream Air Traffic step returns StreamResult with observations. mocks["FlightBlenderClient"] = fb_client diff --git a/web-editor/src/components/ScenarioEditor.tsx b/web-editor/src/components/ScenarioEditor.tsx index 5567ee4e..b8ec44b4 100644 --- a/web-editor/src/components/ScenarioEditor.tsx +++ b/web-editor/src/components/ScenarioEditor.tsx @@ -72,7 +72,7 @@ const ScenarioEditorContent = () => { }, blue_sky_air_traffic_simulator_settings: { number_of_aircraft: 3, - simulation_duration_seconds: 30, + simulation_duration: 30, single_or_multiple_sensors: "multiple", sensor_ids: ["562e6297036a4adebb4848afcd1ede90"] } diff --git a/web-editor/src/components/ScenarioEditor/ConfigEditor.tsx b/web-editor/src/components/ScenarioEditor/ConfigEditor.tsx index 5e04771c..6187bd17 100644 --- a/web-editor/src/components/ScenarioEditor/ConfigEditor.tsx +++ b/web-editor/src/components/ScenarioEditor/ConfigEditor.tsx @@ -483,8 +483,8 @@ export const ConfigEditor: React.FC = ({ config, onUpdateConf updateBlueSkyAirTrafficSimulator('simulation_duration_seconds', parseInt(e.target.value))} + value={config.blue_sky_air_traffic_simulator_settings?.simulation_duration || 30} + onChange={(e) => updateBlueSkyAirTrafficSimulator('simulation_duration', parseInt(e.target.value))} min="1" max="3600" /> @@ -561,8 +561,8 @@ export const ConfigEditor: React.FC = ({ config, onUpdateConf updateBayesianAirTrafficSimulator('simulation_duration_seconds', parseInt(e.target.value))} + value={config.bayesian_air_traffic_simulator_settings?.simulation_duration || 30} + onChange={(e) => updateBayesianAirTrafficSimulator('simulation_duration', parseInt(e.target.value))} min="1" max="3600" /> diff --git a/web-editor/src/types/scenario.ts b/web-editor/src/types/scenario.ts index be484eb2..90f66285 100644 --- a/web-editor/src/types/scenario.ts +++ b/web-editor/src/types/scenario.ts @@ -111,14 +111,14 @@ export interface AirTrafficSimulatorSettings { export interface BlueSkyAirTrafficSimulatorSettings { number_of_aircraft?: number; - simulation_duration_seconds?: number; + simulation_duration?: number; single_or_multiple_sensors?: string; sensor_ids?: string[]; } export interface BayesianAirTrafficSimulatorSettings { number_of_aircraft?: number; - simulation_duration_seconds?: number; + simulation_duration?: number; single_or_multiple_sensors?: string; sensor_ids?: string[]; }