shortdeck1.0

shortdeck_arena/agent.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .simulation import Simulation
+
+
+class Agent:
+    def __init__(self, pid: int):
+        self.pid = pid
+
+    def try_act(self, sim: 'Simulation'):
+        return None
+
+    def __str__(self):
+        return f"Agent({self.pid})"
+
+
+class HumanAgent(Agent):
+    def try_act(self, sim: 'Simulation'):
+        return None
+
+    def __str__(self):
+        return "HumanAgent"
+
+
+class RandomAgent(Agent):
+    def try_act(self, sim: 'Simulation'):
+        import random
+        info = sim.node_info()
+        choices = ["fold", "call"]
+        if info.get("bet_max", 0) > 0:
+            choices.append("bet")
+
+        action = random.choice(choices)
+        if action == "bet":
+            amt = random.randint(max(1, info.get("bet_min", 1)), info.get("bet_max", 1))
+            sim.apply_action(self.pid, "bet", amt)
+        else:
+            sim.apply_action(self.pid, action)

shortdeck_arena/card.py

@@ -0,0 +1,52 @@
+"""ShortDeck card model (6-A, 36 cards)."""
+from __future__ import annotations
+
+from enum import IntEnum
+from typing import List
+
+
+class Suit(IntEnum):
+    S = 0
+    H = 1
+    D = 2
+    C = 3
+
+    def __str__(self) -> str:
+        return "shdc"[self.value]
+
+
+class Rank(IntEnum):
+    R6 = 6
+    R7 = 7
+    R8 = 8
+    R9 = 9
+    RT = 10
+    RJ = 11
+    RQ = 12
+    RK = 13
+    RA = 14
+
+    def __str__(self):
+        if self.value <= 9:
+            return str(self.value)
+        return {10: "T", 11: "J", 12: "Q", 13: "K", 14: "A"}[self.value]
+
+
+class Card:
+    def __init__(self, rank: Rank, suit: Suit):
+        self.rank = rank
+        self.suit = suit
+
+    def __repr__(self) -> str:
+        return f"{str(self.rank)}{str(self.suit)}"
+
+    def __str__(self) -> str:
+        return self.__repr__()
+
+    @classmethod
+    def all_short(cls) -> List["Card"]:
+        cards: List[Card] = []
+        for r in [Rank.R6, Rank.R7, Rank.R8, Rank.R9, Rank.RT, Rank.RJ, Rank.RQ, Rank.RK, Rank.RA]:
+            for s in Suit:
+                cards.append(Card(r, s))
+        return cards

shortdeck_arena/main.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from pathlib import Path
+from .agent import HumanAgent, RandomAgent
+from .simulation import Simulation
+
+
+def main():
+    agents = [HumanAgent(0), RandomAgent(1)]
+    sim = Simulation(agents)
+
+    print("Player cards:")
+    for i in range(len(agents)):
+        print(i, sim.player_cards(i))
+
+    print("Random agent acting...")
+    agents[1].try_act(sim)
+    print("History:", sim.history)
+    sim.dump_data(Path.cwd() / "shortdeck_arena_history.jsonl")
+
+
+if __name__ == "__main__":
+    main()

shortdeck_arena/simulation.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import json
+import random
+from pathlib import Path
+from typing import List, Dict, Optional
+
+from typing import TYPE_CHECKING
+if TYPE_CHECKING:
+    from .agent import Agent
+from .card import Card
+
+
+class Simulation:
+    def __init__(self, agents: List[Agent]):
+        self.agents = agents
+        self.history: List[Dict] = []
+        self.cards: List[Card] = []
+        self.saved = False
+        self.new_round()
+
+    def new_round(self):
+        self.history = []
+        self.cards = Card.all_short()
+        random.shuffle(self.cards)
+        self.saved = False
+
+    def player_cards(self, pid: int) -> List[Card]:
+        return self.cards[pid * 2 : pid * 2 + 2]
+
+    def board_cards(self, street: str) -> List[Card]:
+        nplayers = len(self.agents)
+        idx_start = nplayers * 2
+        if street == "flop":
+            return self.cards[idx_start: idx_start + 3]
+        if street == "turn":
+            return self.cards[idx_start: idx_start + 4]
+        if street == "river":
+            return self.cards[idx_start: idx_start + 5]
+        return []
+
+    def node_info(self) -> Dict:
+        return {"bet_min": 1, "bet_max": 100}
+
+    def apply_action(self, pid: int, action: str, amount: Optional[int] = None):
+        self.history.append({"pid": pid, "action": action, "amount": amount})
+
+    def to_save_data(self) -> Dict:
+        players = [f"Agent{a.pid}" for a in self.agents]
+        return {
+            "history": self.history,
+            "players": players,
+            "player_cards": ["".join(str(c) for c in self.player_cards(i)) for i in range(len(self.agents))],
+            "board": "".join(str(c) for c in self.board_cards("river")),
+        }
+
+    def dump_data(self, path: Path | None = None):
+        if self.saved:
+            return
+        if path is None:
+            path = Path.cwd() / "shortdeck_arena_history.jsonl"
+        with path.open("a", encoding="utf-8") as f:
+            f.write(json.dumps(self.to_save_data(), ensure_ascii=False))
+            f.write("\n")
+        self.saved = True

