pybullet_fleet.plugins package

Submodules

pybullet_fleet.plugins.battery_plugin module

BatteryPlugin — per-agent battery simulation.

Drains/charges battery SOC each step based on the plugin’s rate attributes and the agent’s current motion/charging state.

The plugin owns all battery state (rates, SOC, charging flag). Agent exposes delegate properties for convenience:

agent.battery_soc       # → plugin.soc
agent.battery_plugin    # → the BatteryPlugin instance (or None)
agent.is_charging       # → plugin.is_charging
agent.set_charging(b)   # → plugin.set_charging(b)

Created automatically by Agent.from_params() when listed in AgentSpawnParams.plugins, or attached manually.

Example (via plugins list in YAML):

plugins:
  - type: battery
    config:
      initial_soc: 0.8
      discharge_rate: 0.002

Example (manual):

from pybullet_fleet.plugins.battery_plugin import BatteryPlugin
agent.add_plugin(BatteryPlugin(agent, discharge_rate=0.002))

Example (custom subclass):

class TemperatureBattery(BatteryPlugin):
    def on_update(self, dt: float) -> None:
        temp_factor = 1.0 + 0.02 * (self.agent.user_data.get("temp", 25) - 25)
        if self.is_charging:
            self.soc = min(1.0, self.soc + self.charge_rate * dt)
        elif self.agent.is_moving:
            self.soc = max(0.0, self.soc - self.discharge_rate * dt * temp_factor)

agent.remove_plugin(BatteryPlugin)
agent.add_plugin(TemperatureBattery(agent, initial_soc=0.9, discharge_rate=0.005))
class pybullet_fleet.plugins.battery_plugin.BatteryPlugin(agent, initial_soc=1.0, discharge_rate=0.001, charge_rate=0.005, idle_rate=0.0)

Bases: AgentPlugin

Default linear battery drain/charge plugin.

Owns all battery state as plain attributes:

  • discharge_rate — SOC drain per second while moving

  • charge_rate — SOC gain per second while charging

  • idle_rate — SOC drain per second while idle

  • soc — current state of charge [0.0, 1.0]

  • is_charging — whether the agent is charging

Update logic:

  • charging → soc += charge_rate * dt

  • moving → soc -= discharge_rate * dt

  • idle → soc -= idle_rate * dt

SOC is clamped to [0.0, 1.0].

set_charging(charging)

Start or stop charging.

Parameters:

charging (bool) – True to start, False to stop.

Return type:

None

on_update(dt)

Update battery SOC based on agent’s current state.

Return type:

None

pybullet_fleet.plugins.workcell_plugin module

WorkcellPlugin — ROS-free dispenser/ingestor simulation logic.

Extracts the pure simulation logic from WorkcellHandler so that dispense/ingest cycles can run without ROS. The ROS bridge (WorkcellHandler) delegates to this plugin and handles only messaging (subscriptions, publishers, result/state messages).

Dispenser flow:

  1. dispense(workcell_name, robot) checks for nearby pickable SimObjects (like Gazebo TeleportDispenser’s fill_dispenser). If none found, a fallback cargo box is spawned.

  2. A PickAction(target_position=...) is queued on the robot. The action itself resolves the nearest pickable object.

  3. on_step() monitors pending actions, records item initial positions on completion, and emits completion events.

Ingestor flow:

  1. ingest(workcell_name, robot) queues a DropAction on the carrier robot.

  2. On completion the item is scheduled for return-home (teleport back to its dispenser position after a configurable delay).

Config keys (passed via config dict):

item_shape:                              # ShapeParams-compatible dict (default: box)
  shape_type: box
  half_extents: [0.15, 0.15, 0.1]
  rgba_color: [0.8, 0.5, 0.2, 1.0]
item_mass: 0.5
item_search_radius: 1.0
attach_offset: [0.0, 0.0, 0.15, 0.0, 0.0, 0.0]  # [dx, dy, dz, roll, pitch, yaw]
spawn_fallback: true
spawn_offset: [0.0, 0.0, 0.3, 0.0, 0.0, 0.0]  # [dx, dy, dz, roll, pitch, yaw]
return_home: true
return_home_delay: 5.0
overrides:              # per-workcell config (optional)
  dispenser_1:
    position: [8.5, -0.5]       # z defaults to 0
    item_search_radius: 2.0
    spawn_fallback: false
  dispenser_2:
    position: [1.0, 2.0, 5.0]   # multi-floor (z=5)
    item_shape:                  # ShapeParams dict (shape_type inferred if omitted)
      mesh_path: "mesh/coke_can.obj"
      mesh_scale: [0.01, 0.01, 0.01]
      rgba_color: [1.0, 0.0, 0.0, 1.0]
    item_mass: 0.3

