64 lines
2.3 KiB
Python
64 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import heapq
|
|
import itertools
|
|
from dataclasses import dataclass, field
|
|
from typing import Callable
|
|
|
|
|
|
class GameClock:
|
|
"""Converts real elapsed time into discrete game ticks at a fixed rate."""
|
|
|
|
def __init__(self, tick_interval: float = 1.0):
|
|
self.tick_interval = tick_interval
|
|
self.total_time = 0.0
|
|
self.tick_count = 0
|
|
self._accumulator = 0.0
|
|
|
|
def advance(self, dt: float) -> int:
|
|
"""Advance by dt real seconds; returns how many whole ticks just elapsed."""
|
|
self.total_time += dt
|
|
self._accumulator += dt
|
|
ticks = int(self._accumulator // self.tick_interval)
|
|
if ticks:
|
|
self._accumulator -= ticks * self.tick_interval
|
|
self.tick_count += ticks
|
|
return ticks
|
|
|
|
|
|
@dataclass(order=True)
|
|
class _ScheduledEvent:
|
|
run_at: float
|
|
seq: int
|
|
event_id: str = field(compare=False)
|
|
callback: Callable[[], None] = field(compare=False)
|
|
interval: float | None = field(default=None, compare=False)
|
|
|
|
|
|
class Scheduler:
|
|
"""Min-heap of one-off or repeating callbacks, run against a game-time value you drive (e.g. GameClock.total_time)."""
|
|
|
|
def __init__(self):
|
|
self._heap: list[_ScheduledEvent] = []
|
|
self._seq = itertools.count()
|
|
|
|
def schedule_at(self, run_at: float, event_id: str, callback: Callable[[], None], interval: float | None = None) -> None:
|
|
heapq.heappush(self._heap, _ScheduledEvent(run_at, next(self._seq), event_id, callback, interval))
|
|
|
|
def schedule_in(self, now: float, delay: float, event_id: str, callback: Callable[[], None], interval: float | None = None) -> None:
|
|
self.schedule_at(now + delay, event_id, callback, interval)
|
|
|
|
def run_due(self, now: float) -> list[str]:
|
|
"""Runs (and pops) every event due at or before `now`; repeating events are rescheduled. Returns the ids run."""
|
|
ran: list[str] = []
|
|
while self._heap and self._heap[0].run_at <= now:
|
|
event = heapq.heappop(self._heap)
|
|
event.callback()
|
|
ran.append(event.event_id)
|
|
if event.interval is not None:
|
|
self.schedule_at(event.run_at + event.interval, event.event_id, event.callback, event.interval)
|
|
return ran
|
|
|
|
def pending_count(self) -> int:
|
|
return len(self._heap)
|