The Game Loop

The engine runs a game as tasks, once per rendered frame, with a frame time that varies. Game logic – movement, physics, timers – is easier to get right at a fixed rate. The framework provides that rate and the pieces that run on it:

  • acron.fixed_loop – GameLoop, a fixed-step loop with render interpolation;

  • acron.timers – countdowns ticked by the fixed step;

  • acron.signals – per-object signals;

  • acron.governor – trading render resolution for a steady frame rate.

The fixed-step loop

GameLoop keeps a wall-clock accumulator and runs as many fixed steps as the elapsed time calls for. Each step, in this order:

  1. calls _physics_process(dt) on every registered object;

  2. steps the physics world (physics.do_physics(dt)), which moves the Bullet bodies and syncs their nodes;

  3. calls update_visuals() on every registered object that has one, to place visuals against the final transforms;

  4. calls end_step() on the input map, so an action’s “just pressed” lasts exactly one step.

from direct.showbase.ShowBase import ShowBase
from acron import GameLoop, PhysicsWorld, InputMap

base = ShowBase()
physics = PhysicsWorld()
inputs = InputMap(base, held={"move_forward": ["w"]},
                  edges={"jump": ["space"]})
loop = GameLoop(physics, base.task_mgr, base.clock, input_map=inputs)

loop.register(player)    # player._physics_process(dt) runs at 60 Hz
loop.start()
base.run()

GameLoop(physics=None, task_mgr=None, clock=None, input_map=None, rate_hz=60, max_steps=5, max_debt_s=None):

physics

A PhysicsWorld, anything with a do_physics(dt), or None for a loop that only runs the callbacks.

task_mgr, clock

The task manager and clock. start() adds a task (named acron_fixed_loop) that feeds clock.get_dt() to advance. A loop that is never started does not need them.

input_map

An InputMap whose end_step() runs after every step.

rate_hz

The step rate; fixed_dt is 1 / rate_hz.

max_steps, max_debt_s

What happens when the loop falls behind; see Falling behind.

register(obj) and unregister(obj) add and remove objects. An object needs a _physics_process(dt) method; update_visuals() and _process(dt) are optional. Setting loop.paused = True stops the steps; the time that passes while paused is dropped, so nothing catches up on resume.

Render interpolation

The world advances in whole steps, but frames are drawn at the display’s rate. A camera or model placed at the body’s position every frame moves in uneven jumps, more visibly the faster it goes. loop.alpha, from 0 to 1, is how far the current frame is between the last two steps. Record the position before and after each step, and draw at interpolated(prev, curr, loop.alpha), which is prev + (curr - prev) * alpha:

from acron import interpolated

class Ball:
    def __init__(self, body, model):
        self.body = body          # a CharacterBody3D, say
        self.model = model
        self.prev = self.curr = body.np.get_pos()

    def _physics_process(self, dt):
        self.prev = self.curr
        self.body.move_and_slide(dt)
        self.curr = self.body.np.get_pos()

    def update_visuals(self):
        self.model.set_pos(interpolated(self.prev, self.curr, loop.alpha))

update_visuals() runs after every fixed step, but alpha only changes between frames, so call loop.update_visuals() once per frame as well, from a task of your own; without it, visuals still move in whole steps.

Per-frame work

loop.process(dt) calls _process(dt) on every registered object that has one. Call it once per frame with the frame time, for work that belongs to the frame rather than the step: effects that integrate per frame, and the animations that drive them, so both keep one clock.

def per_frame(task):
    dt = base.clock.get_dt()
    loop.process(dt)
    loop.update_visuals()
    return task.cont

base.task_mgr.add(per_frame, "per_frame", sort=1)

Falling behind

When more time arrives than the loop should simulate at once, the max_debt_s setting decides:

  • max_debt_s=None (the default): run at most max_steps steps, then drop the rest. A game that is drawing the world must not fast-forward it.

  • max_debt_s=seconds: run every whole step that is due, so a simulation that others predict as tick * dt stays in step; but debt beyond max_debt_s seconds (a stall, a debugger, a sleeping laptop) is dropped first. This is the setting for a server. A value below one step is refused with a ValueError.

