File size: 2,260 Bytes
35341c0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | """
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
|