`fastsim.extensions`

46 个类

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.

class

ExtensionManager

静态类2 个方法

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.

参数

参数名	类型	默认值	说明
`sim`	FastSim	—	The active FastSim instance to hook extensions into.

返回值

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.

参数

参数名	类型	默认值	说明
`name`	str	—	Extension name as registered (e.g. 'benchmark', 'server').

返回值

Extension — The active Extension instance.

class

Extension

抽象类1 个方法

fastsim.extensions.extension.Extension继承自ABC

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.

参数

参数名	类型	默认值	说明
`sim`	FastSim	—	The running FastSim instance.

返回值

None

class

DataCollector

3 个方法

fastsim.extensions.data_collector.DataCollector继承自Extension

Extension that collects observations from registered observers each tick.

enablemethod

def enable(self, sim: FastSim) → None

Hook observers into the step loop.

参数

参数名	类型	默认值	说明
`sim`	FastSim	—	The running FastSim instance.

返回值

None

register_observermethod

def register_observer(self, observer: Observer) → None

Add an Observer instance to the collection pipeline.

参数

参数名	类型	默认值	说明
`observer`	Observer	—	Observer to register.

返回值

None

register_from_configmethod

def register_from_config(self) → None

Instantiate and register all observers defined in the extension config.

返回值

None

继承自Extension1

enablemethod

def enable(self, sim: FastSim) → None

Activate the extension against the provided FastSim instance.

class

Recorder

1 个方法

fastsim.extensions.recorder.Recorder继承自Extension

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.

参数

参数名	类型	默认值	说明
`sim`	FastSim	—	The running FastSim instance.

返回值

None

继承自Extension1

enablemethod

def enable(self, sim: FastSim) → None

Activate the extension against the provided FastSim instance.

class

Replayer

1 个方法

fastsim.extensions.replayer.Replayer继承自Extension

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.

参数

参数名	类型	默认值	说明
`sim`	FastSim	—	The running FastSim instance.

返回值

None

继承自Extension1

enablemethod

def enable(self, sim: FastSim) → None

Activate the extension against the provided FastSim instance.

class

Benchmark

1 个方法

fastsim.extensions.benchmark.Benchmark继承自Extension

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.

参数

参数名	类型	默认值	说明
`sim`	FastSim	—	The running FastSim instance.

返回值

None

继承自Extension1

enablemethod

def enable(self, sim: FastSim) → None

Activate the extension against the provided FastSim instance.

class

Server

1 个方法

fastsim.extensions.server.Server继承自Extension

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.

参数

参数名	类型	默认值	说明
`sim`	FastSim	—	The running FastSim instance.

返回值

None

继承自Extension1

enablemethod

def enable(self, sim: FastSim) → None

Activate the extension against the provided FastSim instance.

class

Policy

抽象类6 个方法

fastsim.extensions.benchmark.policy.Policy继承自ABC

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.

返回值

None

warmupmethod

def warmup(self, benchmark_observation: Any) → None

Perform any model warm-up using an initial observation.

参数

参数名	类型	默认值	说明
`benchmark_observation`	Any	—	Raw observation from the benchmark environment.

返回值

None

needs_observationmethod

def needs_observation(self) → bool

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

返回值

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.

参数

参数名	类型	默认值	说明
`benchmark_observation`	Any	—	Raw observation data.

返回值

dict — Preprocessed observation dict.

compute_actionmethod

def compute_action(self, observation: dict) → dict

Run inference and return a raw action dict.

参数

参数名	类型	默认值	说明
`observation`	dict	—	Preprocessed observation dict.

返回值

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.

参数

参数名	类型	默认值	说明
`action`	dict	—	Raw action dict from compute_action.

返回值

BenchmarkAction — The fully constructed action object.

class

Action

抽象类2 个方法

fastsim.extensions.action.Action继承自ABC

Abstract action executed by the task/benchmark system.

actmethod

def act(self) → ActionResultStatus

Execute the action and return a result status.

返回值

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.

返回值

bool

class

GraspAction

0 个方法

fastsim.extensions.action.GraspAction继承自Action

Close the gripper to grasp a target object.

继承自Action2

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 个方法

fastsim.extensions.action.MoveAction继承自Action

Move the end-effector to a target pose.

继承自Action2

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 个方法

fastsim.extensions.action.ReleaseAction继承自Action

Open the gripper to release a currently held object.

继承自Action2

actmethod

def act(self) → ActionResultStatus

参数

参数名	类型	默认值	说明
`action_type`	str	—	Registered action type name (e.g. 'pick', 'place').
`config`	dict	—	Action configuration parameters.

返回值

Action — Constructed Action instance.

class

Goal

抽象类1 个方法

fastsim.extensions.goal.Goal继承自ABC

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.

返回值

bool

class

PoseGoal

0 个方法

fastsim.extensions.goal.PoseGoal继承自Goal

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

继承自Goal1

is_satisfiedmethod

def is_satisfied(self) → bool

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

class

OnTopGoal

0 个方法

fastsim.extensions.goal.OnTopGoal继承自Goal

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

继承自Goal1

is_satisfiedmethod

def is_satisfied(self) → bool

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

class

InsideGoal

0 个方法

fastsim.extensions.goal.InsideGoal继承自Goal

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

继承自Goal1

is_satisfiedmethod

def is_satisfied(self) → bool

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

class

GoalFactory

静态类1 个方法

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.

参数

参数名	类型	默认值	说明
`goal_type`	str	—	Registered goal type name (e.g. 'pose', 'on_top').
`config`	dict	—	Goal configuration parameters.

返回值

Goal — Constructed Goal instance.

class

Observer

抽象类1 个方法

fastsim.extensions.observer.Observer继承自ABC

Abstract data observer that collects state information each tick.