Either way, loop.last_skip is the simulated time the last advance dropped (0.0 when nothing was).

Driving the loop yourself

advance(dt) adds dt seconds, runs the steps that are due, updates alpha and returns how many steps ran. start() is this with the engine’s clock; a program with a clock of its own, such as a dedicated server on time.monotonic(), calls it directly.

step(n=1) runs exactly n steps without looking at any clock. Use it in tests: in an offscreen test the global clock barely advances between frames, so an accumulator would run no steps at all.

loop = GameLoop(physics)
loop.register(player)
loop.step(60)                       # one simulated second
assert player.body.is_on_floor()

Timers

Timer is a countdown that emits a timeout signal when it runs out. Unlike doMethodLater, which follows the wall clock, a timer only advances when it is ticked – by the fixed step – so it pauses with the game and runs the same in a test.

Timer(wait_time=1.0, one_shot=True, autostart=False, group=None):

  • start(wait_time=None) starts the countdown again, optionally with a new wait_time; stop() stops it; is_stopped() tells.

  • time_left is the time remaining.

  • tick(dt) advances it. A one-shot timer stops when it fires; a repeating one starts its next period.

A TimerGroup ticks a set of timers at once: group.new(...) makes a timer in the group, group.add(timer) adds one, and group.tick(dt) ticks them all. One group per game object, ticked from its _physics_process, is typical.

from acron.timers import TimerGroup

class Turret:
    def __init__(self):
        self.timers = TimerGroup()
        self.reload = self.timers.new(wait_time=0.8)
        self.reload.timeout.connect(self.on_reloaded)
        self.loaded = True

    def fire(self):
        if self.loaded:
            self.loaded = False
            self.reload.start()

    def on_reloaded(self):
        self.loaded = True

    def _physics_process(self, dt):
        self.timers.tick(dt)

Signals

The engine’s messenger is a single bus of events named by strings. Signal belongs to one object instead: each enemy can have its own died signal, without making up a unique event name for each.

from acron import Signal

class Enemy:
    def __init__(self):
        self.died = Signal("died")

enemy = Enemy()
enemy.died.connect(lambda: print("one fewer"))
enemy.died.emit()

connect(callback) adds a callback once (connecting it again does nothing), disconnect(callback) removes it, and emit(*args) calls each callback with the arguments. A callback may disconnect itself while the signal is being emitted. To pass extra arguments, connect a lambda or a functools.partial.

Resolution and frame time

acron.governor treats render resolution as a budget. It assumes the scene is drawn into an offscreen buffer (for example by a FilterManager) and only ever scales that buffer, never the window.

pixel_scale(width, height, budget_pixels, render_scale=1.0, floor=0.25) returns the scale that keeps a width by height framebuffer within budget_pixels pixels in total, times render_scale, and no less than floor. Capping the pixel count, rather than a ratio, gives the same cost on a high-DPI display as on any other.

Governor watches a smoothed frame time and changes its scale once per period seconds: down by step_down when frames take longer than slow_ms, back up by step_up when they take less than fast_ms, between floor and 1.0. update(dt) folds in one frame and returns True when the scale moved. It is plain arithmetic, not a task, so it can be tested with a list of frame times. The defaults (period=1.1, slow_ms=23.0, fast_ms=13.5, step_down=0.08, step_up=0.06, floor=0.62) suit a 60 Hz target; adaptive=False holds the scale.

force_scene_scale(manager, win, scale, current) sets a FilterManager’s scene buffer to scale times the window size, and returns the new (width, height), or None when it did not change. The manager refits its buffers to the window on every window event, so call it again after a resize.

from acron.governor import Governor, force_scene_scale, pixel_scale

governor = Governor()
size = (0, 0)

def fit(task):
    global size
    governor.update(base.clock.get_dt())
    scale = pixel_scale(base.win.get_x_size(), base.win.get_y_size(),
                        budget_pixels=1920 * 1080,
                        render_scale=governor.scale)
    new = force_scene_scale(manager, base.win, scale, size)
    if new is not None:
        size = new
    return task.cont

base.task_mgr.add(fit, "fit_scene")