daniel156161
c704fbc742
Build and Push Docker Container / build-and-push (push) Successful in 8m3s
- Add bitboard-accelerated Prism and versioned Supreme snake implementations. - Add database-backed move benchmarks and focused strategy tests. - Normalize gameplay storage while preserving replay compatibility. - Add deterministic game quality scoring and replay retention tiers. - Add backup-first SQLite cleanup, verification, and replacement tooling. - Add safe compact-plus-delta database merging with conflict detection. - Extend SQLite and PostgreSQL schemas for replay and quality metadata. - Add PostgreSQL development service and pytest import configuration. - Update gameplay documentation and the quart_common submodule revision.
85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
"""Normalized gameplay turn storage and replay hydration.
|
|
|
|
A turn is split into one board row plus one snake row per participating snake.
|
|
Static snake identity belongs to ``game_snakes``. This avoids storing complete
|
|
snake payloads in the board, snakes, you, and snake-turn columns simultaneously.
|
|
"""
|
|
|
|
from typing import Callable
|
|
|
|
def compact_turn_json(board:dict) -> tuple[dict, list, dict, list, list]:
|
|
"""Return compatibility JSON plus canonical food and hazard values."""
|
|
return {}, [], {}, board.get("food", []), board.get("hazards", [])
|
|
|
|
def hydrate_replay_turns(game_row, turn_rows, snake_rows, decode_json:Callable) -> list[dict]:
|
|
rows_by_turn:dict[int, list] = {}
|
|
for row in snake_rows:
|
|
rows_by_turn.setdefault(int(row["turn"]), []).append(row)
|
|
|
|
turns = []
|
|
for row in turn_rows:
|
|
turn = int(row["turn"])
|
|
stored_board = decode_json(row["board_state_json"]) or {}
|
|
food = decode_json(row["food_json"])
|
|
hazards = decode_json(row["hazards_json"])
|
|
|
|
snakes = []
|
|
api_snakes = []
|
|
for snake_row in rows_by_turn.get(turn, []):
|
|
body = decode_json(snake_row["body_json"]) or []
|
|
api_snake = {
|
|
"id": snake_row["snake_id"],
|
|
"name": snake_row["snake_name"],
|
|
"health": snake_row["health"],
|
|
"length": snake_row["length"],
|
|
"head": {"x": snake_row["head_x"], "y": snake_row["head_y"]},
|
|
"body": body,
|
|
}
|
|
if snake_row["latency"] is not None:
|
|
api_snake["latency"] = snake_row["latency"]
|
|
api_snakes.append(api_snake)
|
|
snakes.append({
|
|
"snake_id": snake_row["snake_id"],
|
|
"snake_name": snake_row["snake_name"],
|
|
"health": snake_row["health"],
|
|
"length": snake_row["length"],
|
|
"head": api_snake["head"],
|
|
"body": body,
|
|
"is_you": bool(snake_row["is_you"]),
|
|
"inferred_move": snake_row["inferred_move"],
|
|
"latency": snake_row["latency"],
|
|
})
|
|
|
|
board = stored_board or {
|
|
"width": game_row["width"],
|
|
"height": game_row["height"],
|
|
"food": food or [],
|
|
"hazards": hazards or [],
|
|
"snakes": api_snakes,
|
|
}
|
|
if food is None:
|
|
food = board.get("food", [])
|
|
if hazards is None:
|
|
hazards = board.get("hazards", [])
|
|
|
|
you = decode_json(row["you_json"]) or {}
|
|
if not you:
|
|
you = next(
|
|
(snake for snake in api_snakes if snake["id"] == game_row["your_snake_id"]),
|
|
{},
|
|
)
|
|
|
|
turns.append({
|
|
"turn": turn,
|
|
"observed_at": row["observed_at"],
|
|
"my_move": row["my_move"],
|
|
"my_thinking": decode_json(row["my_thinking_json"]),
|
|
"board": board,
|
|
"food": food or [],
|
|
"hazards": hazards or [],
|
|
"you": you,
|
|
"snakes": snakes,
|
|
})
|
|
|
|
return turns
|