apply_sensorium_screen_patch.py

#!/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()