100 lines
2.7 KiB
Python
100 lines
2.7 KiB
Python
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, player_id) -> 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, player_id, action, amount):
|
|
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) -> 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, player_id, interval, seed):
|
|
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) -> 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())
|