← Python SDK / services

PYTHONPUBLIC API

fastsim.api.services

Public Python API for fastsim.api.services; signatures are extracted directly from FastSim source.

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

Public symbols 15

class

ControlAxisDescriptor

L76

class ControlAxisDescriptor(object)

One ordered command axis and its accepted units.

Field	Type	Default
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.

Field	Type	Default
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.

Field	Type	Default
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.

Field	Type	Default
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.

Field	Type	Default
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.

Field	Type	Default
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.

Field	Type	Default
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