"""Lobby board: sandbox (solo) + mvp (multiplayer). Seat claims are server-authoritative.""" from __future__ import annotations import random import string import time import uuid from typing import Any, Optional _LETTERS = string.ascii_uppercase SANDBOX_IDS = [f"0{ch}" for ch in _LETTERS[:10]] # 0A … 0J MVP_IDS = [f"M{ch}" for ch in _LETTERS[:8]] # MA … MH SANDBOX_CAP = 1 MVP_CAP = 8 # MVP: start only after roster (>2 players) is unchanged for this long. MVP_MIN_PLAYERS = 3 # more than 2 MVP_STABLE_SEC = 30.0 SEAT_STALE_SEC = 8.0 STARTING_TIMEOUT_SEC = 40.0 LIVE_STALE_SEC = 40.0 # Car assets under assets/cars/.glb CAR_IDS = [ "porsche_gt2rs", "corvette_zr1", "ford_gt", "lambo_sc18", "mclaren_600lt", "mustang_roush", ] DEFAULT_CAR = "porsche_gt2rs" def _now() -> float: return time.time() def _empty_sandbox(lobby_id: str) -> dict[str, Any]: return { "id": lobby_id, "mode": "sandbox", "seats": {f"S-{i}": None for i in range(1, SANDBOX_CAP + 1)}, "status": "open", # open | starting | live "last_change": _now(), "match_id": None, "_stable_since": None, "_roster_sig": "", "_cars": {}, } def _empty_mvp(lobby_id: str) -> dict[str, Any]: seats: dict[str, Optional[str]] = {f"P-{i}": None for i in range(1, MVP_CAP + 1)} return { "id": lobby_id, "mode": "mvp", "seats": seats, "status": "open", "last_change": _now(), "match_id": None, "_stable_since": None, "_roster_sig": "", "_cars": {}, } def _roster_sig(seats: dict[str, Any]) -> str: """Stable signature of who occupies which seat.""" parts = [f"{seat}={user or ''}" for seat, user in sorted(seats.items())] return "|".join(parts) class LobbyBoard: def __init__(self) -> None: self.lobbies: dict[str, dict[str, Any]] = {} for lid in SANDBOX_IDS: self.lobbies[lid] = _empty_sandbox(lid) for lid in MVP_IDS: self.lobbies[lid] = _empty_mvp(lid) # username -> (lobby_id, seat) self.by_user: dict[str, tuple[str, str]] = {} def snapshot(self) -> dict[str, Any]: self._tick_stale_and_start() return { "sandbox": [self._public(self.lobbies[lid]) for lid in SANDBOX_IDS], "mvp": [self._public(self.lobbies[lid]) for lid in MVP_IDS], "serverTime": _now(), } def _public(self, lobby: dict[str, Any]) -> dict[str, Any]: filled = sum(1 for v in lobby["seats"].values() if v) countdown = None if ( lobby["mode"] == "mvp" and lobby["status"] == "open" and filled >= MVP_MIN_PLAYERS and lobby.get("_stable_since") is not None ): remaining = MVP_STABLE_SEC - (_now() - float(lobby["_stable_since"])) countdown = max(0.0, round(remaining, 1)) return { "id": lobby["id"], "mode": lobby["mode"], "seats": dict(lobby["seats"]), "status": lobby["status"], "lastChange": lobby["last_change"], "matchId": lobby["match_id"], "filled": filled, "capacity": len(lobby["seats"]), "countdown": countdown, } def _clear_user(self, username: str) -> None: prev = self.by_user.pop(username, None) if not prev: return lid, seat = prev lobby = self.lobbies.get(lid) if lobby and lobby["seats"].get(seat) == username: lobby["seats"][seat] = None lobby.get("_avatars", {}).pop(username, None) lobby.get("_hb", {}).pop(username, None) lobby.get("_cars", {}).pop(username, None) lobby["last_change"] = _now() self._note_roster_change(lobby) if lobby["status"] in ("live", "starting") and not any(lobby["seats"].values()): self._reset_lobby(lobby) def _note_roster_change(self, lobby: dict[str, Any]) -> None: """Join/leave resets MVP stability timer.""" sig = _roster_sig(lobby["seats"]) lobby["_roster_sig"] = sig filled = sum(1 for v in lobby["seats"].values() if v) if lobby["mode"] == "mvp" and lobby["status"] == "open" and filled >= MVP_MIN_PLAYERS: lobby["_stable_since"] = _now() else: lobby["_stable_since"] = None def _reset_lobby(self, lobby: dict[str, Any]) -> None: mode = lobby["mode"] lid = lobby["id"] if mode == "sandbox": self.lobbies[lid] = _empty_sandbox(lid) else: self.lobbies[lid] = _empty_mvp(lid) def leave(self, username: str) -> dict[str, Any]: self._clear_user(username) return {"ok": True} def claim( self, mode: str, lobby_id: str, username: str, seat: str, avatar_url: str | None = None, ) -> dict[str, Any]: username = (username or "").strip() if not username: return {"ok": False, "error": "username required"} lobby = self.lobbies.get(lobby_id) if not lobby or lobby["mode"] != mode: return {"ok": False, "error": "lobby not found"} if seat not in lobby["seats"]: return {"ok": False, "error": "invalid seat"} if lobby["status"] in ("live", "starting"): return {"ok": False, "error": "match already starting"} if lobby["seats"][seat] is not None: return {"ok": False, "error": "seat taken"} self._clear_user(username) lobby["seats"][seat] = username lobby["last_change"] = _now() lobby.setdefault("_hb", {})[username] = _now() avatars = lobby.setdefault("_avatars", {}) if avatar_url and str(avatar_url).strip(): avatars[username] = str(avatar_url).strip() elif username not in avatars: avatars[username] = f"https://huggingface.co/avatars/{username}" self.by_user[username] = (lobby_id, seat) self._note_roster_change(lobby) started = self._maybe_start(lobby) return { "ok": True, "lobby": self._public(lobby), "seat": seat, "started": started, "matchId": lobby["match_id"], } def heartbeat(self, username: str) -> None: prev = self.by_user.get(username) if not prev: return lid, _seat = prev lobby = self.lobbies.get(lid) if not lobby: return lobby.setdefault("_hb", {})[username] = _now() def _tick_stale_and_start(self) -> None: now = _now() stale_users: list[str] = [] for username, (lid, _seat) in list(self.by_user.items()): lobby = self.lobbies.get(lid) if not lobby: stale_users.append(username) continue hb = lobby.get("_hb", {}).get(username, lobby["last_change"]) status = lobby["status"] if status == "open" and now - hb > SEAT_STALE_SEC: stale_users.append(username) elif status == "live" and now - hb > LIVE_STALE_SEC: stale_users.append(username) elif status == "starting" and now - lobby["last_change"] > STARTING_TIMEOUT_SEC: stale_users.append(username) for u in stale_users: self._clear_user(u) for lobby in list(self.lobbies.values()): if lobby["status"] == "starting" and now - lobby["last_change"] > STARTING_TIMEOUT_SEC: self._reset_lobby(lobby) elif lobby["status"] == "live": seated = [u for u in lobby["seats"].values() if u] if seated and all( now - lobby.get("_hb", {}).get(u, 0) > LIVE_STALE_SEC for u in seated ): self._reset_lobby(lobby) elif lobby["status"] == "open": self._maybe_start(lobby) def _maybe_start(self, lobby: dict[str, Any]) -> bool: if lobby["status"] != "open": return False mode = lobby["mode"] seats = lobby["seats"] filled = [s for s, u in seats.items() if u] if mode == "sandbox": if filled: return self._mark_starting(lobby) return False # MVP: >2 players in the same seats for MVP_STABLE_SEC without join/leave. if mode == "mvp": n = len(filled) if n < MVP_MIN_PLAYERS: lobby["_stable_since"] = None return False sig = _roster_sig(seats) if sig != lobby.get("_roster_sig"): lobby["_roster_sig"] = sig lobby["_stable_since"] = _now() return False if lobby.get("_stable_since") is None: lobby["_stable_since"] = _now() return False if (_now() - float(lobby["_stable_since"])) >= MVP_STABLE_SEC: return self._mark_starting(lobby) return False return False def _assign_cars(self, lobby: dict[str, Any]) -> None: cars = lobby.setdefault("_cars", {}) if lobby["mode"] == "sandbox": for user in lobby["seats"].values(): if user: cars[user] = DEFAULT_CAR return # MVP: random unique cars when possible, then recycle. pool = CAR_IDS[:] random.shuffle(pool) i = 0 for user in lobby["seats"].values(): if not user: continue cars[user] = pool[i % len(pool)] i += 1 def _mark_starting(self, lobby: dict[str, Any]) -> bool: lobby["status"] = "starting" lobby["match_id"] = str(uuid.uuid4()) lobby["last_change"] = _now() lobby["_stable_since"] = None self._assign_cars(lobby) return True def mark_live(self, lobby_id: str) -> None: lobby = self.lobbies.get(lobby_id) if lobby and lobby["status"] == "starting": lobby["status"] = "live" def players_in(self, lobby_id: str) -> list[dict[str, str]]: lobby = self.lobbies.get(lobby_id) if not lobby: return [] avatars = lobby.get("_avatars") or {} cars = lobby.get("_cars") or {} out = [] for seat, user in lobby["seats"].items(): if user: car_id = cars.get(user) or ( DEFAULT_CAR if lobby["mode"] == "sandbox" else random.choice(CAR_IDS) ) out.append( { "username": user, "seat": seat, "side": "ffa", "carId": car_id, "avatarUrl": avatars.get(user) or f"https://huggingface.co/avatars/{user}", } ) return out def get(self, lobby_id: str) -> Optional[dict[str, Any]]: return self.lobbies.get(lobby_id) board = LobbyBoard()