observemethod

def observe(self) → Observation

Sample the current state and return an Observation.

返回值

Observation — Collected observation snapshot.

class

RobotObserver

0 个方法

fastsim.extensions.observer.RobotObserver继承自Observer

Observes robot joint positions, velocities, and EE pose.

继承自Observer1

observemethod

def observe(self) → Observation

Sample the current state and return an Observation.

class

SensorObserver

0 个方法

fastsim.extensions.observer.SensorObserver继承自Observer

Observes raw data from a named sensor entity.

继承自Observer1

observemethod

def observe(self) → Observation

Sample the current state and return an Observation.

class

ObjectObserver

0 个方法

fastsim.extensions.observer.ObjectObserver继承自Observer

Observes the pose and state of a named object entity.

继承自Observer1

observemethod

def observe(self) → Observation

Sample the current state and return an Observation.

class

TaskObserver

0 个方法

fastsim.extensions.observer.TaskObserver继承自Observer

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

继承自Observer1

observemethod

def observe(self) → Observation

Sample the current state and return an Observation.

class

ObserverFactory

静态类1 个方法

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.

参数

参数名	类型	默认值	说明
`observer_type`	str	—	Registered observer type name (e.g. 'robot', 'sensor').
`config`	dict	—	Observer configuration parameters.

返回值

Observer — Constructed Observer instance.

class

Observation

1 个方法

fastsim.extensions.observer.Observation继承自dict

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).

参数

参数名	类型	默认值	说明
`other`	Observation	—	Observation to merge.

返回值

None

class

Postprocessor

抽象类1 个方法

fastsim.extensions.postprocessor.Postprocessor继承自ABC

Abstract post-processing step applied to recorded trajectory data.

postprocessmethod

def postprocess(self, trajectory: Trajectory) → None

Apply this post-processing step to a trajectory.

参数

参数名	类型	默认值	说明
`trajectory`	Trajectory	—	Trajectory to process.

返回值

None

class

ExportHdf5

0 个方法

fastsim.extensions.postprocessor.ExportHdf5继承自Postprocessor

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

继承自Postprocessor1

postprocessmethod

def postprocess(self, trajectory: Trajectory) → None

Apply this post-processing step to a trajectory.

class

ExportVideo

0 个方法

fastsim.extensions.postprocessor.ExportVideo继承自Postprocessor

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

继承自Postprocessor1

postprocessmethod

def postprocess(self, trajectory: Trajectory) → None

Apply this post-processing step to a trajectory.

class

ExportPngDepth

0 个方法

fastsim.extensions.postprocessor.ExportPngDepth继承自Postprocessor

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

继承自Postprocessor1

postprocessmethod

def postprocess(self, trajectory: Trajectory) → None

Apply this post-processing step to a trajectory.

class

ExportPreviewVideo

0 个方法

fastsim.extensions.postprocessor.ExportPreviewVideo继承自Postprocessor

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

继承自Postprocessor1

postprocessmethod

def postprocess(self, trajectory: Trajectory) → None

Apply this post-processing step to a trajectory.

class

PostprocessorFactory

静态类1 个方法

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.

参数

参数名	类型	默认值	说明
`postprocessor_type`	str	—	Registered postprocessor type name (e.g. 'hdf5', 'video').
`config`	dict	—	Postprocessor configuration parameters.

返回值

Postprocessor — Constructed Postprocessor instance.

class

Backend

抽象类1 个方法

fastsim.extensions.backend.Backend继承自ABC

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

writemethod

def write(self, trajectory: Trajectory) → None

Write a trajectory to the backend.

参数

参数名	类型	默认值	说明
`trajectory`	Trajectory	—	Trajectory data to persist.

返回值

None

class

LocalFileSystemBackend

0 个方法

fastsim.extensions.backend.LocalFileSystemBackend继承自Backend

Saves trajectory data to the local filesystem.

继承自Backend1

writemethod

def write(self, trajectory: Trajectory) → None

Write a trajectory to the backend.

class

Ros2Backend

0 个方法

fastsim.extensions.backend.Ros2Backend继承自Backend

Publishes trajectory data as ROS2 topic messages.

继承自Backend1

writemethod

def write(self, trajectory: Trajectory) → None

Write a trajectory to the backend.

class

BackendFactory

静态类1 个方法

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.

参数

参数名	类型	默认值	说明
`backend_type`	str	—	Registered backend type (e.g. 'local', 'ros2').
`config`	dict	—	Backend configuration parameters.

返回值

Backend — Constructed Backend instance.

class

TrajectoryTrack

2 个方法

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.

参数

参数名	类型	默认值	说明
`timestamp`	float	—	Simulation time in seconds.
`data`	Any	—	Sample data (array, scalar, or dict).

返回值

None

get_samplesmethod

def get_samples(self) → list

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

返回值

list — List of (timestamp, data) pairs.

class

Trajectory

3 个方法

fastsim.extensions.trajectory.Trajectory

Container for multiple TrajectoryTracks forming a complete episode recording.

add_trackmethod

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

参数

参数名	类型	默认值	说明
`name`	str	—	Channel name for the track.
`track`	TrajectoryTrack	—	Track instance to add.

返回值

None

get_trackmethod

def get_track(self, name: str) → TrajectoryTrack

Retrieve a named track.

参数

参数名	类型	默认值	说明
`name`	str	—	Track channel name.

返回值

TrajectoryTrack — The track associated with the given name.

get_track_namesmethod

def get_track_names(self) → list

Return all registered track channel names.

返回值

list — List of track name strings.

class

TrajectoryLoader

静态类1 个方法

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.

参数

参数名	类型	默认值	说明
`source`	str	—	File path or backend identifier for the trajectory.

返回值

Trajectory — Loaded Trajectory object.