Air Traffic Client Unification phase 1 by atti92 · Pull Request #90 · openutm/verification

atti92 · 2026-01-31T12:50:39Z

No description provided.

Copilot

Pull request overview

This PR introduces a unified interface for air traffic streaming operations, implementing "Air Traffic Client Unification phase 1". The changes replace multiple provider-specific scenario steps with a single "Stream Air Traffic" step that supports multiple data sources (GeoJSON, BlueSky, Bayesian, OpenSky) and delivery targets (Flight Blender, AMQP, none).

Changes:

Introduces a provider/streamer architecture with protocol definitions and factory functions
Implements four air traffic providers (GeoJSON, BlueSky, Bayesian, OpenSky) that wrap existing clients
Implements three streamers (NullStreamer, FlightBlenderStreamer, AMQPStreamer) for data delivery
Adds a unified AirTrafficStepClient with a single stream_air_traffic step
Includes comprehensive unit tests for factories and data models
Provides an example YAML scenario demonstrating the new unified interface

Reviewed changes

Copilot reviewed 20 out of 21 changed files in this pull request and generated 19 comments.

Show a summary per file

File	Description
`src/openutm_verification/core/providers/protocol.py`	Defines the AirTrafficProvider protocol for data sources
`src/openutm_verification/core/providers/geojson_provider.py`	GeoJSON provider wrapping AirTrafficClient
`src/openutm_verification/core/providers/bluesky_provider.py`	BlueSky simulator provider wrapping BlueSkyClient
`src/openutm_verification/core/providers/bayesian_provider.py`	Bayesian track generation provider wrapping BayesianTrafficClient
`src/openutm_verification/core/providers/opensky_provider.py`	OpenSky Network live data provider wrapping OpenSkyClient
`src/openutm_verification/core/providers/factory.py`	Factory function for creating providers by type
`src/openutm_verification/core/providers/__init__.py`	Module exports for providers
`src/openutm_verification/core/streamers/protocol.py`	Defines AirTrafficStreamer protocol and StreamResult dataclass
`src/openutm_verification/core/streamers/null_streamer.py`	Null streamer for data collection without delivery
`src/openutm_verification/core/streamers/flight_blender_streamer.py`	Flight Blender API streamer wrapping FlightBlenderClient
`src/openutm_verification/core/streamers/amqp_streamer.py`	AMQP streamer (placeholder implementation)
`src/openutm_verification/core/streamers/factory.py`	Factory function for creating streamers by type
`src/openutm_verification/core/streamers/__init__.py`	Module exports for streamers
`src/openutm_verification/core/steps/air_traffic_step.py`	Unified AirTrafficStepClient with stream_air_traffic method
`src/openutm_verification/core/steps/__init__.py`	Module exports for steps
`src/openutm_verification/core/execution/dependencies.py`	Registers AirTrafficStepClient as a dependency
`tests/test_stream_air_traffic.py`	Unit tests for factories, protocols, and step registration
`scenarios/stream_air_traffic_example.yaml`	Example scenario demonstrating the unified interface

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

src/openutm_verification/core/streamers/flight_blender_streamer.py

tests/test_stream_air_traffic.py

src/openutm_verification/core/streamers/flight_blender_streamer.py

src/openutm_verification/core/providers/opensky_provider.py

tests/test_stream_air_traffic.py

src/openutm_verification/core/providers/bayesian_provider.py

scenarios/stream_air_traffic_example.yaml

Copilot

Pull request overview

Copilot reviewed 20 out of 21 changed files in this pull request and generated 11 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-02-13T20:05:18Z

src/openutm_verification/core/streamers/flight_blender_streamer.py

+            logger.error(f"Flight Blender streaming failed: {e}")
+            return self._make_result(
+                success=False,
+                provider_name=provider.name,
+                duration_seconds=duration_seconds,
+                errors=[str(e)],
+                observations=observations,
+            )


Catching Exception here converts submission failures into a StreamResult and prevents exceptions from propagating to the @scenario_step wrapper. That can cause the scenario step to be reported as PASS even when Flight Blender submission fails. Prefer letting exceptions bubble up (or re-raising FlightBlenderError), and only returning success=False if the step client converts that into a FAIL StepResult/exception.

Suggested change

logger.error(f"Flight Blender streaming failed: {e}")

return self._make_result(

success=False,

provider_name=provider.name,

duration_seconds=duration_seconds,

errors=[str(e)],

observations=observations,

)

# Let exceptions propagate so that the scenario_step wrapper can

# correctly mark this step as failed, rather than converting all

# failures into a StreamResult.

logger.exception(f"Flight Blender streaming failed: {e}")

raise

Copilot · 2026-02-13T20:05:19Z

