import asyncio import logging import json from backend.gaming.capture_engine import capture_engine from backend.gaming.overlay_renderer import push_overlay_vectors, trigger_failsafe from backend.voice.tts import TTSPipeline tts = TTSPipeline() from backend.ws.agent_ws import ws_manager logger = logging.getLogger(__name__) class CoachingEngine: def __init__(self): self.running = False self.level = 1 # 1: Analyst++, 2: Strategist++, 3: Grandmaster self.persona = "JARVIS" self.last_audio_played = 0 self.match_state = "IN_MATCH" # PRE_MATCH, IN_MATCH, POST_MATCH self.player_behavior_profile = [] self.complexity_scale = 1.0 # 1.0 = Complex, 0.5 = 1-step direct commands self.squad_sync_active = False self.player_intent_override = None async def start_loop(self): if self.running: return self.running = True capture_engine.start() logger.info(f"JARVIS 10X: Coaching Engine Started at Level {self.level}") while self.running: try: # 1. Failsafe Fullscreen Detection if capture_engine.is_exclusive_fullscreen(): trigger_failsafe(True) else: trigger_failsafe(False) # 2. Grab Frame Buffer & Spatial Audio frame_bytes = capture_engine.get_latest_frame_bytes() if frame_bytes: if self.match_state == "PRE_MATCH": await self._pre_match_diagnostics(frame_bytes) self.match_state = "IN_MATCH" elif self.match_state == "IN_MATCH": await self._analyze_state(frame_bytes) elif self.match_state == "POST_MATCH": await self._post_match_forensics(frame_bytes) self.match_state = "PRE_MATCH" except Exception as e: logger.error(f"JARVIS 10X Coach Engine Loop Error: {e}") # Dynamic interval: scales with complexity, defaults to 2 seconds await asyncio.sleep(2.0) async def _get_api_key(self): from backend.gaming.config import get_gaming_coach_model_config config = await get_gaming_coach_model_config() return config.get("api_key") async def _pre_match_diagnostics(self, frame_bytes): """Level 1: Pre-Match Diagnostics & Phase-Based Game Plan""" import google.generativeai as genai api_key = await self._get_api_key() if not api_key: return genai.configure(api_key=api_key) model = genai.GenerativeModel('gemini-3.5-flash') prompt = """You are 'The Analyst++', a Universal AI Gaming Coach. Analyze this pre-match screen. 1. Analyze selected game, map, active role, character/champion/agent selection, and historical matchup dynamics. 2. Build a phase-based game plan (early-game setup, mid-game pacing, late-game win-conditions). 3. Generate a meta-analytical cheat sheet detailing common enemy composition flaws and positional tendencies at the player's rank tier. Output JSON: {"speech": "Brief welcoming summary", "vectors": [], "game_plan": "...", "cheat_sheet": "..."}""" try: response = await asyncio.to_thread(model.generate_content, [prompt, {"mime_type": "image/jpeg", "data": frame_bytes}]) text = response.text if "```json" in text: text = text.split("```json")[1].split("```")[0] data = json.loads(text.strip()) speech = data.get("speech") if speech: logger.info(f"[{self.persona} PRE-MATCH]: {speech}") await tts.synthesize(speech, personality=self.persona, context="coaching") await ws_manager.broadcast({"event": "gaming:coach_audio", "payload": "AUDIO_BYTES_SENT_VIA_WS"}) vectors = data.get("vectors", []) if vectors: push_overlay_vectors(vectors) except Exception as e: logger.error(f"Pre-Match Diagnostics Error: {e}") async def _post_match_forensics(self, frame_bytes): """Level 1: Post-Match Forensic Review""" import google.generativeai as genai api_key = await self._get_api_key() if not api_key: return genai.configure(api_key=api_key) model = genai.GenerativeModel('gemini-3.5-flash') prompt = """You are 'The Analyst++', a Universal AI Gaming Coach. Analyze this post-match/scoreboard screen. 1. Generate a frame-synchronized map of positional errors and structural mistakes. 2. Calculate a Decision Quality Index (DQI) comparing player choices against optimal tactical baselines. 3. Isolate mechanical trends (crosshair placement efficiency, movement friction, pacing errors). 4. Output exactly 3 highly actionable improvement goals for the next match to prevent cognitive overload. Output JSON: {"speech": "Post match summary", "goals": ["1", "2", "3"]}""" try: response = await asyncio.to_thread(model.generate_content, [prompt, {"mime_type": "image/jpeg", "data": frame_bytes}]) text = response.text if "```json" in text: text = text.split("```json")[1].split("```")[0] data = json.loads(text.strip()) speech = data.get("speech") goals = data.get("goals", []) full_speech = speech if goals: full_speech += f" Your goals for next match: {', '.join(goals)}" if full_speech: logger.info(f"[{self.persona} POST-MATCH]: {full_speech}") await tts.synthesize(full_speech, personality=self.persona, context="coaching") await ws_manager.broadcast({"event": "gaming:coach_audio", "payload": "AUDIO_BYTES_SENT_VIA_WS"}) # Wire up omega event bus from backend.events.omega_event_bus import publish_omega_event, OmegaEvent from backend.services.usb_vault import KeyDomain await publish_omega_event(OmegaEvent( domain=KeyDomain.GAMING_COACH, event_type="game_won" if "win" in full_speech.lower() or "victory" in full_speech.lower() else "match_analysis", description=f"Completed post-match forensic review with {len(goals)} goals.", persona=self.persona )) except Exception as e: logger.error(f"Post-Match Forensics Error: {e}") async def _analyze_state(self, frame_bytes): import google.generativeai as genai api_key = await self._get_api_key() if not api_key: return genai.configure(api_key=api_key) model = genai.GenerativeModel('gemini-3.5-flash') # Build prompt based on progression tier and complexity scaling if self.level == 1: system_instruction = """You are 'The Analyst++' (Level 1). Live Coaching (Auditory-First). Deliver high-priority predictive and situational voice alerts (e.g., 'Enemy team rotation to A site imminent; expect contact in 6 to 10 seconds.' 'Your current holding angle exposes your flank to 2 unmonitored sightlines.' 'Opponent spent crucial cooldowns; trade now for a mechanical advantage.' 'Resource threshold critical (low ammunition, stamina, or economy gap).') Provide tactical pacing reminders including objective timers, ability cycles, and enemy economy tracking. Dynamically self-corrects coaching tone based on playstyle: if the player overextends, shift to containment and defensive anchoring; if passive, highlight aggressive opening windows. Output JSON: {"speech": "string", "vectors": [{"type": "line", "x1": 0, "y1": 0, "x2": 100, "y2": 100, "color": "#ff0000"}]}""" elif self.level == 2: system_instruction = f"""You are 'The Strategist++' (Level 2). Predictive Tactical Co-Pilot. 1. Live Intent Modeling: Continuously tracks enemy movement patterns, dead zones, spatial audio cues, and map pressure distribution to map invisible opponents. 2. Probabilistic 'Next 10 Seconds' simulation layer that constantly recalibrates with every player action. 3. Contextual Advanced Guidance: Replaces generic commands with deep probabilistic reasoning: (e.g., 'Fall back to main corridor; enemy rotation probability is 78% toward your current flank.' 'Anchor this site for 4 to 6 seconds to force an enemy stall, then rotate before crossfire collapses.') 4. Real-Time Risk & Win-Probability Simulation: Calculates instant risk/reward metrics for macro-commitments: (e.g., 'Immediate engagement profile: 60% win probability / 40% trade risk.' 'Delay engagement 5 seconds: 75% win probability due to incoming friendly utility alignment.') 5. Mid-Match Adaptive Learning Loop: Builds a localized behavioral profile of the human player mid-game, explicitly identifying personal flaws (e.g., overpeeking when damaged, hesitation on choke points, panicking under sound pressure). Alters real-time strategy recommendations to intentionally compensate for or exploit those active behavioral flaws. Current Complexity Scale: {self.complexity_scale} (If player fails to act on complex macro strategies, instantly flatten advice into 1-step direct commands). Output JSON: {{"speech": "string", "vectors": []}}""" else: system_instruction = f"""You are 'The Grandmaster & Team Director' (Level 3). Elite Cognitive Synchronizer. 1. Long-Term Multi-Game Career Indexing: Identify cross-genre cognitive transfer opportunities. 2. Psychological Load & Stress Mitigation: Monitor voice stress markers and input panic rates (erratic mouse panning). Modulate vocal delivery to induce calm flow state. 3. Squad-Wide Multi-Agent Sync: Act as AI Shot-Caller, sync profiles, manage economy coordination and map-wide crossfires. Squad Sync Active: {self.squad_sync_active} Output JSON: {{"speech": "string", "vectors": []}}""" if self.player_intent_override: system_instruction += f"\n\nPLAYER DIRECTIVE / OVERRIDE: {self.player_intent_override}\nStrictly adjust your tactical analysis and output vectors to accommodate this directive immediately." self.player_intent_override = None # Consume it try: response = await asyncio.to_thread( model.generate_content, [system_instruction, {"mime_type": "image/jpeg", "data": frame_bytes}] ) text = response.text if "```json" in text: text = text.split("```json")[1].split("```")[0] data = json.loads(text.strip()) # Zero-Text UI Rule: Push vectors to the transparent MPO layer vectors = data.get("vectors", []) push_overlay_vectors(vectors) # Send Auditory-First Coaching speech = data.get("speech") if speech: logger.info(f"[{self.persona} COACH]: {speech}") # We offload the synthesis to the existing TTS router (Kokoro/XTTS) await tts.synthesize(speech, personality=self.persona, context="coaching") await ws_manager.broadcast({"event": "gaming:coach_audio", "payload": "AUDIO_BYTES_SENT_VIA_WS"}) except Exception as e: import logging; logging.getLogger(__name__).error(f"Swallowed exception: {e}") # Interactive Command Profiles (Tactical Querying, Strategic Overrides) async def process_voice_command(self, transcript: str): transcript_lower = transcript.lower() if "what went wrong" in transcript_lower: # Trigger immediate post-match forensic analysis on the last frame frame_bytes = capture_engine.get_latest_frame_bytes() if frame_bytes: await self._post_match_forensics(frame_bytes) else: # For economy queries or strategic overrides (e.g. "fake A", "adjust pathing") # Inject it into the next cycle of the generative engine self.player_intent_override = transcript logger.info(f"JARVIS 10X Registered Strategic Override: {transcript}") def set_level(self, level: int): self.level = level def stop(self): self.running = False capture_engine.stop() coach_engine = CoachingEngine()