Core

The OpenLab API entry points.

core

HiveBrAgent

HiveBrAgent(robot_id=None, config=None, **params)

Bases: ABC

Decentralized (agent-centric) behavior for a single B-bot.

The engine instantiates this class once per robot and calls loop with a frozen state snapshot plus the states of its neighbors.

Source code in src/hivebr/core.py

def __init__(self, robot_id: int | None = None, config: dict | None = None,
             **params):
    self.id = robot_id
    # The engine config is exposed so controllers can read values such as
    # `waypoints.goal_tolerance` straight from the YAML file.
    self.cfg = config
    self.config = config
    self.params = params

id instance-attribute

id = robot_id

cfg instance-attribute

cfg = config

config instance-attribute

config = config

params instance-attribute

params = params

setup

setup()

Optional one-time initialization.

Source code in src/hivebr/core.py

def setup(self) -> None:
    """Optional one-time initialization."""

loop abstractmethod

loop(state, neighbors)

Return the desired (v, omega) for this agent.

Source code in src/hivebr/core.py

@abstractmethod
def loop(
    self,
    state: RobotState,
    neighbors: dict[int, RobotState],
) -> "Command | tuple[float, float]":
    """Return the desired (v, omega) for this agent."""

reset

reset()

Optional per-episode reset.

Source code in src/hivebr/core.py

def reset(self) -> None:
    """Optional per-episode reset."""

SwarmController

Bases: ABC

Centralized (global-centric) control for the whole fleet.

The engine instantiates this class exactly once and passes the states of all robots, expecting one command per robot id back.

setup

setup()

Optional one-time initialization.

Source code in src/hivebr/core.py

def setup(self) -> None:
    """Optional one-time initialization."""

loop abstractmethod

loop(states)

Return a mapping {robot_id: command}.

Source code in src/hivebr/core.py

@abstractmethod
def loop(self, states: dict[int, RobotState]) -> dict[int, "Command | tuple[float, float]"]:
    """Return a mapping ``{robot_id: command}``."""

reset

reset()

Optional per-episode reset.

Source code in src/hivebr/core.py

def reset(self) -> None:
    """Optional per-episode reset."""

SwarmExperiment

SwarmExperiment(
    robots=1,
    backend="navsim2d",
    config=None,
    dt=0.03,
    orientation=None,
    goal_tolerance=None,
    comm_radius=None,
)

Entry point that binds user code to a backend (navsim2d or ROS 2).

Source code in src/hivebr/core.py

def __init__(
    self,
    robots=1,
    backend: str = "navsim2d",
    config=None,
    dt: float = 0.03,
    orientation: bool | None = None,
    goal_tolerance: float | None = None,
    comm_radius: float | None = None,
):
    self.dt = float(dt)
    self.backend_name = backend
    self.config = config
    self.orientation = orientation
    self.goal_tolerance = goal_tolerance
    self.comm_radius = comm_radius
    self.agent_classes: dict[int, type[HiveBrAgent]] = {}
    self.agent_params: dict[int, dict] = {}
    self.routes: dict[int, list] = {}
    self.controller: SwarmController | None = None
    self.agents: dict[int, HiveBrAgent] = {}
    self._mode: str | None = None
    self._backend = None
    self._num_robots, self._fleet = _parse_robots(robots)

dt instance-attribute

dt = float(dt)

backend_name instance-attribute

backend_name = backend

config instance-attribute

config = config

orientation instance-attribute

orientation = orientation

goal_tolerance instance-attribute

goal_tolerance = goal_tolerance

comm_radius instance-attribute

comm_radius = comm_radius

agent_classes instance-attribute

agent_classes = {}

agent_params instance-attribute

agent_params = {}

routes instance-attribute

routes = {}

controller instance-attribute

controller = None

agents instance-attribute

agents = {}

simulator property

simulator

Simulador (engine) usado pelo backend navsim2d, se já criado.

Útil para acessar sim.logger depois de run(), inclusive em experimentos descentralizados.

assign_agent

assign_agent(
    robot_id=None,
    robot_ids=None,
    agent_class=None,
    **params
)

Source code in src/hivebr/core.py

