pybullet_fleet.devices package

Submodules

pybullet_fleet.devices.door module

Door device — Agent subclass with URDF joint control.

class pybullet_fleet.devices.door.DoorParams(visual_shape=None, collision_shape=None, initial_pose=<factory>, mass=0.0, pickable=False, name=None, visual_frame_pose=None, collision_frame_pose=None, collision_mode=CollisionMode.NORMAL_3D, user_data=<factory>, urdf_path=None, motion_mode=MotionMode.DIFFERENTIAL, use_fixed_base=False, ik_params=None, controller=None, plugins=None, global_scaling=1.0, joint_tolerance=None, open_positions=None, close_positions=<factory>)

Bases: AgentSpawnParams

Spawn parameters for a Door device.

Extends AgentSpawnParams with door-specific joint targets.

open_positions: Joint name → target angle/position for the “open” state. Required — None raises TypeError in __post_init__.

close_positions: Joint name → target angle/position for the “closed” state.

open_positions: Dict[str, float] | None = None

close_positions: Dict[str, float]

classmethod from_dict(config)

Create DoorParams from a config dict.

Delegates base-field parsing to AgentSpawnParams.from_dict and adds door-specific fields (open_positions, close_positions).

Return type:: DoorParams

class pybullet_fleet.devices.door.Door(body_id, urdf_path=None, motion_mode=MotionMode.DIFFERENTIAL, use_fixed_base=False, collision_mode=CollisionMode.NORMAL_3D, sim_core=None, name=None, user_data=None, mass=None, ik_params=None, joint_tolerance=None, controller=None, notify_spawn=True, pickable=False)

Bases: Agent

Door controlled via URDF revolute or prismatic joint.

Uses JointAction for open/close transitions. Door state is derived from the current JointAction status and joint positions.

Config (YAML):

type: door
urdf_path: robots/door_hinge.urdf
open_positions: {hinge: 1.57}
close_positions: {hinge: 0.0}

classmethod from_params(spawn_params, sim_core=None)

Create a Door from DoorParams.

Raises:: TypeError – If spawn_params is not a DoorParams instance or open_positions is missing.
Return type:: Door

property door_state: DoorState: Derived door state.

request_open()

Open the door (enqueue JointAction to open positions).

Return type:: None

request_close()

Close the door (enqueue JointAction to close positions).

Return type:: None

pybullet_fleet.devices.elevator module

Elevator device — Agent subclass with prismatic Z joint and auto-attach.

Objects (agents, sim_objects) on the platform are automatically attached before movement and detached on arrival, so they physically ride the elevator without any external teleport logic.

class pybullet_fleet.devices.elevator.ElevatorParams(visual_shape=None, collision_shape=None, initial_pose=<factory>, mass=0.0, pickable=False, name=None, visual_frame_pose=None, collision_frame_pose=None, collision_mode=CollisionMode.NORMAL_3D, user_data=<factory>, urdf_path=None, motion_mode=MotionMode.DIFFERENTIAL, use_fixed_base=False, ik_params=None, controller=None, plugins=None, global_scaling=1.0, joint_tolerance=None, floors=None, initial_floor='', joint_name='lift', platform_link='platform', request_policy=ElevatorRequestPolicy.REJECT, max_speed=None, max_accel=None, max_decel=None)

Bases: AgentSpawnParams

Spawn parameters for an Elevator device.

Extends AgentSpawnParams with elevator-specific floor/joint config.

floors: Floor name → joint position mapping. Required — None raises TypeError in __post_init__.

initial_floor: Name of the floor the elevator starts at.

joint_name: Name of the prismatic joint that moves the platform.

platform_link: Name of the link with the platform collision box.

request_policy: Handling for floor requests received while moving.

max_speed: Cabin speed in m/s. None uses the URDF’s <limit velocity="..."> on joint_name, which is where the speed otherwise has to be set – so without this, two installations that differ only in how fast the cabin travels need two URDFs.

max_accel: Cabin acceleration in m/s^2. None keeps the cabin at full speed from the first step, as before. Giving it makes the travel a trapezoid, which is what a real cabin does: 0.6 m at 0.4 m/s takes 1.5 s flat and 1.9 s with 1.0 m/s^2 ramps.

max_decel: Cabin deceleration. Defaults to max_accel.

floors: Dict[str, float] | None = None

initial_floor: str = ''

joint_name: str = 'lift'

platform_link: str = 'platform'

request_policy: ElevatorRequestPolicy | str = 'reject'

max_speed: float | None = None

max_accel: float | None = None

max_decel: float | None = None

classmethod from_dict(config)

Create ElevatorParams from a config dict.

Delegates base-field parsing to AgentSpawnParams.from_dict and adds elevator-specific fields (floors, initial_floor, joint_name, platform_link).

Return type:: ElevatorParams

class pybullet_fleet.devices.elevator.Elevator(body_id, urdf_path=None, motion_mode=MotionMode.DIFFERENTIAL, use_fixed_base=False, collision_mode=CollisionMode.NORMAL_3D, sim_core=None, name=None, user_data=None, mass=None, ik_params=None, joint_tolerance=None, controller=None, notify_spawn=True, pickable=False)

Bases: Agent

Elevator with named floors and automatic passenger attachment.

Uses JointAction for floor transitions. Before movement starts, all sim_objects whose AABB overlaps the platform collision box are attached to the platform link. On arrival they are detached.

Config (YAML):

type: elevator
urdf_path: robots/elevator.urdf
use_fixed_base: true
floors: {L1: 0.0, L2: 8.0, L3: 16.0}
initial_floor: L1
joint_name: lift
platform_link: platform
request_policy: queue  # reject (default), replace_next, or queue

classmethod from_params(spawn_params, sim_core=None)

Create an Elevator from ElevatorParams.

Raises:: TypeError – If spawn_params is not an ElevatorParams instance or floors is missing.
Return type:: Elevator

property current_floor: str

Floor where the elevator currently is (or departed from while moving).

While the elevator is in transit this returns the departure floor, not the destination. Use target_floor for the destination.

property target_floor: str: Destination floor (equals current_floor when idle).

property available_floors: List[str]: List of available floor names.

property is_moving: bool: True while elevator is transitioning between floors.

property state: ElevatorState: Current backend-neutral elevator lifecycle state.

property request_policy: ElevatorRequestPolicy: Policy applied to requests received while the cabin is moving.

property pending_floors: tuple[str, ...]: Requested floors to visit after the current transition.

property passengers: List[SimObject]: Currently attached passengers (read-only copy).

request_floor(floor_name)

Request the elevator to move to the given floor.

Attaches all objects on the platform, then starts the joint action.

Parameters:: floor_name (str) – Name of the target floor (must exist in floors).
Return type:: ElevatorRequestResult
Returns:: Whether the request was accepted immediately, queued, replaced, or rejected according to request_policy.

update(dt)

Per-step update. Detaches passengers when movement completes.

Return type:: None

begin_motion(target_height)

Command the PyBullet lift joint to move to target_height.

Return type:: None

motion_in_progress()

Observe whether the cabin is still moving.

The action alone is not enough once the lift joint has a motion profile. JointAction completes as soon as the joint is inside its tolerance, but a ramped joint is still braking through the last millimetres after that – measured on a 0.6 m ride at 0.4 m/s with 1 m/s^2 ramps, the action finished at 1.78 s with the cabin 7.2 mm short, and the joint settled at 1.88 s. Taking the action’s word for it let the state machine mark the cabin arrived and release its passengers while it was still moving.

Return type:: bool

attach_platform_passengers()

Attach platform occupants using PyBullet constraints.

Return type:: int

detach_passengers()

Release constrained passengers after a completed lift motion.

Return type:: int

Module contents

Infrastructure devices: doors, elevators.

class pybullet_fleet.devices.Door(body_id, urdf_path=None, motion_mode=MotionMode.DIFFERENTIAL, use_fixed_base=False, collision_mode=CollisionMode.NORMAL_3D, sim_core=None, name=None, user_data=None, mass=None, ik_params=None, joint_tolerance=None, controller=None, notify_spawn=True, pickable=False)

Bases: Agent

Door controlled via URDF revolute or prismatic joint.

Uses JointAction for open/close transitions. Door state is derived from the current JointAction status and joint positions.

Config (YAML):

type: door
urdf_path: robots/door_hinge.urdf
open_positions: {hinge: 1.57}
close_positions: {hinge: 0.0}

classmethod from_params(spawn_params, sim_core=None)

Create a Door from DoorParams.

Raises:: TypeError – If spawn_params is not a DoorParams instance or open_positions is missing.
Return type:: Door

property door_state: DoorState: Derived door state.

request_open()

Open the door (enqueue JointAction to open positions).

Return type:: None

request_close()

Close the door (enqueue JointAction to close positions).

Return type:: None

class pybullet_fleet.devices.DoorParams(visual_shape=None, collision_shape=None, initial_pose=<factory>, mass=0.0, pickable=False, name=None, visual_frame_pose=None, collision_frame_pose=None, collision_mode=CollisionMode.NORMAL_3D, user_data=<factory>, urdf_path=None, motion_mode=MotionMode.DIFFERENTIAL, use_fixed_base=False, ik_params=None, controller=None, plugins=None, global_scaling=1.0, joint_tolerance=None, open_positions=None, close_positions=<factory>)

Bases: AgentSpawnParams

Spawn parameters for a Door device.

Extends AgentSpawnParams with door-specific joint targets.

open_positions: Joint name → target angle/position for the “open” state. Required — None raises TypeError in __post_init__.

close_positions: Joint name → target angle/position for the “closed” state.

open_positions: Dict[str, float] | None = None

close_positions: Dict[str, float]

classmethod from_dict(config)

Create DoorParams from a config dict.

Delegates base-field parsing to AgentSpawnParams.from_dict and adds door-specific fields (open_positions, close_positions).

Return type:: DoorParams

class pybullet_fleet.devices.Elevator(body_id, urdf_path=None, motion_mode=MotionMode.DIFFERENTIAL, use_fixed_base=False, collision_mode=CollisionMode.NORMAL_3D, sim_core=None, name=None, user_data=None, mass=None, ik_params=None, joint_tolerance=None, controller=None, notify_spawn=True, pickable=False)

Bases: Agent

Elevator with named floors and automatic passenger attachment.

Uses JointAction for floor transitions. Before movement starts, all sim_objects whose AABB overlaps the platform collision box are attached to the platform link. On arrival they are detached.

Config (YAML):

type: elevator
urdf_path: robots/elevator.urdf
use_fixed_base: true
floors: {L1: 0.0, L2: 8.0, L3: 16.0}
initial_floor: L1
joint_name: lift
platform_link: platform
request_policy: queue  # reject (default), replace_next, or queue

classmethod from_params(spawn_params, sim_core=None)

Create an Elevator from ElevatorParams.

Raises:: TypeError – If spawn_params is not an ElevatorParams instance or floors is missing.
Return type:: Elevator

property current_floor: str

Floor where the elevator currently is (or departed from while moving).

While the elevator is in transit this returns the departure floor, not the destination. Use target_floor for the destination.

property target_floor: str: Destination floor (equals current_floor when idle).

property available_floors: List[str]: List of available floor names.

property is_moving: bool: True while elevator is transitioning between floors.

property state: ElevatorState: Current backend-neutral elevator lifecycle state.

property request_policy: ElevatorRequestPolicy: Policy applied to requests received while the cabin is moving.

property pending_floors: tuple[str, ...]: Requested floors to visit after the current transition.

property passengers: List[SimObject]: Currently attached passengers (read-only copy).

request_floor(floor_name)

Request the elevator to move to the given floor.

Attaches all objects on the platform, then starts the joint action.

Parameters:: floor_name (str) – Name of the target floor (must exist in floors).
Return type:: ElevatorRequestResult
Returns:: Whether the request was accepted immediately, queued, replaced, or rejected according to request_policy.

update(dt)

Per-step update. Detaches passengers when movement completes.

Return type:: None

begin_motion(target_height)

Command the PyBullet lift joint to move to target_height.

Return type:: None

motion_in_progress()

Observe whether the cabin is still moving.

The action alone is not enough once the lift joint has a motion profile. JointAction completes as soon as the joint is inside its tolerance, but a ramped joint is still braking through the last millimetres after that – measured on a 0.6 m ride at 0.4 m/s with 1 m/s^2 ramps, the action finished at 1.78 s with the cabin 7.2 mm short, and the joint settled at 1.88 s. Taking the action’s word for it let the state machine mark the cabin arrived and release its passengers while it was still moving.

Return type:: bool

attach_platform_passengers()

Attach platform occupants using PyBullet constraints.

Return type:: int

detach_passengers()

Release constrained passengers after a completed lift motion.

Return type:: int

class pybullet_fleet.devices.ElevatorParams(visual_shape=None, collision_shape=None, initial_pose=<factory>, mass=0.0, pickable=False, name=None, visual_frame_pose=None, collision_frame_pose=None, collision_mode=CollisionMode.NORMAL_3D, user_data=<factory>, urdf_path=None, motion_mode=MotionMode.DIFFERENTIAL, use_fixed_base=False, ik_params=None, controller=None, plugins=None, global_scaling=1.0, joint_tolerance=None, floors=None, initial_floor='', joint_name='lift', platform_link='platform', request_policy=ElevatorRequestPolicy.REJECT, max_speed=None, max_accel=None, max_decel=None)

Bases: AgentSpawnParams

Spawn parameters for an Elevator device.

Extends AgentSpawnParams with elevator-specific floor/joint config.

floors: Floor name → joint position mapping. Required — None raises TypeError in __post_init__.

initial_floor: Name of the floor the elevator starts at.

joint_name: Name of the prismatic joint that moves the platform.

platform_link: Name of the link with the platform collision box.

request_policy: Handling for floor requests received while moving.

max_speed: Cabin speed in m/s. None uses the URDF’s <limit velocity="..."> on joint_name, which is where the speed otherwise has to be set – so without this, two installations that differ only in how fast the cabin travels need two URDFs.

max_accel: Cabin acceleration in m/s^2. None keeps the cabin at full speed from the first step, as before. Giving it makes the travel a trapezoid, which is what a real cabin does: 0.6 m at 0.4 m/s takes 1.5 s flat and 1.9 s with 1.0 m/s^2 ramps.

max_decel: Cabin deceleration. Defaults to max_accel.

floors: Dict[str, float] | None = None

initial_floor: str = ''

joint_name: str = 'lift'

platform_link: str = 'platform'

request_policy: ElevatorRequestPolicy | str = 'reject'

max_speed: float | None = None

max_accel: float | None = None

max_decel: float | None = None

classmethod from_dict(config)

Create ElevatorParams from a config dict.

Delegates base-field parsing to AgentSpawnParams.from_dict and adds elevator-specific fields (floors, initial_floor, joint_name, platform_link).

Return type:: ElevatorParams

class pybullet_fleet.devices.ElevatorRequestPolicy(value)

Bases: str, Enum

How requests received while the cabin is moving are handled.

REJECT = 'reject'

REPLACE_NEXT = 'replace_next'

QUEUE = 'queue'

class pybullet_fleet.devices.ElevatorRequestResult(value)

Bases: str, Enum

Outcome of an ElevatorStateMachine.request_floor() call.

ACCEPTED = 'accepted'

QUEUED = 'queued'

REPLACED = 'replaced'

REJECTED = 'rejected'

class pybullet_fleet.devices.ElevatorState(value)

Bases: str, Enum

Lifecycle states exposed by an elevator device.

IDLE = 'idle'

MOVING = 'moving'

ARRIVED = 'arrived'

class pybullet_fleet.devices.ElevatorStateMachine(floors, initial_floor, adapter, request_policy=ElevatorRequestPolicy.REJECT)

Bases: object

Coordinate floor requests independently of a physics backend.

ARRIVED is observable after a completed transition until the next update or a newly accepted floor request.

property is_moving: bool: Whether the cabin is currently in a physical movement transition.

property pending_floors: tuple[str, ...]: Floor requests waiting for the current transition to complete.

request_floor(floor_name)

Accept, queue, replace, or reject a requested destination.

The physical cabin is never redirected mid-motion. REPLACE_NEXT replaces the pending destination after the current trip; QUEUE preserves requests in arrival order.

Return type:: ElevatorRequestResult

update()

Observe the adapter and complete an arrival when movement has ended.

Returns True only on the update that observes arrival.

Return type:: bool