from __future__ import annotations import argparse import csv import hashlib import json import math import platform from collections import Counter, defaultdict, deque from dataclasses import asdict, dataclass from datetime import datetime, timezone from pathlib import Path from statistics import median import numpy as np from PIL import Image, ImageDraw, ImageFont from vizdoom import ( AutomapMode, Button, DoomGame, GameVariable, ScreenFormat, ScreenResolution, __version__ as vizdoom_version, scenarios_path, ) TICS_PER_SECOND = 35 ROOT = Path(__file__).resolve().parent OUTPUTS = ROOT / "outputs" @dataclass(frozen=True) class MenuAction: name: str button: Button duration_tics: int @dataclass class TrialResult: scenario: str seed: int controller: str menu_order: list[str] scan_dwell_tics: int scan_dwell_ms: float success: bool presses: int scan_highlights: int scan_wait_tics: int action_tics: int engine_tics: int elapsed_engine_seconds: float total_reward: float final_health: float final_armor: float final_killcount: float policy_decisions: int selected_actions: dict[str, int] path_sector_count: int | None path_waypoint_count: int | None final_distance_to_goal: float | None end_reason: str def normalize_degrees(value: float) -> float: return (value + 180.0) % 360.0 - 180.0 def point_in_sector(x: float, y: float, sector) -> bool: crossings = 0 for line in sector.lines: y1, y2 = float(line.y1), float(line.y2) if (y1 > y) == (y2 > y): continue x_at_y = float(line.x1) + (y - y1) * (float(line.x2) - float(line.x1)) / (y2 - y1) if x_at_y > x: crossings += 1 return crossings % 2 == 1 def normalized_segment(line) -> tuple[tuple[float, float], tuple[float, float]]: first = (round(float(line.x1), 4), round(float(line.y1), 4)) second = (round(float(line.x2), 4), round(float(line.y2), 4)) return tuple(sorted((first, second))) # type: ignore[return-value] def locate_sector(sectors, point: tuple[float, float]) -> int: matches = [index for index, sector in enumerate(sectors) if point_in_sector(*point, sector)] if not matches: raise RuntimeError(f"Point {point} was not inside any exposed sector") return matches[0] def plan_sector_path(state, start: tuple[float, float], goal: tuple[float, float]): start_sector = locate_sector(state.sectors, start) goal_sector = locate_sector(state.sectors, goal) if start_sector == goal_sector: return [start_sector], [goal] portals: dict[tuple[tuple[float, float], tuple[float, float]], list[int]] = defaultdict(list) for sector_index, sector in enumerate(state.sectors): for line in sector.lines: if not line.is_blocking: portals[normalized_segment(line)].append(sector_index) graph: dict[int, list[tuple[int, tuple[float, float]]]] = defaultdict(list) for segment, sector_indices in portals.items(): unique = sorted(set(sector_indices)) if len(unique) != 2: continue midpoint = ( (segment[0][0] + segment[1][0]) / 2.0, (segment[0][1] + segment[1][1]) / 2.0, ) left, right = unique graph[left].append((right, midpoint)) graph[right].append((left, midpoint)) queue = deque([start_sector]) parent: dict[int, tuple[int, tuple[float, float]] | None] = {start_sector: None} while queue: current = queue.popleft() if current == goal_sector: break for neighbor, portal in graph[current]: if neighbor not in parent: parent[neighbor] = (current, portal) queue.append(neighbor) if goal_sector not in parent: raise RuntimeError(f"No portal path from sector {start_sector} to {goal_sector}") sector_path = [] portal_path = [] cursor = goal_sector while True: sector_path.append(cursor) link = parent[cursor] if link is None: break cursor, portal = link portal_path.append(portal) sector_path.reverse() portal_path.reverse() return sector_path, [*portal_path, goal] class MazePolicy: def __init__(self, state, start: tuple[float, float], goal: tuple[float, float]): self.goal = goal self.sector_path, self.waypoints = plan_sector_path(state, start, goal) self.waypoint_index = 0 self.decisions = 0 def desired_action(self, game: DoomGame, _state) -> str: self.decisions += 1 x = game.get_game_variable(GameVariable.POSITION_X) y = game.get_game_variable(GameVariable.POSITION_Y) while self.waypoint_index < len(self.waypoints) - 1: wx, wy = self.waypoints[self.waypoint_index] if math.hypot(wx - x, wy - y) > 28.0: break self.waypoint_index += 1 wx, wy = self.waypoints[self.waypoint_index] desired_angle = math.degrees(math.atan2(wy - y, wx - x)) % 360.0 angle = game.get_game_variable(GameVariable.ANGLE) error = normalize_degrees(desired_angle - angle) if error > 8.0: return "turn_left" if error < -8.0: return "turn_right" return "forward" def distance_to_goal(self, game: DoomGame) -> float: x = game.get_game_variable(GameVariable.POSITION_X) y = game.get_game_variable(GameVariable.POSITION_Y) return math.hypot(self.goal[0] - x, self.goal[1] - y) class CombatPolicy: def __init__(self): self.decisions = 0 def desired_action(self, _game: DoomGame, state) -> str: self.decisions += 1 targets = [label for label in state.labels if label.object_category == "Monster"] if not targets: return "attack" target = max(targets, key=lambda label: label.width * label.height) center_x = target.x + target.width / 2.0 if center_x < 148: return "strafe_left" if center_x > 172: return "strafe_right" return "attack" def make_game(scenario: str, seed: int) -> DoomGame: game = DoomGame() game.load_config(str(Path(scenarios_path) / f"{scenario}.cfg")) game.set_window_visible(False) game.set_screen_resolution(ScreenResolution.RES_320X240) game.set_screen_format(ScreenFormat.RGB24) game.set_labels_buffer_enabled(True) game.set_depth_buffer_enabled(True) game.set_automap_buffer_enabled(True) game.set_automap_mode(AutomapMode.OBJECTS_WITH_SIZE) game.set_sectors_info_enabled(True) game.set_objects_info_enabled(True) for variable in ( GameVariable.POSITION_X, GameVariable.POSITION_Y, GameVariable.ANGLE, GameVariable.HEALTH, GameVariable.ARMOR, GameVariable.KILLCOUNT, ): game.add_available_game_variable(variable) game.set_seed(seed) game.init() game.new_episode() return game def action_vector(game: DoomGame, selected: Button | None) -> list[int]: return [int(selected == button) for button in game.get_available_buttons()] def save_frame(buffer: np.ndarray, path: Path) -> None: Image.fromarray(buffer).save(path) def run_trial( scenario: str, seed: int, controller: str, menu: list[MenuAction], scan_dwell_tics: int, evidence: bool = False, ) -> TrialResult: game = make_game(scenario, seed) first_state = game.get_state() assert first_state is not None policy: MazePolicy | CombatPolicy if scenario == "my_way_home": player = next(item for item in first_state.objects if item.name == "DoomPlayer") goal = next(item for item in first_state.objects if item.name == "GreenArmor") policy = MazePolicy( first_state, (float(player.position_x), float(player.position_y)), (float(goal.position_x), float(goal.position_y)), ) else: policy = CombatPolicy() trial_id = f"{scenario}-seed-{seed}-{controller}-{scan_dwell_tics}" evidence_dir = OUTPUTS / "evidence" / trial_id if evidence: evidence_dir.mkdir(parents=True, exist_ok=True) save_frame(first_state.screen_buffer, evidence_dir / "000-start.png") presses = 0 scan_highlights = 0 scan_wait_tics = 0 action_tics = 0 selected_actions: Counter[str] = Counter() last_buffer = first_state.screen_buffer.copy() selected_log = [] while not game.is_episode_finished(): selected: MenuAction | None = None highlight_index = 0 while selected is None and not game.is_episode_finished(): state = game.get_state() if state is None: break last_buffer = state.screen_buffer.copy() highlighted = menu[highlight_index] desired = policy.desired_action(game, state) scan_highlights += 1 if highlighted.name == desired: selected = highlighted break game.make_action(action_vector(game, None), scan_dwell_tics) scan_wait_tics += scan_dwell_tics highlight_index = (highlight_index + 1) % len(menu) if selected is None or game.is_episode_finished(): break game.make_action(action_vector(game, selected.button), selected.duration_tics) action_tics += selected.duration_tics presses += 1 selected_actions[selected.name] += 1 selected_log.append( { "press": presses, "controller_tic": scan_wait_tics + action_tics, "action": selected.name, "action_tics": selected.duration_tics, "x": game.get_game_variable(GameVariable.POSITION_X), "y": game.get_game_variable(GameVariable.POSITION_Y), "angle": game.get_game_variable(GameVariable.ANGLE), } ) if evidence and presses in {1, 10, 25, 50, 75, 100} and not game.is_episode_finished(): state = game.get_state() if state is not None: save_frame(state.screen_buffer, evidence_dir / f"{presses:03d}-press.png") scenario_timeout = 300 if scenario == "basic" else 2100 engine_tics = min(scan_wait_tics + action_tics, scenario_timeout) total_reward = game.get_total_reward() health = game.get_game_variable(GameVariable.HEALTH) armor = game.get_game_variable(GameVariable.ARMOR) killcount = game.get_game_variable(GameVariable.KILLCOUNT) success = armor > 0 if scenario == "my_way_home" else killcount > 0 end_reason = "goal" if success else ("death" if health <= 0 else "timeout") final_distance = policy.distance_to_goal(game) if isinstance(policy, MazePolicy) else None if evidence: save_frame(last_buffer, evidence_dir / "999-final-visible-state.png") (evidence_dir / "switch-events.json").write_text( json.dumps(selected_log, indent=2), encoding="utf-8" ) result = TrialResult( scenario=scenario, seed=seed, controller=controller, menu_order=[item.name for item in menu], scan_dwell_tics=scan_dwell_tics, scan_dwell_ms=round(scan_dwell_tics / TICS_PER_SECOND * 1000.0, 3), success=success, presses=presses, scan_highlights=scan_highlights, scan_wait_tics=scan_wait_tics, action_tics=action_tics, engine_tics=engine_tics, elapsed_engine_seconds=round(engine_tics / TICS_PER_SECOND, 3), total_reward=round(total_reward, 4), final_health=health, final_armor=armor, final_killcount=killcount, policy_decisions=policy.decisions, selected_actions=dict(sorted(selected_actions.items())), path_sector_count=len(policy.sector_path) if isinstance(policy, MazePolicy) else None, path_waypoint_count=len(policy.waypoints) if isinstance(policy, MazePolicy) else None, final_distance_to_goal=round(final_distance, 3) if final_distance is not None else None, end_reason=end_reason, ) game.close() return result def summarize(results: list[TrialResult]) -> list[dict]: groups: dict[tuple, list[TrialResult]] = defaultdict(list) for result in results: key = ( result.scenario, result.controller, tuple(result.menu_order), result.scan_dwell_tics, ) groups[key].append(result) summaries = [] for (scenario, controller, menu_order, dwell), trials in sorted(groups.items()): successes = [trial for trial in trials if trial.success] summaries.append( { "scenario": scenario, "controller": controller, "menu_order": list(menu_order), "scan_dwell_tics": dwell, "scan_dwell_ms": trials[0].scan_dwell_ms, "trials": len(trials), "successes": len(successes), "success_rate": round(len(successes) / len(trials), 4), "median_presses_on_success": median([trial.presses for trial in successes]) if successes else None, "median_engine_seconds_on_success": median( [trial.elapsed_engine_seconds for trial in successes] ) if successes else None, "median_scan_wait_seconds_on_success": round( median([trial.scan_wait_tics for trial in successes]) / TICS_PER_SECOND, 3, ) if successes else None, } ) return summaries def build_evidence_sheet() -> None: panels = [ ( OUTPUTS / "evidence" / "my_way_home-seed-42-linear-auto-scan-18" / "000-start.png", "A. SEED 42 START, 514 MS SCAN", ), ( OUTPUTS / "evidence" / "my_way_home-seed-42-linear-auto-scan-18" / "025-press.png", "B. AFTER SWITCH PRESS 25", ), ( OUTPUTS / "evidence" / "my_way_home-seed-42-linear-auto-scan-18" / "999-final-visible-state.png", "C. LAST FRAME BEFORE GOAL", ), ( OUTPUTS / "evidence" / "my_way_home-seed-7-linear-auto-scan-18" / "999-final-visible-state.png", "D. SEED 7 AT THE TIMEOUT", ), ] missing = [str(path) for path, _ in panels if not path.exists()] if missing: raise RuntimeError(f"Missing evidence frames: {missing}") font = ImageFont.load_default(size=18) canvas = Image.new("RGB", (660, 540), "white") draw = ImageDraw.Draw(canvas) for index, (path, label) in enumerate(panels): column = index % 2 row = index // 2 x = column * 330 y = row * 270 frame = Image.open(path).convert("RGB") canvas.paste(frame, (x + 5, y + 5)) draw.rectangle((x + 5, y + 5, x + 324, y + 244), outline="black", width=2) draw.text((x + 8, y + 248), label, fill="black", font=font) canvas.save(OUTPUTS / "one-switch-maze-evidence.png") def write_outputs(results: list[TrialResult], stem: str = "results") -> None: OUTPUTS.mkdir(parents=True, exist_ok=True) trial_rows = [asdict(result) for result in results] summary_rows = summarize(results) payload = { "experiment": "one-switch-vizdoom-v0.1", "generated_at": datetime.now(timezone.utc).isoformat(), "environment": { "python": platform.python_version(), "platform": platform.platform(), "vizdoom": vizdoom_version, "tic_rate_hz": TICS_PER_SECOND, }, "claim_boundary": ( "An automated observer selected Doom actions through one binary switch channel. " "This tests controller sufficiency and engine execution, not human usability." ), "summaries": summary_rows, "trials": trial_rows, } output_path = OUTPUTS / f"{stem}.json" output_path.write_text(json.dumps(payload, indent=2), encoding="utf-8") payload_hash = hashlib.sha256(output_path.read_bytes()).hexdigest() (OUTPUTS / f"{stem}.sha256").write_text( f"{payload_hash} {stem}.json\n", encoding="utf-8" ) trial_filename = "trials.csv" if stem == "results" else f"{stem}-trials.csv" with (OUTPUTS / trial_filename).open("w", newline="", encoding="utf-8") as handle: fieldnames = [ "scenario", "seed", "controller", "scan_dwell_tics", "scan_dwell_ms", "success", "presses", "scan_highlights", "scan_wait_tics", "action_tics", "engine_tics", "elapsed_engine_seconds", "total_reward", "final_health", "final_armor", "final_killcount", "end_reason", ] writer = csv.DictWriter(handle, fieldnames=fieldnames) writer.writeheader() for row in trial_rows: writer.writerow({field: row[field] for field in fieldnames}) print(json.dumps(summary_rows, indent=2)) print(f"Wrote {output_path}") print(f"SHA-256 {payload_hash}") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run the one-switch ViZDoom experiment") parser.add_argument("--quick", action="store_true", help="Run seed 7 and the fastest dwell only") return parser.parse_args() def main() -> None: args = parse_args() seeds = [7] if args.quick else [7, 19, 42, 73, 101, 1337] dwell_values = [4] if args.quick else [4, 9, 18] menus = { "combat-forward": [ MenuAction("attack", Button.ATTACK, 4), MenuAction("strafe_left", Button.MOVE_LEFT, 3), MenuAction("strafe_right", Button.MOVE_RIGHT, 3), ], "maze-forward": [ MenuAction("forward", Button.MOVE_FORWARD, 8), MenuAction("turn_left", Button.TURN_LEFT, 2), MenuAction("turn_right", Button.TURN_RIGHT, 2), ], } results = [] for dwell in dwell_values: for seed in seeds: results.append( run_trial( "basic", seed, "linear-auto-scan", menus["combat-forward"], dwell, evidence=(seed == 7 and dwell in {dwell_values[0], dwell_values[-1]}), ) ) results.append( run_trial( "my_way_home", seed, "linear-auto-scan", menus["maze-forward"], dwell, evidence=(seed == 7 and dwell in {dwell_values[0], dwell_values[-1]}) or (not args.quick and seed == 42 and dwell == dwell_values[-1]), ) ) write_outputs(results, "results-quick" if args.quick else "results") if not args.quick: build_evidence_sheet() if __name__ == "__main__": main()