#!/usr/bin/env -S uv run --script # /// script # requires-python = ">=3.10" # dependencies = ["instantdb>=1.0.44"] # /// """ INSTANT ARENA — a real-time multiplayer terminal game, powered by InstantDB. Everyone who runs this joins the SAME shared grid. Walk around with WASD / arrow keys and grab the `$` coins to score. You see every other player move in real time, and there's a live scoreboard. The whole multiplayer layer is just one InstantDB live query (`db.subscribe_query`) plus `db.transact` for moves. Run it: uv run arena.py # play (asks for your name) uv run arena.py "Ada" # play with a name uv run arena.py --bot # add a headless AI player (no terminal UI) uv run arena.py --bot "Botniss" # ... with a name uv run arena.py --watch # spectate the live scoreboard (no TTY needed) uv run arena.py --reset # clear the arena (remove all players + coins) How the real-time layer works ----------------------------- * A background thread runs `AsyncInstant.subscribe_query({"players":{}, "coins":{}})` and keeps a local snapshot of the world fresh. * The render loop draws that snapshot at ~25fps and applies your own moves optimistically so input feels instant. * A writer thread batches your position/score changes into `db.transact(...)`. """ from __future__ import annotations import argparse import asyncio import contextlib import os import queue import random import threading import time import uuid from instantdb import AsyncInstant, Instant, InstantAPIError, id # --------------------------------------------------------------------------- # # Credentials. Env wins; otherwise we fall back to the baked-in arena so the # file is shareable as-is (see README for the security trade-off of shipping an # admin token in a throwaway game app). # --------------------------------------------------------------------------- # APP_ID = os.environ.get("INSTANT_APP_ID") or "103fde8a-637b-4c9e-a24e-e069c59b9f09" ADMIN_TOKEN = ( os.environ.get("INSTANT_APP_ADMIN_TOKEN") or "178e9c9f-46e9-4f3d-b1ff-0dc733199cbf" ) # --------------------------------------------------------------------------- # # Game constants. GRID_W/GRID_H define a FIXED shared coordinate space — every # client must agree on these or positions won't line up. # --------------------------------------------------------------------------- # GRID_W = 44 GRID_H = 16 TARGET_COINS = 14 ACTIVE_MS = 8000 # players unseen for this long are hidden / off the board HEARTBEAT_MS = 2000 # how often we re-publish our position so we stay "alive" BOT_TICK = 0.3 # seconds between bot moves # Palette: (name, ansi_index). curses.COLOR_* equals these indices. PALETTE = [ ("red", 1), ("green", 2), ("yellow", 3), ("blue", 4), ("magenta", 5), ("cyan", 6), ] COIN_CHAR = "$" BOT_NAMES = [ "Botniss", "Clawde", "Pac-Bot", "Coinbot", "Grabby", "Roomba", "Hodlr", "Sir-Loots", "Magneto", "Scrooge", ] # --------------------------------------------------------------------------- # # Small helpers # --------------------------------------------------------------------------- # def now_ms() -> int: return int(time.time() * 1000) def clamp(v: int, lo: int, hi: int) -> int: return lo if v < lo else hi if v > hi else v def name_to_char(name: str) -> str: for ch in name: if ch.isalnum(): return ch.upper() return "@" def random_empty_cell(occupied: set[tuple[int, int]]) -> tuple[int, int]: """A random grid cell not in `occupied` (best effort if the grid is packed).""" if len(occupied) >= GRID_W * GRID_H: return (random.randrange(GRID_W), random.randrange(GRID_H)) while True: cell = (random.randrange(GRID_W), random.randrange(GRID_H)) if cell not in occupied: return cell def step_toward(x: int, y: int, tx: int, ty: int) -> tuple[int, int]: """One 4-directional step from (x,y) toward (tx,ty).""" if x != tx and (y == ty or random.random() < 0.5): return (1 if tx > x else -1, 0) if y != ty: return (0, 1 if ty > y else -1) if x != tx: return (1 if tx > x else -1, 0) return (0, 0) # Fixed namespace so every client derives the SAME cold-start coin ids. COIN_SEED_NS = uuid.UUID("a1f0c0de-1abe-4abe-8abe-c0ffee000000") def deterministic_seed_chunks(db): """Cold-start seed (count == 0): identical ids + positions for every client. Two players who join a freshly reset arena at the same instant both run this and produce byte-identical rows, so the concurrent transacts upsert the same TARGET_COINS coins instead of doubling them. Coin grabs are atomic (delete + create in one transact), so the live count never dips below TARGET, which means this is the only place duplication could happen. """ rng = random.Random(0xC0FFEE) # fixed seed -> same layout everywhere occ: set[tuple[int, int]] = set() chunks = [] ts = now_ms() for i in range(TARGET_COINS): while True: cell = (rng.randrange(GRID_W), rng.randrange(GRID_H)) if cell not in occ: break occ.add(cell) cid = str(uuid.uuid5(COIN_SEED_NS, f"coin-{i}")) chunks.append(db.tx.coins[cid].update({"x": cell[0], "y": cell[1], "createdAt": ts})) return chunks def coin_topup_chunks(db, current_coins: list[dict], avoid: set[tuple[int, int]]): """Build transactions that bring the coin count up to TARGET_COINS. Tops up rather than blind-seeds so a partially-populated arena heals, and a small random jitter before calling this (see callers) keeps two simultaneous fresh joiners from both seeding a full set. """ need = TARGET_COINS - len(current_coins) if need <= 0: return [] occ = {(c["x"], c["y"]) for c in current_coins} | set(avoid) chunks = [] ts = now_ms() for _ in range(need): cx, cy = random_empty_cell(occ) occ.add((cx, cy)) chunks.append(db.tx.coins[id()].update({"x": cx, "y": cy, "createdAt": ts})) return chunks # --------------------------------------------------------------------------- # # Human client (curses) # --------------------------------------------------------------------------- # class ArenaClient: def __init__(self, name: str) -> None: self.name = name[:14] or "Player" self.char = name_to_char(self.name) self.color = random.randrange(len(PALETTE)) self.pid = id() self.db = Instant(app_id=APP_ID, admin_token=ADMIN_TOKEN) # Local, authoritative copy of *my* state (optimistic — drawn instantly). self.me = {"x": 0, "y": 0, "score": 0} self.lock = threading.Lock() self.players: list[dict] = [] # latest snapshot from subscribe_query self.coins: list[dict] = [] self.status = "connecting…" self.connected = False self.dirty = False # my position changed, needs flushing self.last_write = 0 self.txq: "queue.Queue[list]" = queue.Queue() self.stop = threading.Event() self._writer: threading.Thread | None = None # ---- lifecycle -------------------------------------------------------- # def connect(self) -> None: res = self.db.query({"players": {}, "coins": {}}) players = res.get("players", []) coins = res.get("coins", []) occ = {(c["x"], c["y"]) for c in coins} | {(p["x"], p["y"]) for p in players} self.me["x"], self.me["y"] = random_empty_cell(occ) ts = now_ms() self.db.transact( self.db.tx.players[self.pid].update( { "name": self.name, "char": self.char, "x": self.me["x"], "y": self.me["y"], "color": self.color, "score": 0, "lastSeen": ts, "sessionId": self.pid, "bot": False, } ) ) # Seed/top-up coins. Cold start uses a deterministic layout so # simultaneous joiners don't double-seed (see deterministic_seed_chunks). coins = self.db.query({"coins": {}}).get("coins", []) if not coins: # Another player may seed the same deterministic ids at the same # instant; the loser gets "Record not unique" — that's fine, their # coins are already there and the subscription will deliver them. with contextlib.suppress(InstantAPIError): self.db.transact(deterministic_seed_chunks(self.db)) else: topup = coin_topup_chunks(self.db, coins, {(self.me["x"], self.me["y"])}) if topup: self.db.transact(topup) with self.lock: self.players = players self.coins = coins self.last_write = ts threading.Thread(target=self._sub_thread, daemon=True).start() self._writer = threading.Thread(target=self._writer_thread, daemon=True) self._writer.start() def shutdown(self) -> None: self.stop.set() if self._writer is not None: self._writer.join(timeout=1.0) with contextlib.suppress(Exception): self.db.transact(self.db.tx.players[self.pid].delete()) with contextlib.suppress(Exception): self.db.close() # ---- background threads ---------------------------------------------- # def _sub_thread(self) -> None: async def run() -> None: adb = AsyncInstant(app_id=APP_ID, admin_token=ADMIN_TOKEN) try: async with adb.subscribe_query({"players": {}, "coins": {}}) as sub: async for payload in sub: if self.stop.is_set(): break if payload["type"] == "ok": data = payload["data"] or {} with self.lock: self.players = data.get("players", []) self.coins = data.get("coins", []) self.connected = True self.status = "live" elif payload["type"] == "error": with self.lock: self.connected = False self.status = f"reconnecting… ({payload['error']})" except Exception as e: # noqa: BLE001 — surface in the status line with self.lock: self.connected = False self.status = f"subscription stopped: {e}" finally: with contextlib.suppress(Exception): await adb.aclose() with contextlib.suppress(Exception): asyncio.run(run()) def _writer_thread(self) -> None: while not self.stop.is_set(): chunks: list = [] # 1) drain discrete events (coin grabs) with contextlib.suppress(queue.Empty): while True: chunks.extend(self.txq.get_nowait()) # 2) position/heartbeat flush ts = now_ms() with self.lock: need_pos = self.dirty or (ts - self.last_write >= HEARTBEAT_MS) mx, my, ms = self.me["x"], self.me["y"], self.me["score"] if need_pos: self.dirty = False self.last_write = ts if need_pos: chunks.append( self.db.tx.players[self.pid].update( {"x": mx, "y": my, "score": ms, "lastSeen": ts} ) ) if chunks: try: self.db.transact(chunks) except Exception as e: # noqa: BLE001 with self.lock: self.status = f"write error: {e}" self.stop.wait(0.05) # ---- input ------------------------------------------------------------ # def try_move(self, dx: int, dy: int) -> None: ts = now_ms() with self.lock: nx = clamp(self.me["x"] + dx, 0, GRID_W - 1) ny = clamp(self.me["y"] + dy, 0, GRID_H - 1) if (nx, ny) == (self.me["x"], self.me["y"]): return self.me["x"], self.me["y"] = nx, ny hit = next((c for c in self.coins if c["x"] == nx and c["y"] == ny), None) if hit is None: self.dirty = True return # Grabbed a coin: score, remove it, and spawn a fresh one elsewhere. self.me["score"] += 1 self.coins = [c for c in self.coins if c["id"] != hit["id"]] occ = {(c["x"], c["y"]) for c in self.coins} occ |= {(p["x"], p["y"]) for p in self.players} occ.add((nx, ny)) ncx, ncy = random_empty_cell(occ) new_id = id() self.coins = self.coins + [ {"id": new_id, "x": ncx, "y": ncy, "createdAt": ts} ] self.txq.put( [ self.db.tx.players[self.pid].update( {"x": nx, "y": ny, "score": self.me["score"], "lastSeen": ts} ), self.db.tx.coins[hit["id"]].delete(), self.db.tx.coins[new_id].update( {"x": ncx, "y": ncy, "createdAt": ts} ), ] ) self.dirty = False self.last_write = ts # ---- rendering -------------------------------------------------------- # def play(self, stdscr) -> None: import curses curses.curs_set(0) stdscr.timeout(40) # ~25fps; getch() returns -1 when idle _init_colors(curses) while not self.stop.is_set(): self._render(stdscr, curses) ch = stdscr.getch() if ch == -1 or ch == curses.KEY_RESIZE: continue if ch in (ord("q"), ord("Q")): break elif ch in (ord("w"), ord("W"), curses.KEY_UP): self.try_move(0, -1) elif ch in (ord("s"), ord("S"), curses.KEY_DOWN): self.try_move(0, 1) elif ch in (ord("a"), ord("A"), curses.KEY_LEFT): self.try_move(-1, 0) elif ch in (ord("d"), ord("D"), curses.KEY_RIGHT): self.try_move(1, 0) def _render(self, stdscr, curses) -> None: stdscr.erase() scr_h, scr_w = stdscr.getmaxyx() arena_top, arena_left = 2, 0 panel_left = GRID_W + 3 min_cols = panel_left + 30 min_rows = arena_top + GRID_H + 2 + 1 if scr_w < min_cols or scr_h < min_rows: _addstr( stdscr, 0, 0, f"Please resize your terminal to at least {min_cols}x{min_rows} " f"(currently {scr_w}x{scr_h}).", 0, scr_w, ) stdscr.refresh() return with self.lock: players = list(self.players) coins = list(self.coins) me = dict(self.me) status = self.status connected = self.connected ts = now_ms() dim = curses.A_DIM bold = curses.A_BOLD # Header _addstr(stdscr, 0, 0, " I N S T A N T A R E N A", bold, scr_w) _addstr( stdscr, 1, 0, " move: WASD / arrows grab the $ q: quit", dim, scr_w, ) # Arena border border = curses.color_pair(_BORDER_PAIR) top = "+" + "-" * GRID_W + "+" _addstr(stdscr, arena_top, arena_left, top, border, scr_w) for gy in range(GRID_H): _addstr(stdscr, arena_top + 1 + gy, arena_left, "|", border, scr_w) _addstr(stdscr, arena_top + 1 + gy, arena_left + GRID_W + 1, "|", border, scr_w) _addstr(stdscr, arena_top + 1 + GRID_H, arena_left, top, border, scr_w) def put(gx: int, gy: int, ch: str, attr: int) -> None: if 0 <= gx < GRID_W and 0 <= gy < GRID_H: _addstr(stdscr, arena_top + 1 + gy, arena_left + 1 + gx, ch, attr, scr_w) # Coins coin_attr = curses.color_pair(_COIN_PAIR) | bold for c in coins: put(c["x"], c["y"], COIN_CHAR, coin_attr) # Other players (skip mine — drawn from local state below) for p in players: if p.get("sessionId") == self.pid: continue if ts - p.get("lastSeen", 0) > ACTIVE_MS: continue attr = curses.color_pair(p.get("color", 0) % len(PALETTE) + 1) | bold put(p.get("x", 0), p.get("y", 0), (p.get("char") or "@")[:1], attr) # Me (highlighted) me_attr = curses.color_pair(self.color + 1) | bold | curses.A_REVERSE put(me["x"], me["y"], self.char, me_attr) # Scoreboard panel self._render_panel(stdscr, curses, players, coins, me, status, connected, ts, panel_left, arena_top, scr_w, scr_h) # Footer foot_row = arena_top + GRID_H + 2 dot = "●" if connected else "○" _addstr( stdscr, foot_row, 0, f" you: {self.name} score: {me['score']} {dot} {status}", dim, scr_w, ) stdscr.refresh() def _render_panel(self, stdscr, curses, players, coins, me, status, connected, ts, x0, y0, scr_w, scr_h) -> None: bold = curses.A_BOLD dim = curses.A_DIM # Merge snapshot players with my local (more up-to-date) state. board: dict[str, dict] = {} for p in players: if ts - p.get("lastSeen", 0) <= ACTIVE_MS: board[p.get("sessionId", p["id"])] = p board[self.pid] = { "sessionId": self.pid, "name": self.name, "color": self.color, "score": me["score"], } rows = sorted( board.values(), key=lambda p: (-int(p.get("score", 0)), str(p.get("name", ""))), ) top_score = rows[0].get("score", 0) if rows else 0 _addstr(stdscr, y0, x0, "SCOREBOARD", bold, scr_w) line_y = y0 + 1 max_line = y0 + GRID_H # keep inside the arena's vertical extent for i, p in enumerate(rows, 1): if line_y > max_line: break is_me = p.get("sessionId") == self.pid score = int(p.get("score", 0)) crown = "*" if score == top_score and score > 0 else " " label = str(p.get("name", "?"))[:13] if is_me: label += " (you)" text = f"{crown}{i:>2}. {label:<19}{score:>3}" attr = curses.color_pair(int(p.get("color", 0)) % len(PALETTE) + 1) if is_me: attr |= bold | curses.A_REVERSE _addstr(stdscr, line_y, x0, text, attr, scr_w) line_y += 1 # Stats at the bottom of the panel n_online = len(board) stat_y = y0 + GRID_H - 2 if stat_y > line_y + 1: _addstr(stdscr, stat_y, x0, f"players online : {n_online}", dim, scr_w) _addstr(stdscr, stat_y + 1, x0, f"coins on grid : {len(coins)}", dim, scr_w) conn_attr = curses.color_pair(_GREEN_PAIR) if connected else curses.A_DIM _addstr(stdscr, stat_y + 2, x0, "● live" if connected else "○ offline", conn_attr, scr_w) # curses color-pair ids (set up in _init_colors) _COIN_PAIR = 7 _BORDER_PAIR = 8 _GREEN_PAIR = 9 def _init_colors(curses) -> None: curses.start_color() try: curses.use_default_colors() bg = -1 except curses.error: bg = curses.COLOR_BLACK for i, (_, col) in enumerate(PALETTE): curses.init_pair(i + 1, col, bg) curses.init_pair(_COIN_PAIR, curses.COLOR_YELLOW, bg) curses.init_pair(_BORDER_PAIR, curses.COLOR_BLUE, bg) curses.init_pair(_GREEN_PAIR, curses.COLOR_GREEN, bg) def _addstr(stdscr, y: int, x: int, text: str, attr: int, scr_w: int) -> None: """addstr that never throws on edge cells and clips to the screen width.""" if x >= scr_w: return text = text[: max(0, scr_w - x)] with contextlib.suppress(Exception): stdscr.addstr(y, x, text, attr) def run_human(name: str) -> None: import curses client = ArenaClient(name) print(f"Joining INSTANT ARENA as '{client.name}' …") try: client.connect() curses.wrapper(client.play) finally: client.shutdown() # always remove our player row, even if connect() failed print(f"You left the arena. Final score: {client.me['score']}. Thanks for playing!") # --------------------------------------------------------------------------- # # Bot mode (headless, pure async — also doubles as an end-to-end smoke test) # --------------------------------------------------------------------------- # async def run_bot(name: str, life: int | None = None) -> None: adb = AsyncInstant(app_id=APP_ID, admin_token=ADMIN_TOKEN) pid = id() color = random.randrange(len(PALETTE)) char = name_to_char(name) world = {"players": [], "coins": []} stop = asyncio.Event() res = await adb.query({"players": {}, "coins": {}}) coins = res.get("coins", []) players = res.get("players", []) occ = {(c["x"], c["y"]) for c in coins} | {(p["x"], p["y"]) for p in players} me = {"x": 0, "y": 0, "score": 0} me["x"], me["y"] = random_empty_cell(occ) ts = now_ms() await adb.transact( adb.tx.players[pid].update( { "name": name[:14], "char": char, "x": me["x"], "y": me["y"], "color": color, "score": 0, "lastSeen": ts, "sessionId": pid, "bot": True, } ) ) # Seed/top-up coins. Cold start uses a deterministic layout so simultaneous # joiners don't double-seed (see deterministic_seed_chunks). coins = (await adb.query({"coins": {}})).get("coins", []) if not coins: # Loser of a simultaneous seed gets "Record not unique" — ignore it. with contextlib.suppress(InstantAPIError): await adb.transact(deterministic_seed_chunks(adb)) else: topup = coin_topup_chunks(adb, coins, {(me["x"], me["y"])}) if topup: await adb.transact(topup) async def consume() -> None: try: async with adb.subscribe_query({"players": {}, "coins": {}}) as sub: async for payload in sub: if stop.is_set(): break if payload["type"] == "ok": data = payload["data"] or {} world["players"] = data.get("players", []) world["coins"] = data.get("coins", []) elif payload["type"] == "error": print(f"[{name}] sub error: {payload['error']}", flush=True) except asyncio.CancelledError: pass consumer = asyncio.create_task(consume()) print( f"[{name}] joined as '{char}' (color {PALETTE[color][0]}) " f"at ({me['x']},{me['y']})", flush=True, ) start = now_ms() ticks = 0 try: while not stop.is_set(): coins = world["coins"] others = world["players"] ts = now_ms() if coins: target = min( coins, key=lambda c: abs(c["x"] - me["x"]) + abs(c["y"] - me["y"]), ) dx, dy = step_toward(me["x"], me["y"], target["x"], target["y"]) else: dx, dy = random.choice([(1, 0), (-1, 0), (0, 1), (0, -1)]) nx = clamp(me["x"] + dx, 0, GRID_W - 1) ny = clamp(me["y"] + dy, 0, GRID_H - 1) if (nx, ny) == (me["x"], me["y"]): dx, dy = random.choice([(1, 0), (-1, 0), (0, 1), (0, -1)]) nx = clamp(me["x"] + dx, 0, GRID_W - 1) ny = clamp(me["y"] + dy, 0, GRID_H - 1) me["x"], me["y"] = nx, ny hit = next((c for c in coins if c["x"] == nx and c["y"] == ny), None) if hit is not None: me["score"] += 1 occ = {(c["x"], c["y"]) for c in coins if c["id"] != hit["id"]} occ |= {(p["x"], p["y"]) for p in others} occ.add((nx, ny)) ncx, ncy = random_empty_cell(occ) chunks = [ adb.tx.players[pid].update( {"x": nx, "y": ny, "score": me["score"], "lastSeen": ts} ), adb.tx.coins[hit["id"]].delete(), adb.tx.coins[id()].update( {"x": ncx, "y": ncy, "createdAt": ts} ), ] print(f"[{name}] grabbed a coin! score={me['score']}", flush=True) else: chunks = [ adb.tx.players[pid].update( {"x": nx, "y": ny, "score": me["score"], "lastSeen": ts} ) ] try: await adb.transact(chunks) except Exception as e: # noqa: BLE001 print(f"[{name}] write error: {e}", flush=True) ticks += 1 if ticks % 20 == 0: online = sum( 1 for p in others if ts - p.get("lastSeen", 0) < ACTIVE_MS ) print( f"[{name}] score={me['score']} pos=({me['x']},{me['y']}) " f"players_online={online} coins={len(coins)}", flush=True, ) if life is not None and now_ms() - start > life * 1000: break await asyncio.sleep(BOT_TICK) except (KeyboardInterrupt, asyncio.CancelledError): pass finally: stop.set() consumer.cancel() with contextlib.suppress(Exception): await consumer with contextlib.suppress(Exception): await adb.transact(adb.tx.players[pid].delete()) with contextlib.suppress(Exception): await adb.aclose() print(f"[{name}] left the arena. final score={me['score']}", flush=True) # --------------------------------------------------------------------------- # # Watch mode (headless spectator — great for verifying the live query) # --------------------------------------------------------------------------- # async def run_watch() -> None: adb = AsyncInstant(app_id=APP_ID, admin_token=ADMIN_TOKEN) print("Watching INSTANT ARENA live … (Ctrl-C to stop)\n", flush=True) try: async with adb.subscribe_query({"players": {}, "coins": {}}) as sub: async for payload in sub: if payload["type"] == "error": print("error:", payload["error"], flush=True) continue data = payload["data"] or {} ts = now_ms() players = [ p for p in data.get("players", []) if ts - p.get("lastSeen", 0) < ACTIVE_MS ] players.sort( key=lambda p: (-int(p.get("score", 0)), str(p.get("name", ""))) ) coins = data.get("coins", []) print( f"=== INSTANT ARENA === players online: {len(players)} " f"coins: {len(coins)}", flush=True, ) for i, p in enumerate(players[:12], 1): tag = " (bot)" if p.get("bot") else "" print( f" {i:>2}. {str(p.get('name', '?')):<16}{p.get('score', 0):>4}" f" @({p.get('x')},{p.get('y')}){tag}", flush=True, ) print(flush=True) except (KeyboardInterrupt, asyncio.CancelledError): pass finally: with contextlib.suppress(Exception): await adb.aclose() # --------------------------------------------------------------------------- # # Reset (admin maintenance) # --------------------------------------------------------------------------- # def run_reset() -> None: db = Instant(app_id=APP_ID, admin_token=ADMIN_TOKEN) res = db.query({"players": {}, "coins": {}}) players = res.get("players", []) coins = res.get("coins", []) chunks = [db.tx.players[p["id"]].delete() for p in players] chunks += [db.tx.coins[c["id"]].delete() for c in coins] if chunks: db.transact(chunks) print(f"Reset arena: removed {len(players)} player(s) and {len(coins)} coin(s).") # --------------------------------------------------------------------------- # # Entry point # --------------------------------------------------------------------------- # def prompt_name() -> str: try: name = input("Enter your name: ").strip() except EOFError: name = "" return name or "Player" def main() -> None: ap = argparse.ArgumentParser( description="INSTANT ARENA — a real-time multiplayer terminal game (InstantDB)." ) ap.add_argument("name", nargs="?", default=None, help="your display name") ap.add_argument("--bot", action="store_true", help="run a headless AI player") ap.add_argument("--watch", action="store_true", help="spectate the live scoreboard") ap.add_argument("--reset", action="store_true", help="remove all players and coins") ap.add_argument( "--life", type=int, default=None, help="bot only: seconds to play before leaving (default: forever)", ) args = ap.parse_args() if args.reset: run_reset() elif args.watch: asyncio.run(run_watch()) elif args.bot: asyncio.run(run_bot(args.name or random.choice(BOT_NAMES), life=args.life)) else: run_human(args.name or prompt_name()) if __name__ == "__main__": main()