lsdtools.runtime reference

The runtime service's public contract for the packages that author scenes and simulation
systems: lazy, headless.

A scene package's steps call ``build_scene_tables`` and ``scene_layer_payload``; its templates
derive identities with ``scene_id_for``; its declarations name ``OBJECT_NAMESPACE`` and
``SCENE_LAYER_KIND``; its tree seams ask ``member_kind`` which runtime role an entity plays.
A package that plays in a composition registers its participant with ``register_system`` (a
``SimulationSystem`` built per loaded composition, its ``SystemPhases`` around the fixed step,
its ``rank`` among the others) and reaches the loaded scene through the ``SceneRuntime`` the
driver hands its factory; ``BUILT_IN_PARTICIPANTS`` are the names the runtime keeps. Importing
this facade loads no engine, no driver and no view.

BUILT_IN_PARTICIPANTS = ('scene', 'lighting', 'ui', 'timeline')

CONTRACTS_PIN: Final

CONTRACTS_PIN = f'lsd-runtime-scene-v{SCENE_SCHEMA_VERSION}'

DEFAULT_SHADER_ID: Final

DEFAULT_SHADER_ID = 'lsd-shader/standard'

OBJECTS_DEFAULTS: Final

OBJECTS_DEFAULTS = {'scale': (1.0, 1.0, 1.0), 'shader': DEFAULT_SHADER_ID, 'enabled': True}

OBJECT_ARCHETYPE: Final

OBJECT_ARCHETYPE = 'object'

OBJECT_NAMESPACE: Final

OBJECT_NAMESPACE = 'lsd-runtime/object@1'

Explicit capability for state derived entirely from authored inputs and absolute tick.

The owner installs this as ``pure_seek`` during construction. ``apply`` must reproduce
its complete output after Timeline endpoints have been applied, including backward
jumps, without events, history, root integration or external input. Stateful owners
remain on the existing StateOwner/capture path. This is not reload-factory admission.

apply: Callable[[int], None]

REFUSED_COLUMNS: Final

REFUSED_COLUMNS = {'shape': _SHAPE_REASON, 'shape_kind': _SHAPE_REASON, 'extent': _EXTENT_REASON, 'ex': _EXTENT_REASON, 'e0': _EXTENT_REASON, 'extent_x': _EXTENT_REASON, 'archetype': "a scene holds runtime objects only: cameras and actors are a camera controller's, lights a lighting entity's"}

SCENE_CONTRACT: Final

SCENE_CONTRACT = 'lsd:runtime'

SCENE_FEATURE_KIND: Final

SCENE_FEATURE_KIND = 'runtime_object'

SCENE_LAYER_KIND: Final

SCENE_LAYER_KIND = 'runtime_objects'

SCENE_LAYER_VERSION: Final

SCENE_LAYER_VERSION = 4

SCENE_PORTS: Final

SCENE_PORTS = ('scene', 'objects')

SCENE_SCHEMA_VERSION: Final

SCENE_SCHEMA_VERSION = 4

scene: pa.Table

objects: pa.Table

feature_reference: Mapping[str, Any]

feature_reference = None

from_tables(cls, tables: Mapping[str, pa.Table]) -> 'SceneBundle'

cls: (unannotated)
tables: Mapping[str, pa.Table]

tables(self) -> dict[str, pa.Table]

self: (unannotated)

scene_id(self) -> str

self: (unannotated)

name(self) -> str

self: (unannotated)

up_axis(self) -> str

self: (unannotated)

