pybullet_fleet.controllers package

Submodules

pybullet_fleet.controllers.batch_base module

Shared base class for vectorized (batched) kinematic controllers.

A BatchKinematicController manages N agents at once using NumPy arrays for per-agent state, replacing the per-agent Python dispatch that KinematicController.compute() performs in the default path.

Lifecycle

  1. User constructs the controller (no agents yet).

  2. User registers each agent via c.register_agent(agent). The first registration auto-binds the controller to the agent’s sim_core and appends it to sim_core._batch_controllers so step_once() will drive it. All subsequent agents must belong to the same sim_core.

  3. Each sim step, sim_core.step_once() calls c.batch_advance(dt) during Phase 1, which computes new poses for all registered agents and writes them via the buffered batch API sim_core.set_poses().

  4. Agents may be unregistered at runtime. When the last agent leaves, the controller auto-detaches from sim_core._batch_controllers.

Subclass responsibilities

Concrete subclasses implement batch_advance(dt) and any controller-specific state (e.g. trajectory parameters). Subclasses should call self._resize_state(n) from _on_agents_changed to keep their own arrays sized with self._agents.

class pybullet_fleet.controllers.batch_base.BatchKinematicController

Bases: Controller

Vectorized base for kinematic controllers that drive many agents at once.

Inherits from Controller directly (not KinematicController): the per-agent compute(agent, dt) -> bool contract does not apply here since batch controllers act through batch_advance() during Phase 1 of sim_core.step_once(). Per-agent kinematic parameters (max_vel, accel, etc.) are read from each Agent at set_path time.

To register a custom batch controller, set _registry_name on the subclass and import the module once:

class MyBatchController(BatchKinematicController):
    _registry_name = "my_batch"

    def batch_advance(self, dt): ...

# Usage:
#   AgentManager(sim_core=sim, fleet_controller={"type": "my_batch"})
register_agent(agent)

Register agent with this batch controller.

On the first call the controller records agent.sim_core so that _apply_phase1() can call set_poses. The controller is not attached to sim_core._batch_controllers directly; step_once() discovers it through the owning AgentManager that is registered with the sim.

Return type:

int

Returns:

The agent’s row index in this controller’s state arrays.

Raises:

ValueError – If the agent is already registered, has no sim_core, or belongs to a different sim_core than previously-registered agents.

unregister_agent(agent)

Unregister agent. Compacts the state arrays by swapping with last row.

Raises:

KeyError – If the agent is not registered.

Return type:

None

reset()

Unregister all agents and reset arrays.

Return type:

None

synchronized_set_path(agent, path, **kwargs)

Set one path without racing a concurrent simulation step.

Return type:

None

synchronized_cancel_path(agent)

Cancel one path without racing a concurrent simulation step.

Return type:

None

synchronized_batch_advance(dt)

Advance vectorized trajectory state without concurrent path writes.

Return type:

ndarray

abstractmethod batch_advance(dt)

Compute one timestep for all registered agents.

Implementations must:

  1. Fill self._pos_buf and self._orn_buf with new poses for all agents.

  2. Fill self._moved_mask with True for agents whose pose changed.

  3. Call self._apply_phase1() to write the moved rows back to sim.

Return type:

ndarray

Returns:

self._moved_mask — a (N,) boolean array.

compute(agent, dt)

Compute one step of control.

Return type:

bool

Returns:

True if the agent moved, False otherwise.

set_velocity(**kwargs)
Return type:

None

pybullet_fleet.controllers.batch_base.resolve_batch_controller_key(entry)

Resolve a batch controller from a registry name or a dotted import path.

Mirrors the type: / class: resolution used by per-agent controllers and plugins:

  • Registry name (e.g. "batch_omni") — looked up case-insensitively in BATCH_CONTROLLER_REGISTRY (populated when a subclass sets _registry_name).

  • Dotted path (e.g. "my_pkg.MyBatchController") — imported dynamically and validated to be a BatchKinematicController subclass. Lets custom batch controllers be selected from YAML without pre-importing their module.

Raises:

ValueError – If entry is neither a known registry name nor a dotted path resolving to a BatchKinematicController subclass.

Return type:

Type[BatchKinematicController]

pybullet_fleet.controllers.batch_differential module

Vectorized differential-drive batch controller.

Manages many differential-drive agents at once using NumPy arrays. Each registered agent runs the standard ROTATE → FORWARD lifecycle per waypoint:

  1. ROTATE: trapezoidal-velocity TPI on rotation angle + quaternion slerp.

  2. FORWARD: trapezoidal-velocity TPI on straight-line distance.

A single batch_advance(dt) call evaluates both phases for all agents without per-agent Python dispatch and writes the resulting poses via sim_core.set_poses.

