qianrui.mmmy
353 lines
12 KiB
Python
353 lines
12 KiB
Python
"""Pure helpers for rendering observations and prompting human actions.
|
|
|
|
This module is intentionally I/O-injected and dict-based so that both the
|
|
in-process :class:`HumanAgent` and the standalone HTTP human client can share
|
|
the same presentation and validation logic.
|
|
|
|
Design rationale:
|
|
- Functions accept the *dict* form of an observation (the same payload that
|
|
:meth:`Observation.to_dict` produces and that flows over HTTP). That keeps
|
|
the helpers agnostic to whether the caller has a real ``Observation``
|
|
object or a freshly parsed JSON document.
|
|
- Reader/Writer callables are passed in (rather than reading ``stdin`` /
|
|
writing ``stdout`` directly) so the helpers stay testable and reusable in
|
|
any context (CLI, sockets, GUI, etc.).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Callable
|
|
|
|
# Type aliases keep the public function signatures self-documenting and make
|
|
# it trivial to swap in alternative IO backends (e.g. async streams).
|
|
Reader = Callable[[str], str]
|
|
Writer = Callable[[str], None]
|
|
|
|
# Mapping from internal one-letter suit codes to Unicode pip glyphs. Defined
|
|
# at module scope so it is cheap to look up and easy to override in tests.
|
|
SUIT_GLYPHS: dict[str, str] = {
|
|
"s": "\u2660", # ♠ spades
|
|
"h": "\u2665", # ♥ hearts
|
|
"c": "\u2663", # ♣ clubs
|
|
"d": "\u2666", # ♦ diamonds
|
|
}
|
|
|
|
|
|
def pretty_card(label: str) -> str:
|
|
"""Render a two-character card label like ``"8h"`` as ``"♥8"``.
|
|
|
|
Designed as a small, total function so it can be reused anywhere card
|
|
strings need to be displayed (terminal, future GUI, logs). Unknown
|
|
suits fall through unchanged so we never crash on malformed data.
|
|
"""
|
|
if not isinstance(label, str) or len(label) < 2:
|
|
return str(label)
|
|
rank, suit = label[0], label[1].lower()
|
|
glyph = SUIT_GLYPHS.get(suit)
|
|
if glyph is None:
|
|
return label
|
|
return f"{glyph}{rank}"
|
|
|
|
|
|
def _format_cards(cards: list[Any], empty_text: str) -> str:
|
|
"""Render a list of card labels using :func:`pretty_card` with separators.
|
|
|
|
Extracted as a tiny helper so both the board and hole-cards lines share
|
|
one definition of "what an empty hand looks like".
|
|
"""
|
|
if not cards:
|
|
return empty_text
|
|
return " ".join(pretty_card(str(card)) for card in cards)
|
|
|
|
|
|
def render_observation(observation: dict[str, Any]) -> str:
|
|
"""Render an observation dict as a multi-line, human-readable block.
|
|
|
|
Returning a single string (instead of writing directly) lets the caller
|
|
decide where the output should go, and keeps the function pure and easy
|
|
to unit-test.
|
|
"""
|
|
lines: list[str] = []
|
|
lines.append("=" * 60)
|
|
lines.append(
|
|
f"Game {observation.get('game_id')} | Hand #{observation.get('hand_number')} "
|
|
f"| Street: {observation.get('street')}"
|
|
)
|
|
lines.append(
|
|
f"Blinds {observation.get('small_blind')}/{observation.get('big_blind')} "
|
|
f"| Button seat: {observation.get('button_seat')} "
|
|
f"| Pot: {observation.get('pot')}"
|
|
)
|
|
|
|
board = observation.get("board") or []
|
|
hole_cards = observation.get("hole_cards") or []
|
|
lines.append(f"Board : {_format_cards(board, '(empty)')}")
|
|
lines.append(
|
|
f"Your hand : {_format_cards(hole_cards, '(none)')}"
|
|
f" (seat {observation.get('seat')}, id={observation.get('player_id')})"
|
|
)
|
|
|
|
lines.append("-" * 60)
|
|
lines.append("Players:")
|
|
current_id = observation.get("player_id")
|
|
for player in observation.get("players", []):
|
|
marker = _player_marker(player, current_id)
|
|
lines.append(
|
|
f" {marker} seat {int(player.get('seat', 0)):>2} "
|
|
f"| {str(player.get('name', '')):<16} "
|
|
f"| stack {int(player.get('stack', 0)):>6} "
|
|
f"| street_bet {int(player.get('street_bet', 0)):>6} "
|
|
f"| total_bet {int(player.get('total_bet', 0)):>6}"
|
|
)
|
|
|
|
lines.append("-" * 60)
|
|
min_raise_to = observation.get("min_raise_to")
|
|
lines.append(
|
|
f"To call: {observation.get('to_call')} "
|
|
f"| Min raise to: {min_raise_to if min_raise_to is not None else '-'}"
|
|
)
|
|
|
|
lines.append("Recent actions:")
|
|
history = observation.get("action_history") or []
|
|
if not history:
|
|
lines.append(" (no actions yet)")
|
|
else:
|
|
for record in history[-8:]:
|
|
lines.append(
|
|
f" [{str(record.get('street', '')):<7}] "
|
|
f"{str(record.get('player_id', '')):<12} "
|
|
f"-> {str(record.get('action', '')):<6} "
|
|
f"amount={record.get('amount', 0)}"
|
|
)
|
|
|
|
lines.append("=" * 60)
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def _player_marker(player: dict[str, Any], current_player_id: Any) -> str:
|
|
"""Produce a single-character marker describing a player's status.
|
|
|
|
``*`` highlights the player who is currently to act, ``F`` flags a folded
|
|
seat, ``A`` an all-in seat. Isolated as a helper to keep the rendering
|
|
loop free of cosmetic branching.
|
|
"""
|
|
if player.get("player_id") == current_player_id:
|
|
return "*"
|
|
if player.get("folded"):
|
|
return "F"
|
|
if player.get("all_in"):
|
|
return "A"
|
|
return " "
|
|
|
|
|
|
def prompt_action(
|
|
legal_actions: list[dict[str, Any]],
|
|
reader: Reader,
|
|
writer: Writer,
|
|
) -> dict[str, Any]:
|
|
"""Drive an interactive menu and return a chosen ``{action, amount}`` dict.
|
|
|
|
The function loops until a valid choice is made because, for an
|
|
interactive debugger, treating typos as fatal would be hostile. The
|
|
returned dict matches the JSON schema accepted by ``PlayerAction.from_dict``.
|
|
A trailing separator + blank line is emitted right before returning so
|
|
consecutive turns are visually separated in the terminal log.
|
|
"""
|
|
if not legal_actions:
|
|
raise RuntimeError("no legal actions available")
|
|
|
|
while True:
|
|
writer("Choose an action:\n")
|
|
for index, action in enumerate(legal_actions, start=1):
|
|
writer(f" [{index}] {format_legal_action(action)}\n")
|
|
|
|
raw = reader("Enter choice number: ").strip()
|
|
choice = _parse_choice(raw, len(legal_actions))
|
|
if choice is None:
|
|
writer("Invalid choice, please try again.\n")
|
|
continue
|
|
|
|
selected = legal_actions[choice - 1]
|
|
action_type = str(selected["action"])
|
|
|
|
if action_type in {"bet", "raise"}:
|
|
amount = _prompt_amount(
|
|
int(selected["min_amount"]),
|
|
int(selected["max_amount"]),
|
|
reader,
|
|
writer,
|
|
)
|
|
if amount is None:
|
|
# Operator cancelled the amount entry; redisplay the menu.
|
|
continue
|
|
_emit_turn_separator(writer)
|
|
return {"action": action_type, "amount": amount}
|
|
|
|
_emit_turn_separator(writer)
|
|
return {"action": action_type, "amount": int(selected.get("amount") or 0)}
|
|
|
|
|
|
def _emit_turn_separator(writer: Writer) -> None:
|
|
"""Print a divider plus a blank line to delimit consecutive turns.
|
|
|
|
Centralised so the exact glyph/length of the separator can be changed
|
|
in one place if the visual style ever needs tweaking.
|
|
"""
|
|
line = "~" * 60
|
|
writer(line + "\n\n")
|
|
|
|
|
|
# ANSI control sequence: ``ESC[2J`` clears the entire screen and ``ESC[H``
|
|
# moves the cursor back to the top-left. Kept as a module constant so any
|
|
# caller can reuse the exact same sequence and tests can monkey-patch it.
|
|
CLEAR_SCREEN_SEQUENCE = "\x1b[2J\x1b[H"
|
|
|
|
|
|
def clear_screen(writer: Writer) -> None:
|
|
"""Wipe the terminal via ANSI control sequences.
|
|
|
|
Implemented as a tiny helper rather than each caller inlining the escape
|
|
code so we have a single location to swap in alternative strategies
|
|
(e.g. printing many newlines on terminals that ignore ANSI).
|
|
"""
|
|
writer(CLEAR_SCREEN_SEQUENCE)
|
|
|
|
|
|
def render_game_state(game_state: dict[str, Any]) -> str:
|
|
"""Render a full ``GameManager.to_dict()`` snapshot for terminal display.
|
|
|
|
The resulting block is intended for the standalone HTTP human client's
|
|
``POST /game`` callback so the operator sees the up-to-date table state
|
|
plus a per-hand digest (winners, awards, showdown hole cards).
|
|
"""
|
|
lines: list[str] = []
|
|
lines.append("#" * 60)
|
|
lines.append(
|
|
f"GAME UPDATE game_id={game_state.get('game_id')} "
|
|
f"status={game_state.get('status')} hand={game_state.get('hand_number')}"
|
|
)
|
|
lines.append(
|
|
f"Blinds {game_state.get('small_blind')}/{game_state.get('big_blind')} "
|
|
f"| Button seat: {game_state.get('button_seat')} "
|
|
f"| Starting stack: {game_state.get('starting_stack')}"
|
|
)
|
|
|
|
lines.append("-" * 60)
|
|
lines.append("Stacks:")
|
|
for player in game_state.get("players", []):
|
|
flags = _player_flags(player)
|
|
lines.append(
|
|
f" seat {int(player.get('seat', 0)):>2} "
|
|
f"| {str(player.get('name', '')):<16} "
|
|
f"| stack {int(player.get('stack', 0)):>6} "
|
|
f"| {flags}"
|
|
)
|
|
|
|
hands = game_state.get("hands") or []
|
|
lines.append("-" * 60)
|
|
lines.append(f"Hands played: {len(hands)}")
|
|
for hand in hands:
|
|
lines.extend(_render_hand_digest(hand))
|
|
|
|
lines.append("#" * 60)
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def _player_flags(player: dict[str, Any]) -> str:
|
|
"""Render the boolean state of a player as a compact tag list."""
|
|
tags: list[str] = []
|
|
if player.get("folded"):
|
|
tags.append("folded")
|
|
if player.get("all_in"):
|
|
tags.append("all_in")
|
|
if not player.get("in_hand"):
|
|
tags.append("out")
|
|
return ",".join(tags) if tags else "active"
|
|
|
|
|
|
def _render_hand_digest(hand: dict[str, Any]) -> list[str]:
|
|
"""Render a single hand summary as a compact, multi-line digest.
|
|
|
|
Kept separate from :func:`render_game_state` so the per-hand format can
|
|
be reused or extended (e.g. detailed action log) without entangling
|
|
with the table-level header layout.
|
|
"""
|
|
lines: list[str] = []
|
|
lines.append(
|
|
f" Hand #{hand.get('hand_number')} "
|
|
f"| button_seat={hand.get('button_seat')} "
|
|
f"| board: {_format_cards(hand.get('board') or [], '(folded out)')}"
|
|
)
|
|
|
|
awards = hand.get("awards") or []
|
|
if not awards:
|
|
lines.append(" (no awards recorded)")
|
|
for award in awards:
|
|
winners = ", ".join(str(w) for w in award.get("winners") or [])
|
|
hand_value = award.get("hand_value") or {}
|
|
value_label = hand_value.get("name") or "-"
|
|
lines.append(
|
|
f" pot {int(award.get('amount', 0)):>6} -> "
|
|
f"{winners or '(no winner)'} ({value_label})"
|
|
)
|
|
|
|
showdown = hand.get("showdown_hands") or {}
|
|
if showdown:
|
|
lines.append(" showdown:")
|
|
for player_id, cards in showdown.items():
|
|
lines.append(
|
|
f" {player_id}: {_format_cards(cards, '(empty)')}"
|
|
)
|
|
return lines
|
|
|
|
|
|
def format_legal_action(action: dict[str, Any]) -> str:
|
|
"""Render one legal-action dict as a one-line description for the menu."""
|
|
action_type = str(action["action"])
|
|
if action_type in {"bet", "raise"}:
|
|
return (
|
|
f"{action_type} (street_total in "
|
|
f"[{action['min_amount']}, {action['max_amount']}])"
|
|
)
|
|
if action_type in {"call", "all_in"}:
|
|
return f"{action_type} {action.get('amount', 0)}"
|
|
return action_type
|
|
|
|
|
|
def _parse_choice(raw: str, upper: int) -> int | None:
|
|
"""Parse a 1-based menu index, returning ``None`` on any out-of-range input."""
|
|
if not raw.isdigit():
|
|
return None
|
|
value = int(raw)
|
|
if not 1 <= value <= upper:
|
|
return None
|
|
return value
|
|
|
|
|
|
def _prompt_amount(
|
|
min_amount: int,
|
|
max_amount: int,
|
|
reader: Reader,
|
|
writer: Writer,
|
|
) -> int | None:
|
|
"""Prompt for a bet/raise street-total in ``[min_amount, max_amount]``.
|
|
|
|
Returning ``None`` lets the caller back out of an accidental selection
|
|
(operator just presses Enter on an empty line).
|
|
"""
|
|
while True:
|
|
raw = reader(
|
|
f"Enter target street total in [{min_amount}, {max_amount}] "
|
|
f"(blank to cancel): "
|
|
).strip()
|
|
if raw == "":
|
|
return None
|
|
if not raw.lstrip("-").isdigit():
|
|
writer("Amount must be an integer.\n")
|
|
continue
|
|
value = int(raw)
|
|
if not min_amount <= value <= max_amount:
|
|
writer(f"Amount {value} out of range [{min_amount}, {max_amount}].\n")
|
|
continue
|
|
return value
|