From e91023be29d97e0244b2b72691be87154bfd18da Mon Sep 17 00:00:00 2001 From: Daniel Fourman Date: Wed, 15 Jul 2026 12:32:24 +0300 Subject: [PATCH 1/8] Add telemetry observer/sinks and headless batch simulation Adds a telemetry package (observer + sinks) wired into the live game loop, new /telemetry, /simulate, /simulate/{job_id} server routes, and a simulate.py CLI for running batch games between two drivers without the websocket/UI rate throttle. Co-Authored-By: Claude Sonnet 5 --- Dockerfile | 2 +- README.md | 38 ++++++++++ rose/engine/logic.py | 71 +++++++++++++---- rose/engine/server.py | 130 +++++++++++++++++++++++++++++++- rose/engine/simulate.py | 108 ++++++++++++++++++++++++++ rose/engine/test_server.py | 129 +++++++++++++++++++++++++++++++ rose/engine/test_simulate.py | 115 ++++++++++++++++++++++++++++ rose/telemetry/__init__.py | 0 rose/telemetry/observer.py | 60 +++++++++++++++ rose/telemetry/sinks.py | 127 +++++++++++++++++++++++++++++++ rose/telemetry/test_observer.py | 73 ++++++++++++++++++ simulate.py | 55 ++++++++++++++ 12 files changed, 893 insertions(+), 15 deletions(-) create mode 100644 rose/engine/simulate.py create mode 100644 rose/engine/test_server.py create mode 100644 rose/engine/test_simulate.py create mode 100644 rose/telemetry/__init__.py create mode 100644 rose/telemetry/observer.py create mode 100644 rose/telemetry/sinks.py create mode 100644 rose/telemetry/test_observer.py create mode 100644 simulate.py diff --git a/Dockerfile b/Dockerfile index 0d9d715..4587714 100644 --- a/Dockerfile +++ b/Dockerfile @@ -24,4 +24,4 @@ ENV PORT 8880 EXPOSE 8880 # Define the command to run your app using CMD which defines your runtime -CMD ["sh", "-c", "python rose/main.py --listen 0.0.0.0 --track ${TRACK} --port ${PORT}"] +CMD ["sh", "-c", "python main.py --listen 0.0.0.0 --track ${TRACK} --port ${PORT}"] diff --git a/README.md b/README.md index 7e3ba42..9f7517e 100644 --- a/README.md +++ b/README.md @@ -129,3 +129,41 @@ The response in `JSON` format should include the car name and the recommended ac } } ``` + +## Batch simulation (headless automation) + +`simulate.py` runs many full games back-to-back between two drivers, without a +websocket/UI and without the live game's rate throttle, then writes aggregated +win/loss/tie stats to a JSON file. This is the same HTTP driver contract used by +the live engine (see `Testing your driver` above), so it works against any real +`rose-game-ai` driver process, community driver, or your own `mydriver.py`. + +```bash +# Start two drivers first, e.g.: +# (cd ../rose-game-ai && python main.py --driver mydriver.py --port 8081) +# (cd ../rose-game-ai && python main.py --driver examples/driver.py --port 8082) + +python simulate.py \ + --drivers http://127.0.0.1:8081 http://127.0.0.1:8082 \ + --games 50 \ + --track random \ + --output batch_stats.json +``` + +`batch_stats.json` contains per-driver `wins`/`losses`/`ties`/`avg_score`, plus a +`per_game` breakdown: + +```json +{ + "games": 50, + "track_type": "random", + "drivers": ["http://127.0.0.1:8081", "http://127.0.0.1:8082"], + "results": { + "DriverA": {"wins": 27, "losses": 21, "ties": 2, "avg_score": 612.4}, + "DriverB": {"wins": 21, "losses": 27, "ties": 2, "avg_score": 588.9} + }, + "per_game": [{"scores": {"DriverA": 620, "DriverB": 590}, "winner": "DriverA"}] +} +``` + +Run `python simulate.py --help` for all options. diff --git a/rose/engine/logic.py b/rose/engine/logic.py index 9f92aba..b31b552 100644 --- a/rose/engine/logic.py +++ b/rose/engine/logic.py @@ -14,9 +14,15 @@ async def initialize_game(state): - """Reset game settings and return re-initialized track and players.""" + """Reset game settings and return re-initialized track and players. + + Does NOT touch state["running"] — the game loop only polls the reset + flag periodically, so clobbering running here would silently drop a + Run request issued shortly after Reset, before this had a chance to run. + Callers that want reset to also stop the game must set running=0 + themselves, atomically with the reset request. + """ state["reset"] = None - state["running"] = 0 state["timeleft"] = config.game_duration track = initialize_track(state["track_type"] != "same") players = await initialize_players(state["drivers"]) @@ -74,13 +80,23 @@ async def initialize_players(drivers): return players -async def game_loop(state, active_websockets): +def determine_winner(players): + """Return the name of the highest-scoring player, or None if tied/empty.""" + if not players: + return None + best_score = max(player.score for player in players) + leaders = [player.name for player in players if player.score == best_score] + return leaders[0] if len(leaders) == 1 else None + + +async def game_loop(state, active_websockets, telemetry=None): """ Asynchronously execute the game loop, using provided state and active websockets. Args: state (dict): Dictionary containing game state data (rate, running status, time left, etc.). active_websockets (set): A set of active websocket connections for communication. + telemetry (TelemetryObserver, optional): observer notified on each tick. Defaults to None. Returns: None @@ -103,7 +119,7 @@ async def game_loop(state, active_websockets): if state["running"] == 1: # Start executing a step in the game task = asyncio.create_task( - game_step(state, players, track, active_websockets) + game_step(state, players, track, active_websockets, telemetry) ) # Pause the game loop for a specified duration, based on the rate defined in the state @@ -121,7 +137,33 @@ async def game_loop(state, active_websockets): await asyncio.sleep(1) -async def game_step(state, players, track, active_websockets): +async def play_tick(players, track, telemetry=None, step_index=None): + """ + Execute a single tick of game logic: fetch drivers' actions, advance the + track, and score the resulting player actions. Shared by the live + websocket-driven game loop and headless batch simulation. + + Args: + players (list): List of Player objects. + track (Track): the game track. + telemetry (TelemetryObserver, optional): observer notified after scoring. + step_index (int, optional): tick counter passed through to the observer. + """ + + # Fetch players actions using an asynchronous HTTP session + await net.fetch_drivers_actions(players, track.matrix()) + + # Update track + track.update() + + # Process the actions of the players + score.process(players, track) + + if telemetry is not None: + telemetry.on_step(step_index, players, track) + + +async def game_step(state, players, track, active_websockets, telemetry=None): """ Execute a game step: Update the track, fetch drivers' actions, process actions, and update websockets. @@ -130,17 +172,13 @@ async def game_step(state, players, track, active_websockets): players (list): List of Player objects. track (Track): the game track. active_websockets (Any): Active websockets for communication (assuming a suitable data structure). + telemetry (TelemetryObserver, optional): observer notified after scoring. """ try: - # Fetch players actions using an asynchronous HTTP session - await net.fetch_drivers_actions(players, track.matrix()) - - # Update track - track.update() - - # Process the actions of the players - score.process(players, track) + await play_tick( + players, track, telemetry, config.game_duration - state["timeleft"] + ) # Send data to all WebSocket connections await net.update_websockets(True, state, players, track, active_websockets) @@ -148,6 +186,13 @@ async def game_step(state, players, track, active_websockets): # Progress the game's timer state["timeleft"] -= 1 + if telemetry is not None and state["timeleft"] < 1: + result = { + "scores": {player.name: player.score for player in players}, + "winner": determine_winner(players), + } + telemetry.on_game_end(players, result) + except asyncio.CancelledError: log.info("Game step was canceled!") raise diff --git a/rose/engine/server.py b/rose/engine/server.py index cb9f09b..1598cad 100644 --- a/rose/engine/server.py +++ b/rose/engine/server.py @@ -1,10 +1,17 @@ +import asyncio import json +import logging +import uuid import aiohttp from aiohttp import web from rose.engine import config from rose.engine import logic +from rose.engine import simulate +from rose.telemetry.sinks import LiveSink + +log = logging.getLogger("server") # Global active_websockets # IMPORTANT - shared with game loop in game.py @@ -14,6 +21,13 @@ # IMPORTANT - shared with game loop in game.py state = {"rate": None, "running": None, "reset": None, "drivers": [], "timeleft": None} +# Global telemetry sink, fed by the game loop, read by the /telemetry route. +telemetry = LiveSink() + +# Global batch-simulation jobs, keyed by job id. +# {"status": "running" | "done" | "error", "result": dict or None, "error": str or None} +simulation_jobs = {} + async def admin_handler(request): """ @@ -47,6 +61,11 @@ async def admin_handler(request): state["reset"] = int(reset) except ValueError: return web.Response(text="Invalid reset provided", status=400) + # Stop the game on reset, unless this same request also asked to run + # (formerly done as a side effect inside initialize_game(), which + # raced with a Run request issued shortly after Reset). + if running is None: + state["running"] = 0 # This expects the drivers to be passed as a comma-separated list in the query param # e.g., ?drivers=http://localhost:8081/drv2,http://driver.com:8090/ @@ -60,6 +79,111 @@ async def admin_handler(request): return web.Response(text=json.dumps(state)) +async def telemetry_handler(request): + """Return recent per-tick history and recently finished game results.""" + return web.json_response(telemetry.snapshot()) + + +async def telemetry_clear_handler(request): + """Clear the displayed tick log and match history.""" + telemetry.clear() + return web.json_response(telemetry.snapshot()) + + +async def simulate_start_handler(request): + """ + Start a batch simulation as a background job. Each round is driven + through the same shared game state/websockets as manual play, so it + plays out live on the main game screen, same as a human clicking + Reset then Run would see, one round after another. + + Expected POST JSON body: + {"drivers": [url1, url2], "games": , "track": "same"|"random"} + + Returns: + {"job_id": } with status 202, so the caller can poll + GET /simulate/{job_id} for progress/results. + """ + try: + payload = await request.json() + except json.JSONDecodeError: + return web.Response(text="Invalid JSON body", status=400) + + drivers = payload.get("drivers") + if not isinstance(drivers, list) or len(drivers) != 2: + return web.Response(text="Exactly 2 driver URLs are required", status=400) + + try: + games = int(payload.get("games", 10)) + except (TypeError, ValueError): + return web.Response(text="Invalid games value", status=400) + + track_type = payload.get("track", "random") + if track_type not in ("same", "random"): + return web.Response(text="track must be 'same' or 'random'", status=400) + + job_id = uuid.uuid4().hex + simulation_jobs[job_id] = {"status": "running", "result": None, "error": None} + asyncio.create_task(_run_simulation_job(job_id, drivers, games, track_type)) + + return web.json_response({"job_id": job_id}, status=202) + + +ROUND_POLL_INTERVAL_S = 0.2 +INTER_ROUND_PAUSE_S = 2 + + +async def _run_live_batch(drivers, games, track_type): + """ + Play `games` rounds back-to-back through the shared live game loop, so + each one is visible on the main game screen (same state/websockets a + human's Reset+Run drives), and collect their results as they finish. + + Relies on the game loop's own reset/running handling (see + logic.game_loop / logic.initialize_game): setting reset=1 and running=1 + together starts a fresh round immediately, and telemetry.on_game_end + fires once it's done, which is what we poll for below. + """ + per_game = [] + for game_index in range(games): + finished_before = telemetry.result_count() + + state["drivers"] = list(drivers) + state["track_type"] = track_type + state["reset"] = 1 + state["running"] = 1 + + while telemetry.result_count() == finished_before: + await asyncio.sleep(ROUND_POLL_INTERVAL_S) + + result = telemetry.latest_result() + per_game.append(result) + log.info("live batch game %d/%d: %s", game_index + 1, games, result) + + if game_index < games - 1: + await asyncio.sleep(INTER_ROUND_PAUSE_S) + + return simulate._aggregate(drivers, games, track_type, per_game) + + +async def _run_simulation_job(job_id, drivers, games, track_type): + """Run a batch simulation and store its outcome for simulate_status_handler.""" + try: + result = await _run_live_batch(drivers, games, track_type) + simulation_jobs[job_id] = {"status": "done", "result": result, "error": None} + except Exception as e: + simulation_jobs[job_id] = {"status": "error", "result": None, "error": str(e)} + + +async def simulate_status_handler(request): + """Return the status/result of a batch-simulation job started via POST /simulate.""" + job = simulation_jobs.get(request.match_info["job_id"]) + if job is None: + return web.Response(text="Unknown job id", status=404) + + return web.json_response(job) + + async def websocket_handler(request): """ Handle WebSocket connections, echoing received messages with a prefix. @@ -123,6 +247,10 @@ async def run( # Add application routes app.router.add_get("/ws", websocket_handler) app.router.add_post("/admin", admin_handler) + app.router.add_get("/telemetry", telemetry_handler) + app.router.add_post("/telemetry/clear", telemetry_clear_handler) + app.router.add_post("/simulate", simulate_start_handler) + app.router.add_get("/simulate/{job_id}", simulate_status_handler) runner = aiohttp.web.AppRunner(app) await runner.setup() @@ -138,4 +266,4 @@ async def run( # Start game loop # IMPORTANT: state and active_websockets are references, changes in this file will affect the game loop. - await logic.game_loop(state, active_websockets) + await logic.game_loop(state, active_websockets, telemetry) diff --git a/rose/engine/simulate.py b/rose/engine/simulate.py new file mode 100644 index 0000000..4915522 --- /dev/null +++ b/rose/engine/simulate.py @@ -0,0 +1,108 @@ +"""Headless batch simulation. + +Runs full games against real rose-game-ai driver HTTP servers (the same +GET/POST contract used by the live engine, see rose.engine.net), but without +websockets and without the wall-clock throttle used by the live game loop, so +games run back-to-back as fast as the drivers respond. Useful for automated +evaluation of drivers (e.g. CI, tournaments) and for feeding a telemetry sink. +""" + +import logging + +from rose.engine import config +from rose.engine import logic +from rose.engine.track import Track + +log = logging.getLogger("simulate") + + +async def run_single_game(drivers, track_type, telemetry=None): + """ + Play one full headless game between the given drivers. + + Args: + drivers (list of str): driver URLs, one per player. + track_type (str): "same" or "random", see Track. + telemetry (TelemetryObserver, optional): observer notified per tick. + + Returns: + dict: {"scores": {driver_name: score, ...}, "winner": name_or_None}. + + Raises: + RuntimeError: if any driver failed to respond during initialization. + """ + track = Track(track_type != "same") + track.reset() + + players = await logic.initialize_players(drivers) + if len(players) != len(drivers): + raise RuntimeError( + f"Only {len(players)}/{len(drivers)} drivers responded to initialization" + ) + + if telemetry is not None: + telemetry.on_game_start(track, players, {"track_type": track_type}) + + for step_index in range(config.game_duration): + await logic.play_tick(players, track, telemetry, step_index) + + result = { + "scores": {player.name: player.score for player in players}, + "winner": logic.determine_winner(players), + } + + if telemetry is not None: + telemetry.on_game_end(players, result) + + return result + + +async def run_batch(drivers, games, track_type, telemetry=None): + """ + Play `games` consecutive headless games between the given drivers and + aggregate win/loss/tie stats per driver. + + Games run sequentially: rose-game-ai's reference server is a blocking + socketserver.TCPServer, so concurrent games would just queue on it anyway. + + Returns: + dict: aggregated stats, see rose.engine.simulate module docstring + for shape, or the README for an example. + """ + per_game = [] + for game_index in range(games): + result = await run_single_game(drivers, track_type, telemetry) + per_game.append(result) + log.info("game %d/%d: %s", game_index + 1, games, result) + + return _aggregate(drivers, games, track_type, per_game) + + +def _aggregate(drivers, games, track_type, per_game): + names = sorted({name for result in per_game for name in result["scores"]}) + results = { + name: {"wins": 0, "losses": 0, "ties": 0, "avg_score": 0.0} for name in names + } + + for result in per_game: + winner = result["winner"] + for name, score in result["scores"].items(): + results[name]["avg_score"] += score + if winner is None: + results[name]["ties"] += 1 + elif name == winner: + results[name]["wins"] += 1 + else: + results[name]["losses"] += 1 + + for name in names: + if games: + results[name]["avg_score"] = round(results[name]["avg_score"] / games, 2) + + return { + "games": games, + "track_type": track_type, + "drivers": list(drivers), + "results": results, + "per_game": per_game, + } diff --git a/rose/engine/test_server.py b/rose/engine/test_server.py new file mode 100644 index 0000000..9debc04 --- /dev/null +++ b/rose/engine/test_server.py @@ -0,0 +1,129 @@ +import asyncio + +from aiohttp import web +from aiohttp.test_utils import TestClient +from aiohttp.test_utils import TestServer + +from rose.common import actions +from rose.engine import config +from rose.engine import server +from rose.telemetry.sinks import LiveSink + + +class FakePlayer: + def __init__(self, name, score): + self.name = name + self.score = score + + def state(self): + return {"name": self.name, "score": self.score} + + +def build_app(): + app = web.Application() + app.router.add_get("/telemetry", server.telemetry_handler) + app.router.add_post("/simulate", server.simulate_start_handler) + app.router.add_get("/simulate/{job_id}", server.simulate_status_handler) + return app + + +def make_driver_app(name, action_fn): + async def get_handler(request): + return web.json_response({"info": {"name": name}}) + + async def post_handler(request): + payload = await request.json() + return web.json_response({"info": {"name": name, "action": action_fn(payload)}}) + + driver_app = web.Application() + driver_app.router.add_get("/", get_handler) + driver_app.router.add_post("/", post_handler) + return driver_app + + +class FakeDriver: + def __init__(self, name, action_fn): + self._server = TestServer(make_driver_app(name, action_fn)) + + async def __aenter__(self): + await self._server.start_server() + return str(self._server.make_url("/")) + + async def __aexit__(self, *exc_info): + await self._server.close() + + +def test_telemetry_handler_returns_recent_history_and_results(monkeypatch): + sink = LiveSink() + sink.on_step(0, [FakePlayer("A", 10), FakePlayer("B", 5)], track=None) + sink.on_game_end( + players=None, result={"scores": {"A": 100, "B": 90}, "winner": "A"} + ) + monkeypatch.setattr(server, "telemetry", sink) + + async def scenario(): + async with TestClient(TestServer(build_app())) as client: + resp = await client.get("/telemetry") + return resp.status, await resp.json() + + status, body = asyncio.run(scenario()) + + assert status == 200 + assert len(body["history"]) == 1 + assert body["history"][0]["players"] == [ + {"name": "A", "score": 10}, + {"name": "B", "score": 5}, + ] + assert body["recent_results"] == [{"scores": {"A": 100, "B": 90}, "winner": "A"}] + + +def test_simulate_endpoint_rejects_bad_payloads(): + async def scenario(): + async with TestClient(TestServer(build_app())) as client: + resp = await client.post("/simulate", json={"drivers": ["only-one"]}) + return resp.status + + assert asyncio.run(scenario()) == 400 + + +def test_simulate_status_unknown_job_is_404(): + async def scenario(): + async with TestClient(TestServer(build_app())) as client: + resp = await client.get("/simulate/does-not-exist") + return resp.status + + assert asyncio.run(scenario()) == 404 + + +def test_simulate_job_runs_in_background_and_completes(monkeypatch): + monkeypatch.setattr(config, "game_duration", 4) + + async def scenario(): + async with FakeDriver( + "DriverA", lambda payload: actions.NONE + ) as url_a, FakeDriver( + "DriverB", lambda payload: actions.NONE + ) as url_b, TestClient( + TestServer(build_app()) + ) as client: + start_resp = await client.post( + "/simulate", + json={"drivers": [url_a, url_b], "games": 2, "track": "same"}, + ) + assert start_resp.status == 202 + job_id = (await start_resp.json())["job_id"] + + for _ in range(200): + status_resp = await client.get(f"/simulate/{job_id}") + body = await status_resp.json() + if body["status"] != "running": + return body + await asyncio.sleep(0.01) + + raise AssertionError("job never finished") + + job = asyncio.run(scenario()) + + assert job["status"] == "done" + assert job["result"]["games"] == 2 + assert set(job["result"]["results"].keys()) == {"DriverA", "DriverB"} diff --git a/rose/engine/test_simulate.py b/rose/engine/test_simulate.py new file mode 100644 index 0000000..6c83bfd --- /dev/null +++ b/rose/engine/test_simulate.py @@ -0,0 +1,115 @@ +import asyncio +import random + +import pytest +from aiohttp import web +from aiohttp.test_utils import TestServer + +from rose.common import actions +from rose.common import obstacles +from rose.engine import config +from rose.engine import simulate + + +def make_driver_app(name, action_fn): + """A minimal stand-in for a rose-game-ai driver HTTP server.""" + + async def get_handler(request): + return web.json_response({"info": {"name": name}}) + + async def post_handler(request): + payload = await request.json() + return web.json_response({"info": {"name": name, "action": action_fn(payload)}}) + + app = web.Application() + app.router.add_get("/", get_handler) + app.router.add_post("/", post_handler) + return app + + +class FakeDriver: + """Runs a driver app on an ephemeral local port for the duration of a `with` block.""" + + def __init__(self, name, action_fn): + self._server = TestServer(make_driver_app(name, action_fn)) + + async def __aenter__(self): + await self._server.start_server() + return str(self._server.make_url("/")) + + async def __aexit__(self, *exc_info): + await self._server.close() + + +def test_run_single_game_ties_when_drivers_behave_identically(monkeypatch): + monkeypatch.setattr(config, "game_duration", 6) + + async def scenario(): + async with FakeDriver( + "DriverA", lambda payload: actions.NONE + ) as url_a, FakeDriver("DriverB", lambda payload: actions.NONE) as url_b: + return await simulate.run_single_game([url_a, url_b], track_type="same") + + result = asyncio.run(scenario()) + + assert result["winner"] is None + assert result["scores"]["DriverA"] == result["scores"]["DriverB"] + + +def test_run_single_game_determines_a_winner(monkeypatch): + # Pin the obstacle stream so the outcome doesn't depend on real randomness: + # every generated row is a CRACK placed exactly where both (stationary) + # players sit in their lane. + monkeypatch.setattr(config, "game_duration", 12) + monkeypatch.setattr(obstacles, "get_random_obstacle", lambda: obstacles.CRACK) + monkeypatch.setattr(random, "choice", lambda seq: 1) + + async def scenario(): + async with FakeDriver( + "DriverA", lambda payload: actions.JUMP + ) as url_a, FakeDriver("DriverB", lambda payload: actions.NONE) as url_b: + return await simulate.run_single_game([url_a, url_b], track_type="same") + + result = asyncio.run(scenario()) + + assert result["winner"] == "DriverA" + assert result["scores"]["DriverA"] > result["scores"]["DriverB"] + + +def test_run_single_game_raises_if_a_driver_does_not_respond(): + async def scenario(): + async with FakeDriver("DriverA", lambda payload: actions.NONE) as url_a: + return await simulate.run_single_game( + [url_a, "http://127.0.0.1:1/unreachable"], track_type="same" + ) + + with pytest.raises(RuntimeError): + asyncio.run(scenario()) + + +def test_run_batch_aggregates_win_loss_tie_stats(monkeypatch): + monkeypatch.setattr(config, "game_duration", 5) + + async def scenario(): + async with FakeDriver( + "DriverA", lambda payload: actions.NONE + ) as url_a, FakeDriver("DriverB", lambda payload: actions.NONE) as url_b: + stats = await simulate.run_batch([url_a, url_b], games=3, track_type="same") + return stats, url_a, url_b + + stats, url_a, url_b = asyncio.run(scenario()) + + assert stats["games"] == 3 + assert stats["track_type"] == "same" + assert stats["drivers"] == [url_a, url_b] + assert len(stats["per_game"]) == 3 + + for name in ("DriverA", "DriverB"): + assert stats["results"][name]["ties"] == 3 + assert stats["results"][name]["wins"] == 0 + assert stats["results"][name]["losses"] == 0 + + assert ( + stats["results"]["DriverA"]["avg_score"] + == stats["results"]["DriverB"]["avg_score"] + ) diff --git a/rose/telemetry/__init__.py b/rose/telemetry/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/rose/telemetry/observer.py b/rose/telemetry/observer.py new file mode 100644 index 0000000..69cb058 --- /dev/null +++ b/rose/telemetry/observer.py @@ -0,0 +1,60 @@ +"""Telemetry observer interface for the game engine. + +An observer is notified at well-defined points in the game lifecycle +(game start, each tick, game end). It is purely additive: nothing in +the engine requires an observer to be present, and the default +NullObserver is a no-op so existing call sites are unaffected. +""" + + +class TelemetryObserver(object): + """Base class for telemetry observers. Override any subset of hooks.""" + + def on_game_start(self, track, players, meta=None): + """Called once, right after players/track are initialized. + + Args: + track (Track): the initialized game track. + players (list of Player): the initialized players. + meta (dict, optional): extra context (e.g. track_type). + """ + + def on_step(self, step_index, players, track): + """Called once per tick, right after score.process() has run. + + Args: + step_index (int): 0-based tick counter for this game. + players (list of Player): players with up-to-date state. + track (Track): track with up-to-date state. + """ + + def on_game_end(self, players, result): + """Called once, after the last tick of a game. + + Args: + players (list of Player): players in their final state. + result (dict): {"scores": {name: score}, "winner": name_or_None}. + """ + + +class NullObserver(TelemetryObserver): + """No-op observer, used as the implicit default everywhere.""" + + +class CompositeObserver(TelemetryObserver): + """Fans out every hook to a list of child observers.""" + + def __init__(self, observers): + self._observers = list(observers) + + def on_game_start(self, track, players, meta=None): + for observer in self._observers: + observer.on_game_start(track, players, meta) + + def on_step(self, step_index, players, track): + for observer in self._observers: + observer.on_step(step_index, players, track) + + def on_game_end(self, players, result): + for observer in self._observers: + observer.on_game_end(players, result) diff --git a/rose/telemetry/sinks.py b/rose/telemetry/sinks.py new file mode 100644 index 0000000..d455b70 --- /dev/null +++ b/rose/telemetry/sinks.py @@ -0,0 +1,127 @@ +"""Concrete TelemetryObserver sinks.""" + +import collections +import json + +from rose.telemetry.observer import TelemetryObserver + + +class InMemorySink(TelemetryObserver): + """Keeps per-step snapshots and the final result in memory. + + Handy for batch simulation, where the run is over long before anyone + could tail a file, and for tests. + """ + + def __init__(self): + self.games = [] + self._current_game = None + + def on_game_start(self, track, players, meta=None): + self._current_game = { + "meta": meta or {}, + "steps": [], + "result": None, + } + self.games.append(self._current_game) + + def on_step(self, step_index, players, track): + self._current_game["steps"].append( + { + "step": step_index, + "players": [player.state() for player in players], + "track": track.state(), + } + ) + + def on_game_end(self, players, result): + self._current_game["result"] = result + + +class LiveSink(TelemetryObserver): + """Keeps a bounded in-memory history for the live, running game server. + + Unlike InMemorySink, this isn't grouped per-game: it's a flat rolling log + of recent ticks plus a rolling log of recently finished games, meant to be + polled over HTTP (see rose.engine.server's `/telemetry` route) by a + dashboard that just wants "what's been happening lately". + """ + + def __init__(self, max_ticks=500, max_results=20): + self._history = collections.deque(maxlen=max_ticks) + self._recent_results = collections.deque(maxlen=max_results) + self._total_finished = 0 + + def on_step(self, step_index, players, track): + self._history.append( + { + "step": step_index, + "players": [player.state() for player in players], + } + ) + + def on_game_end(self, players, result): + self._recent_results.append(result) + self._total_finished += 1 + + def result_count(self): + """Total number of finished games recorded so far (monotonic). + + Deliberately backed by a counter, not len(_recent_results) — that + deque is capped at max_results, so its length would plateau and + break callers polling for "has another game finished" past that cap. + """ + return self._total_finished + + def latest_result(self): + """Most recently finished game's result, or None if none yet.""" + return self._recent_results[-1] if self._recent_results else None + + def clear(self): + """Clear the displayed tick log and match history. + + Deliberately leaves _total_finished untouched: it's used by + server._run_live_batch to detect "has another round finished", + and resetting it here would break any batch job in flight. + """ + self._history.clear() + self._recent_results.clear() + + def snapshot(self): + """Return a JSON-serializable view of recent ticks and results.""" + return { + "history": list(self._history), + "recent_results": list(self._recent_results), + } + + +class JSONLSink(TelemetryObserver): + """Appends one JSON object per event to a file, one event per line. + + This is a natural feed for a future dashboard to tail (`tail -f` or a + simple polling reader), without the dashboard needing to understand + engine internals beyond the JSON shape written here. + """ + + def __init__(self, path): + self._path = path + + def on_game_start(self, track, players, meta=None): + self._append({"event": "game_start", "meta": meta or {}}) + + def on_step(self, step_index, players, track): + self._append( + { + "event": "step", + "step": step_index, + "players": [player.state() for player in players], + "track": track.state(), + } + ) + + def on_game_end(self, players, result): + self._append({"event": "game_end", "result": result}) + + def _append(self, record): + with open(self._path, "a") as f: + f.write(json.dumps(record) + "\n") diff --git a/rose/telemetry/test_observer.py b/rose/telemetry/test_observer.py new file mode 100644 index 0000000..1a20ca7 --- /dev/null +++ b/rose/telemetry/test_observer.py @@ -0,0 +1,73 @@ +from rose.telemetry.observer import CompositeObserver +from rose.telemetry.observer import NullObserver +from rose.telemetry.observer import TelemetryObserver +from rose.telemetry.sinks import InMemorySink + + +def test_null_observer_is_a_noop(): + observer = NullObserver() + + # None of these should raise, regardless of arguments. + observer.on_game_start(track=None, players=None) + observer.on_step(step_index=0, players=None, track=None) + observer.on_game_end(players=None, result=None) + + +def test_in_memory_sink_records_game_lifecycle(): + sink = InMemorySink() + + class FakePlayer: + def __init__(self, name, score): + self.name = name + self.score = score + + def state(self): + return {"name": self.name, "score": self.score} + + class FakeTrack: + def state(self): + return [] + + players = [FakePlayer("A", 0), FakePlayer("B", 0)] + track = FakeTrack() + + sink.on_game_start(track, players, {"track_type": "same"}) + sink.on_step(0, players, track) + sink.on_step(1, players, track) + result = {"scores": {"A": 10, "B": 5}, "winner": "A"} + sink.on_game_end(players, result) + + assert len(sink.games) == 1 + game = sink.games[0] + assert game["meta"] == {"track_type": "same"} + assert len(game["steps"]) == 2 + assert game["steps"][0]["step"] == 0 + assert game["result"] == result + + +def test_in_memory_sink_tracks_multiple_games_independently(): + sink = InMemorySink() + + sink.on_game_start(track=None, players=[]) + sink.on_game_end(players=[], result={"scores": {}, "winner": None}) + + sink.on_game_start(track=None, players=[]) + sink.on_game_end(players=[], result={"scores": {}, "winner": None}) + + assert len(sink.games) == 2 + + +def test_composite_observer_fans_out_to_children(): + calls = [] + + class RecordingObserver(TelemetryObserver): + def __init__(self, tag): + self.tag = tag + + def on_step(self, step_index, players, track): + calls.append((self.tag, step_index)) + + composite = CompositeObserver([RecordingObserver("a"), RecordingObserver("b")]) + composite.on_step(3, players=None, track=None) + + assert calls == [("a", 3), ("b", 3)] diff --git a/simulate.py b/simulate.py new file mode 100644 index 0000000..91c18db --- /dev/null +++ b/simulate.py @@ -0,0 +1,55 @@ +import argparse +import asyncio +import json +import logging + +from rose.engine import simulate + + +def main(): + parser = argparse.ArgumentParser( + description="Run headless batch simulations of the engine against rose-game-ai drivers." + ) + parser.add_argument( + "-d", + "--drivers", + nargs=2, + required=True, + metavar=("DRIVER1_URL", "DRIVER2_URL"), + help="Exactly two driver URLs to simulate (matches the game's 2-player design).", + ) + parser.add_argument( + "-g", "--games", type=int, default=10, help="Number of games to simulate." + ) + parser.add_argument( + "-t", + "--track", + choices=["same", "random"], + default="random", + help="Choose the track type. Can be 'same' or 'random'.", + ) + parser.add_argument( + "-o", + "--output", + default="batch_stats.json", + help="Path to write the aggregated JSON stats to.", + ) + parser.add_argument( + "--log", default="WARNING", help="Set the logging level. E.g. --log DEBUG" + ) + + args = parser.parse_args() + + logging.basicConfig(level=getattr(logging, args.log.upper())) + + stats = asyncio.run(simulate.run_batch(args.drivers, args.games, args.track)) + + with open(args.output, "w") as f: + json.dump(stats, f, indent=2) + + print(f"Simulated {args.games} games -> {args.output}") + print(json.dumps(stats["results"], indent=2)) + + +if __name__ == "__main__": + main() From 386e2282dd6959f909e5f540df02f0cac327a3f3 Mon Sep 17 00:00:00 2001 From: Daniel Fourman Date: Wed, 15 Jul 2026 13:20:59 +0300 Subject: [PATCH 2/8] Add coin object with a one-time 5-coin jackpot bonus Coins are worthless individually; collecting exactly config.coins_for_jackpot (5) awards a one-time config.score_jackpot (50) bonus. Coins collected beyond that grant no further bonus. Co-Authored-By: Claude Sonnet 5 --- rose/common/obstacles.py | 3 +- rose/common/test_obstacles.py | 4 ++- rose/engine/config.py | 3 ++ rose/engine/player.py | 3 ++ rose/engine/score.py | 27 ++++++++++++++ rose/engine/test_player.py | 1 + rose/engine/test_score.py | 66 +++++++++++++++++++++++++++++++++++ 7 files changed, 105 insertions(+), 2 deletions(-) diff --git a/rose/common/obstacles.py b/rose/common/obstacles.py index 3b84f9c..2174eb5 100644 --- a/rose/common/obstacles.py +++ b/rose/common/obstacles.py @@ -9,8 +9,9 @@ BIKE = "bike" # NOQA WATER = "water" # NOQA BARRIER = "barrier" # NOQA +COIN = "coin" # NOQA -ALL = (NONE, CRACK, TRASH, PENGUIN, BIKE, WATER, BARRIER) +ALL = (NONE, CRACK, TRASH, PENGUIN, BIKE, WATER, BARRIER, COIN) def get_random_obstacle(): diff --git a/rose/common/test_obstacles.py b/rose/common/test_obstacles.py index 1b5d688..9ce6e0c 100644 --- a/rose/common/test_obstacles.py +++ b/rose/common/test_obstacles.py @@ -6,6 +6,7 @@ BIKE, WATER, BARRIER, + COIN, ALL, get_random_obstacle, ) @@ -19,10 +20,11 @@ def test_constants(): assert BIKE == "bike" assert WATER == "water" assert BARRIER == "barrier" + assert COIN == "coin" def test_all_constant(): - assert ALL == (NONE, CRACK, TRASH, PENGUIN, BIKE, WATER, BARRIER) + assert ALL == (NONE, CRACK, TRASH, PENGUIN, BIKE, WATER, BARRIER, COIN) def test_get_random_obstacle(): diff --git a/rose/engine/config.py b/rose/engine/config.py index 38e343d..104ca49 100644 --- a/rose/engine/config.py +++ b/rose/engine/config.py @@ -18,3 +18,6 @@ score_pickup = 10 score_jump = 5 score_brake = 4 + +coins_for_jackpot = 5 +score_jackpot = 50 diff --git a/rose/engine/player.py b/rose/engine/player.py index 208dc0e..8a17799 100644 --- a/rose/engine/player.py +++ b/rose/engine/player.py @@ -34,6 +34,7 @@ def __init__(self, name, car, lane): self.response_time = None self.score = None self.pickups = None + self.coins = None self.misses = None self.hits = None self.breaks = None @@ -48,6 +49,7 @@ def reset(self): self.response_time = None self.score = 0 self.pickups = 0 + self.coins = 0 self.misses = 0 self.hits = 0 self.breaks = 0 @@ -79,6 +81,7 @@ def state(self): "lane": self.lane, "score": self.score, "pickups": self.pickups, + "coins": self.coins, "misses": self.misses, "hits": self.hits, "breaks": self.breaks, diff --git a/rose/engine/score.py b/rose/engine/score.py index 1936432..5860b31 100644 --- a/rose/engine/score.py +++ b/rose/engine/score.py @@ -158,6 +158,33 @@ def process(players, track): log.debug("player %s missed %s", player.name, obstacle) + elif obstacle == obstacles.COIN: + if player.action == actions.PICKUP: + # Move forward and collect a coin. Coins are worth nothing + # on their own; only the coin that completes the jackpot + # tally awards points, and further coins award nothing more. + track.clear(player.x, player.y) + player.coins += 1 + points = config.score_move_forward + if player.coins == config.coins_for_jackpot: + points += config.score_jackpot + player.score += points + player.pickups += 1 + + log.debug( + "player %s picked up %s (%d/%d): got %d points", + player.name, + obstacle, + player.coins, + config.coins_for_jackpot, + points, + ) + else: + # Move forward leaving the obstacle on the track + player.score += config.score_move_forward + + log.debug("player %s missed %s", player.name, obstacle) + # Here we can end the game when player gets out of # the track bounds. For now, just keep the player at the same # location. diff --git a/rose/engine/test_player.py b/rose/engine/test_player.py index 3ddac3b..fdd0e5d 100644 --- a/rose/engine/test_player.py +++ b/rose/engine/test_player.py @@ -80,6 +80,7 @@ def test_player_state(): "lane": 1, "score": 0, "pickups": 0, + "coins": 0, "misses": 0, "hits": 0, "breaks": 0, diff --git a/rose/engine/test_score.py b/rose/engine/test_score.py index 680b7e3..95f7372 100644 --- a/rose/engine/test_score.py +++ b/rose/engine/test_score.py @@ -121,6 +121,72 @@ def test_other(self, action): self.assert_keep_obstacle() +class TestCoin(SinglePlayerTest): + """ + Handling coins + + Coins are worthless on their own. Only collecting exactly + `config.coins_for_jackpot` of them awards a one-time jackpot bonus; + coins collected beyond that award no further bonus. + """ + + obstacle = obstacles.COIN + + def pickup(self): + # track.clear() consumes the coin each tick, so put a fresh one + # back down before every pickup to simulate collecting one per tick. + self.track.set(self.x, self.y, self.obstacle) + self.player.action = actions.PICKUP + self.process() + + def test_pickup_below_jackpot(self): + self.pickup() + assert self.player.coins == 1 + self.assert_score(self.score) + self.assert_remove_obstacle() + + def test_pickup_reaches_jackpot(self): + for _ in range(config.coins_for_jackpot): + self.pickup() + assert self.player.coins == config.coins_for_jackpot + expected = ( + config.coins_for_jackpot * config.score_move_forward + + config.score_jackpot + ) + assert self.player.score == self.score + expected + + def test_pickup_beyond_jackpot_gets_no_extra_bonus(self): + for _ in range(config.coins_for_jackpot + 2): + self.pickup() + assert self.player.coins == config.coins_for_jackpot + 2 + expected = ( + config.coins_for_jackpot + 2 + ) * config.score_move_forward + config.score_jackpot + assert self.player.score == self.score + expected + + def test_right(self): + self.player.action = actions.RIGHT + self.process() + self.assert_move_right() + self.assert_keep_obstacle() + + def test_left(self): + self.player.action = actions.LEFT + self.process() + self.assert_move_left() + self.assert_keep_obstacle() + + @pytest.mark.parametrize( + "action", [a for a in FORWARD_ACTIONS if a != actions.PICKUP] + ) + def test_other(self, action): + self.player.action = action + self.process() + self.assert_score(self.score) + assert self.player.coins == 0 + self.assert_keep_obstacle() + + class MagicActionTest(SinglePlayerTest): """ Handling obstacles with magic action From 36012b358373883d9d1c11fcf61a1872814eb433 Mon Sep 17 00:00:00 2001 From: Daniel Fourman Date: Wed, 15 Jul 2026 15:53:22 +0300 Subject: [PATCH 3/8] Add per-obstacle outcome counters and running/round state for telemetry Splits the blended hits/pickups counters into per-obstacle success/fail/miss data the redesigned telemetry dashboard needs: wall_hits, water_hits, crack_hits, coin_misses, and a finally-wired-up misses (penguin miss was tracked in the field but never incremented). Also exposes a `running` flag and `total_finished` round count on GET/POST /telemetry(/clear), and attaches each player's full end-of-game state to game results so "Recent games" can show a real per-driver stat breakdown instead of just final score. Co-Authored-By: Claude Sonnet 5 --- rose/engine/logic.py | 1 + rose/engine/player.py | 12 ++++++++++++ rose/engine/score.py | 5 +++++ rose/engine/server.py | 10 ++++++++-- rose/engine/simulate.py | 4 +++- rose/engine/test_player.py | 4 ++++ rose/engine/test_score.py | 7 +++++++ rose/telemetry/sinks.py | 1 + 8 files changed, 41 insertions(+), 3 deletions(-) diff --git a/rose/engine/logic.py b/rose/engine/logic.py index b31b552..6b7ca02 100644 --- a/rose/engine/logic.py +++ b/rose/engine/logic.py @@ -190,6 +190,7 @@ async def game_step(state, players, track, active_websockets, telemetry=None): result = { "scores": {player.name: player.score for player in players}, "winner": determine_winner(players), + "players": {player.name: player.state() for player in players}, } telemetry.on_game_end(players, result) diff --git a/rose/engine/player.py b/rose/engine/player.py index 8a17799..3e089a9 100644 --- a/rose/engine/player.py +++ b/rose/engine/player.py @@ -36,7 +36,11 @@ def __init__(self, name, car, lane): self.pickups = None self.coins = None self.misses = None + self.coin_misses = None self.hits = None + self.wall_hits = None + self.water_hits = None + self.crack_hits = None self.breaks = None self.jumps = None self.collisions = None @@ -51,7 +55,11 @@ def reset(self): self.pickups = 0 self.coins = 0 self.misses = 0 + self.coin_misses = 0 self.hits = 0 + self.wall_hits = 0 + self.water_hits = 0 + self.crack_hits = 0 self.breaks = 0 self.collisions = 0 self.jumps = 0 @@ -83,7 +91,11 @@ def state(self): "pickups": self.pickups, "coins": self.coins, "misses": self.misses, + "coin_misses": self.coin_misses, "hits": self.hits, + "wall_hits": self.wall_hits, + "water_hits": self.water_hits, + "crack_hits": self.crack_hits, "breaks": self.breaks, "jumps": self.jumps, "collisions": self.collisions, diff --git a/rose/engine/score.py b/rose/engine/score.py index 5860b31..e4bc9f7 100644 --- a/rose/engine/score.py +++ b/rose/engine/score.py @@ -70,6 +70,7 @@ def process(players, track): player.y += 1 player.score += config.score_move_backward player.hits += 1 + player.wall_hits += 1 log.debug( "player %s hit %s: lost %d points, moved back to %d,%d", @@ -99,6 +100,7 @@ def process(players, track): player.y += 1 player.score += config.score_move_backward player.hits += 1 + player.crack_hits += 1 log.debug( "player %s hit %s: lost %d points, moved back to %d,%d", @@ -128,6 +130,7 @@ def process(players, track): player.y += 1 player.score += config.score_move_backward player.hits += 1 + player.water_hits += 1 log.debug( "player %s hit %s: lost %d points, moved back to %d,%d", @@ -155,6 +158,7 @@ def process(players, track): else: # Move forward leaving the obstacle on the track player.score += config.score_move_forward + player.misses += 1 log.debug("player %s missed %s", player.name, obstacle) @@ -182,6 +186,7 @@ def process(players, track): else: # Move forward leaving the obstacle on the track player.score += config.score_move_forward + player.coin_misses += 1 log.debug("player %s missed %s", player.name, obstacle) diff --git a/rose/engine/server.py b/rose/engine/server.py index 1598cad..0030070 100644 --- a/rose/engine/server.py +++ b/rose/engine/server.py @@ -79,15 +79,21 @@ async def admin_handler(request): return web.Response(text=json.dumps(state)) +def _telemetry_snapshot(): + snapshot = telemetry.snapshot() + snapshot["running"] = bool(state.get("running")) + return snapshot + + async def telemetry_handler(request): """Return recent per-tick history and recently finished game results.""" - return web.json_response(telemetry.snapshot()) + return web.json_response(_telemetry_snapshot()) async def telemetry_clear_handler(request): """Clear the displayed tick log and match history.""" telemetry.clear() - return web.json_response(telemetry.snapshot()) + return web.json_response(_telemetry_snapshot()) async def simulate_start_handler(request): diff --git a/rose/engine/simulate.py b/rose/engine/simulate.py index 4915522..5fcfd36 100644 --- a/rose/engine/simulate.py +++ b/rose/engine/simulate.py @@ -26,7 +26,8 @@ async def run_single_game(drivers, track_type, telemetry=None): telemetry (TelemetryObserver, optional): observer notified per tick. Returns: - dict: {"scores": {driver_name: score, ...}, "winner": name_or_None}. + dict: {"scores": {driver_name: score, ...}, "winner": name_or_None, + "players": {driver_name: Player.state(), ...}}. Raises: RuntimeError: if any driver failed to respond during initialization. @@ -49,6 +50,7 @@ async def run_single_game(drivers, track_type, telemetry=None): result = { "scores": {player.name: player.score for player in players}, "winner": logic.determine_winner(players), + "players": {player.name: player.state() for player in players}, } if telemetry is not None: diff --git a/rose/engine/test_player.py b/rose/engine/test_player.py index fdd0e5d..19c0e7d 100644 --- a/rose/engine/test_player.py +++ b/rose/engine/test_player.py @@ -82,7 +82,11 @@ def test_player_state(): "pickups": 0, "coins": 0, "misses": 0, + "coin_misses": 0, "hits": 0, + "wall_hits": 0, + "water_hits": 0, + "crack_hits": 0, "breaks": 0, "jumps": 0, "collisions": 0, diff --git a/rose/engine/test_score.py b/rose/engine/test_score.py index 95f7372..87b7b38 100644 --- a/rose/engine/test_score.py +++ b/rose/engine/test_score.py @@ -119,6 +119,7 @@ def test_other(self, action): self.process() self.assert_score(self.score) self.assert_keep_obstacle() + assert self.player.misses == 1 class TestCoin(SinglePlayerTest): @@ -184,6 +185,7 @@ def test_other(self, action): self.process() self.assert_score(self.score) assert self.player.coins == 0 + assert self.player.coin_misses == 1 self.assert_keep_obstacle() @@ -198,6 +200,7 @@ class MagicActionTest(SinglePlayerTest): # Must be defined in subclass action = None magic_score = None + hit_field = None @pytest.mark.parametrize("action", FORWARD_ACTIONS) def test_forward(self, action): @@ -209,6 +212,7 @@ def test_forward(self, action): else: self.assert_move_back() self.assert_remove_obstacle() + assert getattr(self.player, self.hit_field) == 1 def test_right(self): self.player.action = actions.RIGHT @@ -227,12 +231,14 @@ class TestCrack(MagicActionTest): magic_score = config.score_jump obstacle = obstacles.CRACK action = actions.JUMP + hit_field = "crack_hits" class TestWater(MagicActionTest): magic_score = config.score_brake obstacle = obstacles.WATER action = actions.BRAKE + hit_field = "water_hits" class TurnTest(SinglePlayerTest): @@ -261,6 +267,7 @@ def test_other(self, action): # TODO: decrease points on redundant action? self.assert_move_back_no_punish() self.assert_remove_obstacle() + assert self.player.wall_hits == 1 class TestTrash(TurnTest): diff --git a/rose/telemetry/sinks.py b/rose/telemetry/sinks.py index d455b70..bfff198 100644 --- a/rose/telemetry/sinks.py +++ b/rose/telemetry/sinks.py @@ -92,6 +92,7 @@ def snapshot(self): return { "history": list(self._history), "recent_results": list(self._recent_results), + "total_finished": self._total_finished, } From 196cb94dbb55b97fd4c5426e6b783660b1926021 Mon Sep 17 00:00:00 2001 From: Daniel Fourman Date: Wed, 15 Jul 2026 16:22:49 +0300 Subject: [PATCH 4/8] Track wall-clock round duration for the telemetry summary table Adds duration_seconds to the game-end result (both the live loop and the headless simulate.py path), timed from round start to round end. Meaningful mainly for the headless/automation path, where rounds run back-to-back as fast as the drivers respond rather than on a fixed wall-clock cadence. Co-Authored-By: Claude Sonnet 5 --- rose/engine/logic.py | 5 +++++ rose/engine/simulate.py | 6 +++++- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/rose/engine/logic.py b/rose/engine/logic.py index 6b7ca02..296a8ee 100644 --- a/rose/engine/logic.py +++ b/rose/engine/logic.py @@ -1,6 +1,7 @@ import asyncio import logging import random +import time import aiohttp from rose.engine import config @@ -24,6 +25,7 @@ async def initialize_game(state): """ state["reset"] = None state["timeleft"] = config.game_duration + state["game_started_at"] = time.monotonic() track = initialize_track(state["track_type"] != "same") players = await initialize_players(state["drivers"]) return track, players @@ -191,6 +193,9 @@ async def game_step(state, players, track, active_websockets, telemetry=None): "scores": {player.name: player.score for player in players}, "winner": determine_winner(players), "players": {player.name: player.state() for player in players}, + "duration_seconds": round( + time.monotonic() - state["game_started_at"], 2 + ), } telemetry.on_game_end(players, result) diff --git a/rose/engine/simulate.py b/rose/engine/simulate.py index 5fcfd36..97543b5 100644 --- a/rose/engine/simulate.py +++ b/rose/engine/simulate.py @@ -8,6 +8,7 @@ """ import logging +import time from rose.engine import config from rose.engine import logic @@ -27,7 +28,8 @@ async def run_single_game(drivers, track_type, telemetry=None): Returns: dict: {"scores": {driver_name: score, ...}, "winner": name_or_None, - "players": {driver_name: Player.state(), ...}}. + "players": {driver_name: Player.state(), ...}, + "duration_seconds": float}. Raises: RuntimeError: if any driver failed to respond during initialization. @@ -44,6 +46,7 @@ async def run_single_game(drivers, track_type, telemetry=None): if telemetry is not None: telemetry.on_game_start(track, players, {"track_type": track_type}) + started_at = time.monotonic() for step_index in range(config.game_duration): await logic.play_tick(players, track, telemetry, step_index) @@ -51,6 +54,7 @@ async def run_single_game(drivers, track_type, telemetry=None): "scores": {player.name: player.score for player in players}, "winner": logic.determine_winner(players), "players": {player.name: player.state() for player in players}, + "duration_seconds": round(time.monotonic() - started_at, 2), } if telemetry is not None: From f8b77c5e4116001c869a04ec14a718c10ba87d0a Mon Sep 17 00:00:00 2001 From: Daniel Fourman Date: Wed, 15 Jul 2026 16:38:59 +0300 Subject: [PATCH 5/8] Remove the coin/jackpot feature entirely Coins are gone from the game, matching upstream RedHat-Israel/rose-game-engine: no COIN obstacle, no coins_for_jackpot/score_jackpot config, no coin/coin_misses player counters, no coin scoring branch. Kept everything else added since (wall_hits/water_hits/crack_hits/misses fix, running flag, duration_seconds, players-in-result) since none of that is coin-specific. Co-Authored-By: Claude Sonnet 5 --- rose/common/obstacles.py | 3 +- rose/common/test_obstacles.py | 4 +-- rose/engine/config.py | 3 -- rose/engine/player.py | 6 ---- rose/engine/score.py | 28 --------------- rose/engine/test_player.py | 2 -- rose/engine/test_score.py | 67 ----------------------------------- 7 files changed, 2 insertions(+), 111 deletions(-) diff --git a/rose/common/obstacles.py b/rose/common/obstacles.py index 2174eb5..3b84f9c 100644 --- a/rose/common/obstacles.py +++ b/rose/common/obstacles.py @@ -9,9 +9,8 @@ BIKE = "bike" # NOQA WATER = "water" # NOQA BARRIER = "barrier" # NOQA -COIN = "coin" # NOQA -ALL = (NONE, CRACK, TRASH, PENGUIN, BIKE, WATER, BARRIER, COIN) +ALL = (NONE, CRACK, TRASH, PENGUIN, BIKE, WATER, BARRIER) def get_random_obstacle(): diff --git a/rose/common/test_obstacles.py b/rose/common/test_obstacles.py index 9ce6e0c..1b5d688 100644 --- a/rose/common/test_obstacles.py +++ b/rose/common/test_obstacles.py @@ -6,7 +6,6 @@ BIKE, WATER, BARRIER, - COIN, ALL, get_random_obstacle, ) @@ -20,11 +19,10 @@ def test_constants(): assert BIKE == "bike" assert WATER == "water" assert BARRIER == "barrier" - assert COIN == "coin" def test_all_constant(): - assert ALL == (NONE, CRACK, TRASH, PENGUIN, BIKE, WATER, BARRIER, COIN) + assert ALL == (NONE, CRACK, TRASH, PENGUIN, BIKE, WATER, BARRIER) def test_get_random_obstacle(): diff --git a/rose/engine/config.py b/rose/engine/config.py index 104ca49..38e343d 100644 --- a/rose/engine/config.py +++ b/rose/engine/config.py @@ -18,6 +18,3 @@ score_pickup = 10 score_jump = 5 score_brake = 4 - -coins_for_jackpot = 5 -score_jackpot = 50 diff --git a/rose/engine/player.py b/rose/engine/player.py index 3e089a9..3ccae82 100644 --- a/rose/engine/player.py +++ b/rose/engine/player.py @@ -34,9 +34,7 @@ def __init__(self, name, car, lane): self.response_time = None self.score = None self.pickups = None - self.coins = None self.misses = None - self.coin_misses = None self.hits = None self.wall_hits = None self.water_hits = None @@ -53,9 +51,7 @@ def reset(self): self.response_time = None self.score = 0 self.pickups = 0 - self.coins = 0 self.misses = 0 - self.coin_misses = 0 self.hits = 0 self.wall_hits = 0 self.water_hits = 0 @@ -89,9 +85,7 @@ def state(self): "lane": self.lane, "score": self.score, "pickups": self.pickups, - "coins": self.coins, "misses": self.misses, - "coin_misses": self.coin_misses, "hits": self.hits, "wall_hits": self.wall_hits, "water_hits": self.water_hits, diff --git a/rose/engine/score.py b/rose/engine/score.py index e4bc9f7..10889e5 100644 --- a/rose/engine/score.py +++ b/rose/engine/score.py @@ -162,34 +162,6 @@ def process(players, track): log.debug("player %s missed %s", player.name, obstacle) - elif obstacle == obstacles.COIN: - if player.action == actions.PICKUP: - # Move forward and collect a coin. Coins are worth nothing - # on their own; only the coin that completes the jackpot - # tally awards points, and further coins award nothing more. - track.clear(player.x, player.y) - player.coins += 1 - points = config.score_move_forward - if player.coins == config.coins_for_jackpot: - points += config.score_jackpot - player.score += points - player.pickups += 1 - - log.debug( - "player %s picked up %s (%d/%d): got %d points", - player.name, - obstacle, - player.coins, - config.coins_for_jackpot, - points, - ) - else: - # Move forward leaving the obstacle on the track - player.score += config.score_move_forward - player.coin_misses += 1 - - log.debug("player %s missed %s", player.name, obstacle) - # Here we can end the game when player gets out of # the track bounds. For now, just keep the player at the same # location. diff --git a/rose/engine/test_player.py b/rose/engine/test_player.py index 19c0e7d..e93efc8 100644 --- a/rose/engine/test_player.py +++ b/rose/engine/test_player.py @@ -80,9 +80,7 @@ def test_player_state(): "lane": 1, "score": 0, "pickups": 0, - "coins": 0, "misses": 0, - "coin_misses": 0, "hits": 0, "wall_hits": 0, "water_hits": 0, diff --git a/rose/engine/test_score.py b/rose/engine/test_score.py index 87b7b38..048d44e 100644 --- a/rose/engine/test_score.py +++ b/rose/engine/test_score.py @@ -122,73 +122,6 @@ def test_other(self, action): assert self.player.misses == 1 -class TestCoin(SinglePlayerTest): - """ - Handling coins - - Coins are worthless on their own. Only collecting exactly - `config.coins_for_jackpot` of them awards a one-time jackpot bonus; - coins collected beyond that award no further bonus. - """ - - obstacle = obstacles.COIN - - def pickup(self): - # track.clear() consumes the coin each tick, so put a fresh one - # back down before every pickup to simulate collecting one per tick. - self.track.set(self.x, self.y, self.obstacle) - self.player.action = actions.PICKUP - self.process() - - def test_pickup_below_jackpot(self): - self.pickup() - assert self.player.coins == 1 - self.assert_score(self.score) - self.assert_remove_obstacle() - - def test_pickup_reaches_jackpot(self): - for _ in range(config.coins_for_jackpot): - self.pickup() - assert self.player.coins == config.coins_for_jackpot - expected = ( - config.coins_for_jackpot * config.score_move_forward - + config.score_jackpot - ) - assert self.player.score == self.score + expected - - def test_pickup_beyond_jackpot_gets_no_extra_bonus(self): - for _ in range(config.coins_for_jackpot + 2): - self.pickup() - assert self.player.coins == config.coins_for_jackpot + 2 - expected = ( - config.coins_for_jackpot + 2 - ) * config.score_move_forward + config.score_jackpot - assert self.player.score == self.score + expected - - def test_right(self): - self.player.action = actions.RIGHT - self.process() - self.assert_move_right() - self.assert_keep_obstacle() - - def test_left(self): - self.player.action = actions.LEFT - self.process() - self.assert_move_left() - self.assert_keep_obstacle() - - @pytest.mark.parametrize( - "action", [a for a in FORWARD_ACTIONS if a != actions.PICKUP] - ) - def test_other(self, action): - self.player.action = action - self.process() - self.assert_score(self.score) - assert self.player.coins == 0 - assert self.player.coin_misses == 1 - self.assert_keep_obstacle() - - class MagicActionTest(SinglePlayerTest): """ Handling obstacles with magic action From efa4b53981856fdff7968fcc7056309e1bb3acf5 Mon Sep 17 00:00:00 2001 From: Daniel Fourman Date: Wed, 15 Jul 2026 18:30:41 +0300 Subject: [PATCH 6/8] Reset displayed round numbering on telemetry clear Clear history previously left the round count monotonically increasing forever (by design, since result_count() backs the live-batch "has another round finished" poll and must never go backwards or reset mid-job). Split that: result_count() stays untouched across a clear, but a new _round_offset captures total_finished at clear time so the number shown to users (via snapshot()'s total_finished) restarts at 1 for the next game, without disturbing the internal counter. Co-Authored-By: Claude Sonnet 5 --- rose/telemetry/sinks.py | 20 ++++++++++++++------ rose/telemetry/test_observer.py | 32 ++++++++++++++++++++++++++++++++ 2 files changed, 46 insertions(+), 6 deletions(-) diff --git a/rose/telemetry/sinks.py b/rose/telemetry/sinks.py index bfff198..312dd2b 100644 --- a/rose/telemetry/sinks.py +++ b/rose/telemetry/sinks.py @@ -51,6 +51,10 @@ def __init__(self, max_ticks=500, max_results=20): self._history = collections.deque(maxlen=max_ticks) self._recent_results = collections.deque(maxlen=max_results) self._total_finished = 0 + # Snapshot of _total_finished as of the last clear(), subtracted out + # for display so round numbering restarts at 1 after a clear, without + # touching _total_finished itself (see result_count()/clear() below). + self._round_offset = 0 def on_step(self, step_index, players, track): self._history.append( @@ -78,21 +82,25 @@ def latest_result(self): return self._recent_results[-1] if self._recent_results else None def clear(self): - """Clear the displayed tick log and match history. - - Deliberately leaves _total_finished untouched: it's used by - server._run_live_batch to detect "has another round finished", - and resetting it here would break any batch job in flight. + """Clear the displayed tick log, match history, and round numbering. + + Deliberately leaves _total_finished itself untouched: it's used by + server._run_live_batch to detect "has another round finished", and + resetting it here would break any batch job in flight. Round + numbering for display is reset instead via _round_offset (see + snapshot()), so the next finished game shows as round 1 again + without disturbing that internal counter. """ self._history.clear() self._recent_results.clear() + self._round_offset = self._total_finished def snapshot(self): """Return a JSON-serializable view of recent ticks and results.""" return { "history": list(self._history), "recent_results": list(self._recent_results), - "total_finished": self._total_finished, + "total_finished": self._total_finished - self._round_offset, } diff --git a/rose/telemetry/test_observer.py b/rose/telemetry/test_observer.py index 1a20ca7..ad90ab9 100644 --- a/rose/telemetry/test_observer.py +++ b/rose/telemetry/test_observer.py @@ -2,6 +2,7 @@ from rose.telemetry.observer import NullObserver from rose.telemetry.observer import TelemetryObserver from rose.telemetry.sinks import InMemorySink +from rose.telemetry.sinks import LiveSink def test_null_observer_is_a_noop(): @@ -57,6 +58,37 @@ def test_in_memory_sink_tracks_multiple_games_independently(): assert len(sink.games) == 2 +def test_live_sink_clear_resets_displayed_round_numbering(): + sink = LiveSink() + + sink.on_game_end(players=None, result={"scores": {"A": 10}, "winner": "A"}) + sink.on_game_end(players=None, result={"scores": {"A": 10}, "winner": "A"}) + assert sink.snapshot()["total_finished"] == 2 + + sink.clear() + assert sink.snapshot()["total_finished"] == 0 + assert sink.snapshot()["recent_results"] == [] + + sink.on_game_end(players=None, result={"scores": {"A": 10}, "winner": "A"}) + assert sink.snapshot()["total_finished"] == 1 + + +def test_live_sink_clear_does_not_disturb_result_count(): + # result_count() backs server._run_live_batch's "did another round + # finish" polling and must stay monotonic across a clear(), even though + # the displayed total_finished resets for round numbering. + sink = LiveSink() + + sink.on_game_end(players=None, result={"scores": {"A": 10}, "winner": "A"}) + assert sink.result_count() == 1 + + sink.clear() + assert sink.result_count() == 1 + + sink.on_game_end(players=None, result={"scores": {"A": 10}, "winner": "A"}) + assert sink.result_count() == 2 + + def test_composite_observer_fans_out_to_children(): calls = [] From 0c2f8066cf27f2edd2452a61e3af1eefc2a4569e Mon Sep 17 00:00:00 2001 From: Daniel Fourman Date: Thu, 16 Jul 2026 10:37:47 +0300 Subject: [PATCH 7/8] Fix test_simulate_job_runs_in_background_and_completes hanging _run_live_batch() drives batch games through the shared game_loop (the same reset/running state a human's Reset+Run would set), not through simulate.py's headless runner. The test never started that loop, so telemetry.result_count() never advanced and the job polled forever until timing out. Start logic.game_loop() as a background task for the duration of the test (mirroring what server.run() does in production) and cancel it afterward, and give the poll loop enough budget to clear the mandatory inter-round pause. --- rose/engine/test_server.py | 71 ++++++++++++++++++++++++++------------ 1 file changed, 49 insertions(+), 22 deletions(-) diff --git a/rose/engine/test_server.py b/rose/engine/test_server.py index 9debc04..631fc50 100644 --- a/rose/engine/test_server.py +++ b/rose/engine/test_server.py @@ -1,4 +1,5 @@ import asyncio +import contextlib from aiohttp import web from aiohttp.test_utils import TestClient @@ -6,6 +7,7 @@ from rose.common import actions from rose.engine import config +from rose.engine import logic from rose.engine import server from rose.telemetry.sinks import LiveSink @@ -96,31 +98,56 @@ async def scenario(): def test_simulate_job_runs_in_background_and_completes(monkeypatch): + # _run_live_batch (behind /simulate) drives games through the shared + # game_loop, same as manual play would -- it doesn't call simulate.py's + # headless runner. So this test must actually run that loop in the + # background, same as server.run() does in production, or state/reset + # signals never get picked up and the job hangs forever. monkeypatch.setattr(config, "game_duration", 4) + server.state.update( + { + "rate": 50, + "running": 0, + "reset": None, + "drivers": [], + "timeleft": None, + "track_type": "same", + } + ) async def scenario(): - async with FakeDriver( - "DriverA", lambda payload: actions.NONE - ) as url_a, FakeDriver( - "DriverB", lambda payload: actions.NONE - ) as url_b, TestClient( - TestServer(build_app()) - ) as client: - start_resp = await client.post( - "/simulate", - json={"drivers": [url_a, url_b], "games": 2, "track": "same"}, - ) - assert start_resp.status == 202 - job_id = (await start_resp.json())["job_id"] - - for _ in range(200): - status_resp = await client.get(f"/simulate/{job_id}") - body = await status_resp.json() - if body["status"] != "running": - return body - await asyncio.sleep(0.01) - - raise AssertionError("job never finished") + loop_task = asyncio.create_task( + logic.game_loop(server.state, server.active_websockets, server.telemetry) + ) + try: + async with FakeDriver( + "DriverA", lambda payload: actions.NONE + ) as url_a, FakeDriver( + "DriverB", lambda payload: actions.NONE + ) as url_b, TestClient( + TestServer(build_app()) + ) as client: + start_resp = await client.post( + "/simulate", + json={"drivers": [url_a, url_b], "games": 2, "track": "same"}, + ) + assert start_resp.status == 202 + job_id = (await start_resp.json())["job_id"] + + # Budget comfortably above the mandatory 2s INTER_ROUND_PAUSE_S + # between the 2 games plus tick overhead. + for _ in range(400): + status_resp = await client.get(f"/simulate/{job_id}") + body = await status_resp.json() + if body["status"] != "running": + return body + await asyncio.sleep(0.02) + + raise AssertionError("job never finished") + finally: + loop_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await loop_task job = asyncio.run(scenario()) From d13a0c38ae667909c3437ab6166adca228d48938 Mon Sep 17 00:00:00 2001 From: Daniel Fourman Date: Thu, 16 Jul 2026 15:35:53 +0300 Subject: [PATCH 8/8] Remove unused global state declarations flagged by flake8 F824 state is a dict that's mutated in place (state["key"] = ...), never reassigned, so global was never required. --- rose/engine/server.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/rose/engine/server.py b/rose/engine/server.py index 0030070..4c68181 100644 --- a/rose/engine/server.py +++ b/rose/engine/server.py @@ -39,8 +39,6 @@ async def admin_handler(request): Returns: aiohttp.web.Response: A response indicating the new game rate or an error message. """ - global state - rate = request.rel_url.query.get("rate") if rate: try: @@ -240,8 +238,6 @@ async def run( theme (str): Path to the static them resources directory. track_type (str): Type of track can be "random" or "same". """ - global state - state["rate"] = initial_rate state["running"] = 1 if initial_running else 0 state["drivers"] = initial_drivers