← Python SDK / application

PYTHONPUBLIC API

fastsim.api.application

fastsim.api.application 的当前公共 Python API；签名从 vNext AST 直接提取。

ApplicationStateSnapshot ApplicationControlStatus ApplicationControlError ApplicationControlOperation ApplicationControlClient ApplicationSceneClient ApplicationVisualizationClient ApplicationFrameClient ApplicationGeometryResource ApplicationGeometryClient ApplicationScenarioView CameraRgbSubscription ApplicationKinematicsClient FastSimApplication application_plan application

公开符号 16

class

ApplicationStateSnapshot

L133

class ApplicationStateSnapshot(object)

One coherent committed world/observation envelope.

字段	类型	默认值
world	WorldStateSnapshot	—
observation	ObservationSnapshot	—

run_id

ApplicationStateSnapshot.run_id(self) -> str

当前源码未提供公开 docstring。

generation

ApplicationStateSnapshot.generation(self) -> int

当前源码未提供公开 docstring。

sim_tick

ApplicationStateSnapshot.sim_tick(self) -> int

当前源码未提供公开 docstring。

sim_time_s

ApplicationStateSnapshot.sim_time_s(self) -> float

当前源码未提供公开 docstring。

class

ApplicationControlStatus

L175

class ApplicationControlStatus(object)

Portable progress for one application-owned control operation.

字段	类型	默认值
operation_id	str	—
chunk_id	str	—
status	ChunkStatus	—
done	bool	—
progress	ChunkProgress	—

class

ApplicationControlError

L195

class ApplicationControlError(RuntimeError)

Stable direct-application input or capability error.

class

ApplicationControlOperation

L602

class ApplicationControlOperation(object)

Stable non-blocking handle; cancellation linearizes through ControlService.

status

async ApplicationControlOperation.status(self) -> ApplicationControlStatus

当前源码未提供公开 docstring。

result

async ApplicationControlOperation.result(self, timeout: float | None=None) -> ChunkResult

当前源码未提供公开 docstring。

cancel

async ApplicationControlOperation.cancel(self, timeout: float | None=None) -> bool

当前源码未提供公开 docstring。

release

async ApplicationControlOperation.release(self) -> bool

当前源码未提供公开 docstring。

class

ApplicationControlClient

L642

class ApplicationControlClient(object)

Physical direct-control surface for joint paths and generic tracks.

targets

async ApplicationControlClient.targets(self) -> tuple[ControlTargetDescriptor, ...]

当前源码未提供公开 docstring。

state

async ApplicationControlClient.state(self, timeout: float | None=None) -> ControlContext

当前源码未提供公开 docstring。

subscribe_applied_frames

async ApplicationControlClient.subscribe_applied_frames(self, *, capacity: int=64, max_item_bytes: int=1048576, overload: ControlAuditOverloadPolicy | str=ControlAuditOverloadPolicy.DROP_OLDEST) -> ScopedControlAuditSubscription

Subscribe to bounded, history-free ``APPLIED_FRAME`` audit envelopes. Every non-gap item is a :class:`ControlAuditEnvelope` whose kind is ``APPLIED_FRAME`` and whose payload is an :class:`AppliedControlFrame`. The payload is directly serializable with ``applied_frame_to_dict`` or ``encode_applied_frame``. Closing this revocable subscription removes its audit demand immediately.

submit_joints

async ApplicationControlClient.submit_joints(self, *, actor: str, group: str, path: Sequence[Sequence[float]], dt: float, preempt: bool=False, timeout: float | None=None) -> ApplicationControlOperation

当前源码未提供公开 docstring。

joints

async ApplicationControlClient.joints(self, **kwargs: Any) -> ChunkResult

当前源码未提供公开 docstring。

submit_tracks

async ApplicationControlClient.submit_tracks(self, tracks: Sequence[ControlTrack], *, preempt: bool=False, timeout: float | None=None) -> ApplicationControlOperation

当前源码未提供公开 docstring。

tracks

async ApplicationControlClient.tracks(self, tracks: Sequence[ControlTrack], *, preempt: bool=False, timeout: float | None=None) -> ChunkResult

当前源码未提供公开 docstring。

class

ApplicationSceneClient

L766

class ApplicationSceneClient(_ApplicationReadClient)

当前源码未提供公开 docstring。

catalog

async ApplicationSceneClient.catalog(self, query: SceneQuery | None=None)

当前源码未提供公开 docstring。

state

async ApplicationSceneClient.state(self, query: SceneQuery | None=None)

当前源码未提供公开 docstring。

set_pose

async ApplicationSceneClient.set_pose(self, entity_id: str, *, xyz_m: Sequence[float], quat_xyzw: Sequence[float]=(0.0, 0.0, 0.0, 1.0), generation: int | None=None, command_id: str | None=None, deadline_s: float | None=None) -> SceneCommandResult

