chuanqi/apilot/stubs/scheduler.py

# apilot/stubs/scheduler.py
"""计时轮: Python 侧驱动, Lua 侧全局函数注入。对齐游戏 50ms tick 节奏。"""
from __future__ import annotations
import itertools
from typing import Callable


class Scheduler:
    def __init__(self) -> None:
        self._ids = itertools.count(1)
        self._timers: dict[int, tuple[Callable[[], None], float, float, bool]] = {}
        # id -> (fn, interval, elapsed, repeat)

    def schedule(self, fn: Callable[[], None], interval_s: float) -> int:
        eid = next(self._ids)
        self._timers[eid] = (fn, interval_s, 0.0, True)
        return eid

    def delay(self, fn: Callable[[], None], delay_s: float) -> int:
        eid = next(self._ids)
        self._timers[eid] = (fn, delay_s, 0.0, False)
        return eid

    def unschedule(self, eid: int) -> None:
        self._timers.pop(eid, None)

    def tick(self, dt: float) -> None:
        for eid in list(self._timers):
            entry = self._timers.get(eid)
            if entry is None:  # 回调内已 unschedule 本 tick 未遍历的条目
                continue
            fn, interval, elapsed, repeat = entry
            elapsed += dt
            if elapsed >= interval:
                if repeat:
                    self._timers[eid] = (fn, interval, elapsed - interval, True)
                else:
                    self._timers.pop(eid)
                fn()
            else:
                self._timers[eid] = (fn, interval, elapsed, repeat)


def install_scheduler(lua, sched: Scheduler) -> None:
    py_sched = sched  # Lupa: Python 对象直接可见

    def lua_schedule(fn, interval: float, loops=None) -> int:
        # 游戏里 Schedule(fn, interval, loops) 第三参 nil/缺省=循环, >0=一次性
        if loops is None or loops == 0:
            return py_sched.schedule(fn, float(interval))
        return py_sched.delay(fn, float(interval))

    def lua_perform_delay(fn, delay_s: float) -> int:
        return py_sched.delay(fn, float(delay_s))

    def lua_unschedule(eid: int) -> None:
        py_sched.unschedule(int(eid))

    lua.globals()["PerformWithDelayGlobal"] = lua_perform_delay
    lua.globals()["Schedule"] = lua_schedule
    lua.globals()["UnSchedule"] = lua_unschedule
    lua.globals()["SL_getScheduler"] = lambda: py_sched