src/openutm_verification/core/streamers/flight_blender_streamer.py

+            total_observations=total_observations,
+            total_batches=total_batches,
+            errors=errors or [],
+            observations=observations or [],


StreamResult.observations is optional (... | None) so downstream steps can distinguish 'not captured' vs 'captured but empty'. _make_result currently forces observations to [] even when the caller doesn't intend to include observations, which changes that meaning (and diverges from the dataclass default of None). Consider preserving None unless observations are explicitly provided.

Suggested change

observations=observations or [],

observations=observations if observations is not None else None,

Copilot · 2026-02-13T20:05:19Z

src/openutm_verification/core/providers/geojson_provider.py

+            return await client.generate_simulated_air_traffic_data(
+                config_path=self._config_path,
+                duration=effective_duration,
+            )


AirTrafficClient.generate_simulated_air_traffic_data is decorated with @scenario_step, so calling it returns a StepResult, not a raw list[list[FlightObservationSchema]]. Returning that value here will break downstream consumers (e.g., streamers summing len(batch)), and will also wrap nested step results unexpectedly. Consider calling the undecorated implementation (e.g., via .__wrapped__) or refactoring the client to expose a non-step async method for providers to use.

Suggested change

return await client.generate_simulated_air_traffic_data(

config_path=self._config_path,

duration=effective_duration,

)

# Call the undecorated implementation to return raw observations,

# not a StepResult produced by the @scenario_step decorator.

raw_observations = await client.generate_simulated_air_traffic_data.__wrapped__( # type: ignore[attr-defined]

client,

config_path=self._config_path,

duration=effective_duration,

)

return raw_observations

Copilot · 2026-02-13T20:05:19Z

src/openutm_verification/core/streamers/flight_blender_streamer.py