shortdeck_arena_history.jsonl

@@ -0,0 +1,6 @@
+{"history": [{"pid": 0, "action": "check", "amount": null}], "players": ["Agent0", "Agent1"], "player_cards": ["8s7s", "8cQd"], "board": "Js9d6hAs8d"}
+{"history": [{"pid": 0, "action": "check", "amount": null}], "players": ["Agent0", "Agent1"], "player_cards": ["Jc6d", "9hKc"], "board": "Th7dJd6s7h"}
+{"history": [{"pid": 0, "action": "check", "amount": null}], "players": ["Agent0", "Agent1"], "player_cards": ["9c7c", "7s7d"], "board": "9sAcQc9hAh"}
+{"history": [{"pid": 0, "action": "check", "amount": null}], "players": ["Agent0", "Agent1"], "player_cards": ["Ac6s", "8hQs"], "board": "9s8s9h7cJh"}
+{"history": [{"pid": 0, "action": "check", "amount": null}], "players": ["Agent0", "Agent1"], "player_cards": ["Kc6c", "9hKs"], "board": "8dJs7h7dTs"}
+{"history": [{"pid": 0, "action": "check", "amount": null}], "players": ["Agent0", "Agent1"], "player_cards": ["AhTd", "QcTc"], "board": "9hKdKcQh7h"}

shortdeck_server/arena_adapter.py

@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+from typing import List, Dict, Optional
+from pathlib import Path
+
+from shortdeck_arena.simulation import Simulation
+from shortdeck_arena.agent import HumanAgent
+import uuid
+
+
+class ArenaGame:
+    def __init__(self, starting_stack: int = 1000, max_players: int = 6):
+        self.agents = []
+        self.player_names: List[str] = []
+        self.starting_stack = starting_stack
+        self.max_players = max_players
+        self.sim: Optional[Simulation] = None
+        self.stacks: List[int] = []
+        self.current_turn: int = 0
+        self.pot: int = 0
+        self.game_id = [str(name) for name in self.player_names]
+
+    def join_game(self, name: str) -> int:
+        if len(self.player_names) >= self.max_players:
+            raise ValueError("table full")
+        pid = len(self.player_names)
+        self.player_names.append(name)
+        agent = HumanAgent(pid)
+        self.agents.append(agent)
+        self.stacks.append(self.starting_stack)
+
+        self.sim = Simulation(self.agents)
+        return pid
+
+    def info(self, player_id) -> Dict:
+   
+        if not self.sim:
+            return {
+                "game_id": self.game_id,
+                "players": self.player_names,
+                "stacks": [],
+                "dealer_index": 0,
+                "current_turn": 0,
+                "pot": 0,
+                "stage": "preflop",
+                "actions": {},
+                "player_cards": [],
+                "board_cards": [],
+            }
+
+        player_cards = []
+        board_cards = []
+        try:
+            if player_id is not None and 0 <= player_id < len(self.agents):
+                player_cards = [str(c) for c in self.sim.player_cards(player_id)]
+        except Exception:
+            player_cards = []
+
+        try:
+    
+            board_cards = [str(c) for c in self.sim.board_cards("river")]
+        except Exception:
+            board_cards = []
+
+        return {
+            "game_id": self.game_id,
+            "players": self.player_names,
+            "stacks": list(self.stacks),
+            "dealer_index": 0,
+            "current_turn": self.current_turn,
+            "pot": self.pot,
+            "stage": "preflop",
+            "actions": {"bet_min": 1, "bet_max": 100},
+            "player_cards": player_cards,
+            "board_cards": board_cards,
+        }
+
+    def apply_action(self, pid: int, action: str, amount: Optional[int] = None):
+        if not self.sim:
+            raise ValueError("no game")
+        if pid != self.current_turn:
+            raise ValueError("not your turn")
+
+        action = action.lower()
+        if action == "check":
+            pass
+        elif action == "bet":
+            if amount is None:
+                raise ValueError("bet requires amount")
+            if amount < 0:
+                raise ValueError("invalid amount")
+            if amount > self.stacks[pid]:
+
+                amount = self.stacks[pid]
+            self.stacks[pid] -= amount
+            self.pot += amount
+        elif action == "fold":
+            self.stacks[pid] = 0
+        else:
+            raise ValueError(f"unknown action: {action}")
+
+        self.sim.apply_action(pid, action, amount)
+        self.sim.dump_data(Path.cwd() / "shortdeck_arena_history.jsonl")
+        if len(self.agents) > 0:
+            self.current_turn = (self.current_turn + 1) % len(self.agents)
+
+    @property
+    def history(self) -> List[Dict]:
+        if not self.sim:
+            return []
+        return self.sim.history

