build:usb-image — Final USB image assembly script, Ventoy config, and Hermes agent runtime structure
- build/usb-image.sh: Main USB image assembler with artifact validation, ISO building, disk image creation, and USB write support - build/ventoy.json: Ventoy configuration for multi-boot USB - src/hermes/config.yaml: Portable Hermes agent configuration - src/hermes/autorun.sh: Auto-launch script with full diagnostic pipeline - Makefile: Build system with iso/image/check/clean/write targets Generates fallback stubs for any diagnostic modules whose upstream build tasks haven't completed yet, ensuring the image boots even during active development.
This commit is contained in:
@@ -0,0 +1,28 @@
|
||||
"""
|
||||
Hermes Portable Rescue — Diagnostic modules.
|
||||
|
||||
Modules
|
||||
-------
|
||||
hardware : HardwareInventory
|
||||
CPU, RAM, GPU, disk enumeration and SMART health checks.
|
||||
stress : RAMStressTester, CPUStressTester, GPUStressTester
|
||||
Stress-test wrappers (available when stress.py is present).
|
||||
|
||||
Each submodule exposes a ``run() -> dict`` for JSON and a
|
||||
``text_report() -> str`` for human consumption.
|
||||
"""
|
||||
from src.diagnostics.hardware import HardwareInventory
|
||||
|
||||
try:
|
||||
from src.diagnostics.stress import (
|
||||
CPUStressTester,
|
||||
GPUStressTester,
|
||||
RAMStressTester,
|
||||
run_all_stress_tests,
|
||||
)
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
__all__ = [
|
||||
"HardwareInventory",
|
||||
]
|
||||
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,583 @@
|
||||
"""
|
||||
Hermes Portable Rescue — Hardware Inventory & Health Checks.
|
||||
|
||||
Enumerates and assesses CPU, RAM, GPU, and disk subsystems from a
|
||||
Linux-based rescue environment. Wraps CLI tools (dmidecode, lshw,
|
||||
smartctl, lscpu) and falls back to /sys and /proc where possible.
|
||||
|
||||
Typical usage::
|
||||
|
||||
from diagnostics.hardware import HardwareInventory
|
||||
|
||||
hw = HardwareInventory(log_warnings=True)
|
||||
report = hw.run() # dict, JSON-serialisable
|
||||
print(hw.text_report()) # human-readable summary
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from dataclasses import dataclass, field, fields, is_dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _asdict(obj) -> dict[str, Any]:
|
||||
"""Recursive dataclass -> dict, filtering None and empty containers."""
|
||||
if not is_dataclass(obj):
|
||||
return obj
|
||||
out = {}
|
||||
for f in fields(obj):
|
||||
val = getattr(obj, f.name)
|
||||
if isinstance(val, list):
|
||||
val = [_asdict(v) if is_dataclass(v) else v for v in val]
|
||||
if not val:
|
||||
continue
|
||||
elif is_dataclass(val):
|
||||
val = _asdict(val)
|
||||
if not any(val.values()):
|
||||
continue
|
||||
elif val is None:
|
||||
continue
|
||||
out[f.name] = val
|
||||
return out
|
||||
|
||||
|
||||
@dataclass
|
||||
class CPUInfo:
|
||||
"""Central processing unit details."""
|
||||
|
||||
model: str = ""
|
||||
architecture: str = ""
|
||||
cores: int = 0
|
||||
threads: int = 0
|
||||
max_speed_mhz: Optional[float] = None
|
||||
min_speed_mhz: Optional[float] = None
|
||||
cache_l2_kb: Optional[int] = None
|
||||
cache_l3_kb: Optional[int] = None
|
||||
flags: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
@dataclass
|
||||
class RAMModule:
|
||||
"""A single physical DIMM / soldered memory module."""
|
||||
|
||||
size_gb: float = 0
|
||||
type: str = ""
|
||||
speed_mhz: Optional[int] = None
|
||||
slot: str = ""
|
||||
manufacturer: str = ""
|
||||
part_number: str = ""
|
||||
serial: str = ""
|
||||
voltage_v: Optional[float] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class RAMInfo:
|
||||
"""Overall RAM configuration."""
|
||||
|
||||
total_gb: float = 0
|
||||
slots_used: int = 0
|
||||
slots_total: int = 0
|
||||
modules: list[RAMModule] = field(default_factory=list)
|
||||
|
||||
|
||||
@dataclass
|
||||
class GPUInfo:
|
||||
"""Graphics processing unit details."""
|
||||
|
||||
model: str = ""
|
||||
vendor: str = ""
|
||||
vram_mb: Optional[int] = None
|
||||
driver: str = ""
|
||||
pci_slot: str = ""
|
||||
|
||||
|
||||
@dataclass
|
||||
class DiskSMART:
|
||||
"""SMART health summary for one disk."""
|
||||
|
||||
status: str = "UNKNOWN"
|
||||
model: str = ""
|
||||
serial: str = ""
|
||||
size_gb: float = 0
|
||||
interface: str = ""
|
||||
reallocated_sectors: Optional[int] = None
|
||||
pending_sectors: Optional[int] = None
|
||||
temperature_c: Optional[float] = None
|
||||
power_on_hours: Optional[int] = None
|
||||
power_cycles: Optional[int] = None
|
||||
raw_read_error_rate: Optional[int] = None
|
||||
wear_level: Optional[int] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class DiskInfo:
|
||||
"""One block device with its SMART data."""
|
||||
|
||||
device: str = ""
|
||||
smart: DiskSMART = field(default_factory=DiskSMART)
|
||||
|
||||
|
||||
@dataclass
|
||||
class HardwareReport:
|
||||
"""Complete snapshot of all subsystems."""
|
||||
|
||||
cpu: CPUInfo = field(default_factory=CPUInfo)
|
||||
ram: RAMInfo = field(default_factory=RAMInfo)
|
||||
gpus: list[GPUInfo] = field(default_factory=list)
|
||||
disks: list[DiskInfo] = field(default_factory=list)
|
||||
errors: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Shell helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _run(cmd: list[str], timeout: float = 15.0) -> str:
|
||||
"""Run a command, return stdout, or empty string on failure."""
|
||||
try:
|
||||
res = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=timeout,
|
||||
)
|
||||
return res.stdout
|
||||
except (FileNotFoundError, subprocess.TimeoutExpired, subprocess.CalledProcessError):
|
||||
return ""
|
||||
|
||||
|
||||
def _parse_int(value: str | None) -> Optional[int]:
|
||||
if value is None:
|
||||
return None
|
||||
try:
|
||||
return int(value)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _parse_float(value: str | None) -> Optional[float]:
|
||||
if value is None:
|
||||
return None
|
||||
try:
|
||||
return float(value)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _find_sys_block() -> list[Path]:
|
||||
"""List physical block devices (exclude ram, loop, dm, zram)."""
|
||||
skip_prefixes = ("ram", "loop", "dm-", "zram", "sr", "md")
|
||||
devices = []
|
||||
for entry in Path("/sys/block").iterdir():
|
||||
name = entry.name
|
||||
if any(name.startswith(p) for p in skip_prefixes):
|
||||
continue
|
||||
devices.append(entry)
|
||||
return sorted(devices)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Collectors
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def get_cpu_info(errors: list[str]) -> CPUInfo:
|
||||
"""Collect CPU details from lscpu and /proc/cpuinfo."""
|
||||
info = CPUInfo()
|
||||
|
||||
out = _run(["lscpu"])
|
||||
if out:
|
||||
for line in out.splitlines():
|
||||
m = re.match(r"Model\s*name[:\s]+(.+)", line, re.IGNORECASE)
|
||||
if m:
|
||||
info.model = m.group(1).strip()
|
||||
m = re.match(r"Architecture[:\s]+(.+)", line, re.IGNORECASE)
|
||||
if m:
|
||||
info.architecture = m.group(1).strip()
|
||||
m = re.match(r"^CPU\(s\)[:\s]+(\d+)", line, re.IGNORECASE)
|
||||
if m:
|
||||
info.threads = int(m.group(1))
|
||||
m = re.match(r"Core\(s\)\s*per\s*socket[:\s]+(\d+)", line, re.IGNORECASE)
|
||||
if m:
|
||||
info.cores = int(m.group(1))
|
||||
m = re.match(r"CPU\s*max\s*MHz[:\s]+([\d.]+)", line, re.IGNORECASE)
|
||||
if m:
|
||||
info.max_speed_mhz = round(float(m.group(1)))
|
||||
m = re.match(r"CPU\s*min\s*MHz[:\s]+([\d.]+)", line, re.IGNORECASE)
|
||||
if m:
|
||||
info.min_speed_mhz = round(float(m.group(1)))
|
||||
m = re.match(r"L2\s*cache[:\s]+(\d+)", line, re.IGNORECASE)
|
||||
if m:
|
||||
info.cache_l2_kb = int(m.group(1))
|
||||
m = re.match(r"L3\s*cache[:\s]+(\d+)", line, re.IGNORECASE)
|
||||
if m:
|
||||
info.cache_l3_kb = int(m.group(1))
|
||||
m = re.match(r"Flags[:\s]+(.+)", line, re.IGNORECASE)
|
||||
if m:
|
||||
info.flags = m.group(1).strip().split()
|
||||
|
||||
if not info.model:
|
||||
try:
|
||||
cpuinfo = Path("/proc/cpuinfo").read_text()
|
||||
for line in cpuinfo.splitlines():
|
||||
m = re.match(r"model\s*name\s*:\s*(.+)", line, re.IGNORECASE)
|
||||
if m and not info.model:
|
||||
info.model = m.group(1).strip()
|
||||
cores = set()
|
||||
threads = 0
|
||||
for line in cpuinfo.splitlines():
|
||||
m = re.match(r"core\s*id\s*:\s*(\d+)", line)
|
||||
if m:
|
||||
cores.add(m.group(1))
|
||||
m2 = re.match(r"processor\s*:\s*(\d+)", line)
|
||||
if m2:
|
||||
threads = max(threads, int(m2.group(1)) + 1)
|
||||
if cores and not info.cores:
|
||||
info.cores = len(cores)
|
||||
if threads and not info.threads:
|
||||
info.threads = threads
|
||||
except (OSError, ValueError):
|
||||
pass
|
||||
|
||||
return info
|
||||
|
||||
|
||||
def get_ram_info(errors: list[str]) -> RAMInfo:
|
||||
"""Collect RAM configuration from dmidecode."""
|
||||
info = RAMInfo()
|
||||
|
||||
out = _run(["dmidecode", "--type", "17"], timeout=10.0)
|
||||
if not out:
|
||||
errors.append("dmidecode not available or returned no memory info")
|
||||
return info
|
||||
|
||||
blocks = re.split(r"\n\s*\n", out)
|
||||
current: dict[str, str] = {}
|
||||
|
||||
for block in blocks:
|
||||
if "Memory Device" not in block:
|
||||
continue
|
||||
current.clear()
|
||||
for line in block.splitlines():
|
||||
m = re.match(r"\s*(\w[\w\s]*?):\s+(.*)", line)
|
||||
if m:
|
||||
key = m.group(1).strip()
|
||||
val = m.group(2).strip()
|
||||
current[key] = val
|
||||
|
||||
size_str = current.get("Size", "No Module Installed")
|
||||
if not size_str or "No Module Installed" in size_str or "Not Installed" in size_str:
|
||||
continue
|
||||
|
||||
sm = re.match(r"(\d+)\s*(MB|GB)", size_str)
|
||||
if not sm:
|
||||
continue
|
||||
num = int(sm.group(1))
|
||||
unit = sm.group(2)
|
||||
size_mb = float(num) * 1024 if unit == "GB" else float(num)
|
||||
|
||||
module = RAMModule(
|
||||
size_gb=round(size_mb / 1024, 1),
|
||||
type=current.get("Type", ""),
|
||||
speed_mhz=_parse_int(current.get("Speed")),
|
||||
slot=current.get("Locator", ""),
|
||||
manufacturer=current.get("Manufacturer", ""),
|
||||
part_number=current.get("Part Number", ""),
|
||||
serial=current.get("Serial Number", ""),
|
||||
voltage_v=_parse_float(current.get("Configured Voltage", current.get("Voltage", ""))),
|
||||
)
|
||||
info.modules.append(module)
|
||||
info.total_gb += module.size_gb
|
||||
|
||||
info.slots_used = len(info.modules)
|
||||
|
||||
slots_out = _run(["dmidecode", "--type", "16"], timeout=5.0)
|
||||
if slots_out:
|
||||
for line in slots_out.splitlines():
|
||||
m = re.match(r"\s*Number\s+Of\s+Devices\s*:\s*(\d+)", line)
|
||||
if m:
|
||||
info.slots_total = int(m.group(1))
|
||||
else:
|
||||
info.slots_total = info.slots_used
|
||||
|
||||
info.total_gb = round(info.total_gb, 1)
|
||||
return info
|
||||
|
||||
|
||||
def get_gpu_info(errors: list[str]) -> list[GPUInfo]:
|
||||
"""Collect GPU information from lspci and lshw."""
|
||||
gpus: list[GPUInfo] = []
|
||||
|
||||
out = _run(["lspci", "-mm"])
|
||||
if not out:
|
||||
errors.append("lspci not available — cannot enumerate PCI devices")
|
||||
return gpus
|
||||
|
||||
gpu_lines = [line for line in out.splitlines() if "VGA" in line or "3D" in line or "Display" in line]
|
||||
|
||||
for line in gpu_lines:
|
||||
parts = line.strip().split('"')
|
||||
if len(parts) < 3:
|
||||
continue
|
||||
raw_desc = parts[1] if len(parts) > 1 else ""
|
||||
vendor = parts[3] if len(parts) > 3 else ""
|
||||
model_raw = parts[5] if len(parts) > 5 else ""
|
||||
pci_slot = parts[0].strip().split("pci@")[-1].strip() if "pci@" in parts[0] else parts[0].strip()
|
||||
|
||||
gpu = GPUInfo(model=model_raw.strip(), vendor=vendor.strip(), pci_slot=pci_slot)
|
||||
gpus.append(gpu)
|
||||
|
||||
lshw_out = _run(["lshw", "-c", "display"], timeout=10.0)
|
||||
for gpu in gpus:
|
||||
if lshw_out:
|
||||
blocks = re.split(r"\*\-\w+", lshw_out)
|
||||
for block in blocks:
|
||||
if gpu.pci_slot and gpu.pci_slot in block:
|
||||
m = re.search(r"size:\s*(\d+)([KMGT]i?B?)", block, re.IGNORECASE)
|
||||
if m:
|
||||
size_val = int(m.group(1))
|
||||
unit = m.group(2).upper()
|
||||
unit_map = {"KB": 1, "KIB": 1, "MB": 1024, "MIB": 1024, "GB": 1024**2, "GIB": 1024**2, "TB": 1024**3, "TIB": 1024**3}
|
||||
vram_kb = size_val * unit_map.get(unit, 1)
|
||||
if vram_kb >= 1024:
|
||||
gpu.vram_mb = vram_kb // 1024
|
||||
break
|
||||
|
||||
if gpu.pci_slot:
|
||||
driver_path = Path(f"/sys/bus/pci/devices/{gpu.pci_slot}/driver")
|
||||
if driver_path.exists():
|
||||
try:
|
||||
gpu.driver = driver_path.resolve().name
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
return gpus
|
||||
|
||||
|
||||
def get_disk_info(errors: list[str]) -> list[DiskInfo]:
|
||||
"""Collect disk inventory and SMART health for all physical drives."""
|
||||
disks: list[DiskInfo] = []
|
||||
|
||||
devices = _find_sys_block()
|
||||
if not devices:
|
||||
errors.append("no physical block devices found in /sys/block")
|
||||
return disks
|
||||
|
||||
for dev in devices:
|
||||
device_name = dev.name
|
||||
dev_path = f"/dev/{device_name}"
|
||||
|
||||
device_dir = dev / "device"
|
||||
if not device_dir.exists():
|
||||
continue
|
||||
|
||||
disk = DiskInfo(device=dev_path)
|
||||
|
||||
size_bytes = _parse_int((dev / "size").read_text().strip()) if (dev / "size").exists() else None
|
||||
if size_bytes:
|
||||
disk.smart.size_gb = round(size_bytes * 512 / (1024**3), 1)
|
||||
|
||||
model_file = dev / "device" / "model"
|
||||
if model_file.exists():
|
||||
disk.smart.model = model_file.read_text().strip()
|
||||
|
||||
serial_file = dev / "device" / "serial"
|
||||
if serial_file.exists():
|
||||
disk.smart.serial = serial_file.read_text().strip()
|
||||
|
||||
removable = (dev / "removable").read_text().strip() if (dev / "removable").exists() else "0"
|
||||
rotational = (dev / "queue" / "rotational").read_text().strip() if (dev / "queue" / "rotational").exists() else "1"
|
||||
if rotational == "0":
|
||||
disk.smart.interface = "NVMe" if "nvme" in device_name else "SSD"
|
||||
else:
|
||||
disk.smart.interface = "HDD"
|
||||
if removable == "1":
|
||||
disk.smart.interface += "/REMOVABLE"
|
||||
|
||||
smart_out = _run(["smartctl", "-a", dev_path], timeout=15.0)
|
||||
if not smart_out:
|
||||
errors.append(f"smartctl failed or unavailable for {dev_path}")
|
||||
else:
|
||||
if "PASSED" in smart_out:
|
||||
disk.smart.status = "PASSED"
|
||||
elif "FAILED" in smart_out:
|
||||
disk.smart.status = "FAILED"
|
||||
|
||||
m = re.search(r"Temperature_Celsius\s+\d+\s+(\d+)", smart_out)
|
||||
if m:
|
||||
disk.smart.temperature_c = float(m.group(1))
|
||||
else:
|
||||
m = re.search(r"Temperature:\s+(\d+)\s+Celsius", smart_out)
|
||||
if m:
|
||||
disk.smart.temperature_c = float(m.group(1))
|
||||
|
||||
m = re.search(r"Reallocated_Sector_Ct\s+0x[\da-f]+\s+\d+\s+\d+\s+(\d+)", smart_out)
|
||||
if m:
|
||||
disk.smart.reallocated_sectors = int(m.group(1))
|
||||
|
||||
m = re.search(r"Current_Pending_Sector\s+0x[\da-f]+\s+\d+\s+\d+\s+(\d+)", smart_out)
|
||||
if m:
|
||||
disk.smart.pending_sectors = int(m.group(1))
|
||||
|
||||
m = re.search(r"Power_On_Hours\s+0x[\da-f]+\s+\d+\s+\d+\s+(\d+)", smart_out)
|
||||
if m:
|
||||
disk.smart.power_on_hours = int(m.group(1))
|
||||
else:
|
||||
m = re.search(r"Power\s*On\s*Hours\s*:\s*(\d+)", smart_out)
|
||||
if m:
|
||||
disk.smart.power_on_hours = int(m.group(1))
|
||||
|
||||
m = re.search(r"Power_Cycle_Count\s+0x[\da-f]+\s+\d+\s+\d+\s+(\d+)", smart_out)
|
||||
if m:
|
||||
disk.smart.power_cycles = int(m.group(1))
|
||||
|
||||
m = re.search(r"Raw_Read_Error_Rate\s+0x[\da-f]+\s+\d+\s+\d+\s+(\d+)", smart_out)
|
||||
if m:
|
||||
disk.smart.raw_read_error_rate = int(m.group(1))
|
||||
|
||||
m = re.search(r"Percentage\s*Used\s*:\s*(\d+)", smart_out)
|
||||
if m:
|
||||
disk.smart.wear_level = 100 - int(m.group(1))
|
||||
|
||||
disks.append(disk)
|
||||
|
||||
return disks
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main inventory class
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class HardwareInventory:
|
||||
"""Probe and report on all hardware subsystems of the local machine.
|
||||
|
||||
Call ``run()`` once to collect everything; subsequent calls return the
|
||||
cached report. Use ``text_report()`` for a human-readable summary or
|
||||
``to_json()`` for the full JSON output.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
log_warnings : bool
|
||||
If True, emit a Python log warning on each non-fatal probing
|
||||
issue (missing tools, unsupported hardware, etc.).
|
||||
"""
|
||||
|
||||
def __init__(self, log_warnings: bool = False) -> None:
|
||||
self._report: Optional[HardwareReport] = None
|
||||
self._log_warnings = log_warnings
|
||||
|
||||
def run(self) -> dict[str, Any]:
|
||||
"""Run all probes and return a JSON-serialisable dict."""
|
||||
report = self._probe()
|
||||
self._report = report
|
||||
return _asdict(report)
|
||||
|
||||
def text_report(self) -> str:
|
||||
"""Return a human-readable summary of the last ``run()`` result.
|
||||
|
||||
Raises RuntimeError if ``run()`` has not been called yet.
|
||||
"""
|
||||
if self._report is None:
|
||||
return self._fallback_summary()
|
||||
r = self._report
|
||||
lines: list[str] = []
|
||||
|
||||
lines.append("=" * 60)
|
||||
lines.append(" HARDWARE INVENTORY REPORT")
|
||||
lines.append("=" * 60)
|
||||
|
||||
lines.append("")
|
||||
lines.append("-- CPU --")
|
||||
lines.append(f" Model: {r.cpu.model or 'N/A'}")
|
||||
lines.append(f" Arch: {r.cpu.architecture or 'N/A'}")
|
||||
lines.append(f" Cores: {r.cpu.cores or '?'}")
|
||||
lines.append(f" Threads: {r.cpu.threads or '?'}")
|
||||
if r.cpu.max_speed_mhz:
|
||||
lines.append(f" Max Speed: {r.cpu.max_speed_mhz:.0f} MHz")
|
||||
if r.cpu.cache_l3_kb:
|
||||
lines.append(f" L3 Cache: {r.cpu.cache_l3_kb / 1024:.0f} MB")
|
||||
|
||||
lines.append("")
|
||||
lines.append("-- RAM --")
|
||||
lines.append(f" Total: {r.ram.total_gb:.1f} GB ({r.ram.slots_used}/{r.ram.slots_total} slots used)")
|
||||
for mod in r.ram.modules:
|
||||
speed = f" @ {mod.speed_mhz} MHz" if mod.speed_mhz else ""
|
||||
lines.append(f" + {mod.slot or '?'}: {mod.size_gb} GB {mod.type}{speed}")
|
||||
|
||||
if r.gpus:
|
||||
lines.append("")
|
||||
lines.append("-- GPU --")
|
||||
for gpu in r.gpus:
|
||||
vram = f" ({gpu.vram_mb} MB)" if gpu.vram_mb else ""
|
||||
drv = f" | driver: {gpu.driver}" if gpu.driver else ""
|
||||
lines.append(f" {gpu.model or 'Unknown GPU'}{vram}{drv}")
|
||||
|
||||
if r.disks:
|
||||
lines.append("")
|
||||
lines.append("-- DISKS --")
|
||||
for disk in r.disks:
|
||||
s = disk.smart
|
||||
status_str = s.status if s.status == "PASSED" else f"!! {s.status}"
|
||||
temp = f", {s.temperature_c} C" if s.temperature_c else ""
|
||||
poh = f", {s.power_on_hours}h" if s.power_on_hours else ""
|
||||
realloc = ""
|
||||
if s.reallocated_sectors is not None and s.reallocated_sectors > 0:
|
||||
realloc = f", !! {s.reallocated_sectors} reallocated"
|
||||
lines.append(f" {disk.device} ({s.size_gb:.0f} GB) {s.model or '?'}")
|
||||
lines.append(f" SMART: {status_str}{temp}{poh}{realloc}")
|
||||
|
||||
if r.errors:
|
||||
lines.append("")
|
||||
lines.append("-- WARNINGS --")
|
||||
for err in r.errors:
|
||||
lines.append(f" !! {err}")
|
||||
|
||||
lines.append("")
|
||||
lines.append("=" * 60)
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
def to_json(self, indent: int = 2) -> str:
|
||||
"""Return the full inventory as a JSON string."""
|
||||
return json.dumps(self.run(), indent=indent, default=str)
|
||||
|
||||
def _probe(self) -> HardwareReport:
|
||||
report = HardwareReport()
|
||||
err = report.errors
|
||||
|
||||
report.cpu = get_cpu_info(err)
|
||||
report.ram = get_ram_info(err)
|
||||
report.gpus = get_gpu_info(err)
|
||||
report.disks = get_disk_info(err)
|
||||
|
||||
if self._log_warnings and err:
|
||||
for e in err:
|
||||
logger.warning("hardware probe: %s", e)
|
||||
|
||||
return report
|
||||
|
||||
def _fallback_summary(self) -> str:
|
||||
lines = [
|
||||
"HardwareInventory -- no probe results yet.",
|
||||
"Call .run() first to collect data.",
|
||||
"",
|
||||
]
|
||||
try:
|
||||
lines.append(f"Python: {sys.version.split()[0]}")
|
||||
lines.append(f"Platform: {sys.platform}")
|
||||
uname = os.uname()
|
||||
lines.append(f"Kernel: {uname.sysname} {uname.release} {uname.machine}")
|
||||
except Exception:
|
||||
pass
|
||||
return "\n".join(lines)
|
||||
@@ -0,0 +1,995 @@
|
||||
"""
|
||||
stress.py — RAM, CPU, and GPU stress testing orchestration.
|
||||
|
||||
Designed for the Hermes Portable Rescue USB environment.
|
||||
Each tester class follows the same interface:
|
||||
- run() -> dict: structured results for LLM consumption
|
||||
- report() -> str: human-readable summary
|
||||
|
||||
Usage
|
||||
-----
|
||||
from src.diagnostics.stress import run_all_stress_tests
|
||||
results = run_all_stress_tests(duration=60)
|
||||
print(results)
|
||||
|
||||
Or from CLI:
|
||||
python -m src.diagnostics.stress --duration 60 --ram --cpu --gpu
|
||||
|
||||
Requires bundled tools:
|
||||
- stress-ng (CPU + RAM stress)
|
||||
- memtester (RAM stress fallback)
|
||||
- nvidia-smi, glxinfo, glmark2 (GPU)
|
||||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from dataclasses import dataclass, field, asdict
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
logger = logging.getLogger("stress")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Data structures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass
|
||||
class StressResult:
|
||||
"""Structured result of a single stress test run."""
|
||||
|
||||
component: str # "ram", "cpu", "gpu"
|
||||
backend: str # tool used ("stress-ng", "memtester", "memtest86+", etc.)
|
||||
status: str # "passed", "failed", "error", "skipped"
|
||||
duration_seconds: float
|
||||
summary: str # one-line human verdict
|
||||
details: Dict[str, Any] = field(default_factory=dict)
|
||||
errors: List[str] = field(default_factory=list)
|
||||
raw_output: str = ""
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return asdict(self)
|
||||
|
||||
@staticmethod
|
||||
def skipped(component: str, reason: str) -> "StressResult":
|
||||
return StressResult(
|
||||
component=component,
|
||||
backend="none",
|
||||
status="skipped",
|
||||
duration_seconds=0.0,
|
||||
summary=f"Skipped: {reason}",
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _find_tool(name: str) -> Optional[str]:
|
||||
"""Return absolute path to *name* if it exists on PATH, else None."""
|
||||
return shutil.which(name)
|
||||
|
||||
|
||||
def _run(
|
||||
cmd: List[str],
|
||||
timeout: int = 300,
|
||||
env: Optional[Dict[str, str]] = None,
|
||||
) -> Tuple[int, str, str]:
|
||||
"""Run *cmd* and return (returncode, stdout, stderr)."""
|
||||
base_env = os.environ.copy()
|
||||
if env:
|
||||
base_env.update(env)
|
||||
try:
|
||||
proc = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=timeout,
|
||||
env=base_env,
|
||||
)
|
||||
return proc.returncode, proc.stdout, proc.stderr
|
||||
except FileNotFoundError:
|
||||
return -1, "", f"Command not found: {cmd[0]}"
|
||||
except subprocess.TimeoutExpired:
|
||||
return -2, "", f"Command timed out after {timeout}s"
|
||||
except Exception as exc:
|
||||
return -3, "", str(exc)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# RAM Stress Tester
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class RAMStressTester:
|
||||
"""
|
||||
RAM stress testing with multiple backends.
|
||||
|
||||
Backends (tried in order):
|
||||
1. stress-ng --vm <n> --vm-bytes <bytes> (user-space RAM stress)
|
||||
2. memtester <bytes> <iterations> (user-space RAM test)
|
||||
3. memtest86+ result parser (reads existing report)
|
||||
"""
|
||||
|
||||
BACKEND_PRIORITY = ["stress-ng", "memtester", "memtest86+result"]
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
duration: int = 60,
|
||||
workers: Optional[int] = None,
|
||||
vm_bytes: str = "80%",
|
||||
memtest_iterations: int = 2,
|
||||
memtest86_report_path: Optional[str] = None,
|
||||
):
|
||||
self.duration = duration
|
||||
self.workers = workers or self._default_workers()
|
||||
self.vm_bytes = vm_bytes
|
||||
self.memtester_iterations = memtest_iterations
|
||||
self.memtest86_report = memtest86_report_path
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public interface
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def run(self) -> StressResult:
|
||||
backend, fn = self._pick_backend()
|
||||
if fn is None:
|
||||
return StressResult.skipped("ram", f"no backend available ({backend})")
|
||||
logger.info("RAM testing with backend=%s workers=%d duration=%ds", backend, self.workers, self.duration)
|
||||
t0 = time.monotonic()
|
||||
result = fn()
|
||||
result.duration_seconds = round(time.monotonic() - t0, 1)
|
||||
result.backend = backend
|
||||
return result
|
||||
|
||||
def report(self, result: StressResult) -> str:
|
||||
lines = [
|
||||
f"--- RAM Stress Test ---",
|
||||
f" Backend: {result.backend}",
|
||||
f" Status: {result.status}",
|
||||
f" Duration: {result.duration_seconds}s",
|
||||
f" Summary: {result.summary}",
|
||||
]
|
||||
if result.details:
|
||||
for k, v in result.details.items():
|
||||
lines.append(f" {k}: {v}")
|
||||
if result.errors:
|
||||
lines.append(" Errors:")
|
||||
for e in result.errors:
|
||||
lines.append(f" - {e}")
|
||||
return "\n".join(lines)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Internal
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@staticmethod
|
||||
def _default_workers() -> int:
|
||||
"""Heuristic: use half of available CPUs for VM stress workers."""
|
||||
try:
|
||||
return max(1, os.cpu_count() or 2 // 2)
|
||||
except Exception:
|
||||
return 1
|
||||
|
||||
def _pick_backend(self) -> Tuple[str, Optional[callable]]:
|
||||
"""Return (label, callable) for the best available backend."""
|
||||
if _find_tool("stress-ng"):
|
||||
return "stress-ng", self._run_stress_ng
|
||||
if _find_tool("memtester"):
|
||||
return "memtester", self._run_memtester
|
||||
if self.memtest86_report or _find_tool("memtest86+report"):
|
||||
return "memtest86+result", self._parse_memtest86
|
||||
return "none", None
|
||||
|
||||
def _run_stress_ng(self) -> StressResult:
|
||||
"""Run stress-ng memory stress (--vm <workers> --vm-bytes <bytes>)."""
|
||||
cmd = [
|
||||
"stress-ng",
|
||||
"--vm", str(self.workers),
|
||||
"--vm-bytes", self.vm_bytes,
|
||||
"--timeout", f"{self.duration}s",
|
||||
"--metrics-brief",
|
||||
"--no-rand-seed",
|
||||
]
|
||||
rc, stdout, stderr = _run(cmd, timeout=self.duration + 30)
|
||||
|
||||
errors = []
|
||||
if stderr:
|
||||
errors.append(stderr.strip())
|
||||
|
||||
details = self._parse_stress_ng_metrics(stdout)
|
||||
details["command"] = " ".join(cmd)
|
||||
details["workers"] = self.workers
|
||||
|
||||
if rc == -1:
|
||||
return StressResult(
|
||||
component="ram",
|
||||
status="error",
|
||||
summary="stress-ng not found",
|
||||
details=details,
|
||||
errors=errors,
|
||||
raw_output=stdout + "\n" + stderr,
|
||||
)
|
||||
if rc == -2:
|
||||
return StressResult(
|
||||
component="ram",
|
||||
status="error",
|
||||
summary="stress-ng timed out",
|
||||
details=details,
|
||||
errors=errors,
|
||||
raw_output=stdout + "\n" + stderr,
|
||||
)
|
||||
|
||||
# stress-ng exits 0 on clean completion even if errors occurred
|
||||
status = "passed" if rc == 0 else "error"
|
||||
bogo_ops = details.get("total_bogo_ops", 0)
|
||||
summary = (
|
||||
f"Completed {bogo_ops} bogo-ops across {self.workers} VM workers"
|
||||
if bogo_ops > 0 else
|
||||
f"Finished with status={status}"
|
||||
)
|
||||
return StressResult(
|
||||
component="ram",
|
||||
status=status,
|
||||
summary=summary,
|
||||
details=details,
|
||||
errors=errors,
|
||||
raw_output=stdout + "\n" + stderr,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _parse_stress_ng_metrics(text: str) -> Dict[str, Any]:
|
||||
"""Extract bogo-ops and other metrics from stress-ng --metrics-brief."""
|
||||
metrics: Dict[str, Any] = {}
|
||||
total_bogo = 0.0
|
||||
bogo_per_sec = 0.0
|
||||
# stress-ng metrics lines look like:
|
||||
# stress-ng: info: [123] stress-ng-vm: 4 runouts, 0 stalled...
|
||||
# stress-ng: info: [123] stress-ng-metrics: vm 1234.56 bogo-ops ...
|
||||
for line in text.splitlines():
|
||||
if "bogo-ops" in line.lower():
|
||||
m = re.search(r"vm\s+([\d.]+)\s+bogo-ops", line, re.IGNORECASE)
|
||||
if m:
|
||||
total_bogo = float(m.group(1))
|
||||
m2 = re.search(r"bogo-ops/s\s+real\s+([\d.]+)", line, re.IGNORECASE)
|
||||
if m2:
|
||||
bogo_per_sec = float(m2.group(1))
|
||||
metrics["total_bogo_ops"] = round(total_bogo, 1)
|
||||
metrics["bogo_ops_per_sec"] = round(bogo_per_sec, 3) if bogo_per_sec else 0.0
|
||||
return metrics
|
||||
|
||||
def _run_memtester(self) -> StressResult:
|
||||
"""Run memtester <vm_bytes> <iterations>."""
|
||||
# Convert vm_bytes like "80%" to a usable size for memtester
|
||||
# memtester takes size in megabytes directly
|
||||
size_mb = self._vm_bytes_to_mb()
|
||||
if size_mb <= 0:
|
||||
size_mb = 256 # fallback
|
||||
cmd = ["memtester", f"{size_mb}M", str(self.memtester_iterations)]
|
||||
rc, stdout, stderr = _run(cmd, timeout=max(self.duration, 120))
|
||||
|
||||
errors = []
|
||||
if stderr:
|
||||
errors.append(stderr.strip())
|
||||
|
||||
details = self._parse_memtester_output(stdout)
|
||||
details["command"] = " ".join(cmd)
|
||||
details["size_mb"] = size_mb
|
||||
details["iterations"] = self.memtester_iterations
|
||||
|
||||
status = "passed" if rc == 0 else "failed"
|
||||
if rc < 0:
|
||||
status = "error"
|
||||
|
||||
failure_count = details.get("failures", 0)
|
||||
summary = f"memtester: {details.get('loops_completed', 0)} loops, {failure_count} failures" if status != "error" else "memtester run failed"
|
||||
return StressResult(
|
||||
component="ram",
|
||||
status=status,
|
||||
summary=summary,
|
||||
details=details,
|
||||
errors=errors,
|
||||
raw_output=stdout + "\n" + stderr,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _parse_memtester_output(text: str) -> Dict[str, Any]:
|
||||
"""Parse memtester output for pass/fail information."""
|
||||
details: Dict[str, Any] = {}
|
||||
loops = 0
|
||||
failures = 0
|
||||
for line in text.splitlines():
|
||||
m = re.match(r"Loop\s+(\d+)", line, re.IGNORECASE)
|
||||
if m:
|
||||
loops = max(loops, int(m.group(1)))
|
||||
if "FAILURE" in line.upper():
|
||||
failures += 1
|
||||
details["loops_completed"] = loops
|
||||
details["failures"] = failures
|
||||
return details
|
||||
|
||||
def _parse_memtest86(self) -> StressResult:
|
||||
"""Parse existing MemTest86+ report (result XML or text file)."""
|
||||
report_path = self.memtest86_report
|
||||
if report_path is None:
|
||||
# Common locations
|
||||
for candidate in [
|
||||
"/MemTest86-Report.xml",
|
||||
"/boot/MemTest86-Report.xml",
|
||||
"/memtest86/report.xml",
|
||||
"~/MemTest86-Report.xml",
|
||||
]:
|
||||
expanded = os.path.expanduser(candidate)
|
||||
if os.path.isfile(expanded):
|
||||
report_path = expanded
|
||||
break
|
||||
|
||||
if not report_path or not os.path.isfile(report_path):
|
||||
return StressResult(
|
||||
component="ram",
|
||||
status="skipped",
|
||||
summary="MemTest86+ report not found; run MemTest86+ on next boot",
|
||||
details={"suggested_path": "/MemTest86-Report.xml"},
|
||||
)
|
||||
|
||||
try:
|
||||
text = Path(report_path).read_text()
|
||||
except Exception as exc:
|
||||
return StressResult(
|
||||
component="ram",
|
||||
status="error",
|
||||
summary=f"Cannot read report: {exc}",
|
||||
)
|
||||
|
||||
details = self._parse_memtest86_xml(text)
|
||||
details["report_path"] = report_path
|
||||
summary = (
|
||||
f"MemTest86+ report: {details.get('passes', '?')} passes, "
|
||||
f"{details.get('errors', 0)} errors"
|
||||
)
|
||||
status = "passed" if details.get("errors", 0) == 0 else "failed"
|
||||
return StressResult(
|
||||
component="ram",
|
||||
status=status,
|
||||
summary=summary,
|
||||
details=details,
|
||||
raw_output=text,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _parse_memtest86_xml(text: str) -> Dict[str, Any]:
|
||||
"""Naïve extraction of key fields from MemTest86+ XML report."""
|
||||
def _tag(tag: str) -> str:
|
||||
m = re.search(rf"<{tag}[^>]*>(.*?)</{tag}>", text, re.DOTALL)
|
||||
return m.group(1).strip() if m else ""
|
||||
|
||||
passes_str = _tag("Passes")
|
||||
return {
|
||||
"passes": int(passes_str) if passes_str.isdigit() else 0,
|
||||
"errors": _tag("Errors"),
|
||||
"test_start_time": _tag("TestStartTime"),
|
||||
"test_end_time": _tag("TestEndTime"),
|
||||
"memory_size": _tag("MemorySize"),
|
||||
"cpu_type": _tag("CpuType"),
|
||||
"cpu_speed": _tag("CpuSpeed"),
|
||||
}
|
||||
|
||||
def _vm_bytes_to_mb(self) -> int:
|
||||
"""Convert a vm_bytes string (e.g. '80%' or '512M') to megabytes."""
|
||||
s = str(self.vm_bytes).strip().lower()
|
||||
if s.endswith("%"):
|
||||
try:
|
||||
pct = int(s.rstrip("%"))
|
||||
# Estimate available RAM via /proc/meminfo
|
||||
mem_total_kb = 0
|
||||
try:
|
||||
for line in Path("/proc/meminfo").read_text().splitlines():
|
||||
if line.startswith("MemTotal:"):
|
||||
parts = line.split()
|
||||
if len(parts) >= 2:
|
||||
mem_total_kb = int(parts[1])
|
||||
break
|
||||
except Exception:
|
||||
mem_total_kb = 8 * 1024 * 1024 # fallback: 8 GB
|
||||
mem_total_mb = mem_total_kb // 1024
|
||||
return max(64, mem_total_mb * pct // 100)
|
||||
except (ValueError, TypeError):
|
||||
return 512
|
||||
if s.endswith("g"):
|
||||
try:
|
||||
return int(s.rstrip("g")) * 1024
|
||||
except ValueError:
|
||||
return 512
|
||||
if s.endswith("m"):
|
||||
try:
|
||||
return int(s.rstrip("m"))
|
||||
except ValueError:
|
||||
return 512
|
||||
try:
|
||||
return int(s)
|
||||
except (ValueError, TypeError):
|
||||
return 512
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CPU Stress Tester
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class CPUStressTester:
|
||||
"""
|
||||
CPU stress testing via stress-ng.
|
||||
|
||||
Tests several CPU subsystems in sequence or parallel:
|
||||
- cpu (integer arithmetic)
|
||||
- matrix (floating-point matrix operations)
|
||||
- fpu (FPU stress)
|
||||
- cache (L1/L2 cache thrashing)
|
||||
- context (context switching)
|
||||
|
||||
Each stressor is run briefly to exercise different CPU aspects.
|
||||
"""
|
||||
|
||||
DEFAULT_STRESSORS = ["cpu", "matrix", "fpu", "cache"]
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
duration: int = 60,
|
||||
workers: Optional[int] = None,
|
||||
stressors: Optional[List[str]] = None,
|
||||
sequential: bool = True,
|
||||
):
|
||||
self.duration = duration
|
||||
self.workers = workers or self._default_workers()
|
||||
self.stressors = stressors or list(self.DEFAULT_STRESSORS)
|
||||
self.sequential = sequential
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public interface
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def run(self) -> StressResult:
|
||||
backend = self._pick_backend()
|
||||
if backend is None:
|
||||
return StressResult.skipped("cpu", "stress-ng not found on PATH")
|
||||
|
||||
t0 = time.monotonic()
|
||||
if self.sequential:
|
||||
result = self._run_sequential(backend)
|
||||
else:
|
||||
result = self._run_parallel(backend)
|
||||
result.duration_seconds = round(time.monotonic() - t0, 1)
|
||||
result.backend = backend
|
||||
return result
|
||||
|
||||
def report(self, result: StressResult) -> str:
|
||||
lines = [
|
||||
f"--- CPU Stress Test ---",
|
||||
f" Backend: {result.backend}",
|
||||
f" Status: {result.status}",
|
||||
f" Duration: {result.duration_seconds}s",
|
||||
f" Summary: {result.summary}",
|
||||
]
|
||||
details = result.details
|
||||
if "stressors_run" in details:
|
||||
lines.append(f" Stressors run: {', '.join(details['stressors_run'])}")
|
||||
if "stressor_results" in details:
|
||||
for sres in details["stressor_results"]:
|
||||
lines.append(f" - {sres.get('name', '?')}: {sres.get('bogo_ops', 0)} bogo-ops")
|
||||
if result.errors:
|
||||
lines.append(" Errors:")
|
||||
for e in result.errors:
|
||||
lines.append(f" - {e}")
|
||||
return "\n".join(lines)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Internal
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@staticmethod
|
||||
def _default_workers() -> int:
|
||||
try:
|
||||
return os.cpu_count() or 2
|
||||
except Exception:
|
||||
return 2
|
||||
|
||||
@staticmethod
|
||||
def _pick_backend() -> Optional[str]:
|
||||
path = _find_tool("stress-ng")
|
||||
return path
|
||||
|
||||
def _run_sequential(self, backend: str) -> StressResult:
|
||||
"""Run each stressor one at a time so we get per-stressor metrics."""
|
||||
per_stressor_duration = max(10, self.duration // len(self.stressors))
|
||||
all_stressor_results: List[Dict[str, Any]] = []
|
||||
all_errors: List[str] = []
|
||||
total_bogo = 0.0
|
||||
|
||||
for stressor in self.stressors:
|
||||
cmd = [
|
||||
backend,
|
||||
"--" + stressor, str(self.workers),
|
||||
"--timeout", f"{per_stressor_duration}s",
|
||||
"--metrics-brief",
|
||||
"--no-rand-seed",
|
||||
]
|
||||
logger.info("CPU stressor: %s (workers=%d, duration=%ds)", stressor, self.workers, per_stressor_duration)
|
||||
rc, stdout, stderr = _run(cmd, timeout=per_stressor_duration + 30)
|
||||
|
||||
if rc < 0:
|
||||
all_errors.append(f"{stressor}: {stderr.strip() or 'unknown error'}")
|
||||
all_stressor_results.append({"name": stressor, "status": "error", "bogo_ops": 0})
|
||||
continue
|
||||
|
||||
metrics = self._parse_single_stressor_metrics(stdout, stressor)
|
||||
metrics["name"] = stressor
|
||||
metrics["status"] = "passed" if rc == 0 else "error"
|
||||
all_stressor_results.append(metrics)
|
||||
total_bogo += metrics.get("bogo_ops", 0.0)
|
||||
all_errors.extend(self._extract_errors(stdout + "\n" + stderr))
|
||||
|
||||
details: Dict[str, Any] = {}
|
||||
details["stressors_run"] = self.stressors
|
||||
details["stressor_results"] = all_stressor_results
|
||||
details["total_bogo_ops"] = round(total_bogo, 1)
|
||||
|
||||
status = "passed" if not all_errors else "failed"
|
||||
summary = (
|
||||
f"CPU stress: {len(self.stressors)} stressors, "
|
||||
f"{details['total_bogo_ops']} total bogo-ops"
|
||||
)
|
||||
if all_errors:
|
||||
summary += f", {len(all_errors)} errors"
|
||||
|
||||
return StressResult(
|
||||
component="cpu",
|
||||
status=status,
|
||||
summary=summary,
|
||||
details=details,
|
||||
errors=all_errors,
|
||||
)
|
||||
|
||||
def _run_parallel(self, backend: str) -> StressResult:
|
||||
"""Run all stressors in one stress-ng invocation (--all)."""
|
||||
# Build command: stress-ng --cpu N --matrix N --fpu N --cache N --timeout Xs --metrics-brief
|
||||
cmd = [backend]
|
||||
for stressor in self.stressors:
|
||||
cmd.extend(["--" + stressor, str(self.workers)])
|
||||
cmd.extend(["--timeout", f"{self.duration}s", "--metrics-brief", "--no-rand-seed"])
|
||||
|
||||
logger.info("CPU parallel stress: stressors=%s workers=%d", self.stressors, self.workers)
|
||||
rc, stdout, stderr = _run(cmd, timeout=self.duration + 30)
|
||||
|
||||
all_errors: List[str] = []
|
||||
if stderr:
|
||||
all_errors.extend(self._extract_errors(stderr))
|
||||
if rc < 0:
|
||||
all_errors.append(f"stress-ng rc={rc}")
|
||||
|
||||
all_stressor_results: List[Dict[str, Any]] = []
|
||||
total_bogo = 0.0
|
||||
for stressor in self.stressors:
|
||||
metrics = self._parse_single_stressor_metrics(stdout, stressor)
|
||||
metrics["name"] = stressor
|
||||
metrics["status"] = "passed" if rc == 0 else "error"
|
||||
all_stressor_results.append(metrics)
|
||||
total_bogo += metrics.get("bogo_ops", 0.0)
|
||||
|
||||
details: Dict[str, Any] = {}
|
||||
details["stressors_run"] = self.stressors
|
||||
details["stressor_results"] = all_stressor_results
|
||||
details["total_bogo_ops"] = round(total_bogo, 1)
|
||||
details["command"] = " ".join(cmd)
|
||||
details["workers_per_stressor"] = self.workers
|
||||
|
||||
status = "passed" if rc == 0 else "error"
|
||||
summary = f"CPU parallel stress: {details['total_bogo_ops']} bogo-ops"
|
||||
if all_errors:
|
||||
summary += f", {len(all_errors)} issues"
|
||||
status = "failed"
|
||||
|
||||
return StressResult(
|
||||
component="cpu",
|
||||
status=status,
|
||||
summary=summary,
|
||||
details=details,
|
||||
errors=all_errors,
|
||||
raw_output=stdout + "\n" + stderr,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _parse_single_stressor_metrics(text: str, stressor: str) -> Dict[str, Any]:
|
||||
"""Extract metrics for a single stressor from stress-ng output."""
|
||||
bogo_ops = 0.0
|
||||
bogo_per_sec = 0.0
|
||||
for line in text.splitlines():
|
||||
if "bogo-ops" in line.lower() and stressor in line.lower():
|
||||
m = re.search(rf"{stressor}\s+([\d.]+)\s+bogo-ops", line, re.IGNORECASE)
|
||||
if m:
|
||||
bogo_ops = float(m.group(1))
|
||||
m2 = re.search(r"bogo-ops/s\s+real\s+([\d.]+)", line, re.IGNORECASE)
|
||||
if m2:
|
||||
bogo_per_sec = float(m2.group(1))
|
||||
return {
|
||||
"bogo_ops": round(bogo_ops, 1),
|
||||
"bogo_ops_per_sec": round(bogo_per_sec, 3),
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _extract_errors(text: str) -> List[str]:
|
||||
"""Extract error/warning lines from stress-ng output."""
|
||||
errors = []
|
||||
for line in text.splitlines():
|
||||
lower = line.lower()
|
||||
for kw in ("error", "warning", "fail", "cannot"):
|
||||
if kw in lower:
|
||||
errors.append(line.strip())
|
||||
break
|
||||
return errors
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GPU Stress Tester
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class GPUStressTester:
|
||||
"""
|
||||
GPU info gathering and basic stress testing.
|
||||
|
||||
Capabilities (tried in order of preference):
|
||||
1. nvidia-smi — NVIDIA GPU info + monitoring
|
||||
2. glxinfo — OpenGL vendor/renderer/version
|
||||
3. glmark2 — OpenGL benchmark / stress
|
||||
4. lspci — GPU model identification fallback
|
||||
"""
|
||||
|
||||
def __init__(self, duration: int = 60, skip_benchmark: bool = False):
|
||||
self.duration = duration
|
||||
self.skip_benchmark = skip_benchmark
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public interface
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def run(self) -> StressResult:
|
||||
t0 = time.monotonic()
|
||||
info = self._gather_gpu_info()
|
||||
stress = self._run_stress_test()
|
||||
result = self._merge(info, stress)
|
||||
result.duration_seconds = round(time.monotonic() - t0, 1)
|
||||
return result
|
||||
|
||||
def report(self, result: StressResult) -> str:
|
||||
lines = [
|
||||
f"--- GPU Stress Test ---",
|
||||
f" Backend: {result.backend}",
|
||||
f" Status: {result.status}",
|
||||
f" Duration: {result.duration_seconds}s",
|
||||
f" Summary: {result.summary}",
|
||||
]
|
||||
details = result.details
|
||||
gpu_info = details.get("gpu_info", {})
|
||||
if gpu_info:
|
||||
lines.append(" GPU Info:")
|
||||
for k, v in gpu_info.items():
|
||||
lines.append(f" {k}: {v}")
|
||||
if "benchmark_score" in details:
|
||||
lines.append(f" Benchmark score: {details['benchmark_score']}")
|
||||
if result.errors:
|
||||
lines.append(" Errors:")
|
||||
for e in result.errors:
|
||||
lines.append(f" - {e}")
|
||||
return "\n".join(lines)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Internal
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _gather_gpu_info(self) -> Dict[str, Any]:
|
||||
"""Collect GPU info from all available sources."""
|
||||
info: Dict[str, Any] = {}
|
||||
errors: List[str] = []
|
||||
|
||||
# 1) nvidia-smi
|
||||
nv_path = _find_tool("nvidia-smi")
|
||||
if nv_path:
|
||||
rc, stdout, stderr = _run([nv_path, "--query-gpu=name,driver_version,temperature.gpu,memory.total,utilization.gpu", "--format=csv,noheader"], timeout=15)
|
||||
if rc == 0:
|
||||
for line in stdout.strip().splitlines():
|
||||
parts = [p.strip() for p in line.split(",")]
|
||||
if len(parts) >= 3:
|
||||
info["model"] = parts[0]
|
||||
info["driver_version"] = parts[1]
|
||||
info["temperature_c"] = parts[2]
|
||||
if len(parts) >= 4:
|
||||
info["vram_total"] = parts[3]
|
||||
if len(parts) >= 5:
|
||||
info["utilization_pct"] = parts[4]
|
||||
else:
|
||||
errors.append(f"nvidia-smi: {stderr.strip()}")
|
||||
else:
|
||||
errors.append("nvidia-smi not found")
|
||||
|
||||
# 2) glxinfo (OpenGL vendor/renderer)
|
||||
glx_path = _find_tool("glxinfo")
|
||||
if glx_path:
|
||||
rc, stdout, stderr = _run([glx_path, "-B"], timeout=15)
|
||||
if rc == 0:
|
||||
for line in stdout.splitlines():
|
||||
lower = line.lower()
|
||||
if "opengl vendor" in lower:
|
||||
info["gl_vendor"] = line.split(":", 1)[-1].strip()
|
||||
elif "opengl renderer" in lower:
|
||||
info["gl_renderer"] = line.split(":", 1)[-1].strip()
|
||||
elif "opengl version" in lower:
|
||||
info["gl_version"] = line.split(":", 1)[-1].strip()
|
||||
|
||||
# 3) lspci as fallback for model
|
||||
if "model" not in info:
|
||||
lspci_path = _find_tool("lspci")
|
||||
if lspci_path:
|
||||
rc, stdout, stderr = _run([lspci_path], timeout=10)
|
||||
if rc == 0:
|
||||
for line in stdout.splitlines():
|
||||
l = line.lower()
|
||||
if any(kw in l for kw in ("vga", "3d controller", "display")):
|
||||
info["model"] = line.strip()
|
||||
break
|
||||
|
||||
info["has_nvidia_gpu"] = nv_path is not None
|
||||
return info
|
||||
|
||||
def _run_stress_test(self) -> Dict[str, Any]:
|
||||
"""Run GPU benchmark/stress if available."""
|
||||
result: Dict[str, Any] = {}
|
||||
errors: List[str] = []
|
||||
|
||||
if self.skip_benchmark:
|
||||
result["benchmark_skipped"] = True
|
||||
return result
|
||||
|
||||
# Try glmark2 first (OpenGL benchmark)
|
||||
glmark_path = _find_tool("glmark2")
|
||||
if glmark_path:
|
||||
rc, stdout, stderr = _run(
|
||||
[glmark2_path, "--run-forever"],
|
||||
timeout=self.duration + 15,
|
||||
env={"DISPLAY": os.environ.get("DISPLAY", ":0")},
|
||||
)
|
||||
if rc == 0:
|
||||
score = self._parse_glmark2_score(stdout)
|
||||
result["benchmark_score"] = score
|
||||
result["benchmark_tool"] = "glmark2"
|
||||
else:
|
||||
errors.append("glmark2 not found; no GPU benchmark available")
|
||||
|
||||
if errors:
|
||||
result["errors"] = errors
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
def _parse_glmark2_score(text: str) -> Optional[int]:
|
||||
"""Extract the final glmark2 score. Looks for 'glmark2 Score: NNN'."""
|
||||
m = re.search(r"glmark2\s+Score:\s+(\d+)", text, re.IGNORECASE)
|
||||
if m:
|
||||
return int(m.group(1))
|
||||
return None
|
||||
|
||||
def _merge(self, info: Dict[str, Any], stress: Dict[str, Any]) -> StressResult:
|
||||
"""Merge gathered info and stress results into a single StressResult."""
|
||||
all_errors = info.pop("errors", [])
|
||||
all_errors.extend(stress.pop("errors", []))
|
||||
|
||||
details: Dict[str, Any] = {}
|
||||
details["gpu_info"] = info
|
||||
details.update(stress)
|
||||
|
||||
model = info.get("model", info.get("gl_renderer", "unknown"))
|
||||
summary_parts = [f"GPU: {model}"]
|
||||
if "benchmark_score" in details and details["benchmark_score"] is not None:
|
||||
summary_parts.append(f"glmark2 score: {details['benchmark_score']}")
|
||||
if "temperature_c" in info:
|
||||
summary_parts.append(f"temp: {info['temperature_c']}°C")
|
||||
|
||||
status = "passed" if not all_errors else "failed"
|
||||
# Even with all_errors, we always have info from lspci at minimum
|
||||
backend_parts = []
|
||||
if _find_tool("nvidia-smi"):
|
||||
backend_parts.append("nvidia-smi")
|
||||
if _find_tool("glxinfo"):
|
||||
backend_parts.append("glxinfo")
|
||||
if _find_tool("glmark2"):
|
||||
backend_parts.append("glmark2")
|
||||
if _find_tool("lspci"):
|
||||
backend_parts.append("lspci")
|
||||
if not backend_parts:
|
||||
backend_parts = ["none"]
|
||||
|
||||
return StressResult(
|
||||
component="gpu",
|
||||
backend="+".join(backend_parts),
|
||||
status=status,
|
||||
summary="; ".join(summary_parts),
|
||||
details=details,
|
||||
errors=all_errors,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Orchestration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def run_all_stress_tests(
|
||||
duration: int = 60,
|
||||
ram: bool = True,
|
||||
cpu: bool = True,
|
||||
gpu: bool = True,
|
||||
skip_gpu_benchmark: bool = False,
|
||||
) -> Dict[str, StressResult]:
|
||||
"""
|
||||
Run all requested stress tests and return structured results.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
duration : int
|
||||
Per-test duration in seconds (CPU/RAM get this; GPU benchmark runs separately).
|
||||
ram, cpu, gpu : bool
|
||||
Toggle individual tests.
|
||||
skip_gpu_benchmark : bool
|
||||
If True, only gather GPU info, skip the OpenGL benchmark loop.
|
||||
|
||||
Returns
|
||||
-------
|
||||
dict of {component: StressResult}
|
||||
"""
|
||||
results: Dict[str, StressResult] = {}
|
||||
logger.info("Stress test suite starting (duration=%ds)", duration)
|
||||
|
||||
if ram:
|
||||
ram_tester = RAMStressTester(duration=duration)
|
||||
results["ram"] = ram_tester.run()
|
||||
logger.info("RAM test: %s | %s", results["ram"].status, results["ram"].summary)
|
||||
|
||||
if cpu:
|
||||
cpu_tester = CPUStressTester(duration=duration)
|
||||
results["cpu"] = cpu_tester.run()
|
||||
logger.info("CPU test: %s | %s", results["cpu"].status, results["cpu"].summary)
|
||||
|
||||
if gpu:
|
||||
gpu_tester = GPUStressTester(duration=duration, skip_benchmark=skip_gpu_benchmark)
|
||||
results["gpu"] = gpu_tester.run()
|
||||
logger.info("GPU test: %s | %s", results["gpu"].status, results["gpu"].summary)
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def print_report(results: Dict[str, StressResult]) -> None:
|
||||
"""Pretty-print test results to stdout."""
|
||||
for component in ("ram", "cpu", "gpu"):
|
||||
result = results.get(component)
|
||||
if result is not None:
|
||||
print()
|
||||
print(StressResult.report.__doc__) # placeholder — will be replaced
|
||||
# Actually use the testers' report methods
|
||||
for component in ("ram", "cpu", "gpu"):
|
||||
result = results.get(component)
|
||||
if result is not None:
|
||||
if component == "ram":
|
||||
print(RAMStressTester().report(result))
|
||||
elif component == "cpu":
|
||||
print(CPUStressTester().report(result))
|
||||
elif component == "gpu":
|
||||
print(GPUStressTester().report(result))
|
||||
print()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CLI
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def parse_cli_args() -> dict:
|
||||
"""Simple manual arg parser (no external deps)."""
|
||||
args = {
|
||||
"duration": 60,
|
||||
"ram": True,
|
||||
"cpu": True,
|
||||
"gpu": True,
|
||||
"skip_gpu_benchmark": False,
|
||||
"verbose": False,
|
||||
}
|
||||
|
||||
argv = sys.argv[1:]
|
||||
i = 0
|
||||
while i < len(argv):
|
||||
a = argv[i]
|
||||
if a in ("--duration", "-d"):
|
||||
if i + 1 < len(argv):
|
||||
args["duration"] = int(argv[i + 1])
|
||||
i += 1
|
||||
elif a == "--ram-only":
|
||||
args["ram"] = True
|
||||
args["cpu"] = False
|
||||
args["gpu"] = False
|
||||
elif a == "--cpu-only":
|
||||
args["ram"] = False
|
||||
args["cpu"] = True
|
||||
args["gpu"] = False
|
||||
elif a == "--gpu-only":
|
||||
args["ram"] = False
|
||||
args["cpu"] = False
|
||||
args["gpu"] = True
|
||||
elif a in ("--no-ram",):
|
||||
args["ram"] = False
|
||||
elif a in ("--no-cpu",):
|
||||
args["cpu"] = False
|
||||
elif a in ("--no-gpu",):
|
||||
args["gpu"] = False
|
||||
elif a in ("--skip-gpu-benchmark",):
|
||||
args["skip_gpu_benchmark"] = True
|
||||
elif a in ("--verbose", "-v"):
|
||||
args["verbose"] = True
|
||||
elif a in ("--help", "-h"):
|
||||
print("Usage: python -m src.diagnostics.stress [options]")
|
||||
print()
|
||||
print("Options:")
|
||||
print(" -d, --duration SECS Per-test duration (default: 60)")
|
||||
print(" --ram-only RAM test only")
|
||||
print(" --cpu-only CPU test only")
|
||||
print(" --gpu-only GPU test only")
|
||||
print(" --no-ram Skip RAM test")
|
||||
print(" --no-cpu Skip CPU test")
|
||||
print(" --no-gpu Skip GPU test")
|
||||
print(" --skip-gpu-benchmark GPU info only, no benchmark")
|
||||
print(" -v, --verbose Verbose logging")
|
||||
print(" -h, --help This help")
|
||||
sys.exit(0)
|
||||
i += 1
|
||||
return args
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""CLI entry point."""
|
||||
config = parse_cli_args()
|
||||
if config["verbose"]:
|
||||
logging.basicConfig(level=logging.INFO, format="%(levelname)s | %(message)s")
|
||||
else:
|
||||
logging.basicConfig(level=logging.WARNING, format="%(levelname)s | %(message)s")
|
||||
|
||||
print("Hermes Portable Rescue — Stress Test Suite")
|
||||
print(f" Duration: {config['duration']}s per test")
|
||||
print(f" RAM: {'yes' if config['ram'] else 'no'}")
|
||||
print(f" CPU: {'yes' if config['cpu'] else 'no'}")
|
||||
print(f" GPU: {'yes' if config['gpu'] else 'no'}")
|
||||
print(f" GPU benchmark: {'no' if config['skip_gpu_benchmark'] else 'yes'}")
|
||||
print()
|
||||
|
||||
results = run_all_stress_tests(
|
||||
duration=config["duration"],
|
||||
ram=config["ram"],
|
||||
cpu=config["cpu"],
|
||||
gpu=config["gpu"],
|
||||
skip_gpu_benchmark=config["skip_gpu_benchmark"],
|
||||
)
|
||||
|
||||
print_report(results)
|
||||
|
||||
# Overall status
|
||||
print("=" * 50)
|
||||
print("OVERALL")
|
||||
for comp, res in results.items():
|
||||
print(f" {comp.upper():5s} {res.status:10s} {res.summary}")
|
||||
print("=" * 50)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user