rework that the sqlite and postgresql backend uses the same code to build the data for the dashboard
This commit is contained in:
@@ -98,3 +98,184 @@ class GameplayBackendTemplate:
|
||||
return "duel"
|
||||
|
||||
return ruleset.get("name") or "standard"
|
||||
|
||||
# ── shared output builders ─────────────────────────────────────────────────
|
||||
|
||||
def _ts_to_str(self, value) -> str|None:
|
||||
"""Normalize a timestamp: pass str through, call .isoformat() on datetime."""
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, str):
|
||||
return value
|
||||
return value.isoformat()
|
||||
|
||||
def _build_summary_output(self, totals, by_type_rows, recent_rows) -> dict:
|
||||
return {
|
||||
"total_games": int(totals["total_games"] or 0),
|
||||
"running_games": int(totals["running_games"] or 0),
|
||||
"finished_games": int(totals["finished_games"] or 0),
|
||||
"wins": int(totals["wins"] or 0),
|
||||
"losses": int(totals["losses"] or 0),
|
||||
"avg_turns_finished": round(float(totals["avg_turns"] or 0.0), 2),
|
||||
"by_game_type": [{
|
||||
"game_type": row["type_label"],
|
||||
"total": int(row["total"]),
|
||||
"wins": int(row["wins"]),
|
||||
"losses": int(row["losses"]),
|
||||
} for row in by_type_rows],
|
||||
"recent_games": [{
|
||||
"game_id": row["game_id"],
|
||||
"started_at": self._ts_to_str(row["started_at"]),
|
||||
"ended_at": self._ts_to_str(row["ended_at"]),
|
||||
"map": row["map_name"],
|
||||
"ruleset": row["ruleset_name"],
|
||||
"game_type": row["game_type"],
|
||||
"snake": row["your_snake_name"],
|
||||
"snake_type": row["your_snake_type"],
|
||||
"snake_version": row["your_snake_version"],
|
||||
"winner_you": bool(row["winner_you"]),
|
||||
"final_turn": int(row["final_turn"] or 0),
|
||||
"status": row["status"],
|
||||
} for row in recent_rows],
|
||||
}
|
||||
|
||||
def _build_game_list_output(self, rows) -> list[dict]:
|
||||
return [{
|
||||
"game_id": row["game_id"],
|
||||
"started_at": self._ts_to_str(row["started_at"]),
|
||||
"ended_at": self._ts_to_str(row["ended_at"]),
|
||||
"map": row["map_name"],
|
||||
"source": row["source"],
|
||||
"ruleset": row["ruleset_name"],
|
||||
"game_type": row["game_type"],
|
||||
"snake": row["your_snake_name"],
|
||||
"snake_type": row["your_snake_type"],
|
||||
"snake_version": row["your_snake_version"],
|
||||
"winner_you": bool(row["winner_you"]),
|
||||
"winner_name": row["winner_name"],
|
||||
"final_turn": int(row["final_turn"] or 0),
|
||||
"status": row["status"],
|
||||
} for row in rows]
|
||||
|
||||
def _build_game_replay_output(self, game_row, turn_rows, snake_rows, decode_json) -> dict:
|
||||
"""Build the full replay dict.
|
||||
|
||||
decode_json: callable applied to raw column values that may be JSON strings
|
||||
(SQLite) or already-decoded objects (PostgreSQL). Pass self._from_json for
|
||||
SQLite; pass (lambda x: x) for PostgreSQL.
|
||||
"""
|
||||
snakes_by_turn:dict[int, list[dict]] = {}
|
||||
for row in snake_rows:
|
||||
snakes_by_turn.setdefault(int(row["turn"]), []).append({
|
||||
"snake_id": row["snake_id"],
|
||||
"snake_name": row["snake_name"],
|
||||
"health": row["health"],
|
||||
"length": row["length"],
|
||||
"head": {"x": row["head_x"], "y": row["head_y"]},
|
||||
"body": decode_json(row["body_json"]) or [],
|
||||
"is_you": bool(row["is_you"]),
|
||||
"inferred_move": row["inferred_move"],
|
||||
"latency": row["latency"],
|
||||
})
|
||||
|
||||
return {
|
||||
"game": {
|
||||
"game_id": game_row["game_id"],
|
||||
"started_at": self._ts_to_str(game_row["started_at"]),
|
||||
"ended_at": self._ts_to_str(game_row["ended_at"]),
|
||||
"width": game_row["width"],
|
||||
"height": game_row["height"],
|
||||
"source": game_row["source"],
|
||||
"map": game_row["map_name"],
|
||||
"ruleset_name": game_row["ruleset_name"],
|
||||
"ruleset_version": game_row["ruleset_version"],
|
||||
"game_type": game_row["game_type"],
|
||||
"your_snake_id": game_row["your_snake_id"],
|
||||
"your_snake_name": game_row["your_snake_name"],
|
||||
"your_snake_type": game_row["your_snake_type"],
|
||||
"your_snake_version": game_row["your_snake_version"],
|
||||
"winner_name": game_row["winner_name"],
|
||||
"winner_you": bool(game_row["winner_you"]),
|
||||
"final_turn": int(game_row["final_turn"] or 0),
|
||||
"status": game_row["status"],
|
||||
},
|
||||
"turns": [
|
||||
{
|
||||
"turn": int(row["turn"]),
|
||||
"observed_at": self._ts_to_str(row["observed_at"]),
|
||||
"my_move": row["my_move"],
|
||||
"my_thinking": decode_json(row["my_thinking_json"]),
|
||||
"board": decode_json(row["board_state_json"]),
|
||||
"food": decode_json(row["food_json"]) or [],
|
||||
"hazards": decode_json(row["hazards_json"]) or [],
|
||||
"you": decode_json(row["you_json"]) or {},
|
||||
"snakes": snakes_by_turn.get(int(row["turn"]), []),
|
||||
}
|
||||
for row in turn_rows
|
||||
],
|
||||
}
|
||||
|
||||
# ── shared write-path helpers ──────────────────────────────────────────────
|
||||
|
||||
def _build_snake_turn_params(
|
||||
self,
|
||||
snakes:list[dict],
|
||||
you_id:str|None,
|
||||
game_id:str|None,
|
||||
turn:int,
|
||||
previous_positions:dict,
|
||||
) -> list[tuple]:
|
||||
"""Return one parameter tuple per snake for a snake_turns INSERT."""
|
||||
result = []
|
||||
for snake in snakes:
|
||||
snake_id = snake.get("id")
|
||||
if snake_id is None:
|
||||
continue
|
||||
head = snake.get("head", {})
|
||||
head_x = head.get("x")
|
||||
head_y = head.get("y")
|
||||
new_head = (
|
||||
(int(head_x), int(head_y))
|
||||
if head_x is not None and head_y is not None
|
||||
else None
|
||||
)
|
||||
inferred = self._infer_direction(previous_positions.get(snake_id), new_head)
|
||||
result.append((
|
||||
game_id,
|
||||
turn,
|
||||
snake_id,
|
||||
snake.get("name"),
|
||||
snake.get("health"),
|
||||
snake.get("length"),
|
||||
head_x,
|
||||
head_y,
|
||||
snake.get("body", []),
|
||||
snake_id == you_id,
|
||||
inferred,
|
||||
snake.get("latency"),
|
||||
))
|
||||
return result
|
||||
|
||||
def _extract_game_end_params(self, game_state:dict) -> tuple:
|
||||
"""Return (game_id, winner_name, winner_you, turn) from a game_end payload."""
|
||||
game = game_state.get("game", {})
|
||||
game_id = game.get("id")
|
||||
board = game_state.get("board", {})
|
||||
snakes = list(board.get("snakes", []))
|
||||
you = self._extract_you(game_state)
|
||||
winner_name = next((s.get("name") for s in snakes if s.get("name")), None)
|
||||
you_id = you.get("id")
|
||||
winner_you = any(s.get("id") == you_id for s in snakes)
|
||||
turn = int(game_state.get("turn", 0))
|
||||
return game_id, winner_name, winner_you, turn
|
||||
|
||||
def _calculate_survivor(self, snake_rows, your_snake_id:str|None) -> tuple[bool, str|None]:
|
||||
"""Return (winner_you, survivor_name) from a list of snake_turns rows."""
|
||||
survivor_ids = [s["snake_id"] for s in snake_rows if s["snake_id"]]
|
||||
winner_you = bool(
|
||||
your_snake_id
|
||||
and your_snake_id in survivor_ids
|
||||
and len(survivor_ids) == 1
|
||||
)
|
||||
survivor_name = next((s["snake_name"] for s in snake_rows if s["snake_name"]), None)
|
||||
return winner_you, survivor_name
|
||||
|
||||
Reference in New Issue
Block a user