Set a rigid or articulated entity asset-root pose in world coordinates. On USD backends the target is the spawned entity Prim, never an inferred articulation root link. Query root-link state explicitly when it is required.

begin_drag

async ApplicationSceneClient.begin_drag(self, entity_id: str, *, grab_point_world_m: Sequence[float], generation: int | None=None, command_id: str | None=None, drag_id: str | None=None, deadline_s: float | None=None) -> SceneCommandResult

Begin one explicit backend-neutral kinematic drag transaction.

attach

async ApplicationSceneClient.attach(self, child_entity_id: str, *, parent_entity_id: str, parent_link_name: str | None=None, child_link_name: str | None=None, parent_T_child_xyz_m: Sequence[float] | None=None, parent_T_child_quat_xyzw: Sequence[float]=(0.0, 0.0, 0.0, 1.0), generation: int | None=None, command_id: str | None=None, attachment_id: str | None=None, deadline_s: float | None=None) -> SceneCommandResult

Fix a rigid body or articulation link to another physical endpoint.

detach

async ApplicationSceneClient.detach(self, child_entity_id: str, attachment_id: str, *, generation: int | None=None, command_id: str | None=None, deadline_s: float | None=None) -> SceneCommandResult

Remove one active attachment and leave the child at its committed pose.

update_drag

async ApplicationSceneClient.update_drag(self, entity_id: str, drag_id: str, *, xyz_m: Sequence[float], quat_xyzw: Sequence[float]=(0.0, 0.0, 0.0, 1.0), generation: int | None=None, command_id: str | None=None, deadline_s: float | None=None) -> SceneCommandResult

当前源码未提供公开 docstring。

end_drag

async ApplicationSceneClient.end_drag(self, entity_id: str, drag_id: str, *, generation: int | None=None, command_id: str | None=None, deadline_s: float | None=None) -> SceneCommandResult

当前源码未提供公开 docstring。

cancel_drag

async ApplicationSceneClient.cancel_drag(self, entity_id: str, drag_id: str, *, generation: int | None=None, command_id: str | None=None, deadline_s: float | None=None) -> SceneCommandResult

当前源码未提供公开 docstring。

class

ApplicationVisualizationClient

L946

class ApplicationVisualizationClient(object)

Asynchronous application facade for native and portable debug overlays.

publish

async ApplicationVisualizationClient.publish(self, primitives: Sequence[VisualizationPrimitive], *, mesh_resources: Sequence[VisualizationMeshResource]=(), event_id: str | None=None, generation: int | None=None, timeout: float | None=None) -> VisualizationPublishResult

当前源码未提供公开 docstring。

clear

当前源码未提供公开 docstring。

metric

async ApplicationVisualizationClient.metric(self, series_id: str, value: float | Mapping[str, float], *, unit: str='1', units: Mapping[str, str] | None=None, metadata: Mapping[str, object] | None=None) -> VisualizationMetricRecord

当前源码未提供公开 docstring。

event

async ApplicationVisualizationClient.event(self, event_id: str, message: str, *, category: str='application', severity: VisualizationSeverity | str=VisualizationSeverity.INFO, metadata: Mapping[str, object] | None=None) -> VisualizationEventRecord

当前源码未提供公开 docstring。

snapshot

async ApplicationVisualizationClient.snapshot(self) -> VisualizationSnapshot

当前源码未提供公开 docstring。

class

ApplicationFrameClient

L1052

class ApplicationFrameClient(_ApplicationReadClient)

当前源码未提供公开 docstring。

catalog

async ApplicationFrameClient.catalog(self, query: FrameQuery | None=None)

当前源码未提供公开 docstring。

lookup

async ApplicationFrameClient.lookup(self, target_frame: str, source_frame: str, *, generation: int | None=None)

当前源码未提供公开 docstring。

transform

async ApplicationFrameClient.transform(self, source_frame: str, target_frame: str, *, generation: int | None=None)

Return ``target_frame_T_source_frame`` for explicit source/target order.

class

ApplicationGeometryResource

L1080

class ApplicationGeometryResource(object)

Async owner-loop proxy for one provider-owned geometry resource lease.

descriptor

async ApplicationGeometryResource.descriptor(self)

当前源码未提供公开 docstring。

read

async ApplicationGeometryResource.read(self, offset: int=0, length: int | None=None) -> bytes

当前源码未提供公开 docstring。

async ApplicationGeometryResource.close(self) -> None

当前源码未提供公开 docstring。

class

ApplicationGeometryClient

L1127

class ApplicationGeometryClient(_ApplicationReadClient)

当前源码未提供公开 docstring。

capabilities

async ApplicationGeometryClient.capabilities(self)

当前源码未提供公开 docstring。

catalog

async ApplicationGeometryClient.catalog(self, query: GeometryQuery | None=None)

