"""Graphics application layer: engine wrapper with node tree support."""
import logging
import weakref
from collections.abc import Callable
from typing import Any
from .engine import Engine
from .platform import AVAILABLE_BACKENDS
log = logging.getLogger(__name__)
__all__ = ["App", "AVAILABLE_BACKENDS"]
def _get_sdl3_type() -> type:
"""Return the Sdl3Backend class, or a dummy type if not importable."""
try:
from .platform._sdl3 import Sdl3Backend
return Sdl3Backend
except ImportError, ModuleNotFoundError:
return type(None) # will never match isinstance()
[docs]
class App:
"""Graphics application wrapper. Supports both raw callbacks and node tree scenes.
Args:
title: Window title string.
width: Window width in pixels (must be >= 1).
height: Window height in pixels (must be >= 1).
backend: Windowing backend name. One of ``"glfw"`` (desktop),
``"sdl3"`` (touch/mobile), ``"qt"`` (embedding). ``None`` auto-detects.
physics_fps: Fixed-timestep physics simulation rate (Hz). Independent
of the display rate; the run loop drives physics via an
accumulator so changing this never affects render cadence.
target_fps: Maximum display frame rate. ``None`` (default) means
uncapped (vsync-gated only). Mirrors Godot's ``Engine.max_fps``,
Unity's ``Application.targetFrameRate``, and Bevy's
``FrameRateLimit::Limit``. Combine with ``vsync=False`` to make
the cap dominant on high-refresh displays.
visible: Open a visible window. Set ``False`` for headless tests.
vsync: Enable vertical sync at present time.
bg_colour: Background clear colour. ``"transparent"``, an RGBA tuple,
or ``None`` to use the theme background.
render_thread: Per-process override forcing the pipelined render thread
on. ``False`` (default) uses the synchronous render path unless a
``WorldEnvironment`` in the scene authors ``render_mode='pipelined'``.
``True`` forces pipelining regardless of the authored value.
``WorldEnvironment.render_mode`` is the authored home for this
choice; this flag is a runtime override only.
Usage with node tree (recommended)::
app = App(title="My Game", width=1280, height=720)
app.run(MyGameScene()) # Node subclass with ready()/process()
Usage with raw callbacks::
app = App(title="My Game")
app.run(update=my_update, render=my_render)
After ``run()`` has created the window, ``window_title``, ``window_size``,
and ``active_backend`` reflect the live state.
"""
# Weakref to the most recently constructed App, exposed via App.current().
# Lets non-Node helpers (audio code, utilities) reach the running app
# without a circular dependency. Held as a weakref so test suites that
# spin up many short-lived Apps don't leak them.
_current_ref: weakref.ReferenceType[App] | None = None
[docs]
@classmethod
def current(cls) -> App | None:
"""Return the running App instance, or ``None`` if no App is alive.
Prefer ``node.app`` from inside Node subclasses (after
``on_enter_tree``); use this from non-Node helpers like audio
callbacks, utility modules, or tools that don't have a node handle.
"""
ref = cls._current_ref
return ref() if ref is not None else None
def __init__(
self,
title: str = "SimVX",
width: int = 1280,
height: int = 720,
backend: str | None = None,
physics_fps: int = 60,
target_fps: int | None = None,
visible: bool = True,
vsync: bool = True,
bg_colour: tuple[float, float, float, float] | str | None = None,
audio_sample_rate: int = 48000,
audio_channels: int = 2,
render_thread: bool = False,
multi_gpu: bool = False,
splash: Any = None,
splash_min_time: float | None = None,
unbranded: bool = False,
**_kwargs,
):
if not isinstance(title, str):
raise TypeError(f"App title must be a string, got {type(title).__name__}")
if not isinstance(width, int) or not isinstance(height, int):
raise TypeError("App width and height must be integers")
if width < 1 or height < 1:
raise ValueError(f"App width and height must be >= 1, got {width}x{height}")
if target_fps is not None and target_fps < 1:
raise ValueError(f"target_fps must be >= 1 or None, got {target_fps}")
if backend is not None and backend not in AVAILABLE_BACKENDS:
raise ValueError(f"Unknown backend {backend!r}. Allowed: {list(AVAILABLE_BACKENDS)}")
if not isinstance(audio_sample_rate, int) or audio_sample_rate <= 0:
raise ValueError(f"audio_sample_rate must be a positive int, got {audio_sample_rate!r}")
if audio_channels not in (1, 2):
raise ValueError(f"audio_channels must be 1 (mono) or 2 (stereo), got {audio_channels!r}")
if splash not in (None, True, False):
from simvx.core.ui.splash import SplashScreen
if not isinstance(splash, SplashScreen):
raise TypeError(f"App splash must be None, a bool, or a SplashScreen, got {type(splash).__name__}")
if splash_min_time is not None and splash_min_time < 0:
raise ValueError(f"splash_min_time must be >= 0 or None, got {splash_min_time!r}")
if _kwargs:
raise TypeError(
f"App() got unexpected keyword argument(s): {sorted(_kwargs)}. "
f"Valid: title, width, height, backend, physics_fps, target_fps, visible, "
f"vsync, bg_colour, audio_sample_rate, audio_channels, render_thread, multi_gpu, "
f"splash, splash_min_time, unbranded."
)
self.title = title
self.width = width
self.height = height
self._backend_name = backend
self._physics_fps = physics_fps
self._target_fps = target_fps
self._visible = visible
self._vsync = vsync
self._bg_colour = bg_colour
self._audio_sample_rate = audio_sample_rate
self._audio_channels = audio_channels
# Per-process override forcing the pipelined render thread on (or off).
# ``WorldEnvironment.render_mode`` is the authored home for this choice
# (``'default'`` vs ``'pipelined'``); passing ``render_thread=True`` here
# overrides any authored value for this run. Default OFF keeps the
# synchronous, byte-identical render path. The frame loop reads this in
# a later wave; for now it is plumbed and stored only.
self._render_thread = render_thread
# Explicit-multi-adapter opt-in. OFF by default: a single-GPU box and
# any multi-GPU box that does not opt in run the single-device path
# byte-identical (no secondary VkDevice, no cross-device transfer). When
# ON *and* the engine reports > 1 physical device, init creates an
# independent logical device per GPU and the offload renderer activates.
# The active multi-device path is verified on the 4x Arc Pro B70 rig.
self._multi_gpu = multi_gpu
# Boot splash configuration. ``None``
# resolves to the branded default face iff ``branding.show_splash()`` is
# true and this run is windowed+visible; ``True`` forces it on (any
# integration); ``False`` suppresses the splash (the persistent
# attribution watermark then applies on branded builds); a SplashScreen
# instance is a game-authored face. ``splash_min_time`` (seconds)
# overrides the splash's own ``min_display_time`` when not None.
# ``unbranded`` is one half of the two-step branding gate
# (``branding.unbranded_allowed``); alone it has no effect.
self._splash = splash
self._splash_min_time = splash_min_time
self._unbranded = unbranded
# One-shot guard so the pipelined-mode unpacketised-content warning logs
# at most once per run, not every frame.
self._warned_pipelined_unpacketised = False
self._engine: Engine | None = None
# Per-run SubViewport offscreen render-target manager (graphics-side
# driver for core.SubViewport render-to-texture). Created in setup().
self._sub_viewports: Any = None
# Per-run RenderView offscreen render-target manager (graphics-side
# driver for core.RenderView main-scene-to-texture, D6). Created in setup().
self._render_views: Any = None
# Global slow-mo / hitstop / fast-forward multiplier applied to the
# per-frame ``dt`` *before* it reaches ``tree.tick()`` and
# ``tree.physics_tick()``. ``1.0`` = real-time; ``0.3`` = 30%
# speed; ``0.0`` = paused (tick/physics still run, but with zero
# dt). Audio pitching and coroutine pacing are intentionally NOT
# affected by this knob: only the scene-tree dt is scaled.
self._time_scale: float = 1.0
# Snapshot of per-frame telemetry from the most recent
# ``run_headless`` call. Captured BEFORE the engine shuts down so
# tests / harnesses can read renderer counters after the call
# returns (the engine nulls its ``_renderer`` in shutdown).
# Keys populated today:
# ``draw2d_draw_count`` : 2D bindless-item draw count of the final
# fully rendered frame.
# ``gpu_phase_times`` : per-pass GPU timings (label → ms),
# copied from ``Engine.gpu_phase_times``.
# ``frames_rendered`` : total update() calls that produced a frame.
# ``occlusion_drawn`` : instances that survived the Hi-Z cull and
# were drawn this frame. Present only while
# occlusion culling is active.
# ``occlusion_total`` : frustum-visible instances submitted to the
# cull (pre-cull). ``occlusion_total -
# occlusion_drawn`` is the number culled.
# Present only while occlusion is active.
# ``occlusion_phase1`` : two-phase set A (predicted occluders drawn
# into the scratch depth this frame).
# ``occlusion_phase2`` : two-phase set B survivors (newly disoccluded
# objects admitted against the fresh Hi-Z).
# phase1 + phase2 == occlusion_drawn.
# ``view_occlusion_drawn``/``view_occlusion_total``: the same
# drawn / pre-cull pair summed across every
# per-view occlusion bundle (SubViewport /
# RenderView with ``use_occlusion=True``).
# Present only when such a view rendered.
self.last_telemetry: dict[str, Any] = {}
type(self)._current_ref = weakref.ref(self)
[docs]
def toggle_fullscreen(self) -> None:
"""Toggle fullscreen mode."""
if self._engine:
self._engine.toggle_fullscreen()
[docs]
@property
def is_fullscreen(self) -> bool:
return self._engine.is_fullscreen if self._engine else False
[docs]
@property
def engine(self) -> Engine | None:
"""Access the graphics engine (available after run() starts)."""
return self._engine
[docs]
@property
def capabilities(self):
"""Immutable :class:`RenderCapabilities` snapshot (host + GPU).
Forwards to ``app.engine.capabilities``. ``None`` until ``run()`` has
initialised the Vulkan engine. Read-only; rendering config flows through
``WorldEnvironment`` / ``App`` construction, never through this object.
"""
return self._engine.capabilities if self._engine is not None else None
[docs]
@property
def texture_manager(self):
"""The renderer's :class:`TextureManager` (available after run() starts).
Convenience accessor: forwards to ``app.engine.texture_manager`` so
scenes that upload runtime-generated pixels (e.g. rasterised text)
don't need to reach into ``_engine``.
"""
return self._engine.texture_manager if self._engine is not None else None
[docs]
@property
def scene_adapter(self):
"""Scene adapter bridging SceneTree to the renderer (available after run() starts)."""
return getattr(self, "_scene_adapter", None)
[docs]
def register_lut(self, tex_id: int, lut_data) -> int:
"""Register a 3D colour-grading LUT under ``tex_id`` (available after run() starts).
Forwards to :meth:`Engine.register_lut`. Drive the LUT from a scene via
``WorldEnvironment.lut_tex_id`` + ``lut_enabled``; the engine binds the
registered 3D LUT in the tonemap pass.
"""
if self._engine is None:
raise RuntimeError("register_lut: app is not running yet")
return self._engine.register_lut(tex_id, lut_data)
[docs]
@property
def active_backend(self) -> str | None:
"""Return the resolved backend name (e.g. ``"glfw"``), or ``None`` if
the engine has not yet initialised a window."""
if self._engine is None:
return None
return self._engine._resolved_backend_name
@property
def window_title(self) -> str:
"""Current window title. Writable once the window exists."""
if self._engine is not None:
return self._engine.title
return self.title
[docs]
@window_title.setter
def window_title(self, value: str) -> None:
if not isinstance(value, str):
raise TypeError(f"window_title must be a string, got {type(value).__name__}")
self.title = value
if self._engine is not None:
self._engine.title = value
window = getattr(self._engine, "_window", None)
if window is not None and hasattr(window, "set_title"):
window.set_title(value)
@property
def window_size(self) -> tuple[int, int]:
"""Current window ``(width, height)``. Writable once the window exists."""
if self._engine is not None:
return (self._engine.width, self._engine.height)
return (self.width, self.height)
[docs]
@window_size.setter
def window_size(self, value: tuple[int, int]) -> None:
if not isinstance(value, tuple | list) or len(value) != 2:
raise TypeError("window_size must be a (width, height) tuple")
w, h = int(value[0]), int(value[1])
if w < 1 or h < 1:
raise ValueError(f"window_size must be >= 1x1, got {w}x{h}")
self.width = w
self.height = h
if self._engine is not None:
self._engine.window_size = (w, h)
[docs]
@property
def cursor_pos(self) -> tuple[float, float]:
"""Current cursor position in screen coordinates, ``(0.0, 0.0)`` before run()."""
if self._engine is not None:
return self._engine.cursor_pos
return (0.0, 0.0)
[docs]
@property
def vsync(self) -> bool:
"""Whether vertical sync is currently enabled."""
if self._engine is not None:
return self._engine.vsync
return self._vsync
[docs]
def set_vsync(self, value: bool) -> None:
"""Toggle vsync at runtime.
Before ``run()`` this just stores the boot-time value. After
``run()``, the engine recreates the swapchain with the new
present mode (FIFO when on, MAILBOX/IMMEDIATE when off).
"""
value = bool(value)
self._vsync = value
engine = getattr(self, "_engine", None)
if engine is not None:
engine.set_vsync(value)
@property
def time_scale(self) -> float:
"""Global time-scale multiplier for the scene-tree tick.
Applied to ``dt`` before delivery to ``tree.process()`` and
``tree.physics_process()``. ``1.0`` is real-time, ``0.3`` is 30%
speed (slow-motion / hitstop), ``2.0`` is double speed, ``0.0`` is
paused (still ticks, with zero dt). Default ``1.0``.
Scope is intentionally narrow: only the scene-tree dt is scaled.
Audio playback rate, coroutine pacing, and the wall-clock physics
accumulator are NOT touched: slowing audio + coroutines requires a
separate design pass. Read :attr:`simvx.core.SceneTree.now` for
the scaled scene clock.
"""
return self._time_scale
[docs]
@time_scale.setter
def time_scale(self, value: float) -> None:
v = float(value)
if v < 0.0:
raise ValueError(f"App.time_scale must be >= 0, got {v}")
self._time_scale = v
[docs]
@property
def last_draw2d_draw_count(self) -> int | None:
"""The 2D draw count (bindless item submit) from the most recent
``run_headless`` call's final rendered frame.
``None`` if no run has happened yet.
"""
return self.last_telemetry.get("draw2d_draw_count")
def _populate_telemetry(self, *, frames_rendered: int, occlusion_wait: bool = False) -> None:
"""Refresh ``self.last_telemetry`` from the live renderer.
Shared by the headless and windowed loops so both expose the same
per-frame counters. ``occlusion_*`` keys are added only while the
renderer's Hi-Z occlusion cull is active (zero overhead otherwise: the
telemetry read is skipped entirely when the gate flag is off).
``occlusion_wait`` forces a ``vkDeviceWaitIdle`` before reading the
culled indirect commands back from GPU memory. The windowed loop leaves
it ``False`` (it reads the previous frame's already-completed dispatch,
kept fresh by the prepass); headless capture sets it ``True`` so the
single final read is exact.
"""
r = self._engine._renderer if self._engine else None
if r is None:
return
t = self.last_telemetry
if hasattr(r, "draw2d_draw_count"):
t["draw2d_draw_count"] = int(r.draw2d_draw_count)
t["gpu_phase_times"] = dict(getattr(self._engine, "gpu_phase_times", {}) or {})
t["frames_rendered"] = frames_rendered
bufs = getattr(r, "_buffers", None)
if bufs is not None:
t["transform_capacity"] = int(getattr(bufs, "transform_capacity", 0))
t["transform_high_water"] = int(getattr(bufs, "transform_high_water", 0))
sr = getattr(r, "_scene_renderer", None)
if getattr(r, "_occlusion_culling_enabled", False) and sr is not None:
drawn = int(sr.read_occlusion_telemetry(wait=occlusion_wait))
phase1 = int(getattr(sr, "last_phase1_count", 0))
t["occlusion_drawn"] = drawn
t["occlusion_total"] = int(getattr(sr, "_last_pre_cull_count", 0))
t["occlusion_phase1"] = phase1
t["occlusion_phase2"] = max(0, drawn - phase1)
else:
for k in ("occlusion_drawn", "occlusion_total", "occlusion_phase1", "occlusion_phase2"):
t.pop(k, None)
# Per-view occlusion: aggregate across every SubViewport /
# RenderView bundle (views with ``use_occlusion=True`` only). Zero-cost
# when unused: both stores are empty dicts and nothing is read back.
bundles = [
b
for mgr in (getattr(self, "_sub_viewports", None), getattr(self, "_render_views", None))
if mgr is not None
for b in getattr(mgr, "_occlusion", {}).values()
]
if bundles:
t["view_occlusion_drawn"] = sum(b.read_drawn(wait=occlusion_wait) for b in bundles)
t["view_occlusion_total"] = sum(b.last_pre_cull_count for b in bundles)
else:
for k in ("view_occlusion_drawn", "view_occlusion_total"):
t.pop(k, None)
[docs]
def run(
self,
root_or_update: Any = None,
*,
update: Callable[[], None] | None = None,
render: Callable[[object, tuple[int, int]], None] | None = None,
) -> None:
"""Run the graphics engine.
Args:
root_or_update: Either a Node (scene root) or a per-frame update callback.
When a Node is passed, SceneTree and input are set up automatically.
update: Per-frame callback (alternative to the positional arg).
render: Custom render callback (cmd, extent). Only used in callback mode.
"""
from simvx.core import Node
if isinstance(root_or_update, Node):
self._run_with_tree(root_or_update)
else:
cb = root_or_update or update
self._run_with_callbacks(cb, render)
def _init_audio_backend(self, tree) -> None:
"""Attach an audio backend to the scene tree (best-effort).
`make_backend()` prefers the native ma_engine path (built via
`simvx build-audio`) and falls back to the pure-Python mixer if
the native extension isn't loaded, so games still produce sound
on systems without a C compiler at install time. Sample rate
and channel count come from the `App(audio_sample_rate=,
audio_channels=)` constructor kwargs.
"""
try:
from simvx.core.audio_backend import make_backend
tree.install_audio_backend(
make_backend(
sample_rate=self._audio_sample_rate,
nchannels=self._audio_channels,
)
)
except (ImportError, OSError) as exc:
# Justified: the audio_backend module (or its native dependency) may be
# absent, or no audio device is present: the engine still runs silently.
# ``make_backend`` already degrades to the null backend internally, so the
# only escapes here are an import failure or a device-level OSError. A
# deliberate ``AudioBackendUnavailable`` (SIMVX_ALLOW_LEGACY_AUDIO=0) is
# NOT caught: it is a fail-loud opt-in and must propagate.
log.warning("Audio backend unavailable; running without sound: %s", exc, exc_info=True)
@staticmethod
def _shutdown_audio(tree) -> None:
"""Stop the audio backend so its playback thread doesn't block process exit."""
backend = getattr(tree, "_audio_backend", None)
if backend and hasattr(backend, "shutdown"):
backend.shutdown()
def _run_with_callbacks(
self,
update: Callable | None,
render: Callable | None,
) -> None:
"""Run with raw per-frame update/render callbacks (no scene tree)."""
self._engine = Engine(
width=self.width,
height=self.height,
title=self.title,
backend=self._backend_name,
visible=self._visible,
vsync=self._vsync,
target_fps=self._target_fps,
)
self._engine._multi_gpu_requested = self._multi_gpu
if self._bg_colour == "transparent":
self._engine.clear_colour = [0.0, 0.0, 0.0, 0.0]
elif isinstance(self._bg_colour, tuple | list):
self._engine.clear_colour = list(self._bg_colour)
else:
from simvx.core.ui.theme import get_theme as _gt
c = _gt().bg_black
self._engine.clear_colour = [c[0], c[1], c[2], c[3]]
self._engine.run(callback=update, render=render)
def _resolve_pipelined(self, tree: Any) -> bool:
"""Resolve whether to run the pipelined render thread for this scene.
``App(render_thread=True)`` is a per-process override that forces it on.
Otherwise the authored home is a ``WorldEnvironment`` in the scene with
``render_mode='pipelined'``. Default is the synchronous path.
"""
if self._render_thread:
return True
from simvx.core import WorldEnvironment
env = tree.root.find(WorldEnvironment) if tree.root is not None else None
return env is not None and getattr(env, "render_mode", "default") == "pipelined"
def _warn_pipelined_unpacketised(self, tree: Any) -> None:
"""Warn ONCE if a pipelined scene uses content the packet does not snapshot.
Tilemap layers, 2D lights/occluders, 3D-overlay text, and SubViewport
render-to-texture are now packetised and render correctly in pipelined
mode. Reflection-probe capture remains deferred (its IBL convolution /
cube-array copy state machine is still too GPU-stateful to replay from a
packet), and RenderView capture is deferred the same way (D6: it re-renders
the live MAIN tree, which the render thread must never walk). Detect either
in a pipelined scene after ``submit_scene`` and log a single clear warning
that the affected capture renders stale this run.
"""
if self._warned_pipelined_unpacketised:
return
renderer = self._engine.renderer
stale: list[str] = []
try:
if renderer._collect_reflection_probes():
stale.append("reflection-probe capture")
except Exception: # noqa: BLE001 - detection must never break the frame
log.debug("pipelined-limitation detection failed", exc_info=True)
try:
from simvx.core import RenderView
if tree.root is not None and tree.root.find(RenderView) is not None:
stale.append("RenderView capture")
except Exception: # noqa: BLE001 - detection must never break the frame
log.debug("pipelined-limitation detection failed", exc_info=True)
if stale:
self._warned_pipelined_unpacketised = True
log.warning(
"pipelined render mode does not yet packetise %s: this will render "
"stale this run. Cube, Draw2D, tilemap, 2D-light, text, and "
"SubViewport content are packetised and unaffected. Packetising "
"reflection probes and RenderViews is a known follow-up.",
", ".join(stale),
)
def _run_with_tree(self, root_node: Any) -> None:
"""Run with a node tree: full engine integration."""
from .frame_loop import FrameLoop, RealTimeClock, SwapchainSink, WindowedIntegration
loop = FrameLoop(
self,
root_node,
clock=RealTimeClock(),
sink=SwapchainSink(),
integration=WindowedIntegration(),
visible=self._visible,
vsync=self._vsync,
)
loop.run()
[docs]
def quit(self) -> None:
"""Request a clean shutdown of the running app.
Safe to call from process/physics_process or input handlers. The main
loop exits at the end of the current frame; audio and Vulkan resources
are torn down in the usual finally-paths. Prefer this over ``sys.exit``
which can leave background threads (e.g. miniaudio) alive and stall
process exit.
"""
if self._engine is not None:
self._engine._running = False
tree = getattr(self._engine, "_scene_tree", None)
if tree is not None:
tree.quit()
# ------------------------------------------------------------------
# Headless rendering (for tests and CI)
# ------------------------------------------------------------------
[docs]
def run_headless(
self,
root_node: Any,
*,
frames: int = 1,
on_frame: Callable[[int, float], bool | None] | None = None,
capture_frames: list[int] | None = (),
capture_fn: Callable[[int], bool] | None = None,
) -> list:
"""Run the engine headlessly for *frames* frames and return captured pixels.
Args:
root_node: Scene root node.
frames: Total number of frames to simulate.
on_frame: Optional callback invoked with (frame_index, time) before each frame.
Return False to stop early.
capture_frames: Which frame indices to capture. Defaults to ``()``: no capture.
Per-frame swapchain readback costs ~4 ms/frame (``vkQueueWaitIdle`` +
``vkDeviceWaitIdle`` + host copy), so the default is no-capture so that
tests/CI/benchmarks measure real frame work, not host-side readback.
Pass an explicit list of frame indices (e.g. ``[frames - 1]`` for the
final frame) to opt in, or ``None`` to capture every frame.
capture_fn: Dynamic capture predicate: called with frame index, captures if True.
Takes precedence over *capture_frames* when provided.
Returns:
List of (H, W, 4) uint8 RGBA numpy arrays, one per captured frame
(empty when the default ``capture_frames=()`` is used).
"""
from .frame_loop import FixedStepClock, FrameLoop, HeadlessIntegration, OffscreenSink
sink = OffscreenSink(capture_frames=capture_frames, capture_fn=capture_fn)
loop = FrameLoop(
self,
root_node,
clock=FixedStepClock(frames=frames),
sink=sink,
integration=HeadlessIntegration(on_frame=on_frame),
visible=False,
vsync=False,
)
loop.run()
return sink.captured
# ------------------------------------------------------------------
# Streaming (JPEG over WebSocket to browser)
# ------------------------------------------------------------------
[docs]
def run_streaming(
self,
root_node: Any,
server: Any,
) -> None:
"""Run the engine headlessly and stream frames to browsers over WebSocket.
Args:
root_node: Scene root node.
server: A ``StreamingServer`` instance (from ``simvx.graphics.streaming``).
"""
from .frame_loop import FrameLoop, RealTimeClock, StreamingIntegration, StreamSink
loop = FrameLoop(
self,
root_node,
clock=RealTimeClock(),
sink=StreamSink(server),
integration=StreamingIntegration(server),
visible=False,
vsync=False,
)
loop.run()