389 lines
20 KiB
Python
389 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""Arcade room controller and browser operator; Python standard library only."""
|
|
import argparse
|
|
import hmac
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import time
|
|
import uuid
|
|
from urllib.parse import urlparse
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
|
|
|
|
class Room:
|
|
def __init__(self, database, clock=time.time, cabinet_id="cabinet_01"):
|
|
self.clock = clock
|
|
self.cabinet_id = cabinet_id
|
|
self.db = sqlite3.connect(database, check_same_thread=False)
|
|
self.db.execute("CREATE TABLE IF NOT EXISTS room (id INTEGER PRIMARY KEY, value TEXT NOT NULL)")
|
|
self.db.execute("CREATE TABLE IF NOT EXISTS cabinet_rooms (id TEXT PRIMARY KEY, value TEXT NOT NULL)")
|
|
row = self.db.execute("SELECT value FROM cabinet_rooms WHERE id=?", (cabinet_id,)).fetchone()
|
|
if row is None and cabinet_id == "cabinet_01":
|
|
row = self.db.execute("SELECT value FROM room WHERE id=1").fetchone()
|
|
self.state = json.loads(row[0]) if row else dict(
|
|
session="", phase="idle", reward="not requested", commands=[], seen=[], history=[])
|
|
# Never carry live connectivity across a controller restart.
|
|
self.last_seen = 0
|
|
self.boot = self.state.get("boot", "")
|
|
self.version = "unknown"
|
|
self.gpio = "unknown"
|
|
self.state.setdefault("game", "pacman")
|
|
self.state.setdefault("mappings", {"game_won": ["coin_release"]})
|
|
self.save()
|
|
|
|
def save(self):
|
|
with self.db:
|
|
self.db.execute("INSERT OR REPLACE INTO cabinet_rooms VALUES (?, ?)", (self.cabinet_id, json.dumps(self.state)))
|
|
|
|
def log(self, message):
|
|
self.state["history"].append(dict(time=self.clock(), message=message))
|
|
self.state["history"] = self.state["history"][-100:]
|
|
|
|
def online(self):
|
|
return self.last_seen > 0 and self.clock() - self.last_seen < 4
|
|
|
|
def snapshot(self):
|
|
self.expire()
|
|
return dict(self.state, online=self.online(), cabinet=self.cabinet_id,
|
|
version=self.version, gpio=self.gpio,
|
|
last_seen=self.last_seen, protocol=2)
|
|
|
|
def expire(self):
|
|
for command in self.state["commands"]:
|
|
if command["status"] == "queued" and self.clock() > command["expires"]:
|
|
command["status"] = "unknown"
|
|
if command["type"] in ("reward", "output"):
|
|
self.state["reward" if command["type"] == "reward" else "last_output"] = "unknown — inspect hardware; not automatically retried"
|
|
else:
|
|
self.state["phase"] = "needs_prepare"
|
|
self.log(command["type"] + " acknowledgement timed out")
|
|
self.save()
|
|
|
|
def command(self, kind):
|
|
command = dict(id=str(uuid.uuid4()), session=self.state["session"], type=kind,
|
|
status="queued", expires=self.clock() + 15)
|
|
if kind == "prepare":
|
|
command.update(game=self.state["game"], variant=self.state.get("variant", "standard"))
|
|
self.state["commands"].append(command)
|
|
self.log("Queued " + kind)
|
|
return command
|
|
|
|
def operator(self, action):
|
|
self.expire()
|
|
if not self.online(): raise ValueError("Cabinet is offline; no command sent")
|
|
if any(c["status"] == "queued" and c["type"] not in ("output", "reward") for c in self.state["commands"]):
|
|
raise ValueError("Wait for the pending command or its timeout")
|
|
if action in ("prepare", "reset"):
|
|
if action == "prepare" and self.state["phase"] not in ("idle", "won", "lost", "needs_prepare"):
|
|
raise ValueError("Reset the active session before preparing again")
|
|
self.state.update(session=str(uuid.uuid4()), phase="preparing" if action == "prepare" else "resetting",
|
|
reward="not requested", commands=[], seen=[])
|
|
self.command(action)
|
|
elif action == "start":
|
|
if self.state["phase"] != "ready": raise ValueError("Prepare the cabinet first")
|
|
self.state["phase"] = "starting"
|
|
self.command("start")
|
|
else:
|
|
raise ValueError("Unknown operator action")
|
|
self.save()
|
|
return self.snapshot()
|
|
|
|
def poll(self, data):
|
|
if data.get("cabinet") != self.cabinet_id or data.get("protocol") not in (1, 2):
|
|
raise ValueError("Wrong cabinet or protocol version")
|
|
boot = data.get("boot")
|
|
if not isinstance(boot, str) or not 1 <= len(boot) <= 100:
|
|
raise ValueError("Missing cabinet boot identity")
|
|
if self.boot and boot != self.boot:
|
|
self.state.update(session="", phase="needs_prepare", commands=[], seen=[], reward="not requested")
|
|
self.log("Cabinet restarted; operator must prepare a new session")
|
|
self.boot = boot
|
|
self.state["boot"] = boot
|
|
self.last_seen = self.clock()
|
|
self.version = str(data.get("version", "unknown"))[:80]
|
|
self.gpio = str(data.get("gpio", "unknown"))[:100]
|
|
self.expire()
|
|
reports = data.get("reports", [])
|
|
if not isinstance(reports, list) or len(reports) > 50: raise ValueError("Invalid reports")
|
|
for report in reports:
|
|
if not isinstance(report, dict): raise ValueError("Invalid report")
|
|
if report.get("session") != self.state["session"]: continue
|
|
if report.get("kind") == "ack":
|
|
self.ack(report)
|
|
elif report.get("kind") == "event":
|
|
self.event(report)
|
|
self.save()
|
|
return dict(ok=True, session=self.state["session"],
|
|
commands=[c for c in self.state["commands"] if c["status"] == "queued"])
|
|
|
|
def ack(self, report):
|
|
command = next((c for c in self.state["commands"] if c["id"] == report.get("id")), None)
|
|
if command is None or command["status"] != "queued": return
|
|
success = report.get("ok") is True
|
|
command["status"] = "accepted" if success else "failed"
|
|
self.log("%s: %s" % (command["type"], str(report.get("detail", command["status"]))[:200]))
|
|
if command["type"] == "output":
|
|
self.state["last_output"] = "accepted (physical state unconfirmed)" if success else "failed or uncertain"
|
|
elif command["type"] == "reward":
|
|
statuses = [c["status"] for c in self.state["commands"] if c["type"] == "reward"]
|
|
if any(s in ("failed", "unknown") for s in statuses):
|
|
self.state["reward"] = "failed or uncertain — inspect cabinet"
|
|
elif all(s == "accepted" for s in statuses):
|
|
self.state["reward"] = "GPIO accepted (physical delivery unconfirmed)"
|
|
else:
|
|
self.state["reward"] = "requested"
|
|
elif success:
|
|
self.state["phase"] = {"prepare": "ready", "reset": "waiting" if self.state.get("gate") else "idle", "start": "playing"}[command["type"]]
|
|
if command["type"] == "prepare" and self.state.pop("auto_start", False):
|
|
self.state["phase"] = "starting"
|
|
self.command("start")
|
|
if command["type"] == "prepare":
|
|
for action in self.state["mappings"].get("game_activated", []):
|
|
self.command("output")["action_id"] = action
|
|
else:
|
|
self.state["phase"] = "needs_prepare"
|
|
|
|
def event(self, report):
|
|
event_id = report.get("id")
|
|
if not isinstance(event_id, str) or not 1 <= len(event_id) <= 100: return
|
|
if event_id in self.state["seen"]: return
|
|
self.state["seen"].append(event_id)
|
|
self.state["seen"] = self.state["seen"][-500:]
|
|
name = report.get("event")
|
|
if name == "started" and self.state["phase"] in ("ready", "starting"):
|
|
self.state["phase"] = "playing"
|
|
elif name == "won" and self.state["phase"] == "playing":
|
|
self.state["phase"] = "won"
|
|
if self.state["reward"] == "not requested" and self.state["mappings"].get("game_won"):
|
|
self.state["reward"] = "requested"
|
|
elif name == "lost" and self.state["phase"] == "playing":
|
|
self.state["phase"] = "lost"
|
|
else:
|
|
return
|
|
self.log(self.state["game"] + " " + name)
|
|
for action in self.state["mappings"].get("game_" + name, []):
|
|
command = self.command("reward" if name == "won" else "output")
|
|
command["action_id"] = action
|
|
|
|
|
|
class Fleet:
|
|
"""Editor catalog plus independently acknowledged cabinet sessions."""
|
|
def __init__(self, database, catalog, clock=time.time):
|
|
self.catalog = catalog
|
|
self.rooms = {key: Room(database, clock, key) for key in catalog["cabinets"]}
|
|
|
|
def compatible(self, cabinet, game):
|
|
definition = self.catalog["games"][game]
|
|
capabilities = set(self.catalog["cabinets"][cabinet]["capabilities"])
|
|
alternatives = definition.get("alternatives", [])
|
|
return set(definition["requirements"]) <= capabilities and (not alternatives or any(set(a) <= capabilities for a in alternatives))
|
|
|
|
def snapshot(self):
|
|
cabinets = {}
|
|
for key, room in self.rooms.items():
|
|
cabinets[key] = dict(room.snapshot(), definition=self.catalog["cabinets"][key],
|
|
compatible_games=[g for g in self.catalog["games"] if self.compatible(key, g)])
|
|
# Retain the single-cabinet desktop UI response while the browser offers the fleet view.
|
|
first = cabinets.get("cabinet_01", next(iter(cabinets.values())))
|
|
return dict(first, cabinets=cabinets, catalog=self.catalog)
|
|
|
|
def validate_assignment(self, item):
|
|
cabinet, game = item.get("cabinet"), item.get("game")
|
|
if cabinet not in self.rooms or game not in self.catalog["games"]:
|
|
raise ValueError("Unknown cabinet or game")
|
|
if not self.compatible(cabinet, game): raise ValueError(f"{game} cannot run on {cabinet}: hardware requirements not met")
|
|
if item.get("variant", "standard") != "standard": raise ValueError("Only standard variants are supported by this runtime")
|
|
allowed = set(self.catalog["cabinets"][cabinet]["actions"])
|
|
for event, actions in item.get("mappings", {}).items():
|
|
if event not in ("game_activated", "game_started", "game_won", "game_lost"):
|
|
raise ValueError(f"Scenario event {event} is not implemented in managed runtime")
|
|
if not set(actions) <= allowed: raise ValueError("Scenario refers to an unknown hardware action")
|
|
room = self.rooms[cabinet]
|
|
room.expire()
|
|
if not room.online(): raise ValueError(f"{cabinet} is offline")
|
|
if room.state["phase"] not in ("idle", "won", "lost", "needs_prepare"):
|
|
raise ValueError(f"Reset {cabinet} before preparing a new assignment")
|
|
if any(c["status"] == "queued" for c in room.state["commands"]):
|
|
raise ValueError(f"{cabinet} has a pending command")
|
|
|
|
def operator(self, action, data=None):
|
|
data = data or {}
|
|
if action == "prepare_scenario":
|
|
scenario_id = data.get("scenario")
|
|
scenario = self.catalog["scenarios"].get(scenario_id)
|
|
if not scenario: raise ValueError("Unknown scenario")
|
|
assignments = scenario["assignments"]
|
|
if not assignments or len({a["cabinet"] for a in assignments}) != len(assignments):
|
|
raise ValueError("Scenario must have unique, nonempty cabinet assignments")
|
|
# Validate the whole plan before sending anything to any cabinet.
|
|
for item in assignments: self.validate_assignment(item)
|
|
by_id = {a["cabinet"]: a for a in assignments}
|
|
for item in assignments:
|
|
visited = {item["cabinet"]}
|
|
source = item.get("activate_after_win", "")
|
|
while source:
|
|
if source not in by_id: raise ValueError("Dependency cabinet must be assigned in the same scenario")
|
|
if source in visited: raise ValueError("Cabinet activation dependencies cannot form a cycle")
|
|
visited.add(source)
|
|
source = by_id[source].get("activate_after_win", "")
|
|
for item in assignments: self.prepare(item, scenario_id)
|
|
# Bind dependencies to this particular run, never an earlier win.
|
|
for item in assignments:
|
|
source = item.get("activate_after_win")
|
|
if source:
|
|
room = self.rooms[item["cabinet"]]
|
|
room.state["gate"] = dict(cabinet=source, session=self.rooms[source].state["session"])
|
|
room.log("Waiting for " + source + " to win")
|
|
room.save()
|
|
elif action == "prepare":
|
|
cabinet = data.get("cabinet", "cabinet_01")
|
|
item = dict(cabinet=cabinet, game=data.get("game", "pacman"), mappings={})
|
|
self.validate_assignment(item)
|
|
# Manual overrides do not inherit a reward from the previous scenario.
|
|
self.prepare(item, "manual override")
|
|
else:
|
|
cabinet = data.get("cabinet", "cabinet_01")
|
|
if cabinet not in self.rooms: raise ValueError("Unknown cabinet")
|
|
if action == "start" and self.rooms[cabinet].state.get("gate"):
|
|
raise ValueError("This cabinet starts automatically after its dependency wins")
|
|
self.rooms[cabinet].operator(action)
|
|
if action == "reset":
|
|
self.rooms[cabinet].state.pop("gate", None)
|
|
self.rooms[cabinet].state.pop("auto_start", None)
|
|
self.rooms[cabinet].save()
|
|
self.advance()
|
|
return self.snapshot()
|
|
|
|
def prepare(self, item, scenario):
|
|
room = self.rooms[item["cabinet"]]
|
|
room.state.update(game=item["game"], mappings=item.get("mappings", {}),
|
|
variant=item.get("variant", "standard"), scenario=scenario)
|
|
room.state.pop("gate", None)
|
|
room.state.pop("auto_start", None)
|
|
room.operator("reset" if item.get("activate_after_win") else "prepare")
|
|
room.log("Assignment from " + scenario)
|
|
room.save()
|
|
|
|
def poll(self, data):
|
|
cabinet = data.get("cabinet")
|
|
if data.get("protocol") != 2: raise ValueError("Update cabinet export: controller requires protocol 2")
|
|
if cabinet not in self.rooms: raise ValueError("Unknown cabinet")
|
|
self.rooms[cabinet].poll(data)
|
|
self.advance()
|
|
room = self.rooms[cabinet]
|
|
return dict(ok=True, session=room.state["session"],
|
|
commands=[c for c in room.state["commands"] if c["status"] == "queued"])
|
|
|
|
def advance(self):
|
|
# Invalidate descendants when their source is reset/reassigned/restarted.
|
|
# Iterate to propagate invalidation through longer chains.
|
|
for _ in self.rooms:
|
|
for room in self.rooms.values():
|
|
gate = room.state.get("gate")
|
|
if not gate: continue
|
|
source = self.rooms.get(gate["cabinet"])
|
|
if source is None or source.state["session"] != gate["session"]:
|
|
room.state.update(session=str(uuid.uuid4()), phase="needs_prepare", commands=[], seen=[], gate=None, auto_start=False)
|
|
room.log("Dependency session ended; prepare the scenario again")
|
|
room.save()
|
|
for room in self.rooms.values():
|
|
gate = room.state.get("gate")
|
|
if not gate or room.state["phase"] != "waiting" or not room.online(): continue
|
|
source = self.rooms[gate["cabinet"]]
|
|
if source.state["phase"] == "won":
|
|
room.state.update(phase="preparing", auto_start=True)
|
|
room.command("prepare")
|
|
room.log("Activated by " + gate["cabinet"] + " win")
|
|
room.save()
|
|
|
|
|
|
def handler_for(room, operator_token, cabinet_token):
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
if self.path not in ("/", "/index.html"):
|
|
self.respond(404, {"error": "Not found"})
|
|
return
|
|
body = (Path(__file__).parent / "operator.html").read_bytes()
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/html; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.send_header("X-Frame-Options", "DENY")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_POST(self):
|
|
try:
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
if not 0 <= length <= 65536: raise ValueError("Request too large")
|
|
data = json.loads(self.rfile.read(length) or b"{}")
|
|
if not isinstance(data, dict): raise ValueError("Expected object")
|
|
expected = operator_token
|
|
if self.path == "/cabinet/poll":
|
|
expected = cabinet_token.get(data.get("cabinet"), "") if isinstance(cabinet_token, dict) else cabinet_token
|
|
if not expected:
|
|
self.respond(401, {"error": "Cabinet credentials not provisioned"})
|
|
return
|
|
if not hmac.compare_digest(self.headers.get("Authorization", ""), "Bearer " + expected):
|
|
self.respond(401, {"error": "Unauthorized"})
|
|
return
|
|
origin = self.headers.get("Origin")
|
|
if origin and urlparse(origin).netloc != self.headers.get("Host"):
|
|
raise ValueError("Cross-origin requests are disabled")
|
|
if self.path == "/cabinet/poll": result = room.poll(data)
|
|
elif self.path == "/operator/state": result = room.snapshot()
|
|
elif self.path == "/operator/command":
|
|
result = room.operator(data.get("action"), data) if isinstance(room, Fleet) else room.operator(data.get("action"))
|
|
else: raise ValueError("Unknown endpoint")
|
|
self.respond(200, result)
|
|
except (ValueError, TypeError, KeyError) as exc:
|
|
self.respond(400, {"error": str(exc)})
|
|
|
|
def setup(self):
|
|
super().setup()
|
|
self.connection.settimeout(3)
|
|
|
|
def respond(self, code, data):
|
|
body = json.dumps(data).encode()
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, *_args): pass
|
|
return Handler
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--bind", default="127.0.0.1")
|
|
parser.add_argument("--port", type=int, default=8766)
|
|
parser.add_argument("--database", default="arcade_room.sqlite3")
|
|
parser.add_argument("--catalog", default=str(Path(__file__).with_name("catalog.json")))
|
|
args = parser.parse_args()
|
|
operator = os.environ.get("ARCADE_OPERATOR_TOKEN", "")
|
|
cabinet = os.environ.get("ARCADE_CABINET_TOKEN", "")
|
|
try:
|
|
tokens = json.loads(os.environ.get("ARCADE_CABINET_TOKENS", "{}"))
|
|
except ValueError:
|
|
parser.error("ARCADE_CABINET_TOKENS must be a JSON object")
|
|
if not isinstance(tokens, dict): parser.error("ARCADE_CABINET_TOKENS must be a JSON object")
|
|
if cabinet: tokens.setdefault("cabinet_01", cabinet)
|
|
if not isinstance(tokens, dict) or not tokens or len(operator) < 16 or any(not isinstance(t, str) or len(t) < 16 or t == operator for t in tokens.values()) or len(set(tokens.values())) != len(tokens):
|
|
parser.error("Set an operator token and unique cabinet tokens (16+ characters); use ARCADE_CABINET_TOKEN for cabinet_01 or ARCADE_CABINET_TOKENS JSON for multiple cabinets")
|
|
catalog = json.loads(Path(args.catalog).read_text(encoding="utf-8"))
|
|
room = Fleet(Path(args.database), catalog)
|
|
server = HTTPServer((args.bind, args.port), handler_for(room, operator, tokens))
|
|
print(f"Arcade controller ready at http://{args.bind}:{args.port}/ — {len(room.rooms)} cabinets", flush=True)
|
|
try: server.serve_forever()
|
|
except KeyboardInterrupt: pass
|
|
finally:
|
|
server.server_close()
|
|
for cabinet_room in room.rooms.values(): cabinet_room.db.close()
|
|
|
|
|
|
if __name__ == "__main__": main()
|