Skip to content

Source code reference

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

load(category, scenario_name) async

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

Source code in bluebird_api/routes.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
@router.post("/load/{category}/{scenario_name}", tags=["Control"])
async def load(category: str, scenario_name: str) -> bool:  # noqa: ARG001
    """
    End any existing run, then create a new Runner and load a given simulator scenario.
    """

    if RunnerStore.current_runner is not None:
        await RunnerStore.current_runner.delete()

    RunnerStore.current_runner = Runner(category, scenario_name)

    # start the task
    task = asyncio.create_task(RunnerStore.current_runner.run_main())

    # add the task to the background tasks set and have it auto-remove its reference from the set when done
    background_tasks.add(task)
    task.add_done_callback(background_tasks.remove)

    return True

simulator(request) async

The simulator dependency, available to endpoints using the RunnerDep dependency, provides access to the runner.

Source code in bluebird_api/routes.py 
19
20
21
22
23
24
async def simulator(request: Request):
    """
    The simulator dependency, available to endpoints using the RunnerDep dependency, provides access to the runner.
    """

    request.state.runner = RunnerStore.current_runner

ActionInput

Bases: BaseModel

An action to be sent to the simulator.

Source code in bluebird_api/models.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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.

The first thing available is the RunnerABC class, which is implemented for BluebirdATC in ./runner.py.

Secondly, and most importantly, 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 ..runnerabc import RunnerDep @core_router.post("/close", tags=["Control"]) async def close(runner: RunnerDep) -> bool: await runner.delete() return True

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.

RunnerABC

Bases: ABC

Source code in bluebird_api/runnerabc.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
class RunnerABC(ABC):
    kill: bool
    category: str
    scenario_name: str
    sim: Simulator
    running: bool
    evolve_period: float
    tick_frequency_period: float
    tick: int
    hmi: dict

    @abstractmethod
    def initialise_simulator(self, *args, **kwargs) -> Simulator:  # noqa: ANN002, ANN003
        """
        Function to instantiate the simulator class, or variations of it for each use case.

        This function is designed to be a transparent function taking all the arguments passed into the constructor of
        the runner class from the load endpoint.
        """
        pass

    @abstractmethod
    async def delete(self):
        pass

    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,
            )
        )

    def __init__(self, category: str, scenario_name: str, log_name: str | None = None):
        """
        Constructor of the Runner classes.

        Although the argumets of this function are currently hardcoded, if a usecase requires different ones they will
        be replaced by args and kwargs placeholders.

        Arguments
        ---------
        category: str
            The category of scenario to load
        scenario_name: str
            The specific scenario of the category to load
        log_name: str
            The name to store the logs for the run.
        """
        self.category = category
        self.scenario_name = scenario_name
        self.sim = self.initialise_simulator(self.category, self.scenario_name, log_filename=log_name)
        self.running = False
        self.evolve_period = 6.0
        self.tick_frequency_period = 6.0
        self.kill = False
        self.time_of_next_tick = datetime.min
        self.tick = 0
        self.hmi = defaultdict(lambda: {"selected_aircraft": ""})

        # allow 'None' and 'ALL' sectors to also select aircraft
        self.hmi["None"] = {"selected_aircraft": ""}
        self.hmi["ALL"] = {"selected_aircraft": ""}
        # try a lock to avoid concurrency bugs
        self._sim_lock = asyncio.Lock()

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

        while True:
            if self.running and datetime.now() >= self.time_of_next_tick:
                async with self._sim_lock:
                    start_time = datetime.now()
                    self.time_of_next_tick = start_time + timedelta(seconds=self.tick_frequency_period)

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

            if self.kill:
                self.category = None
                self.scenario_name = None
                break

            await asyncio.sleep(0.1)

__init__(category, scenario_name, log_name=None)

Constructor of the Runner classes.

Although the argumets of this function are currently hardcoded, if a usecase requires different ones they will be replaced by args and kwargs placeholders.

Arguments

category: str The category of scenario to load scenario_name: str The specific scenario of the category to load log_name: str The name to store the logs for the run.

Source code in bluebird_api/runnerabc.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def __init__(self, category: str, scenario_name: str, log_name: str | None = None):
    """
    Constructor of the Runner classes.

    Although the argumets of this function are currently hardcoded, if a usecase requires different ones they will
    be replaced by args and kwargs placeholders.

    Arguments
    ---------
    category: str
        The category of scenario to load
    scenario_name: str
        The specific scenario of the category to load
    log_name: str
        The name to store the logs for the run.
    """
    self.category = category
    self.scenario_name = scenario_name
    self.sim = self.initialise_simulator(self.category, self.scenario_name, log_filename=log_name)
    self.running = False
    self.evolve_period = 6.0
    self.tick_frequency_period = 6.0
    self.kill = False
    self.time_of_next_tick = datetime.min
    self.tick = 0
    self.hmi = defaultdict(lambda: {"selected_aircraft": ""})

    # allow 'None' and 'ALL' sectors to also select aircraft
    self.hmi["None"] = {"selected_aircraft": ""}
    self.hmi["ALL"] = {"selected_aircraft": ""}
    # try a lock to avoid concurrency bugs
    self._sim_lock = asyncio.Lock()

initialise_simulator(*args, **kwargs) abstractmethod

Function to instantiate the simulator class, or variations of it for each use case.

This function is designed to be a transparent function taking all the arguments passed into the constructor of the runner class from the load endpoint.

Source code in bluebird_api/runnerabc.py 
46
47
48
49
50
51
52
53
54
@abstractmethod
def initialise_simulator(self, *args, **kwargs) -> Simulator:  # noqa: ANN002, ANN003
    """
    Function to instantiate the simulator class, or variations of it for each use case.

    This function is designed to be a transparent function taking all the arguments passed into the constructor of
    the runner class from the load endpoint.
    """
    pass

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/runnerabc.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
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 state, and making it available for the endpoint that uses it. See module documentation for more details on usage, and an example.

This FastAPI dependency will throw a HTTP exception if the runner isinstance is not found, therefore this does not need to be handled by the individual endpoint.

Source code in bluebird_api/runnerabc.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
async def runner(request: Request) -> RunnerABC:  # noqa: ARG001
    """
    Function taking the runner information from the state, and making it available for the endpoint that uses it. See
    module documentation for more details on usage, and an example.

    This FastAPI dependency will throw a HTTP exception if the runner isinstance is not found, therefore this does not
    need to be handled by the individual endpoint.
    """
    runner = request.state.runner

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

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

    request.state.runner = None
    return runner

The runner implementation for the general BluebirdATC API. This runner is responsible for the initialisation of the simulator class.

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