Air Traffic Client Unification Design Spec

Status: Implemented — legacy removal pending
Date: 2026-01-31
Last Updated: 2026-04-12

Executive Summary

Refactor air traffic data handling into a clean Provider + Streamer architecture, exposing a single unified scenario step: "Stream Air Traffic" with configurable provider, duration, and target.

1. Implementation Status

1.1 Status Summary

1.2 Old Step References — All Migrated ✅

All scenarios, docs, and README files use "Stream Air Traffic".
No remaining references to old step names.

1.3 Legacy @internal_step Decorators

All 9 legacy decorators switched from @scenario_step to @internal_step.
This keeps StepResult wrapping, timing, logging, and error handling intact,
but the methods no longer appear in STEP_REGISTRY or the UI.

2. Implemented Architecture

┌─────────────────────────────────────────────────────────────────────┐
│                         PROVIDERS                                   │
│              (Pure data acquisition - no system knowledge)          │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│   GeoJSONProvider   BlueSkyProvider   BayesianProvider   OpenSky    │
│    (synthetic)       (simulator)       (track-gen)       (live)     │
│                                                                     │
│   All implement: AirTrafficProvider protocol                        │
│   All constructed via from_kwargs() + create_provider() factory     │
│                                                                     │
│   LatencyProviderWrapper: decorator for data_quality="latency"      │
│                                                                     │
└──────────────────────────────┬──────────────────────────────────────┘
                               │
                               ▼
                    list[FlightObservationSchema]
                               │
                               ▼
┌─────────────────────────────────────────────────────────────────────┐
│                         STREAMERS                                   │
│                  (Delivery to target systems)                       │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│   FlightBlenderStreamer        AMQPStreamer        NullStreamer     │
│     (HTTP POST)              (placeholder)         (testing)        │
│     normal | varying                                                │
│                                                                     │
│   All implement: AirTrafficStreamer protocol                        │
│   All constructed via from_kwargs() + create_streamer() factory     │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

2.1 Current File Structure

src/openutm_verification/core/
├── steps/
│   └── air_traffic_step.py          # ✅ AirTrafficStepClient + "Stream Air Traffic"
│
├── providers/
│   ├── __init__.py                  # ✅ Exports ProviderType, DataQualityType, create_provider
│   ├── protocol.py                  # ✅ AirTrafficProvider protocol
│   ├── factory.py                   # ✅ ProviderType enum, create_provider(), _QUALITY_WRAPPERS registry
│   ├── latency.py                   # ✅ DataQualityType, LatencyProviderWrapper, shift_timestamps()
│   ├── geojson_provider.py          # ✅ Wraps AirTrafficClient (legacy indirection)
│   ├── bluesky_provider.py          # ✅ Wraps BlueSkyClient (legacy indirection)
│   ├── bayesian_provider.py         # ✅ Wraps BayesianTrafficClient (legacy indirection)
│   └── opensky_provider.py          # ✅ Wraps OpenSkyClient (legacy indirection)
│
├── streamers/
│   ├── __init__.py                  # ✅ Exports TargetType, RefreshModeType, create_streamer
│   ├── protocol.py                  # ✅ AirTrafficStreamer protocol, StreamResult
│   ├── factory.py                   # ✅ TargetType enum, create_streamer()
│   ├── flight_blender_streamer.py   # ✅ Normal + varying refresh modes
│   ├── amqp_streamer.py             # ⚠️ Placeholder (logs warning, returns error StreamResult)
│   └── null_streamer.py             # ✅ Returns observations without delivery
│
├── clients/
│   ├── air_traffic/                 # ⏳ Legacy — wrapped by providers, pending Phase D/E
│   │   ├── base_client.py           #    3 Settings classes, 3 base clients
│   │   ├── air_traffic_client.py    #    3 @internal_steps (GeoJSON)
│   │   ├── blue_sky_client.py       #    2 @internal_steps (BlueSky)
│   │   └── bayesian_air_traffic_client.py  # 3 @internal_steps (Bayesian)
│   └── opensky/                     # ⏳ Legacy — wrapped by OpenSkyProvider, pending Phase D/E
│       ├── base_client.py           #    OpenSkySettings, BaseOpenSkyAPIClient
│       └── opensky_client.py        #    1 @internal_step ("Fetch OpenSky Data")

2.2 Key Design Decisions (as implemented)

1. Providers wrap legacy clients rather than replacing them — each provider instantiates an old client internally via from_kwargs().
2. data_quality parameter replaces old *_with_latency step variants — now a provider wrapper concern.
3. refresh_mode parameter replaces old *_at_random_refresh_rates — now a streamer concern.
4. AirTrafficStepClient reads config defaults from AppConfig when step arguments omit them.
5. FlightPhase.CRUISE assigned to the unified step (old steps used FlightPhase.PRE_FLIGHT).

