← Python SDK / services

PYTHONPUBLIC API

fastsim.api.services

fastsim.api.services 的公共 Python API；签名从 FastSim 源码直接提取。

ControlAxisDescriptor ControlCommandSpaceDescriptor ControlControllerExecutionDescriptor ControlEmbeddedBaseJointDescriptor ControlTargetDescriptor RunInfoSnapshot PublishedPluginEvent ScopedControlSession ScopedControlService ScopedControlTargetService ScopedObservationSubscription ScopedObservationChannelSubscription ScopedObservationService RunInfoService EventPublishService

公开符号 15

class

ControlAxisDescriptor

L76

class ControlAxisDescriptor(object)

One ordered command axis and its accepted units.

字段	类型	默认值
name	str	—
units	tuple[str, ...]	—

class

ControlCommandSpaceDescriptor

L96

class ControlCommandSpaceDescriptor(object)

Portable resource-specific contract for one command space. Frame-capable spaces always carry an explicit ``allowed_frames`` tuple. A required-frame space needs at least one entry; an optional space may use an empty tuple to advertise frameless-only operation. Frame-forbidden spaces may use either ``None`` or an empty tuple. ``unit_rule`` states whether the advertised per-axis unit choices are fixed, must be selected uniformly across all axes, or may be selected per axis. This is a track-target contract. Terminal policies and controller convergence are exposed separately by the enclosing target's controller execution descriptors.

字段	类型	默认值
command_space_id	str	—
controller_ids	tuple[str, ...]	—
axes	tuple[ControlAxisDescriptor, ...]	—
frame_requirement	str	—
allowed_frames	tuple[str, ...] \| None	—
default_frame	str \| None	—
required_semantics	Mapping[str, tuple[str, ...]]	—
unit_rule	str	'per_axis'
supports_linear	bool	True

class

ControlControllerExecutionDescriptor

L189

class ControlControllerExecutionDescriptor(object)

Portable chunk-execution choices for one target-compatible controller.

字段	类型	默认值
controller_id	str	—
terminal_policies	tuple[str, ...]	—
supports_convergence	bool	—

class

ControlEmbeddedBaseJointDescriptor

L214

class ControlEmbeddedBaseJointDescriptor(object)

Capture-safe manifest declaration for an embedded physical SE(2) base.

字段	类型	默认值
declaration_id	str	—
physical_joint_names	tuple[str, str, str]	—
units	tuple[str, str, str]	—
supported_models	tuple[str, ...]	—
controller_id	str	—

class

ControlTargetDescriptor

L249

class ControlTargetDescriptor(object)

One immutable target visible through a plugin's exact control grant.

字段	类型	默认值
binding_role	str	—
entity_id	str	—
resource_group	str	—
controller_ids	tuple[str, ...]	—
controller_executions	tuple[ControlControllerExecutionDescriptor, ...]	—
command_space_ids	tuple[str, ...]	—
command_spaces	tuple[ControlCommandSpaceDescriptor, ...]	—
max_command_rate_hz	float	—
embedded_base_joints	ControlEmbeddedBaseJointDescriptor \| None	None

class

RunInfoSnapshot

L323

class RunInfoSnapshot(object)

Small immutable run identity safe to expose outside the core.

字段	类型	默认值
run_id	str	—
plan_digest	str	—
name	str	—
backend	str	—

class

PublishedPluginEvent

L341

class PublishedPluginEvent(object)

One immutable, core-namespaced event emitted by a plugin.

字段	类型	默认值
run_id	str	—
plan_digest	str	—
instance	str	—
topic	str	—
sequence	int	—
published_monotonic_ns	int	—
payload	Mapping[str, object]	—

qualified_topic

PublishedPluginEvent.qualified_topic(self) -> str

Return the unspoofable instance namespace followed by the local topic.

class

ScopedControlSession

L422

class ScopedControlSession(Protocol)

Asynchronous control session with a policy fixed by the composition root.

session_id

ScopedControlSession.session_id(self) -> str

run_id

ScopedControlSession.run_id(self) -> str

source_id

ScopedControlSession.source_id(self) -> str

closed

ScopedControlSession.closed(self) -> bool

state

ScopedControlSession.state(self) -> ControlSessionState

submit

async ScopedControlSession.submit(self, chunk: ControlChunk, *, planning_submission: object | None=None) -> ControlSubmission

cancel

