qinglong
278 lines
11 KiB
Python
278 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
import os
|
|
import time
|
|
import platform
|
|
import logging
|
|
from typing import Dict, Any, Tuple
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class SystemInfoCollector:
|
|
def __init__(self):
|
|
self.is_android = 'ANDROID_ROOT' in os.environ or os.path.exists('/system/bin/getprop')
|
|
self.psutil = None
|
|
if not self.is_android:
|
|
try:
|
|
import psutil
|
|
self.psutil = psutil
|
|
except ImportError as e:
|
|
logger.debug(f"psutil not available: {e}")
|
|
else:
|
|
logger.info("🤖 Android 平台识别，启用原生采集")
|
|
|
|
# State for speed calculation (rx_bytes, tx_bytes, timestamp)
|
|
self._prev_net: Tuple[int, int, float] | None = None
|
|
|
|
# ── Raw counters (used by both total and speed) ──
|
|
|
|
def _read_net_bytes(self) -> Tuple[int, int]:
|
|
"""Read raw (rx_bytes, tx_bytes) since boot from system"""
|
|
try:
|
|
if self.psutil:
|
|
io = self.psutil.net_io_counters()
|
|
return (io.bytes_recv, io.bytes_sent)
|
|
except Exception:
|
|
pass
|
|
|
|
# Android / Linux: parse /proc/net/dev
|
|
try:
|
|
rx = tx = 0
|
|
with open('/proc/net/dev', 'r') as f:
|
|
for line in f:
|
|
if ':' not in line or 'lo' in line:
|
|
continue
|
|
parts = line.split(':')[1].split()
|
|
rx += int(parts[0])
|
|
if len(parts) > 8:
|
|
tx += int(parts[8])
|
|
return (rx, tx)
|
|
except Exception as e:
|
|
logger.warning(f"Network read failed: {e}")
|
|
return (0, 0)
|
|
|
|
# ── Public collectors ──
|
|
|
|
def get_all(self) -> Dict[str, Any]:
|
|
return {
|
|
"platform": {
|
|
"system": platform.system(),
|
|
"machine": platform.machine(),
|
|
"release": platform.release(),
|
|
"python": platform.python_version(),
|
|
},
|
|
"cpu": self._get_cpu(),
|
|
"memory": self._get_memory(),
|
|
"swap": self._get_swap(),
|
|
"disk": self._get_disk(),
|
|
"partitions": self._get_partitions(),
|
|
"process": self._get_process(),
|
|
"network": self._get_network(),
|
|
"net_speed": self._get_network_speed(),
|
|
"temperature": self._get_temperature(),
|
|
}
|
|
|
|
def _get_cpu(self) -> Dict[str, Any]:
|
|
if self.psutil:
|
|
return {
|
|
"percent": self.psutil.cpu_percent(interval=0.1),
|
|
"cores": self.psutil.cpu_count(),
|
|
"load_avg": os.getloadavg() if hasattr(os, 'getloadavg') else [0, 0, 0]
|
|
}
|
|
try:
|
|
load = os.getloadavg() if hasattr(os, "getloadavg") else [0.0, 0.0, 0.0]
|
|
cores = os.cpu_count() or 1
|
|
percent = min(100.0, (load[0] / cores) * 100)
|
|
return {"percent": round(percent, 1), "cores": cores, "load_avg": load}
|
|
except Exception as e:
|
|
logger.error(f"CPU info error: {e}", exc_info=True)
|
|
return {"percent": 0, "cores": 0, "load_avg": [0, 0, 0]}
|
|
|
|
def _get_memory(self) -> Dict[str, Any]:
|
|
if self.psutil:
|
|
m = self.psutil.virtual_memory()
|
|
return {"total_gb": round(m.total / 1073741824, 1),
|
|
"used_gb": round(m.used / 1073741824, 1),
|
|
"percent": m.percent}
|
|
try:
|
|
mem = {}
|
|
with open('/proc/meminfo') as f:
|
|
for line in f:
|
|
parts = line.split()
|
|
if len(parts) >= 2:
|
|
mem[parts[0].rstrip(':')] = int(parts[1]) * 1024
|
|
t = mem.get('MemTotal', 1)
|
|
a = mem.get('MemAvailable', mem.get('MemFree', 0))
|
|
return {"total_gb": round(t / 1073741824, 1),
|
|
"used_gb": round((t - a) / 1073741824, 1),
|
|
"percent": round(((t - a) / t) * 100, 1)}
|
|
except Exception as e:
|
|
logger.warning(f"Memory info failed: {e}")
|
|
return {"total_gb": 0, "used_gb": 0, "percent": 0}
|
|
|
|
def _get_swap(self) -> Dict[str, Any]:
|
|
if self.psutil:
|
|
try:
|
|
s = self.psutil.swap_memory()
|
|
return {"total_gb": round(s.total / 1073741824, 1),
|
|
"used_gb": round(s.used / 1073741824, 1),
|
|
"percent": s.percent}
|
|
except Exception:
|
|
pass
|
|
try:
|
|
with open('/proc/meminfo') as f:
|
|
mem = {}
|
|
for line in f:
|
|
parts = line.split()
|
|
if len(parts) >= 2:
|
|
mem[parts[0].rstrip(':')] = int(parts[1]) * 1024
|
|
t = mem.get('SwapTotal', 0)
|
|
f_swap = mem.get('SwapFree', mem.get('SwapCached', 0))
|
|
if t > 0:
|
|
used = t - f_swap
|
|
return {"total_gb": round(t / 1073741824, 1),
|
|
"used_gb": round(used / 1073741824, 1),
|
|
"percent": round((used / t) * 100, 1)}
|
|
except Exception:
|
|
pass
|
|
return {"total_gb": 0, "used_gb": 0, "percent": 0}
|
|
|
|
_SKIP_FS = {'tmpfs', 'devtmpfs', 'devfs', 'overlay', 'squashfs', 'proc', 'sysfs',
|
|
'cgroup', 'cgroup2', 'debugfs', 'tracefs', 'fusectl', 'configfs',
|
|
'securityfs', 'pstore', 'efivarfs', 'autofs', 'ramfs', 'hugetlbfs',
|
|
'mqueue', 'bpf', 'binfmt_misc', 'rpc_pipefs', 'nfsd', 'smb3fs_ctl',
|
|
'snapfuse', 'fuse.gvfsd-fuse', 'fuse.portal'}
|
|
|
|
def _get_partitions(self) -> list:
|
|
"""获取存储分区列表 (过滤虚拟文件系统)"""
|
|
parts = []
|
|
if self.psutil:
|
|
try:
|
|
for p in self.psutil.disk_partitions(all=False):
|
|
if p.fstype and p.fstype.lower() in self._SKIP_FS:
|
|
continue
|
|
try:
|
|
usage = self.psutil.disk_usage(p.mountpoint)
|
|
# Windows: 用盘符作为设备名 (C:, D:)
|
|
dev = p.device
|
|
if os.name == 'nt':
|
|
dev = p.mountpoint.rstrip('\\') # C:\ → C:
|
|
parts.append({
|
|
"device": dev,
|
|
"mount": p.mountpoint,
|
|
"fstype": p.fstype or "",
|
|
"total_gb": round(usage.total / 1073741824, 1),
|
|
"used_gb": round(usage.used / 1073741824, 1),
|
|
"percent": usage.percent,
|
|
})
|
|
except (PermissionError, OSError):
|
|
dev = p.device
|
|
if os.name == 'nt':
|
|
dev = p.mountpoint.rstrip('\\')
|
|
parts.append({
|
|
"device": dev,
|
|
"mount": p.mountpoint,
|
|
"fstype": p.fstype or "",
|
|
"total_gb": 0, "used_gb": 0, "percent": 0,
|
|
})
|
|
except Exception:
|
|
pass
|
|
if not parts:
|
|
# Fallback: just root partition
|
|
disk = self._get_disk()
|
|
if disk.get("total_gb", 0) > 0:
|
|
parts.append({
|
|
"device": "/",
|
|
"mount": "/",
|
|
"fstype": "",
|
|
"total_gb": disk["total_gb"],
|
|
"used_gb": disk["used_gb"],
|
|
"percent": disk["percent"],
|
|
})
|
|
return parts
|
|
|
|
def _get_disk(self) -> Dict[str, Any]:
|
|
if self.psutil:
|
|
try:
|
|
d = self.psutil.disk_usage('/')
|
|
return {"total_gb": round(d.total / 1073741824, 1),
|
|
"used_gb": round(d.used / 1073741824, 1),
|
|
"percent": d.percent}
|
|
except Exception:
|
|
pass
|
|
return {"total_gb": 0, "used_gb": 0, "percent": 0}
|
|
|
|
def _get_temperature(self) -> Dict[str, Any]:
|
|
if self.psutil:
|
|
try:
|
|
temps = self.psutil.sensors_temperatures()
|
|
if temps:
|
|
for name, entries in temps.items():
|
|
for e in entries:
|
|
if e.current > 0:
|
|
return {"name": e.label or name, "current": round(e.current, 1)}
|
|
except Exception:
|
|
pass
|
|
# Android fallback: read thermal zone
|
|
try:
|
|
for i in range(10):
|
|
tz = f'/sys/class/thermal/thermal_zone{i}/temp'
|
|
ttype = f'/sys/class/thermal/thermal_zone{i}/type'
|
|
if os.path.exists(tz) and os.path.exists(ttype):
|
|
with open(ttype) as f:
|
|
name = f.read().strip()
|
|
if 'cpu' in name.lower() or 'soc' in name.lower():
|
|
with open(tz) as f:
|
|
temp = int(f.read().strip()) / 1000.0
|
|
return {"name": name, "current": round(temp, 1)}
|
|
except Exception:
|
|
pass
|
|
return {"name": "", "current": 0}
|
|
|
|
def _get_process(self) -> Dict[str, Any]:
|
|
"""Current process info (PID + RSS memory)"""
|
|
pid = os.getpid()
|
|
mem_mb = 0
|
|
try:
|
|
if self.psutil:
|
|
import psutil
|
|
mem_mb = round(psutil.Process(pid).memory_info().rss / 1048576, 1)
|
|
else:
|
|
# Parse /proc/self/status for VmRSS
|
|
with open('/proc/self/status', 'r') as f:
|
|
for line in f:
|
|
if line.startswith('VmRSS:'):
|
|
mem_mb = round(int(line.split()[1]) / 1024, 1)
|
|
break
|
|
except Exception as e:
|
|
logger.debug(f"Process info failed: {e}")
|
|
return {"pid": pid, "memory_mb": mem_mb}
|
|
|
|
def _get_network(self) -> Dict[str, Any]:
|
|
"""Cumulative RX/TX since boot (MB)"""
|
|
try:
|
|
rx, tx = self._read_net_bytes()
|
|
return {"rx_mb": round(rx / 1048576, 1),
|
|
"tx_mb": round(tx / 1048576, 1)}
|
|
except Exception as e:
|
|
logger.warning(f"Network info failed: {e}")
|
|
return {"rx_mb": 0, "tx_mb": 0}
|
|
|
|
def _get_network_speed(self) -> Dict[str, Any]:
|
|
"""Real-time RX/TX speed (bytes/sec), computed from delta between calls"""
|
|
now = time.time()
|
|
rx, tx = self._read_net_bytes()
|
|
|
|
if self._prev_net:
|
|
prx, ptx, pt = self._prev_net
|
|
dt = now - pt
|
|
speed_rx = (rx - prx) / dt if dt > 0 and rx >= prx else 0
|
|
speed_tx = (tx - ptx) / dt if dt > 0 and tx >= ptx else 0
|
|
else:
|
|
speed_rx = speed_tx = 0
|
|
|
|
self._prev_net = (rx, tx, now)
|
|
return {"rx_bytes_sec": round(speed_rx, 1),
|
|
"tx_bytes_sec": round(speed_tx, 1)}
|