simvx.graphics.frame_loop

FrameLoop: the single per-frame engine drive seam (clock + sink + step).

Historically App._run_with_tree (visible), App.run_headless, App.run_streaming, and the editor’s PlayMode each hand-rolled the same physics -> interpolate -> tick -> collect 2D items -> submit_scene -> present/capture core plus an identical pipelined frame_driver. FrameLoop owns that core once; the per-loop differences are expressed as three pluggable pieces:

• Clock: how wall time advances and the physics-stepping strategy (RealTimeClock accumulator vs FixedStepClock single-step).

• Sink: where the rendered frame goes and how capture is wired (SwapchainSink present, OffscreenSink readback, StreamSink push).

• Integration: input source + windowed extras (callbacks, gamepad, resize, picking) – supplied per loop.

The LLM agent live-session and the editor viewport are just FixedStepClock (externally driven) + an offscreen or swapchain sink. This module is migrated one consumer at a time; run_headless is the first to land on it.

Module Contents

Classes

FrameClock	Decides per-frame dt and how physics/logic advance.
FixedStepClock	Fixed timestep: one physics step per frame, interpolate at the step boundary.
RealTimeClock	Wall-clock dt with a fixed-timestep physics accumulator (windowed / streaming).
FrameSink	Where the frame goes; supplies the render-thread capture + sync readback.
SwapchainSink	Present to the window; no readback. Windowed games.
OffscreenSink	Read frames back into captured (headless tests, agent capture).
StreamSink	Push each drawn frame to connected browser clients over WebSocket.
FrameIntegration	Per-loop input source and windowed extras.
HeadlessIntegration	Batch headless: InputSimulator drives Input; capture + telemetry at the end.
StreamingIntegration	Browser streaming: drain queued client input, push readback to clients.
WindowedIntegration	Visible game window: platform input callbacks, gamepad, resize, picking.
FrameService	A unit of work drained once per frame on the loop (main) thread.
FrameServices	Ordered registry of :class:FrameService objects drained each frame.
FrameLoop	One per-frame engine drive seam, parameterised by clock + sink + integration.

Functions

active_loops

Snapshot of currently-running :class:FrameLoop instances (most recent last).

Data

log

API

simvx.graphics.frame_loop.log

‘getLogger(…)’

simvx.graphics.frame_loop.active_loops() → list[FrameLoop]

Snapshot of currently-running :class:FrameLoop instances (most recent last).

class simvx.graphics.frame_loop.FrameClock

Decides per-frame dt and how physics/logic advance.

on_begin(app: simvx.graphics.app.App) → None

Called once after setup, before the first frame.

should_stop(frame_idx: int) → bool

abstractmethod advance(app: simvx.graphics.app.App) → float

abstractmethod step_physics(app: simvx.graphics.app.App, tree: Any, frame_dt: float) → None

tick_logic(app: simvx.graphics.app.App, tree: Any, frame_dt: float) → None

class simvx.graphics.frame_loop.FixedStepClock(*, frames: int | None = None, dt: float | None = None)

Bases: simvx.graphics.frame_loop.FrameClock

Fixed timestep: one physics step per frame, interpolate at the step boundary.

With frames set this is the headless batch clock; with frames=None and an explicit dt it is the externally-driven clock the agent / editor use (the caller advances frames, no internal cap).

Initialization

should_stop(frame_idx: int) → bool

advance(app: simvx.graphics.app.App) → float

step_physics(app: simvx.graphics.app.App, tree: Any, frame_dt: float) → None

tick_logic(app: simvx.graphics.app.App, tree: Any, frame_dt: float) → None

on_begin(app: simvx.graphics.app.App) → None

class simvx.graphics.frame_loop.RealTimeClock

Bases: simvx.graphics.frame_loop.FrameClock

Wall-clock dt with a fixed-timestep physics accumulator (windowed / streaming).

Initialization

on_begin(app: simvx.graphics.app.App) → None

advance(app: simvx.graphics.app.App) → float

step_physics(app: simvx.graphics.app.App, tree: Any, frame_dt: float) → None

should_stop(frame_idx: int) → bool

tick_logic(app: simvx.graphics.app.App, tree: Any, frame_dt: float) → None

class simvx.graphics.frame_loop.FrameSink

Where the frame goes; supplies the render-thread capture + sync readback.

ring_capacity: int

2

render_thread_capture() → collections.abc.Callable[[int, Any], None] | None

Callback the pipelined RenderThread hands a captured frame to (or None).

wants_capture(frame_idx: int) → bool

Whether to arm a before-present readback of this frame.

The one decision, asked by both paths at the same point in the frame: the synchronous loop before _draw_frame and the render thread before the packet’s _draw_frame. A sink that answers False pays nothing – no copy is recorded and no device drain happens – which is what makes capture_frames=() mean what it says on both.

capture_synchronous(loop: simvx.graphics.frame_loop.FrameLoop, frame_idx: int) → None

