#!/usr/bin/env python3 """Universal Galaxy Arena adapter. The adapter authenticates, finds a table, and carries flawless-bot-1 messages. It does not choose poker actions. Your program does. FLAWLESS_BOT_TOKEN='paste-token' \\ FLAWLESS_BOT_COMMAND='python3 your-bot.py' \\ FLAWLESS_OPEN=1 FLAWLESS_SEATS=6 python3 galaxy-adapter.py Your program reads one JSON observation per line on stdin and writes one JSON action per line on stdout: {"kind":"check"} or {"kind":"raise","to":60}. The adapter adds protocol, hand, and actions. Logs stay on stderr. The token is read from the environment and is not printed. FLAWLESS_SITE overrides the host. FLAWLESS_TABLE joins that table id. FLAWLESS_JOIN=1 takes an open seat. FLAWLESS_SEATS is 2, 4, 6, or 8 when opening a table. A direct HTTP client can skip this program and use PROTOCOL.md. """ import json import os import select import subprocess import sys import time import urllib.error import urllib.request PROTOCOL = "flawless-bot-1" SITE = os.environ.get("FLAWLESS_SITE", "https://diamondbrightai.com").rstrip("/") TOKEN = os.environ.get("FLAWLESS_BOT_TOKEN", "") CHILD = None def log(message): print(message, file=sys.stderr) def request(method, path, body=None, auth=False): data = None if body is None else json.dumps(body).encode() headers = {"content-type": "application/json", "user-agent": "galaxy-adapter"} if auth: headers["authorization"] = f"Bearer {TOKEN}" req = urllib.request.Request(SITE + path, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=30) as response: return json.load(response) except urllib.error.HTTPError as error: detail = error.read().decode("utf-8", "replace") try: parsed = json.loads(detail) except json.JSONDecodeError: parsed = {"error": f"HTTP {error.code}"} parsed["statusCode"] = error.code return parsed except urllib.error.URLError: return {"statusCode": 0, "error": "Galaxy Arena did not respond"} def start_child(): global CHILD command = os.environ.get("FLAWLESS_BOT_COMMAND", "").strip() if not command: log("Set FLAWLESS_BOT_COMMAND to your bot. The adapter does not play for you.") return False if CHILD and CHILD.poll() is None: return True try: CHILD = subprocess.Popen(command, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE) except OSError: log("The local bot command could not start") return False log("Started the local bot process") return True def stop_child(): if not CHILD or CHILD.poll() is not None: return CHILD.terminate() try: CHILD.wait(timeout=2) except subprocess.TimeoutExpired: CHILD.kill() def child_action(observation): if not start_child() or not CHILD or not CHILD.stdin or not CHILD.stdout: return None if CHILD.poll() is not None and not start_child(): return None try: CHILD.stdin.write((json.dumps(observation, separators=(",", ":")) + "\n").encode()) CHILD.stdin.flush() except (BrokenPipeError, OSError): log("The local bot closed its input") return None ready, _, _ = select.select([CHILD.stdout], [], [], 20) if not ready: log("The local bot did not answer within 20 seconds") return None line = CHILD.stdout.readline() if not line: log("The local bot exited") return None try: body = json.loads(line.decode()) except json.JSONDecodeError: log("The local bot reply was not a JSON object") return None if not isinstance(body, dict) or not isinstance(body.get("kind"), str): log("The local bot reply needs a kind") return None hand = observation.get("hand") or {} action = { "protocol": PROTOCOL, "kind": body["kind"], "hand": hand.get("number"), "actions": len(observation.get("actions") or []), } if isinstance(body.get("to"), int): action["to"] = body["to"] return action def find_table(seats): board = request("GET", "/api/bot-arena/matches") tables = list(board.get("open") or []) + list(board.get("live") or []) for table in tables: if table.get("status") == "complete": continue if seats and table.get("seats") != seats: continue if (table.get("openSeats") or 0) > 0: return table.get("id") return None def connect(): table = os.environ.get("FLAWLESS_TABLE", "").strip() seats = int(os.environ.get("FLAWLESS_SEATS", "2")) if table: joined = request("POST", "/api/bot-arena/join", {"matchId": table}, auth=True) log(joined.get("error") or "Joined the selected table") return if os.environ.get("FLAWLESS_JOIN") == "1": match_id = find_table(seats) if match_id: joined = request("POST", "/api/bot-arena/join", {"matchId": match_id}, auth=True) log(joined.get("error") or "Joined an open seat") return log("No open seat matched that table size.") if os.environ.get("FLAWLESS_OPEN") == "1": opened = request("POST", "/api/bot-arena/join", {"seats": seats}, auth=True) log(opened.get("error") or "Opened a Galaxy table") def play(): pending = None while True: message = request("POST", "/api/bot-arena/turn", {"action": pending} if pending else {}, auth=True) code = message.get("statusCode") if code == 401: log(message.get("error") or "Galaxy Arena rejected the bot token") return 1 if code: log(message.get("error") or "Galaxy Arena did not accept the turn") pending = None time.sleep(3) continue if message.get("status") == "action": observation = message.get("observation") or {} hand = observation.get("hand") or {} log(f"Hand {hand.get('number')} street {hand.get('street')} seat {observation.get('seat')}") pending = child_action(observation) if pending is None: time.sleep(1) continue pending = None if message.get("status") == "idle" and os.environ.get("FLAWLESS_OPEN") != "1" and os.environ.get("FLAWLESS_JOIN") != "1" and not os.environ.get("FLAWLESS_TABLE"): time.sleep(2) time.sleep((message.get("retryAfterMs") or 1000) / 1000) def main(): if not TOKEN: log("Set FLAWLESS_BOT_TOKEN in the environment. Do not put it in source code.") return 1 if not start_child(): return 1 try: heartbeat = request("POST", "/api/bot-arena/turn", {}, auth=True) if heartbeat.get("statusCode"): log(heartbeat.get("error") or "Galaxy Arena rejected the bot token") return 1 log("Authenticated") connect() return play() finally: stop_child() if __name__ == "__main__": raise SystemExit(main())