Source code reference

The routes module builds the router for the provided endpoint and adds any endpoints only available for BluebirdATC, including loading which is implementation dependent.

load(category, scenario_name, runner_store) async

End any existing run, then create a new Runner and load a given simulator scenario.

Source code in bluebird_api/routes.py

@router.post("/load/{category}/{scenario_name}", tags=["Control"])
async def load(category: str, scenario_name: str, runner_store: RunnerStoreDep) -> bool:  # noqa: ARG001
    """
    End any existing run, then create a new Runner and load a given simulator scenario.
    """
    if runner_store.current_runner is not None:
        await runner_store.delete()

    runner_store.initialise_from_category(category, scenario_name)

    runner_store.start()

    return True

ActionInput

Bases: BaseModel

An action to be sent to the simulator.

Source code in bluebird_api/models.py

class ActionInput(BaseModel):
    """An action to be sent to the simulator."""

    agent: str = Field(
        description="The agent performing the action.",
        json_schema_extra={"example": "atc_1"},
    )
    callsign: str = Field(
        description="The callsign of the flight being acted on.",
        json_schema_extra={"example": "AIR123"},
    )
    kind: typing.Literal[tuple(SUPPORTED_ACTIONS_LIST)] = Field(
        description="The kind of action to perform. Must be a string that is one of the supported actions.",
        json_schema_extra={"example": "change_heading_to"},
    )
    value: typing.Any = Field(
        description=(
            "The value associated with the action. The type of this value depends on the action being performed."
        ),
        json_schema_extra={"example": 90},
    )
    sector: str = Field(
        description="The sector that the action is being performed in.",
        json_schema_extra={"example": "sector_1"},
    )

This module provides the interfaces used in all endpoints to obtain data from the runner and simulator that is being requested.

This module provides the FastAPI dependency for the runner, used to access the simulator instance, RunnerDep. To use this dependency, create an endpoint as would be done normally. As one of the arguments to the function, include the RunnerDep type alias as shown in the example below and it will be available to interact with.

from ..runner import RunnerDep @core_router.get("/wind_field", tags=["State"]) async def wind_field(runner: RunnerDep) -> WindField | None: return runner.sim.manager.environment.wind_field

If during resolution of the runner, for example trying to find the runner, the runner is not available, a HTTP error 404 (Not found) will be returned before even running the function above.

Runner dataclass

Bases: Generic[TSimulator]

Source code in bluebird_api/runner.py

@dataclass(init=True, slots=True)
class Runner(typing.Generic[TSimulator]):
    sim: TSimulator
    task: asyncio.Task | None = None
    running: bool = False
    evolve_period: float = 6
    tick_frequency_period: float = 6
    tick: int = 0
    kill: bool = False
    time_of_next_tick: datetime = datetime.min
    hmi: defaultdict[str, HmiRunnerInformation] = field(
        default_factory=lambda: defaultdict(lambda: HmiRunnerInformation(selected_aircraft=None))
    )

    def start(self):
        self.task = asyncio.create_task(self.run_main())

    async def close(self):
        self.kill = True

        if self.task is not None:
            try:
                await asyncio.wait_for(self.task, timeout=10)
            except Exception as e:
                logger.exception(f"Runner task failed during shutdown: {e}")
            finally:
                self.task = None

        await self.sim.async_save(autosave=False, end_save=True)
        await self.sim.async_close()
        await asyncio.sleep(3)

    def log_simrate(self):
        """
        Creates an entry in the file logs with the current tick frequency and evolve period of the runner.

        See documentation for the event_logger.log_sim_event for more information.
        """
        self.sim.manager.event_logger.log_sim_event(
            SimRateUpdate(
                simulation_datetime=self.sim.manager.environment.datetime,
                tick_frequency=self.tick_frequency_period,
                evolve_period=self.evolve_period,
            )
        )

    async def run_main(self):
        self.time_of_next_tick = datetime.now()

        while True:
            if self.kill:
                break

            if self.running and datetime.now() >= self.time_of_next_tick:
                start_time = datetime.now()
                self.time_of_next_tick = start_time + timedelta(seconds=self.tick_frequency_period)

                await self.sim.async_evolve(self.evolve_period)
                self.tick += 1
                logger.info(f"evolve time: {datetime.now() - start_time}")

            await asyncio.sleep(0.1)

log_simrate()

Creates an entry in the file logs with the current tick frequency and evolve period of the runner.

See documentation for the event_logger.log_sim_event for more information.

Source code in bluebird_api/runner.py

def log_simrate(self):
    """
    Creates an entry in the file logs with the current tick frequency and evolve period of the runner.

    See documentation for the event_logger.log_sim_event for more information.
    """
    self.sim.manager.event_logger.log_sim_event(
        SimRateUpdate(
            simulation_datetime=self.sim.manager.environment.datetime,
            tick_frequency=self.tick_frequency_period,
            evolve_period=self.evolve_period,
        )
    )

runner(request) async

Function taking the runner information from the store, and making it available for the endpoint that uses it. See module documentation for more details on usage, and an example.

Source code in bluebird_api/runner.py

async def runner(request: Request) -> Runner[Simulator]:  # noqa: ARG001
    """
    Function taking the runner information from the store, and making it available for the endpoint that uses it. See
    module documentation for more details on usage, and an example.
    """
    runner = request.app.state.runner_store.current_runner

    if runner is None:
        raise HTTPException(404, "Runner instance not found")

    if not isinstance(runner, Runner):
        raise HTTPException(500, "The runner passed is not a valid type.")

    return runner

runner_store(request) async

Function taking the runner store.

Source code in bluebird_api/runner.py

async def runner_store(request: Request) -> RunnerStore[Runner, Simulator]:
    """
    Function taking the runner store.
    """
    return request.app.state.runner_store

This module contains all the routers for the different features available in BluebirdATC. Note that, unlike in compiled programming languages where disabled features are not included in the machine code, not including certain routers just means that the endpoint is not available but the logic will continue to be available in BluebirdATC, and therefore logged.

All the following routers are independent on the implementation of the storage of the runners and get this through the fastapi dependency. See the respective documentation in ../runnerabc.py.

Note the endpoint for loading a run, that the HMI and most clients expect, is not included here as they are dependent on the implementation of the store, which is dependent on the use case. An example of this endpoint, and any other endpoints not included within the groups defined here are available in ../routes.py