Synchronous-mode readback of the just-drawn frame (no-op by default).

class simvx.graphics.frame_loop.SwapchainSink

Bases: simvx.graphics.frame_loop.FrameSink

Present to the window; no readback. Windowed games.

ring_capacity

2

render_thread_capture() → collections.abc.Callable[[int, Any], None] | None

wants_capture(frame_idx: int) → bool

capture_synchronous(loop: simvx.graphics.frame_loop.FrameLoop, frame_idx: int) → None

class simvx.graphics.frame_loop.OffscreenSink(*, capture_frames: collections.abc.Sequence[int] | None = (), capture_fn: collections.abc.Callable[[int], bool] | None = None)

Bases: simvx.graphics.frame_loop.FrameSink

Read frames back into captured (headless tests, agent capture).

Initialization

ring_capacity

1

render_thread_capture() → collections.abc.Callable[[int, Any], None]

wants_capture(frame_idx: int) → bool

capture_synchronous(loop: simvx.graphics.frame_loop.FrameLoop, frame_idx: int) → None

class simvx.graphics.frame_loop.StreamSink(server: Any)

Bases: simvx.graphics.frame_loop.FrameSink

Push each drawn frame to connected browser clients over WebSocket.

Initialization

ring_capacity

2

render_thread_capture() → collections.abc.Callable[[int, Any], None]

wants_capture(frame_idx: int) → bool

Read frames back only while somebody is watching.

Without this the pipelined path streams to nobody: it would read every frame back and drop the pixels, which costs a full-frame copy and a device drain per frame for a session with no clients connected.

capture_synchronous(loop: simvx.graphics.frame_loop.FrameLoop, frame_idx: int) → None

class simvx.graphics.frame_loop.FrameIntegration

Per-loop input source and windowed extras.

reset_globals: bool

False

boot_splash: bool

False

on_run_begin(loop: simvx.graphics.frame_loop.FrameLoop) → None

After the engine + tree exist, before the loop (e.g. start a stream server).

on_run_end(loop: simvx.graphics.frame_loop.FrameLoop) → None

In the run() finally, before audio shutdown (e.g. stop a stream server).

on_setup(loop: simvx.graphics.frame_loop.FrameLoop) → None

Extra setup after the adapter exists (input callbacks, clipboard).

pre_tick(loop: simvx.graphics.frame_loop.FrameLoop) → bool

Input drain / window sync before the tick. Return False to stop the loop.

mid_tick(loop: simvx.graphics.frame_loop.FrameLoop) → None

Between physics and logic tick (theme sync on the windowed path).

post_tick(loop: simvx.graphics.frame_loop.FrameLoop) → None

After the 2D draw collection (mouse picking on the windowed and headless paths).

per_frame_telemetry(loop: simvx.graphics.frame_loop.FrameLoop) → None

Live per-frame telemetry (the windowed and streaming paths).

on_loop_end(loop: simvx.graphics.frame_loop.FrameLoop) → None

At the stop boundary (headless captures final telemetry here).

pre_render(loop: simvx.graphics.frame_loop.FrameLoop, cmd: Any) → None

Offscreen passes before the main render pass (SubViewports, probes).

class simvx.graphics.frame_loop.HeadlessIntegration(on_frame: collections.abc.Callable[[int, float], bool | None] | None = None)

Bases: simvx.graphics.frame_loop.FrameIntegration

Batch headless: InputSimulator drives Input; capture + telemetry at the end.

Initialization

reset_globals

True

pre_tick(loop: simvx.graphics.frame_loop.FrameLoop) → bool

post_tick(loop: simvx.graphics.frame_loop.FrameLoop) → None

on_loop_end(loop: simvx.graphics.frame_loop.FrameLoop) → None

boot_splash: bool

False

on_run_begin(loop: simvx.graphics.frame_loop.FrameLoop) → None

on_run_end(loop: simvx.graphics.frame_loop.FrameLoop) → None

on_setup(loop: simvx.graphics.frame_loop.FrameLoop) → None

mid_tick(loop: simvx.graphics.frame_loop.FrameLoop) → None

per_frame_telemetry(loop: simvx.graphics.frame_loop.FrameLoop) → None

pre_render(loop: simvx.graphics.frame_loop.FrameLoop, cmd: Any) → None

class simvx.graphics.frame_loop.StreamingIntegration(server: Any)

Bases: simvx.graphics.frame_loop.FrameIntegration

Browser streaming: drain queued client input, push readback to clients.

Initialization

on_run_begin(loop: simvx.graphics.frame_loop.FrameLoop) → None

on_run_end(loop: simvx.graphics.frame_loop.FrameLoop) → None

pre_tick(loop: simvx.graphics.frame_loop.FrameLoop) → bool

pre_render(loop: simvx.graphics.frame_loop.FrameLoop, cmd: Any) → None

per_frame_telemetry(loop: simvx.graphics.frame_loop.FrameLoop) → None

