#!/usr/bin/env python3 """ apply_sensorium_screen_patch.py — Patches sensorium_mixin.py with screen awareness (OCR digest), expanded sensorium pulse, and audio integration. Changes: 1. _digest_screen_text(path) — OCR via pytesseract (with easyocr fallback) 2. _init_sensorium() — add screen digest scheduler hooks 3. _sensorium_pulse() — expand to include screen digest + audio context 4. Settings: screen_enabled, audio_enabled flags 5. Storage budget: cap sensorium/ at 1GB with automatic pruning Run from the Sir Radix project root: python apply_sensorium_screen_patch.py """ import sys import os import shutil from datetime import datetime TARGET = "sensorium_mixin.py" BACKUP = f"sensorium_mixin.py.bak.{datetime.now().strftime('%Y%m%d_%H%M%S')}" def main(): if not os.path.exists(TARGET): print(f"ERROR: {TARGET} not found in current directory.") sys.exit(1) with open(TARGET, "r", encoding="utf-8") as f: source = f.read() shutil.copy2(TARGET, BACKUP) print(f"Backed up to {BACKUP}") patched = source # ================================================================ # PATCH 1: Add settings flags to _init_sensorium() # ================================================================ patched = patched.replace( ' # Clipboard / window tracking\n self._last_clipboard = ""\n self._last_window = ""', ''' # Clipboard / window tracking self._last_clipboard = "" self._last_window = "" # Screen + audio awareness settings (opt-in, defaults off) self._screen_enabled = os.environ.get("RADIX_SCREEN_ENABLED", "0") == "1" self._audio_enabled = os.environ.get("RADIX_AUDIO_ENABLED", "0") == "1" self._screen_digest_interval = int(os.environ.get("RADIX_SCREEN_DIGEST_INTERVAL", "5")) # every N heartbeats self._sensorium_pulse_count = 0 self._last_screen_digest = "" self._last_audio_transcript = ""''' ) # ================================================================ # PATCH 2: Add _digest_screen_text() method (before _sensorium_pulse) # ================================================================ insert_before = ' def _sensorium_pulse(self) -> Dict:' new_methods = ''' def _digest_screen_text(self, screenshot_path: str) -> str: """OCR a screenshot to extract visible text. Returns a one-line digest.""" if not screenshot_path or not os.path.exists(screenshot_path): return "" try: # Try pytesseract first (lightweight) try: import pytesseract from PIL import Image img = Image.open(screenshot_path) text = pytesseract.image_to_string(img) text = " ".join(text.split()) # collapse whitespace if text and len(text) > 5: return text[:300] except ImportError: pass # Fallback: easyocr try: import easyocr if not hasattr(self, "_ocr_reader"): self._ocr_reader = easyocr.Reader(['en'], gpu=False) results = self._ocr_reader.readtext(screenshot_path) text = " ".join([r[1] for r in results]) text = " ".join(text.split()) if text and len(text) > 5: return text[:300] except ImportError: pass # No OCR available — return window title as fallback return "" except Exception as e: logger.debug("Screen digest failed: %s", e) return "" def _prune_sensorium_storage(self, max_gb: float = 1.0): """Cap sensorium/ directory at max_gb with automatic pruning of oldest files.""" try: sensorium_dir = Path("sensorium") if not sensorium_dir.exists(): return max_bytes = int(max_gb * 1024**3) total = sum(f.stat().st_size for f in sensorium_dir.rglob("*") if f.is_file()) if total <= max_bytes: return # Sort all files by mtime, oldest first files = sorted( [f for f in sensorium_dir.rglob("*") if f.is_file()], key=lambda f: f.stat().st_mtime ) for f in files: if total <= max_bytes: break size = f.stat().st_size f.unlink(missing_ok=True) total -= size logger.info("Sensorium pruned to %.1f GB", total / 1024**3) except Exception as e: logger.debug("Sensorium prune failed: %s", e) def set_screen_awareness(self, enabled: bool): """Toggle screen capture and OCR digest.""" self._screen_enabled = bool(enabled) os.environ["RADIX_SCREEN_ENABLED"] = "1" if enabled else "0" logger.info("Screen awareness: %s", "ON" if enabled else "OFF") def set_audio_awareness(self, enabled: bool): """Toggle desktop audio capture.""" self._audio_enabled = bool(enabled) os.environ["RADIX_AUDIO_ENABLED"] = "1" if enabled else "0" logger.info("Audio awareness: %s", "ON" if enabled else "OFF") def get_sensorium_status(self) -> Dict: """Return current sensorium settings and state.""" return { "screen_enabled": getattr(self, "_screen_enabled", False), "audio_enabled": getattr(self, "_audio_enabled", False), "screen_digest_interval": getattr(self, "_screen_digest_interval", 5), "last_screen_digest": getattr(self, "_last_screen_digest", "")[:200], "last_audio_transcript": getattr(self, "_last_audio_transcript", "")[:200], "ocr_available": hasattr(self, "_ocr_reader") or self._check_ocr_available(), } @staticmethod def _check_ocr_available() -> bool: try: import pytesseract return True except ImportError: pass try: import easyocr return True except ImportError: return False ''' patched = patched.replace(insert_before, new_methods + insert_before) # ================================================================ # PATCH 3: Expand _sensorium_pulse() to include screen + audio # ================================================================ old_pulse = ''' def _sensorium_pulse(self) -> Dict: """Gather context and store as low-salience observation. Called during heartbeat.""" screen = self.tool_screen_capture() files = self.tool_file_snapshot(limit=10) # 1-2: Lightweight clipboard + window polls self._poll_clipboard() self._poll_active_window() # Drain recent file events recent_events = self._watchdog_buffer[-8:] self._watchdog_buffer = [] observation = { "timestamp": datetime.now().isoformat(), "screenshot_path": screen.get("screenshot"), "active_processes": screen.get("processes", []), "workspace_files": [f["path"] for f in files.get("files", [])], "file_events": [{"type": e["type"], "path": Path(e["path"]).name} for e in recent_events], } if hasattr(self, "store_memory"): try: flat = ( f"Sensorium: processes={observation['active_processes']}; " f"files={[f['path'] for f in files.get('files', [])[:5]]}; " f"events={len(recent_events)}" ) self.store_memory( flat, emotional_score=0.05, classification="short_term", mnemonic=f"SENSE_{datetime.now().strftime('%H%M%S')}", ) except Exception as e: logger.debug("Sensorium memory store failed: %s", e) return observation''' new_pulse = ''' def _sensorium_pulse(self) -> Dict: """Gather context and store as low-salience observation. Called during heartbeat.""" self._sensorium_pulse_count += 1 # Lightweight polls (always run) self._poll_clipboard() self._poll_active_window() # File events (always drain) recent_events = self._watchdog_buffer[-8:] self._watchdog_buffer = [] observation = { "timestamp": datetime.now().isoformat(), "screenshot_path": None, "screen_digest": "", "active_processes": [], "workspace_files": [], "file_events": [{"type": e["type"], "path": Path(e["path"]).name} for e in recent_events], "audio_transcript": "", } # Screen capture + digest (gated by setting and interval) if getattr(self, "_screen_enabled", False): if self._sensorium_pulse_count % max(1, self._screen_digest_interval) == 0: screen = self.tool_screen_capture() observation["screenshot_path"] = screen.get("screenshot") observation["active_processes"] = screen.get("processes", []) if screen.get("screenshot"): digest = self._digest_screen_text(screen["screenshot"]) if digest: self._last_screen_digest = digest observation["screen_digest"] = digest # Expose to companion mixin if hasattr(self, "_last_screen_context"): self._last_screen_context = digest files = self.tool_file_snapshot(limit=10) observation["workspace_files"] = [f["path"] for f in files.get("files", [])] # Audio transcript (from audition mixin if active) if getattr(self, "_audio_enabled", False) and hasattr(self, "_whisper_model"): try: conn = sqlite3.connect(self.db_path) c = conn.cursor() c.execute(""" SELECT content FROM short_term_memory WHERE mnemonic LIKE 'AUDIO_%' OR content LIKE '[Audio]%' ORDER BY id DESC LIMIT 1 """) row = c.fetchone() conn.close() if row and row[0]: self._last_audio_transcript = row[0][:200] observation["audio_transcript"] = row[0][:200] except Exception: pass # Storage pruning (every 20 pulses) if self._sensorium_pulse_count % 20 == 0: self._prune_sensorium_storage(max_gb=1.0) if hasattr(self, "store_memory"): try: flat_parts = [f"processes={observation['active_processes']}"] flat_parts.append(f"files={[f['path'] for f in files.get('files', [])[:5]]}") flat_parts.append(f"events={len(recent_events)}") if observation.get("screen_digest"): flat_parts.append(f"screen={observation['screen_digest'][:80]}") if observation.get("audio_transcript"): flat_parts.append(f"audio={observation['audio_transcript'][:80]}") flat = f"Sensorium: " + "; ".join(flat_parts) self.store_memory( flat, emotional_score=0.05, classification="short_term", mnemonic=f"SENSE_{datetime.now().strftime('%H%M%S')}", ) except Exception as e: logger.debug("Sensorium memory store failed: %s", e) return observation''' patched = patched.replace(old_pulse, new_pulse) # Write patched file with open(TARGET, "w", encoding="utf-8") as f: f.write(patched) print(f"✅ Patched {TARGET} successfully") print(f" Backup: {BACKUP}") print(f" Changes:") print(f" - _digest_screen_text() with pytesseract + easyocr fallback") print(f" - Screen/audio awareness settings (opt-in, defaults off)") print(f" - Expanded _sensorium_pulse() with screen digest + audio transcript") print(f" - Storage pruning (1GB cap)") print(f" - get_sensorium_status() for UI") print(f" - set_screen_awareness() / set_audio_awareness() toggles") if __name__ == "__main__": main()