+                return self._make_result(
+                    success=result.get("success", False),


FlightBlenderClient.submit_simulated_air_traffic is decorated with @scenario_step, so result here will be a StepResult, not a dict. Using result.get(...) will raise an AttributeError at runtime. Consider using the raw result (result.result) or bypassing the decorator (e.g., via .__wrapped__), or adding a non-step client method for streamer usage.

Suggested change

return self._make_result(

success=result.get("success", False),

# `submit_simulated_air_traffic` may be decorated with @scenario_step,

# in which case it returns a StepResult wrapper with the actual

# result available as `.result`. Fallback to `result` itself if

# no such attribute exists.

raw_result = getattr(result, "result", result)

return self._make_result(

success=raw_result.get("success", False),

Copilot · 2026-02-13T20:05:19Z

src/openutm_verification/core/steps/air_traffic_step.py

+    @scenario_step("Stream Air Traffic")
+    async def stream_air_traffic(
+        self,
+        provider: ProviderType,
+        duration: int,


Because this method is decorated with @scenario_step, callers receive a StepResult wrapper at runtime, not a raw StreamResult. The return type annotation currently advertises StreamResult, which is misleading for API consumers and static typing. Consider updating the annotation to a StepResult[StreamResult] (or returning an explicit StepResult).

Copilot · 2026-02-13T20:05:20Z

src/openutm_verification/core/providers/opensky_provider.py

+        self._viewport = viewport or DEFAULT_SWITZERLAND_VIEWPORT
+        self._duration = duration or 30
+


self._duration is set in __init__, but get_observations() always performs a single fetch_data() call and never uses either the instance duration or the duration override parameter. This makes the argument and docstring misleading. Either implement duration-based polling/aggregation, or remove/rename the parameter to reflect the actual behavior.

Copilot · 2026-02-13T20:05:20Z

src/openutm_verification/core/providers/bluesky_provider.py

+        self._config_path = config_path or ""
+        self._number_of_aircraft = number_of_aircraft or 2
+        self._duration = duration or 30
+        self._sensor_ids = sensor_ids or []
+        self._session_ids = session_ids or []


Like GeoJSONProvider, this defaults config_path to "" and stores it in _config_path, so omitting config_path will override any configured defaults and likely fail when the underlying BlueSky client loads the scenario. Consider validating config_path early with a helpful error, or sourcing a default path from application config when omitted.

Copilot · 2026-02-13T20:05:20Z

src/openutm_verification/core/providers/bluesky_provider.py

+        )
+
+        async with BlueSkyClient(settings) as client:
+            return await client.generate_bluesky_sim_air_traffic_data(


BlueSkyClient.generate_bluesky_sim_air_traffic_data is a @scenario_step, so this call returns a StepResult wrapper rather than the observation list. That will break the provider/streamer interface (and likely causes type errors at runtime). Consider calling the undecorated method (via .__wrapped__) or adding a non-step helper on the client for providers to call.

Suggested change

return await client.generate_bluesky_sim_air_traffic_data(

generate_raw = client.generate_bluesky_sim_air_traffic_data.__wrapped__

return await generate_raw(

client,

Copilot · 2026-02-13T20:05:20Z

src/openutm_verification/core/providers/bayesian_provider.py

+        )
+
+        async with BayesianTrafficClient(settings) as client:
+            result = await client.generate_bayesian_sim_air_traffic_data(


BayesianTrafficClient.generate_bayesian_sim_air_traffic_data is decorated with @scenario_step, so it returns a StepResult wrapper rather than the observation list. That wrapper is truthy, so result if result else [] will return the StepResult and break downstream processing. Consider bypassing the wrapper (e.g., .__wrapped__) or exposing a non-step client method for providers.

Suggested change

result = await client.generate_bayesian_sim_air_traffic_data(

# Bypass the @scenario_step wrapper to get the raw observation list

generate_fn = getattr(

BayesianTrafficClient.generate_bayesian_sim_air_traffic_data,

"__wrapped__",

BayesianTrafficClient.generate_bayesian_sim_air_traffic_data,

)

result = await generate_fn(

client,

Copilot · 2026-02-13T20:05:21Z

src/openutm_verification/core/steps/air_traffic_step.py

+        return await streamer_instance.stream_from_provider(
+            provider=provider_instance,
+            duration_seconds=duration,
+        )


The @scenario_step wrapper always marks the step as PASS unless an exception is raised or a StepResult with FAIL is returned. As written, this step returns StreamResult even when StreamResult.success is False, which will still produce a PASS step and can hide streaming failures. Consider raising an appropriate exception when streaming fails (e.g., FlightBlenderError/OpenSkyError where relevant) or returning an explicit StepResult with FAIL status based on the StreamResult.success flag.

Suggested change

return await streamer_instance.stream_from_provider(

provider=provider_instance,

duration_seconds=duration,

)

stream_result = await streamer_instance.stream_from_provider(

provider=provider_instance,

duration_seconds=duration,

)

# If streaming failed, raise an exception so the scenario step is marked as failed.

if not stream_result.success:

raise RuntimeError(f"Air traffic streaming failed: {stream_result}")

return stream_result

atti92 added 3 commits January 31, 2026 13:38

Fix dependencies in pyproject

c49acda

Fix missing config for PRs

80459b9

Air Traffic Client Unification phase 1

b48fe66

atti92 requested review from RomanPszonka and Copilot January 31, 2026 12:50

Copilot started reviewing on behalf of atti92 January 31, 2026 12:50 View session

Base automatically changed from fix-pyproject to main January 31, 2026 12:51

Copilot AI reviewed Jan 31, 2026

View reviewed changes

atti92 self-assigned this Jan 31, 2026

atti92 added 6 commits February 11, 2026 21:34

change tests

a5b9304

const

f22a68d

consistent factory

c5c3355

more error handling in FB streamer

36df5df

test context manager

cc0f46e

security

7a38e2d

atti92 marked this pull request as ready for review February 11, 2026 20:44

atti92 requested a review from Copilot February 13, 2026 19:56

Copilot started reviewing on behalf of atti92 February 13, 2026 19:57 View session

Copilot AI reviewed Feb 13, 2026

View reviewed changes

	observations=observations or [],
	observations=observations if observations is not None else None,

		return self._make_result(
		success=result.get("success", False),

-                return self._make_result(
-                    success=result.get("success", False),
+                # `submit_simulated_air_traffic` may be decorated with @scenario_step,
+                # in which case it returns a StepResult wrapper with the actual
+                # result available as `.result`. Fallback to `result` itself if
+                # no such attribute exists.
+                raw_result = getattr(result, "result", result)
+                return self._make_result(
+                    success=raw_result.get("success", False),

		self._viewport = viewport or DEFAULT_SWITZERLAND_VIEWPORT
		self._duration = duration or 30

-            return await client.generate_bluesky_sim_air_traffic_data(
+            generate_raw = client.generate_bluesky_sim_air_traffic_data.__wrapped__
+            return await generate_raw(
+                client,

-            result = await client.generate_bayesian_sim_air_traffic_data(
+            # Bypass the @scenario_step wrapper to get the raw observation list
+            generate_fn = getattr(
+                BayesianTrafficClient.generate_bayesian_sim_air_traffic_data,
+                "__wrapped__",
+                BayesianTrafficClient.generate_bayesian_sim_air_traffic_data,
+            )
+            result = await generate_fn(
+                client,

Comments

Conversation

atti92 commented Jan 31, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Copilot AI Feb 13, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 13, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 13, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 13, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 13, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 13, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 13, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 13, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 13, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 13, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant