Python SDK /fastsim.extensions

fastsim.extensions

46 classes

Optional capability modules (recording, benchmark, ROS, HTTP server, data collection, …) that plug into the simulation lifecycle. Enabled via config and wired through ExtensionManager. Also contains the action/goal/observer/backend subsystems used by these modules.

ExtensionManager Extension DataCollector Recorder Replayer Benchmark Server Policy Action GraspAction MoveAction ReleaseAction AcquireAction PickAction PlaceAction PushAction PullAction InsertAction ClickAction TwistAction ActionFactory Goal PoseGoal OnTopGoal InsideGoal GoalFactory Observer RobotObserver SensorObserver ObjectObserver TaskObserver ObserverFactory Observation Postprocessor ExportHdf5 ExportVideo ExportPngDepth ExportPreviewVideo PostprocessorFactory Backend LocalFileSystemBackend Ros2Backend BackendFactory TrajectoryTrack Trajectory TrajectoryLoader
class

ExtensionManager

static-only2 methods
fastsim.extensions.ExtensionManager

Static manager that enables, wires, and provides access to all configured extensions.

enable_all_extensionsstaticmethod
@staticmethod def enable_all_extensions(sim: FastSim) → None

Activate all extensions listed in the extension config.

Parameters
NameTypeDefaultDescription
simFastSim—The active FastSim instance to hook extensions into.
Returns
None
Called automatically by setup_all_before_loop().
get_enabled_extensionstaticmethod
@staticmethod def get_enabled_extension(name: str) → Extension

Retrieve a running extension instance by name.

Parameters
NameTypeDefaultDescription
namestr—Extension name as registered (e.g. 'benchmark', 'server').
Returns
Extension — The active Extension instance.
class

Extension

abstract1 method
fastsim.extensions.extension.ExtensionextendsABC

Abstract base class for all FastSim extensions.

Implement enable(sim) to hook into the FastSim lifecycle (register step callbacks, set up resources, etc.). Extensions are instantiated and enabled by ExtensionManager.

enablemethod
def enable(self, sim: FastSim) → None

Activate the extension against the provided FastSim instance.

Parameters
NameTypeDefaultDescription
simFastSim—The running FastSim instance.
Returns
None
class

DataCollector

3 methods
fastsim.extensions.data_collector.DataCollectorextendsExtension

Extension that collects observations from registered observers each tick.

enablemethod
def enable(self, sim: FastSim) → None

Hook observers into the step loop.

Parameters
NameTypeDefaultDescription
simFastSim—The running FastSim instance.
Returns
None
register_observermethod
def register_observer(self, observer: Observer) → None

Add an Observer instance to the collection pipeline.

Parameters
NameTypeDefaultDescription
observerObserver—Observer to register.
Returns
None
register_from_configmethod
def register_from_config(self) → None

Instantiate and register all observers defined in the extension config.

Returns
None
Inherited fromExtension1
enablemethod
def enable(self, sim: FastSim) → None

Activate the extension against the provided FastSim instance.

class

Recorder

1 method
fastsim.extensions.recorder.RecorderextendsExtension

Extension that records simulation data to a configured backend.

enablemethod
def enable(self, sim: FastSim) → None

Initialize the backend and start recording on each tick.

Parameters
NameTypeDefaultDescription
simFastSim—The running FastSim instance.
Returns
None
Inherited fromExtension1
enablemethod
def enable(self, sim: FastSim) → None

Activate the extension against the provided FastSim instance.

class

Replayer

1 method
fastsim.extensions.replayer.ReplayerextendsExtension

Extension that replays previously recorded trajectory data into the simulation.

enablemethod
def enable(self, sim: FastSim) → None

Load trajectory data and begin playback each tick.

Parameters
NameTypeDefaultDescription
simFastSim—The running FastSim instance.
Returns
None
Inherited fromExtension1
enablemethod
def enable(self, sim: FastSim) → None

Activate the extension against the provided FastSim instance.

class

Benchmark

1 method
fastsim.extensions.benchmark.BenchmarkextendsExtension

Extension that runs policy benchmark episodes and collects success metrics.

enablemethod
def enable(self, sim: FastSim) → None

Wire the benchmark into the step loop and start episode management.

Parameters
NameTypeDefaultDescription
simFastSim—The running FastSim instance.
Returns
None
Inherited fromExtension1
enablemethod
def enable(self, sim: FastSim) → None

Activate the extension against the provided FastSim instance.

class

Server

1 method
fastsim.extensions.server.ServerextendsExtension

Extension that exposes a REST HTTP API for remote control of the simulation.

enablemethod
def enable(self, sim: FastSim) → None

Start the HTTP server thread and register API routes.

Parameters
NameTypeDefaultDescription
simFastSim—The running FastSim instance.
Returns
None
Inherited fromExtension1
enablemethod
def enable(self, sim: FastSim) → None

Activate the extension against the provided FastSim instance.

class

Policy

abstract6 methods
fastsim.extensions.benchmark.policy.PolicyextendsABC

Abstract base class for benchmark policies. Implement all abstract methods to create a custom policy.

A Policy defines the observe → preprocess → compute → postprocess pipeline that runs each benchmark episode. Override needs_observation() to control when observations are requested.

resetmethod
def reset(self) → None

Reset internal policy state at the start of each episode.

Returns
None
warmupmethod
def warmup(self, benchmark_observation: Any) → None

Perform any model warm-up using an initial observation.

Parameters
NameTypeDefaultDescription
benchmark_observationAny—Raw observation from the benchmark environment.
Returns
None
needs_observationmethod
def needs_observation(self) → bool

Return True if the policy requires a fresh observation this step.

Returns
bool — True to request a new observation, False to reuse the last.
preprocess_observationmethod
def preprocess_observation(self, benchmark_observation: Any) → dict

Convert raw benchmark observation to a model-ready dict.

Parameters
NameTypeDefaultDescription
benchmark_observationAny—Raw observation data.
Returns
dict — Preprocessed observation dict.
compute_actionmethod
def compute_action(self, observation: dict) → dict

Run inference and return a raw action dict.

Parameters
NameTypeDefaultDescription
observationdict—Preprocessed observation dict.
Returns
dict — Raw action dict from the model.
postprocess_actionmethod
def postprocess_action(self, action: dict) → BenchmarkAction

Convert raw model output to a BenchmarkAction ready for execution.

Parameters
NameTypeDefaultDescription
actiondict—Raw action dict from compute_action.
Returns
BenchmarkAction — The fully constructed action object.
class

Action

abstract2 methods
fastsim.extensions.action.ActionextendsABC

Abstract action executed by the task/benchmark system.

actmethod
def act(self) → ActionResultStatus

Execute the action and return a result status.

Returns
ActionResultStatus — Result of action execution.
is_goal_satisfiedmethod
def is_goal_satisfied(self) → bool

Return True if the post-condition goal of this action is met.

Returns
bool
class

GraspAction

0 methods
fastsim.extensions.action.GraspActionextendsAction

Close the gripper to grasp a target object.

Inherited fromAction2
actmethod
def act(self) → ActionResultStatus

Execute the action and return a result status.

is_goal_satisfiedmethod
def is_goal_satisfied(self) → bool

Return True if the post-condition goal of this action is met.

class

MoveAction

0 methods
fastsim.extensions.action.MoveActionextendsAction

Move the end-effector to a target pose.

Inherited fromAction2
actmethod
def act(self) → ActionResultStatus

Execute the action and return a result status.

is_goal_satisfiedmethod
def is_goal_satisfied(self) → bool

Return True if the post-condition goal of this action is met.

class

ReleaseAction

0 methods
fastsim.extensions.action.ReleaseActionextendsAction

Open the gripper to release a currently held object.

Inherited fromAction2
actmethod
def act(self) → ActionResultStatus

Execute the action and return a result status.

is_goal_satisfiedmethod
def is_goal_satisfied(self) → bool

Return True if the post-condition goal of this action is met.

class

AcquireAction

0 methods
fastsim.extensions.action.AcquireActionextendsGraspAction,MoveAction

Combined move-to + grasp action for acquiring an object.

This class is used via inheritance; no additional methods documented.See methods:GraspAction/MoveAction
class

PickAction

0 methods
fastsim.extensions.action.PickActionextendsAcquireAction

Pick up an object from its current location.

This class is used via inheritance; no additional methods documented.See methods:AcquireAction
class

PlaceAction

0 methods
fastsim.extensions.action.PlaceActionextendsReleaseAction,MoveAction

Move to a target pose and release the held object.

This class is used via inheritance; no additional methods documented.See methods:ReleaseAction/MoveAction
class

PushAction

0 methods
fastsim.extensions.action.PushActionextendsGraspAction,MoveAction

Push an object by contacting and driving it to a new location.

This class is used via inheritance; no additional methods documented.See methods:GraspAction/MoveAction
class

PullAction

0 methods
fastsim.extensions.action.PullActionextendsAcquireAction

Pull an object toward the robot.

This class is used via inheritance; no additional methods documented.See methods:AcquireAction
class

InsertAction

0 methods
fastsim.extensions.action.InsertActionextendsPlaceAction

Insert an object into a slot or receptacle.

This class is used via inheritance; no additional methods documented.See methods:PlaceAction
class

ClickAction

0 methods
fastsim.extensions.action.ClickActionextendsPushAction

Click or press a button/switch by making contact and applying force.

This class is used via inheritance; no additional methods documented.See methods:PushAction
class

TwistAction

0 methods
fastsim.extensions.action.TwistActionextendsAcquireAction

Grasp and rotate an object (e.g. turn a knob or open a jar).

This class is used via inheritance; no additional methods documented.See methods:AcquireAction
class

ActionFactory

static-only1 method
fastsim.extensions.action.ActionFactory

Static factory for creating Action instances from a config dict.

createstaticmethod
@staticmethod def create(action_type: str, config: dict) → Action

Instantiate an Action by type name and config parameters.

Parameters
NameTypeDefaultDescription
action_typestr—Registered action type name (e.g. 'pick', 'place').
configdict—Action configuration parameters.
Returns
Action — Constructed Action instance.
class

Goal

abstract1 method
fastsim.extensions.goal.GoalextendsABC

Abstract goal condition evaluated after an action to determine task success.

is_satisfiedmethod
def is_satisfied(self) → bool

Evaluate the goal condition and return True if it is met.

Returns
bool
class

PoseGoal

0 methods
fastsim.extensions.goal.PoseGoalextendsGoal

Goal satisfied when an entity reaches a target pose within a tolerance.

Inherited fromGoal1
is_satisfiedmethod
def is_satisfied(self) → bool

Evaluate the goal condition and return True if it is met.

class

OnTopGoal

0 methods
fastsim.extensions.goal.OnTopGoalextendsGoal

Goal satisfied when a target object is resting on top of a reference object.

Inherited fromGoal1
is_satisfiedmethod
def is_satisfied(self) → bool

Evaluate the goal condition and return True if it is met.

class

InsideGoal

0 methods
fastsim.extensions.goal.InsideGoalextendsGoal

Goal satisfied when a target object is contained inside a specified container.

Inherited fromGoal1
is_satisfiedmethod
def is_satisfied(self) → bool

Evaluate the goal condition and return True if it is met.

class

GoalFactory

static-only1 method
fastsim.extensions.goal.GoalFactory

Static factory for creating Goal instances from a config dict.

createstaticmethod
@staticmethod def create(goal_type: str, config: dict) → Goal

Instantiate a Goal by type name and config parameters.

Parameters
NameTypeDefaultDescription
goal_typestr—Registered goal type name (e.g. 'pose', 'on_top').
configdict—Goal configuration parameters.
Returns
Goal — Constructed Goal instance.
class

Observer

abstract1 method
fastsim.extensions.observer.ObserverextendsABC

Abstract data observer that collects state information each tick.

observemethod
def observe(self) → Observation

Sample the current state and return an Observation.

Returns
Observation — Collected observation snapshot.
class

RobotObserver

0 methods
fastsim.extensions.observer.RobotObserverextendsObserver

Observes robot joint positions, velocities, and EE pose.

Inherited fromObserver1
observemethod
def observe(self) → Observation

Sample the current state and return an Observation.

class

SensorObserver

0 methods
fastsim.extensions.observer.SensorObserverextendsObserver

Observes raw data from a named sensor entity.

Inherited fromObserver1
observemethod
def observe(self) → Observation

Sample the current state and return an Observation.

class

ObjectObserver

0 methods
fastsim.extensions.observer.ObjectObserverextendsObserver

Observes the pose and state of a named object entity.

Inherited fromObserver1
observemethod
def observe(self) → Observation

Sample the current state and return an Observation.

class

TaskObserver

0 methods
fastsim.extensions.observer.TaskObserverextendsObserver

Observes task execution status (current action, progress, success flag).

Inherited fromObserver1
observemethod
def observe(self) → Observation

Sample the current state and return an Observation.

class

ObserverFactory

static-only1 method
fastsim.extensions.observer.ObserverFactory

Static factory for creating Observer instances from a config dict.

createstaticmethod
@staticmethod def create(observer_type: str, config: dict) → Observer

Instantiate an Observer by type name and config parameters.

Parameters
NameTypeDefaultDescription
observer_typestr—Registered observer type name (e.g. 'robot', 'sensor').
configdict—Observer configuration parameters.
Returns
Observer — Constructed Observer instance.
class

Observation

1 method
fastsim.extensions.observer.Observationextendsdict

Dict-like container for a single observation snapshot from one or more observers.

Observation subclasses dict, so all standard dict operations are available. Keys are observer-defined channel names; values are the sampled data arrays or scalars.

mergemethod
def merge(self, other: Observation) → None

Merge another Observation into this one (in-place union of keys).

Parameters
NameTypeDefaultDescription
otherObservation—Observation to merge.
Returns
None
class

Postprocessor

abstract1 method
fastsim.extensions.postprocessor.PostprocessorextendsABC

Abstract post-processing step applied to recorded trajectory data.

postprocessmethod
def postprocess(self, trajectory: Trajectory) → None

Apply this post-processing step to a trajectory.

Parameters
NameTypeDefaultDescription
trajectoryTrajectory—Trajectory to process.
Returns
None
class

ExportHdf5

0 methods
fastsim.extensions.postprocessor.ExportHdf5extendsPostprocessor

Post-processor that exports trajectory data to an HDF5 file.

Inherited fromPostprocessor1
postprocessmethod
def postprocess(self, trajectory: Trajectory) → None

Apply this post-processing step to a trajectory.

class

ExportVideo

0 methods
fastsim.extensions.postprocessor.ExportVideoextendsPostprocessor

Post-processor that renders recorded camera frames as an MP4 video.

Inherited fromPostprocessor1
postprocessmethod
def postprocess(self, trajectory: Trajectory) → None

Apply this post-processing step to a trajectory.

class

ExportPngDepth

0 methods
fastsim.extensions.postprocessor.ExportPngDepthextendsPostprocessor

Post-processor that saves recorded depth frames as individual PNG images.

Inherited fromPostprocessor1
postprocessmethod
def postprocess(self, trajectory: Trajectory) → None

Apply this post-processing step to a trajectory.

class

ExportPreviewVideo

0 methods
fastsim.extensions.postprocessor.ExportPreviewVideoextendsPostprocessor

Post-processor that produces a compact preview video from recorded data.

Inherited fromPostprocessor1
postprocessmethod
def postprocess(self, trajectory: Trajectory) → None