Usage (standalone, no ROS):

plugin = sim.register_plugin(WorkcellPlugin, config={
    "item_search_radius": 1.0,
})
# After sim init:
item, pick = plugin.dispense("dispenser_1", robot)
# ... sim steps until pick completes ...
drop = plugin.ingest("ingestor_1", robot)
class pybullet_fleet.plugins.workcell_plugin.WorkcellKind(value)

Bases: Enum

Type of workcell operation.

DISPENSER = 'dispenser'
INGESTOR = 'ingestor'
class pybullet_fleet.plugins.workcell_plugin.PendingAction(action, cargo, kind, robot, workcell_name, on_complete=None)

Bases: object

A pick/drop action waiting for completion.

action: Any
cargo: SimObject | None
kind: WorkcellKind
robot: Agent
workcell_name: str
on_complete: Callable | None = None
class pybullet_fleet.plugins.workcell_plugin.WorkcellConfig(position=None, search_radius=1.0, spawn_fallback=True, spawn_offset=<factory>, item_visual=None, item_mass=0.5)

Bases: object

Resolved per-workcell configuration.

position: Tuple[float, float, float] | None = None
search_radius: float = 1.0
spawn_fallback: bool = True
spawn_offset: List[float]
item_visual: ShapeParams | None = None
item_mass: float = 0.5
class pybullet_fleet.plugins.workcell_plugin.WorkcellPlugin(sim_core, item_shape=None, item_mass=0.5, item_search_radius=1.0, attach_offset=None, spawn_fallback=True, spawn_offset=None, return_home_delay=5.0, return_home=True, overrides=None)

Bases: SimPlugin

ROS-free workcell (dispenser/ingestor) simulation logic.

Provides public methods for dispense/ingest operations and monitors pending actions in on_step(). Completion is reported via an optional on_complete callback passed to each operation.

dispense(workcell_name, robot, key=None, on_complete=None)

Queue a PickAction on robot to pick the nearest item at workcell_name.

Uses PickAction(target_position=...) so the action itself resolves the nearest pickable object within item_search_radius. If no pickable item exists near the workcell, a fallback cargo box is spawned first.

Parameters:

  • workcell_name (str) – Name of the dispenser workcell.

  • robot (Agent) – The Agent to pick the item.

  • key (Optional[str]) – Unique key for tracking (auto-generated if None).

  • on_complete (Optional[Callable]) – Callback (success: bool, pending: PendingAction) called when the PickAction completes or fails.

Return type:

Tuple[Optional[SimObject], Any]

Returns:

(None, pick_action) on success (item resolved lazily by PickAction), (None, None) if workcell position unknown.

ingest(workcell_name, robot, key=None, on_complete=None)

Queue a DropAction on robot at the ingestor position.

Parameters:

  • workcell_name (str) – Name of the ingestor workcell.

  • robot (Agent) – The Agent carrying cargo.

  • key (Optional[str]) – Unique key for tracking (auto-generated if None).

  • on_complete (Optional[Callable]) – Callback (success: bool, pending: PendingAction) called when the DropAction completes or fails.

Return type:

Optional[Any]

Returns:

DropAction on success, None on failure.

find_nearest_robot(workcell_name, candidates)

Return the candidate Agent nearest to workcell_name.

Parameters:

  • workcell_name (str) – Workcell to measure distance from.

  • candidates (List[Agent]) – List of Agent instances to search.

Return type:

Optional[Agent]

Returns:

Nearest Agent, or None if candidates is empty.

find_nearest_carrier(workcell_name, candidates)

Return the nearest candidate carrying cargo (attached objects).

Falls back to the nearest candidate if no carrier is found.

Parameters:

  • workcell_name (str) – Workcell to measure distance from.

  • candidates (List[Agent]) – List of Agent instances to search.

Return type:

Optional[Agent]

Returns:

Nearest carrier Agent, or nearest candidate as fallback.

get_workcell_position(name)

Look up a workcell’s XYZ position.

Positions are defined in config overrides:

overrides:
  dispenser_1:
    position: [8.5, -0.5]       # z defaults to 0
  dispenser_2:
    position: [1.0, 2.0, 5.0]   # multi-floor
Return type:

Optional[Tuple[float, float, float]]