当前源码未提供公开 docstring。

transforms

async ApplicationGeometryClient.transforms(self, query: GeometryQuery | None=None)

当前源码未提供公开 docstring。

resolve

async ApplicationGeometryClient.resolve(self, geometry_id: str, representation: GeometryRepresentation | str | None=None)

当前源码未提供公开 docstring。

class

ApplicationScenarioView

L1156

class ApplicationScenarioView(object)

Immutable Scenario data exactly as frozen into the Run.

字段	类型	默认值
source	FrozenDict[str, object]	—
scene	FrozenDict[str, object]	—
behavior	FrozenDict[str, object] \| None	—
evaluation	FrozenDict[str, object] \| None	—
digest	str	—

class

CameraRgbSubscription

L1166

class CameraRgbSubscription(object)

One generation-fenced, capacity-one asynchronous camera RGB stream.

entity_id

CameraRgbSubscription.entity_id(self) -> str

当前源码未提供公开 docstring。

run_id

CameraRgbSubscription.run_id(self) -> str

当前源码未提供公开 docstring。

generation

CameraRgbSubscription.generation(self) -> int

当前源码未提供公开 docstring。

capacity

CameraRgbSubscription.capacity(self) -> int

当前源码未提供公开 docstring。

dropped_frames

CameraRgbSubscription.dropped_frames(self) -> int

当前源码未提供公开 docstring。

closed

CameraRgbSubscription.closed(self) -> bool

当前源码未提供公开 docstring。

receive

async CameraRgbSubscription.receive(self, timeout: float | None=None) -> CameraRgbFrame

Return the latest selected frame, always newer than the prior result.

async CameraRgbSubscription.close(self) -> None

Remove capture demand promptly; repeated calls are harmless.

class

ApplicationKinematicsClient

L1378

class ApplicationKinematicsClient(object)

Beginner-first, solve-only kinematics client for one Application.

descriptor

async ApplicationKinematicsClient.descriptor(self) -> KinematicsSolverDescriptor

Return the explicitly selected solver's validated descriptor.

solve

async ApplicationKinematicsClient.solve(self, robot: str, base_frame: str, tip_frame: str, *, xyz_m: Sequence[float], quat_xyzw: Sequence[float], reference_frame: str | None=None, request_id: str | None=None, group: str | None=None, start: object | None=None, locked_joints: Sequence[object]=(), position_tolerance_m: float=0.0001, orientation_tolerance_rad: float=0.001, max_iterations: int=256, timeout_s: float=1.0, collision_mode: str='none', seed: int | None=None, max_solutions: int=1) -> IKSolveResponse

Solve an EE pose without sending any control command. ``robot`` accepts either a canonical entity ID such as ``robots.arm`` or its Scenario alias such as ``arm``. Omitting ``start`` captures the current joint state atomically. Multi-group robots may pass ``group`` to select the exact ordered chain state.

solve_request

async ApplicationKinematicsClient.solve_request(self, request: IKRequest) -> IKSolveResponse

Solve one advanced canonical SPI request exactly as supplied.

class

FastSimApplication

L1483

class FastSimApplication(object)

Asynchronous facade for one immutable direct-development Run.

run_id

FastSimApplication.run_id(self) -> str

当前源码未提供公开 docstring。

plan_digest

FastSimApplication.plan_digest(self) -> str

当前源码未提供公开 docstring。

authority

FastSimApplication.authority(self) -> ApplicationAuthority

当前源码未提供公开 docstring。

control

FastSimApplication.control(self) -> ApplicationControlClient

当前源码未提供公开 docstring。

scene

FastSimApplication.scene(self) -> ApplicationSceneClient

当前源码未提供公开 docstring。

frames

FastSimApplication.frames(self) -> ApplicationFrameClient

当前源码未提供公开 docstring。

geometry

FastSimApplication.geometry(self) -> ApplicationGeometryClient

当前源码未提供公开 docstring。

kinematics

FastSimApplication.kinematics(self) -> ApplicationKinematicsClient

Access the lazy solve-only kinematics service.

visualization

FastSimApplication.visualization(self) -> ApplicationVisualizationClient

Access portable debug drawing without a backend SDK.

scenario

FastSimApplication.scenario(self) -> ApplicationScenarioView

当前源码未提供公开 docstring。

open

async FastSimApplication.open(self, timeout: float | None=None) -> FastSimApplication

当前源码未提供公开 docstring。

prepare

async FastSimApplication.prepare(self, timeout: float | None=None) -> ApplicationSnapshot

当前源码未提供公开 docstring。

start

async FastSimApplication.start(self, timeout: float | None=None) -> ApplicationSnapshot

当前源码未提供公开 docstring。

pause

async FastSimApplication.pause(self, timeout: float | None=None) -> ApplicationSnapshot

当前源码未提供公开 docstring。

resume