Apply this post-processing step to a trajectory.

class

PostprocessorFactory

static-only1 method
fastsim.extensions.postprocessor.PostprocessorFactory

Static factory for creating Postprocessor instances from a config dict.

createstaticmethod
@staticmethod def create(postprocessor_type: str, config: dict) → Postprocessor

Instantiate a Postprocessor by type name and config parameters.

Parameters
NameTypeDefaultDescription
postprocessor_typestr—Registered postprocessor type name (e.g. 'hdf5', 'video').
configdict—Postprocessor configuration parameters.
Returns
Postprocessor — Constructed Postprocessor instance.
class

Backend

abstract1 method
fastsim.extensions.backend.BackendextendsABC

Abstract storage/transport backend used by Recorder to persist trajectory data.

writemethod
def write(self, trajectory: Trajectory) → None

Write a trajectory to the backend.

Parameters
NameTypeDefaultDescription
trajectoryTrajectory—Trajectory data to persist.
Returns
None
class

LocalFileSystemBackend

0 methods
fastsim.extensions.backend.LocalFileSystemBackendextendsBackend

Saves trajectory data to the local filesystem.

Inherited fromBackend1
writemethod
def write(self, trajectory: Trajectory) → None

Write a trajectory to the backend.

class

Ros2Backend

0 methods
fastsim.extensions.backend.Ros2BackendextendsBackend

Publishes trajectory data as ROS2 topic messages.

Inherited fromBackend1
writemethod
def write(self, trajectory: Trajectory) → None

Write a trajectory to the backend.

class

BackendFactory

static-only1 method
fastsim.extensions.backend.BackendFactory

Static factory for creating Backend instances from a config dict.

createstaticmethod
@staticmethod def create(backend_type: str, config: dict) → Backend

Instantiate a Backend by type name and config parameters.

Parameters
NameTypeDefaultDescription
backend_typestr—Registered backend type (e.g. 'local', 'ros2').
configdict—Backend configuration parameters.
Returns
Backend — Constructed Backend instance.
class

TrajectoryTrack

2 methods
fastsim.extensions.trajectory.TrajectoryTrack

A single named data stream (channel) within a trajectory.

Each TrajectoryTrack holds a sequence of timestamped samples for one observable (e.g. joint positions, camera frames). Multiple tracks combine to form a Trajectory.

appendmethod
def append(self, timestamp: float, data: Any) → None

Append a timestamped sample to this track.

Parameters
NameTypeDefaultDescription
timestampfloat—Simulation time in seconds.
dataAny—Sample data (array, scalar, or dict).
Returns
None
get_samplesmethod
def get_samples(self) → list

Return all samples as a list of (timestamp, data) tuples.

Returns
list — List of (timestamp, data) pairs.
class

Trajectory

3 methods
fastsim.extensions.trajectory.Trajectory

Container for multiple TrajectoryTracks forming a complete episode recording.

add_trackmethod
def add_track(self, name: str, track: TrajectoryTrack) → None

Register a named track into this trajectory.

Parameters
NameTypeDefaultDescription
namestr—Channel name for the track.
trackTrajectoryTrack—Track instance to add.
Returns
None
get_trackmethod
def get_track(self, name: str) → TrajectoryTrack

Retrieve a named track.

Parameters
NameTypeDefaultDescription
namestr—Track channel name.
Returns
TrajectoryTrack — The track associated with the given name.
get_track_namesmethod
def get_track_names(self) → list

Return all registered track channel names.

Returns
list — List of track name strings.
class

TrajectoryLoader

static-only1 method
fastsim.extensions.trajectory.TrajectoryLoader

Loads Trajectory objects from a storage backend.

loadmethod
def load(self, source: str) → Trajectory

Load a trajectory from a path or identifier.

Parameters
NameTypeDefaultDescription
sourcestr—File path or backend identifier for the trajectory.
Returns
Trajectory — Loaded Trajectory object.