diff --git a/README.md b/README.md index 70c4227..7df91f9 100644 --- a/README.md +++ b/README.md @@ -1,17 +1,21 @@ # tavolo -A platform for multiplayer card games. Three games are included: +A platform for multiplayer card games. Four games are included: **scopone scientifico** — the four-player, fixed-partnership Italian card -game — **briscola**, in its classic two-player and four-player -partnership (*briscola a quattro*) variants, and **briscola chiamata** — -the five-player variant with a calling auction and a secret partner: +game — **scopa**, in its classic two-player and four-player partnership +(*scopa a quattro*) variants, with the *asso pigliatutto* option turning +either into **scopa d'assi** — **briscola**, in its classic two-player +and four-player partnership (*briscola a quattro*) variants, and +**briscola chiamata** — the five-player variant with a calling auction +and a secret partner: - **`server/`** — backend: Python + [kaya](https://github.com/woggioni/kaya) framework, OIDC login, live games in Redis, match statistics in Postgres. - A monorepo of four distributions sharing the `tavolo` namespace: the - game-independent `tavolo-platform`, the `tavolo-scopone` and - `tavolo-briscola` games, and the `tavolo-app` composition root wiring - them together. See [server/README.md](server/README.md). + A monorepo of five distributions sharing the `tavolo` namespace: the + game-independent `tavolo-platform`, the `tavolo-scopone`, + `tavolo-scopa` and `tavolo-briscola` games, and the `tavolo-app` + composition root wiring them together. See + [server/README.md](server/README.md). - **`web/`** — frontend: Rust + [Sycamore](https://sycamore.dev) compiled to WebAssembly, built with [Trunk](https://trunkrs.dev) and styled with [Bootstrap](https://getbootstrap.com) 5.3 (MIT, vendored under @@ -40,15 +44,16 @@ echo "127.0.0.1 mockoauth" | sudo tee -a /etc/hosts ## Between hands When a hand ends but the match is not decided, the game pauses on a -**scoring summary screen**: scopone players see how each category was won -(carte, denara, settebello, primiera, scope — plus napola and asso -pigliatutto when enabled), briscola players see the deal's points and any -*patta*; both see the running totals and must click "Understood" before -the next hand is dealt. If someone is away the next hand is dealt -automatically after `HAND_ACK_TIMEOUT_SECONDS` (default 30s). The -match-ending hand is explained on the final screen. Briscola chiamata -plays a single deal per match, so it goes straight from the last trick to -the final scoring screen; a 60–60 *patta* is replayed immediately. +**scoring summary screen**: scopone and scopa players see how each +category was won (carte, denara, settebello, primiera, scope — plus +napola and asso pigliatutto when enabled), briscola players see the +deal's points and any *patta*; both see the running totals and must click +"Understood" before the next hand is dealt. If someone is away the next +hand is dealt automatically after `HAND_ACK_TIMEOUT_SECONDS` (default +30s). The match-ending hand is explained on the final screen. Briscola +chiamata plays a single deal per match, so it goes straight from the last +trick to the final scoring screen; a 60–60 *patta* is replayed +immediately. ## On your turn @@ -80,10 +85,11 @@ Backend (from `server/`): cd server python3 -m venv .venv .venv/bin/pip install -r requirements.txt -.venv/bin/pip install -e ./packages/tavolo-platform -e ./packages/tavolo-scopone -e ./packages/tavolo-briscola -e './packages/tavolo-app[dev]' +.venv/bin/pip install -e ./packages/tavolo-platform -e ./packages/tavolo-scopone -e ./packages/tavolo-scopa -e ./packages/tavolo-briscola -e './packages/tavolo-app[dev]' .venv/bin/python -m unittest discover -s tests -t . .venv/bin/python -m unittest discover -s packages/tavolo-platform/tests .venv/bin/python -m unittest discover -s packages/tavolo-scopone/tests +.venv/bin/python -m unittest discover -s packages/tavolo-scopa/tests .venv/bin/python -m unittest discover -s packages/tavolo-briscola/tests ``` diff --git a/server/Dockerfile b/server/Dockerfile index eda6d65..d929eb0 100644 --- a/server/Dockerfile +++ b/server/Dockerfile @@ -58,13 +58,13 @@ RUN --mount=type=cache,target=/root/.cache/pip \ && /opt/venv/bin/pip install --upgrade pip \ && /opt/venv/bin/pip install -r requirements.txt -# The four sibling distributions, installed from the local tree (PEP 420 +# The five sibling distributions, installed from the local tree (PEP 420 # namespace packages sharing the `tavolo` namespace). --no-deps: every # third-party dependency — including the tavolo-app `otel` extra — is # already pinned in requirements.txt. COPY server/packages/ ./packages/ RUN --mount=type=cache,target=/root/.cache/pip \ - /opt/venv/bin/pip install --no-deps ./packages/tavolo-platform ./packages/tavolo-scopone ./packages/tavolo-briscola ./packages/tavolo-app + /opt/venv/bin/pip install --no-deps ./packages/tavolo-platform ./packages/tavolo-scopone ./packages/tavolo-scopa ./packages/tavolo-briscola ./packages/tavolo-app # server/pyproject.toml only carries the [tool.aerich] section (the # db-migrate compose service runs `aerich upgrade` with working_dir=/app, # next to ./migrations). diff --git a/server/README.md b/server/README.md index 347e6fc..26ae19d 100644 --- a/server/README.md +++ b/server/README.md @@ -1,10 +1,11 @@ # tavolo (backend) The backend for a multiplayer card-game platform, built on the -[kaya](../kaya) framework. It is a monorepo of four distributions +[kaya](../kaya) framework. It is a monorepo of five distributions sharing the `tavolo` namespace (see [`packages/tavolo-platform/README.md`](packages/tavolo-platform/README.md), -[`packages/tavolo-scopone/README.md`](packages/tavolo-scopone/README.md) +[`packages/tavolo-scopone/README.md`](packages/tavolo-scopone/README.md), +[`packages/tavolo-scopa/README.md`](packages/tavolo-scopa/README.md) and [`packages/tavolo-briscola/README.md`](packages/tavolo-briscola/README.md)): @@ -15,6 +16,10 @@ and - **`tavolo-scopone`** (`packages/tavolo-scopone/`) — scopone scientifico, the four-player, fixed-partnership variant of the classic Italian card game: pure rules plus the platform adapter. +- **`tavolo-scopa`** (`packages/tavolo-scopa/`) — scopa, the classic + Italian fishing game, in its two-player (`scopa`) and four-player + fixed-partnership (`scopa_a_quattro`) variants, with the optional asso + pigliatutto rule (scopa d'assi): pure rules plus the platform adapters. - **`tavolo-briscola`** (`packages/tavolo-briscola/`) — briscola, the Italian trick-taking game, in its classic two-player (`briscola`), four-player partnership (`briscola_a_quattro`) and five-player @@ -236,9 +241,11 @@ aggregate with `max`, not `sum`. scoped per game), start/finish timestamps, and the game-specific outcome as reported by the engine (`result` JSON — for scopone: both teams' final scores, winner, target score, hands played and the per-hand - audit; for briscola: the final scores, winner, target score, deals - played and the per-deal audit; for briscola chiamata: the calling side's - points against the defenders', the declared target, the called card and + audit; for scopa: both sides' final scores, winner, target score, the + asso pigliatutto flag, hands played and the per-hand audit; for + briscola: the final scores, winner, target score, deals played and the + per-deal audit; for briscola chiamata: the calling side's points + against the defenders', the declared target, the called card and whether the caller played alone). - `match_player` — one row per participant: the OIDC `sub`, display name, seat, game-defined team label (nullable), whether they won, the points @@ -278,7 +285,7 @@ All endpoints except `/api/health`, `/api/docs`, `/api/openapi.json`, | Method | Path | Description | |---|---|---| | `GET` | `/api/game-types` | The games the platform can host (for the creation dropdown), each with its `options_schema` | -| `POST` | `/api/games` | Create a lobby game. Optional body `{"game_type": "scopone_scientifico", "options": {"target_score": 11, "napola": true, "asso_pigliatutto": false}}`; briscola games use `{"game_type": "briscola"}` or `{"game_type": "briscola_a_quattro"}` with an optional `{"target_score": 61}`, and briscola chiamata uses `{"game_type": "briscola_chiamata"}` with no options. Returns the lobby (`{id, join_code, players, seats_open, ...}`) | +| `POST` | `/api/games` | Create a lobby game. Optional body `{"game_type": "scopone_scientifico", "options": {"target_score": 11, "napola": true, "asso_pigliatutto": false}}`; scopa games use `{"game_type": "scopa"}` or `{"game_type": "scopa_a_quattro"}` with an optional `{"target_score": 11, "asso_pigliatutto": false}` (the asso pigliatutto rule makes it scopa d'assi); briscola games use `{"game_type": "briscola"}` or `{"game_type": "briscola_a_quattro"}` with an optional `{"target_score": 61}`, and briscola chiamata uses `{"game_type": "briscola_chiamata"}` with no options. Returns the lobby (`{id, join_code, players, seats_open, ...}`) | | `POST` | `/api/games/join` | Join with `{"code": "ABC123"}`. The last player triggers the start and gets the game view back | | `GET` | `/api/games/preview?code=ABC123` | Pre-join summary of the lobby behind a code: game name/description, seated players, open seats and the engine-echoed options. `404` when the code is unknown | | `GET` | `/api/games/{id}` | Game snapshot. Seated players get the personalized view (own hand plus the join code); anyone else gets the read-only spectator view: table, scores, captured counts and scope, no hands and no join code | @@ -312,7 +319,7 @@ platform handles `state` (resend the snapshot) itself; every other action is dispatched to the session's game engine with the rest of the message as its payload. Spectators may only send `state`/`sync`: any other action is answered with an `error` whose `code` is `forbidden`. For -scopone scientifico: +scopone scientifico and scopa (both variants): ```json {"action": "play", "card": "07D", "capture": ["02D", "05C"]} @@ -384,11 +391,11 @@ for the called-card choice (a random suit of the won rank). When a hand finishes but the match continues, the game enters the `hand_end` phase instead of dealing immediately: the state carries the scoring breakdown of the hand just played (`last_hand` with an `award` -map naming the team that won each scopone category; `last_deal` with the -deal's points and running totals for briscola), the `acknowledged` seats -and a `hand_end_deadline`. The frontend renders this as a screen every -player must dismiss. A play attempted in this phase is rejected with an -`illegal_move` error. +map naming the side that won each scopone/scopa category; `last_deal` +with the deal's points and running totals for briscola), the +`acknowledged` seats and a `hand_end_deadline`. The frontend renders this +as a screen every player must dismiss. A play attempted in this phase is +rejected with an `illegal_move` error. ## Rules implemented @@ -417,6 +424,36 @@ player must dismiss. A play attempted in this phase is rejected with an configurable per game via the `target_score` creation option) with a clear lead; a tie at or above the target is broken by another hand. +### Scopa + +- 40-card Italian deck; three cards per player and four cards face up on + the table at the start of every hand. When every hand is empty the + dealer deals three more cards to each player (no new table cards) until + the deck is exhausted: two players play six rounds per hand, four + players three. +- A card captures a **single card of equal rank** (mandatory when present) + or a **combination of table cards whose ranks sum to its own**; when any + capture exists it must be taken. Remaining table cards at the end of the + hand go to the player who made the last capture. +- Emptying the table scores a *scopa* (+1), except on the last play of the + hand (a sweep on the last play of a round still scores when the deck has + cards left). +- Hand points: `carte`, `denara`, `settebello` and `primiera` (same values + and all-four-suits requirement as scopone), plus one point per scopa. + Ties award nothing. There is no napola. +- Optional *asso pigliatutto* rule (per-game `asso_pigliatutto` creation + option, default off; scopa d'assi): an ace played when no equal rank is + on the table sweeps every card on the table — or captures itself when + the table is empty. The capture never scores a scopa, but it counts as + the last capture, and the mandatory equal-rank capture still applies + when another ace is on the table. +- The match ends when a side reaches the target score (default 11, + configurable per game via the `target_score` creation option) with a + clear lead; a tie at or above the target is broken by another hand. +- `scopa` is the classic two-player game; `scopa_a_quattro` seats four + players in two fixed partnerships (seats 0/2 vs 1/3, partners + opposite). + ### Briscola - 40-card Italian deck, three cards per player. The card turned up after @@ -470,16 +507,18 @@ player must dismiss. A play attempted in this phase is rejected with an ```sh python3 -m venv .venv .venv/bin/pip install -r requirements.txt -.venv/bin/pip install -e ./packages/tavolo-platform -e ./packages/tavolo-scopone -e ./packages/tavolo-briscola -e './packages/tavolo-app[dev]' +.venv/bin/pip install -e ./packages/tavolo-platform -e ./packages/tavolo-scopone -e ./packages/tavolo-scopa -e ./packages/tavolo-briscola -e './packages/tavolo-app[dev]' # app integration suite: .venv/bin/python -m unittest discover -s tests -t . # per-package suites (each package is self-contained): .venv/bin/python -m unittest discover -s packages/tavolo-platform/tests .venv/bin/python -m unittest discover -s packages/tavolo-scopone/tests +.venv/bin/python -m unittest discover -s packages/tavolo-scopa/tests .venv/bin/python -m unittest discover -s packages/tavolo-briscola/tests # type checks (see packages/tavolo-app/README.md for the app modules): .venv/bin/python -m mypy -p tavolo.platform .venv/bin/python -m mypy -p tavolo.scopone +.venv/bin/python -m mypy -p tavolo.scopa .venv/bin/python -m mypy -p tavolo.briscola ``` @@ -538,6 +577,12 @@ packages/tavolo-scopone/ # scopone scientifico game (own pyproject + tests) ├── engine.py # pure scopone scientifico rules ├── errors.py # scopone-specific rule violations └── plugin.py # ScoponeEngine: the GameEngine adapter +packages/tavolo-scopa/ # scopa game (own pyproject + tests) + src/tavolo/scopa/ + ├── state.py # ScopaState / PlayerState / Card / Move, JSON (de)serialization + ├── engine.py # pure scopa rules (two- and four-player) + ├── errors.py # scopa-specific rule violations + └── plugin.py # ScopaEngine + ScopaQuattroEngine adapters packages/tavolo-briscola/ # briscola game (own pyproject + tests) src/tavolo/briscola/ ├── state.py # BriscolaState / PlayerState / Card / Trick, JSON (de)serialization diff --git a/server/packages/tavolo-app/README.md b/server/packages/tavolo-app/README.md index 20b15ba..4d6e135 100644 --- a/server/packages/tavolo-app/README.md +++ b/server/packages/tavolo-app/README.md @@ -2,15 +2,17 @@ The deployable tavolo application: the composition root wiring [`tavolo-platform`](../tavolo-platform/README.md) to the -[`tavolo-scopone`](../tavolo-scopone/README.md) and +[`tavolo-scopone`](../tavolo-scopone/README.md), +[`tavolo-scopa`](../tavolo-scopa/README.md) and [`tavolo-briscola`](../tavolo-briscola/README.md) games. ## Contents - `app.py` — assembles the `KayaApp` (session/OIDC/Tortoise/OpenAPI mixins plus `PlatformMixin` and the deadline scheduler), registers the - `ScoponeEngine`, `BriscolaEngine` and `BriscolaQuattroEngine` with - their configured timeouts, and exposes `tavolo.app:app` for granian. + `ScoponeEngine`, `ScopaEngine`, `ScopaQuattroEngine`, `BriscolaEngine`, + `BriscolaQuattroEngine` and `BriscolaChiamataEngine` with their + configured timeouts, and exposes `tavolo.app:app` for granian. - `config.py` — environment-driven frozen `Settings`. - `telemetry.py` — the optional OpenTelemetry wiring beyond kaya-otel: meter/tracer/logger providers, the diff --git a/server/packages/tavolo-app/pyproject.toml b/server/packages/tavolo-app/pyproject.toml index 44a471c..f80ad65 100644 --- a/server/packages/tavolo-app/pyproject.toml +++ b/server/packages/tavolo-app/pyproject.toml @@ -11,6 +11,7 @@ requires-python = ">=3.10" dependencies = [ "tavolo-platform", "tavolo-scopone", + "tavolo-scopa", "tavolo-briscola", "kaya-core", "kaya-cors", diff --git a/server/packages/tavolo-app/src/tavolo/app.py b/server/packages/tavolo-app/src/tavolo/app.py index 698b666..15583b6 100644 --- a/server/packages/tavolo-app/src/tavolo/app.py +++ b/server/packages/tavolo-app/src/tavolo/app.py @@ -48,6 +48,7 @@ from tavolo.platform import GameRegistry, Platform, PlatformMixin from tavolo.platform.deadlines import DeadlineScheduler, DeadlineSchedulerMixin from tavolo.platform.store import GameStore, InMemoryGameStore, RedisGameStore from tavolo.platform.tortoise_mixin import TortoiseMixin +from tavolo.scopa import ScopaEngine, ScopaQuattroEngine from tavolo.scopone import ScoponeEngine from .config import Settings, settings @@ -136,6 +137,18 @@ registry.register( hand_ack_timeout_seconds=settings.hand_ack_timeout_seconds, ) ) +registry.register( + ScopaEngine( + turn_timeout_seconds=settings.turn_timeout_seconds, + hand_ack_timeout_seconds=settings.hand_ack_timeout_seconds, + ) +) +registry.register( + ScopaQuattroEngine( + turn_timeout_seconds=settings.turn_timeout_seconds, + hand_ack_timeout_seconds=settings.hand_ack_timeout_seconds, + ) +) registry.register( BriscolaEngine( turn_timeout_seconds=settings.turn_timeout_seconds, diff --git a/server/packages/tavolo-platform/README.md b/server/packages/tavolo-platform/README.md index 615a3a3..f32c6d7 100644 --- a/server/packages/tavolo-platform/README.md +++ b/server/packages/tavolo-platform/README.md @@ -48,7 +48,8 @@ itself ships no game. Implement `GameEngine` (see `engine.py` for the full contract), register it in a `GameRegistry`, and mount `PlatformMixin(Platform(...))` on a `KayaApp` — see the composition root in `tavolo.app`. The canonical -examples are [`tavolo-scopone`](../tavolo-scopone/README.md) and +examples are [`tavolo-scopone`](../tavolo-scopone/README.md), +[`tavolo-scopa`](../tavolo-scopa/README.md) and [`tavolo-briscola`](../tavolo-briscola/README.md). ## Development (from `server/`) diff --git a/server/packages/tavolo-scopa/README.md b/server/packages/tavolo-scopa/README.md new file mode 100644 index 0000000..c01e93a --- /dev/null +++ b/server/packages/tavolo-scopa/README.md @@ -0,0 +1,42 @@ +# tavolo-scopa + +Scopa — the classic Italian fishing game — in its two-player form and in +the four-player fixed-partnership variant (*scopa a quattro*), as +[`tavolo-platform`](../tavolo-platform/README.md) game implementations. +The optional *asso pigliatutto* creation option turns either variant into +**scopa d'assi**. + +## Contents + +- `state.py` — `ScopaState` (pure game data: phases, players, hands, + table, the deck persisting between rounds, scores, deadlines), + `PlayerState`, `Card`, `Move`, with JSON (de)serialization. No session + envelope, no transport, no I/O. +- `engine.py` — the pure rules engine: deck, legal captures, plays, + auto-play, round re-dealing, hand scoring (carte, denara, settebello, + primiera, scope), the optional asso pigliatutto capture, hand-end + acknowledgements. Deterministic and I/O-free apart from logging, so the + whole rule set is unit-testable. +- `errors.py` — scopa-specific errors (`CardNotInHand`); every other + failure mode is a shared `tavolo.platform.errors` subclass. +- `plugin.py` — `ScopaEngine` (two players) and `ScopaQuattroEngine` + (four players in fixed partnerships): the platform-facing adapters. + They translate create/join/websocket actions/deadlines into + rules-engine calls and back, validate the + `target_score`/`asso_pigliatutto` creation options, and extract the + `MatchResult` (teams, winner, per-player scores and the match summary + persisted as the match's JSON `result`). + +Timeouts are constructor arguments (`turn_timeout_seconds`, +`hand_ack_timeout_seconds`), wired from the environment by the +application composition root. + +## Development (from `server/`) + +```sh +.venv/bin/python -m unittest discover -s packages/tavolo-scopa/tests +.venv/bin/python -m mypy -p tavolo.scopa +``` + +`test_engine.py` covers the pure rules; `test_plugin.py` covers the +platform contract (actions, deadlines, serialization, results). diff --git a/server/packages/tavolo-scopa/pyproject.toml b/server/packages/tavolo-scopa/pyproject.toml new file mode 100644 index 0000000..60347a8 --- /dev/null +++ b/server/packages/tavolo-scopa/pyproject.toml @@ -0,0 +1,22 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "tavolo-scopa" +version = "0.1.0" +description = "Scopa game implementation for the tavolo platform" +readme = "README.md" +requires-python = ">=3.10" +dependencies = [ + "tavolo-platform", +] + +[tool.setuptools.packages.find] +where = ["src"] +namespaces = true + +[tool.mypy] +python_version = "3.12" +ignore_missing_imports = true +plugins = [] diff --git a/server/packages/tavolo-scopa/src/tavolo/scopa/__init__.py b/server/packages/tavolo-scopa/src/tavolo/scopa/__init__.py new file mode 100644 index 0000000..2e1d889 --- /dev/null +++ b/server/packages/tavolo-scopa/src/tavolo/scopa/__init__.py @@ -0,0 +1,15 @@ +"""Scopa: the classic Italian fishing game. + +The pure rules (:mod:`tavolo.scopa.engine`, :mod:`tavolo.scopa.state`) +know nothing about HTTP, Redis or Postgres; :mod:`tavolo.scopa.plugin` +adapts them to the platform's +:class:`~tavolo.platform.engine.GameEngine` contract so the +game-independent platform can host them. Two variants are provided: +:class:`ScopaEngine` (two players) and :class:`ScopaQuattroEngine` (four +players in fixed partnerships). +""" +from __future__ import annotations + +from .plugin import ScopaEngine, ScopaQuattroEngine + +__all__ = ["ScopaEngine", "ScopaQuattroEngine"] diff --git a/server/packages/tavolo-scopa/src/tavolo/scopa/engine.py b/server/packages/tavolo-scopa/src/tavolo/scopa/engine.py new file mode 100644 index 0000000..2d54d85 --- /dev/null +++ b/server/packages/tavolo-scopa/src/tavolo/scopa/engine.py @@ -0,0 +1,589 @@ +"""Pure rules engine for scopa. + +Every function here is deterministic and I/O-free (the only side effect is +debug logging): it mutates (or reads) :class:`~tavolo.scopa.state.ScopaState` +and raises :class:`~tavolo.platform.errors.GameError` subclasses on rule +violations. This makes the whole rule set unit-testable without Redis, +Postgres or HTTP. The platform-facing adapter is +:class:`~tavolo.scopa.plugin.ScopaEngine`. + +Rules implemented +----------------- +* 40-card Italian deck; three cards per player and four cards face up on + the table at the start of every hand; the rest of the deck stays out of + play until the next round of the hand. +* When every hand is empty the dealer deals three more cards to each + player (no new table cards) and play resumes; this repeats until the + deck is exhausted. Two players play six rounds per hand, four players + three. +* 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. When any capture exists it must be taken. +* Emptying the table with a capture is a **scopa** (+1), except on the very + last play of a hand — a sweep on the last play of a round still scores + when the deck has cards left. +* 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. + With four players the partners' piles and scope are pooled. +* Optional ``asso_pigliatutto`` rule (disabled by default; scopa d'assi): + an ace played when no equal rank is on the table sweeps the whole table + — or captures itself, leaving the table empty, when the table is bare. + The sweep never scores a scopa, even when cards remain in hand or in the + deck, but it counts as a capture for the last-taker rule. When another + ace is on the table the mandatory equal-rank capture still applies. +* The match ends when a side 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 logging import getLogger +from typing import Dict, List, Optional, Sequence, Tuple + +from tavolo.platform.errors import ( + AlreadyJoined, + GameFinished, + GameNotStarted, + IllegalMove, + LobbyFull, + NotYourTurn, +) + +from .errors import CardNotInHand + +from .state import ( + DEFAULT_PLAYER_COUNT, + DEFAULT_TARGET_SCORE, + PHASE_FINISHED, + PHASE_HAND_END, + PHASE_LOBBY, + PHASE_PLAYING, + PLAYER_COUNTS, + SUITS, + TEAM_NAMES, + Card, + Move, + PlayerState, + ScopaState, + parse_card, +) + +log = getLogger(__name__) + +# Number of cards dealt to each player at a time. +HAND_SIZE = 3 +# Cards dealt face up on the table at the start of every hand. +TABLE_CARDS = 4 +# Default seconds the hand-end summary waits before dealing anyway. Games +# carry their own copy in ``ScopaState.hand_ack_timeout``. +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 +# ``ScopaState.turn_timeout``. +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, *, asso_pigliatutto: bool = False +) -> 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, when ``asso_pigliatutto`` is enabled and ``card`` is an ace, + the only capture is the whole table — or an empty capture when the + table is bare, the ace taking itself. Without the option 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] + if asso_pigliatutto and card.rank == 1: + return [list(table)] + + 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( + creator_sub: str, + creator_name: str, + player_count: int = DEFAULT_PLAYER_COUNT, + target_score: int = DEFAULT_TARGET_SCORE, + hand_ack_timeout: int = DEFAULT_HAND_ACK_TIMEOUT_SECONDS, + turn_timeout: int = DEFAULT_TURN_TIMEOUT_SECONDS, + asso_pigliatutto: bool = False, +) -> ScopaState: + """Create a lobby game with the creator seated first.""" + if player_count not in PLAYER_COUNTS: + raise IllegalMove( + "player_count must be one of: " + + ", ".join(str(count) for count in PLAYER_COUNTS) + ) + if target_score < 1 or target_score > 100: + raise IllegalMove("target_score must be between 1 and 100") + return ScopaState( + player_count=player_count, + target_score=target_score, + asso_pigliatutto=asso_pigliatutto, + phase=PHASE_LOBBY, + players=[PlayerState(sub=creator_sub, name=creator_name, seat=0)], + hand_ack_timeout=hand_ack_timeout, + turn_timeout=turn_timeout, + ) + + +def join_game(state: ScopaState, 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) >= state.player_count: + raise LobbyFull("game is full") + seat = len(state.players) + state.players.append(PlayerState(sub=sub, name=name, seat=seat)) + if len(state.players) == state.player_count: + start_game(state) + + +def start_game(state: ScopaState) -> None: + """Deal the first hand and switch the game to playing.""" + if len(state.players) != state.player_count: + raise GameNotStarted( + f"need exactly {state.player_count} players to start" + ) + state.phase = PHASE_PLAYING + _deal_hand(state) + + +def _set_turn_deadline(state: ScopaState) -> 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 _player_at(state: ScopaState, seat: int) -> PlayerState: + for player in state.players: + if player.seat == seat: + return player + raise IllegalMove(f"no player in seat {seat}") + + +def _deal_hand(state: ScopaState) -> None: + """Start a new hand: shuffle, lay the table, deal the first round.""" + deck = shuffled_deck() + for player in state.players: + player.hand = [] + player.captured = [] + player.scope = 0 + state.table = [] + state.last_taker = None + for _ in range(TABLE_CARDS): + state.table.append(deck.pop()) + state.deck = deck + _deal_round(state) + log.debug( + "hand %d dealt (dealer seat %d, %d players, %d on the table)", + state.hand_number, + state.dealer, + state.player_count, + len(state.table), + ) + + +def _deal_round(state: ScopaState) -> None: + """Deal three cards to every player and give the lead to the player + after the dealer.""" + for player in state.players: + player.hand = [] + for _ in range(HAND_SIZE): + for offset in range(state.player_count): + player = _player_at(state, (state.dealer + 1 + offset) % state.player_count) + player.hand.append(state.deck.pop()) + state.turn = (state.dealer + 1) % state.player_count + _set_turn_deadline(state) + + +def play( + state: ScopaState, + 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:`~tavolo.platform.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 [])] + table_before = list(state.table) + options = legal_captures( + table_before, card, asso_pigliatutto=state.asso_pigliatutto + ) + # An asso pigliatutto capture: the ace takes the whole table because no + # equal rank was on it, or takes itself when the table is bare. Such a + # capture never scores a scopa. + pigliatutto = ( + state.asso_pigliatutto + and card.rank == 1 + and all(c.rank != card.rank for c in table_before) + ) + + 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: a + # sweep on the last play of a round still scores when the deck has + # cards left to deal. + cards_left = bool(state.deck) or any(p.hand for p in state.players) + if not state.table and cards_left and not pigliatutto: + 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, + pigliatutto=pigliatutto, + ) + + if all(not p.hand for p in state.players): + if state.deck: + # Same hand continues with a fresh round of three cards each. + _deal_round(state) + else: + _end_hand(state) + else: + state.turn = (state.turn + 1) % state.player_count + _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. + + An empty request matches only an empty option — the asso pigliatutto + self-capture — so requesting nothing never satisfies a real capture. + """ + wanted = sorted(c.code for c in requested) + for option in options: + if sorted(c.code for c in option) == wanted: + return list(option) + return None + + +def auto_play(state: ScopaState, 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, asso_pigliatutto=state.asso_pigliatutto + ) + capture = chooser.choice(options) if options else None + play( + state, + player.sub, + card.code, + [c.code for c in capture] if capture is not None else None, + ) + + +def _end_hand(state: ScopaState) -> 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 every player has acknowledged (or by the + hand-end timeout fired by the platform's deadline scheduler). 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 + log.debug( + "hand %d scored A+%d B+%d (totals %d-%d)", + state.hand_number, + points[0], + points[1], + a, + b, + ) + reached = max(a, b) >= state.target_score + if reached and a != b: + state.phase = PHASE_FINISHED + state.winner = 0 if a > b else 1 + log.debug("match ended, team %s wins", TEAM_NAMES[state.winner]) + 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: ScopaState, sub: str) -> None: + """Record that ``sub`` has read the hand-end summary. + + When every player has 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) < state.player_count: + return + + state.hand_number += 1 + state.dealer = (state.dealer + 1) % state.player_count + 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: ScopaState) -> Tuple[List[int], Dict[str, object]]: + """Compute the hand points for both sides (index 0 = team A). + + With four players the partners' piles and scopa counts are pooled; with + two each player is their own side. + """ + 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 side 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] + + return points, { + "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, + } + + +def state_for_player(state: ScopaState, sub: Optional[str]) -> Dict[str, object]: + """Serialize ``state`` hiding other players' hands. + + Hands are reduced to a count, except for the requesting player's own + hand. Non-seated viewers — including spectators, for whom ``sub`` is + ``None`` — simply see no hand at all. The platform merges the session + envelope (``id``, ``join_code`` for seated players, ``game_type``, + ``spectator``) into the view itself. + """ + 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] = { + "phase": state.phase, + "player_count": state.player_count, + "target_score": state.target_score, + "asso_pigliatutto": state.asso_pigliatutto, + "hand_number": state.hand_number, + "dealer": state.dealer, + "turn": state.turn, + "deck_count": len(state.deck), + "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, asso_pigliatutto=state.asso_pigliatutto + ) + if options: + legal_moves[hand_card.code] = [ + [c.code for c in option] for option in options + ] + payload["legal_moves"] = legal_moves + return payload diff --git a/server/packages/tavolo-scopa/src/tavolo/scopa/errors.py b/server/packages/tavolo-scopa/src/tavolo/scopa/errors.py new file mode 100644 index 0000000..5565df0 --- /dev/null +++ b/server/packages/tavolo-scopa/src/tavolo/scopa/errors.py @@ -0,0 +1,34 @@ +"""Scopa errors. + +Most failure modes are game-independent and live in +:mod:`tavolo.platform.errors`; only genuinely scopa-specific errors +are defined here. +""" +from __future__ import annotations + +from tavolo.platform.errors import ( + AlreadyJoined, + GameError, + GameFinished, + GameNotFound, + GameNotStarted, + IllegalMove, + LobbyFull, + NotYourTurn, +) + +__all__ = [ + "AlreadyJoined", + "CardNotInHand", + "GameError", + "GameFinished", + "GameNotFound", + "GameNotStarted", + "IllegalMove", + "LobbyFull", + "NotYourTurn", +] + + +class CardNotInHand(GameError): + """The played card is not held by the player.""" diff --git a/server/packages/tavolo-scopa/src/tavolo/scopa/plugin.py b/server/packages/tavolo-scopa/src/tavolo/scopa/plugin.py new file mode 100644 index 0000000..943b203 --- /dev/null +++ b/server/packages/tavolo-scopa/src/tavolo/scopa/plugin.py @@ -0,0 +1,294 @@ +"""The platform-facing adapter for scopa. + +:class:`ScopaEngine` (classic two-player) and +:class:`ScopaQuattroEngine` (four-player fixed partnerships) implement +:class:`~tavolo.platform.engine.GameEngine` on top of the pure rules in +:mod:`tavolo.scopa.engine`: they translate platform calls (create, join, +websocket actions, deadlines) into rules-engine calls and back, keeping +all scopa knowledge inside this package. Nothing outside ``tavolo.scopa`` +needs to know about phases, turns or hand scoring. +""" +from __future__ import annotations + +from datetime import datetime, timezone +from logging import getLogger +from typing import Any, Dict, Mapping, Optional + +from tavolo.platform.engine import ( + Deadline, + GameEngine, + GameSession, + MatchResult, + PlayerResult, + Seat, +) +from tavolo.platform.errors import GameError, IllegalMove + +from . import engine +from .state import ( + DEFAULT_TARGET_SCORE, + PHASE_FINISHED, + PHASE_HAND_END, + PHASE_LOBBY, + PHASE_PLAYING, + TEAM_NAMES, + ScopaState, +) + +log = getLogger(__name__) + + +def _deadline_ms(iso: Optional[str]) -> Optional[int]: + """Epoch milliseconds for an ISO-8601 deadline, ``None`` when absent + or unparseable.""" + if not iso: + return None + try: + return int(datetime.fromisoformat(iso).timestamp() * 1000) + except ValueError: + return None + + +class _ScopaBase(GameEngine): + """Shared platform adapter for the two-player and four-player games.""" + + player_count: int + options_schema = { + "type": "object", + "properties": { + "target_score": { + "type": "integer", + "minimum": 1, + "maximum": 100, + "default": DEFAULT_TARGET_SCORE, + "description": "Match points the winning side must reach.", + }, + "asso_pigliatutto": { + "type": "boolean", + "default": False, + "description": "Scopa d'assi: an ace sweeps the whole " + "table (no scopa)", + }, + }, + } + + def __init__( + self, + turn_timeout_seconds: int = engine.DEFAULT_TURN_TIMEOUT_SECONDS, + hand_ack_timeout_seconds: int = engine.DEFAULT_HAND_ACK_TIMEOUT_SECONDS, + ) -> None: + self._turn_timeout = turn_timeout_seconds + self._hand_ack_timeout = hand_ack_timeout_seconds + + # -- lobby ------------------------------------------------------------ + + def create(self, session: GameSession, options: Mapping[str, Any]) -> None: + target_score = options.get("target_score", DEFAULT_TARGET_SCORE) + if isinstance(target_score, bool) or not isinstance(target_score, int): + raise IllegalMove("target_score must be an integer") + asso_pigliatutto = options.get("asso_pigliatutto", False) + if not isinstance(asso_pigliatutto, bool): + raise IllegalMove("asso_pigliatutto must be a boolean") + creator = session.players[0] + session.state = engine.create_game( + creator_sub=creator.user_sub, + creator_name=creator.display_name, + player_count=self.player_count, + target_score=target_score, + hand_ack_timeout=self._hand_ack_timeout, + turn_timeout=self._turn_timeout, + asso_pigliatutto=asso_pigliatutto, + ) + session.players[0] = Seat( + user_sub=creator.user_sub, + display_name=creator.display_name, + team=TEAM_NAMES[0], + ) + + def join(self, session: GameSession, user_sub: str, display_name: str) -> None: + seat = len(session.players) + engine.join_game(session.state, user_sub, display_name) + # join_game raises before appending on any violation, so the seat + # list stays in sync with the engine's players. + session.players.append( + Seat( + user_sub=user_sub, + display_name=display_name, + team=TEAM_NAMES[seat % 2], + ) + ) + + def in_lobby(self, session: GameSession) -> bool: + return session.state.phase == PHASE_LOBBY + + def lobby_view(self, session: GameSession) -> Dict[str, Any]: + state: ScopaState = session.state + return { + "phase": state.phase, + "target_score": state.target_score, + "asso_pigliatutto": state.asso_pigliatutto, + } + + # -- play ------------------------------------------------------------- + + def handle_action( + self, session: GameSession, user_sub: str, action: str, payload: Mapping[str, Any] + ) -> None: + state: ScopaState = session.state + if action == "play": + card = payload.get("card") + capture = payload.get("capture") + if not isinstance(card, str): + raise IllegalMove("'card' must be a card code string") + if capture is not None and ( + not isinstance(capture, list) + or any(not isinstance(item, str) for item in capture) + ): + raise IllegalMove("'capture' must be a list of card codes") + try: + engine.play(state, user_sub, card, capture) + except ValueError: + raise IllegalMove("invalid card code") + elif action == "ack": + engine.acknowledge_hand(state, user_sub) + else: + raise IllegalMove(f"unknown action: {action!r}") + + def view_for(self, session: GameSession, user_sub: Optional[str]) -> Dict[str, Any]: + return engine.state_for_player(session.state, user_sub) + + def is_finished(self, session: GameSession) -> bool: + return session.state.phase == PHASE_FINISHED + + def game_over_view(self, session: GameSession, user_sub: Optional[str]) -> Dict[str, Any]: + state: ScopaState = session.state + return { + "scores": {"A": state.scores[0], "B": state.scores[1]}, + "winner": None if state.winner is None else TEAM_NAMES[state.winner], + } + + # -- (de)serialization ------------------------------------------------- + + def state_to_json(self, state: Any) -> Dict[str, Any]: + assert isinstance(state, ScopaState) + return state.to_json() + + def state_from_json(self, data: Mapping[str, Any]) -> Any: + return ScopaState.from_json(dict(data)) + + # -- deadlines ---------------------------------------------------------- + + def next_deadline(self, session: GameSession) -> Optional[Deadline]: + state: ScopaState = session.state + if state.phase == PHASE_PLAYING and state.turn_deadline: + due_ms = _deadline_ms(state.turn_deadline) + if due_ms is None: + return None + return Deadline( + kind="turn", + due_at=datetime.fromtimestamp(due_ms / 1000, tz=timezone.utc), + token=f"turn:{state.hand_number}:{state.turn}:{due_ms}", + ) + if state.phase == PHASE_HAND_END and state.hand_end_deadline: + due_ms = _deadline_ms(state.hand_end_deadline) + if due_ms is None: + return None + return Deadline( + kind="hand_end", + due_at=datetime.fromtimestamp(due_ms / 1000, tz=timezone.utc), + token=f"hand_end:{state.hand_number}:{due_ms}", + ) + return None + + def fire_deadline(self, session: GameSession, kind: str, token: str) -> None: + current = self.next_deadline(session) + if current is None or current.kind != kind or current.token != token: + # Overtaken by events (a play landed in time, the hand was + # acknowledged, the deadline moved): nothing to do. + raise GameError("stale deadline") + state: ScopaState = session.state + if kind == "turn": + engine.auto_play(state) + actor = state.last_move.seat if state.last_move is not None else None + log.info( + "auto-played for seat %s (turn timeout, hand %d)", + actor, + state.hand_number, + ) + elif kind == "hand_end": + for player in state.players: + engine.acknowledge_hand(state, player.sub) + log.info( + "hand %d auto-advanced after the acknowledgement timeout", + state.hand_number, + ) + else: # pragma: no cover - next_deadline never emits other kinds + raise GameError(f"unknown deadline kind: {kind!r}") + + # -- results ------------------------------------------------------------- + + def result(self, session: GameSession) -> MatchResult: + state: ScopaState = session.state + if state.winner is None: + raise GameError("no result: the match is not finished") + teams = [ + [p.sub for p in state.players if p.team == 0], + [p.sub for p in state.players if p.team == 1], + ] + return MatchResult( + teams=teams, + winner_team=state.winner, + players=[ + PlayerResult( + user_sub=player.sub, + seat=player.seat, + won=player.team == state.winner, + team=TEAM_NAMES[player.team], + score=float(state.scores[player.team]), + details={ + "scope": player.scope, + "captured_cards": len(player.captured), + }, + ) + for player in state.players + ], + summary={ + "team_a_score": state.scores[0], + "team_b_score": state.scores[1], + "winner_team": TEAM_NAMES[state.winner], + "target_score": state.target_score, + "asso_pigliatutto": state.asso_pigliatutto, + "hands_played": state.hand_number, + "hand_scores": state.hand_scores, + }, + ) + + +class ScopaEngine(_ScopaBase): + """Classic two-player scopa as a platform game engine.""" + + id = "scopa" + name = "Scopa" + description = ( + "The classic two-player Italian fishing game: three cards each, " + "four on the table, capture by rank or sum and sweep for scope. " + "First to the target score wins." + ) + min_players = 2 + max_players = 2 + player_count = 2 + + +class ScopaQuattroEngine(_ScopaBase): + """Four-player partnership scopa as a platform game engine.""" + + id = "scopa_a_quattro" + name = "Scopa a quattro" + description = ( + "Scopa for four players in fixed partnerships: three cards each, " + "four on the table, capture by rank or sum and sweep for scope. " + "First side to the target score wins." + ) + min_players = 4 + max_players = 4 + player_count = 4 diff --git a/server/packages/tavolo-scopa/src/tavolo/scopa/py.typed b/server/packages/tavolo-scopa/src/tavolo/scopa/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/server/packages/tavolo-scopa/src/tavolo/scopa/state.py b/server/packages/tavolo-scopa/src/tavolo/scopa/state.py new file mode 100644 index 0000000..70dbb95 --- /dev/null +++ b/server/packages/tavolo-scopa/src/tavolo/scopa/state.py @@ -0,0 +1,261 @@ +"""In-memory representation of a scopa game. + +The whole mutable game lives in :class:`ScopaState`, which is serialized +to and from plain JSON for storage inside the platform's session envelope +(see :mod:`tavolo.platform.store`). Keeping the representation JSON-native +means the store needs no custom codecs and the state is inspectable with +``redis-cli``. + +The same state serves the classic two-player game and the four-player +partnership variant (``player_count``): with two players the two seats +are individual sides, with four they are fixed partnerships (0/2 vs 1/3), +and either way the side is the seat's parity. Within a hand the deck is +dealt in rounds of three cards per player and persists across rounds in +:attr:`ScopaState.deck`. + +The state contains only game data: the platform owns the session envelope +(id, join code, seats, timestamps, stats persistence) — see +:class:`tavolo.platform.engine.GameSession`. + +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 parity: seats 0 and 2 form team A +# (index 0), seats 1 and 3 form team B (index 1). In the two-player game +# that makes the two players their own sides (seats 0 and 1); in the +# four-player game partners sit opposite each other. +TEAM_A = 0 +TEAM_B = 1 +TEAM_NAMES = ("A", "B") + +PHASE_LOBBY = "lobby" +PHASE_PLAYING = "playing" +# Between hands of an unfinished match: scoring summary shown to every +# player; the next hand is dealt once everyone acknowledges (or the +# hand-end timeout elapses). +PHASE_HAND_END = "hand_end" +PHASE_FINISHED = "finished" + +DEFAULT_TARGET_SCORE = 11 +DEFAULT_PLAYER_COUNT = 2 +PLAYER_COUNTS = (2, 4) + + +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.""" + return Card.parse(str(code)) + + +@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 + # The play was an asso pigliatutto capture (the whole table, or the + # ace itself when the table was empty). + pigliatutto: 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, + "pigliatutto": self.pigliatutto, + } + + @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)), + pigliatutto=bool(data.get("pigliatutto", 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 ScopaState: + # 2 (individual) or 4 (fixed partnerships). Fixed for the session. + player_count: int = DEFAULT_PLAYER_COUNT + target_score: int = DEFAULT_TARGET_SCORE + # Whether the asso pigliatutto rule is scored (scopa d'assi: an ace + # sweeps the whole table when no equal rank is on it, or captures + # itself when the table is empty; the capture never scores a scopa). + # Default off, also for states serialized before the option existed. + asso_pigliatutto: bool = False + phase: str = PHASE_LOBBY + players: List[PlayerState] = field(default_factory=list) + table: List[Card] = field(default_factory=list) + # Cards of the current hand not dealt yet: scopa deals three cards per + # player at a time, and re-deals when every hand is empty until the + # deck is exhausted. + deck: 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) + # The most recent play in the current hand, for move announcements. + last_move: Optional[Move] = None + # While phase == "hand_end": seats that acknowledged the summary, and + # when the auto-continue timeout fires. + acked: List[int] = field(default_factory=list) + hand_end_deadline: Optional[str] = None + # Seconds the hand-end summary waits before dealing anyway. + hand_ack_timeout: int = 30 + # While phase == "playing": when the server plays a random legal card + # for the player on turn. + turn_deadline: Optional[str] = None + turn_timeout: int = 30 + + # -- serialization ---------------------------------------------------- + + def to_json(self) -> Dict[str, Any]: + return { + "player_count": self.player_count, + "target_score": self.target_score, + "asso_pigliatutto": self.asso_pigliatutto, + "phase": self.phase, + "players": [p.to_json() for p in self.players], + "table": [c.to_json() for c in self.table], + "deck": [c.to_json() for c in self.deck], + "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), + "last_move": self.last_move.to_json() if self.last_move else None, + "acked": list(self.acked), + "hand_end_deadline": self.hand_end_deadline, + "hand_ack_timeout": self.hand_ack_timeout, + "turn_deadline": self.turn_deadline, + "turn_timeout": self.turn_timeout, + } + + @staticmethod + def from_json(data: Dict[str, Any]) -> "ScopaState": + return ScopaState( + player_count=int(data.get("player_count", DEFAULT_PLAYER_COUNT)), + target_score=int(data.get("target_score", DEFAULT_TARGET_SCORE)), + asso_pigliatutto=bool(data.get("asso_pigliatutto", False)), + 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", [])], + deck=[Card.from_json(c) for c in data.get("deck", [])], + 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", [])), + last_move=Move.from_json(data["last_move"]) if data.get("last_move") else None, + acked=[int(s) for s in data.get("acked", [])], + hand_end_deadline=data.get("hand_end_deadline"), + hand_ack_timeout=int(data.get("hand_ack_timeout", 30)), + turn_deadline=data.get("turn_deadline"), + turn_timeout=int(data.get("turn_timeout", 30)), + ) + + # -- helpers ---------------------------------------------------------- + + def player_for(self, sub: Optional[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 diff --git a/server/packages/tavolo-scopa/tests/test_engine.py b/server/packages/tavolo-scopa/tests/test_engine.py new file mode 100644 index 0000000..1361cc2 --- /dev/null +++ b/server/packages/tavolo-scopa/tests/test_engine.py @@ -0,0 +1,616 @@ +"""Rule engine tests: deals, captures, scope, scoring and full matches.""" +from __future__ import annotations + +import random +import unittest + +from tavolo.scopa import engine +from tavolo.scopa.errors import ( + AlreadyJoined, + CardNotInHand, + GameFinished, + GameNotStarted, + IllegalMove, + NotYourTurn, +) +from tavolo.scopa.state import ( + PHASE_FINISHED, + PHASE_HAND_END, + PHASE_PLAYING, + Card, + PlayerState, + ScopaState, +) + + +def card(code: str) -> Card: + return Card.parse(code) + + +def make_state( + hands, + table, + turn: int = 0, + *, + dealer: int = 0, + captured=None, + scope=None, + target: int = 11, + last_taker=None, + deck=None, + asso_pigliatutto: bool = False, +) -> ScopaState: + """Build a controlled game state directly (bypassing the deal).""" + state = ScopaState( + player_count=len(hands), + target_score=target, + asso_pigliatutto=asso_pigliatutto, + phase=PHASE_PLAYING, + dealer=dealer, + turn=turn, + last_taker=last_taker, + ) + for seat, hand in enumerate(hands): + state.players.append( + PlayerState(sub=f"p{seat}", name=f"p{seat}", seat=seat, + hand=[card(c) for c in hand]) + ) + if captured is not None: + for player, codes in zip(state.players, captured): + player.captured = [card(c) for c in codes] + if scope is not None: + for player, value in zip(state.players, scope): + player.scope = value + state.table = [card(c) for c in table] + state.deck = [card(c) for c in (deck or [])] + return state + + +class DeckTest(unittest.TestCase): + def test_full_deck_has_40_unique_cards(self) -> None: + deck = engine.full_deck() + self.assertEqual(40, len(deck)) + self.assertEqual(40, len({c.code for c in deck})) + self.assertEqual(4, len({c.suit for c in deck})) + self.assertEqual(4, sum(1 for c in deck if c.rank == 7)) + + def test_shuffled_deck_is_permutation(self) -> None: + deck = engine.shuffled_deck() + self.assertEqual( + sorted(c.code for c in engine.full_deck()), + sorted(c.code for c in deck), + ) + + +class CaptureTest(unittest.TestCase): + def test_equal_card_is_mandatory(self) -> None: + table = [card("05C"), card("02D"), card("03S")] + options = engine.legal_captures(table, card("05D")) + self.assertEqual([["05C"]], [[c.code for c in o] for o in options]) + + def test_sum_combination(self) -> None: + table = [card("01C"), card("03C"), card("02S")] + options = engine.legal_captures(table, card("04D")) + self.assertEqual([["01C", "03C"]], [[c.code for c in o] for o in options]) + + def test_multiple_equal_cards_each_a_separate_option(self) -> None: + table = [card("05C"), card("05S")] + options = engine.legal_captures(table, card("05D")) + self.assertEqual( + [["05C"], ["05S"]], sorted([[c.code for c in o] for o in options]) + ) + + def test_no_capture(self) -> None: + table = [card("09C"), card("08S")] + self.assertEqual([], engine.legal_captures(table, card("02D"))) + + def test_play_without_capture_places_card_on_table(self) -> None: + state = make_state([["02D", "03D"], ["01C"]], table=["09C"]) + engine.play(state, "p0", "02D") + self.assertIn("02D", [c.code for c in state.table]) + self.assertNotIn("02D", [c.code for c in state.players[0].hand]) + self.assertEqual(1, state.turn) + + def test_play_capture_and_scopa(self) -> None: + state = make_state([["02D", "09C"], ["01C"]], table=["02C"]) + engine.play(state, "p0", "02D", ["02C"]) + self.assertEqual(1, state.players[0].scope) + self.assertEqual([], state.table) + self.assertEqual( + ["02C", "02D"], [c.code for c in state.players[0].captured] + ) + # The move is recorded for the "who played what" announcement. + assert state.last_move is not None + self.assertEqual(0, state.last_move.seat) + self.assertEqual("p0", state.last_move.name) + self.assertEqual("02D", state.last_move.card) + self.assertEqual(["02C"], state.last_move.captured) + self.assertTrue(state.last_move.scopa) + + def test_play_without_capture_records_move(self) -> None: + state = make_state([["02D", "03D"], ["01C"]], table=["09C"]) + engine.play(state, "p0", "02D") + assert state.last_move is not None + self.assertEqual("02D", state.last_move.card) + self.assertEqual([], state.last_move.captured) + self.assertFalse(state.last_move.scopa) + + def test_illegal_combination_when_equal_card_present(self) -> None: + state = make_state([["05D"], ["01C"]], + table=["05C", "02D", "03S"]) + with self.assertRaises(IllegalMove): + engine.play(state, "p0", "05D", ["02D", "03S"]) + + def test_illegal_capture_rejected(self) -> None: + state = make_state([["04D"], ["01C"]], table=["02C", "03S"]) + with self.assertRaises(IllegalMove): + engine.play(state, "p0", "04D", ["02C"]) + + def test_no_capture_requested_when_capture_possible(self) -> None: + state = make_state([["05D"], ["01C"]], table=["05C"]) + with self.assertRaises(IllegalMove): + engine.play(state, "p0", "05D") + + def test_not_your_turn(self) -> None: + state = make_state([["02D"], ["03D"]], table=[], turn=1) + with self.assertRaises(NotYourTurn): + engine.play(state, "p0", "02D") + + def test_card_not_in_hand(self) -> None: + state = make_state([["02D"], ["03D"]], table=[]) + with self.assertRaises(CardNotInHand): + engine.play(state, "p0", "07D") + + def test_finished_game_rejects_moves(self) -> None: + state = make_state([["02D"], ["03D"]], table=[]) + state.phase = PHASE_FINISHED + with self.assertRaises(GameFinished): + engine.play(state, "p0", "02D") + + +class DealTest(unittest.TestCase): + def test_create_game_defaults(self) -> None: + state = engine.create_game("p0", "p0") + self.assertEqual(2, state.player_count) + self.assertEqual(11, state.target_score) + self.assertFalse(state.asso_pigliatutto) + self.assertEqual("p0", state.players[0].sub) + self.assertEqual("lobby", state.phase) + + def test_create_game_player_count_validation(self) -> None: + for bad in (1, 3, 5, 0): + with self.assertRaises(IllegalMove): + engine.create_game("p0", "p0", player_count=bad) + + def test_join_starts_two_player_game(self) -> None: + state = engine.create_game("p0", "p0") + engine.join_game(state, "p1", "p1") + self.assertEqual(PHASE_PLAYING, state.phase) + self.assertEqual(4, len(state.table)) + self.assertEqual(4, len({c.code for c in state.table})) + for player in state.players: + self.assertEqual(3, len(player.hand)) + # 40 - 4 on the table - 6 in hand. + self.assertEqual(30, len(state.deck)) + self.assertEqual(1, state.turn) + + def test_join_starts_four_player_game(self) -> None: + state = engine.create_game("p0", "p0", player_count=4) + for seat in range(1, 4): + engine.join_game(state, f"p{seat}", f"p{seat}") + self.assertEqual(PHASE_PLAYING, state.phase) + self.assertEqual(4, len(state.table)) + for player in state.players: + self.assertEqual(3, len(player.hand)) + # 40 - 4 on the table - 12 in hand. + self.assertEqual(24, len(state.deck)) + self.assertEqual(1, state.turn) + for seat in range(4): + self.assertEqual(seat % 2, engine._player_at(state, seat).team) + + def test_join_errors(self) -> None: + state = engine.create_game("p0", "p0") + with self.assertRaises(AlreadyJoined): + engine.join_game(state, "p0", "p0") + engine.join_game(state, "p1", "p1") + # The lobby filled up and the match started. + with self.assertRaises(GameNotStarted): + engine.join_game(state, "p2", "p2") + + def test_start_requires_full_table(self) -> None: + state = engine.create_game("p0", "p0", player_count=4) + engine.join_game(state, "p1", "p1") + with self.assertRaises(GameNotStarted): + engine.start_game(state) + + +class RoundTest(unittest.TestCase): + def test_redeal_when_hands_empty_and_deck_has_cards(self) -> None: + deck = [f"{rank:02d}{suit}" for rank in (4, 5, 6) + for suit in ("D", "C")] + state = make_state([["02D"], ["03D"]], table=["09C"], + deck=deck, turn=0, dealer=0) + engine.play(state, "p0", "02D") + self.assertEqual(1, state.turn) + # The last card of the round: the next three cards are dealt. + engine.play(state, "p1", "03D") + self.assertEqual(PHASE_PLAYING, state.phase) + for player in state.players: + self.assertEqual(3, len(player.hand)) + self.assertEqual([], state.deck) + # The table is never topped up. + self.assertEqual( + {"09C", "02D", "03D"}, {c.code for c in state.table} + ) + # The player after the dealer leads every round again. + self.assertEqual(1, state.turn) + + def test_redeal_gives_the_lead_back_to_the_non_dealer(self) -> None: + deck = [f"{rank:02d}{suit}" for rank in (4, 5, 6) + for suit in ("D", "C")] + state = make_state([["02D"], ["03D"]], table=[], deck=deck, + turn=0, dealer=1) + engine.play(state, "p0", "02D") + engine.play(state, "p1", "03D") + # Dealer seat 1: the lead is seat 0 again. + self.assertEqual(0, state.turn) + + def test_hand_ends_when_deck_is_exhausted(self) -> None: + state = make_state([["02D"], ["03D"]], table=[], deck=[], target=100) + engine.play(state, "p0", "02D") + engine.play(state, "p1", "03D") + self.assertEqual(PHASE_HAND_END, state.phase) + + +class ScopaTest(unittest.TestCase): + def test_scopa_on_the_last_play_of_a_round_with_deck_left(self) -> None: + deck = [f"{rank:02d}{suit}" for rank in (4, 5, 6) + for suit in ("D", "C")] + state = make_state([["02D"], []], table=["02C"], deck=deck, turn=0) + engine.play(state, "p0", "02D", ["02C"]) + # Cards remain in the deck: the sweep is a scopa, and the round is + # re-dealt immediately. + self.assertEqual(1, state.players[0].scope) + assert state.last_move is not None + self.assertTrue(state.last_move.scopa) + for player in state.players: + self.assertEqual(3, len(player.hand)) + + def test_no_scopa_on_the_last_play_of_the_hand(self) -> None: + state = make_state([["02D"], []], table=["02C"], deck=[], + turn=0, target=100) + engine.play(state, "p0", "02D", ["02C"]) + self.assertEqual(0, state.players[0].scope) + assert state.last_move is not None + self.assertFalse(state.last_move.scopa) + self.assertEqual(PHASE_HAND_END, state.phase) + self.assertEqual(0, state.hand_scores[-1]["scope"]["A"]) + + +class AssoPigliatuttoTest(unittest.TestCase): + def test_ace_sweeps_the_whole_table_without_scopa(self) -> None: + # p1 still holds cards, so emptying the table would normally score + # a scopa: the sweep must suppress it. + state = make_state( + [["01D", "09C"], ["04D"]], + table=["05C", "02D", "03S"], + asso_pigliatutto=True, + ) + engine.play(state, "p0", "01D", ["05C", "02D", "03S"]) + self.assertEqual([], state.table) + self.assertEqual( + {"05C", "02D", "03S", "01D"}, + {c.code for c in state.players[0].captured}, + ) + self.assertEqual(0, state.players[0].scope) + assert state.last_move is not None + self.assertTrue(state.last_move.pigliatutto) + self.assertFalse(state.last_move.scopa) + + def test_legal_captures_offer_only_the_whole_table(self) -> None: + table = [card("05C"), card("02D"), card("03S")] + options = engine.legal_captures( + table, card("01D"), asso_pigliatutto=True + ) + self.assertEqual( + [["05C", "02D", "03S"]], [[c.code for c in o] for o in options] + ) + # A bare table leaves only the self-capture option; without the + # rule the ace stays on the table. + self.assertEqual( + [[]], engine.legal_captures([], card("01D"), asso_pigliatutto=True) + ) + self.assertEqual( + [], engine.legal_captures([], card("01D"), asso_pigliatutto=False) + ) + # Non-ace cards keep the normal rules. + self.assertEqual( + [], + engine.legal_captures(table, card("04B"), asso_pigliatutto=True), + ) + + def test_equal_rank_ace_blocks_the_sweep(self) -> None: + state = make_state( + [["01D"], ["04D"]], + table=["01C", "02D"], + asso_pigliatutto=True, + ) + options = engine.legal_captures( + state.table, card("01D"), asso_pigliatutto=True + ) + self.assertEqual([["01C"]], [[c.code for c in o] for o in options]) + with self.assertRaises(IllegalMove): + engine.play(state, "p0", "01D", ["01C", "02D"]) + # A rejected play consumes the card, so retry on a fresh state. + state = make_state( + [["01D"], ["04D"]], + table=["01C", "02D"], + asso_pigliatutto=True, + ) + engine.play(state, "p0", "01D", ["01C"]) + self.assertEqual(["02D"], [c.code for c in state.table]) + assert state.last_move is not None + self.assertFalse(state.last_move.pigliatutto) + + def test_disabled_by_default(self) -> None: + state = make_state([["01D", "09C"], ["04D"]], table=["05C", "02D"]) + engine.play(state, "p0", "01D") + # Without the rule an ace captures nothing here and stays on the + # table. + self.assertEqual({"05C", "02D", "01D"}, {c.code for c in state.table}) + assert state.last_move is not None + self.assertFalse(state.last_move.pigliatutto) + # ... and on a bare table it stays there too. + state = make_state([["01D", "09C"], ["04D"]], table=[]) + engine.play(state, "p0", "01D") + self.assertEqual(["01D"], [c.code for c in state.table]) + self.assertEqual([], state.players[0].captured) + + def test_ace_on_empty_table_captures_itself(self) -> None: + # The self-capture makes p0 the last taker and scores no scopa. + state = make_state( + [["01D", "09C"], ["04D"]], + table=[], + last_taker=1, + asso_pigliatutto=True, + ) + engine.play(state, "p0", "01D") + self.assertEqual([], state.table) + self.assertEqual(["01D"], [c.code for c in state.players[0].captured]) + self.assertEqual(0, state.players[0].scope) + self.assertEqual(0, state.last_taker) + assert state.last_move is not None + self.assertTrue(state.last_move.pigliatutto) + self.assertEqual([], state.last_move.captured) + self.assertFalse(state.last_move.scopa) + + def test_non_ace_on_empty_table_stays(self) -> None: + state = make_state( + [["02D", "09C"], ["04D"]], table=[], asso_pigliatutto=True + ) + engine.play(state, "p0", "02D") + self.assertEqual(["02D"], [c.code for c in state.table]) + self.assertEqual([], state.players[0].captured) + assert state.last_move is not None + self.assertFalse(state.last_move.pigliatutto) + + +class HandEndTest(unittest.TestCase): + def test_table_swept_to_last_taker(self) -> None: + state = make_state([["02D"], []], table=["05C", "04D"], + last_taker=1, deck=[], target=100) + engine.play(state, "p0", "02D") + captured = {c.code for c in state.players[1].captured} + self.assertEqual({"05C", "04D", "02D"}, captured) + self.assertEqual(PHASE_HAND_END, state.phase) + + def test_match_ends_at_target_with_clear_lead(self) -> None: + state = make_state([["02D"], []], table=[], deck=[], turn=0, + target=1, captured=[["07D"], []]) + engine.play(state, "p0", "02D") + self.assertEqual(PHASE_FINISHED, state.phase) + self.assertEqual(0, state.winner) + + def test_tie_at_target_plays_another_hand(self) -> None: + state = make_state( + [["02D"], []], + table=[], + deck=[], + target=1, + captured=[["01D", "02C"], ["02D", "01C"]], + ) + state.scores = [1, 1] + engine.play(state, "p0", "02D") + # A 0-0 hand at the target is not a win: another hand is played. + self.assertEqual(PHASE_HAND_END, state.phase) + self.assertIsNone(state.winner) + + def test_acknowledge_deals_next_hand(self) -> None: + state = engine.create_game("p0", "p0") + engine.join_game(state, "p1", "p1") + state.phase = PHASE_HAND_END + engine.acknowledge_hand(state, "p0") + self.assertEqual(PHASE_HAND_END, state.phase) + engine.acknowledge_hand(state, "p1") + self.assertEqual(PHASE_PLAYING, state.phase) + self.assertEqual(2, state.hand_number) + self.assertEqual(1, state.dealer) + self.assertEqual(4, len(state.table)) + for player in state.players: + self.assertEqual(3, len(player.hand)) + + def test_acknowledge_rejects_outside_hand_end(self) -> None: + state = engine.create_game("p0", "p0") + engine.join_game(state, "p1", "p1") + with self.assertRaises(IllegalMove): + engine.acknowledge_hand(state, "p0") + + def test_hand_summary_has_no_napola(self) -> None: + state = make_state([["02D"], []], table=[], deck=[], target=1, + captured=[["07D"], []]) + engine.play(state, "p0", "02D") + self.assertNotIn("napola", state.hand_scores[-1]) + + +class ScoringTest(unittest.TestCase): + def test_primiera_values_and_all_suits_requirement(self) -> None: + self.assertEqual(70, engine.primiera_score( + [card(c) for c in ["07D", "06C", "01S", "05B"]])) + self.assertEqual(0, engine.primiera_score( + [card(c) for c in ["07D", "06C", "01S"]])) + self.assertEqual(40, engine.primiera_score( + [card(c) for c in ["08D", "09C", "10S", "10B"]])) + + def test_two_player_hand_points(self) -> None: + state = make_state( + [[], []], + table=[], + captured=[ + ["07D", "06C", "01S", "05B"], + ["03D", "04C", "07S", "02B"], + ], + scope=[1, 0], + ) + points, details = engine.hand_points(state) + # Carte and denara tie; A takes settebello and primiera, plus one + # scopa: A+3. + self.assertEqual({"A": 4, "B": 4}, details["cards"]) + self.assertEqual({"A": 1, "B": 1}, details["denara"]) + self.assertEqual({"A": True, "B": False}, details["settebello"]) + self.assertEqual({"A": 70, "B": 60}, details["primiera"]) + self.assertEqual({"A": 1, "B": 0}, details["scope"]) + self.assertEqual([3, 0], points) + + def test_ties_award_nothing(self) -> None: + state = make_state( + [[], []], + table=[], + captured=[ + ["06C", "01S", "05B", "02D"], + ["06S", "01B", "05D", "02C"], + ], + scope=[0, 0], + ) + points, _ = engine.hand_points(state) + # Equal cards, equal denara, equal primiera and no settebello: + # everything ties, so no points at all. + self.assertEqual([0, 0], points) + + def test_four_player_pools_the_partners(self) -> None: + state = make_state( + [[], [], [], []], + table=[], + captured=[ + ["07D", "06C", "01S", "05B"], # seat 0, team A + ["03D", "04C", "07S", "02B"], # seat 1, team B + ["02D"], # seat 2, team A + ["10D", "10C", "10S", "10B"], # seat 3, team B + ], + scope=[1, 0, 0, 2], + ) + points, details = engine.hand_points(state) + self.assertEqual([3, 3], points) + self.assertEqual({"A": 5, "B": 8}, details["cards"]) + self.assertEqual({"A": 2, "B": 2}, details["denara"]) + self.assertEqual({"A": True, "B": False}, details["settebello"]) + self.assertEqual({"A": 70, "B": 60}, details["primiera"]) + self.assertEqual({"A": 1, "B": 2}, details["scope"]) + self.assertNotIn("napola", details) + + +class SerializationTest(unittest.TestCase): + def test_state_roundtrip(self) -> None: + state = engine.create_game( + "p0", "p0", player_count=4, target_score=16, + asso_pigliatutto=True, + ) + for seat in range(1, 4): + engine.join_game(state, f"p{seat}", f"p{seat}") + restored = ScopaState.from_json(state.to_json()) + self.assertEqual(4, restored.player_count) + self.assertEqual(16, restored.target_score) + self.assertTrue(restored.asso_pigliatutto) + self.assertEqual( + [c.code for c in state.deck], [c.code for c in restored.deck] + ) + self.assertEqual( + [c.code for c in state.table], [c.code for c in restored.table] + ) + self.assertEqual(state.phase, restored.phase) + self.assertEqual(state.turn, restored.turn) + self.assertEqual( + [p.sub for p in state.players], [p.sub for p in restored.players] + ) + + def test_asso_pigliatutto_defaults_off(self) -> None: + state = make_state([["02D"], ["03D"]], table=[]) + data = state.to_json() + del data["asso_pigliatutto"] + self.assertFalse(ScopaState.from_json(data).asso_pigliatutto) + + def test_player_count_defaults_to_two(self) -> None: + data = ScopaState().to_json() + del data["player_count"] + self.assertEqual(2, ScopaState.from_json(data).player_count) + + +class AutoPlayTest(unittest.TestCase): + def test_auto_play_applies_a_legal_move(self) -> None: + state = engine.create_game("p0", "p0") + engine.join_game(state, "p1", "p1") + rng = random.Random(7) + player = engine._player_at(state, state.turn) + hand_size = len(player.hand) + engine.auto_play(state, rng) + self.assertEqual(hand_size - 1, len(player.hand)) + assert state.last_move is not None + self.assertEqual(player.seat, state.last_move.seat) + + +class FullMatchTest(unittest.TestCase): + def _play_out(self, player_count: int, target: int) -> ScopaState: + state = engine.create_game("p0", "p0", player_count=player_count, + target_score=target) + for seat in range(1, player_count): + engine.join_game(state, f"p{seat}", f"p{seat}") + rng = random.Random(1234 + player_count) + steps = 0 + while state.phase != PHASE_FINISHED and steps < 100_000: + if state.phase == PHASE_HAND_END: + for player in state.players: + engine.acknowledge_hand(state, player.sub) + continue + engine.auto_play(state, rng) + steps += 1 + self.assertEqual(PHASE_FINISHED, state.phase) + self.assertIsNotNone(state.winner) + self.assertGreaterEqual(max(state.scores), target) + self.assertNotEqual(state.scores[0], state.scores[1]) + return state + + def test_random_two_player_match_reaches_a_winner(self) -> None: + state = self._play_out(2, 5) + for hand in state.hand_scores: + self.assertIn("cards", hand) + self.assertIn("scope", hand) + + def test_random_four_player_match_reaches_a_winner(self) -> None: + state = self._play_out(4, 5) + self.assertEqual(4, len(state.players)) + + def test_all_cards_are_captured_by_the_end_of_a_hand(self) -> None: + state = engine.create_game("p0", "p0") + engine.join_game(state, "p1", "p1") + rng = random.Random(99) + # Finish exactly one hand (target 100 keeps the match alive). + state.target_score = 100 + while state.phase != PHASE_HAND_END: + engine.auto_play(state, rng) + self.assertIsNotNone(state.last_taker) + self.assertEqual( + 40, sum(len(p.captured) for p in state.players) + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/server/packages/tavolo-scopa/tests/test_plugin.py b/server/packages/tavolo-scopa/tests/test_plugin.py new file mode 100644 index 0000000..856872a --- /dev/null +++ b/server/packages/tavolo-scopa/tests/test_plugin.py @@ -0,0 +1,335 @@ +"""Adapter tests: the platform contract over the pure scopa rules.""" +from __future__ import annotations + +import unittest +from datetime import datetime, timedelta, timezone + +from tavolo.platform import GameSession, Seat +from tavolo.platform.errors import ( + AlreadyJoined, + GameError, + GameFinished, + GameNotStarted, + IllegalMove, +) +from tavolo.scopa import ScopaEngine, ScopaQuattroEngine +from tavolo.scopa import engine as rules +from tavolo.scopa.state import ( + PHASE_FINISHED, + PHASE_HAND_END, + PHASE_PLAYING, + ScopaState, +) + +ENGINES = (ScopaEngine, ScopaQuattroEngine) + + +def _session(engine, **options) -> GameSession: + session = GameSession( + id="s1", + game_type=engine.id, + join_code="CODE01", + creator_sub="alice", + players=[Seat(user_sub="alice", display_name="Alice")], + ) + engine.create(session, options) + return session + + +def _started(engine, **options) -> GameSession: + session = _session(engine, **options) + for seat in range(1, engine.max_players): + engine.join(session, f"p{seat}", f"P{seat}") + return session + + +def _drive_to_end(engine, session: GameSession) -> None: + """Play the match out with the first legal move each turn.""" + moves = 0 + while not engine.is_finished(session) and moves < 200_000: + state = session.state + if state.phase == PHASE_HAND_END: + for player in state.players: + engine.handle_action(session, player.sub, "ack", {}) + continue + player = next(p for p in state.players if p.seat == state.turn) + played = player.hand[0] + options = rules.legal_captures( + state.table, played, asso_pigliatutto=state.asso_pigliatutto + ) + payload = {"card": played.code} + if options: + payload["capture"] = [c.code for c in options[0]] + engine.handle_action(session, player.sub, "play", payload) + moves += 1 + + +class CreateTest(unittest.TestCase): + def test_create_seats_creator_on_side_a(self) -> None: + for engine_type in ENGINES: + engine = engine_type() + session = _session(engine) + self.assertEqual("A", session.players[0].team) + self.assertIsInstance(session.state, ScopaState) + self.assertEqual(11, session.state.target_score) + self.assertFalse(session.state.asso_pigliatutto) + + def test_create_options(self) -> None: + engine = ScopaEngine() + session = _session(engine, target_score=21, asso_pigliatutto=True) + self.assertEqual(21, session.state.target_score) + self.assertTrue(session.state.asso_pigliatutto) + + def test_create_rejects_bad_options(self) -> None: + engine = ScopaEngine() + with self.assertRaises(IllegalMove): + _session(engine, target_score=0) + with self.assertRaises(IllegalMove): + _session(engine, target_score=101) + with self.assertRaises(IllegalMove): + _session(engine, target_score="eleven") + with self.assertRaises(IllegalMove): + _session(engine, target_score=True) + with self.assertRaises(IllegalMove): + _session(engine, asso_pigliatutto="yes") + + def test_timeouts_come_from_the_engine(self) -> None: + engine = ScopaEngine(turn_timeout_seconds=7, hand_ack_timeout_seconds=9) + session = _session(engine) + self.assertEqual(7, session.state.turn_timeout) + self.assertEqual(9, session.state.hand_ack_timeout) + + +class JoinTest(unittest.TestCase): + def test_join_starts_two_player_game(self) -> None: + engine = ScopaEngine() + session = _session(engine) + self.assertTrue(engine.in_lobby(session)) + engine.join(session, "bob", "Bob") + self.assertFalse(engine.in_lobby(session)) + self.assertEqual(PHASE_PLAYING, session.state.phase) + self.assertEqual(["A", "B"], [p.team for p in session.players]) + + def test_join_starts_four_player_game_with_teams(self) -> None: + engine = ScopaQuattroEngine() + session = _session(engine) + for seat, team in ((1, "B"), (2, "A"), (3, "B")): + engine.join(session, f"p{seat}", f"P{seat}") + self.assertEqual(team, session.players[-1].team) + self.assertFalse(engine.in_lobby(session)) + self.assertEqual(PHASE_PLAYING, session.state.phase) + + def test_join_errors(self) -> None: + engine = ScopaEngine() + session = _session(engine) + with self.assertRaises(AlreadyJoined): + engine.join(session, "alice", "Alice") + engine.join(session, "bob", "Bob") + with self.assertRaises(GameNotStarted): + engine.join(session, "carol", "Carol") + with self.assertRaises(GameNotStarted): + engine.join(session, "alice", "Alice") + + +class ActionTest(unittest.TestCase): + def test_unknown_action_rejected(self) -> None: + engine = ScopaEngine() + session = _started(engine) + with self.assertRaises(IllegalMove): + engine.handle_action(session, "alice", "dance", {}) + + def test_play_validates_payload(self) -> None: + engine = ScopaEngine() + session = _started(engine) + with self.assertRaises(IllegalMove): + engine.handle_action(session, "p1", "play", {"card": 42}) + with self.assertRaises(IllegalMove): + engine.handle_action(session, "p1", "play", {}) + with self.assertRaises(IllegalMove): + engine.handle_action( + session, "p1", "play", {"card": "01D", "capture": "02C"} + ) + + def test_invalid_card_code_is_illegal_move(self) -> None: + engine = ScopaEngine() + session = _started(engine) + with self.assertRaises(IllegalMove): + engine.handle_action(session, "p1", "play", {"card": "nope"}) + + def test_finished_match_rejects_actions(self) -> None: + engine = ScopaEngine() + session = _started(engine, target_score=1) + _drive_to_end(engine, session) + self.assertTrue(engine.is_finished(session)) + with self.assertRaises(GameFinished): + engine.handle_action(session, "alice", "play", {"card": "01D"}) + + +class ViewTest(unittest.TestCase): + def test_view_hides_other_hands_and_reports_the_deck(self) -> None: + engine = ScopaEngine() + session = _started(engine) + view = engine.view_for(session, "alice") + players = {p["seat"]: p for p in view["players"]} + self.assertIn("hand", players[0]) + self.assertNotIn("hand", players[1]) + self.assertEqual(2, view["player_count"]) + self.assertEqual(30, view["deck_count"]) + self.assertEqual(4, len(view["table"])) + # The envelope is the platform's job, not the view's. + self.assertNotIn("id", view) + self.assertNotIn("join_code", view) + self.assertNotIn("game_type", view) + + def test_view_for_spectator_hides_all_hands(self) -> None: + engine = ScopaQuattroEngine() + session = _started(engine) + view = engine.view_for(session, None) + for player in view["players"]: + self.assertNotIn("hand", player) + self.assertNotIn("your_turn", view) + self.assertNotIn("legal_moves", view) + self.assertIn("table", view) + self.assertEqual(4, view["player_count"]) + + def test_lobby_view(self) -> None: + engine = ScopaEngine() + session = _session(engine, target_score=16, asso_pigliatutto=True) + lobby = engine.lobby_view(session) + self.assertEqual("lobby", lobby["phase"]) + self.assertEqual(16, lobby["target_score"]) + self.assertTrue(lobby["asso_pigliatutto"]) + + +class SerializationTest(unittest.TestCase): + def test_state_roundtrip(self) -> None: + engine = ScopaQuattroEngine() + session = _started(engine) + restored = engine.state_from_json(engine.state_to_json(session.state)) + self.assertIsInstance(restored, ScopaState) + self.assertEqual(session.state.phase, restored.phase) + self.assertEqual(session.state.turn, restored.turn) + self.assertEqual( + [p.sub for p in session.state.players], + [p.sub for p in restored.players], + ) + self.assertEqual( + [c.code for c in session.state.deck], + [c.code for c in restored.deck], + ) + + +class DeadlineTest(unittest.TestCase): + def test_no_deadline_in_lobby(self) -> None: + engine = ScopaEngine() + self.assertIsNone(engine.next_deadline(_session(engine))) + + def test_turn_deadline_and_revalidation(self) -> None: + engine = ScopaEngine() + session = _started(engine) + deadline = engine.next_deadline(session) + assert deadline is not None + self.assertEqual("turn", deadline.kind) + # A forged token is stale. + with self.assertRaises(GameError): + engine.fire_deadline(session, "turn", "turn:1:1:0") + turn_before = session.state.turn + engine.fire_deadline(session, deadline.kind, deadline.token) + self.assertNotEqual(turn_before, session.state.turn) + + def test_stale_deadline_after_play(self) -> None: + engine = ScopaEngine() + session = _started(engine) + deadline = engine.next_deadline(session) + assert deadline is not None + state = session.state + player = next(p for p in state.players if p.seat == state.turn) + played = player.hand[0] + options = rules.legal_captures(state.table, played) + payload = {"card": played.code} + if options: + payload["capture"] = [c.code for c in options[0]] + engine.handle_action(session, player.sub, "play", payload) + with self.assertRaises(GameError): + engine.fire_deadline(session, deadline.kind, deadline.token) + + def test_hand_end_deadline_advances(self) -> None: + engine = ScopaEngine() + session = _started(engine) + state = session.state + # Force the hand-end phase with an imminent deadline. + state.phase = PHASE_HAND_END + state.hand_end_deadline = ( + datetime.now(timezone.utc) + timedelta(seconds=60) + ).isoformat() + deadline = engine.next_deadline(session) + assert deadline is not None + self.assertEqual("hand_end", deadline.kind) + engine.fire_deadline(session, deadline.kind, deadline.token) + self.assertEqual(PHASE_PLAYING, session.state.phase) + self.assertEqual(2, session.state.hand_number) + + +class ResultTest(unittest.TestCase): + def test_result_of_finished_two_player_match(self) -> None: + engine = ScopaEngine() + session = _started(engine, target_score=1) + _drive_to_end(engine, session) + result = engine.result(session) + self.assertEqual(2, len(result.teams)) + self.assertEqual(1, len(result.teams[0])) + self.assertEqual(1, len(result.teams[1])) + self.assertIn(result.winner_team, (0, 1)) + self.assertEqual(2, len(result.players)) + for player in result.players: + self.assertEqual( + player.won, + player.team == ("A" if result.winner_team == 0 else "B"), + ) + self.assertIn("team_a_score", result.summary) + self.assertIn("hands_played", result.summary) + self.assertIn("asso_pigliatutto", result.summary) + + def test_result_of_finished_four_player_match(self) -> None: + engine = ScopaQuattroEngine() + session = _started(engine, target_score=1) + _drive_to_end(engine, session) + result = engine.result(session) + self.assertEqual([2, 2], [len(team) for team in result.teams]) + self.assertEqual(4, len(result.players)) + + def test_result_requires_finished_match(self) -> None: + engine = ScopaEngine() + with self.assertRaises(GameError): + engine.result(_started(engine)) + + def test_game_over_view(self) -> None: + engine = ScopaEngine() + session = _started(engine) + session.state.phase = PHASE_FINISHED + session.state.winner = 1 + session.state.scores = [3, 11] + over = engine.game_over_view(session, "alice") + self.assertEqual({"A": 3, "B": 11}, over["scores"]) + self.assertEqual("B", over["winner"]) + + def test_registry_metadata(self) -> None: + two = ScopaEngine() + self.assertEqual("scopa", two.id) + self.assertEqual("Scopa", two.name) + self.assertEqual(2, two.min_players) + self.assertEqual(2, two.max_players) + four = ScopaQuattroEngine() + self.assertEqual("scopa_a_quattro", four.id) + self.assertEqual("Scopa a quattro", four.name) + self.assertEqual(4, four.min_players) + self.assertEqual(4, four.max_players) + for engine in (two, four): + self.assertIn("target_score", engine.options_schema["properties"]) + self.assertIn( + "asso_pigliatutto", engine.options_schema["properties"] + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/server/requirements.txt b/server/requirements.txt index fe03cde..55e6c09 100644 --- a/server/requirements.txt +++ b/server/requirements.txt @@ -4,10 +4,12 @@ # # pip-compile --allow-unsafe --extra otel --extra-index-url=https://pypi.org/simple --index-url=https://gitea.woggioni.net/api/packages/woggioni/pypi/simple --output-file=../../requirements.txt pyproject.toml # -# NOTE: the tavolo-platform, tavolo-scopone, tavolo-briscola and tavolo-app -# sibling packages are NOT pinned here: they are installed from ./packages in +# NOTE: the tavolo-platform, tavolo-scopone, tavolo-scopa, tavolo-briscola +# and tavolo-app sibling packages are NOT pinned here: they are installed +# from ./packages in # the Dockerfile and in development (pip install -e ./packages/tavolo-platform -# -e ./packages/tavolo-scopone -e ./packages/tavolo-briscola +# -e ./packages/tavolo-scopone -e ./packages/tavolo-scopa +# -e ./packages/tavolo-briscola # -e ./packages/tavolo-app). All of their # third-party dependencies are direct dependencies of tavolo-app, so this # lock covers the full closure. The otel extra (kaya-otel and its diff --git a/server/tests/test_routes_games.py b/server/tests/test_routes_games.py index 01ae1c5..c88b31d 100644 --- a/server/tests/test_routes_games.py +++ b/server/tests/test_routes_games.py @@ -302,6 +302,8 @@ class GameTypesRouteTest(unittest.TestCase): self.assertEqual( [ "scopone_scientifico", + "scopa", + "scopa_a_quattro", "briscola", "briscola_a_quattro", "briscola_chiamata", @@ -318,19 +320,39 @@ class GameTypesRouteTest(unittest.TestCase): self.assertIn( "asso_pigliatutto", scopone["options_schema"]["properties"] ) - briscola = results[1] + scopa = results[1] + self.assertEqual("Scopa", scopa["name"]) + self.assertTrue(scopa["description"]) + self.assertEqual(2, scopa["min_players"]) + self.assertEqual(2, scopa["max_players"]) + self.assertIn("target_score", scopa["options_schema"]["properties"]) + self.assertIn( + "asso_pigliatutto", scopa["options_schema"]["properties"] + ) + scopa_quattro = results[2] + self.assertEqual("Scopa a quattro", scopa_quattro["name"]) + self.assertTrue(scopa_quattro["description"]) + self.assertEqual(4, scopa_quattro["min_players"]) + self.assertEqual(4, scopa_quattro["max_players"]) + self.assertIn( + "target_score", scopa_quattro["options_schema"]["properties"] + ) + self.assertIn( + "asso_pigliatutto", scopa_quattro["options_schema"]["properties"] + ) + briscola = results[3] self.assertEqual("Briscola", briscola["name"]) self.assertTrue(briscola["description"]) self.assertEqual(2, briscola["min_players"]) self.assertEqual(2, briscola["max_players"]) self.assertIn("target_score", briscola["options_schema"]["properties"]) - quattro = results[2] + quattro = results[4] self.assertEqual("Briscola a quattro", quattro["name"]) self.assertTrue(quattro["description"]) self.assertEqual(4, quattro["min_players"]) self.assertEqual(4, quattro["max_players"]) self.assertIn("target_score", quattro["options_schema"]["properties"]) - chiamata = results[3] + chiamata = results[5] self.assertEqual("Briscola chiamata", chiamata["name"]) self.assertTrue(chiamata["description"]) self.assertEqual(5, chiamata["min_players"]) diff --git a/server/tests/test_scopa.py b/server/tests/test_scopa.py new file mode 100644 index 0000000..9ea7cc9 --- /dev/null +++ b/server/tests/test_scopa.py @@ -0,0 +1,412 @@ +"""Scopa integration tests: lobby, websocket play, timeouts and summaries. + +The per-package suites cover the rules and the engine contract; these +tests wire both scopa variants through the full platform stack (REST, +WebSocket, deadline scheduler). +""" +from __future__ import annotations + +import asyncio +import contextlib +import unittest + +from httpx import ASGITransport, AsyncClient +from httpx_ws import aconnect_ws +from httpx_ws.transport import ASGIWebSocketTransport + +from tavolo.app import app, game_store, platform +from tavolo.scopa.state import ( + PHASE_HAND_END, + PHASE_PLAYING, + Card, + PlayerState, + ScopaState, +) +from tavolo.platform import GameSession, Seat +from tests.helpers import async_test, make_user, oidc_user, ws_users + + +class ScopaRoutesTest(unittest.TestCase): + @async_test + async def test_create_and_start_two_player_game(self) -> None: + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: + with oidc_user("alice"): + created = await client.post( + "/api/games", + json={ + "game_type": "scopa", + "options": {"target_score": 21, "asso_pigliatutto": True}, + }, + ) + self.assertEqual(201, created.status_code) + body = created.json() + self.assertEqual("scopa", body["game_type"]) + self.assertEqual("lobby", body["phase"]) + self.assertEqual(1, body["seats_open"]) + self.assertEqual(21, body["target_score"]) + self.assertTrue(body["asso_pigliatutto"]) + self.assertEqual("A", body["players"][0]["team"]) + code = body["join_code"] + + with oidc_user("bob"): + started = await client.post("/api/games/join", json={"code": code}) + self.assertEqual(200, started.status_code) + state = started.json() + self.assertEqual("playing", state["phase"]) + self.assertEqual(2, len(state["players"])) + self.assertEqual(["A", "B"], [p["team"] for p in state["players"]]) + for participant in state["players"]: + self.assertEqual(3, participant["cards_left"]) + self.assertEqual(4, len(state["table"])) + self.assertEqual(30, state["deck_count"]) + self.assertEqual(1, state["turn"]) + self.assertTrue(state["asso_pigliatutto"]) + # Bob's view carries his own hand only. + self.assertIn("hand", state["players"][1]) + self.assertNotIn("hand", state["players"][0]) + + # A non-player watches without a hand or the join code. + watched = await client.get(f"/api/games/{state['id']}") + self.assertEqual(200, watched.status_code) + spectator = watched.json() + self.assertTrue(spectator["spectator"]) + self.assertNotIn("join_code", spectator) + for participant in spectator["players"]: + self.assertNotIn("hand", participant) + + @async_test + async def test_create_and_start_four_player_game(self) -> None: + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: + with oidc_user("alice"): + created = await client.post( + "/api/games", json={"game_type": "scopa_a_quattro"} + ) + self.assertEqual(201, created.status_code) + body = created.json() + self.assertEqual("scopa_a_quattro", body["game_type"]) + self.assertEqual(3, body["seats_open"]) + code = body["join_code"] + + for player in ("bob", "carol"): + with oidc_user(player): + joined = await client.post("/api/games/join", json={"code": code}) + self.assertEqual("lobby", joined.json()["phase"]) + + with oidc_user("dave"): + started = await client.post("/api/games/join", json={"code": code}) + state = started.json() + self.assertEqual("playing", state["phase"]) + self.assertEqual(["A", "B", "A", "B"], [p["team"] for p in state["players"]]) + for participant in state["players"]: + self.assertEqual(3, participant["cards_left"]) + self.assertEqual(4, len(state["table"])) + self.assertEqual(24, state["deck_count"]) + + @async_test + async def test_preview_shows_scopa_options(self) -> None: + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: + with oidc_user("alice"): + created = await client.post( + "/api/games", + json={ + "game_type": "scopa", + "options": {"target_score": 21, "asso_pigliatutto": True}, + }, + ) + code = created.json()["join_code"] + with oidc_user("bob"): + preview = await client.get(f"/api/games/preview?code={code}") + self.assertEqual(200, preview.status_code) + body = preview.json() + self.assertEqual("scopa", body["game_type"]) + self.assertEqual("Scopa", body["game_name"]) + self.assertTrue(body["game_description"]) + self.assertEqual(21, body["target_score"]) + self.assertTrue(body["asso_pigliatutto"]) + self.assertEqual(1, body["seats_open"]) + + @async_test + async def test_create_rejects_bad_options(self) -> None: + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: + for bad in (0, 101, "eleven", True): + with oidc_user("alice"): + response = await client.post( + "/api/games", + json={"game_type": "scopa", "options": {"target_score": bad}}, + ) + self.assertEqual(400, response.status_code, bad) + with oidc_user("alice"): + response = await client.post( + "/api/games", + json={"game_type": "scopa", "options": {"asso_pigliatutto": "yes"}}, + ) + self.assertEqual(400, response.status_code) + + +class ScopaWebSocketTest(unittest.TestCase): + async def _started_two_player(self, client: AsyncClient) -> dict: + with oidc_user("alice"): + created = await client.post("/api/games", json={"game_type": "scopa"}) + code = created.json()["join_code"] + with oidc_user("bob"): + started = await client.post("/api/games/join", json={"code": code}) + return started.json() + + @async_test + async def test_play_trails_or_captures(self) -> None: + api_transport = ASGITransport(app=app) + async with AsyncClient(transport=api_transport, base_url="http://127.0.0.1") as client: + state = await self._started_two_player(client) + game_id = state["id"] + + ws_transport = ASGIWebSocketTransport(app=app) + async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client: + with ws_users([make_user("bob")]): + async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as bob_ws: + first = await bob_ws.receive_json() + self.assertEqual("state", first["type"]) + game = first["game"] + self.assertEqual(1, game["turn"]) + self.assertTrue(game.get("your_turn")) + self.assertEqual(4, len(game["table"])) + card = game["players"][1]["hand"][0] + options = (game.get("legal_moves") or {}).get(card) + payload = {"action": "play", "card": card} + if options: + payload["capture"] = options[0] + await bob_ws.send_json(payload) + + update = await bob_ws.receive_json() + self.assertEqual("state", update["type"]) + game = update["game"] + self.assertEqual(0, game["turn"]) + self.assertIsNotNone(game["last_move"]) + self.assertEqual(1, game["last_move"]["seat"]) + self.assertEqual(card, game["last_move"]["card"]) + if options: + self.assertEqual(2, game["players"][1]["captured_count"]) + else: + self.assertEqual(5, len(game["table"])) + + @async_test + async def test_illegal_move_returns_error(self) -> None: + api_transport = ASGITransport(app=app) + async with AsyncClient(transport=api_transport, base_url="http://127.0.0.1") as client: + state = await self._started_two_player(client) + game_id = state["id"] + + ws_transport = ASGIWebSocketTransport(app=app) + async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client: + with ws_users([make_user("alice")]): + async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as alice_ws: + await alice_ws.receive_json() + # Bob (seat 1) is on turn, not Alice. + await alice_ws.send_json({"action": "play", "card": "01D"}) + error = await alice_ws.receive_json() + self.assertEqual("error", error["type"]) + + @async_test + async def test_spectator_cannot_act(self) -> None: + api_transport = ASGITransport(app=app) + async with AsyncClient(transport=api_transport, base_url="http://127.0.0.1") as client: + state = await self._started_two_player(client) + game_id = state["id"] + + ws_transport = ASGIWebSocketTransport(app=app) + async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client: + with ws_users([make_user("mallory")]): + async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as ws: + first = await ws.receive_json() + self.assertTrue(first["game"]["spectator"]) + self.assertNotIn("join_code", first["game"]) + for player in first["game"]["players"]: + self.assertNotIn("hand", player) + await ws.send_json({"action": "play", "card": "01D"}) + error = await ws.receive_json() + self.assertEqual("forbidden", error["code"]) + + @async_test + async def test_asso_pigliatutto_sweeps_the_table(self) -> None: + # A controlled scopa d'assi position: Alice's ace sweeps the whole + # table, scores no scopa and counts as the last capture. + state = ScopaState( + player_count=2, + target_score=100, + asso_pigliatutto=True, + phase=PHASE_PLAYING, + dealer=0, + turn=0, + players=[ + PlayerState( + sub="alice", name="Alice", seat=0, + hand=[Card.parse("01D"), Card.parse("09C")], + ), + PlayerState( + sub="bob", name="Bob", seat=1, + hand=[Card.parse("04D")], + ), + ], + table=[Card.parse("05C"), Card.parse("02D")], + ) + engine = platform.registry.require("scopa") + session = GameSession( + id="scopa-asso-pigliatutto-1", + game_type="scopa", + join_code="SA0001", + creator_sub="alice", + players=[ + Seat(user_sub="alice", display_name="Alice", team="A"), + Seat(user_sub="bob", display_name="Bob", team="B"), + ], + ) + session.state = state + await game_store.save(session) + + ws_transport = ASGIWebSocketTransport(app=app) + async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client: + with ws_users([make_user("alice")]): + async with aconnect_ws(f"/ws/games/{session.id}", ws_client) as ws: + first = await ws.receive_json() + self.assertEqual(0, first["game"]["turn"]) + self.assertTrue(first["game"].get("your_turn")) + await ws.send_json( + {"action": "play", "card": "01D", "capture": ["05C", "02D"]} + ) + update = await ws.receive_json() + game = update["game"] + self.assertEqual([], game["table"]) + self.assertEqual(3, game["players"][0]["captured_count"]) + self.assertEqual(0, game["players"][0]["scope"]) + self.assertEqual(0, game["scores"]["A"]) + self.assertTrue(game["last_move"]["pigliatutto"]) + self.assertFalse(game["last_move"]["scopa"]) + + @async_test + async def test_hand_end_ack_flow(self) -> None: + # A one-play hand with target 100 so the match continues to the + # scoring summary. + state = ScopaState( + player_count=2, + target_score=100, + phase=PHASE_PLAYING, + dealer=0, + turn=0, + players=[ + PlayerState( + sub="alice", name="Alice", seat=0, + hand=[Card.parse("02D")], + ), + PlayerState(sub="bob", name="Bob", seat=1), + ], + table=[Card.parse("02C")], + ) + session = GameSession( + id="scopa-hand-end-1", + game_type="scopa", + join_code="SH0001", + creator_sub="alice", + players=[ + Seat(user_sub="alice", display_name="Alice", team="A"), + Seat(user_sub="bob", display_name="Bob", team="B"), + ], + ) + session.state = state + await game_store.save(session) + + ws_transport = ASGIWebSocketTransport(app=app) + async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client: + async with contextlib.AsyncExitStack() as stack: + with ws_users([make_user("alice"), make_user("bob")]): + sockets = [ + await stack.enter_async_context( + aconnect_ws(f"/ws/games/{session.id}", ws_client) + ) + for _ in range(2) + ] + for ws in sockets: + await ws.receive_json() + + # Alice captures the last table card: the hand ends. + await sockets[0].send_json( + {"action": "play", "card": "02D", "capture": ["02C"]} + ) + summaries = [await ws.receive_json() for ws in sockets] + for summary in summaries: + self.assertEqual("state", summary["type"]) + self.assertEqual(PHASE_HAND_END, summary["game"]["phase"]) + self.assertEqual([], summary["game"]["acknowledged"]) + self.assertIsNotNone(summary["game"]["hand_end_deadline"]) + hand = summary["game"]["last_hand"] + self.assertEqual({"A": 2, "B": 0}, hand["cards"]) + self.assertEqual("A", hand["award"]["carte"]) + + # Both acknowledge; the second ack deals the next hand. + await sockets[0].send_json({"action": "ack"}) + updates = [await ws.receive_json() for ws in sockets] + for update in updates: + self.assertEqual(PHASE_HAND_END, update["game"]["phase"]) + self.assertEqual([0], update["game"]["acknowledged"]) + await sockets[1].send_json({"action": "ack"}) + updates = [await ws.receive_json() for ws in sockets] + for update in updates: + self.assertEqual(PHASE_PLAYING, update["game"]["phase"]) + self.assertEqual(2, update["game"]["hand_number"]) + for player in update["game"]["players"]: + self.assertEqual(3, player["cards_left"]) + + +class ScopaTurnTimeoutTest(unittest.TestCase): + @async_test + async def test_turn_timeout_auto_plays_a_card(self) -> None: + engine = platform.registry.require("scopa") + session = GameSession( + id="scopa-turn-timeout-1", + game_type=engine.id, + join_code="ST0001", + creator_sub="alice", + players=[ + Seat(user_sub="alice", display_name="Alice", team="A"), + ], + ) + engine.create(session, {}) + session.state.turn_timeout = 1 + engine.join(session, "bob", "Bob") + await game_store.save(session) + + ws_transport = ASGIWebSocketTransport(app=app) + async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client: + with ws_users([make_user("alice")]): + async with aconnect_ws(f"/ws/games/{session.id}", ws_client) as ws: + first = await ws.receive_json() + # Bob (seat 1) is first to act and never connects. + self.assertEqual(1, first["game"]["turn"]) + self.assertIsNotNone(first["game"]["turn_deadline"]) + + update = None + for _ in range(20): + try: + update = await asyncio.wait_for(ws.receive_json(), timeout=2) + except asyncio.TimeoutError: + break + if ( + update.get("type") == "state" + and update["game"]["last_move"] is not None + ): + break + self.assertIsNotNone(update) + assert update is not None + # The absent player was auto-played: the move is on the + # board and the turn advanced. + self.assertEqual(1, update["game"]["last_move"]["seat"]) + self.assertEqual(0, update["game"]["turn"]) + players = {p["seat"]: p for p in update["game"]["players"]} + self.assertEqual(2, players[1]["cards_left"]) + self.assertEqual(3, players[0]["cards_left"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/server/tests/test_stats.py b/server/tests/test_stats.py index fef38c7..b838aa0 100644 --- a/server/tests/test_stats.py +++ b/server/tests/test_stats.py @@ -19,6 +19,7 @@ from tavolo.platform.elo import INITIAL_RATING from tavolo.platform.errors import IllegalMove from tavolo.platform.models import Match, MatchPlayer, PlayerRating from tavolo.platform.stats import save_match_result +from tavolo.scopa import engine as scopa_rules from tavolo.scopone import engine as rules from tests.helpers import async_test, oidc_user @@ -318,5 +319,132 @@ class ChiamataStatsTest(unittest.TestCase): ) +class ScopaStatsTest(unittest.TestCase): + async def _finished_session( + self, player_count: int = 2, asso_pigliatutto: bool = False + ) -> GameSession: + engine = platform.registry.require( + "scopa" if player_count == 2 else "scopa_a_quattro" + ) + session = GameSession( + id=f"stats-scopa-{player_count}", + game_type=engine.id, + join_code="SC0001", + creator_sub="alice", + players=[Seat(user_sub="alice", display_name="Alice")], + ) + engine.create( + session, + {"target_score": 1, "asso_pigliatutto": asso_pigliatutto}, + ) + for name in PLAYERS[1:player_count]: + engine.join(session, name, name.capitalize()) + moves = 0 + while not engine.is_finished(session) and moves < 200000: + state = session.state + if state.phase == "hand_end": + for player in state.players: + engine.handle_action(session, player.sub, "ack", {}) + continue + player = next(p for p in state.players if p.seat == state.turn) + played = player.hand[0] + options = scopa_rules.legal_captures( + state.table, played, + asso_pigliatutto=state.asso_pigliatutto, + ) + payload = {"card": played.code} + if options: + payload["capture"] = [c.code for c in options[0]] + engine.handle_action(session, player.sub, "play", payload) + moves += 1 + assert engine.is_finished(session) + return session + + @async_test + async def test_finished_two_player_match_persisted_with_summary(self) -> None: + await tortoise_mixin._bind() + ctx = tortoise_mixin._ctx + assert ctx is not None + engine = platform.registry.require("scopa") + session = await self._finished_session(2, asso_pigliatutto=True) + with ctx: + await save_match_result(session, engine) + self.assertEqual(1, await Match.all().count()) + self.assertEqual(2, await MatchPlayer.all().count()) + + match = await Match.all().first() + assert match is not None + self.assertEqual("scopa", match.game_type) + summary = match.result + for key in ( + "team_a_score", + "team_b_score", + "winner_team", + "target_score", + "asso_pigliatutto", + "hands_played", + "hand_scores", + ): + self.assertIn(key, summary) + self.assertTrue(summary["asso_pigliatutto"]) + + winner_team = summary["winner_team"] + players = {p.user_sub: p for p in await MatchPlayer.all()} + for row in players.values(): + self.assertEqual(row.won, row.team == winner_team) + # A two-player match moves Elo by 16 points each. + ratings = {r.user_sub: r.rating for r in await PlayerRating.all()} + self.assertEqual( + {INITIAL_RATING + 16, INITIAL_RATING - 16}, set(ratings.values()) + ) + + @async_test + async def test_finished_four_player_match_persisted_with_teams(self) -> None: + await tortoise_mixin._bind() + ctx = tortoise_mixin._ctx + assert ctx is not None + engine = platform.registry.require("scopa_a_quattro") + session = await self._finished_session(4) + with ctx: + await save_match_result(session, engine) + self.assertEqual(4, await MatchPlayer.all().count()) + match = await Match.all().first() + assert match is not None + self.assertEqual("scopa_a_quattro", match.game_type) + players = await MatchPlayer.all() + self.assertEqual({"A", "B"}, {p.team for p in players}) + # Partnership Elo: the winning side gains 16 each. + ratings = {r.user_sub: r.rating for r in await PlayerRating.all()} + winners = {p.user_sub for p in players if p.won} + self.assertEqual( + {INITIAL_RATING + 16}, {ratings[sub] for sub in winners} + ) + self.assertEqual( + {INITIAL_RATING - 16}, + {ratings[sub] for sub in ratings if sub not in winners}, + ) + + @async_test + async def test_history_exposes_scopa_result(self) -> None: + await tortoise_mixin._bind() + ctx = tortoise_mixin._ctx + assert ctx is not None + engine = platform.registry.require("scopa") + session = await self._finished_session(2) + with ctx: + await save_match_result(session, engine) + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: + with oidc_user("alice"): + history = await client.get("/api/me/matches") + self.assertEqual(200, history.status_code) + results = history.json()["results"] + self.assertEqual(1, len(results)) + self.assertEqual("scopa", results[0]["game_type"]) + self.assertIn("team_a_score", results[0]["result"]) + self.assertIn("your_elo_delta", results[0]) + self.assertEqual(2, len(results[0]["players"])) + + if __name__ == "__main__": unittest.main() diff --git a/web/src/components/summary.rs b/web/src/components/summary.rs index 5260dc9..28b63d0 100644 --- a/web/src/components/summary.rs +++ b/web/src/components/summary.rs @@ -4,27 +4,56 @@ //! language, how each scoring category played out and how the running //! totals moved toward the target. Every player must acknowledge it before //! the next hand is dealt (or the server-side timeout deals anyway). +//! +//! The sides are named from the game: the players' names for individual +//! games (classic two-player scopa), "Team A"/"Team B" for partnership +//! games (scopone scientifico, scopa a quattro). use sycamore::prelude::*; use crate::components::card::{card_back, card_img}; -use crate::model::{GameView, HandSummary, Scores}; +use crate::model::{is_individual_game, GameView, HandSummary, Scores}; use crate::ws::GameSocket; +/// Names of the two sides: the players' names for games without +/// partnerships, "Team A"/"Team B" otherwise. +fn side_names(game: &GameView) -> (String, String) { + if is_individual_game(&game.game_type) { + ( + game.player_name_for_team("A"), + game.player_name_for_team("B"), + ) + } else { + ("Team A".to_string(), "Team B".to_string()) + } +} + +/// The display label for the side that won a category ("A"/"B"). +fn side_label(game: &GameView, team: &str) -> String { + let (a, b) = side_names(game); + match team { + "A" => a, + "B" => b, + other => other.to_string(), + } +} + /// One explanatory row: icon, title, plain-language sentence, points chip. +/// `team` is the winning side's letter, or `None` for a tie. fn award_row( icon: View, title: &'static str, text: String, - winner: Option, + team: Option<&str>, + name: &str, points: &'static str, ) -> View { - let cls = match winner.as_deref() { + let cls = match team { Some("A") => "score-row team-a", Some("B") => "score-row team-b", _ => "score-row tie", }; - let chip = match &winner { - Some(t) => format!("Team {t} {points}"), + let chip = match team { + Some(_) => format!("{name} {points}"), None => "tie".to_string(), }; view! { @@ -48,69 +77,109 @@ fn winner_first(a: i32, b: i32, winner: Option<&String>) -> (i32, i32) { } /// The five scoring rows of a completed hand. -pub fn summary_rows(summary: HandSummary) -> View { +pub fn summary_rows(game: &GameView, summary: HandSummary) -> View { + // Carte + let carte = summary.award.carte.clone(); + let carte_name = carte + .as_deref() + .map(|t| side_label(game, t)) + .unwrap_or_default(); + let carte_text = match &carte { + Some(t) => { + let (w, l) = winner_first(summary.cards.a, summary.cards.b, Some(t)); + format!( + "{} captured more cards ({w} vs {l})", + side_label(game, t) + ) + } + None => format!( + "Both sides captured {} cards — no point", + summary.cards.a + ), + }; + // Denara + let denara = summary.award.denara.clone(); + let denara_name = denara + .as_deref() + .map(|t| side_label(game, t)) + .unwrap_or_default(); + let denara_text = match &denara { + Some(t) => { + let (w, l) = winner_first(summary.denara.a, summary.denara.b, Some(t)); + format!( + "{} collected more denari cards ({w} vs {l})", + side_label(game, t) + ) + } + None => format!( + "Both sides collected {} denari cards — no point", + summary.denara.a + ), + }; + // Settebello + let settebello = summary.award.settebello.clone(); + let settebello_name = settebello + .as_deref() + .map(|t| side_label(game, t)) + .unwrap_or_default(); + let settebello_text = match &settebello { + Some(t) => format!( + "{} captured the 7 of denari — the Settebello", + side_label(game, t) + ), + None => "Nobody captured the Settebello".to_string(), + }; + // Primiera + let primiera = summary.award.primiera.clone(); + let primiera_name = primiera + .as_deref() + .map(|t| side_label(game, t)) + .unwrap_or_default(); + let primiera_text = match &primiera { + Some(t) => { + let (w, l) = winner_first(summary.primiera.a, summary.primiera.b, Some(t)); + format!( + "{} holds the strongest primiera ({w} vs {l})", + side_label(game, t) + ) + } + None => format!( + "Both primiere are worth {} — no point", + summary.primiera.a + ), + }; + let rows: Vec = vec![ - // Carte award_row( card_back("score-mini"), "Carte", - match &summary.award.carte { - Some(t) => { - let (w, l) = winner_first(summary.cards.a, summary.cards.b, Some(t)); - format!("Team {t} captured more cards ({w} vs {l})") - } - None => format!( - "Both teams captured {} cards — no point", - summary.cards.a - ), - }, - summary.award.carte.clone(), + carte_text, + carte.as_deref(), + &carte_name, "+1", ), - // Denara award_row( card_img("02D".to_string(), "score-mini"), - "Denari", - match &summary.award.denara { - Some(t) => { - let (w, l) = winner_first(summary.denara.a, summary.denara.b, Some(t)); - format!("Team {t} collected more denari cards ({w} vs {l})") - } - None => format!( - "Both teams collected {} denari cards — no point", - summary.denara.a - ), - }, - summary.award.denara.clone(), + "Denara", + denara_text, + denara.as_deref(), + &denara_name, "+1", ), - // Settebello award_row( card_img("07D".to_string(), "score-mini"), "Settebello", - match &summary.award.settebello { - Some(t) => format!("Team {t} captured the 7 of denari — the Settebello"), - None => "Nobody captured the Settebello".to_string(), - }, - summary.award.settebello.clone(), + settebello_text, + settebello.as_deref(), + &settebello_name, "+1", ), - // Primiera award_row( card_img("10D".to_string(), "score-mini"), "Primiera", - match &summary.award.primiera { - Some(t) => { - let (w, l) = - winner_first(summary.primiera.a, summary.primiera.b, Some(t)); - format!("Team {t} holds the strongest primiera ({w} vs {l})") - } - None => format!( - "Both primiere are worth {} — no point", - summary.primiera.a - ), - }, - summary.award.primiera.clone(), + primiera_text, + primiera.as_deref(), + &primiera_name, "+1", ), // Napola (only when the rule is enabled for this match) @@ -124,15 +193,17 @@ pub fn summary_rows(summary: HandSummary) -> View { }; let text = match (&summary.award.napola, n) { (Some(t), 10) => format!( - "Team {t} swept the whole denari suit — napola! Instant match win" + "{} swept the whole denari suit — napola! Instant match win", + side_label(game, t) ), (Some(t), n) => format!( - "Team {t} captured {n} consecutive denari from the ace" + "{} captured {n} consecutive denari from the ace", + side_label(game, t) ), (None, _) => "No napola this hand".to_string(), }; let chip = match &summary.award.napola { - Some(t) => format!("Team {t} +{n}"), + Some(t) => format!("{} +{n}", side_label(game, t)), None => "tie".to_string(), }; let cls = match summary.award.napola.as_deref() { @@ -154,12 +225,13 @@ pub fn summary_rows(summary: HandSummary) -> View { }, // Scope { + let (name_a, name_b) = side_names(game); let a = summary.scope.a; let b = summary.scope.b; let text = if a == 0 && b == 0 { "No scope this hand".to_string() } else { - format!("Team A made {a} scope · Team B made {b} scope") + format!("{name_a} made {a} scope · {name_b} made {b} scope") }; let chip = format!("+{a} · +{b}"); view! { @@ -193,10 +265,11 @@ pub fn totals(game: &GameView) -> View { .as_ref() .map(|s| (s.team_a_points, s.team_b_points)) .unwrap_or((0, 0)); + let (name_a, name_b) = side_names(game); view! { div(class="totals") { div(class="total-row team-a") { - span(class="total-label") { "Team A" } + span(class="total-label") { (name_a) } div(class="progress") { div(class="progress-fill", style=format!("width: {pct_a}%")) {} } @@ -206,7 +279,7 @@ pub fn totals(game: &GameView) -> View { } } div(class="total-row team-b") { - span(class="total-label") { "Team B" } + span(class="total-label") { (name_b) } div(class="progress") { div(class="progress-fill", style=format!("width: {pct_b}%")) {} } @@ -256,7 +329,7 @@ pub fn hand_summary_modal( } }); - let rows = summary_rows(summary.clone()); + let rows = summary_rows(&game, summary.clone()); let totals_view = totals(&game); let title = format!("Hand {} — results", summary.hand); diff --git a/web/src/model.rs b/web/src/model.rs index a3e3267..deec076 100644 --- a/web/src/model.rs +++ b/web/src/model.rs @@ -193,6 +193,9 @@ pub struct GameView { pub winner: Option, #[serde(default)] pub table: Vec, + /// Scopa: face-down cards left in the deck for the current hand. + #[serde(default)] + pub deck_count: usize, #[serde(default)] pub players: Vec, #[serde(default)] @@ -553,6 +556,22 @@ pub fn fallback_game_types() -> Vec { max_players: 4, options_schema: serde_json::Value::Null, }, + GameTypeInfo { + id: "scopa".to_string(), + name: "Scopa".to_string(), + description: String::new(), + min_players: 2, + max_players: 2, + options_schema: serde_json::Value::Null, + }, + GameTypeInfo { + id: "scopa_a_quattro".to_string(), + name: "Scopa a quattro".to_string(), + description: String::new(), + min_players: 4, + max_players: 4, + options_schema: serde_json::Value::Null, + }, GameTypeInfo { id: "briscola".to_string(), name: "Briscola".to_string(), @@ -863,6 +882,12 @@ pub fn is_briscola(game_type: &str) -> bool { game_type == "briscola" || game_type == "briscola_a_quattro" } +/// Whether a game type is one of the scopa variants (classic two-player +/// scopa and four-player partnership scopa a quattro). +pub fn is_scopa(game_type: &str) -> bool { + game_type == "scopa" || game_type == "scopa_a_quattro" +} + /// Whether a game type is five-player briscola chiamata, which has its /// own table view (auction, secret partner, five seats). pub fn is_chiamata(game_type: &str) -> bool { @@ -888,10 +913,10 @@ pub fn rank_name(rank: i32) -> String { } /// Whether a game is played by two individuals rather than partnerships -/// (classic two-player briscola): team labels and team messages are -/// replaced by the players' names. +/// (classic two-player briscola, classic two-player scopa): team labels +/// and team messages are replaced by the players' names. pub fn is_individual_game(game_type: &str) -> bool { - game_type == "briscola" + game_type == "briscola" || game_type == "scopa" } /// Human-readable suit name for a one-letter code (`D` -> "denari"). @@ -1408,6 +1433,14 @@ mod tests { assert_eq!("?", suit_name("?")); } + #[test] + fn is_scopa_matches_both_variants() { + assert!(is_scopa("scopa")); + assert!(is_scopa("scopa_a_quattro")); + assert!(!is_scopa("scopone_scientifico")); + assert!(!is_scopa("briscola")); + } + #[test] fn rank_names_cover_the_ladder() { assert_eq!("Asso", rank_name(1)); @@ -1419,9 +1452,11 @@ mod tests { } #[test] - fn is_individual_game_only_matches_two_player_briscola() { + fn is_individual_game_matches_two_player_games() { assert!(is_individual_game("briscola")); + assert!(is_individual_game("scopa")); assert!(!is_individual_game("briscola_a_quattro")); + assert!(!is_individual_game("scopa_a_quattro")); assert!(!is_individual_game("scopone_scientifico")); } @@ -1485,6 +1520,8 @@ mod tests { assert_eq!( vec![ "scopone_scientifico", + "scopa", + "scopa_a_quattro", "briscola", "briscola_a_quattro", "briscola_chiamata", @@ -1493,7 +1530,7 @@ mod tests { ); // The lobby's first game (the creation default) must exist. assert_eq!("scopone_scientifico", types[0].id); - assert_eq!(5, types[3].max_players); + assert_eq!(5, types[5].max_players); } #[test] @@ -1724,4 +1761,90 @@ mod tests { assert_eq!("Difensori", b); assert_eq!(Some("Chiamante".to_string()), g.winner); } + + #[test] + fn game_view_parses_scopa_payload() { + let g: GameView = serde_json::from_value(json!({ + "id": "s1", + "join_code": "ABC123", + "game_type": "scopa", + "phase": "playing", + "target_score": 11, + "asso_pigliatutto": true, + "player_count": 2, + "hand_number": 1, + "dealer": 0, + "turn": 1, + "deck_count": 24, + "scores": {"A": 0, "B": 0}, + "table": ["05C", "02D", "03S", "10B"], + "players": [ + {"sub": "alice", "name": "Alice", "seat": 0, "team": "A", + "cards_left": 3, "captured_count": 0, "scope": 0}, + {"sub": "bob", "name": "Bob", "seat": 1, "team": "B", + "cards_left": 3, "captured_count": 0, "scope": 0, + "hand": ["01D", "06C", "09S"]}, + ], + "last_hand": null, + "last_move": {"seat": 0, "name": "Alice", "card": "04D", + "captured": ["01C", "03C"], "scopa": true, + "pigliatutto": false}, + "acknowledged": [], + "hand_end_deadline": null, + "turn_deadline": "2026-01-01T00:00:30+00:00", + "your_turn": true, + "legal_moves": {"01D": [["01C"]]}, + })) + .unwrap(); + assert!(is_scopa(&g.game_type)); + assert!(is_individual_game(&g.game_type)); + assert!(g.asso_pigliatutto); + assert_eq!(2, g.player_count); + assert_eq!(24, g.deck_count); + assert_eq!(4, g.table.len()); + assert_eq!(Some(vec!["01D".to_string(), "06C".to_string(), "09S".to_string()]), + g.players[1].hand.clone()); + let mv = g.last_move.unwrap(); + assert!(mv.scopa); + assert_eq!(2, mv.captured.len()); + } + + #[test] + fn game_view_parses_scopa_a_quattro_hand_summary() { + let g: GameView = serde_json::from_value(json!({ + "id": "s2", + "game_type": "scopa_a_quattro", + "phase": "hand_end", + "player_count": 4, + "scores": {"A": 3, "B": 0}, + "players": [ + {"sub": "a", "name": "Alice", "seat": 0, "team": "A"}, + {"sub": "b", "name": "Bob", "seat": 1, "team": "B"}, + {"sub": "c", "name": "Carol", "seat": 2, "team": "A"}, + {"sub": "d", "name": "Dave", "seat": 3, "team": "B"}, + ], + "last_hand": { + "cards": {"A": 21, "B": 19}, + "denara": {"A": 6, "B": 4}, + "settebello": {"A": true, "B": false}, + "primiera": {"A": 70, "B": 60}, + "scope": {"A": 1, "B": 0}, + "award": { + "carte": "A", "denara": "A", "settebello": "A", + "primiera": "A", + }, + "hand": 1, + "team_a_points": 3, + "team_b_points": 0, + }, + })) + .unwrap(); + assert!(is_scopa(&g.game_type)); + assert!(!is_individual_game(&g.game_type)); + let hand = g.last_hand.unwrap(); + assert_eq!(21, hand.cards.a); + assert_eq!(Some("A".to_string()), hand.award.carte); + assert!(hand.napola.is_none()); + assert_eq!(3, hand.team_a_points); + } } diff --git a/web/src/pages/game.rs b/web/src/pages/game.rs index 3bdf91f..6839101 100644 --- a/web/src/pages/game.rs +++ b/web/src/pages/game.rs @@ -12,11 +12,12 @@ use crate::components::copy::CopyButton; use crate::components::summary::{hand_summary_modal, summary_rows}; use crate::components::toast::toast; use crate::model::{ - card_label, hand_sort_key, is_briscola, is_chiamata, is_individual_game, GameView, MoveView, - PlayerView, Scores, ServerMessage, + card_label, hand_sort_key, is_briscola, is_chiamata, is_individual_game, is_scopa, GameView, + MoveView, PlayerView, Scores, ServerMessage, }; use crate::pages::briscola::briscola_table; use crate::pages::chiamata::{call_picker, chiamata_table}; +use crate::pages::scopa::scopa_table; use crate::ws::{self, GameSocket}; fn send_play(socket: Signal>, card: String, capture: Option>) { @@ -65,7 +66,7 @@ fn seat_panel(game: GameView, seat: usize, position: &'static str) -> View { } /// The last-move announcement strip. -fn move_banner(mv: MoveView) -> View { +pub(crate) fn move_banner(mv: MoveView) -> View { let name = mv.name.clone(); let action = if mv.captured.is_empty() { "played" @@ -428,6 +429,10 @@ pub fn GamePage(id: String) -> View { Some(g) if is_briscola(&g.game_type) => { briscola_table(g, on_hand_card, now) } + // Scopa has its own table view for both variants. + Some(g) if is_scopa(&g.game_type) => { + scopa_table(g, on_hand_card, now) + } // The live table is scopone-specific; other games get a // fallback until they ship their own play view. Some(g) if g.game_type == "scopone_scientifico" => { @@ -450,7 +455,11 @@ pub fn GamePage(id: String) -> View { _ => view! {}, }) (move || match game.get_clone() { - Some(g) if g.phase == "hand_end" && g.game_type == "scopone_scientifico" => { + Some(g) + if g.phase == "hand_end" + && (g.game_type == "scopone_scientifico" + || is_scopa(&g.game_type)) => + { hand_summary_modal(g, socket, now) } Some(g) if g.phase == "hand_end" && is_briscola(&g.game_type) => { @@ -779,8 +788,9 @@ fn capture_picker( fn game_over_view(over: Option<(Scores, Option)>, game: Option) -> View { let briscola = game.as_ref().is_some_and(|g| is_briscola(&g.game_type)); let chiamata = game.as_ref().is_some_and(|g| is_chiamata(&g.game_type)); + let scopa = game.as_ref().is_some_and(|g| is_scopa(&g.game_type)); if let Some(g) = &game { - if g.game_type != "scopone_scientifico" && !briscola && !chiamata { + if g.game_type != "scopone_scientifico" && !briscola && !chiamata && !scopa { return view! {}; } } @@ -793,7 +803,8 @@ fn game_over_view(over: Option<(Scores, Option)>, game: Option None => view! {}, Some((scores, winner)) => { let winner = winner.unwrap_or_else(|| "?".to_string()); - // Two-player briscola has no teams: name the winner instead. + // Games without partnerships (two-player briscola and scopa) + // name the winner instead. // Briscola chiamata names the caller's side (or the defenders). let winner_label = match game.as_ref() { Some(g) if is_individual_game(&g.game_type) => { @@ -805,8 +816,8 @@ fn game_over_view(over: Option<(Scores, Option)>, game: Option let line = format!("{winner_label} wins {} – {}", scores.a, scores.b); // Spectators have no match history to look at. let spectator = game.as_ref().is_some_and(|g| g.spectator); - // Explain the final hand's (scopone) or deal's (briscola, - // briscola chiamata) scoring before the result. + // Explain the final hand's (scopone, scopa) or deal's + // (briscola, briscola chiamata) scoring before the result. let final_summary: View = if briscola || chiamata { match game.as_ref() { Some(g) => match g.last_deal.clone() { @@ -816,8 +827,8 @@ fn game_over_view(over: Option<(Scores, Option)>, game: Option None => view! {}, } } else { - match game.as_ref().and_then(|g| g.last_hand.clone()) { - Some(hand) => summary_rows(hand), + match game.as_ref().and_then(|g| g.last_hand.clone().map(|h| (g.clone(), h))) { + Some((g, hand)) => summary_rows(&g, hand), None => view! {}, } }; diff --git a/web/src/pages/lobby.rs b/web/src/pages/lobby.rs index 0f1f675..ae2fc31 100644 --- a/web/src/pages/lobby.rs +++ b/web/src/pages/lobby.rs @@ -514,7 +514,7 @@ pub fn LobbyPage() -> View { div(class="lobby-hero text-center") { h1(class="display-5 fw-bold") { "tavolo" } p(class="lead text-secondary mb-0") { - "Play scopone scientifico and briscola with friends." + "Play scopone, scopa and briscola with friends." } } // Ongoing games: a "Continue" link back into a diff --git a/web/src/pages/mod.rs b/web/src/pages/mod.rs index 558a10e..f8213b2 100644 --- a/web/src/pages/mod.rs +++ b/web/src/pages/mod.rs @@ -5,3 +5,4 @@ pub mod history; pub mod leaderboard; pub mod lobby; pub mod login; +pub mod scopa; diff --git a/web/src/pages/scopa.rs b/web/src/pages/scopa.rs new file mode 100644 index 0000000..298abe4 --- /dev/null +++ b/web/src/pages/scopa.rs @@ -0,0 +1,237 @@ +//! Scopa table view: classic two-player and four-player partnership. +//! +//! The two variants share this view: four seats render in the usual cross +//! (partners opposite), two seats stack opponent over player. The center +//! holds the cards on the table, the face-down deck and the last-move +//! announcement. +use sycamore::prelude::*; + +use crate::components::card::{card_back, card_img}; +use crate::components::copy::CopyButton; +use crate::model::{ + hand_sort_key, is_individual_game, GameView, PlayerView, Scores, +}; +use crate::pages::game::{move_banner, watch_link}; + +fn player_for_seat(game: &GameView, seat: usize) -> Option { + game.players.iter().find(|p| p.seat == seat).cloned() +} + +/// Panel for one player seat (name, team, hidden card count, stats). +fn seat_panel(game: GameView, seat: usize, position: &'static str) -> View { + let Some(player) = player_for_seat(&game, seat) else { + return view! {}; + }; + let active = game.phase == "playing" && game.turn == seat; + let cls = format!( + "seat seat-{position}{}", + if active { " active" } else { "" } + ); + let name = player.name.clone(); + let team_badge = (!is_individual_game(&game.game_type)) + .then(|| player.team.clone()) + .flatten() + .filter(|team| !team.is_empty()) + .map(|team| view! { span(class="team-badge") { "Team " (team) } }); + let captured = player.captured_count; + let scope = player.scope; + let backs = (0..player.cards_left) + .map(|_| card_back("mini")) + .collect::>(); + view! { + div(class=cls) { + div(class="seat-name") { + (name) + (team_badge) + } + div(class="seat-cards") { (backs) } + div(class="seat-stats") { + (captured) " captured · " (scope) " scope" + } + } + } +} + +/// The face-down deck with the cards left in the current hand. +fn deck_view(game: &GameView) -> View { + let backs = (0..game.deck_count.min(3)) + .map(|_| card_back("stock-card")) + .collect::>(); + let label = format!("{} in deck", game.deck_count); + view! { + div(class="stock") { + div(class="stock-pile") { (backs) } + span(class="stock-count") { (label) } + } + } +} + +/// The main scopa table. +pub fn scopa_table( + game: GameView, + on_hand_card: impl Fn(String) + Copy + 'static, + now: Signal, +) -> View { + // Own seat: the only player entry carrying a hand. Spectators have no + // hand, so they watch from seat 0's perspective. + let viewer_seat = game + .players + .iter() + .find(|p| p.hand.is_some()) + .map(|p| p.seat) + .unwrap_or(0); + let player_count = if game.player_count > 0 { + game.player_count + } else { + game.players.len().max(1) + }; + let my_turn = game.your_turn == Some(true); + let turn_note = if game.phase == "finished" { + "Match finished".to_string() + } else if game.phase == "hand_end" { + "Hand finished".to_string() + } else if my_turn { + "Your turn".to_string() + } else { + let name = player_for_seat(&game, game.turn) + .map(|p| p.name) + .unwrap_or_default(); + format!("{name}'s turn") + }; + let turn_cls = if my_turn { "turn-note you" } else { "turn-note" }; + let countdown = game.turn_deadline.as_ref().map(|deadline| { + // A dynamic closure so only the ticking number re-renders. + let deadline_ms = js_sys::Date::parse(deadline); + view! { + span(class="turn-timer") { + "Auto-play in " + (move || { + ((deadline_ms - now.get_clone()) / 1000.0).ceil().max(0.0) as i32 + }) + "s" + } + } + }); + + let scores = game.scores.unwrap_or(Scores { a: 0, b: 0 }); + let target = game.target_score; + let hand_number = game.hand_number; + // Two-player scopa has no partnerships: show the players' names. + let score_line = if is_individual_game(&game.game_type) { + format!( + "{} {} — {} {} (target {})", + game.player_name_for_team("A"), + scores.a, + scores.b, + game.player_name_for_team("B"), + target + ) + } else { + format!("Team A {} — {} Team B (target {})", scores.a, scores.b, target) + }; + + let (grid_cls, top, left, right) = if player_count == 2 { + ( + "table-grid duo", + seat_panel(game.clone(), (viewer_seat + 1) % 2, "top"), + view! {}, + view! {}, + ) + } else { + ( + "table-grid", + seat_panel(game.clone(), (viewer_seat + 2) % 4, "top"), + seat_panel(game.clone(), (viewer_seat + 1) % 4, "left"), + seat_panel(game.clone(), (viewer_seat + 3) % 4, "right"), + ) + }; + + let viewer = player_for_seat(&game, viewer_seat); + let my_captured = viewer.as_ref().map(|p| p.captured_count).unwrap_or(0); + let my_scope = viewer.as_ref().map(|p| p.scope).unwrap_or(0); + let mut hand = viewer.and_then(|p| p.hand).unwrap_or_default(); + hand.sort_by_key(|c| hand_sort_key(c)); + let hand_cards = hand + .into_iter() + .map(|code| { + let card_view = card_img(code.clone(), "hand-card"); + view! { + button(class="hand-slot", on:click=move |_| { + if my_turn { + on_hand_card(code.clone()); + } + }) { (card_view) } + } + }) + .collect::>(); + let hint = my_turn.then(|| view! { + p(class="text-secondary small text-center") { + "Click a card to play it. If it can capture in several ways you " + "will be asked to choose." + } + }); + + // Spectators have no hand: render the pivot seat as a regular panel + // so its captured count and scope stay visible where the hand would be. + let bottom = if game.spectator { + seat_panel(game.clone(), viewer_seat, "bottom") + } else { + view! { + div(class="seat-bottom") { + div(class="seat-stats own-stats") { + (my_captured) " captured · " (my_scope) " scope" + } + div(class="hand") { (hand_cards) } + (hint) + } + } + }; + + let spectator_badge = game.spectator.then(|| view! { + span(class="spectator-badge") { "Spectating" } + }); + let rule_badge = game.asso_pigliatutto.then(|| view! { + span(class="pigliatutto-badge") { "Asso pigliatutto" } + }); + let watch_link = watch_link(&game.id); + // Precomputed so the final view's closures only capture owned views. + let banner = game.last_move.clone().map(move_banner); + let table_cards = game + .table + .iter() + .map(|c| card_img(c.clone(), "table-card")) + .collect::>(); + let empty_table = game.table.is_empty().then(|| view! { + p(class="table-empty") { "Empty table" } + }); + let deck = deck_view(&game); + + view! { + div(class="table-wrap") { + div(class="hud") { + a(href="/") { "← Lobby" } + span { "Hand " (hand_number) } + (rule_badge) + (spectator_badge) + span(class="hud-scores") { (score_line) } + span(class=turn_cls) { (turn_note) } + (countdown) + CopyButton(text=watch_link, label="Copy watch link".to_string()) + } + div(class=grid_cls) { + (top) + (left) + div(class="center") { + (banner) + div(class="table-cards") { + (table_cards) + (empty_table) + } + (deck) + } + (right) + (bottom) + } + } + } +}