Scope

  • Pose mode only (path following). No set_velocity.

  • MovementDirection.FORWARD, BACKWARD, and AUTO are supported.

  • final_orientation_align performs an in-place rotation to match path[-1].orientation after reaching the last waypoint.

  • 2D navigation friendly (navigation_2d=True flattens goal z).

Numerical equivalence with DifferentialController for the supported scope is within ~1e-6 per step.

class pybullet_fleet.controllers.batch_differential.BatchDifferentialController

Bases: BatchKinematicController

Batched differential-drive pose controller.

set_path(agent, path, direction=None, final_orientation_align=True, **kwargs)

Set a waypoint path.

Parameters:

  • agent (Agent) – A registered agent.

  • path (List[Pose]) – Non-empty list of goal poses.

  • direction (Optional[MovementDirection]) – FORWARD (default), BACKWARD, or AUTO (re-evaluated per waypoint based on heading delta). None defaults to MovementDirection.FORWARD.

  • final_orientation_align (bool) – If True (default), rotate to match path[-1].orientation after reaching the last waypoint.

Return type:

None

batch_advance(dt)

Compute one timestep for all registered agents.

Implementations must:

  1. Fill self._pos_buf and self._orn_buf with new poses for all agents.

  2. Fill self._moved_mask with True for agents whose pose changed.

  3. Call self._apply_phase1() to write the moved rows back to sim.

Return type:

ndarray

Returns:

self._moved_mask — a (N,) boolean array.

pybullet_fleet.controllers.batch_omni module

Vectorized omnidirectional batch controller.

Manages many omni agents at once using NumPy arrays. For each registered agent we store a single straight-line trapezoidal-velocity trajectory; one batch_advance(dt) evaluates all trajectories without per-agent Python dispatch and writes the resulting poses via sim_core.set_poses.

Scope

  • Pose mode only (path following). Velocity commands are not supported in the batched path; agents needing set_velocity should use the per-agent OmniController instead.

  • Multi-waypoint paths are supported.

  • final_orientation_align=True (default) performs an in-place slerp rotation to match path[-1].orientation after the last waypoint.

Numerical equivalence with OmniController for the supported scope is within ~1e-6 per step.

class pybullet_fleet.controllers.batch_omni.BatchOmniController

Bases: BatchKinematicController

Batched omnidirectional pose controller.

set_path(agent, path, final_orientation_align=True, **kwargs)

Set a waypoint path.

Parameters:

  • agent (Agent) – A registered agent.

  • path (List[Pose]) – Non-empty list of goal poses.

  • final_orientation_align (bool) – If True (default), rotate to match path[-1].orientation after reaching the last waypoint.

Return type:

None

batch_advance(dt)

Compute one timestep for all registered agents.

Implementations must:

  1. Fill self._pos_buf and self._orn_buf with new poses for all agents.

  2. Fill self._moved_mask with True for agents whose pose changed.

  3. Call self._apply_phase1() to write the moved rows back to sim.

Return type:

ndarray

Returns:

self._moved_mask — a (N,) boolean array.

pybullet_fleet.controllers.patrol_controller module

Waypoint patrol controller.

class pybullet_fleet.controllers.patrol_controller.PatrolController(params=None, waypoints=None, wait_time=0.0, loop=True)

Bases: Controller

Cycle through waypoints. Delegates actual movement to the base controller by calling agent.set_goal_pose().

Parameters:

  • waypoints (list | None) – List of [x, y, z] positions.

  • wait_time (float) – Seconds to wait at each waypoint before advancing.

  • loop (bool) – If True, restart from first waypoint after reaching last.

compute(agent, dt)

Compute one step of control.

Return type:

bool

Returns:

True if the agent moved, False otherwise.

pybullet_fleet.controllers.random_walk_controller module

Random walk controller.

class pybullet_fleet.controllers.random_walk_controller.RandomWalkController(params=None, radius=5.0, wait_range=(1.0, 5.0))

Bases: Controller

Move to random nearby positions within a radius of the starting point.

Parameters:

  • radius (float) – Maximum distance from origin for random targets.

  • wait_range (tuple | list) – (min, max) seconds to wait at each target.

compute(agent, dt)

Compute one step of control.

Return type:

bool

Returns:

True if the agent moved, False otherwise.

Module contents

High-level controllers for agent behavior.

Includes both behaviour controllers (patrol, random walk) and vectorized batch controllers (BatchKinematicController, BatchOmniController) for the multi-agent NumPy hot path. See docs/architecture/two-phase-step.md for the batch design.

class pybullet_fleet.controllers.BatchDifferentialController

Bases: BatchKinematicController

Batched differential-drive pose controller.

set_path(agent, path, direction=None, final_orientation_align=True, **kwargs)

Set a waypoint path.

