#!/usr/bin/env python3 """Per-module and per-subprocess resource monitoring. Tracks RAM, CPU, disk usage, and thermal readings. Publishes RESOURCE_WARNING, RESOURCE_CRITICAL, and THERMAL_WARNING events. OOM-kills the lowest-priority module when memory exceeds tier limits. Sheds modules at 70C thermal threshold. """ import os import time import logging import threading from pathlib import Path from typing import Optional from core.bus import EventBus from core.state import StateManager logger = logging.getLogger(__name__) # Thermal thresholds (Celsius) THERMAL_WARNING = 55 THERMAL_CRITICAL = 70 # Disk usage threshold DISK_WARNING_PCT = 80 DISK_CRITICAL_PCT = 90 # Check interval CHECK_INTERVAL = 15.0 # seconds class ResourceMonitor: """Monitors system and per-process resources. Integrates with Engine (module processes) and ToolManager (subprocess PIDs) to provide unified resource tracking. """ def __init__(self, bus: EventBus, state: StateManager, config: dict, engine=None, tool_manager=None): self.bus = bus self.state = state self.config = config self.engine = engine self.tool_manager = tool_manager self._thread: Optional[threading.Thread] = None self._running = False # Load tier limits tier = config.get("device", {}).get("platform", "generic") from core.engine import HARDWARE_TIERS, detect_hardware_tier if tier == "auto": tier = detect_hardware_tier() self._tier_limits = HARDWARE_TIERS.get(tier, HARDWARE_TIERS["generic"]) self._tier = tier # Thermal zone path (try common locations) self._thermal_path = self._find_thermal_zone() # State tracking self._thermal_warned = False self._resource_warned = False # ------------------------------------------------------------------ # Lifecycle # ------------------------------------------------------------------ def start(self) -> None: if self._running: return self._running = True self._thread = threading.Thread( target=self._monitor_loop, daemon=True, name="sensor-resource-monitor" ) self._thread.start() logger.info("ResourceMonitor started (tier=%s, thermal_zone=%s)", self._tier, self._thermal_path or "none") def stop(self) -> None: self._running = False if self._thread and self._thread.is_alive(): self._thread.join(timeout=5.0) logger.info("ResourceMonitor stopped") # ------------------------------------------------------------------ # Monitor loop # ------------------------------------------------------------------ def _monitor_loop(self) -> None: while self._running: try: self._check_resources() except Exception: logger.exception("Resource check error") time.sleep(CHECK_INTERVAL) def _check_resources(self) -> None: """Run all resource checks.""" self._check_memory() self._check_disk() self._check_thermal() # ------------------------------------------------------------------ # Memory # ------------------------------------------------------------------ def _check_memory(self) -> None: """Check total memory usage against tier limits.""" mem = self.get_system_memory() max_ram = self._tier_limits.get("max_ram_mb", 0) if max_ram <= 0: return used_mb = mem["used_mb"] if used_mb > max_ram * 0.95: # Critical — OOM kill lowest priority module logger.critical("Memory critical: %.0fMB / %dMB", used_mb, max_ram) self.bus.emit("RESOURCE_CRITICAL", {"resource": "memory", "used_mb": used_mb, "limit_mb": max_ram}, source_module="resource_monitor") self._oom_kill() elif used_mb > max_ram * 0.80: if not self._resource_warned: logger.warning("Memory warning: %.0fMB / %dMB", used_mb, max_ram) self.bus.emit("RESOURCE_WARNING", {"resource": "memory", "used_mb": used_mb, "limit_mb": max_ram}, source_module="resource_monitor") self._resource_warned = True else: self._resource_warned = False def _oom_kill(self) -> None: """Kill the lowest-priority (highest priority number) running module.""" if self.engine is None: return # Find lowest-priority running module worst_name = None worst_priority = -1000 status = self.engine.get_status() for name, info in status.items(): if not info.get("running"): continue prio = info.get("priority", 0) if prio > worst_priority: worst_priority = prio worst_name = name if worst_name: logger.warning("OOM killing module: %s (priority=%d)", worst_name, worst_priority) self.engine.stop(worst_name) self.bus.emit("MODULE_STOPPED", {"module": worst_name, "reason": "oom_kill"}, source_module="resource_monitor") # ------------------------------------------------------------------ # Disk # ------------------------------------------------------------------ def _check_disk(self) -> None: """Check disk usage on the data partition.""" storage_path = self.config.get("storage_path", "/opt/.cache/bb/storage") try: st = os.statvfs(storage_path) except OSError: try: st = os.statvfs("/") except OSError: return total = st.f_blocks * st.f_frsize free = st.f_bavail * st.f_frsize used_pct = ((total - free) / total * 100) if total > 0 else 0 if used_pct >= DISK_CRITICAL_PCT: self.bus.emit("RESOURCE_CRITICAL", {"resource": "disk", "used_pct": round(used_pct, 1), "path": storage_path}, source_module="resource_monitor") elif used_pct >= DISK_WARNING_PCT: self.bus.emit("RESOURCE_WARNING", {"resource": "disk", "used_pct": round(used_pct, 1), "path": storage_path}, source_module="resource_monitor") # ------------------------------------------------------------------ # Thermal # ------------------------------------------------------------------ def _check_thermal(self) -> None: """Read CPU temperature and emit warnings / shed modules.""" temp = self.get_temperature() if temp is None: return if temp >= THERMAL_CRITICAL: logger.critical("Thermal critical: %.1fC — shedding modules", temp) self.bus.emit("THERMAL_WARNING", {"temperature": temp, "action": "shedding"}, source_module="resource_monitor") self._thermal_shed() elif temp >= THERMAL_WARNING: if not self._thermal_warned: logger.warning("Thermal warning: %.1fC", temp) self.bus.emit("THERMAL_WARNING", {"temperature": temp, "action": "warning"}, source_module="resource_monitor") self._thermal_warned = True else: self._thermal_warned = False def _thermal_shed(self) -> None: """Shed (stop) active modules to reduce heat — keep passive + core.""" if self.engine is None: return status = self.engine.get_status() # Stop active modules first (highest priority number = least important) shedded = [] for name, info in sorted(status.items(), key=lambda x: x[1].get("priority", 0), reverse=True): if not info.get("running"): continue if info.get("type") == "active": self.engine.stop(name) shedded.append(name) if shedded: logger.warning("Thermal shed: stopped %s", shedded) # ------------------------------------------------------------------ # Readings # ------------------------------------------------------------------ @staticmethod def get_system_memory() -> dict: """Read system memory from /proc/meminfo.""" info = {"total_mb": 0, "available_mb": 0, "used_mb": 0} try: with open("/proc/meminfo", "r") as f: for line in f: if line.startswith("MemTotal:"): info["total_mb"] = int(line.split()[1]) / 1024 elif line.startswith("MemAvailable:"): info["available_mb"] = int(line.split()[1]) / 1024 info["used_mb"] = info["total_mb"] - info["available_mb"] except (FileNotFoundError, PermissionError, ValueError): pass return {k: round(v, 1) for k, v in info.items()} def get_temperature(self) -> Optional[float]: """Read CPU temperature in Celsius.""" if not self._thermal_path: return None try: with open(self._thermal_path, "r") as f: raw = f.read().strip() temp = int(raw) / 1000.0 # millidegrees return round(temp, 1) except (FileNotFoundError, PermissionError, ValueError): return None @staticmethod def get_cpu_usage() -> float: """Get overall CPU usage percentage from /proc/stat (1-second sample).""" def _read(): with open("/proc/stat", "r") as f: parts = f.readline().split() return [int(x) for x in parts[1:]] try: t1 = _read() time.sleep(1.0) t2 = _read() d = [t2[i] - t1[i] for i in range(len(t1))] idle = d[3] + (d[4] if len(d) > 4 else 0) total = sum(d) return round((1.0 - idle / total) * 100, 1) if total > 0 else 0.0 except Exception: return 0.0 def get_process_resources(self, pid: int) -> dict: """Get RAM/CPU for a specific PID.""" from core.tool_manager import ToolManager return ToolManager._read_proc_stats(pid) def get_snapshot(self) -> dict: """Full resource snapshot.""" return { "system_memory": self.get_system_memory(), "temperature": self.get_temperature(), "tier": self._tier, "tier_limits": self._tier_limits, } # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ @staticmethod def _find_thermal_zone() -> Optional[str]: """Find the CPU thermal zone file.""" base = Path("/sys/class/thermal") if not base.exists(): return None for zone in sorted(base.iterdir()): type_file = zone / "type" temp_file = zone / "temp" if not temp_file.exists(): continue try: ztype = type_file.read_text().strip().lower() if any(k in ztype for k in ("cpu", "soc", "x86_pkg")): return str(temp_file) except (PermissionError, FileNotFoundError): continue # Fallback: first thermal zone with a temp file for zone in sorted(base.iterdir()): temp_file = zone / "temp" if temp_file.exists(): return str(temp_file) return None