tavolo/server/src/scopa/game/state.py

"""In-memory representation of a scopone scientifico game.

The whole mutable game lives in :class:`GameState`, which is serialized to
and from plain JSON for storage in Redis (see :mod:`scopa.store`). Keeping
the representation JSON-native means the store needs no custom codecs and
the state is inspectable with ``redis-cli``.

Deck convention: a 40-card Italian deck. Suits are ``D`` (denari),
``C`` (coppe), ``S`` (spade) and ``B`` (bastoni); ranks are ``1``..``10``.
A card is rendered as ``RRSUIT`` (e.g. ``07D`` is the settebello).
"""
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

SUITS = ("D", "C", "S", "B")
RANKS = tuple(range(1, 11))

# Teams are derived from the seat: seats 0 and 2 form team A (index 0),
# seats 1 and 3 form team B (index 1). Team pairs always sit opposite each
# other, as in real scopone scientifico.
TEAM_A = 0
TEAM_B = 1
TEAM_NAMES = ("A", "B")

PHASE_LOBBY = "lobby"
PHASE_PLAYING = "playing"
PHASE_FINISHED = "finished"

DEFAULT_TARGET_SCORE = 11


def team_of(seat: int) -> int:
    return seat % 2


@dataclass(frozen=True)
class Card:
    rank: int
    suit: str

    def __post_init__(self) -> None:
        if self.suit not in SUITS:
            raise ValueError(f"invalid suit: {self.suit!r}")
        if self.rank not in RANKS:
            raise ValueError(f"invalid rank: {self.rank!r}")

    @property
    def code(self) -> str:
        return f"{self.rank:02d}{self.suit}"

    @staticmethod
    def parse(code: str) -> "Card":
        code = str(code).upper()
        if len(code) != 3 or not code[:2].isdigit():
            raise ValueError(f"invalid card code: {code!r}")
        return Card(rank=int(code[:2]), suit=code[2])

    def to_json(self) -> str:
        return self.code

    @staticmethod
    def from_json(value: Any) -> "Card":
        return Card.parse(str(value))


def parse_card(code: Any) -> Card:
    """Parse a card code, raising :class:`ValueError` on malformed input."""
    try:
        return Card.parse(str(code))
    except ValueError:
        raise


@dataclass
class Move:
    """Record of a single play, broadcast so every client can show who
    played which card and what it captured."""

    seat: int
    name: str
    card: str
    captured: List[str] = field(default_factory=list)
    scopa: bool = False

    def to_json(self) -> Dict[str, Any]:
        return {
            "seat": self.seat,
            "name": self.name,
            "card": self.card,
            "captured": list(self.captured),
            "scopa": self.scopa,
        }

    @staticmethod
    def from_json(data: Dict[str, Any]) -> "Move":
        return Move(
            seat=int(data["seat"]),
            name=str(data["name"]),
            card=str(data["card"]),
            captured=[str(c) for c in data.get("captured", [])],
            scopa=bool(data.get("scopa", False)),
        )


@dataclass
class PlayerState:
    sub: str
    name: str
    seat: int
    hand: List[Card] = field(default_factory=list)
    captured: List[Card] = field(default_factory=list)
    scope: int = 0

    @property
    def team(self) -> int:
        return team_of(self.seat)

    def to_json(self) -> Dict[str, Any]:
        return {
            "sub": self.sub,
            "name": self.name,
            "seat": self.seat,
            "hand": [c.to_json() for c in self.hand],
            "captured": [c.to_json() for c in self.captured],
            "scope": self.scope,
        }

    @staticmethod
    def from_json(data: Dict[str, Any]) -> "PlayerState":
        return PlayerState(
            sub=str(data["sub"]),
            name=str(data["name"]),
            seat=int(data["seat"]),
            hand=[Card.from_json(c) for c in data.get("hand", [])],
            captured=[Card.from_json(c) for c in data.get("captured", [])],
            scope=int(data.get("scope", 0)),
        )


@dataclass
class GameState:
    id: str
    join_code: str
    creator_sub: str
    target_score: int = DEFAULT_TARGET_SCORE
    phase: str = PHASE_LOBBY
    players: List[PlayerState] = field(default_factory=list)
    table: List[Card] = field(default_factory=list)
    dealer: int = 0
    turn: int = 0
    hand_number: int = 1
    scores: List[int] = field(default_factory=lambda: [0, 0])
    winner: Optional[int] = None
    last_taker: Optional[int] = None
    # Per-hand points awarded, for a compact audit trail in the API.
    hand_scores: List[Dict[str, Any]] = field(default_factory=list)
    stats_saved: bool = False
    # ISO-8601 timestamps, used when the match result is written to Postgres.
    created_at: Optional[str] = None
    finished_at: Optional[str] = None
    # The most recent play in the current hand, for move announcements.
    last_move: Optional[Move] = None

    # -- serialization ----------------------------------------------------

    def to_json(self) -> Dict[str, Any]:
        return {
            "id": self.id,
            "join_code": self.join_code,
            "creator_sub": self.creator_sub,
            "target_score": self.target_score,
            "phase": self.phase,
            "players": [p.to_json() for p in self.players],
            "table": [c.to_json() for c in self.table],
            "dealer": self.dealer,
            "turn": self.turn,
            "hand_number": self.hand_number,
            "scores": list(self.scores),
            "winner": self.winner,
            "last_taker": self.last_taker,
            "hand_scores": list(self.hand_scores),
            "stats_saved": self.stats_saved,
            "created_at": self.created_at,
            "finished_at": self.finished_at,
            "last_move": self.last_move.to_json() if self.last_move else None,
        }

    @staticmethod
    def from_json(data: Dict[str, Any]) -> "GameState":
        return GameState(
            id=str(data["id"]),
            join_code=str(data["join_code"]),
            creator_sub=str(data.get("creator_sub", "")),
            target_score=int(data.get("target_score", DEFAULT_TARGET_SCORE)),
            phase=str(data.get("phase", PHASE_LOBBY)),
            players=[PlayerState.from_json(p) for p in data.get("players", [])],
            table=[Card.from_json(c) for c in data.get("table", [])],
            dealer=int(data.get("dealer", 0)),
            turn=int(data.get("turn", 0)),
            hand_number=int(data.get("hand_number", 1)),
            scores=[int(x) for x in data.get("scores", [0, 0])],
            winner=data.get("winner"),
            last_taker=data.get("last_taker"),
            hand_scores=list(data.get("hand_scores", [])),
            stats_saved=bool(data.get("stats_saved", False)),
            created_at=data.get("created_at"),
            finished_at=data.get("finished_at"),
            last_move=Move.from_json(data["last_move"]) if data.get("last_move") else None,
        )

    # -- helpers ----------------------------------------------------------

    def player_for(self, sub: str) -> Optional[PlayerState]:
        for player in self.players:
            if player.sub == sub:
                return player
        return None

    def seated(self, sub: str) -> bool:
        return self.player_for(sub) is not None