def assign_agent(self, robot_id=None, robot_ids=None, agent_class=None, **params) -> "SwarmExperiment":
    if agent_class is None:
        raise ValueError("assign_agent requires an agent_class.")
    ids = []
    if robot_id is not None:
        ids.append(int(robot_id))
    if robot_ids is not None:
        ids.extend(int(i) for i in robot_ids)
    for rid in ids:
        self.agent_classes[rid] = agent_class
        self.agent_params[rid] = dict(params)
    return self

set_controller

set_controller(controller)

Source code in src/hivebr/core.py

def set_controller(self, controller: SwarmController) -> "SwarmExperiment":
    self.controller = controller
    return self

assign_route

assign_route(robot_id, waypoints)

Give a B-bot its waypoint route (each item (x, y) or (x, y, theta)).

Source code in src/hivebr/core.py

def assign_route(self, robot_id, waypoints) -> "SwarmExperiment":
    """Give a B-bot its waypoint route (each item ``(x, y)`` or ``(x, y, theta)``)."""
    self.routes[int(robot_id)] = [list(wp) for wp in waypoints]
    return self

reset

reset()

Source code in src/hivebr/core.py

def reset(self) -> None:
    for agent in self.agents.values():
        agent.reset()
    if self.controller is not None:
        self.controller.reset()
    if self._backend is not None:
        self._backend.reset()

step

step()

Source code in src/hivebr/core.py

def step(self) -> None:
    backend = self._ensure_backend()
    states = backend.get_states()
    commands = self._dispatch(states, self.dt)
    backend.step(commands, self.dt)

run

run()

Source code in src/hivebr/core.py

def run(self) -> None:
    self._build()
    self._ensure_backend().run()

save_log

save_log(path='log.pkl')

Salva o DataLogger do simulador (.pkl ou .csv pela extensão).

Source code in src/hivebr/core.py

def save_log(self, path: str = "log.pkl") -> None:
    """Salva o DataLogger do simulador (``.pkl`` ou ``.csv`` pela extensão)."""
    sim = self.simulator
    if sim is None or getattr(sim, "logger", None) is None:
        print("[SwarmExperiment] Nenhum log disponível (rode o experimento antes).")
        return
    sim.logger.export(path)

plot_log

plot_log()

Abre a janela de plot do DataLogger do simulador.

Source code in src/hivebr/core.py

def plot_log(self) -> None:
    """Abre a janela de plot do DataLogger do simulador."""
    sim = self.simulator
    if sim is None or getattr(sim, "logger", None) is None:
        print("[SwarmExperiment] Nenhum log disponível (rode o experimento antes).")
        return
    sim.logger.plot()

print_collisions

print_collisions()

Imprime as colisões registradas no DataLogger do simulador.

Source code in src/hivebr/core.py

def print_collisions(self) -> None:
    """Imprime as colisões registradas no DataLogger do simulador."""
    sim = self.simulator
    if sim is None or getattr(sim, "logger", None) is None:
        print("[SwarmExperiment] Nenhum log disponível (rode o experimento antes).")
        return
    sim.logger.print_collisions()

close

close()

Source code in src/hivebr/core.py

def close(self) -> None:
    if self._backend is not None:
        self._backend.close()
        self._backend = None

build_engine_config

build_engine_config()

Source code in src/hivebr/core.py

def build_engine_config(self) -> dict:
    cfg = deepcopy(self.config) if self.config is not None else None
    if cfg is None:
        from .engine.simulator import DEFAULT_CONFIG

        cfg = deepcopy(DEFAULT_CONFIG)
    cfg["robots"] = [deepcopy(meta) for meta in self._fleet]
    for index, meta in enumerate(cfg["robots"]):
        meta.setdefault("initial_pose", list(_spread_pose(index)))
        if index in self.routes:
            meta["waypoints"] = [list(wp) for wp in self.routes[index]]

    cfg.setdefault("waypoints", {})
    if self.orientation is not None:
        cfg["waypoints"]["define_orientation"] = bool(self.orientation)
    if self.goal_tolerance is not None:
        cfg["waypoints"]["goal_tolerance"] = float(self.goal_tolerance)
    return cfg