``"z"`` (LSD's elevation axis) or ``"y"``: the axis gravity, jumps and capsules follow.

rows(self) -> int

self: (unannotated)

rows_by_eid(self) -> dict[int, dict[str, Any]]

self: (unannotated)

row_of_eid(self) -> dict[int, int]

self: (unannotated)

Object eid -> object row (the frame row every consumer joins on).

eid_of(self, oid: str) -> Optional[int]

oid: str
self: (unannotated)

describe(self) -> dict[str, Any]

self: (unannotated)

A scene bundle is internally inconsistent (unknown eids, duplicate seats, no camera).

code = 'scene_error'

SceneRuntime(self, bundle: Any, *, driver: 'RuntimeDriver') -> None

The loaded scene as the simulation thread holds it: the engine world and its columns.

Every runtime object is an entity of one ``lsd_ecs.World`` whose index is the object's row;
``position`` / ``orientation`` are zero-copy numpy views over the world's ``position`` and
``rotation`` pages (float64, one row per object of the bundle's objects table); ``enabled`` and
the immutable ``shader`` resource ids are what a published state carries beside them. The
participants' systems run as the world's systems (``bind_systems``): the timeline and the
``before_simulation`` phases in the engine's ``before`` phase, every ``step`` in ``step``, the
``after_simulation`` phases in ``after``, one ``world.step`` per fixed step. ``claimed_by(row)``
names the simulation system that moves a row, so no other system writes it. Systems also reach
the driver's input through ``sample_seats``, ask for a reset through ``request_reset`` and run
latest-value work on the simulation thread through ``schedule``.

__init__(self, bundle: Any, *, driver: 'RuntimeDriver') -> None

bundle: Any
driver: 'RuntimeDriver'
self: (unannotated)

close(self) -> None

self: (unannotated)

Release the engine world and every attached one (their column views become invalid).

bind_systems(self, systems: Sequence[Any], *, timeline: Any) -> None

self: (unannotated)
systems: Sequence[Any]
timeline: Any

Bind the participants to the world as its systems, in the driver's phase order.

The timeline advances first in the engine's ``before`` phase, then every participant's
``before_simulation``; every participant's ``step`` runs in ``step`` (the timeline has no
step of its own); every ``after_simulation`` runs in ``after``; then every ``bind_last``
defines the systems that run last in their phase. The participants come in rank order.
One ``world.step`` per fixed step then runs them all in that order, timing each on the
engine's clock.

step_world(self, dt: float, substeps: int, time_us: int) -> int

dt: float
self: (unannotated)
substeps: int
time_us: int

One fixed step of the world; returns the engine's before-phase nanoseconds (the input lane).

begin_wake(self) -> None

self: (unannotated)

Remember the engine's counters so the wake reports its own deltas.

wake_report(self) -> dict[str, Any]

self: (unannotated)

The engine's numbers for the wake since ``begin_wake``: per-system simulation-thread
time, structural counters as deltas, memory gauges.

attach_world(self, name: str, world: Any) -> None

name: str
self: (unannotated)
world: Any

Hold an engine world of a participant's own beside the scene's, during construction.

The world's systems become profiler rows of this scene under ``<name>.<system>``,
nested inside the interval of the scene system that steps the world (so they are listed
beside the scene's rows and never added to the wake's ``ecs.step_ns``); its gauges
(entities, bytes) stay its own and are described under ``attached``. The participant
steps the world itself, from one of the scene world's systems. The world is closed with
the scene.

attached_worlds(self) -> Mapping[str, Any]

self: (unannotated)

snapshot_world(self) -> None

self: (unannotated)

Read the world's counters for ``describe_world`` (one native call; simulation thread).

The structure behind the description (components, queries, the systems' identities) is
rebuilt only when the layout epoch or a count changed, so a publication pays one counters
read, not a walk of every component page and system profile.

describe_world(self) -> dict[str, Any]

self: (unannotated)

The engine world as data (``describe_runtime``'s ``ecs`` block), from the last
``snapshot_world`` and the last wake report; no native call, so any thread may ask.

A system row carries the last wake's ``step_ns``/``calls``/``entities`` when the profiler
recorded that wake (the capture owns the engine's per-wake numbers); the engine's
cumulative profile is never published here.

apply_windows(self, time_us: int) -> None

self: (unannotated)
time_us: int

Gate every member by its play window at ``time_us`` (microseconds since Play).

refresh_alive(self) -> np.ndarray

self: (unannotated)

``enabled`` after the member gate, in the persistent array the boards publish.

register_publication_field(self, name: str, values: np.ndarray) -> None

name: str
self: (unannotated)
values: np.ndarray

A system declares fixed-row typed state during construction, before load completes.

The system keeps the same array and writes it only on the simulation thread.
A new shape/schema needs a new loaded world; render frames remain independent.

sample_seats(self, seats: Sequence[SeatActions], tick: int) -> list[Optional[Sample]]

seats: Sequence[SeatActions]
self: (unannotated)
tick: int

One sample per seat from the driver's reader for this fixed step, recorded in the trace.

The driver keeps one sampler per loaded world: its edge state (the sequence each seat
last saw) is part of a capture, and it is reset whenever the reader changes hands.

last_samples(self) -> tuple[int, dict[int, Optional[Sample]]]

self: (unannotated)

The tick the driver last sampled seats at and one sample per sampled seat: what a
system that runs after the actors reads instead of sampling again (edges count once).

sampled_frames(self) -> list[Optional['SeatFrame']]

self: (unannotated)

The live frames the last ``sample_seats`` read, one per seat in that call's order: what
a camera rig reads its own actions from. None for an unbound seat or a replayed trace.

request_reset(self) -> None

self: (unannotated)

Restart the world from Play after the current fixed step (an actor's reset action).

schedule(self, key: Any, callback: Callable[[], bool]) -> bool

callback: Callable[[], bool]
key: Any
self: (unannotated)

Run ``callback`` on the simulation thread at the next wake; a later call with the same
``key`` replaces a pending one (latest value wins, so pointer-rate input never queues).
The callback returns whether it changed what a frame shows; the loop publishes once when
any did and no fixed step published in the same wake. Refused (False) while a seek owns
the world, once the world failed, or after this world was replaced.

register_capability(self, name: str, value: Any) -> None

name: str
self: (unannotated)
value: Any

A system offers one named capability to its peers, during construction only.

row_of(self, entity_id: str, oid: str) -> Optional[int]

entity_id: str
oid: str
self: (unannotated)

The pose row of one member's object, or None when the composition does not play it.

member_active(self, entity_id: str) -> bool

entity_id: str
self: (unannotated)

Whether the member is inside its play window now (its rows are published).

register_presentation(self, name: str, state) -> None

name: str
self: (unannotated)
state: (unannotated)

Register independent numeric presentation rows before this world is published.

A schema or source change needs a new loaded world. A presentation does not create
scene objects, claim transforms, or add a simulation clock.

rows(self) -> int

self: (unannotated)

claim(self, system: Any, rows: Any) -> None

rows: Any
self: (unannotated)
system: Any

reclaim(self, system: Any, rows: Any) -> None

rows: Any
self: (unannotated)
system: Any

Make ``rows`` exactly what this system claims, releasing the ones it gave up.

Staging a timeline document can drop a property, and the row it moved must go back to
whoever else may write it rather than staying claimed by a lane that no longer animates it.

claimed_by(self, row: int) -> Any

row: int
self: (unannotated)

claimed_rows(self) -> list[int]

self: (unannotated)

emit(self, event_type: str, **data: Any) -> None

data: Any
event_type: str
self: (unannotated)

Publish on the driver's bus with the driver's identity (a system's own events).

bus(self) -> Any

self: (unannotated)

registry(self) -> Any

self: (unannotated)

driver_id(self) -> str

self: (unannotated)

composition_id(self) -> str

self: (unannotated)

tick(self) -> int

self: (unannotated)

time_us(self) -> int

self: (unannotated)

Play time now: microseconds since Play, on the driver's fixed-step clock.

time_us_at(self, step: int) -> int

self: (unannotated)
step: int

The moment of play time fixed step ``step`` reaches on the current clock.

sim_time_ns(self) -> int

self: (unannotated)

config(self) -> RuntimeConfig

self: (unannotated)

cancel(self) -> Any

self: (unannotated)

restore_authored(self) -> None

self: (unannotated)

pose_hash(self) -> int

self: (unannotated)

bundle: Any

bundle = bundle

driver: 'RuntimeDriver'

driver = driver

scene_id: (unannotated)

scene_id = str(bundle.scene_id)

up_axis: (unannotated)

up_axis = 1 if str(bundle.up_axis) == 'y' else 2

world: (unannotated)

world = World(max_components=64, entities=max(count, 1), single_page=True, profile=True)

entities: (unannotated)

entities = self.world.create(count)

eid: (unannotated)

eid = self._column('eid')

oid: (unannotated)

oid = list(objects.column('oid').to_pylist())

row_of_eid: (unannotated)

row_of_eid = {int(value): row for row, value in enumerate(self.eid.tolist())}

position: (unannotated)

position = self._column('position')

orientation: (unannotated)

orientation = self._column('rotation')

shader: (unannotated)

shader = tuple((str(shader) for shader in objects.column('shader').to_pylist()))

enabled: (unannotated)

enabled = self._column('enabled')

entity_of_row: tuple[str, ...]

entity_of_row = tuple((str(value) for value in objects.column('entity_id').to_pylist())) if 'entity_id' in objects.column_names else (str(driver.composition_id),) * count

member_gate: (unannotated)

member_gate = self._column('member_gate')

alive: (unannotated)

alive = self._column('alive')

property_leases: dict[tuple[int, str], Any]

property_leases = {}

presentations: (unannotated)

presentations = MappingProxyType({})

capabilities: dict[str, Any]

capabilities = {}

timeline: Any

timeline = None

A registered system that moves some objects of the scene on every fixed step.

name: str

claims(self) -> Sequence[int]

self: (unannotated)

The object rows this system moves; no other system may write them.

A system that lets an actor drive one of its rows offers ``<name>.drive`` through
``SceneRuntime.register_capability`` (``drive(row, sample, *, axes, force_scale,
jump_impulse, invert_second_axis, up_axis)``), which the camera controller calls for an
actor bound to that row's body; the protocol has no drive method.

step(self, dt: float, substeps: int) -> None

dt: float
self: (unannotated)
substeps: int

Advance one fixed step and write the claimed rows' poses into the scene arrays.

state_hash(self) -> int

self: (unannotated)

publish(self, *, final: bool) -> None

final: bool
self: (unannotated)

Publish whatever the system publishes (a Trip on the bus, for physics); may be a no-op.

profile(self) -> Any

self: (unannotated)

counters(self) -> Any

self: (unannotated)

describe(self) -> Any

self: (unannotated)

close(self) -> None

self: (unannotated)

Owner validation and commit on the existing Driver command lane.

``prepare(operation, detached_payload)`` must validate without mutating the
world and return a zero-argument commit callback. Admission failures preserve
the world; a failed commit invalidates continuation like a failed fixed step.
The host's action/input authority remains responsible for admitting callers.

prepare: Callable[[str, Any], Callable[[], Any]]

describe: Callable[[], dict] | None

describe = None

Optional work around the Driver's existing simulation stage.

``prepare`` binds peer capabilities after every system has claimed its rows,
before Timeline binds properties. All before callbacks run before any ``step``
(input is sampled and actors are driven here); all after callbacks run after
every ordinary ``step`` (cameras are evaluated here). Callbacks are prepared
once per world and use the same thread, fixed dt and publication transaction.
``reconstruct`` rebuilds derived output from restored state at a completed tick;
it does not qualify the owner for an absolute pure seek. ``bind_native`` binds the
owner's native lane to the scene's engine world (``scene.world``) right after its Python
``step`` was bound, so the native system runs after that step's pre-work in the same phase;
an owner with a native lane keeps its Python ``step`` for the work before the native step.
``bind_last`` is called with the scene after every participant's phases and native lanes are
bound, in rank order: an owner that defines its own engine systems there
(``scene.world.system``) runs them last in each phase, after every other participant's work
of that phase, and the driver adopts every new engine system as a profiler row of its own.

prepare: Callable[[], None] | None

prepare = None

before_simulation: Callable[[float, int], None] | None

before_simulation = None

after_simulation: Callable[[float, int], None] | None

after_simulation = None

reconstruct: Callable[[int], None] | None

reconstruct = None

bind_native: Callable[[Any], None] | None

bind_native = None

bind_last: Callable[[Any], None] | None

bind_last = None

UP_AXES: Final

UP_AXES = ('y', 'z')

build_scene_tables(scene_id: str, objects_in: pa.Table, *, up_axis: str='z', name: str='') -> dict[str, pa.Table]

name: str
objects_in: pa.Table
scene_id: str
up_axis: str

The stamped ``scene`` and ``objects`` tables of one scene from its authored objects.

declares_runtime(entity: Any) -> bool

entity: Any

Whether the entity plays a runtime role at all.

declares_scene(entity: Any) -> bool

entity: Any

Whether the entity declares exactly one coordinate for every scene port.

declaring_step(entity: Any, port: str) -> Any

entity: Any
port: str

The step of ``entity`` that declares output ``port`` exactly once, found by its coordinate.

Editors open a member's authored document through the step that declares the participant's
port (a UI entity's ``ui``, an animation entity's ``animation``); they never know the step's
class, which belongs to a package.

member_kind(entity: Any) -> Optional[str]

entity: Any

The one runtime role the entity plays, or None.

Raises ``ValueError`` when the entity declares the outputs of several participants: an
entity owns the objects of one role; what plays beside it is another entity of the same
composition.

mesh_resources_fingerprint(objects, *, base_dir=None)

base_dir: (unannotated)
objects: (unannotated)

Content fingerprint for a delivery owner's cache, including paired faces bytes.

mint_oid(name: Optional[Any], index: int) -> str

index: int
name: Optional[Any]

The authored key when the table gave none: the name when present, else ``object_<row>``.

object_eid(scene_id: str, oid: str) -> int

oid: str
scene_id: str

The identity of runtime object ``oid`` of scene ``scene_id``.

object_feature_reference() -> dict[str, Any]

The detached feature-reference descriptor for runtime objects.

register_participant(kind: str, ports: Mapping[str, str], *, factory: Optional[Factory]=None, description: str='', rank: int=DEFAULT_RANK) -> None

description: str
factory: Optional[Factory]
kind: str
ports: Mapping[str, str]
rank: int

Register (or re-register) the participant called ``kind``.

``ports`` are the pipeline outputs a member entity of this kind declares, ``{port: kind}`` in
declaration order (a ``kind`` is what the host reads from the port: a ``table``, a ``path``
or a ``trip``). A ``factory`` makes the participant a simulation system; a system with no
ports has no member entity and is offered every composition (a probe, a host-owned system).
``rank`` orders the systems (lower binds first; registration order within a rank).

register_system(name: str, factory: Factory, *, port: Optional[str]=None, port_kind: str='table', ports: Optional[Mapping[str, str]]=None, description: str='', rank: int=DEFAULT_RANK) -> None

description: str
factory: Factory
name: str
port: Optional[str]
port_kind: str
ports: Optional[Mapping[str, str]]
rank: int

Register the simulation system called ``name``: a participant with a factory.

Declare the ports its members' tables travel on as ``ports`` (``{port: kind}``, ordered:
the first names the role) or, for one port, as the ``port`` / ``port_kind`` shorthand.
``rank`` is where its systems bind among the others (physics 100, cameras 200, animation 300,
kernels 400, audio 800, calibration 900; ``DEFAULT_RANK`` otherwise).

scene_id_for(entity_id: Optional[str], explicit: Optional[str]=None) -> str

entity_id: Optional[str]
explicit: Optional[str]

The scene id: the explicit parameter when set, else derived from the owning entity id.

scene_layer_payload(authored: Any, objects: pa.Table, scene: Optional[pa.Table]=None, *, base_dir=None) -> LayerPayload

authored: Any
base_dir: (unannotated)
objects: pa.Table
scene: Optional[pa.Table]

The ``runtime_objects`` viewer layer: the authored table (the edit target, passed as the
loader's own artifact so an edit rewrites what the scene is built from) beside the built objects.

stable_eid(owner: str, archetype: str, oid: str) -> int

archetype: str
oid: str
owner: str

A deterministic 63-bit id for one row; never zero, never negative, never colliding in practice.

unregister_system(kind: str) -> None

kind: str

up_axis_defaults(up_axis: str) -> dict

up_axis: str

The ``up`` vector a scene with this up axis defaults to.

up_axis_of(scene: Optional[pa.Table], default: str='z') -> str

default: str
scene: Optional[pa.Table]

The up axis a built scene table declares (``default`` without one).