504 lines
17 KiB
Python
504 lines
17 KiB
Python
"""Pure rules engine for scopone scientifico.
|
|
|
|
Every function here is deterministic and I/O-free: it mutates (or reads)
|
|
:class:`~scopa.game.state.GameState` and raises
|
|
:class:`~scopa.game.errors.GameError` subclasses on rule violations. This
|
|
makes the whole rule set unit-testable without Redis, Postgres or HTTP.
|
|
|
|
Rules implemented
|
|
-----------------
|
|
* 40-card Italian deck (4 suits x ranks 1-10), ten cards per player, empty
|
|
table at the start of every hand.
|
|
* A card captures either a **single card of equal rank** or a **combination
|
|
of cards whose ranks sum to its own**. When an equal-ranked card is on the
|
|
table that capture is mandatory; the player may not take an alternative
|
|
combination instead.
|
|
* Emptying the table with a capture is a **scopa** (+1), except on the very
|
|
last play of a hand.
|
|
* At the end of a hand the remaining table cards go to the player who made
|
|
the last capture.
|
|
* Hand points: ``carte`` (most captured cards), ``denara`` (most diamond
|
|
cards), ``settebello`` (the 7 of diamonds), ``primiera`` (best
|
|
seven/five/four/three card of each suit, all four suits required), plus
|
|
one point per ``scopa``. Ties on carte/denara/primiera award nothing.
|
|
* The match ends when a team reaches the target score with a clear lead; a
|
|
tie at or above the target is broken by playing another hand.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import random
|
|
from datetime import datetime, timedelta, timezone
|
|
from itertools import combinations
|
|
from typing import Dict, List, Optional, Sequence, Tuple
|
|
|
|
from .errors import (
|
|
AlreadyJoined,
|
|
CardNotInHand,
|
|
GameFinished,
|
|
GameNotStarted,
|
|
IllegalMove,
|
|
LobbyFull,
|
|
NotYourTurn,
|
|
)
|
|
from .state import (
|
|
DEFAULT_TARGET_SCORE,
|
|
PHASE_FINISHED,
|
|
PHASE_HAND_END,
|
|
PHASE_LOBBY,
|
|
PHASE_PLAYING,
|
|
SUITS,
|
|
TEAM_NAMES,
|
|
Card,
|
|
GameState,
|
|
Move,
|
|
PlayerState,
|
|
parse_card,
|
|
)
|
|
|
|
# Number of cards dealt to each player at the start of a hand.
|
|
HAND_SIZE = 10
|
|
PLAYERS = 4
|
|
|
|
# Default seconds the hand-end summary waits before dealing anyway. Games
|
|
# carry their own copy in ``GameState.hand_ack_timeout`` (configurable via
|
|
# the HAND_ACK_TIMEOUT_SECONDS environment variable).
|
|
DEFAULT_HAND_ACK_TIMEOUT_SECONDS = 30
|
|
|
|
# Default seconds a player has to play before the server plays a random
|
|
# legal card for them. Games carry their own copy in
|
|
# ``GameState.turn_timeout`` (configurable via TURN_TIMEOUT_SECONDS).
|
|
DEFAULT_TURN_TIMEOUT_SECONDS = 30
|
|
|
|
# Primiera card values: sevens are best, then sixes, then aces, then the
|
|
# remaining ranks in descending order. All of 8/9/10 are worth 10.
|
|
PRIMIERA_VALUES: Dict[int, int] = {
|
|
7: 21,
|
|
6: 18,
|
|
1: 16,
|
|
5: 15,
|
|
4: 14,
|
|
3: 13,
|
|
2: 12,
|
|
8: 10,
|
|
9: 10,
|
|
10: 10,
|
|
}
|
|
|
|
_rng = random.SystemRandom()
|
|
|
|
|
|
def full_deck() -> List[Card]:
|
|
"""Return the 40 cards of the Italian deck in canonical order."""
|
|
return [Card(rank=rank, suit=suit) for suit in SUITS for rank in range(1, 11)]
|
|
|
|
|
|
def shuffled_deck(rng: Optional[random.Random] = None) -> List[Card]:
|
|
"""Return a shuffled deck. Pass ``rng`` for deterministic tests."""
|
|
deck = full_deck()
|
|
(rng or _rng).shuffle(deck)
|
|
return deck
|
|
|
|
|
|
def legal_captures(table: Sequence[Card], card: Card) -> List[List[Card]]:
|
|
"""Return every legal capture (a list of card sets) for ``card``.
|
|
|
|
If an equal-ranked card is on the table, only those single-card
|
|
captures are returned (the rule forbids taking a combination instead).
|
|
Otherwise every subset of the table whose ranks sum to ``card.rank`` is
|
|
returned.
|
|
"""
|
|
equal = [c for c in table if c.rank == card.rank]
|
|
if equal:
|
|
return [[c] for c in equal]
|
|
|
|
candidates = [c for c in table if c.rank <= card.rank]
|
|
captures: List[List[Card]] = []
|
|
# A sum-equal capture needs at least two cards (single non-equal cards
|
|
# cannot sum to the played card).
|
|
for size in range(2, len(candidates) + 1):
|
|
for combo in combinations(candidates, size):
|
|
if sum(c.rank for c in combo) == card.rank:
|
|
captures.append(list(combo))
|
|
return captures
|
|
|
|
|
|
def create_game(
|
|
game_id: str,
|
|
join_code: str,
|
|
creator_sub: str,
|
|
creator_name: str,
|
|
target_score: int = DEFAULT_TARGET_SCORE,
|
|
hand_ack_timeout: int = DEFAULT_HAND_ACK_TIMEOUT_SECONDS,
|
|
turn_timeout: int = DEFAULT_TURN_TIMEOUT_SECONDS,
|
|
) -> GameState:
|
|
"""Create a lobby game with the creator seated first."""
|
|
if target_score < 1 or target_score > 100:
|
|
raise IllegalMove("target_score must be between 1 and 100")
|
|
return GameState(
|
|
id=game_id,
|
|
join_code=join_code,
|
|
creator_sub=creator_sub,
|
|
target_score=target_score,
|
|
phase=PHASE_LOBBY,
|
|
players=[PlayerState(sub=creator_sub, name=creator_name, seat=0)],
|
|
hand_ack_timeout=hand_ack_timeout,
|
|
turn_timeout=turn_timeout,
|
|
created_at=datetime.now(timezone.utc).isoformat(),
|
|
)
|
|
|
|
|
|
def join_game(state: GameState, sub: str, name: str) -> None:
|
|
"""Seat ``sub`` in the next free chair, starting the match when full."""
|
|
if state.phase != PHASE_LOBBY:
|
|
raise GameNotStarted("game has already started")
|
|
if state.seated(sub):
|
|
raise AlreadyJoined("already joined this game")
|
|
if len(state.players) >= PLAYERS:
|
|
raise LobbyFull("game is full")
|
|
seat = len(state.players)
|
|
state.players.append(PlayerState(sub=sub, name=name, seat=seat))
|
|
if len(state.players) == PLAYERS:
|
|
start_game(state)
|
|
|
|
|
|
def start_game(state: GameState) -> None:
|
|
"""Deal the first hand and switch the game to playing."""
|
|
if len(state.players) != PLAYERS:
|
|
raise GameNotStarted("need exactly four players to start")
|
|
state.phase = PHASE_PLAYING
|
|
_deal_hand(state)
|
|
|
|
|
|
def _set_turn_deadline(state: GameState) -> None:
|
|
"""Arm the auto-play deadline for whoever is on turn."""
|
|
deadline = datetime.now(timezone.utc) + timedelta(seconds=state.turn_timeout)
|
|
state.turn_deadline = deadline.isoformat()
|
|
|
|
|
|
def _deal_hand(state: GameState) -> None:
|
|
deck = shuffled_deck()
|
|
for player in state.players:
|
|
player.hand = []
|
|
player.captured = []
|
|
player.scope = 0
|
|
state.table = []
|
|
state.last_taker = None
|
|
# Dealer rotates each hand; the first card is played by the player to
|
|
# the dealer's left.
|
|
state.turn = (state.dealer + 1) % PLAYERS
|
|
_set_turn_deadline(state)
|
|
for offset in range(HAND_SIZE):
|
|
for seat in range(PLAYERS):
|
|
player = _player_at(state, (state.dealer + 1 + seat) % PLAYERS)
|
|
player.hand.append(deck.pop())
|
|
|
|
|
|
def _player_at(state: GameState, seat: int) -> PlayerState:
|
|
for player in state.players:
|
|
if player.seat == seat:
|
|
return player
|
|
raise IllegalMove(f"no player in seat {seat}")
|
|
|
|
|
|
def play(
|
|
state: GameState,
|
|
sub: str,
|
|
card_code: str,
|
|
capture_codes: Optional[Sequence[str]] = None,
|
|
) -> None:
|
|
"""Apply one move by the player identified by ``sub``.
|
|
|
|
``capture_codes`` selects which table cards to capture; it must be a
|
|
legal capture (see :func:`legal_captures`) when one exists and empty
|
|
otherwise. Raises a :class:`~scopa.game.errors.GameError` subclass on
|
|
any violation.
|
|
"""
|
|
if state.phase == PHASE_FINISHED:
|
|
raise GameFinished("the match is over")
|
|
if state.phase == PHASE_HAND_END:
|
|
raise IllegalMove("the hand is over; acknowledge the summary to continue")
|
|
if state.phase != PHASE_PLAYING:
|
|
raise GameNotStarted("the game has not started yet")
|
|
|
|
player = state.player_for(sub)
|
|
if player is None or player.seat != state.turn:
|
|
raise NotYourTurn("it is not your turn")
|
|
|
|
card = parse_card(card_code)
|
|
if card not in player.hand:
|
|
raise CardNotInHand(f"card {card.code} is not in your hand")
|
|
# Remove the card now so the scopa check below can tell whether this
|
|
# was the last play of the hand.
|
|
player.hand.remove(card)
|
|
|
|
requested = [parse_card(c) for c in (capture_codes or [])]
|
|
options = legal_captures(state.table, card)
|
|
|
|
taken: List[Card] = []
|
|
scopa = False
|
|
if not options:
|
|
if requested:
|
|
raise IllegalMove("no capture is possible with that card")
|
|
state.table.append(card)
|
|
else:
|
|
chosen = _match_option(options, requested)
|
|
if chosen is None:
|
|
raise IllegalMove("the requested capture is not legal")
|
|
taken = chosen
|
|
for captured in chosen:
|
|
state.table.remove(captured)
|
|
player.captured.append(captured)
|
|
player.captured.append(card)
|
|
state.last_taker = player.seat
|
|
# A scopa scores only if cards remain to be played this hand.
|
|
hands_empty = all(not p.hand for p in state.players)
|
|
if not state.table and not hands_empty:
|
|
player.scope += 1
|
|
scopa = True
|
|
|
|
state.last_move = Move(
|
|
seat=player.seat,
|
|
name=player.name,
|
|
card=card.code,
|
|
captured=[c.code for c in taken],
|
|
scopa=scopa,
|
|
)
|
|
|
|
if all(not p.hand for p in state.players):
|
|
_end_hand(state)
|
|
else:
|
|
state.turn = (state.turn + 1) % PLAYERS
|
|
_set_turn_deadline(state)
|
|
|
|
|
|
def _match_option(
|
|
options: Sequence[Sequence[Card]], requested: Sequence[Card]
|
|
) -> Optional[List[Card]]:
|
|
"""Return the option matching ``requested`` exactly, if any."""
|
|
wanted = sorted(c.code for c in requested)
|
|
if not wanted:
|
|
return None
|
|
for option in options:
|
|
if sorted(c.code for c in option) == wanted:
|
|
return list(option)
|
|
return None
|
|
|
|
|
|
def auto_play(state: GameState, rng: Optional[random.Random] = None) -> None:
|
|
"""Play a random legal move for the player currently on turn.
|
|
|
|
A card is drawn at random from that player's hand; if it can capture,
|
|
one of the legal captures is chosen at random (the rules require a
|
|
capture when one exists). Delegates to :func:`play`, so the move is
|
|
fully validated and can end the hand or the match. Pass ``rng`` for
|
|
deterministic tests.
|
|
"""
|
|
if state.phase != PHASE_PLAYING:
|
|
raise GameNotStarted("the game has not started yet")
|
|
player = _player_at(state, state.turn)
|
|
if not player.hand:
|
|
raise IllegalMove("the player on turn has no cards")
|
|
chooser = rng or _rng
|
|
card = chooser.choice(player.hand)
|
|
options = legal_captures(state.table, card)
|
|
capture = chooser.choice(options) if options else None
|
|
play(
|
|
state,
|
|
player.sub,
|
|
card.code,
|
|
[c.code for c in capture] if capture else None,
|
|
)
|
|
|
|
|
|
def _end_hand(state: GameState) -> None:
|
|
"""Sweep the table and score the hand.
|
|
|
|
If the match continues, the game pauses in the ``hand_end`` phase so
|
|
every player can read the scoring summary; the next hand is dealt by
|
|
:func:`acknowledge_hand` once all four players have acknowledged (or by
|
|
the hand-end timeout in the websocket layer). If the match is over the
|
|
game goes to ``finished`` immediately.
|
|
"""
|
|
state.turn_deadline = None
|
|
if state.table and state.last_taker is not None:
|
|
taker = _player_at(state, state.last_taker)
|
|
taker.captured.extend(state.table)
|
|
state.table = []
|
|
|
|
points, details = hand_points(state)
|
|
for team in (0, 1):
|
|
state.scores[team] += points[team]
|
|
details["hand"] = state.hand_number
|
|
details["team_a_points"] = points[0]
|
|
details["team_b_points"] = points[1]
|
|
state.hand_scores.append(details)
|
|
|
|
a, b = state.scores
|
|
reached = max(a, b) >= state.target_score
|
|
if reached and a != b:
|
|
state.phase = PHASE_FINISHED
|
|
state.winner = 0 if a > b else 1
|
|
state.finished_at = datetime.now(timezone.utc).isoformat()
|
|
return
|
|
|
|
# Pause for the scoring summary instead of dealing immediately.
|
|
state.phase = PHASE_HAND_END
|
|
state.acked = []
|
|
deadline = datetime.now(timezone.utc) + timedelta(seconds=state.hand_ack_timeout)
|
|
state.hand_end_deadline = deadline.isoformat()
|
|
|
|
|
|
def acknowledge_hand(state: GameState, sub: str) -> None:
|
|
"""Record that ``sub`` has read the hand-end summary.
|
|
|
|
When all four players have acknowledged, the next hand is dealt.
|
|
Acknowledging twice is a no-op; acknowledging outside the ``hand_end``
|
|
phase raises an error.
|
|
"""
|
|
if state.phase != PHASE_HAND_END:
|
|
raise IllegalMove("no hand summary is waiting for acknowledgement")
|
|
player = state.player_for(sub)
|
|
if player is None:
|
|
raise NotYourTurn("you are not seated in this game")
|
|
if player.seat in state.acked:
|
|
return
|
|
state.acked.append(player.seat)
|
|
if len(state.acked) < PLAYERS:
|
|
return
|
|
|
|
state.hand_number += 1
|
|
state.dealer = (state.dealer + 1) % PLAYERS
|
|
state.acked = []
|
|
state.hand_end_deadline = None
|
|
state.last_move = None
|
|
state.phase = PHASE_PLAYING
|
|
_deal_hand(state)
|
|
|
|
|
|
def primiera_score(captured: Sequence[Card]) -> int:
|
|
"""Return the primiera value of a capture pile (0 if a suit is absent)."""
|
|
best: Dict[str, int] = {}
|
|
for card in captured:
|
|
value = PRIMIERA_VALUES[card.rank]
|
|
if card.suit not in best or value > best[card.suit]:
|
|
best[card.suit] = value
|
|
if len(best) < len(SUITS):
|
|
return 0
|
|
return sum(best.values())
|
|
|
|
|
|
def hand_points(state: GameState) -> Tuple[List[int], Dict[str, object]]:
|
|
"""Compute the hand points for both teams (index 0 = team A)."""
|
|
piles: List[List[Card]] = [[], []]
|
|
scope: List[int] = [0, 0]
|
|
for player in state.players:
|
|
piles[player.team].extend(player.captured)
|
|
scope[player.team] += player.scope
|
|
|
|
points = [0, 0]
|
|
award: Dict[str, Optional[str]] = {}
|
|
# Carte: most captured cards. Ties award nothing.
|
|
cards = [len(piles[0]), len(piles[1])]
|
|
if cards[0] != cards[1]:
|
|
winner = 0 if cards[0] > cards[1] else 1
|
|
points[winner] += 1
|
|
award["carte"] = TEAM_NAMES[winner]
|
|
else:
|
|
award["carte"] = None
|
|
# Denara: most diamond cards. Ties award nothing.
|
|
coins = [
|
|
sum(1 for c in piles[t] if c.suit == "D") for t in (0, 1)
|
|
]
|
|
if coins[0] != coins[1]:
|
|
winner = 0 if coins[0] > coins[1] else 1
|
|
points[winner] += 1
|
|
award["denara"] = TEAM_NAMES[winner]
|
|
else:
|
|
award["denara"] = None
|
|
# Settebello: the 7 of diamonds always belongs to someone.
|
|
settebello = [
|
|
any(c.rank == 7 and c.suit == "D" for c in piles[t]) for t in (0, 1)
|
|
]
|
|
if settebello[0] != settebello[1]:
|
|
winner = 0 if settebello[0] else 1
|
|
points[winner] += 1
|
|
award["settebello"] = TEAM_NAMES[winner]
|
|
# Primiera: highest value, only if the team holds all four suits.
|
|
primiera = [primiera_score(piles[t]) for t in (0, 1)]
|
|
if primiera[0] != primiera[1]:
|
|
winner = 0 if primiera[0] > primiera[1] else 1
|
|
points[winner] += 1
|
|
award["primiera"] = TEAM_NAMES[winner]
|
|
else:
|
|
award["primiera"] = None
|
|
# Scope: one point each.
|
|
points[0] += scope[0]
|
|
points[1] += scope[1]
|
|
|
|
details: Dict[str, object] = {
|
|
"cards": {"A": cards[0], "B": cards[1]},
|
|
"denara": {"A": coins[0], "B": coins[1]},
|
|
"settebello": {"A": settebello[0], "B": settebello[1]},
|
|
"primiera": {"A": primiera[0], "B": primiera[1]},
|
|
"scope": {"A": scope[0], "B": scope[1]},
|
|
"award": award,
|
|
}
|
|
return points, details
|
|
|
|
|
|
def state_for_player(state: GameState, sub: str) -> Dict[str, object]:
|
|
"""Serialize ``state`` hiding other players' hands.
|
|
|
|
Hands are reduced to a count, except for the requesting player's own
|
|
hand. Raises :class:`~scopa.game.errors.GameNotFound`-style access via
|
|
the caller; this function assumes ``sub`` may or may not be seated and
|
|
simply omits the hand for non-seated viewers.
|
|
"""
|
|
viewer = state.player_for(sub)
|
|
players: List[Dict[str, object]] = []
|
|
for player in state.players:
|
|
view: Dict[str, object] = {
|
|
"sub": player.sub,
|
|
"name": player.name,
|
|
"seat": player.seat,
|
|
"team": TEAM_NAMES[player.team],
|
|
"cards_left": len(player.hand),
|
|
"captured_count": len(player.captured),
|
|
"scope": player.scope,
|
|
}
|
|
if viewer is not None and viewer.seat == player.seat:
|
|
view["hand"] = [c.code for c in player.hand]
|
|
players.append(view)
|
|
|
|
payload: Dict[str, object] = {
|
|
"id": state.id,
|
|
"join_code": state.join_code,
|
|
"phase": state.phase,
|
|
"target_score": state.target_score,
|
|
"hand_number": state.hand_number,
|
|
"dealer": state.dealer,
|
|
"turn": state.turn,
|
|
"scores": {"A": state.scores[0], "B": state.scores[1]},
|
|
"winner": None if state.winner is None else TEAM_NAMES[state.winner],
|
|
"table": [c.code for c in state.table],
|
|
"players": players,
|
|
"last_hand": state.hand_scores[-1] if state.hand_scores else None,
|
|
"last_move": state.last_move.to_json() if state.last_move else None,
|
|
"acknowledged": list(state.acked),
|
|
"hand_end_deadline": state.hand_end_deadline,
|
|
"turn_deadline": state.turn_deadline,
|
|
}
|
|
if viewer is not None and state.phase == PHASE_PLAYING and viewer.seat == state.turn:
|
|
payload["your_turn"] = True
|
|
# Only the player on turn receives their legal captures, so all the
|
|
# rule logic stays server-side.
|
|
legal_moves: Dict[str, List[List[str]]] = {}
|
|
for hand_card in viewer.hand:
|
|
options = legal_captures(state.table, hand_card)
|
|
if options:
|
|
legal_moves[hand_card.code] = [
|
|
[c.code for c in option] for option in options
|
|
]
|
|
payload["legal_moves"] = legal_moves
|
|
return payload
|