Add configurable napola rule with instant win on a full denari sweep

This commit is contained in:
2026-09-18 19:14:54 +08:00
parent bf04a9b38d
commit c1e7f70a3b
12 changed files with 250 additions and 5 deletions
+2 -1
View File
@@ -32,7 +32,8 @@ echo "127.0.0.1 mockoauth" | sudo tee -a /etc/hosts
When a hand ends but the match is not decided, the game pauses on a When a hand ends but the match is not decided, the game pauses on a
**scoring summary screen**: every player sees how each category was won **scoring summary screen**: every player sees how each category was won
(carte, denara, settebello, primiera, scope) with the running totals and (carte, denara, settebello, primiera, scope — plus napola when enabled)
with the running totals and
must click "Understood" before the next hand is dealt. If someone is away must click "Understood" before the next hand is dealt. If someone is away
the next hand is dealt automatically after `HAND_ACK_TIMEOUT_SECONDS` the next hand is dealt automatically after `HAND_ACK_TIMEOUT_SECONDS`
(default 30s). The match-ending hand is explained on the final screen. (default 30s). The match-ending hand is explained on the final screen.
+6 -1
View File
@@ -133,7 +133,7 @@ All endpoints except `/api/health`, `/api/docs`, `/api/openapi.json`,
| Method | Path | Description | | Method | Path | Description |
|---|---|---| |---|---|---|
| `GET` | `/api/game-types` | The card games the platform can host (for the creation dropdown) | | `GET` | `/api/game-types` | The card games the platform can host (for the creation dropdown) |
| `POST` | `/api/games` | Create a lobby game. Optional body `{"game_type": "scopone_scientifico", "target_score": 11}`. Returns `{id, join_code}` | | `POST` | `/api/games` | Create a lobby game. Optional body `{"game_type": "scopone_scientifico", "target_score": 11, "napola": true}`. Returns `{id, join_code}` |
| `POST` | `/api/games/join` | Join with `{"code": "ABC123"}`. The fourth player triggers the deal | | `POST` | `/api/games/join` | Join with `{"code": "ABC123"}`. The fourth player triggers the deal |
| `GET` | `/api/games/{id}` | Personalized snapshot (only your own hand is visible) | | `GET` | `/api/games/{id}` | Personalized snapshot (only your own hand is visible) |
| `GET` | `/api/me/matches` | Cursor-paginated match history with final scores (`?limit=&cursor=&game_type=`) | | `GET` | `/api/me/matches` | Cursor-paginated match history with final scores (`?limit=&cursor=&game_type=`) |
@@ -203,6 +203,11 @@ must dismiss. A play attempted in this phase is rejected with an
- Hand points: `carte` (most captured cards), `denara` (most diamonds), - Hand points: `carte` (most captured cards), `denara` (most diamonds),
`settebello` (7♦), `primiera` (best 7/6/5/4 per suit, all four suits `settebello` (7♦), `primiera` (best 7/6/5/4 per suit, all four suits
required), plus one point per scopa. Ties award nothing. required), plus one point per scopa. Ties award nothing.
- Optional *napola* rule (per-game `napola` flag on `POST /api/games`,
default on): the longest run of consecutive denari starting from the
ace scores one point per card once it reaches three cards (A-2-3 = 3,
A-2-3-4 = 4, …). A team that captures the whole denari suit (ace to
king) wins the match instantly, regardless of the score.
- The match ends when a team reaches the target score (default 11, - The match ends when a team reaches the target score (default 11,
configurable per game) with a clear lead; a tie at or above the target is configurable per game) with a clear lead; a tie at or above the target is
broken by another hand. broken by another hand.
+44
View File
@@ -22,6 +22,10 @@ Rules implemented
cards), ``settebello`` (the 7 of diamonds), ``primiera`` (best cards), ``settebello`` (the 7 of diamonds), ``primiera`` (best
seven/five/four/three card of each suit, all four suits required), plus seven/five/four/three card of each suit, all four suits required), plus
one point per ``scopa``. Ties on carte/denara/primiera award nothing. one point per ``scopa``. Ties on carte/denara/primiera award nothing.
* Optional ``napola`` rule (enabled by default): the longest run of
consecutive denari starting from the ace scores one point per card when
it reaches at least three cards (A-2-3 = 3, A-2-3-4 = 4, ...). A team
capturing the whole denari suit (ace to king) wins the match instantly.
* The match ends when a team reaches the target score with a clear lead; a * The match ends when a team reaches the target score with a clear lead; a
tie at or above the target is broken by playing another hand. tie at or above the target is broken by playing another hand.
""" """
@@ -135,6 +139,7 @@ def create_game(
hand_ack_timeout: int = DEFAULT_HAND_ACK_TIMEOUT_SECONDS, hand_ack_timeout: int = DEFAULT_HAND_ACK_TIMEOUT_SECONDS,
turn_timeout: int = DEFAULT_TURN_TIMEOUT_SECONDS, turn_timeout: int = DEFAULT_TURN_TIMEOUT_SECONDS,
game_type: str = "scopone_scientifico", game_type: str = "scopone_scientifico",
napola: bool = True,
) -> GameState: ) -> GameState:
"""Create a lobby game with the creator seated first.""" """Create a lobby game with the creator seated first."""
if target_score < 1 or target_score > 100: if target_score < 1 or target_score > 100:
@@ -145,6 +150,7 @@ def create_game(
creator_sub=creator_sub, creator_sub=creator_sub,
game_type=game_type, game_type=game_type,
target_score=target_score, target_score=target_score,
napola=napola,
phase=PHASE_LOBBY, phase=PHASE_LOBBY,
players=[PlayerState(sub=creator_sub, name=creator_name, seat=0)], players=[PlayerState(sub=creator_sub, name=creator_name, seat=0)],
hand_ack_timeout=hand_ack_timeout, hand_ack_timeout=hand_ack_timeout,
@@ -350,6 +356,17 @@ def _end_hand(state: GameState) -> None:
a, a,
b, b,
) )
# A full napola (the whole denari suit) wins the match outright,
# regardless of the score.
napola = details.get("napola")
if isinstance(napola, dict):
for team, name in enumerate(TEAM_NAMES):
if napola.get(name) == 10:
state.phase = PHASE_FINISHED
state.winner = team
state.finished_at = datetime.now(timezone.utc).isoformat()
log.info("game %s: team %s swept the denari (napola) and wins", state.id, name)
return
reached = max(a, b) >= state.target_score reached = max(a, b) >= state.target_score
if reached and a != b: if reached and a != b:
state.phase = PHASE_FINISHED state.phase = PHASE_FINISHED
@@ -404,6 +421,22 @@ def primiera_score(captured: Sequence[Card]) -> int:
return sum(best.values()) return sum(best.values())
def napola_score(captured: Sequence[Card]) -> int:
"""Return the napola value of a capture pile.
The longest run of consecutive denari starting from the ace scores one
point per card once it reaches three cards (A-2-3 = 3, A-2-3-4 = 4,
...), so the whole suit (ace to king) is worth 10. Shorter runs score
nothing. Only one team can score a napola: the ace of denari belongs
to exactly one capture pile.
"""
ranks = {card.rank for card in captured if card.suit == "D"}
run = 0
while run + 1 in ranks:
run += 1
return run if run >= 3 else 0
def hand_points(state: GameState) -> Tuple[List[int], Dict[str, object]]: def hand_points(state: GameState) -> Tuple[List[int], Dict[str, object]]:
"""Compute the hand points for both teams (index 0 = team A).""" """Compute the hand points for both teams (index 0 = team A)."""
piles: List[List[Card]] = [[], []] piles: List[List[Card]] = [[], []]
@@ -460,6 +493,16 @@ def hand_points(state: GameState) -> Tuple[List[int], Dict[str, object]]:
"scope": {"A": scope[0], "B": scope[1]}, "scope": {"A": scope[0], "B": scope[1]},
"award": award, "award": award,
} }
# Napola (optional rule): consecutive denari from the ace. A run of 10
# means the team swept the whole suit and wins the match instantly.
if state.napola:
napola = [napola_score(piles[t]) for t in (0, 1)]
for team in (0, 1):
points[team] += napola[team]
award["napola"] = next(
(TEAM_NAMES[t] for t in (0, 1) if napola[t] > 0), None
)
details["napola"] = {"A": napola[0], "B": napola[1]}
return points, details return points, details
@@ -493,6 +536,7 @@ def state_for_player(state: GameState, sub: str) -> Dict[str, object]:
"game_type": state.game_type, "game_type": state.game_type,
"phase": state.phase, "phase": state.phase,
"target_score": state.target_score, "target_score": state.target_score,
"napola": state.napola,
"hand_number": state.hand_number, "hand_number": state.hand_number,
"dealer": state.dealer, "dealer": state.dealer,
"turn": state.turn, "turn": state.turn,
+6
View File
@@ -152,6 +152,10 @@ class GameState:
# Defaults so states serialized before game types existed still load. # Defaults so states serialized before game types existed still load.
game_type: str = "scopone_scientifico" game_type: str = "scopone_scientifico"
target_score: int = DEFAULT_TARGET_SCORE target_score: int = DEFAULT_TARGET_SCORE
# Whether the napola rule is scored (denari run from the ace; a full
# suit wins the match instantly). Default on, also for states
# serialized before the option existed.
napola: bool = True
phase: str = PHASE_LOBBY phase: str = PHASE_LOBBY
players: List[PlayerState] = field(default_factory=list) players: List[PlayerState] = field(default_factory=list)
table: List[Card] = field(default_factory=list) table: List[Card] = field(default_factory=list)
@@ -189,6 +193,7 @@ class GameState:
"creator_sub": self.creator_sub, "creator_sub": self.creator_sub,
"game_type": self.game_type, "game_type": self.game_type,
"target_score": self.target_score, "target_score": self.target_score,
"napola": self.napola,
"phase": self.phase, "phase": self.phase,
"players": [p.to_json() for p in self.players], "players": [p.to_json() for p in self.players],
"table": [c.to_json() for c in self.table], "table": [c.to_json() for c in self.table],
@@ -218,6 +223,7 @@ class GameState:
creator_sub=str(data.get("creator_sub", "")), creator_sub=str(data.get("creator_sub", "")),
game_type=str(data.get("game_type", "scopone_scientifico")), game_type=str(data.get("game_type", "scopone_scientifico")),
target_score=int(data.get("target_score", DEFAULT_TARGET_SCORE)), target_score=int(data.get("target_score", DEFAULT_TARGET_SCORE)),
napola=bool(data.get("napola", True)),
phase=str(data.get("phase", PHASE_LOBBY)), phase=str(data.get("phase", PHASE_LOBBY)),
players=[PlayerState.from_json(p) for p in data.get("players", [])], players=[PlayerState.from_json(p) for p in data.get("players", [])],
table=[Card.from_json(c) for c in data.get("table", [])], table=[Card.from_json(c) for c in data.get("table", [])],
+13
View File
@@ -52,6 +52,7 @@ def _lobby_payload(state: GameState) -> Dict[str, Any]:
"join_code": state.join_code, "join_code": state.join_code,
"game_type": state.game_type, "game_type": state.game_type,
"target_score": state.target_score, "target_score": state.target_score,
"napola": state.napola,
"phase": state.phase, "phase": state.phase,
"players": [ "players": [
{"sub": p.sub, "name": p.name, "seat": p.seat, "team": "A" if p.seat % 2 == 0 else "B"} {"sub": p.sub, "name": p.name, "seat": p.seat, "team": "A" if p.seat % 2 == 0 else "B"}
@@ -94,6 +95,12 @@ async def list_game_types(ctx: HttpContext) -> None:
"description": "One of the ids from GET /api/game-types", "description": "One of the ids from GET /api/game-types",
}, },
"target_score": {"type": "integer", "minimum": 1, "maximum": 100}, "target_score": {"type": "integer", "minimum": 1, "maximum": 100},
"napola": {
"type": "boolean",
"default": True,
"description": "Score the napola rule; a full "
"denari sweep wins the match",
},
}, },
} }
} }
@@ -123,6 +130,11 @@ async def create_game(ctx: HttpContext) -> None:
await send_error(ctx, 400, f"unknown game_type: {game_type!r}") await send_error(ctx, 400, f"unknown game_type: {game_type!r}")
return return
napola: Any = body.get("napola", True)
if not isinstance(napola, bool):
await send_error(ctx, 400, "napola must be a boolean")
return
user = oidc_mixin.get_user(ctx) user = oidc_mixin.get_user(ctx)
assert user is not None # enforced by @require_auth assert user is not None # enforced by @require_auth
game_id = str(uuid.uuid4()) game_id = str(uuid.uuid4())
@@ -137,6 +149,7 @@ async def create_game(ctx: HttpContext) -> None:
hand_ack_timeout=settings.hand_ack_timeout_seconds, hand_ack_timeout=settings.hand_ack_timeout_seconds,
turn_timeout=settings.turn_timeout_seconds, turn_timeout=settings.turn_timeout_seconds,
game_type=game_type, game_type=game_type,
napola=napola,
) )
except GameError as exc: except GameError as exc:
await send_error(ctx, 400, str(exc)) await send_error(ctx, 400, str(exc))
+89
View File
@@ -239,6 +239,95 @@ class ScoringTest(unittest.TestCase):
self.assertEqual([0, 0], points) self.assertEqual([0, 0], points)
class NapolaTest(unittest.TestCase):
def test_napola_score_runs(self) -> None:
self.assertEqual(0, engine.napola_score(
[card(c) for c in ["02D", "03D", "04D"]])) # no ace
self.assertEqual(0, engine.napola_score(
[card(c) for c in ["01D", "02D"]])) # too short
self.assertEqual(3, engine.napola_score(
[card(c) for c in ["03D", "01D", "02D"]])) # order-independent
self.assertEqual(4, engine.napola_score(
[card(c) for c in ["01D", "02D", "03D", "04D", "07C"]]))
self.assertEqual(3, engine.napola_score(
[card(c) for c in ["01D", "02D", "03D", "05D"]])) # broken run
self.assertEqual(10, engine.napola_score(
[card(f"{rank:02d}D") for rank in range(1, 11)]))
def test_hand_points_napola(self) -> None:
state = make_state(
[[], [], [], []],
table=[],
captured=[
["01D", "02D", "03D", "04C"], # seat 0, team A
["05D", "06D", "07D", "08D"], # seat 1, team B
["09D", "10D", "01C", "02C"], # seat 2, team A
["03C", "05C", "06C", "07C"], # seat 3, team B
],
)
points, details = engine.hand_points(state)
# Team A has the ace-led run 01D-03D (3 points); team B's denari
# start at the 5, so no napola. Carte tie (8 each), denara to A
# (5 vs 4), settebello to B, primiere tied at 0 (missing suits).
self.assertEqual({"A": 3, "B": 0}, details["napola"])
self.assertEqual("A", details["award"]["napola"])
self.assertEqual([4, 1], points)
def test_napola_disabled(self) -> None:
state = make_state(
[[], [], [], []],
table=[],
captured=[
["01D", "02D", "03D", "04C"],
["05D", "06D", "07D", "08D"],
["09D", "10D", "01C", "02C"],
["03C", "05C", "06C", "07C"],
],
)
state.napola = False
points, details = engine.hand_points(state)
self.assertNotIn("napola", details)
self.assertEqual([1, 1], points)
def test_full_denari_sweep_wins_match_instantly(self) -> None:
# Team A already captured the whole denari suit; the last play of
# the hand cannot capture. Team B leads 50-0, yet the napola ends
# the match in team A's favour, well below the target of 100.
state = make_state(
[["02C"], [], [], []],
table=[],
target=100,
captured=[
[f"{rank:02d}D" for rank in range(1, 11)],
[],
[],
[],
],
)
state.scores = [0, 50]
engine.play(state, "p0", "02C")
self.assertEqual(PHASE_FINISHED, state.phase)
self.assertEqual(0, state.winner)
self.assertLess(state.scores[0], 100)
self.assertEqual(10, state.hand_scores[-1]["napola"]["A"])
def test_napola_serialization_roundtrip(self) -> None:
state = make_state([["02D"], [], [], []], table=[])
self.assertTrue(state.napola)
state.napola = False
self.assertFalse(GameState.from_json(state.to_json()).napola)
# States serialized before the option existed default to enabled.
data = state.to_json()
del data["napola"]
self.assertTrue(GameState.from_json(data).napola)
def test_create_game_napola_default_and_override(self) -> None:
self.assertTrue(engine.create_game("g", "CODE42", "p0", "p0").napola)
self.assertFalse(
engine.create_game("g", "CODE42", "p0", "p0", napola=False).napola
)
class MatchFlowTest(unittest.TestCase): class MatchFlowTest(unittest.TestCase):
def test_join_starts_when_full(self) -> None: def test_join_starts_when_full(self) -> None:
state = engine.create_game("g", "CODE42", "p0", "p0", target_score=11) state = engine.create_game("g", "CODE42", "p0", "p0", target_score=11)
+18
View File
@@ -71,6 +71,24 @@ class GamesRouteTest(unittest.TestCase):
self.assertNotIn("hand", state["players"][0]) self.assertNotIn("hand", state["players"][0])
self.assertEqual(1, state["turn"]) self.assertEqual(1, state["turn"])
@async_test
async def test_create_napola_option(self) -> None:
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
with oidc_user("alice"):
default = await client.post("/api/games", json={})
self.assertEqual(201, default.status_code)
self.assertTrue(default.json()["napola"])
with oidc_user("alice"):
disabled = await client.post("/api/games", json={"napola": False})
self.assertEqual(201, disabled.status_code)
self.assertFalse(disabled.json()["napola"])
with oidc_user("alice"):
invalid = await client.post("/api/games", json={"napola": "yes"})
self.assertEqual(400, invalid.status_code)
@async_test @async_test
async def test_join_errors(self) -> None: async def test_join_errors(self) -> None:
transport = ASGITransport(app=app) transport = ASGITransport(app=app)
+2 -2
View File
@@ -35,9 +35,9 @@ pub async fn game_types() -> Result<Vec<GameTypeInfo>, String> {
Ok(page.results) Ok(page.results)
} }
pub async fn create_game(game_type: &str, target_score: i32) -> Result<GameView, String> { pub async fn create_game(game_type: &str, target_score: i32, napola: bool) -> Result<GameView, String> {
let resp = Request::post("/api/games") let resp = Request::post("/api/games")
.json(&serde_json::json!({ "game_type": game_type, "target_score": target_score })) .json(&serde_json::json!({ "game_type": game_type, "target_score": target_score, "napola": napola }))
.map_err(|e| e.to_string())? .map_err(|e| e.to_string())?
.send() .send()
.await .await
+39
View File
@@ -113,6 +113,45 @@ pub fn summary_rows(summary: HandSummary) -> View {
summary.award.primiera.clone(), summary.award.primiera.clone(),
"+1", "+1",
), ),
// Napola (only when the rule is enabled for this match)
match summary.napola {
None => view! {},
Some(napola) => {
let n = match summary.award.napola.as_deref() {
Some("A") => napola.a,
Some("B") => napola.b,
_ => 0,
};
let text = match (&summary.award.napola, n) {
(Some(t), 10) => format!(
"Team {t} swept the whole denari suit — napola! Instant match win"
),
(Some(t), n) => format!(
"Team {t} captured {n} consecutive denari from the ace"
),
(None, _) => "No napola this hand".to_string(),
};
let chip = match &summary.award.napola {
Some(t) => format!("Team {t} +{n}"),
None => "tie".to_string(),
};
let cls = match summary.award.napola.as_deref() {
Some("A") => "score-row team-a",
Some("B") => "score-row team-b",
_ => "score-row tie",
};
view! {
div(class=cls) {
div(class="score-icon") { (card_img("01D".to_string(), "score-mini")) }
div(class="score-body") {
div(class="score-title") { "Napola" }
div(class="score-text") { (text) }
}
div(class="score-points") { (chip) }
}
}
}
},
// Scope // Scope
{ {
let a = summary.scope.a; let a = summary.scope.a;
+13
View File
@@ -2,6 +2,10 @@
use serde::Deserialize; use serde::Deserialize;
use std::collections::HashMap; use std::collections::HashMap;
fn default_true() -> bool {
true
}
#[derive(Debug, Clone, Deserialize)] #[derive(Debug, Clone, Deserialize)]
#[allow(dead_code)] #[allow(dead_code)]
pub struct User { pub struct User {
@@ -75,6 +79,8 @@ pub struct Award {
pub settebello: Option<String>, pub settebello: Option<String>,
#[serde(default)] #[serde(default)]
pub primiera: Option<String>, pub primiera: Option<String>,
#[serde(default)]
pub napola: Option<String>,
} }
/// The scoring breakdown of one completed hand. /// The scoring breakdown of one completed hand.
@@ -86,6 +92,10 @@ pub struct HandSummary {
pub settebello: TeamBools, pub settebello: TeamBools,
pub primiera: TeamCounts, pub primiera: TeamCounts,
pub scope: TeamCounts, pub scope: TeamCounts,
/// Napola run lengths per team; absent when the rule is disabled (or
/// the summary predates the option).
#[serde(default)]
pub napola: Option<TeamCounts>,
pub award: Award, pub award: Award,
#[serde(default)] #[serde(default)]
pub hand: i32, pub hand: i32,
@@ -104,6 +114,9 @@ pub struct GameView {
/// Which card game this match is (id from /api/game-types). /// Which card game this match is (id from /api/game-types).
#[serde(default)] #[serde(default)]
pub game_type: String, pub game_type: String,
/// Whether the napola rule is scored in this match.
#[serde(default = "default_true")]
pub napola: bool,
pub phase: String, pub phase: String,
#[serde(default)] #[serde(default)]
pub target_score: i32, pub target_score: i32,
+12 -1
View File
@@ -24,6 +24,7 @@ pub fn LobbyPage() -> View {
let code = create_signal(String::new()); let code = create_signal(String::new());
let game_types = create_signal(fallback_game_types()); let game_types = create_signal(fallback_game_types());
let selected_game = create_signal("scopone_scientifico".to_string()); let selected_game = create_signal("scopone_scientifico".to_string());
let napola = create_signal(true);
spawn_local(async move { spawn_local(async move {
match api::me().await { match api::me().await {
@@ -47,8 +48,9 @@ pub fn LobbyPage() -> View {
let on_create = move |target: i32| { let on_create = move |target: i32| {
let game_type = selected_game.get_clone(); let game_type = selected_game.get_clone();
let napola = napola.get();
spawn_local(async move { spawn_local(async move {
match api::create_game(&game_type, target).await { match api::create_game(&game_type, target, napola).await {
Ok(game) => navigate(&format!("/game/{}", game.id)), Ok(game) => navigate(&format!("/game/{}", game.id)),
Err(e) => error.set(Some(e)), Err(e) => error.set(Some(e)),
} }
@@ -101,6 +103,15 @@ pub fn LobbyPage() -> View {
key=|g| g.id.clone(), key=|g| g.id.clone(),
) )
} }
label(class="check", r#for="napola") {
input(id="napola", r#type="checkbox", bind:checked=napola)
" Napola"
}
p(class="hint") {
"A-2-3 of denari scores 3 points, plus 1 per extra "
"consecutive denari card; sweeping the whole suit "
"wins the match instantly."
}
p { "First team to reach the target score wins." } p { "First team to reach the target score wins." }
div(class="target-buttons") { div(class="target-buttons") {
button(class="button", on:click=move |_| on_create(11)) { "Target 11" } button(class="button", on:click=move |_| on_create(11)) { "Target 11" }
+6
View File
@@ -123,6 +123,12 @@ body {
gap: 0.5rem; gap: 0.5rem;
} }
.check {
display: flex;
align-items: center;
gap: 0.4rem;
}
.join-form { .join-form {
display: flex; display: flex;
gap: 0.5rem; gap: 0.5rem;