perception/__init__.py

"""
Perception Layer — ScreenPipe-style desktop telemetry tools.
"""

import os
import platform
import subprocess
import time
from typing import Any


def get_desktop_context() -> dict[str, Any]:
    info = {"platform": platform.platform(), "hostname": platform.node()}
    try:
        if platform.system() == "Linux":
            result = subprocess.run(
                ["ps", "-eo", "pid,comm,etime", "--sort=-%cpu"],
                capture_output=True, text=True, timeout=5,
            )
            info["top_processes"] = result.stdout[:1000]
    except Exception:
        info["top_processes"] = "unavailable"
    info["timestamp"] = time.time()
    return info


def list_recent_activity(limit: int = 10) -> list[dict]:
    try:
        if platform.system() == "Linux":
            result = subprocess.run(
                ["ps", "-eo", "pid,comm,start,etimes", "--sort=-start"],
                capture_output=True, text=True, timeout=5,
            )
            lines = result.stdout.strip().split("\n")[1:limit+1]
            activities = []
            for line in lines:
                parts = line.strip().split(None, 3)
                if len(parts) >= 2:
                    activities.append({
                        "pid": parts[0],
                        "command": parts[1],
                        "start": parts[2] if len(parts) > 2 else "?",
                    })
            return activities
    except Exception:
        pass
    return []


def search_content(query: str, path: str | None = None) -> list[dict]:
    search_dir = path or os.path.expanduser("~/vitalis-mcp-server/storage/memory")
    results = []
    if not os.path.isdir(search_dir):
        return results
    for fname in os.listdir(search_dir):
        fpath = os.path.join(search_dir, fname)
        if os.path.isfile(fpath) and fname.endswith(".jsonl"):
            try:
                with open(fpath) as f:
                    for line in f:
                        if query.lower() in line.lower():
                            results.append({"file": fname, "match": line[:200]})
                            if len(results) >= 10:
                                return results
            except Exception:
                pass
    return results