3. Unified Scenario Step (Implemented)

3.1 Step Signature

@scenario_step("Stream Air Traffic", phase=FlightPhase.CRUISE)
async def stream_air_traffic(
    self,
    provider: ProviderType,                          # geojson | bluesky | bayesian | opensky
    duration: int | None = None,                      # Defaults from config if not set
    target: TargetType = TargetType.FLIGHT_BLENDER,   # flight_blender | amqp | none
    *,
    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: DataQualityType = DataQualityType.NOMINAL,   # nominal | latency
    refresh_mode: RefreshModeType = RefreshModeType.NORMAL,    # normal | varying
) -> StreamResult

3.2 StreamResult

@dataclass
class StreamResult:
    success: bool
    provider: str
    target: str
    duration_seconds: int
    total_observations: int
    total_batches: int
    errors: list[str] = field(default_factory=list)
    observations: list[FlightObservationSchema] | None = None

3.3 Scenarios Using Unified Step

# airtraffic-simulations/bluesky_sim_air_traffic_data.yaml
- step: Stream Air Traffic
  arguments:
    provider: bluesky
    target: flight_blender

# airtraffic-simulations/opensky_live_data.yaml
- step: Stream Air Traffic
  arguments:
    provider: opensky
    target: flight_blender

# airtraffic-simulations/bayesian_sim_air_traffic_data.yaml
- step: Stream Air Traffic
  arguments:
    provider: bayesian

# daa-f3442/*.yaml (many scenarios)
- step: Stream Air Traffic
  arguments:
    provider: bluesky
    target: flight_blender

4. Protocols (Implemented)

4.1 Provider Protocol

# core/providers/protocol.py
@runtime_checkable
class AirTrafficProvider(Protocol):
    @property
    def name(self) -> str: ...

    async def get_observations(
        self, duration: int | None = None,
    ) -> list[FlightObservationSchema]: ...

Note: The stream() async iterator method from the original spec was not implemented.
The current approach uses a single get_observations() call per step invocation.

4.2 Streamer Protocol

# core/streamers/protocol.py
@runtime_checkable
class AirTrafficStreamer(Protocol):
    @property
    def name(self) -> str: ...

    async def stream_from_provider(
        self, provider: AirTrafficProvider, duration_seconds: int,
    ) -> StreamResult: ...

Note: The send_batch() method from the original spec was not implemented.
Streamers receive the full observation list in one call.

5. Remaining Work

Phase A–C: ✅ Complete

Docs, config unification, and step unregistration are all done.

Phase D: Collapse Provider Wrappers into Standalone Providers

Currently each provider wraps its legacy client, creating a double-indirection chain:

Scenario YAML
  → AirTrafficStepClient.stream_air_traffic()
    → create_provider("geojson")
      → GeoJSONProvider.get_observations()
        → AirTrafficClient(AirTrafficSettings).generate_simulated_air_traffic_data()
          ↑ @internal_step decorator wraps result in StepResult
        ← unwrap StepResult.result
      ← list[FlightObservationSchema]
    → create_streamer("flight_blender")
      → FlightBlenderStreamer.stream_from_provider()
        → FlightBlenderClient.submit_simulated_air_traffic()
          ↑ @scenario_step decorator wraps result in StepResult
        ← unwrap StepResult.result
      ← StreamResult

Problems with the current indirection:

1. Double StepResult wrapping — providers call @internal_step methods which wrap
   returns in StepResult, then providers immediately unwrap .result. Pointless overhead.
2. Redundant Settings construction — each get_observations() call builds a throwaway
   AirTrafficSettings/BlueSkyAirTrafficSettings/BayesianAirTrafficSettings Pydantic
   model just to pass values to the legacy client that it already has in self._* fields.
3. Unnecessary context managers — providers instantiate async with LegacyClient(...) as client
   per call. The legacy clients' __aenter__/__aexit__ are no-ops or trivial.
4. AirTrafficClient has a confusing diamond inheritance from both BaseAirTrafficAPIClient
   and BaseBlenderAPIClient, with a dummy base_url="" workaround in __init__. This need
   not survive into the provider era.

Target state — inline the generation logic:

GeoJSONProvider.get_observations()  → json.load + GeoJSONAirtrafficSimulator.generate_air_traffic_data()
BlueSkyProvider.get_observations()  → bluesky scenario parsing (existing logic, ~40 LOC)
BayesianProvider.get_observations() → cam-track-gen API (existing logic, ~30 LOC)
OpenSkyProvider.get_observations()  → httpx + OAuth2 token (existing logic, ~50 LOC)