shortdeck_server/data/[].jsonl

@@ -0,0 +1,4 @@
+{"history": [{"pid": 0, "action": "check", "amount": null}]}
+{"history": [{"pid": 0, "action": "check", "amount": null}, {"pid": 1, "action": "bet", "amount": 10}]}
+{"history": [{"pid": 0, "action": "check", "amount": null}]}
+{"history": [{"pid": 0, "action": "check", "amount": null}, {"pid": 1, "action": "bet", "amount": 10}]}

shortdeck_server/main.py

@@ -0,0 +1,50 @@
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel
+
+from .persistence import append_game_history
+from .arena_adapter import ArenaGame
+
+app = FastAPI(title="shortdeck-server")
+GAME = ArenaGame()
+
+class JoinRequest(BaseModel):
+    name: str
+
+
+class JoinResponse(BaseModel):
+    player_id: int
+    name: str
+
+
+class ActionRequest(BaseModel):
+    player_id: int
+    action: str
+    amount: int | None = None
+
+
+@app.post("/join", response_model=JoinResponse)
+def join(req: JoinRequest):
+    try:
+        player_id = GAME.join_game(req.name)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e)) from e
+    return JoinResponse(player_id=player_id, name=req.name)
+
+
+@app.get("/get_game_state")
+def get_game_state(player_id):
+    try:
+        state = GAME.info(player_id)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e)) from e
+    return state
+
+
+@app.post("/apply_action")
+def apply_action(req: ActionRequest):
+    try:
+        GAME.apply_action(req.player_id, req.action, req.amount)
+        append_game_history(GAME.game_id, {"history": GAME.history})
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e)) from e
+    return {"ok": True}

shortdeck_server/persistence.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any, Dict
+
+DATA_DIR = Path(__file__).parent.joinpath("data")
+DATA_DIR.mkdir(exist_ok=True)
+
+
+def append_game_history(game_id, entry):
+    file = DATA_DIR / f"{game_id}.jsonl"
+    with file.open("a", encoding="utf-8") as f:
+        f.write(json.dumps(entry))
+        f.write("\n")
+

