"""A live room (docs/games.md ยง8): one game of kaggle-environments' Werewolf with people, agents and bots at the seven seats, speaking the room protocol through kit/room.py. Seat "0"-"6" is Player 1-7 and the engine deals the roles (seeded by SS_ROOM's seed, else at random). A bot runs a route's or a field folder's main.py in a process of its own per seat, as in the checks, 3 s a move; a seat on auto, a person out of time and a bot that fails or answers out of turn are played by the engine's deterministic agent. Public things are events; what one seat alone may know (its role, the wolves' picks, the seer's results, the doctor's heals) is told to that seat. At the end the game is left in replays/ in the practice replays' format with a matches.json entry, so it lists as a match. SS_ROOM='{"seats": {"0": {"kind": "human", "name": "ada"}, "1": {"kind": "bot", "field": "deterministic"}}}' \ python3 -s referee/room.py Messages (events are play.py's Story events, seats 0-6, without the private reasoning): event {kind: start, seats: {seat: {name, kind}}, roles: {role: count}, seconds} to each seat {kind: role, seat, role, team, wolves}: its role card (wolves: the werewolves' seats, for a werewolf) event {kind: phase|dawn|out|talk|speech|vote_start|vote|tally, ...}, and {kind: turn, seat, what: chat|vote, seconds} before a person's or an agent's speech or day vote to a seat {kind: heal|saved|inspect|wolf_vote|wolf_pick, ...}: the doctor's, the seer's and the wolves' own night request {to, version, obs: {seat, player, role, team, wolves, day, phase, request, alive, known, events, seconds, raw}, legal}: request is chat, vote, kill, heal or inspect; known is what was told to the seat, events the public ones so far (the last 40), raw Kaggle's own raw_observation, so a Kaggle agent can play from it chat: legal [{type: chat, message: ""}], answered with the message filled in (trimmed, at most 600 characters; left empty, the seat passes) kill|heal|inspect|vote: legal [{type, target: seat}], a day vote also {type: vote, target: null} to abstain an answer may add "reasoning" (kept for the replay), or be the Kaggle agent's own action, as main.py returns it event {kind: over, winner, reason, roles, out}, then end {result, scores}: 1 for each seat of the winning team """ import json, os, random, sys ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path[:0] = [os.path.join(ROOT, 'kit'), os.path.join(ROOT, 'referee')] # kit's room.py, not this file from room import Room, Stop # noqa: E402 (first: play.py puts referee/ ahead of kit/ as it loads) import league # noqa: E402 from kaggle_environments import make # noqa: E402 from kaggle_environments.envs.werewolf.game.actions import ChatAction, HealAction, InspectAction, VoteAction # noqa: E402 from kaggle_environments.envs.werewolf.werewolf import deterministic_agent # noqa: E402 from kaggle_seat import seat as bot_seat # noqa: E402 from play import N, Story, transcript # noqa: E402 SAY, PASS = 600, 'I pass.' ACTION = {'chat': ChatAction, 'vote': VoteAction, 'kill': VoteAction, 'heal': HealAction, 'inspect': InspectAction} SECRET = {'heal': 'Doctor', 'saved': 'Doctor', 'inspect': 'Seer', 'wolf_vote': 'Werewolf', 'wolf_pick': 'Werewolf'} WAKE = {'Doctor': 0, 'Seer': 1} # the judge wakes the doctor, then the seer, then the wolves class Bots: """A folder's bot, one process per seat as in the checks: a bot may keep what it learns about its own game.""" def __init__(self, folder): self.new = lambda: bot_seat(folder, limit=3.0).child self.kids = {None: self.new()} # fails here, as Room expects, when the folder has no bot def ask(self, msg): pid = msg['obs']['raw_observation']['player_id'] if pid not in self.kids: self.kids[pid] = self.kids.pop(None, None) or self.new() return self.kids[pid].ask(msg) def close(self): for c in self.kids.values(): c.close() def request(raw): """What the engine asks of this player now and the players it may name: (chat|vote|kill|heal|inspect, ids).""" phase, role, me = raw['detailed_phase'], raw['role'], raw['player_id'] others = [p for p in raw['alive_players'] if p != me] def listed(name, key): # the request's own list of valid targets return next((v['data'][key] for v in raw['new_player_event_views'] if v['event_name'] == name and (v.get('data') or {}).get(key)), None) if phase == 'DAY_CHAT_AWAIT': return 'chat', [] if phase == 'DAY_VOTING_AWAIT': return 'vote', others if phase == 'NIGHT_AWAIT_ACTIONS': if role == 'Werewolf': return 'kill', listed('vote_request', 'valid_targets') or others if role == 'Doctor': return 'heal', listed('heal_request', 'valid_candidates') or others if role == 'Seer': return 'inspect', listed('inspect_request', 'valid_candidates') or others return None, [] def main(): conf = json.loads(os.environ.get('SS_ROOM') or '{}') seed = int(conf['seed']) if conf.get('seed') is not None else random.SystemRandom().randrange(1 << 31) random.seed(seed) # the engine's own draws: ties, who speaks and votes first room = Room(Bots, field=os.path.join(ROOT, 'field')) env = make('werewolf', configuration={'seed': seed, 'randomize_roles': True}) env.reset(N) config = json.loads(json.dumps(env.configuration, default=str)) seats = [str(i) for i in range(N)] ids = [p.id for p in env.game_state.players] role = [str(p.role.name) for p in env.game_state.players] team = [str(p.role.team) for p in env.game_state.players] wolves = [i for i in range(N) if role[i] == 'Werewolf'] names = [room.label(s) for s in seats] story, public, known, seen = Story(ids), [], {s: [] for s in seats}, 0 def tell(i, e): known[seats[i]].append(e) room.tell(seats[i], e) def publish(): nonlocal seen log = env.info['MODERATOR_OBSERVATION'] for t in range(seen, len(log)): for e in story.step(t, log[t]): e = {k: v for k, v in e.items() if k not in ('t', 'why')} # reasoning stays private until the replay if e['kind'] in SECRET: for i in range(N): if role[i] == SECRET[e['kind']]: tell(i, e) else: public.append(e) room.event(e) seen = len(log) def move(i): s = seats[i] obs = json.loads(json.dumps({**env.state[i].observation, 'step': env.state[0].observation.step}, default=str)) raw = obs['raw_observation'] what, targets = request(raw) stand_in = lambda: deterministic_agent(obs) # noqa: E731 if what is None: return stand_in() # nothing asked: the engine's no-op seat_of = {p: str(ids.index(p)) for p in ids} aim = [seat_of[p] for p in targets if p in seat_of] legal = [{'type': 'chat', 'message': ''}] if what == 'chat' else \ [{'type': what, 'target': t} for t in aim] + ([{'type': 'vote', 'target': None}] if what == 'vote' else []) base = {'day': raw['day'], 'phase': raw['game_state_phase'], 'actor_id': raw['player_id']} def accept(a): if isinstance(a, int) and not isinstance(a, bool) and 0 <= a < len(legal): a = legal[a] if isinstance(a, dict) and 'action_type' in a: # a Kaggle agent's own action if a['action_type'] != ACTION[what].__name__: return None k = a.get('kwargs') or {} t = k.get('target_id') a = {'type': what, 'message': k.get('message'), 'reasoning': k.get('reasoning'), 'target': None if t in (None, '-1') else seat_of.get(t, '?')} if not isinstance(a, dict) or a.get('type') != what: return None why = a.get('reasoning') kw = {**base, 'reasoning': str(why)[:SAY] if why else None} if what == 'chat': text = a.get('message') return ChatAction(**kw, message=text.strip()[:SAY] or PASS).serialize() if isinstance(text, str) else None t = a.get('target') t = None if t is None else str(t) if (t is None and what != 'vote') or (t is not None and t not in aim): return None return ACTION[what](**kw, target_id='-1' if t is None else ids[int(t)]).serialize() limit = room.limit * 3 if what == 'chat' else None # a speech takes longer to write seconds = limit or room.limit if what in ('chat', 'vote') and room.kind(s) in ('human', 'agent'): room.event({'kind': 'turn', 'seat': i, 'what': what, 'seconds': seconds}) view = {'seat': s, 'player': ids[i], 'role': role[i], 'team': team[i], 'wolves': [seats[w] for w in wolves] if role[i] == 'Werewolf' else [], 'day': raw['day'], 'phase': raw['game_state_phase'].lower(), 'request': what, 'alive': [seat_of[p] for p in raw['alive_players']], 'known': known[s], 'events': public[-40:], 'seconds': seconds, 'raw': raw} return room.ask(s, view, legal, auto=stand_in, bot={'obs': obs, 'config': config}, accept=accept, limit=limit) room.event({'kind': 'start', 'seats': {s: {'name': names[i], 'kind': room.kind(s)} for i, s in enumerate(seats)}, 'roles': {r: role.count(r) for r in ('Werewolf', 'Seer', 'Doctor', 'Villager')}, 'seconds': room.limit}) for i in range(N): tell(i, {'kind': 'role', 'seat': i, 'role': role[i], 'team': team[i], 'wolves': wolves if role[i] == 'Werewolf' else []}) publish() stopped = False try: while not env.done: acts = [None] * N for i in sorted((i for i in range(N) if env.state[i].status == 'ACTIVE'), key=lambda i: WAKE.get(role[i], 2)): acts[i] = move(i) env.step(acts) publish() except Stop: stopped = True e = {'kind': 'over', **story.over(None, 'stopped')} public.append(e) room.event(e) end = env.info.get('GAME_END') or {} winner = None if stopped else end.get('winner_team') result = 'stopped' if winner is None else f'{winner.lower()} win, {N - len(story.gone)} left alive' replay = transcript(env.toJSON(), names, {i: room.kind(s) for i, s in enumerate(seats)}) if stopped: replay['events'].append(public[-1]) league.Replays().add(f'room-{seed}.json', replay, 'Room: ' + ', '.join(f'{i + 1} {n}' for i, n in enumerate(names)), names, result) room.end(result, {s: float(winner is not None and team[i] == winner) for i, s in enumerate(seats)}) room.close() if __name__ == '__main__': main()