Each provider would own its own small Settings dataclass (or just __init__ params)
and call the simulator/API directly. No StepResult wrapping/unwrapping, no
legacy client instantiation. The FlightBlenderStreamer similarly calls
FlightBlenderClient methods which are @scenario_step-decorated; the streamer
already handles the StepResult unpacking, but after Phase D the streamer could
call FlightBlenderClient submit methods directly or use httpx natively.

Effort: Medium — each provider is ~100 LOC of straightforward extraction.
Risk: Low — the unified step tests in test_stream_air_traffic.py already mock
at the provider boundary, so the refactor is behind a stable interface.

Phase E: Remove Legacy Code

Once Phase D inlines the logic, delete:

Total: ~670 LOC of dead legacy code removed.

6. Optimization Recommendations

6.1 Eliminate Double StepResult Wrapping (Phase D blocker)

The current provider→legacy-client path results in:

1. @internal_step wraps the legacy method return in StepResult
2. Provider immediately unwraps step_result.result

After Phase D, providers call simulator/API directly — no wrapping overhead.

6.2 Deduplicate Provider Boilerplate

All 3 simulation providers (GeoJSON, BlueSky, Bayesian) share identical
__init__, from_kwargs, and get_observations structure (~40 LOC each).
Consider a BaseSimulationProvider that handles common fields:

class BaseSimulationProvider:
    def __init__(self, *, config_path, number_of_aircraft, duration, sensor_ids, session_ids): ...

    @classmethod
    def from_kwargs(cls, **kwargs) -> Self: ...

    async def get_observations(self, duration=None) -> list[FlightObservationSchema]:
        effective_duration = duration or self._duration
        return await self._generate(effective_duration)

    @abstractmethod
    async def _generate(self, duration: int) -> list[FlightObservationSchema]: ...

This makes each concrete provider ~15 LOC (just name + _generate).
Only pursue this after Phase D lands.

6.3 FlightBlenderStreamer StepResult Handling Fragility

FlightBlenderStreamer.stream_from_provider() has a complex branch handling
StepResult | dict | other returns from FlightBlenderClient.submit_*. This is
because the client's submit methods are @scenario_step-decorated, so callers get
StepResult instead of the raw dict. Options:

• Short-term: Extract the unwrap logic into a small helper (e.g., unwrap_step_result()).
• Long-term (Phase D): Have the streamer call the HTTP submission directly via httpx,
  bypassing the @scenario_step wrapping entirely.

6.4 Provider Factory — Use from_kwargs Consistently

create_provider() currently passes all kwargs through to Provider.from_kwargs().
The factory function duplicates the parameter signature of from_kwargs. Consider
simplifying to just forward **kwargs:

def create_provider(name: ProviderType, *, data_quality=DataQualityType.NOMINAL, **kwargs):
    provider = _PROVIDERS[name].from_kwargs(**kwargs)
    wrapper = _QUALITY_WRAPPERS.get(data_quality)
    return wrapper(provider) if wrapper else provider

This removes the parameter duplication between factory and providers.

6.5 AirTrafficStepClient Has No State

AirTrafficStepClient has empty __aenter__/__aexit__ and no instance state.
It exists purely to satisfy the DI system's expectation of a client class.
This is acceptable for now, but if the DI system is ever extended to support
stateless step functions directly, this wrapper could be removed.

6.6 AMQP Streamer Decision

The AMQP streamer is a placeholder that logs a warning and returns an error StreamResult.
Either implement it when AMQP delivery is needed, or remove it from
TargetType to avoid confusing users who try target: amqp and get silent failures.

7. Prioritized Next Steps

8. Open Questions

1. Should target support multiple destinations? (Not needed currently)
2. Should providers support real-time vs batch mode explicitly? (Not implemented; single get_observations() call is sufficient)
3. Should simulation_duration field name be standardized? Resolved: all use simulation_duration with parse_duration validator.
4. Should the AMQP streamer be fully implemented, kept as placeholder, or removed from TargetType?
5. Should Phase D extract a BaseSimulationProvider ABC, or keep providers fully independent?

9. References

• src/openutm_verification/core/steps/air_traffic_step.py — Unified step implementation
• src/openutm_verification/core/providers/ — Provider protocol, factory, implementations
• src/openutm_verification/core/streamers/ — Streamer protocol, factory, implementations
• src/openutm_verification/core/execution/dependencies.py — AirTrafficStepClient DI registration
• src/openutm_verification/core/execution/scenario_runner.py — @internal_step decorator definition
• src/openutm_verification/core/execution/config_models.py — AirTrafficSimulatorSettings, AppConfig
• src/openutm_verification/core/clients/air_traffic/ — Legacy clients (wrapped by providers, pending removal)
• src/openutm_verification/core/clients/opensky/ — Legacy OpenSky client (wrapped by provider, pending removal)
• tests/test_stream_air_traffic.py — Unified step tests (factories, latency, integration)