Returns:

(x, y, z) tuple, or None if not found.

property pending_actions: Dict[str, PendingAction]

Read-only view of pending actions (key → PendingAction).

on_step(dt)

Check pending actions and process return-home queue.

Return type:

None

on_reset()

Clear all pending actions and return-home queue.

Return type:

None

Module contents

Reusable simulation plugins.

class pybullet_fleet.plugins.BatteryPlugin(agent, initial_soc=1.0, discharge_rate=0.001, charge_rate=0.005, idle_rate=0.0)

Bases: AgentPlugin

Default linear battery drain/charge plugin.

Owns all battery state as plain attributes:

  • discharge_rate — SOC drain per second while moving

  • charge_rate — SOC gain per second while charging

  • idle_rate — SOC drain per second while idle

  • soc — current state of charge [0.0, 1.0]

  • is_charging — whether the agent is charging

Update logic:

  • charging → soc += charge_rate * dt

  • moving → soc -= discharge_rate * dt

  • idle → soc -= idle_rate * dt

SOC is clamped to [0.0, 1.0].

set_charging(charging)

Start or stop charging.

Parameters:

charging (bool) – True to start, False to stop.

Return type:

None

on_update(dt)

Update battery SOC based on agent’s current state.

Return type:

None

class pybullet_fleet.plugins.WorkcellPlugin(sim_core, item_shape=None, item_mass=0.5, item_search_radius=1.0, attach_offset=None, spawn_fallback=True, spawn_offset=None, return_home_delay=5.0, return_home=True, overrides=None)

Bases: SimPlugin

ROS-free workcell (dispenser/ingestor) simulation logic.

Provides public methods for dispense/ingest operations and monitors pending actions in on_step(). Completion is reported via an optional on_complete callback passed to each operation.

dispense(workcell_name, robot, key=None, on_complete=None)

Queue a PickAction on robot to pick the nearest item at workcell_name.

Uses PickAction(target_position=...) so the action itself resolves the nearest pickable object within item_search_radius. If no pickable item exists near the workcell, a fallback cargo box is spawned first.

Parameters:

  • workcell_name (str) – Name of the dispenser workcell.

  • robot (Agent) – The Agent to pick the item.

  • key (Optional[str]) – Unique key for tracking (auto-generated if None).

  • on_complete (Optional[Callable]) – Callback (success: bool, pending: PendingAction) called when the PickAction completes or fails.

Return type:

Tuple[Optional[SimObject], Any]

Returns:

(None, pick_action) on success (item resolved lazily by PickAction), (None, None) if workcell position unknown.

ingest(workcell_name, robot, key=None, on_complete=None)

Queue a DropAction on robot at the ingestor position.

Parameters:

  • workcell_name (str) – Name of the ingestor workcell.

  • robot (Agent) – The Agent carrying cargo.

  • key (Optional[str]) – Unique key for tracking (auto-generated if None).

  • on_complete (Optional[Callable]) – Callback (success: bool, pending: PendingAction) called when the DropAction completes or fails.

Return type:

Optional[Any]

Returns:

DropAction on success, None on failure.

find_nearest_robot(workcell_name, candidates)

Return the candidate Agent nearest to workcell_name.

Parameters:

  • workcell_name (str) – Workcell to measure distance from.

  • candidates (List[Agent]) – List of Agent instances to search.

Return type:

Optional[Agent]

Returns:

Nearest Agent, or None if candidates is empty.

find_nearest_carrier(workcell_name, candidates)

Return the nearest candidate carrying cargo (attached objects).

Falls back to the nearest candidate if no carrier is found.

Parameters:

  • workcell_name (str) – Workcell to measure distance from.

  • candidates (List[Agent]) – List of Agent instances to search.

Return type:

Optional[Agent]

Returns:

Nearest carrier Agent, or nearest candidate as fallback.

get_workcell_position(name)

Look up a workcell’s XYZ position.

Positions are defined in config overrides:

overrides:
  dispenser_1:
    position: [8.5, -0.5]       # z defaults to 0
  dispenser_2:
    position: [1.0, 2.0, 5.0]   # multi-floor
Return type:

Optional[Tuple[float, float, float]]

Returns:

(x, y, z) tuple, or None if not found.

property pending_actions: Dict[str, PendingAction]

Read-only view of pending actions (key → PendingAction).

on_step(dt)

Check pending actions and process return-home queue.

Return type:

None

on_reset()

Clear all pending actions and return-home queue.

Return type:

None