lgbot: review fixes: drift reset after pauses, player-like acceleration, sticky targets with hysteresis, wall backoff; plugin glue tests
Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
1 parent
a998614224
commit
86754715f5
6 files changed
+321
-30
No files matched your search
@@ -147,11 +147,14 @@ def refuse_impostor(status_text):
|
||||
return False
|
||||
|
||||
|
||||
REFUSED = 3
|
||||
|
||||
|
||||
def apply(st, old_aim):
|
||||
run = server()
|
||||
status = run("status")
|
||||
if refuse_impostor(status):
|
||||
return 1
|
||||
return REFUSED
|
||||
bot_in = bot_present(status)
|
||||
for cmd in build_commands(st, readd=(not bot_in or st["aim"] != old_aim), bot_in=bot_in):
|
||||
run(cmd)
|
||||
@@ -162,7 +165,7 @@ def apply(st, old_aim):
|
||||
|
||||
def main(argv, input_fn=input):
|
||||
arg = argv[1] if len(argv) > 1 else ""
|
||||
st = load_state()
|
||||
st = load_state(STATE)
|
||||
if arg == "off":
|
||||
run = server()
|
||||
if refuse_impostor(run("status")):
|
||||
@@ -187,8 +190,10 @@ def main(argv, input_fn=input):
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
print("\ncancelled, nothing changed")
|
||||
return 1
|
||||
save_state(STATE, new)
|
||||
return apply(new, old_aim=st["aim"])
|
||||
rc = apply(new, old_aim=st["aim"])
|
||||
if rc != REFUSED: # saved even if the bot didn't show up (server restarting): --last retries
|
||||
save_state(STATE, new)
|
||||
return rc
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
+89
-2
@@ -1,7 +1,7 @@
|
||||
import math, os, random, sys, unittest
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "home", "minqlx-plugins"))
|
||||
from lgbot import (PRESETS, KEYS, resolve, Strafer, wanted_range, steer, fix_z, pick_target, to_int, is_bot, is_lgbot, StuckGuard)
|
||||
from lgbot import (PRESETS, KEYS, resolve, Strafer, wanted_range, steer, fix_z, pick_target, to_int, is_bot, is_lgbot, StuckGuard, approach, ACCEL)
|
||||
|
||||
|
||||
def preset(n, **kw):
|
||||
@@ -83,6 +83,40 @@ class StraferTest(unittest.TestCase):
|
||||
d = st.step(now, s)
|
||||
|
||||
|
||||
def test_gap_does_not_count_as_drift(self):
|
||||
# bot dead / out of range for 10 s: afterwards it must still strafe both ways (C1)
|
||||
s = preset(5)
|
||||
st, now = Strafer(random.Random(5)), 0
|
||||
for _ in range(80):
|
||||
now += 25; st.step(now, s)
|
||||
now += 10000
|
||||
dirs = []
|
||||
for _ in range(400):
|
||||
now += 25; dirs.append(st.step(now, s))
|
||||
self.assertLess(max(dirs.count(1), dirs.count(-1)) / 400.0, 0.75)
|
||||
|
||||
def test_reset_clears_drift(self):
|
||||
st = Strafer(random.Random(1))
|
||||
st.drift, st.last = 5000, 123
|
||||
st.reset()
|
||||
self.assertEqual((st.drift, st.last), (0, None))
|
||||
|
||||
|
||||
class Accel(unittest.TestCase):
|
||||
def test_limits_change_per_frame(self):
|
||||
v = approach((320, 0), (-320, 0), ACCEL * 0.025)
|
||||
self.assertAlmostEqual(v[0], 320 - ACCEL * 0.025, places=6)
|
||||
|
||||
def test_reaches_target_when_close(self):
|
||||
self.assertEqual(approach((300, 0), (320, 0), 80), (320, 0))
|
||||
|
||||
def test_reversal_takes_human_time(self):
|
||||
v, frames = (320.0, 0.0), 0
|
||||
while v[0] > -319:
|
||||
v = approach(v, (-320, 0), ACCEL * 0.025); frames += 1
|
||||
self.assertTrue(6 <= frames <= 10, frames) # ~150-250 ms, not one frame
|
||||
|
||||
|
||||
class Range(unittest.TestCase):
|
||||
def test_plain(self):
|
||||
self.assertEqual(wanted_range(preset(1), 100, 100), 450)
|
||||
@@ -158,7 +192,7 @@ class Misc(unittest.TestCase):
|
||||
def test_pick_target(self):
|
||||
c = [((300, 0, 0), "near"), ((800, 0, 0), "far")]
|
||||
self.assertEqual(pick_target((0, 0, 0), c), "near")
|
||||
self.assertIsNone(pick_target((0, 0, 0), [((1000, 0, 0), "x")])) # beyond LG reach: AI walks there
|
||||
self.assertIsNone(pick_target((0, 0, 0), [((950, 0, 0), "x")])) # too far to start steering
|
||||
self.assertIsNone(pick_target((0, 0, 0), []))
|
||||
|
||||
def test_pick_target_ignores_other_floors(self):
|
||||
@@ -166,6 +200,17 @@ class Misc(unittest.TestCase):
|
||||
self.assertEqual(pick_target((0, 0, 0), c), "same floor")
|
||||
self.assertIsNone(pick_target((0, 0, 0), [((200, 0, -300), "below")]))
|
||||
|
||||
def test_keeps_current_target_a_bit_past_the_start_limit(self):
|
||||
self.assertEqual(pick_target((0, 0, 0), [((950, 0, 0), "a")], current="a"), "a") # hysteresis
|
||||
self.assertIsNone(pick_target((0, 0, 0), [((1050, 0, 0), "a")], current="a"))
|
||||
|
||||
def test_sticky_target(self):
|
||||
c = [((500, 0, 0), "a"), ((420, 0, 0), "b")]
|
||||
self.assertEqual(pick_target((0, 0, 0), c, current="a"), "a") # b only 80 closer: stay
|
||||
c = [((500, 0, 0), "a"), ((300, 0, 0), "b")]
|
||||
self.assertEqual(pick_target((0, 0, 0), c, current="a"), "b") # clearly closer: switch
|
||||
self.assertEqual(pick_target((0, 0, 0), [((400, 0, 0), "b")], current="a"), "b") # a gone
|
||||
|
||||
|
||||
class Stuck(unittest.TestCase):
|
||||
def test_steers_while_moving(self):
|
||||
@@ -180,6 +225,29 @@ class Stuck(unittest.TestCase):
|
||||
back_on = next(t for t, ok in results if t > first_off and ok)
|
||||
self.assertTrue(1900 <= back_on - first_off + 500 <= 2100) # off for 1.5 s
|
||||
|
||||
def test_backs_off_longer_when_it_keeps_failing(self):
|
||||
g = StuckGuard()
|
||||
offs, prev = [], True
|
||||
for t in range(0, 20000, 25):
|
||||
ok = g.update(t, 320, 50)
|
||||
if prev and not ok:
|
||||
offs.append(t)
|
||||
prev = ok
|
||||
gaps = [b - a for a, b in zip(offs, offs[1:])]
|
||||
self.assertGreater(gaps[1], gaps[0]) # 1.5 s, 3 s, 6 s ... between retries
|
||||
self.assertLessEqual(max(gaps), 6000 + 525)
|
||||
|
||||
def test_backoff_resets_after_clean_steering(self):
|
||||
g = StuckGuard()
|
||||
for t in range(0, 4000, 25):
|
||||
g.update(t, 320, 50) # two releases: streak up
|
||||
for t in range(4000, 12000, 25):
|
||||
g.update(t, 320, 300) # clean for a while
|
||||
res = [(t, g.update(t, 320, 50)) for t in range(12000, 16000, 25)]
|
||||
off = next(t for t, ok in res if not ok)
|
||||
back = next(t for t, ok in res if t > off and ok)
|
||||
self.assertEqual(back - off, 1500) # back to a 1.5 s release
|
||||
|
||||
def test_short_bump_is_fine(self):
|
||||
g = StuckGuard()
|
||||
ok = [g.update(t, 320, 50 if 1000 <= t < 1300 else 300) for t in range(0, 3000, 25)]
|
||||
@@ -267,6 +335,25 @@ class Parsing(unittest.TestCase):
|
||||
self.assertFalse(setup.bot_present("(no output)"))
|
||||
|
||||
|
||||
class Main(unittest.TestCase):
|
||||
def test_refused_run_keeps_saved_setup(self):
|
||||
path = os.path.join(tempfile.mkdtemp(), "lgbot.json")
|
||||
impostor_status = STATUS_WITH_BOT.replace("lgbot 0 bot ", "lgbot 0 1.2.3.4:5")
|
||||
sent = []
|
||||
def run(cmd):
|
||||
sent.append(cmd)
|
||||
return impostor_status if cmd == "status" else ""
|
||||
old_state, old_server = setup.STATE, setup.server
|
||||
setup.STATE, setup.server = path, (lambda: run)
|
||||
try:
|
||||
rc = setup.main(["lgbot"], feeder("1", "5", "n"))
|
||||
finally:
|
||||
setup.STATE, setup.server = old_state, old_server
|
||||
self.assertNotEqual(rc, 0)
|
||||
self.assertFalse(os.path.exists(path))
|
||||
self.assertEqual(sent, ["status"]) # no kick, no set
|
||||
|
||||
|
||||
class Commands(unittest.TestCase):
|
||||
def test_no_details_resets_to_minus_one(self):
|
||||
cmds = setup.build_commands({"aim": 2, "move": 4, "details": {}}, readd=False, bot_in=True)
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
"""The lgbot plugin's frame logic (tick) against a fake minqlx: deaths, gaps, targets, acceleration."""
|
||||
import importlib.util, logging, math, os, sys, types, unittest
|
||||
|
||||
PLUGIN = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "home", "minqlx-plugins", "lgbot.py")
|
||||
BOT_ID = 90071996842377216
|
||||
HUMAN_ID = 76561198000000000
|
||||
|
||||
|
||||
class V:
|
||||
def __init__(self, x, y, z):
|
||||
self.x, self.y, self.z = x, y, z
|
||||
|
||||
|
||||
class FakePlayer:
|
||||
def __init__(self, cid, name, steam_id, pos):
|
||||
self.id, self.clean_name, self.steam_id = cid, name, steam_id
|
||||
self.team, self.is_alive, self.health = "free", True, 100
|
||||
self.pos, self.vel = list(pos), [0.0, 0.0, 0.0]
|
||||
self.sets = []
|
||||
|
||||
def position(self):
|
||||
return V(*self.pos)
|
||||
|
||||
def velocity(self, **kw):
|
||||
if kw:
|
||||
self.vel = [kw.get("x", self.vel[0]), kw.get("y", self.vel[1]), kw.get("z", self.vel[2])]
|
||||
self.sets.append(tuple(self.vel))
|
||||
return True
|
||||
return V(*self.vel)
|
||||
|
||||
|
||||
class World:
|
||||
"""Stands in for the server: players, cvars, a clock; moves players by their velocity."""
|
||||
|
||||
def __init__(self):
|
||||
self.players, self.cvars, self.now = [], {}, 1000.0
|
||||
|
||||
def frame(self, plugin, dt=0.025):
|
||||
self.now += dt
|
||||
plugin.handle_frame()
|
||||
for p in self.players:
|
||||
p.pos[0] += p.vel[0] * 0.85 * dt # friction eats ~15%, like the measured server
|
||||
p.pos[1] += p.vel[1] * 0.85 * dt
|
||||
p.vel[0] *= 0.85
|
||||
p.vel[1] *= 0.85
|
||||
|
||||
|
||||
def load(world):
|
||||
fake = types.ModuleType("minqlx")
|
||||
|
||||
class Plugin:
|
||||
def __init__(self):
|
||||
self.logger = logging.getLogger("lgbot-test")
|
||||
|
||||
def set_cvar_once(self, name, value):
|
||||
world.cvars.setdefault(name, value)
|
||||
|
||||
def get_cvar(self, name):
|
||||
return world.cvars.get(name)
|
||||
|
||||
def add_hook(self, event, fn):
|
||||
pass
|
||||
|
||||
def players(self):
|
||||
return list(world.players)
|
||||
|
||||
def log_exception(plugin=None):
|
||||
raise # surface plugin errors in the test
|
||||
|
||||
fake.Plugin, fake.log_exception = Plugin, log_exception
|
||||
sys.modules["minqlx"] = fake
|
||||
try:
|
||||
spec = importlib.util.spec_from_file_location("lgbot_under_test", PLUGIN)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
finally:
|
||||
del sys.modules["minqlx"]
|
||||
mod.time = types.SimpleNamespace(time=lambda: world.now)
|
||||
plugin = mod.lgbot()
|
||||
plugin.strafer.rng.seed(11) if hasattr(plugin.strafer.rng, "seed") else None
|
||||
return mod, plugin
|
||||
|
||||
|
||||
def sideways_sign(bot, human):
|
||||
# sign of the bot's velocity across the bot->human line
|
||||
dx, dy = human.pos[0] - bot.pos[0], human.pos[1] - bot.pos[1]
|
||||
return 1 if (-dy * bot.vel[0] + dx * bot.vel[1]) > 0 else -1
|
||||
|
||||
|
||||
class Tick(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.w = World()
|
||||
self.bot = FakePlayer(0, "lgbot", BOT_ID, (0, 0, 0))
|
||||
self.human = FakePlayer(1, "player", HUMAN_ID, (450, 0, 0))
|
||||
self.w.players = [self.bot, self.human]
|
||||
self.w.cvars["lgbot_move"] = "5"
|
||||
self.mod, self.plugin = load(self.w)
|
||||
|
||||
def run_frames(self, n):
|
||||
for _ in range(n):
|
||||
self.w.frame(self.plugin)
|
||||
|
||||
def test_steers_when_human_in_range(self):
|
||||
self.run_frames(40)
|
||||
self.assertTrue(self.bot.sets)
|
||||
|
||||
def test_human_named_lgbot_is_left_alone(self):
|
||||
self.bot.steam_id = HUMAN_ID
|
||||
self.run_frames(40)
|
||||
self.assertEqual(self.bot.sets, [])
|
||||
|
||||
def test_strafes_both_ways_after_respawn(self):
|
||||
self.run_frames(80)
|
||||
self.bot.is_alive = False
|
||||
self.run_frames(400) # 10 s dead
|
||||
self.bot.is_alive = True
|
||||
self.bot.pos = [0, 0, 0]
|
||||
signs = []
|
||||
for _ in range(400):
|
||||
self.w.frame(self.plugin)
|
||||
signs.append(sideways_sign(self.bot, self.human))
|
||||
self.assertLess(max(signs.count(1), signs.count(-1)) / 400.0, 0.75)
|
||||
|
||||
def test_strafes_both_ways_after_walking_in(self):
|
||||
self.human.pos = [3000, 0, 0] # far: AI walks
|
||||
self.run_frames(400)
|
||||
self.human.pos = [450, 0, 0]
|
||||
signs = []
|
||||
for _ in range(400):
|
||||
self.w.frame(self.plugin)
|
||||
signs.append(sideways_sign(self.bot, self.human))
|
||||
self.assertLess(max(signs.count(1), signs.count(-1)) / 400.0, 0.75)
|
||||
|
||||
def test_velocity_changes_like_a_player_not_instantly(self):
|
||||
self.run_frames(400)
|
||||
sets = self.bot.sets
|
||||
comp = self.mod.FRICTION_COMP
|
||||
biggest = max(math.hypot(b[0] - a[0] * 0.85, b[1] - a[1] * 0.85) for a, b in zip(sets, sets[1:]))
|
||||
# per frame: at most the acceleration step (plus friction compensation), never a 640 u/s flip
|
||||
self.assertLess(biggest, self.mod.ACCEL * 0.025 * comp + 0.15 * 320 * comp + 1)
|
||||
|
||||
def test_no_target_means_no_steering(self):
|
||||
self.human.team = "spectator"
|
||||
self.run_frames(40)
|
||||
self.assertEqual(self.bot.sets, [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user