async ScopedControlSession.cancel(self, chunk_id: str, timeout: float | None=None) -> ControlOperationTicket[bool]

heartbeat

async ScopedControlSession.heartbeat(self, timeout: float | None=None) -> ControlOperationTicket[None]

observe

async ScopedControlSession.observe(self, timeout: float | None=None) -> ControlContext

query

async ScopedControlSession.query(self, chunk_id: str, timeout: float | None=None) -> ChunkTicket

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

class

ScopedControlService

L459

class ScopedControlService(Protocol)

Factory for plugin-scoped control sessions.

service_id

ScopedControlService.service_id(self) -> str

open_session

async ScopedControlService.open_session(self, timeout: float | None=None) -> ScopedControlSession

class

ScopedControlTargetService

L469

class ScopedControlTargetService(Protocol)

Read-only exact-resource view of discoverable control targets.

service_id

ScopedControlTargetService.service_id(self) -> str

descriptors

ScopedControlTargetService.descriptors(self) -> tuple[ControlTargetDescriptor, ...]

get

ScopedControlTargetService.get(self, binding_role: str, resource_group: str) -> ControlTargetDescriptor

class

ScopedObservationSubscription

L482

class ScopedObservationSubscription(Protocol)

Bounded observation stream that cannot reveal ungranted channels. Application-loop plugins must use :meth:`get_async`; synchronous ``get`` is retained only for a plugin-owned worker thread.

closed

ScopedObservationSubscription.closed(self) -> bool

dropped

ScopedObservationSubscription.dropped(self) -> int

get

ScopedObservationSubscription.get(self, timeout: float | None=None) -> ObservationStreamItem

get_async

async ScopedObservationSubscription.get_async(self, timeout: float | None=None) -> ObservationStreamItem

ScopedObservationSubscription.close(self) -> None

class

ScopedObservationChannelSubscription

L503

class ScopedObservationChannelSubscription(Protocol)

Bounded independently stamped observation-channel stream.

closed

ScopedObservationChannelSubscription.closed(self) -> bool

dropped

ScopedObservationChannelSubscription.dropped(self) -> int

get

ScopedObservationChannelSubscription.get(self, timeout: float | None=None) -> ObservationChannelStreamItem

get_async

async ScopedObservationChannelSubscription.get_async(self, timeout: float | None=None) -> ObservationChannelStreamItem

ScopedObservationChannelSubscription.close(self) -> None

class

ScopedObservationService

L520

class ScopedObservationService(Protocol)

Read-only exact-channel view of the runtime observation service.

service_id

ScopedObservationService.service_id(self) -> str

channels

ScopedObservationService.channels(self) -> tuple[str, ...]

descriptors

ScopedObservationService.descriptors(self) -> tuple[ObservationDescriptor, ...]

latest_snapshot

ScopedObservationService.latest_snapshot(self, *, generation: int | None=None) -> ObservationSnapshot

snapshot

ScopedObservationService.snapshot(self, tick: int, *, generation: int | None=None) -> ObservationSnapshot

latest

ScopedObservationService.latest(self, key: ObservationKey | str, *, generation: int | None=None, current_tick: int | None=None, max_age_ticks: int | None=None, allow_stale: bool=False) -> ObservationSample

ScopedObservationService.subscribe(self, capacity: int=16, overflow: SubscriptionOverflowPolicy=SubscriptionOverflowPolicy.DROP_OLDEST, *, replay_history: bool=False, from_record_sequence: int | None=None) -> ScopedObservationSubscription

subscribe_channels

ScopedObservationService.subscribe_channels(self, capacity: int=16, overflow: SubscriptionOverflowPolicy=SubscriptionOverflowPolicy.DROP_OLDEST, *, replay_history: bool=False, from_record_sequence: int | None=None) -> ScopedObservationChannelSubscription

class

RunInfoService

L566

class RunInfoService(Protocol)

Read-only immutable metadata for the current run.

service_id

RunInfoService.service_id(self) -> str

snapshot

RunInfoService.snapshot(self) -> RunInfoSnapshot

get

RunInfoService.get(self) -> RunInfoSnapshot

class

EventPublishService

L579

class EventPublishService(Protocol)

Namespaced, bounded and portable plugin event publisher.

service_id

EventPublishService.service_id(self) -> str

publish

EventPublishService.publish(self, topic: str, payload: Mapping[str, object] | None=None) -> PublishedPluginEvent