async FastSimApplication.resume(self, timeout: float | None=None) -> ApplicationSnapshot

当前源码未提供公开 docstring。

single_step

async FastSimApplication.single_step(self, timeout: float | None=None) -> ApplicationSnapshot

当前源码未提供公开 docstring。

reset

async FastSimApplication.reset(self, timeout: float | None=None) -> ApplicationSnapshot

当前源码未提供公开 docstring。

stop

async FastSimApplication.stop(self, timeout: float | None=None) -> ApplicationSnapshot

当前源码未提供公开 docstring。

snapshot

async FastSimApplication.snapshot(self) -> ApplicationSnapshot

当前源码未提供公开 docstring。

plugin_status

async FastSimApplication.plugin_status(self) -> PluginHostSnapshot | None

当前源码未提供公开 docstring。

plugin_events

async FastSimApplication.plugin_events(self) -> tuple[PublishedPluginEvent, ...]

Return the bounded immutable history of events published by plugins.

run_info

async FastSimApplication.run_info(self) -> RunInfoSnapshot

Return immutable public identity for this direct Run.

state

async FastSimApplication.state(self) -> ApplicationStateSnapshot

当前源码未提供公开 docstring。

camera_rgb

async FastSimApplication.camera_rgb(self, entity_id: str, timeout: float | None=None) -> CameraRgbFrame

当前源码未提供公开 docstring。

start_fluid_emitter

async FastSimApplication.start_fluid_emitter(self, request: FluidEmitterRequest) -> FluidEmitterStatus

Start one simulation-time particle emitter after prepare.

fluid_emitter_status

async FastSimApplication.fluid_emitter_status(self, emitter_id: str) -> FluidEmitterStatus

当前源码未提供公开 docstring。

stop_fluid_emitter

async FastSimApplication.stop_fluid_emitter(self, emitter_id: str) -> FluidEmitterStatus

Permanently stop one liquid emitter.

set_fluid_emitter_enabled

async FastSimApplication.set_fluid_emitter_enabled(self, emitter_id: str, enabled: bool) -> FluidEmitterStatus

Pause or resume one non-terminal liquid emitter.

create_fluid_reservoir

async FastSimApplication.create_fluid_reservoir(self, spec: FluidReservoirSpec) -> FluidReservoirStatus

Create one conserved liquid reservoir for this Run.

fluid_reservoir_status

async FastSimApplication.fluid_reservoir_status(self, reservoir_id: str) -> FluidReservoirStatus

当前源码未提供公开 docstring。

set_fluid_reservoir_volume

async FastSimApplication.set_fluid_reservoir_volume(self, reservoir_id: str, volume_ml: float) -> FluidReservoirStatus

Refill or drain a reservoir without recreating the scene.

subscribe_camera_rgb

async FastSimApplication.subscribe_camera_rgb(self, entity_id: str, max_rate_hz: float=30.0, capacity: int=1, timeout: float | None=None) -> CameraRgbSubscription

Open a latest-frame RGB subscription pinned to the active generation.

particle_fluid

async FastSimApplication.particle_fluid(self, entity_id: str, timeout: float | None=None) -> ParticleFluidSnapshot

当前源码未提供公开 docstring。

deformable

async FastSimApplication.deformable(self, entity_id: str, timeout: float | None=None) -> DeformableSnapshot

Read one deformable's current world-frame node state.

async FastSimApplication.close(self, timeout: float | None=None) -> ApplicationSnapshot | None

当前源码未提供公开 docstring。

function

application_plan

L2381

application_plan(plan: ExecutionPlan, *, authority: ApplicationAuthority | str=ApplicationAuthority.EXCLUSIVE, timeout: float | None=None, planning_reads: bool=False, kinematics_solver: str=DEFAULT_KINEMATICS_SOLVER_ID, kinematics_config: Mapping[str, object] | None=None) -> FastSimApplication

Create a lazy asynchronous application from one compiled Plan. Planning scene/frame/geometry publication is pay-for-play. Set ``planning_reads=True`` only when those APIs are required; the default avoids locking ``planning.scene@2`` into ordinary control or sensor runs. Disabled planning calls fail with their typed ``SERVICE_ACCESS_DENIED`` error and never return a partial collision scene.

function

application

L2450

application(path: str | Path, lock: str | Path | None=None, *, authority: ApplicationAuthority | str=ApplicationAuthority.EXCLUSIVE, launch_profile: str | None=None, timeout: float | None=None, planning_reads: bool=False, kinematics_solver: str=DEFAULT_KINEMATICS_SOLVER_ID, kinematics_config: Mapping[str, object] | None=None) -> FastSimApplication

Compile a project Run and return its lazy asynchronous application. ``planning_reads`` has the same opt-in semantics as :func:`application_plan`. ``launch_profile`` is an unlocked compilation override and cannot modify an existing lock.