#!/usr/bin/env python3 """ ONI Agent Event Daemon ====================== Continuous event poller that feeds game events to the AI's input stream. Architecture: Game Mod --> Event Queue (via HTTP) --> Event Daemon --> AI Input Stream The daemon: 1. Polls GET /api/state/events?since= every N seconds 2. Classifies events by severity (critical/warning/info) 3. For critical events: immediately triggers full analysis + prints alert 4. For warning events: logs and optionally triggers targeted checks 5. For info events: accumulates and reports periodically 6. Maintains a compact event log for AI context """ import json import os import sys import time import datetime # Add tools to path TOOLS_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'tools') sys.path.insert(0, TOOLS_DIR) # Disable auto-pause for event daemon — we poll frequently and should not pause the game import oni_api oni_api.AUTO_PAUSE_ENABLED = False from oni_api import api_get, api_post, api_url # ── Configuration ────────────────────────────────────────────────────────── POLL_INTERVAL = 5 # seconds between event polls CRITICAL_POLL_INTERVAL = 2 # poll faster when critical events detected MAX_EVENT_HISTORY = 200 # events kept in rolling buffer CRITICAL_SEVERITIES = {'critical', 'duplicantdeath', 'buildingdamage', 'poweroutage'} WARNING_SEVERITIES = {'warning', 'duplicantstress', 'lowoxygen', 'foodshortage'} # ── Event History ────────────────────────────────────────────────────────── class EventHistory: """Rolling buffer of events + statistics for AI context.""" def __init__(self, maxlen=MAX_EVENT_HISTORY): self.events = [] self.maxlen = maxlen self.stats = { 'total': 0, 'critical': 0, 'warning': 0, 'info': 0, 'by_category': {}, 'by_type': {}, 'last_poll_cycle': 0, } def push(self, events): for e in events: self.events.append(e) self.stats['total'] += 1 sev = (e.get('severity') or 'info').lower() cat = e.get('category', 'unknown') etype = e.get('type', 'unknown') if sev in ('critical', 'duplicantdeath', 'buildingdamage'): self.stats['critical'] += 1 elif sev in ('warning',): self.stats['warning'] += 1 else: self.stats['info'] += 1 self.stats['by_category'][cat] = self.stats['by_category'].get(cat, 0) + 1 self.stats['by_type'][etype] = self.stats['by_type'].get(etype, 0) + 1 self.stats['last_poll_cycle'] = e.get('cycle', 0) # Trim if len(self.events) > self.maxlen: self.events = self.events[-self.maxlen:] def get_summary(self): return { 'total_events': self.stats['total'], 'critical_count': self.stats['critical'], 'warning_count': self.stats['warning'], 'info_count': self.stats['info'], 'categories': dict(sorted(self.stats['by_category'].items(), key=lambda x: -x[1])[:10]), 'last_cycle': self.stats['last_poll_cycle'], 'recent_critical': [e for e in self.events[-20:] if (e.get('severity') or '').lower() in CRITICAL_SEVERITIES][-5:], } # ── Event Classifier ────────────────────────────────────────────────────── def classify_event(e): """Return the action type for a given event.""" sev = (e.get('severity') or '').lower() title = (e.get('title') or '').lower() msg = (e.get('message') or '').lower() cat = (e.get('category') or '').lower() if sev in CRITICAL_SEVERITIES: return 'critical' if sev in WARNING_SEVERITIES: return 'warning' if cat == 'action': return 'action_feedback' # Content-based classification combined = title + ' ' + msg if any(w in combined for w in ['suffocat', 'choking', 'no oxygen', 'out of air']): return 'critical' if any(w in combined for w in ['starving', 'food', 'hungry']): return 'warning' if any(w in combined for w in ['heat', 'overheat', 'temperature', 'melt']): return 'warning' if any(w in combined for w in ['power', 'wattage', 'shutoff']): return 'warning' if any(w in combined for w in ['duplicant', 'stress', 'break']): return 'warning' if any(w in combined for w in ['research complete', 'research completed']): return 'info_research' if any(w in combined for w in ['printing pod']): return 'info_printing_pod' return 'info' def format_event_for_ai(e): """Format an event as a structured string for AI input.""" ts = datetime.datetime.fromtimestamp(e.get('timestamp', time.time())).strftime('%H:%M:%S') cycle = e.get('cycle', '?') severity = e.get('severity', 'info').upper() title = e.get('title', '?') message = e.get('message', '') lines = [f"[EVENT {severity}] Cycle {cycle} @ {ts}"] lines.append(f" Title: {title}") if message: lines.append(f" Message: {message}") entity = e.get('entity') if entity: lines.append(f" Entity: {entity}") cell = e.get('cell') if isinstance(cell, int) and cell >= 0: lines.append(f" Cell: {cell}") return '\n'.join(lines) # ── Polling Loop ────────────────────────────────────────────────────────── def poll_loop(event_history): seq = 0 consecutive_errors = 0 print("[ONI Event Daemon] Starting event poll...") print(f"[ONI Event Daemon] Poll interval: {POLL_INTERVAL}s") print() while True: try: data = api_get(f"/api/state/events?since={seq}&limit=50") if 'error' in data: consecutive_errors += 1 if consecutive_errors == 1: print(f"[!] Cannot reach game: {data['error']}") print(" Waiting for game connection...") time.sleep(POLL_INTERVAL * 2) continue consecutive_errors = 0 events = data.get('events', []) next_seq = data.get('next_seq', seq) if events: event_history.push(events) # Classify and report critical_events = [] for e in events: cls = classify_event(e) if cls == 'critical': critical_events.append(e) # Print alert with clear marker print("=" * 56) print(" *** CRITICAL EVENT ***") print(format_event_for_ai(e)) print("=" * 56) print() # Auto-trigger full analysis on critical events _trigger_emergency_analysis(e) elif cls == 'warning': print(format_event_for_ai(e)) print() elif cls == 'info_research': # Research completed — show with unlocks print("=" * 40) print(format_event_for_ai(e)) print(" -> Check what's new: python3 tools/oni_api.py buildable") print("=" * 40) print() elif cls == 'info_printing_pod': # Printing pod ready print("=" * 40) print(format_event_for_ai(e)) print(" -> View options: python3 tools/oni_api.py printing_pod") print("=" * 40) print() else: # Only print non-info events or batch feedback cat = e.get('category', '') if cat != 'general' or cls != 'info': print(format_event_for_ai(e)) print() # If critical events happened, poll faster for a bit if critical_events: seq = next_seq time.sleep(CRITICAL_POLL_INTERVAL) continue seq = next_seq time.sleep(POLL_INTERVAL) except KeyboardInterrupt: print("\n[ONI Event Daemon] Shutting down.") summary = event_history.get_summary() print(f" Total events seen: {summary['total_events']}") print(f" Critical: {summary['critical_count']}, Warning: {summary['warning_count']}") break except Exception as e: consecutive_errors += 1 if consecutive_errors <= 2: print(f"[!] Poll error: {e}") time.sleep(POLL_INTERVAL) def _trigger_emergency_analysis(event): """On critical events, pull game state snapshot for AI context.""" try: print(" -> Triggering emergency snapshot...") game = api_get('/api/state/game') alerts = api_get('/api/state/alert') dups = api_get('/api/state/duplicants') if 'error' not in game: print(f" [SNAPSHOT] Cycle {game.get('cycle', '?')}, " f"{game.get('duplicantCount', '?')} dupes, " f"{game.get('suffocating', 0)} suffocating, " f"{game.get('starving', 0)} starving, " f"{game.get('stressed', 0)} stressed") if isinstance(alerts, list) and alerts: print(f" [ALERTS] {len(alerts)} active:") for a in alerts[:3]: print(f" - [{a.get('severity', '?')}] {a.get('title', '?')}") print() except: pass # ── Main ────────────────────────────────────────────────────────────────── def main(): history = EventHistory() try: poll_loop(history) except KeyboardInterrupt: pass # Print final summary summary = history.get_summary() print() print("=" * 56) print(" Event Daemon Session Summary") print("=" * 56) print(f" Total events: {summary['total_events']}") print(f" Critical: {summary['critical_count']}") print(f" Warning: {summary['warning_count']}") print(f" Info: {summary['info_count']}") print(f" Top categories: {', '.join(summary['categories'].keys())}") print("=" * 56) if __name__ == '__main__': main()