shortdeck_server/random_cli.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import argparse
+import random
+import time
+from typing import Any, Dict, Optional
+
+import requests
+
+
+BASE_PATH = "/get_game_state"
+APPLY_PATH = "/apply_action"
+
+
+def fetch_game_state(base_url: str, player_id: int) -> Dict[str, Any]:
+    resp = requests.get(f"{base_url.rstrip('/')}{BASE_PATH}", params={"player_id": player_id}, timeout=5)
+    resp.raise_for_status()
+    return resp.json()
+
+
+def post_action(base_url: str, player_id: int, action: str, amount: Optional[int]):
+    payload = {"player_id": player_id, "action": action}
+    if amount is not None:
+        payload["amount"] = amount
+    resp = requests.post(f"{base_url.rstrip('/')}{APPLY_PATH}", json=payload, timeout=5)
+    resp.raise_for_status()
+    return resp.json()
+
+
+def choose_random_action(info: Dict[str, Any]) -> Optional[tuple[str, Optional[int]]]:
+    actions = info.get("actions")
+    if not actions:
+        return None
+
+    is_check = actions.get("is_check") or False
+    call_amount = int(actions.get("call", 0))
+    bet_min = int(actions.get("bet_min", 0))
+    bet_max = int(actions.get("bet_max", 0))
+
+    choices = [("fold", None)]
+
+    if call_amount > 0:
+        choices.append(("call", call_amount))
+    if bet_max >= bet_min and bet_max > 0:
+        choices.append(("bet", random.randint(bet_min, bet_max)))
+        choices.append(("allin", bet_max))
+
+    return random.choice(choices)
+
+
+def run_loop(base_url: str, player_id: int, interval: float = 2.0, seed: Optional[int] = None):
+    if seed is not None:
+        random.seed(seed)
+
+    print(f"RandomAgent connecting to {base_url} as player_id={player_id}")
+
+    while True:
+        try:
+            info = fetch_game_state(base_url, player_id)
+        except Exception as e:
+            print(f"failed to fetch game state: {e}")
+            time.sleep(interval)
+            continue
+
+        action = choose_random_action(info)
+        if action is None:
+            time.sleep(interval)
+            continue
+
+        try:
+            act, amt = action
+            print(f"posting action {act} {amt}")
+            post_action(base_url, player_id, act, amt)
+        except Exception as e:
+            print(f"failed to post action: {e}")
+
+        time.sleep(interval)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--server", default="http://localhost:8000")
+    parser.add_argument("--player_id", type=int, default=1)
+    parser.add_argument("--interval", type=float, default=2.0)
+    parser.add_argument("--seed", type=int, default=None)
+    args = parser.parse_args(argv)
+
+    try:
+        run_loop(args.server, args.player_id, args.interval, args.seed)
+    except KeyboardInterrupt:
+        print("stopped")
+        return 0
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

shortdeck_server/requirements.txt

@@ -0,0 +1,4 @@
+fastapi>=0.95.0
+uvicorn>=0.22.0
+pydantic>=1.10.0
+requests>=2.30.0

shortdeck_server/tests/test_game.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from shortdeck_server.arena_adapter import ArenaGame
+
+
+def test_join_and_actions():
+    g = ArenaGame(starting_stack=100, max_players=3)
+    pid0 = g.join_game("aa")
+    pid1 = g.join_game("bb")
+    assert pid0 == 0
+    assert pid1 == 1
+
+    state = g.info()
+    assert state["stacks"] == [100, 100]
+    try:
+        g.apply_action(1, "fold")
+    except ValueError as e:
+        assert "not your turn" in str(e)
+
+    g.apply_action(0, "check")
+    assert g.current_turn == 1
+
+    g.apply_action(1, "bet", 10)
+    assert g.pot == 10
+    assert g.stacks[1] == 90
+    assert g.history[-1]["action"] == "bet"

shortdeck_server/tools/run_smoke.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+import json
+from fastapi.testclient import TestClient
+
+from shortdeck_server.main import app
+
+client = TestClient(app)
+
+def pretty(o):
+    print(json.dumps(o, ensure_ascii=False, indent=2))
+
+
+def run():
+    print('POST /join aa')
+    r = client.post('/join', json={'name':'aa'})
+    pretty(r.json())
+
+    print('POST /join bb')
+    r = client.post('/join', json={'name':'bb'})
+    pretty(r.json())
+
+    print('GET /get_game_state?player_id=0')
+    r = client.get('/get_game_state', params={'player_id':0})
+    pretty(r.json())
+
+    print('POST /apply_action check by pid=0')
+    r = client.post('/apply_action', json={'player_id':0, 'action':'check'})
+    pretty(r.json())
+
+    print('POST /apply_action bet 10 by pid=1')
+    r = client.post('/apply_action', json={'player_id':1, 'action':'bet', 'amount':10})
+    pretty(r.json())
+
+if __name__ == '__main__':
+    run()