AskaEth
f5800cc6c3
- headless: TUI skip now works correctly - example_plugin: keyword args compat (plugin_name=, config=, bridge=) - sysmon: psutil made optional (graceful degrade) - tui_service: SysMonWidget optional - All 22 modules import clean, 23 tests pass, integration verified
91 lines
3.1 KiB
Python
91 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
"""SenSu TUI 系统监控组件 — CPU/内存/磁盘/插件状态"""
|
|
try:
|
|
import psutil
|
|
except ImportError:
|
|
psutil = None
|
|
import time, asyncio, logging
|
|
from textual.widgets import Static
|
|
from textual.reactive import reactive
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def get_system_stats():
|
|
"""获取系统实时状态"""
|
|
try:
|
|
cpu = psutil.cpu_percent(interval=0.1)
|
|
mem = psutil.virtual_memory()
|
|
disk = psutil.disk_usage("/")
|
|
net = psutil.net_io_counters()
|
|
return {
|
|
"cpu": cpu,
|
|
"mem_total": mem.total,
|
|
"mem_used": mem.used,
|
|
"mem_percent": mem.percent,
|
|
"disk_total": disk.total,
|
|
"disk_used": disk.used,
|
|
"disk_percent": disk.percent,
|
|
"net_sent": net.bytes_sent,
|
|
"net_recv": net.bytes_recv,
|
|
"time": time.time(),
|
|
}
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
def format_bytes(b):
|
|
if b < 1024: return f"{b}B"
|
|
if b < 1024**2: return f"{b/1024:.1f}K"
|
|
if b < 1024**3: return f"{b/1024**2:.1f}M"
|
|
return f"{b/1024**3:.1f}G"
|
|
|
|
def make_bar(percent, width=20, filled="█", empty="░"):
|
|
n = int(percent / 100 * width)
|
|
return filled * n + empty * (width - n)
|
|
|
|
class SysMonWidget(Static):
|
|
"""系统监控显示组件"""
|
|
stats = reactive({})
|
|
plugin_info = reactive("")
|
|
|
|
def __init__(self):
|
|
super().__init__("系统监控初始化中...")
|
|
self._refresh_task = None
|
|
|
|
def on_mount(self):
|
|
self._refresh_task = asyncio.create_task(self._periodic_refresh())
|
|
|
|
async def _periodic_refresh(self):
|
|
while True:
|
|
try:
|
|
s = get_system_stats()
|
|
if "error" not in s:
|
|
cpu_bar = make_bar(s["cpu"], 20)
|
|
mem_bar = make_bar(s["mem_percent"], 20)
|
|
disk_bar = make_bar(s["disk_percent"], 20)
|
|
|
|
text = (
|
|
f"[bold cyan]━━━ 系统状态 ━━━[/]\n"
|
|
f"CPU {cpu_bar} {s['cpu']:5.1f}%\n"
|
|
f"MEM {mem_bar} {s['mem_percent']:5.1f}% "
|
|
f"({format_bytes(s['mem_used'])}/{format_bytes(s['mem_total'])})\n"
|
|
f"DISK {disk_bar} {s['disk_percent']:5.1f}% "
|
|
f"({format_bytes(s['disk_used'])}/{format_bytes(s['disk_total'])})\n"
|
|
f"NET ↑{format_bytes(s['net_sent'])} ↓{format_bytes(s['net_recv'])}"
|
|
)
|
|
if self.plugin_info:
|
|
text += f"\n\n[bold yellow]━━━ 插件状态 ━━━[/]\n{self.plugin_info}"
|
|
self.update(text)
|
|
await asyncio.sleep(2)
|
|
except asyncio.CancelledError:
|
|
break
|
|
except Exception as e:
|
|
logger.debug(f"SysMon refresh error: {e}")
|
|
await asyncio.sleep(5)
|
|
|
|
def update_plugins(self, info: str):
|
|
self.plugin_info = info
|
|
|
|
def on_unmount(self):
|
|
if self._refresh_task:
|
|
self._refresh_task.cancel()
|