Parameters:

  • agent (Agent) – A registered agent.

  • path (List[Pose]) – Non-empty list of goal poses.

  • direction (Optional[MovementDirection]) – FORWARD (default), BACKWARD, or AUTO (re-evaluated per waypoint based on heading delta). None defaults to MovementDirection.FORWARD.

  • final_orientation_align (bool) – If True (default), rotate to match path[-1].orientation after reaching the last waypoint.

Return type:

None

batch_advance(dt)

Compute one timestep for all registered agents.

Implementations must:

  1. Fill self._pos_buf and self._orn_buf with new poses for all agents.

  2. Fill self._moved_mask with True for agents whose pose changed.

  3. Call self._apply_phase1() to write the moved rows back to sim.

Return type:

ndarray

Returns:

self._moved_mask — a (N,) boolean array.

class pybullet_fleet.controllers.BatchKinematicController

Bases: Controller

Vectorized base for kinematic controllers that drive many agents at once.

Inherits from Controller directly (not KinematicController): the per-agent compute(agent, dt) -> bool contract does not apply here since batch controllers act through batch_advance() during Phase 1 of sim_core.step_once(). Per-agent kinematic parameters (max_vel, accel, etc.) are read from each Agent at set_path time.

To register a custom batch controller, set _registry_name on the subclass and import the module once:

class MyBatchController(BatchKinematicController):
    _registry_name = "my_batch"

    def batch_advance(self, dt): ...

# Usage:
#   AgentManager(sim_core=sim, fleet_controller={"type": "my_batch"})
register_agent(agent)

Register agent with this batch controller.

On the first call the controller records agent.sim_core so that _apply_phase1() can call set_poses. The controller is not attached to sim_core._batch_controllers directly; step_once() discovers it through the owning AgentManager that is registered with the sim.

Return type:

int

Returns:

The agent’s row index in this controller’s state arrays.

Raises:

ValueError – If the agent is already registered, has no sim_core, or belongs to a different sim_core than previously-registered agents.

unregister_agent(agent)

Unregister agent. Compacts the state arrays by swapping with last row.

Raises:

KeyError – If the agent is not registered.

Return type:

None

reset()

Unregister all agents and reset arrays.

Return type:

None

synchronized_set_path(agent, path, **kwargs)

Set one path without racing a concurrent simulation step.

Return type:

None

synchronized_cancel_path(agent)

Cancel one path without racing a concurrent simulation step.

Return type:

None

synchronized_batch_advance(dt)

Advance vectorized trajectory state without concurrent path writes.

Return type:

ndarray

abstractmethod batch_advance(dt)

Compute one timestep for all registered agents.

Implementations must:

  1. Fill self._pos_buf and self._orn_buf with new poses for all agents.

  2. Fill self._moved_mask with True for agents whose pose changed.

  3. Call self._apply_phase1() to write the moved rows back to sim.

Return type:

ndarray

Returns:

self._moved_mask — a (N,) boolean array.

compute(agent, dt)

Compute one step of control.

Return type:

bool

Returns:

True if the agent moved, False otherwise.

set_velocity(**kwargs)
Return type:

None

class pybullet_fleet.controllers.BatchOmniController

Bases: BatchKinematicController

Batched omnidirectional pose controller.

set_path(agent, path, final_orientation_align=True, **kwargs)

Set a waypoint path.

Parameters:

  • agent (Agent) – A registered agent.

  • path (List[Pose]) – Non-empty list of goal poses.

  • final_orientation_align (bool) – If True (default), rotate to match path[-1].orientation after reaching the last waypoint.

Return type:

None

batch_advance(dt)

Compute one timestep for all registered agents.

Implementations must:

  1. Fill self._pos_buf and self._orn_buf with new poses for all agents.

  2. Fill self._moved_mask with True for agents whose pose changed.

  3. Call self._apply_phase1() to write the moved rows back to sim.

Return type:

ndarray

Returns:

self._moved_mask — a (N,) boolean array.

class pybullet_fleet.controllers.PatrolController(params=None, waypoints=None, wait_time=0.0, loop=True)

Bases: Controller

Cycle through waypoints. Delegates actual movement to the base controller by calling agent.set_goal_pose().

Parameters:

  • waypoints (list | None) – List of [x, y, z] positions.

  • wait_time (float) – Seconds to wait at each waypoint before advancing.

  • loop (bool) – If True, restart from first waypoint after reaching last.

compute(agent, dt)

Compute one step of control.

Return type:

bool

Returns:

True if the agent moved, False otherwise.

class pybullet_fleet.controllers.RandomWalkController(params=None, radius=5.0, wait_range=(1.0, 5.0))

Bases: Controller

Move to random nearby positions within a radius of the starting point.

Parameters:

  • radius (float) – Maximum distance from origin for random targets.

  • wait_range (tuple | list) – (min, max) seconds to wait at each target.

compute(agent, dt)

Compute one step of control.

Return type:

bool

Returns:

True if the agent moved, False otherwise.