Publish App.last_telemetry every frame, as the windowed path does.

A streaming session is a game running on someone else’s screen, so the counters a debug overlay reads have to be there, and pipelined mode is not an exception: the publish is a dict build and a rebind, and takes no lock the render thread wants.

reset_globals: bool

False

boot_splash: bool

False

on_setup(loop: simvx.graphics.frame_loop.FrameLoop) → None

mid_tick(loop: simvx.graphics.frame_loop.FrameLoop) → None

post_tick(loop: simvx.graphics.frame_loop.FrameLoop) → None

on_loop_end(loop: simvx.graphics.frame_loop.FrameLoop) → None

class simvx.graphics.frame_loop.WindowedIntegration

Bases: simvx.graphics.frame_loop.FrameIntegration

Visible game window: platform input callbacks, gamepad, resize, picking.

Gamepad poll / window-resize / theme sync run in mid_tick (after physics, before the logic tick) to match the original _run_with_tree ordering.

Initialization

boot_splash

True

on_run_end(loop: simvx.graphics.frame_loop.FrameLoop) → None

on_setup(loop: simvx.graphics.frame_loop.FrameLoop) → None

mid_tick(loop: simvx.graphics.frame_loop.FrameLoop) → None

post_tick(loop: simvx.graphics.frame_loop.FrameLoop) → None

per_frame_telemetry(loop: simvx.graphics.frame_loop.FrameLoop) → None

reset_globals: bool

False

on_run_begin(loop: simvx.graphics.frame_loop.FrameLoop) → None

pre_tick(loop: simvx.graphics.frame_loop.FrameLoop) → bool

on_loop_end(loop: simvx.graphics.frame_loop.FrameLoop) → None

pre_render(loop: simvx.graphics.frame_loop.FrameLoop, cmd: Any) → None

class simvx.graphics.frame_loop.FrameService

A unit of work drained once per frame on the loop (main) thread.

The contract is the marshalling pattern external drivers (PEP 768 socket bridge, editor tooling) need: a producer thread enqueues work, and this runs on the one thread that owns the SceneTree, AT the post-tick barrier (tree fully ticked, GPU handoff not yet started). Implement

Meth:

on_frame_service; keep it bounded and non-blocking.

on_frame_service(loop: simvx.graphics.frame_loop.FrameLoop) → None

Drain queued work on the loop thread. Must not block the frame.

on_loop_end(loop: simvx.graphics.frame_loop.FrameLoop) → None

Release resources when the owning loop tears down (close sockets, join threads).

class simvx.graphics.frame_loop.FrameServices

Ordered registry of :class:FrameService objects drained each frame.

Loop-level (not integration-level) so every clock/sink/integration combo – windowed, headless, streaming, and the externally-driven agent/editor path – reaches the same barrier through FrameLoop._update with zero per-consumer wiring. Registration is the canonical way to attach a per-frame drain to a running loop; it is dependency-free (pure stdlib list).

Initialization

add(service: simvx.graphics.frame_loop.FrameService) → simvx.graphics.frame_loop.FrameService

Register a service. Returns it for svc = loop.services.add(MyService()).

remove(service: simvx.graphics.frame_loop.FrameService) → None

Deregister a service (idempotent).

__bool__() → bool

run(loop: simvx.graphics.frame_loop.FrameLoop) → None

Drain every registered service on the loop thread, in registration order.

A failing service is fully logged and the remaining services still run; the loop must never be left half-drained or silently killed by one bad service.

close(loop: simvx.graphics.frame_loop.FrameLoop) → None

Tell every service the loop is ending, then clear the registry.

Called from FrameLoop._teardown so a registered bridge (its accept thread + bound socket) is shut down with the loop instead of leaking until process exit.

class simvx.graphics.frame_loop.FrameLoop(app: simvx.graphics.app.App, root_node: Any, *, clock: simvx.graphics.frame_loop.FrameClock, sink: simvx.graphics.frame_loop.FrameSink, integration: simvx.graphics.frame_loop.FrameIntegration, visible: bool, vsync: bool)

One per-frame engine drive seam, parameterised by clock + sink + integration.

Initialization

engine: simvx.graphics.engine.Engine

None

run() → None

Build the engine and drive frames (blocking) until the loop stops.

begin() → None

Externally-driven setup: build the engine, create the window + device, run setup. Pair with :meth:step_frame / :meth:capture / :meth:end. For an External (FixedStepClock) clock – the agent live-session and editor viewport.

step_frame(dt: float | None = None) → bool

Advance exactly one frame; returns False when the loop should stop.

capture(*, scale: float = 1.0, region: tuple[int, int, int, int] | None = None) → Any

Read back the last drawn frame as an (H, W, 4) uint8 RGBA array.

Optional region=[x1,y1,x2,y2] crops and scale (<1.0) nearest-downsamples at the source – the cheap detail/cost levers from the design.

end() → None

Externally-driven teardown (counterpart to :meth:begin).