feat(apilot): python-driven scheduler wheel with lua globals

apilot/stubs/scheduler.py

@@ -0,0 +1,59 @@
+# 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):
+            fn, interval, elapsed, repeat = self._timers[eid]
+            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) -> 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

tests/test_scheduler.py

@@ -0,0 +1,44 @@
+# tests/test_scheduler.py
+import lupa.luajit21 as lj
+from apilot.stubs.scheduler import Scheduler, install_scheduler
+
+def test_interval_and_oneshot():
+    s = Scheduler()
+    hits = []
+    s.schedule(lambda: hits.append("i"), 0.05)
+    s.delay(lambda: hits.append("d"), 0.1)
+    for _ in range(30):
+        s.tick(0.05)
+    assert hits.count("i") >= 2
+    assert hits.count("d") == 1
+
+def test_lua_side_globals():
+    rt = lj.LuaRuntime()
+    from apilot.stubs.noop import install_noop_globals
+    install_noop_globals(rt)
+    s = Scheduler()
+    install_scheduler(rt, s)
+    rt.execute("global = global or {}; global.Scheduler = SL_getScheduler()")
+    # install_scheduler 需暴露一个 Lua 可达引用; 简化: 直接用注入的全局函数
+    rt.execute("n = 0; id = PerformWithDelayGlobal(function() n = n + 1 end, 0)")
+    for _ in range(3):
+        s.tick(0.05)
+    assert rt.eval("n") == 1
+
+def test_schedule_repeats():
+    """Schedule 第三参为 nil 时按循环语义注册。"""
+    rt = lj.LuaRuntime()
+    install_scheduler(rt, Scheduler())
+    rt.execute("n = 0; id = Schedule(function() n = n + 1 end, 0.05, nil)")
+    for _ in range(3):
+        s = rt.globals()["SL_getScheduler"]()
+        s.tick(0.05)
+    assert rt.eval("n") == 3
+
+def test_unschedule():
+    s = Scheduler()
+    hits = []
+    eid = s.schedule(lambda: hits.append(1), 0.05)
+    s.unschedule(eid)
+    s.tick(0.05)
+    assert hits == []