Skip to content

Bridge & Backends

The bridge adapts the OpenLab API to a concrete execution backend. It is the only place that knows about hivebr.engine (or, in the future, ROS 2), keeping the public API backend-agnostic.

bridge

Backend

Bases: ABC

Execution backend contract shared by the simulator and ROS 2 arena.

A backend is the only layer allowed to talk to a concrete engine, so that the public API never imports pygame or ROS 2 directly.

reset abstractmethod 

reset()
Source code in src/hivebr/bridge.py 
16
17
@abstractmethod
def reset(self) -> None: ...

get_states abstractmethod 

get_states()
Source code in src/hivebr/bridge.py 
19
20
@abstractmethod
def get_states(self) -> dict[int, RobotState]: ...

compute abstractmethod 

compute(states, dt)

Return one command per robot id from a frozen fleet snapshot.

Source code in src/hivebr/bridge.py 
22
23
24
@abstractmethod
def compute(self, states: dict[int, RobotState], dt: float) -> dict[int, Command]:
    """Return one command per robot id from a frozen fleet snapshot."""

step abstractmethod 

step(commands, dt)
Source code in src/hivebr/bridge.py 
26
27
@abstractmethod
def step(self, commands: dict[int, Command], dt: float) -> None: ...

run abstractmethod 

run()
Source code in src/hivebr/bridge.py 
29
30
@abstractmethod
def run(self) -> None: ...

close abstractmethod 

close()
Source code in src/hivebr/bridge.py 
32
33
@abstractmethod
def close(self) -> None: ...

NavSim2DBackend 

NavSim2DBackend(experiment)

Bases: Backend

Backend for local simulation using the hivebr.engine renderer/physics.

Source code in src/hivebr/bridge.py 
39
40
41
def __init__(self, experiment) -> None:
    self.experiment = experiment
    self._sim = None

experiment instance-attribute 

experiment = experiment

reset 

reset()
Source code in src/hivebr/bridge.py 
43
44
45
def reset(self) -> None:
    if self._sim is not None:
        self._sim.reset()

get_states 

get_states()
Source code in src/hivebr/bridge.py 
47
48
49
50
51
52
53
54
55
def get_states(self) -> dict[int, RobotState]:
    if self._sim is None:
        return {}
    robots = getattr(self._sim, "robots", [self._sim.robot])
    states = {}
    for i, robot in enumerate(robots):
        goal = self._sim.robot_goal(i) if hasattr(self._sim, "robot_goal") else None
        states[i] = _state_from_robot(i, robot, goal)
    return states

compute 

compute(states, dt)
Source code in src/hivebr/bridge.py 
57
58
def compute(self, states: dict[int, RobotState], dt: float) -> dict[int, Command]:
    return self.experiment._dispatch(states, dt)

step 

step(commands, dt)
Source code in src/hivebr/bridge.py 
60
61
62
63
64
def step(self, commands: dict[int, Command], dt: float) -> None:
    raise NotImplementedError(
        "Headless stepping is not available yet; use run() which drives the "
        "engine's interactive loop through the fleet controller."
    )

run 

run()
Source code in src/hivebr/bridge.py 
66
67
68
69
70
71
def run(self) -> None:
    from .engine import Simulator

    config = self.experiment.build_engine_config()
    self._sim = Simulator(config=config, fleet_controller=self._fleet_controller)
    self._sim.run()

close 

close()
Source code in src/hivebr/bridge.py 
73
74
def close(self) -> None:
    self._sim = None

Ros2Backend 

Ros2Backend(experiment)

Bases: Backend

Placeholder for the physical arena: one ROS 2 node per agent or a single overarching node for a SwarmController.

Source code in src/hivebr/bridge.py 
89
90
def __init__(self, experiment) -> None:
    self.experiment = experiment

experiment instance-attribute 

experiment = experiment

reset 

reset()
Source code in src/hivebr/bridge.py 
92
93
def reset(self) -> None:
    raise NotImplementedError("ROS 2 backend not implemented yet.")

get_states 

get_states()
Source code in src/hivebr/bridge.py 
95
96
def get_states(self) -> dict[int, RobotState]:
    raise NotImplementedError("ROS 2 backend not implemented yet.")

compute 

compute(states, dt)
Source code in src/hivebr/bridge.py 
98
99
def compute(self, states: dict[int, RobotState], dt: float) -> dict[int, Command]:
    return self.experiment._dispatch(states, dt)

step 

step(commands, dt)
Source code in src/hivebr/bridge.py 
101
102
def step(self, commands: dict[int, Command], dt: float) -> None:
    raise NotImplementedError("ROS 2 backend not implemented yet.")

run 

run()
Source code in src/hivebr/bridge.py 
104
105
def run(self) -> None:
    raise NotImplementedError("ROS 2 backend not implemented yet.")

close 

close()
Source code in src/hivebr/bridge.py 
107
108
def close(self) -> None:
    pass

create_backend 

create_backend(name, experiment)
Source code in src/hivebr/bridge.py 
137
138
139
140
141
142
143
def create_backend(name: str, experiment) -> Backend:
    key = str(name).lower()
    if key in ("navsim2d", "nav_sim2d", "local", "sim"):
        return NavSim2DBackend(experiment)
    if key in ("ros2", "ros", "arena"):
        return Ros2Backend(experiment)
    raise ValueError(f"Unknown backend: {name!r} (use 'navsim2d' or 'ros2')")