1824 lines
65 KiB
Python
1824 lines
65 KiB
Python
#!/usr/bin/env python3
|
|
# Copyright (c) 2026 Petr Balvín <opensource@petrbalvin.org> (https://petrbalvin.org)
|
|
# SPDX-License-Identifier: MIT
|
|
|
|
"""System diagnostics — CPU, memory, disk, network, GPU, services, security, performance.
|
|
|
|
No external dependencies — leverages /proc, /sys, and system tools with
|
|
graceful fallbacks when any data source is unavailable.
|
|
|
|
Usage:
|
|
system-diag.py # full report
|
|
system-diag.py --section cpu,memory # only selected sections
|
|
system-diag.py --external-ip # also detect external IPs (3rd-party)
|
|
system-diag.py --json # machine-readable output
|
|
system-diag.py --version
|
|
|
|
Linux only. Progress is written to stderr, the report to stdout.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from collections.abc import Callable
|
|
from concurrent.futures import Future, ThreadPoolExecutor, as_completed
|
|
from typing import Any
|
|
|
|
BOLD = "\033[1m"
|
|
RED = "\033[31m"
|
|
GREEN = "\033[32m"
|
|
YELLOW = "\033[33m"
|
|
CYAN = "\033[36m"
|
|
MAGENTA = "\033[35m"
|
|
DIM = "\033[2m"
|
|
RESET = "\033[0m"
|
|
|
|
|
|
def _supports_colour() -> bool:
|
|
"""Honour NO_COLOR (https://no-color.org) and non-TTY output."""
|
|
if os.environ.get("NO_COLOR") is not None:
|
|
return False
|
|
return sys.stdout.isatty()
|
|
|
|
|
|
if not _supports_colour():
|
|
BOLD = RED = GREEN = YELLOW = CYAN = MAGENTA = DIM = RESET = ""
|
|
|
|
# Letter grade → colour, applied at render time only (never stored in data).
|
|
GRADE_COLOURS = {"A": GREEN, "B": GREEN, "C": YELLOW, "D": YELLOW, "F": RED}
|
|
|
|
VERSION = "1.0.0"
|
|
|
|
WIDTH = 60
|
|
|
|
ALL_SECTIONS = [
|
|
"overview",
|
|
"cpu",
|
|
"memory",
|
|
"disk",
|
|
"network",
|
|
"gpu",
|
|
"services",
|
|
"security",
|
|
"performance",
|
|
"issues",
|
|
]
|
|
|
|
|
|
def _status(msg: str) -> None:
|
|
"""Print progress message to stderr — stays on same line until ``_status_done``."""
|
|
print(f" {msg}...", end="", flush=True, file=sys.stderr)
|
|
|
|
|
|
def _status_done(msg: str = "done") -> None:
|
|
"""Complete the current progress line on stderr."""
|
|
print(f" {msg}", file=sys.stderr)
|
|
|
|
|
|
def run(cmd: list[str], timeout: int = 15) -> tuple[str, str]:
|
|
"""Execute a shell command and return (stdout, stderr) separately.
|
|
|
|
Sets LC_ALL=C (highest locale precedence, overrides user settings) for
|
|
reliable English-language output parsing.
|
|
Returns empty strings on timeout or if the command cannot be run.
|
|
"""
|
|
try:
|
|
env = {**os.environ, "LANG": "C", "LC_ALL": "C"}
|
|
result = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
env=env,
|
|
errors="replace",
|
|
)
|
|
return result.stdout, result.stderr
|
|
except (subprocess.TimeoutExpired, OSError):
|
|
return "", ""
|
|
|
|
|
|
def _read_file(path: str) -> str:
|
|
"""Read a file, returning '' if it doesn't exist or can't be read."""
|
|
try:
|
|
with open(path, encoding="utf-8", errors="replace") as f:
|
|
return f.read().strip()
|
|
except OSError:
|
|
return ""
|
|
|
|
|
|
def _read_lines(path: str) -> list[str]:
|
|
"""Read lines from a file, returning [] if unavailable."""
|
|
try:
|
|
with open(path, encoding="utf-8", errors="replace") as f:
|
|
return [line.strip() for line in f]
|
|
except OSError:
|
|
return []
|
|
|
|
|
|
def _bar(value: float, maximum: float = 100.0, width: int = 20) -> str:
|
|
"""Build a textual usage bar with colour.
|
|
|
|
Returns a string like '████████░░░░░░░░░░░░ 42%' with ANSI colour.
|
|
"""
|
|
ratio = max(0.0, min(1.0, value / maximum)) if maximum > 0 else 0.0
|
|
filled = int(ratio * width)
|
|
if ratio < 0.80:
|
|
colour = GREEN
|
|
elif ratio < 0.90:
|
|
colour = YELLOW
|
|
else:
|
|
colour = RED
|
|
bar_str = colour + "█" * filled + DIM + "░" * (width - filled) + RESET
|
|
return f"{bar_str} {ratio * 100:.0f}%"
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Section: System Overview
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _parse_os_release() -> dict[str, str]:
|
|
"""Parse /etc/os-release into a dict."""
|
|
info: dict[str, str] = {}
|
|
for line in _read_lines("/etc/os-release"):
|
|
if "=" in line:
|
|
key, _, val = line.partition("=")
|
|
info[key.strip()] = val.strip().strip('"')
|
|
return info
|
|
|
|
|
|
def _uptime_human(seconds: float) -> str:
|
|
"""Return human-readable uptime string, e.g. '3 days, 7 hours'."""
|
|
days, rem = divmod(int(seconds), 86400)
|
|
hours, rem = divmod(rem, 3600)
|
|
minutes = rem // 60
|
|
|
|
parts = []
|
|
if days:
|
|
parts.append(f"{days} day{'s' if days != 1 else ''}")
|
|
if hours:
|
|
parts.append(f"{hours} hour{'s' if hours != 1 else ''}")
|
|
if minutes or not parts:
|
|
parts.append(f"{minutes} minute{'s' if minutes != 1 else ''}")
|
|
return ", ".join(parts)
|
|
|
|
|
|
def _boot_time() -> str:
|
|
"""Return boot time from /proc/stat (btime)."""
|
|
for line in _read_lines("/proc/stat"):
|
|
if line.startswith("btime "):
|
|
try:
|
|
ts = int(line.split()[1])
|
|
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(ts))
|
|
except (ValueError, IndexError):
|
|
pass
|
|
return "unknown"
|
|
|
|
|
|
def _load_average() -> tuple[float, float, float]:
|
|
"""Return (1min, 5min, 15min) load averages."""
|
|
content = _read_file("/proc/loadavg")
|
|
try:
|
|
parts = content.split()
|
|
return float(parts[0]), float(parts[1]), float(parts[2])
|
|
except (IndexError, ValueError):
|
|
return (0.0, 0.0, 0.0)
|
|
|
|
|
|
def _logged_users() -> int:
|
|
"""Return the number of unique users with a login session (via ``who``)."""
|
|
out, _ = run(["who"])
|
|
users = {line.split()[0] for line in out.split("\n") if line.strip()}
|
|
return len(users)
|
|
|
|
|
|
def collect_overview() -> dict[str, Any]:
|
|
"""Collect system overview data."""
|
|
os_info = _parse_os_release()
|
|
os_name = os_info.get("PRETTY_NAME", os_info.get("NAME", "unknown"))
|
|
load1, load5, load15 = _load_average()
|
|
|
|
uptime_seconds = 0.0
|
|
uptime_raw = _read_file("/proc/uptime")
|
|
if uptime_raw:
|
|
try:
|
|
uptime_seconds = float(uptime_raw.split()[0])
|
|
except (ValueError, IndexError):
|
|
uptime_seconds = 0.0
|
|
|
|
return {
|
|
"hostname": socket.gethostname() or os.uname().nodename,
|
|
"os": os_name,
|
|
"kernel": os.uname().release,
|
|
"uptime_seconds": uptime_seconds,
|
|
"uptime_human": _uptime_human(uptime_seconds),
|
|
"load_1min": load1,
|
|
"load_5min": load5,
|
|
"load_15min": load15,
|
|
"cpu_cores": os.cpu_count() or 1,
|
|
"users_logged": _logged_users(),
|
|
"boot_time": _boot_time(),
|
|
}
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Section: CPU
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _cpu_model() -> str:
|
|
"""Get CPU model name from /proc/cpuinfo."""
|
|
for line in _read_lines("/proc/cpuinfo"):
|
|
if line.startswith("model name"):
|
|
return line.split(":", 1)[1].strip()
|
|
return "unknown"
|
|
|
|
|
|
def _cpu_arch() -> str:
|
|
"""Get CPU architecture."""
|
|
return os.uname().machine
|
|
|
|
|
|
def _cpu_cores_threads() -> tuple[int, int]:
|
|
"""Return (physical cores, logical threads).
|
|
|
|
"core id" is unique only within one physical package ("physical id"),
|
|
so the (package, core) pair is needed on multi-socket systems.
|
|
"""
|
|
cpuinfo = _read_file("/proc/cpuinfo")
|
|
cores: set[tuple[str, str]] = set()
|
|
threads = 0
|
|
physical_id = "0"
|
|
for line in cpuinfo.split("\n"):
|
|
if line.startswith("physical id"):
|
|
physical_id = line.split(":", 1)[1].strip()
|
|
elif line.startswith("core id"):
|
|
cores.add((physical_id, line.split(":", 1)[1].strip()))
|
|
elif line.startswith("processor"):
|
|
threads += 1
|
|
phys_count = len(cores) if cores else (os.cpu_count() or 1)
|
|
return phys_count, threads if threads else (os.cpu_count() or 1)
|
|
|
|
|
|
def _cpu_frequency() -> str:
|
|
"""Get current CPU frequency in MHz."""
|
|
# Try /proc/cpuinfo first
|
|
for line in _read_lines("/proc/cpuinfo"):
|
|
if "cpu mhz" in line.lower():
|
|
return line.split(":", 1)[1].strip() + " MHz"
|
|
# Try /sys/devices/system/cpu/cpu0/cpufreq/scaling_cur_freq
|
|
freq = _read_file("/sys/devices/system/cpu/cpu0/cpufreq/scaling_cur_freq")
|
|
if freq:
|
|
try:
|
|
return f"{int(freq) / 1000:.0f} MHz"
|
|
except ValueError:
|
|
pass
|
|
return "unknown"
|
|
|
|
|
|
def _cpu_temperature() -> str:
|
|
"""Get CPU temperature from hwmon sensors."""
|
|
base = "/sys/class/hwmon"
|
|
if not os.path.isdir(base):
|
|
return "unknown"
|
|
|
|
for entry in sorted(os.listdir(base)):
|
|
name = _read_file(os.path.join(base, entry, "name"))
|
|
if name.lower() in ("coretemp", "k10temp", "cpu_thermal", "acpitz"):
|
|
for i in range(1, 5):
|
|
temp_raw = _read_file(os.path.join(base, entry, f"temp{i}_input"))
|
|
if temp_raw:
|
|
try:
|
|
temp_c = int(temp_raw) / 1000.0
|
|
return f"{temp_c:.1f}°C"
|
|
except ValueError:
|
|
pass
|
|
# Fallback: try temp1_input without label
|
|
temp_raw = _read_file(os.path.join(base, entry, "temp1_input"))
|
|
if temp_raw:
|
|
try:
|
|
return f"{int(temp_raw) / 1000.0:.1f}°C"
|
|
except ValueError:
|
|
pass
|
|
|
|
return "unknown"
|
|
|
|
|
|
def _parse_cpu_fields(line: str) -> list[int]:
|
|
"""Parse a /proc/stat cpu line into jiffies, dropping the guest columns.
|
|
|
|
guest and guest_nice are already included in user and nice, so keeping
|
|
them in the total would double-count.
|
|
"""
|
|
try:
|
|
return [int(x) for x in line.split()[1:9]]
|
|
except ValueError:
|
|
return []
|
|
|
|
|
|
def _read_cpu_times() -> tuple[list[int], list[list[int]]]:
|
|
"""Read aggregate and per-CPU jiffy counters from /proc/stat."""
|
|
aggregate: list[int] = []
|
|
per_core: list[list[int]] = []
|
|
for line in _read_lines("/proc/stat"):
|
|
if line.startswith("cpu "):
|
|
aggregate = _parse_cpu_fields(line)
|
|
elif line[:3] == "cpu" and line[3:4].isdigit():
|
|
per_core.append(_parse_cpu_fields(line))
|
|
return aggregate, per_core
|
|
|
|
|
|
def _usage_percent(times1: list[int], times2: list[int]) -> float:
|
|
"""Compute usage percentage between two jiffy snapshots.
|
|
|
|
iowait counts as idle time — a process waiting on disk is not using CPU.
|
|
"""
|
|
if len(times1) < 5 or len(times2) < 5:
|
|
return 0.0
|
|
idle_delta = (times2[3] + times2[4]) - (times1[3] + times1[4])
|
|
total_delta = sum(times2) - sum(times1)
|
|
if total_delta <= 0:
|
|
return 0.0
|
|
return round((1 - idle_delta / total_delta) * 100, 1)
|
|
|
|
|
|
def _cpu_usage(sample_interval: float = 0.2) -> tuple[float, list[dict[str, Any]]]:
|
|
"""Sample aggregate and per-core CPU usage over a short interval.
|
|
|
|
Both are computed from the same sampling window, so the per-core view
|
|
matches the aggregate instead of showing since-boot averages.
|
|
"""
|
|
agg1, cores1 = _read_cpu_times()
|
|
time.sleep(sample_interval)
|
|
agg2, cores2 = _read_cpu_times()
|
|
|
|
per_core = [
|
|
{"cpu": str(index), "usage_pct": _usage_percent(c1, c2)}
|
|
for index, (c1, c2) in enumerate(zip(cores1, cores2))
|
|
]
|
|
return _usage_percent(agg1, agg2), per_core
|
|
|
|
|
|
def collect_cpu() -> dict[str, Any]:
|
|
"""Collect CPU data."""
|
|
phys_cores, threads = _cpu_cores_threads()
|
|
usage, per_core = _cpu_usage()
|
|
|
|
return {
|
|
"model": _cpu_model(),
|
|
"architecture": _cpu_arch(),
|
|
"physical_cores": phys_cores,
|
|
"threads": threads,
|
|
"frequency": _cpu_frequency(),
|
|
"temperature": _cpu_temperature(),
|
|
"usage_pct": usage,
|
|
"per_core": per_core,
|
|
}
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Section: Memory
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _parse_meminfo() -> dict[str, int]:
|
|
"""Parse /proc/meminfo into {key: value_in_kb}."""
|
|
mem: dict[str, int] = {}
|
|
for line in _read_lines("/proc/meminfo"):
|
|
parts = line.split()
|
|
if len(parts) >= 2:
|
|
key = parts[0].rstrip(":")
|
|
try:
|
|
mem[key] = int(parts[1])
|
|
except ValueError:
|
|
pass
|
|
return mem
|
|
|
|
|
|
def _top_mem_processes(count: int = 5) -> list[dict[str, Any]]:
|
|
"""Return top N memory-consuming processes using ps."""
|
|
out, _ = run(
|
|
["ps", "-eo", "pid,user,rss,comm", "--sort=-rss", "--no-headers"],
|
|
timeout=5,
|
|
)
|
|
procs: list[dict[str, Any]] = []
|
|
for line in out.strip().split("\n")[:count]:
|
|
parts = line.split(None, 3)
|
|
if len(parts) >= 4:
|
|
try:
|
|
rss_kb = int(parts[2])
|
|
procs.append(
|
|
{
|
|
"pid": int(parts[0]),
|
|
"user": parts[1],
|
|
"rss_mb": round(rss_kb / 1024, 1),
|
|
"command": parts[3],
|
|
}
|
|
)
|
|
except ValueError:
|
|
pass
|
|
return procs
|
|
|
|
|
|
def collect_memory() -> dict[str, Any]:
|
|
"""Collect memory data."""
|
|
mem = _parse_meminfo()
|
|
|
|
total = mem.get("MemTotal", 0) // 1024
|
|
free = mem.get("MemFree", 0) // 1024
|
|
available = mem.get("MemAvailable", 0) // 1024
|
|
used = total - available if available else total - free
|
|
|
|
swap_total = mem.get("SwapTotal", 0) // 1024
|
|
swap_free = mem.get("SwapFree", 0) // 1024
|
|
swap_used = swap_total - swap_free
|
|
|
|
usage_pct = round((used / total) * 100, 1) if total > 0 else 0.0
|
|
swap_pct = round((swap_used / swap_total) * 100, 1) if swap_total > 0 else 0.0
|
|
|
|
return {
|
|
"total_mb": total,
|
|
"used_mb": used,
|
|
"free_mb": free,
|
|
"available_mb": available,
|
|
"usage_pct": usage_pct,
|
|
"swap_total_mb": swap_total,
|
|
"swap_used_mb": swap_used,
|
|
"swap_usage_pct": swap_pct,
|
|
"top_processes": _top_mem_processes(),
|
|
"buffers_mb": mem.get("Buffers", 0) // 1024,
|
|
"cached_mb": mem.get("Cached", 0) // 1024,
|
|
}
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Section: Disk
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _skip_filesystem(fs_type: str, device: str) -> bool:
|
|
"""Return True if this filesystem should be excluded."""
|
|
excluded = {
|
|
"tmpfs",
|
|
"devtmpfs",
|
|
"squashfs",
|
|
"overlay",
|
|
"proc",
|
|
"sysfs",
|
|
"devpts",
|
|
"cgroup",
|
|
"cgroup2",
|
|
"pstore",
|
|
"bpf",
|
|
"debugfs",
|
|
"tracefs",
|
|
"securityfs",
|
|
"hugetlbfs",
|
|
"fuse.gvfsd-fuse",
|
|
"fuse.portal",
|
|
"ramfs",
|
|
"efivarfs",
|
|
"configfs",
|
|
"selinuxfs",
|
|
"systemd-1",
|
|
"mqueue",
|
|
"fusectl",
|
|
"sunrpc",
|
|
"binfmt_misc",
|
|
"autofs",
|
|
"rpc_pipefs",
|
|
"nsfs",
|
|
}
|
|
if fs_type in excluded:
|
|
return True
|
|
if device.startswith("tmpfs") or device.startswith("devtmpfs"):
|
|
return True
|
|
return False
|
|
|
|
|
|
def _mounts() -> list[dict[str, Any]]:
|
|
"""Parse /proc/mounts for real filesystems.
|
|
|
|
``used`` is derived from ``f_bavail`` (space available to unprivileged
|
|
users), so root-reserved blocks count as used — slightly more
|
|
conservative than ``df``, which is deliberate for a health report.
|
|
Bind mounts and btrfs subvolumes share the underlying device, so
|
|
entries are deduplicated by device id to keep the totals honest.
|
|
"""
|
|
fs_list: list[dict[str, Any]] = []
|
|
seen_devices: set[int] = set()
|
|
for line in _read_lines("/proc/mounts"):
|
|
parts = line.split()
|
|
if len(parts) < 6:
|
|
continue
|
|
device, mountpoint, fs_type = parts[0], parts[1], parts[2]
|
|
if _skip_filesystem(fs_type, device):
|
|
continue
|
|
mountpoint = mountpoint.replace("\\040", " ") # spaces are escaped in /proc/mounts
|
|
try:
|
|
st_dev = os.stat(mountpoint).st_dev
|
|
if st_dev in seen_devices:
|
|
continue
|
|
seen_devices.add(st_dev)
|
|
stat = os.statvfs(mountpoint)
|
|
total = stat.f_blocks * stat.f_frsize
|
|
available = stat.f_bavail * stat.f_frsize
|
|
used = total - available
|
|
size_gb = total / (1024**3)
|
|
used_gb = used / (1024**3)
|
|
avail_gb = available / (1024**3)
|
|
use_pct = round((used / total) * 100, 1) if total > 0 else 0.0
|
|
except OSError:
|
|
continue
|
|
|
|
fs_list.append(
|
|
{
|
|
"device": device,
|
|
"mountpoint": mountpoint,
|
|
"fstype": fs_type,
|
|
"size_gb": round(size_gb, 1),
|
|
"used_gb": round(used_gb, 1),
|
|
"available_gb": round(avail_gb, 1),
|
|
"use_pct": use_pct,
|
|
}
|
|
)
|
|
return fs_list
|
|
|
|
|
|
def _disk_io_stats() -> dict[str, Any] | None:
|
|
"""Collect disk I/O stats using iostat, if available.
|
|
|
|
Uses ``iostat -d -x 1 2``: the first report block covers averages since
|
|
boot, the second is a live 1-second sample — only the last block is
|
|
parsed. Columns are located by header name so that layout differences
|
|
between sysstat versions cannot silently misalign the values.
|
|
"""
|
|
out, _ = run(["iostat", "-d", "-x", "1", "2"], timeout=10)
|
|
if not out:
|
|
return None
|
|
|
|
# Split the output into report blocks, each starting at a Device header.
|
|
blocks: list[list[str]] = []
|
|
current: list[str] = []
|
|
for line in out.split("\n"):
|
|
if "Device" in line and "r/s" in line:
|
|
if current:
|
|
blocks.append(current)
|
|
current = [line]
|
|
elif current:
|
|
current.append(line)
|
|
if current:
|
|
blocks.append(current)
|
|
if not blocks:
|
|
return None
|
|
|
|
lines = blocks[-1]
|
|
columns = {name: index for index, name in enumerate(lines[0].split())}
|
|
|
|
def _value(parts: list[str], *names: str) -> float:
|
|
for name in names:
|
|
index = columns.get(name)
|
|
if index is not None and index < len(parts):
|
|
try:
|
|
return float(parts[index])
|
|
except ValueError:
|
|
return 0.0
|
|
return 0.0
|
|
|
|
devices: list[dict[str, Any]] = []
|
|
for line in lines[1:]:
|
|
parts = line.split()
|
|
if not parts or parts[0] == "Device":
|
|
continue
|
|
reads = _value(parts, "r/s")
|
|
writes = _value(parts, "w/s")
|
|
devices.append(
|
|
{
|
|
"device": parts[0],
|
|
"tps": round(reads + writes, 2),
|
|
"read_mb_s": round(_value(parts, "rkB/s", "rMB/s") / 1024, 2),
|
|
"write_mb_s": round(_value(parts, "wkB/s", "wMB/s") / 1024, 2),
|
|
}
|
|
)
|
|
|
|
return {"devices": devices} if devices else None
|
|
|
|
|
|
def _in_container() -> bool:
|
|
"""Best-effort container detection (Podman/Docker markers, overlay rootfs)."""
|
|
if os.path.exists("/run/.containerenv") or os.path.exists("/.dockerenv"):
|
|
return True
|
|
for line in _read_lines("/proc/mounts"):
|
|
parts = line.split()
|
|
if len(parts) >= 3 and parts[1] == "/" and parts[2] == "overlay":
|
|
return True
|
|
return False
|
|
|
|
|
|
def collect_disk() -> dict[str, Any]:
|
|
"""Collect disk data."""
|
|
mounts = _mounts()
|
|
total_gb = sum(m["size_gb"] for m in mounts)
|
|
used_gb = sum(m["used_gb"] for m in mounts)
|
|
critical = [m for m in mounts if m["use_pct"] > 90]
|
|
|
|
return {
|
|
"filesystems": mounts,
|
|
"total_gb": round(total_gb, 1),
|
|
"used_gb": round(used_gb, 1),
|
|
"usage_pct": round((used_gb / total_gb) * 100, 1) if total_gb > 0 else 0.0,
|
|
"critical_fs": critical,
|
|
"io_stats": _disk_io_stats(),
|
|
"container": _in_container(),
|
|
}
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Section: Network
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _net_interfaces() -> list[dict[str, Any]]:
|
|
"""Enumerate non-loopback network interfaces."""
|
|
ifaces: list[dict[str, Any]] = []
|
|
net_path = "/sys/class/net"
|
|
if not os.path.isdir(net_path):
|
|
return ifaces
|
|
|
|
for iface in sorted(os.listdir(net_path)):
|
|
if iface == "lo":
|
|
continue
|
|
operstate = _read_file(os.path.join(net_path, iface, "operstate"))
|
|
speed_raw = _read_file(os.path.join(net_path, iface, "speed"))
|
|
speed = f"{speed_raw} Mbps" if speed_raw and speed_raw != "-1" else "unknown"
|
|
|
|
# Get IPs via ip command
|
|
out, _ = run(["ip", "-br", "addr", "show", iface], timeout=5)
|
|
ipv4: list[str] = []
|
|
ipv6: list[str] = []
|
|
if out:
|
|
parts = out.split()
|
|
skip_next = False
|
|
for part in parts[2:]:
|
|
if part == "peer":
|
|
skip_next = True # next token is the remote end's address
|
|
continue
|
|
if skip_next:
|
|
skip_next = False
|
|
continue
|
|
if "." in part and "/" in part:
|
|
ipv4.append(part.split("/")[0])
|
|
elif ":" in part and "/" in part:
|
|
ipv6.append(part.split("/")[0])
|
|
|
|
ifaces.append(
|
|
{
|
|
"name": iface,
|
|
"state": operstate.upper() if operstate else "UNKNOWN",
|
|
"speed": speed,
|
|
"ipv4": ipv4,
|
|
"ipv6": ipv6,
|
|
}
|
|
)
|
|
return ifaces
|
|
|
|
|
|
def _connection_summary() -> dict[str, int]:
|
|
"""Get connection summary via ss -s."""
|
|
out, _ = run(["ss", "-s"], timeout=5)
|
|
result: dict[str, int] = {}
|
|
for line in out.split("\n"):
|
|
if "TCP:" in line:
|
|
match = re.search(r"estab\s+(\d+)", line)
|
|
if match:
|
|
result["tcp_established"] = int(match.group(1))
|
|
if "Total:" in line:
|
|
match = re.search(r"(\d+)", line)
|
|
if match:
|
|
result["total"] = int(match.group(1))
|
|
return result
|
|
|
|
|
|
def _default_routes() -> dict[str, str]:
|
|
"""Get default IPv6 and IPv4 routes (IPv6 first)."""
|
|
routes: dict[str, str] = {"v6": "", "v4": ""}
|
|
out, _ = run(["ip", "-6", "route", "show", "default"], timeout=5)
|
|
if out.strip():
|
|
routes["v6"] = out.strip()
|
|
out, _ = run(["ip", "-4", "route", "show", "default"], timeout=5)
|
|
if out.strip():
|
|
routes["v4"] = out.strip()
|
|
return routes
|
|
|
|
|
|
def _external_ip() -> dict[str, str]:
|
|
"""Detect external IP addresses via icanhazip.com (IPv6 first).
|
|
|
|
Only called when the user passes --external-ip: it sends a request to
|
|
a third-party service, which also reveals this host's address to it.
|
|
"""
|
|
result: dict[str, str] = {"v6": "unknown", "v4": "unknown"}
|
|
for family, flag in (("v6", "-6"), ("v4", "-4")):
|
|
out, _ = run(
|
|
["curl", flag, "--max-time", "5", "-s", "https://icanhazip.com"],
|
|
timeout=8,
|
|
)
|
|
if out.strip():
|
|
result[family] = out.strip()
|
|
return result
|
|
|
|
|
|
def collect_network(include_external: bool = False) -> dict[str, Any]:
|
|
"""Collect network data. External-IP lookup is opt-in."""
|
|
data: dict[str, Any] = {
|
|
"interfaces": _net_interfaces(),
|
|
"connections": _connection_summary(),
|
|
"routes": _default_routes(),
|
|
}
|
|
if include_external:
|
|
data["external_ip"] = _external_ip()
|
|
return data
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Section: GPU
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
# Matches rocm-smi data lines across ROCm versions, with or without the
|
|
# bracketed card index: "GPU[0]\t\t: Card series:\t…" and "GPU\t\t: …".
|
|
_ROCM_LINE = re.compile(r"^GPU(?:\[(\d+)\])?\s*:\s*([^:]+?)\s*:\s*(.+)$")
|
|
|
|
|
|
def _rocm_smi_bin() -> str | None:
|
|
"""Locate rocm-smi: PATH first (distro packages), ROCm default as fallback."""
|
|
found = shutil.which("rocm-smi")
|
|
if found:
|
|
return found
|
|
default = "/opt/rocm/bin/rocm-smi"
|
|
return default if os.path.isfile(default) else None
|
|
|
|
|
|
def _amd_gpu() -> dict[str, Any] | None:
|
|
"""Collect AMD GPU info via rocm-smi."""
|
|
binary = _rocm_smi_bin()
|
|
if binary is None:
|
|
return None
|
|
out, _ = run([binary, "--showproductname", "--showuse"], timeout=10)
|
|
if not out:
|
|
return None
|
|
|
|
# Group key/value pairs by card id — every rocm-smi data line is
|
|
# prefixed with "GPU[N] :", so looping per line would create one
|
|
# "card" per line instead of one card with many properties.
|
|
cards: dict[str, dict[str, str]] = {}
|
|
for line in out.split("\n"):
|
|
match = _ROCM_LINE.match(line.strip())
|
|
if not match:
|
|
continue
|
|
card_id = match.group(1) or "0"
|
|
key = match.group(2).strip()
|
|
value = match.group(3).strip()
|
|
cards.setdefault(card_id, {"id": card_id})[key] = value
|
|
|
|
if not cards:
|
|
return None
|
|
|
|
info: dict[str, Any] = {"vendor": "AMD", "cards": list(cards.values())}
|
|
vram = _rocm_vram(binary)
|
|
if vram:
|
|
info["vram"] = vram
|
|
return info
|
|
|
|
|
|
def _rocm_vram(binary: str) -> dict[str, Any] | None:
|
|
"""Parse ``rocm-smi --showmeminfo vram`` into MiB totals.
|
|
|
|
Values are reported in bytes ("VRAM Total Memory (B)") and aggregated
|
|
across all cards. Keys are matched exactly: "VRAM Total Used Memory (B)"
|
|
contains both the words "Total" and "Used", so substring matching
|
|
would double-count.
|
|
"""
|
|
out, _ = run([binary, "--showmeminfo", "vram"], timeout=10)
|
|
if not out:
|
|
return None
|
|
|
|
total_bytes = 0
|
|
used_bytes = 0
|
|
for line in out.split("\n"):
|
|
match = _ROCM_LINE.match(line.strip())
|
|
if not match:
|
|
continue
|
|
key = match.group(2).strip()
|
|
try:
|
|
value = int(match.group(3).strip().split()[0])
|
|
except (ValueError, IndexError):
|
|
continue
|
|
if key == "VRAM Total Used Memory (B)":
|
|
used_bytes += value
|
|
elif key == "VRAM Total Memory (B)":
|
|
total_bytes += value
|
|
|
|
if total_bytes <= 0:
|
|
return None
|
|
mib = 1024 * 1024
|
|
return {
|
|
"total_mb": total_bytes // mib,
|
|
"used_mb": used_bytes // mib,
|
|
"free_mb": (total_bytes - used_bytes) // mib,
|
|
}
|
|
|
|
|
|
def _nvidia_gpu() -> dict[str, Any] | None:
|
|
"""Collect NVIDIA GPU info via nvidia-smi."""
|
|
out, _ = run(
|
|
[
|
|
"nvidia-smi",
|
|
"--query-gpu=name,temperature.gpu,utilization.gpu,memory.used,memory.total",
|
|
"--format=csv,noheader,nounits",
|
|
],
|
|
timeout=10,
|
|
)
|
|
if not out:
|
|
return None
|
|
|
|
cards: list[dict[str, Any]] = []
|
|
for line in out.strip().split("\n"):
|
|
parts = [p.strip() for p in line.split(",")]
|
|
if len(parts) >= 3:
|
|
try:
|
|
cards.append(
|
|
{
|
|
"name": parts[0],
|
|
"temperature": parts[1] if len(parts) > 1 else "N/A",
|
|
"utilization_pct": parts[2] if len(parts) > 2 else "N/A",
|
|
"memory": (
|
|
f"{parts[3]} MiB / {parts[4]} MiB" if len(parts) >= 5 else "N/A"
|
|
),
|
|
}
|
|
)
|
|
except (ValueError, IndexError):
|
|
pass
|
|
|
|
return {"vendor": "NVIDIA", "cards": cards} if cards else None
|
|
|
|
|
|
def collect_gpu() -> dict[str, Any] | None:
|
|
"""Collect GPU data. Falls back to lspci when ROCm/nvidia-smi unavailable."""
|
|
nvidia = _nvidia_gpu()
|
|
if nvidia:
|
|
return nvidia
|
|
amd = _amd_gpu()
|
|
if amd:
|
|
return amd
|
|
# Fallback: detect GPU hardware via lspci (no driver needed)
|
|
return _lspci_gpu()
|
|
|
|
|
|
def _lspci_gpu() -> dict[str, Any] | None:
|
|
"""Detect GPU hardware via lspci — works even without vendor drivers.
|
|
|
|
Returns {"unavailable": True} when lspci itself is not installed, so
|
|
the report can say 'detection unavailable' instead of a false 'no GPU'.
|
|
"""
|
|
if shutil.which("lspci") is None:
|
|
return {"unavailable": True}
|
|
out, _ = run(["lspci"], timeout=10)
|
|
if not out:
|
|
return {"unavailable": True}
|
|
cards: list[dict[str, str]] = []
|
|
for line in out.split("\n"):
|
|
lower = line.lower()
|
|
if "vga" in lower or "display" in lower or "3d" in lower:
|
|
# Extract model: "01:00.0 VGA compatible controller: AMD ..."
|
|
parts = line.split(": ", 2)
|
|
model = parts[-1].strip() if len(parts) >= 3 else line.strip()
|
|
vendor = "AMD" if "amd" in lower or "radeon" in lower or "ati" in lower else ""
|
|
vendor = vendor or ("NVIDIA" if "nvidia" in lower else "")
|
|
vendor = vendor or ("Intel" if "intel" in lower else "")
|
|
cards.append({"model": model, "vendor": vendor, "driver": "(not loaded)"})
|
|
if not cards:
|
|
# Probe for AMD compute devices without a display class (e.g. Instinct)
|
|
out2, _ = run(["lspci", "-d", "1002:"], timeout=10)
|
|
if out2:
|
|
for line in out2.strip().split("\n"):
|
|
if line.strip():
|
|
cards.append(
|
|
{
|
|
"model": line.strip(),
|
|
"vendor": "AMD",
|
|
"driver": "(not loaded)",
|
|
}
|
|
)
|
|
if not cards:
|
|
return None
|
|
return {
|
|
"vendor": cards[0]["vendor"] or "unknown",
|
|
"cards": cards,
|
|
"fallback": True,
|
|
}
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Section: Services
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _failed_units() -> list[dict[str, str]] | None:
|
|
"""List failed systemd units, or None when systemctl is unavailable.
|
|
|
|
--plain is required: without it, systemd prefixes failed units with a
|
|
"●" bullet even when output is piped, which breaks field parsing.
|
|
"""
|
|
if shutil.which("systemctl") is None:
|
|
return None
|
|
out, _ = run(["systemctl", "--failed", "--no-legend", "--no-pager", "--plain"], timeout=10)
|
|
units: list[dict[str, str]] = []
|
|
for line in out.strip().split("\n"):
|
|
parts = line.split(None, 4)
|
|
if len(parts) >= 2:
|
|
units.append({"unit": parts[0], "state": parts[1]})
|
|
return units
|
|
|
|
|
|
def _check_service(name: str) -> dict[str, str]:
|
|
"""Check status of a named systemd service."""
|
|
out, _ = run(["systemctl", "is-active", name, "--no-pager"], timeout=5)
|
|
status = out.strip() if out.strip() else "not-found"
|
|
return {"name": name, "status": status}
|
|
|
|
|
|
def collect_services() -> dict[str, Any]:
|
|
"""Collect service data."""
|
|
key_services = ["sshd", "firewalld", "nginx", "postgresql"]
|
|
svc_status = {}
|
|
for svc in key_services:
|
|
svc_status[svc] = _check_service(svc)
|
|
|
|
return {
|
|
"failed_units": _failed_units(),
|
|
"key_services": svc_status,
|
|
}
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Section: Security
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _selinux_status() -> str:
|
|
"""Get SELinux mode: Enforcing, Permissive, Disabled, or 'unknown'.
|
|
|
|
'unknown' means getenforce is not installed at all — callers can then
|
|
distinguish 'no SELinux tooling' from 'SELinux disabled'.
|
|
"""
|
|
out, _ = run(["getenforce"], timeout=5)
|
|
return out.strip() if out.strip() else "unknown"
|
|
|
|
|
|
def _firewalld_info() -> dict[str, Any]:
|
|
"""Get firewalld status.
|
|
|
|
``available`` is False when firewall-cmd is not installed at all, so
|
|
callers can distinguish 'tool absent' from 'firewall disabled'.
|
|
"""
|
|
result: dict[str, Any] = {
|
|
"available": False,
|
|
"active": False,
|
|
"default_zone": "",
|
|
"zones": [],
|
|
}
|
|
out, _ = run(["firewall-cmd", "--state"], timeout=5)
|
|
state = out.strip()
|
|
if not state:
|
|
return result
|
|
result["available"] = True
|
|
result["active"] = state == "running"
|
|
|
|
out, _ = run(["firewall-cmd", "--get-default-zone"], timeout=5)
|
|
if out.strip():
|
|
result["default_zone"] = out.strip()
|
|
|
|
out, _ = run(["firewall-cmd", "--get-active-zones"], timeout=5)
|
|
if out.strip():
|
|
for line in out.split("\n"):
|
|
# Zone names sit at column 0; member lines (" interfaces: …")
|
|
# are indented — indentation must be checked before stripping.
|
|
if not line.strip() or line.startswith((" ", "\t")):
|
|
continue
|
|
result["zones"].append(line.strip())
|
|
|
|
return result
|
|
|
|
|
|
def _failed_ssh_attempts() -> int:
|
|
"""Count failed SSH password attempts today."""
|
|
out, _ = run(
|
|
["journalctl", "-u", "sshd", "--since", "today", "--no-pager"],
|
|
timeout=10,
|
|
)
|
|
if not out:
|
|
# Fall back to the "ssh" unit name used by some distros
|
|
out, _ = run(
|
|
["journalctl", "-u", "ssh", "--since", "today", "--no-pager"],
|
|
timeout=10,
|
|
)
|
|
return out.lower().count("failed password")
|
|
|
|
|
|
def _last_logins(count: int = 3) -> list[str]:
|
|
"""Get last N successful logins."""
|
|
out, _ = run(["last", "-n", str(count), "--no-hostname"], timeout=5)
|
|
logins: list[str] = []
|
|
for line in out.strip().split("\n"):
|
|
if line and "reboot" not in line.lower() and "wtmp" not in line.lower():
|
|
logins.append(line.strip())
|
|
if len(logins) >= count:
|
|
break
|
|
return logins
|
|
|
|
|
|
def _open_ports_count() -> int:
|
|
"""Count listening TCP ports."""
|
|
out, _ = run(["ss", "-tlnp"], timeout=5)
|
|
# Count lines minus header
|
|
lines = [entry for entry in out.strip().split("\n") if entry.strip() and "LISTEN" in entry]
|
|
return len(lines)
|
|
|
|
|
|
def collect_security() -> dict[str, Any]:
|
|
"""Collect security data."""
|
|
return {
|
|
"selinux": _selinux_status(),
|
|
"firewalld": _firewalld_info(),
|
|
"failed_ssh_attempts": _failed_ssh_attempts(),
|
|
"last_logins": _last_logins(),
|
|
"open_ports_count": _open_ports_count(),
|
|
}
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Section: Performance
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _process_counts() -> dict[str, int]:
|
|
"""Count total and zombie processes."""
|
|
total = 0
|
|
zombies = 0
|
|
for entry in os.listdir("/proc"):
|
|
try:
|
|
_pid = int(entry)
|
|
total += 1
|
|
status = _read_file(f"/proc/{entry}/status")
|
|
for line in status.split("\n"):
|
|
if line.startswith("State:"):
|
|
if "Z" in line:
|
|
zombies += 1
|
|
break
|
|
except (ValueError, OSError):
|
|
continue
|
|
return {"total": total, "zombies": zombies}
|
|
|
|
|
|
def _top_cpu_processes(count: int = 5) -> list[dict[str, Any]]:
|
|
"""Return top N CPU-consuming processes."""
|
|
out, _ = run(
|
|
["ps", "-eo", "pid,user,%cpu,comm", "--sort=-%cpu", "--no-headers"],
|
|
timeout=5,
|
|
)
|
|
procs: list[dict[str, Any]] = []
|
|
for line in out.strip().split("\n")[:count]:
|
|
parts = line.split(None, 3)
|
|
if len(parts) >= 4:
|
|
try:
|
|
procs.append(
|
|
{
|
|
"pid": int(parts[0]),
|
|
"user": parts[1],
|
|
"cpu_pct": float(parts[2]),
|
|
"command": parts[3],
|
|
}
|
|
)
|
|
except ValueError:
|
|
pass
|
|
return procs
|
|
|
|
|
|
def _context_switches() -> dict[str, int]:
|
|
"""Get context switches and interrupts from /proc/stat."""
|
|
result: dict[str, int] = {"ctxt": 0, "intr": 0}
|
|
for line in _read_lines("/proc/stat"):
|
|
if line.startswith("ctxt "):
|
|
try:
|
|
result["ctxt"] = int(line.split()[1])
|
|
except (IndexError, ValueError):
|
|
pass
|
|
if line.startswith("intr "):
|
|
try:
|
|
result["intr"] = int(line.split()[1])
|
|
except (IndexError, ValueError):
|
|
pass
|
|
return result
|
|
|
|
|
|
def collect_performance() -> dict[str, Any]:
|
|
"""Collect performance data."""
|
|
counts = _process_counts()
|
|
load1, _, _ = _load_average()
|
|
cores = os.cpu_count() or 1
|
|
|
|
return {
|
|
"processes_total": counts["total"],
|
|
"zombies": counts["zombies"],
|
|
"top_cpu_processes": _top_cpu_processes(),
|
|
"context_switches": _context_switches(),
|
|
"load_vs_cores_ratio": round(load1 / cores, 2),
|
|
}
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Section: Recent Issues
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def collect_issues() -> dict[str, Any]:
|
|
"""Collect recent system issues from journals."""
|
|
out, _ = run(
|
|
["journalctl", "-p", "err", "-n", "10", "--no-pager"],
|
|
timeout=10,
|
|
)
|
|
recent_errors = [
|
|
line.strip()
|
|
for line in out.strip().split("\n")
|
|
if line.strip() and "-- No entries --" not in line
|
|
][:10]
|
|
|
|
out, _ = run(
|
|
["journalctl", "-k", "--no-pager", "--grep", "Oops|BUG", "--output", "short"],
|
|
timeout=10,
|
|
)
|
|
kernel_oops = [
|
|
line.strip()
|
|
for line in out.strip().split("\n")
|
|
if line.strip() and "No entries" not in line
|
|
][:5]
|
|
|
|
# Also check kernel panic
|
|
if not kernel_oops:
|
|
out2, _ = run(
|
|
["journalctl", "-k", "--no-pager", "--grep", "panic", "--output", "short"],
|
|
timeout=10,
|
|
)
|
|
kernel_oops = [
|
|
line.strip()
|
|
for line in out2.strip().split("\n")
|
|
if line.strip() and "No entries" not in line
|
|
][:5]
|
|
|
|
out, _ = run(
|
|
["journalctl", "-k", "--no-pager", "--grep", "Out of memory", "--output", "short"],
|
|
timeout=10,
|
|
)
|
|
oom_events = [
|
|
line.strip()
|
|
for line in out.strip().split("\n")
|
|
if line.strip() and "No entries" not in line
|
|
][-5:]
|
|
|
|
return {
|
|
"recent_errors": recent_errors,
|
|
"kernel_oops": kernel_oops,
|
|
"oom_events": oom_events,
|
|
}
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Health Score
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def compute_health(data: dict[str, Any]) -> dict[str, Any]:
|
|
"""Compute a 0-100 health score with letter grade.
|
|
|
|
Each factor is worth a fixed maximum: CPU 10, memory 10, disk 15,
|
|
SELinux 10, failed services 15, load 10, zombies 10, swap 10,
|
|
firewall 10 (total 100). Factors whose data source is unavailable —
|
|
section not collected, or the required tool not installed — are
|
|
excluded and the score is renormalised against the remaining
|
|
maximum, so missing tooling never looks like a security failure.
|
|
An explicitly disabled control (e.g. SELinux "Disabled") still
|
|
scores 0.
|
|
|
|
Letter mapping: A ≥ 90, B ≥ 80, C ≥ 70, D ≥ 60, F < 60.
|
|
"""
|
|
earned = 0
|
|
possible = 0
|
|
breakdown: dict[str, int] = {}
|
|
maxima: dict[str, int] = {}
|
|
skipped: list[str] = []
|
|
|
|
def _factor(name: str, points: int | None, maximum: int) -> None:
|
|
"""Record one factor; ``points=None`` means 'not assessable'."""
|
|
nonlocal earned, possible
|
|
if points is None:
|
|
skipped.append(name)
|
|
return
|
|
breakdown[name] = points
|
|
maxima[name] = maximum
|
|
earned += points
|
|
possible += maximum
|
|
|
|
cpu_pct = data.get("cpu", {}).get("usage_pct")
|
|
cpu_points = None
|
|
if cpu_pct is not None:
|
|
cpu_points = 10 if cpu_pct < 80 else 8 if cpu_pct < 90 else 4 if cpu_pct < 95 else 0
|
|
_factor("cpu_usage", cpu_points, 10)
|
|
|
|
mem_pct = data.get("memory", {}).get("usage_pct")
|
|
mem_points = None
|
|
if mem_pct is not None:
|
|
mem_points = 10 if mem_pct < 85 else 8 if mem_pct < 90 else 4 if mem_pct < 95 else 0
|
|
_factor("memory_usage", mem_points, 10)
|
|
|
|
critical = data.get("disk", {}).get("critical_fs")
|
|
disk_points = None
|
|
if critical is not None:
|
|
disk_points = 15 if not critical else 10 if len(critical) == 1 else 0
|
|
_factor("disk_usage", disk_points, 15)
|
|
|
|
# "unknown" means getenforce is not installed — not assessable.
|
|
selinux = data.get("security", {}).get("selinux", "unknown")
|
|
selinux_points = None
|
|
if selinux != "unknown":
|
|
selinux_points = 10 if selinux == "Enforcing" else 5 if selinux == "Permissive" else 0
|
|
_factor("selinux", selinux_points, 10)
|
|
|
|
# None means systemctl is not installed — not assessable.
|
|
failed = data.get("services", {}).get("failed_units")
|
|
svc_points = None
|
|
if failed is not None:
|
|
svc_points = 15 if not failed else 10 if len(failed) <= 2 else 5 if len(failed) <= 5 else 0
|
|
_factor("services", svc_points, 15)
|
|
|
|
load1 = data.get("overview", {}).get("load_1min")
|
|
cores = data.get("overview", {}).get("cpu_cores", 1) or 1
|
|
load_points = None
|
|
if load1 is not None:
|
|
load_points = 10 if load1 < cores else 5 if load1 < cores * 2 else 0
|
|
_factor("load", load_points, 10)
|
|
|
|
zombies = data.get("performance", {}).get("zombies")
|
|
zombie_points = None
|
|
if zombies is not None:
|
|
zombie_points = 10 if zombies == 0 else 5 if zombies <= 3 else 0
|
|
_factor("zombies", zombie_points, 10)
|
|
|
|
swap_pct = data.get("memory", {}).get("swap_usage_pct")
|
|
swap_points = None
|
|
if swap_pct is not None:
|
|
swap_points = 10 if swap_pct < 50 else 5 if swap_pct < 75 else 0
|
|
_factor("swap", swap_points, 10)
|
|
|
|
# available=False means firewall-cmd is not installed — not assessable.
|
|
fw = data.get("security", {}).get("firewalld", {})
|
|
fw_points = None
|
|
if fw.get("available"):
|
|
fw_points = 10 if fw.get("active") else 0
|
|
_factor("firewall", fw_points, 10)
|
|
|
|
score = round(100 * earned / possible) if possible > 0 else 0
|
|
|
|
if score >= 90:
|
|
letter = "A"
|
|
elif score >= 80:
|
|
letter = "B"
|
|
elif score >= 70:
|
|
letter = "C"
|
|
elif score >= 60:
|
|
letter = "D"
|
|
else:
|
|
letter = "F"
|
|
|
|
return {
|
|
"score": score,
|
|
"max_score": 100,
|
|
"grade": letter,
|
|
"breakdown": breakdown,
|
|
"maxima": maxima,
|
|
"factors_skipped": skipped,
|
|
}
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Human-readable report
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _section_header(title: str) -> None:
|
|
"""Print a section divider."""
|
|
print(f"\n{BOLD}── {title} ──{RESET}")
|
|
|
|
|
|
def _kv(label: str, value: str, colour: str = "") -> None:
|
|
"""Print a key-value line, optionally wrapped in a colour."""
|
|
if colour:
|
|
print(f" {label:<22s} {colour}{value}{RESET}")
|
|
else:
|
|
print(f" {label:<22s} {value}")
|
|
|
|
|
|
def _kv_status(label: str, ok: bool, ok_text: str = "yes", fail_text: str = "no") -> None:
|
|
"""Print a key-value line with colour-coded boolean status."""
|
|
if ok:
|
|
print(f" {label:<22s} {GREEN}{ok_text}{RESET}")
|
|
else:
|
|
print(f" {label:<22s} {RED}{fail_text}{RESET}")
|
|
|
|
|
|
def print_report(data: dict[str, Any]) -> None:
|
|
"""Render the human-readable diagnostic report to stdout."""
|
|
health = data.get("health", {})
|
|
grade_letter = str(health.get("grade", "?"))
|
|
grade_colour = GRADE_COLOURS.get(grade_letter, RESET)
|
|
|
|
# Present sections in canonical order; numbers adapt to --section subsets.
|
|
order = [name for name in ALL_SECTIONS if name in data]
|
|
|
|
# --- Header ---
|
|
print()
|
|
print(f"{BOLD}{'═' * WIDTH}{RESET}")
|
|
overview = data.get("overview", {})
|
|
print(f" {BOLD}System Diagnostics v{VERSION}{RESET}")
|
|
print(f"{BOLD}{'═' * WIDTH}{RESET}")
|
|
print(f" Hostname: {overview.get('hostname', 'unknown')}")
|
|
print(f" Date: {time.strftime('%Y-%m-%d %H:%M:%S')}")
|
|
print(
|
|
f" Health: {grade_colour}{BOLD}{grade_letter}{RESET} "
|
|
f"({health.get('score', '?')}/{health.get('max_score', 100)})"
|
|
)
|
|
|
|
# --- System Overview ---
|
|
if "overview" in data:
|
|
_section_header(f"{order.index('overview') + 1}. System Overview")
|
|
o = data["overview"]
|
|
_kv("OS", str(o.get("os", "unknown")))
|
|
_kv("Kernel", str(o.get("kernel", "unknown")))
|
|
_kv("Uptime", str(o.get("uptime_human", "unknown")))
|
|
_kv("Boot time", str(o.get("boot_time", "unknown")))
|
|
|
|
load1 = o.get("load_1min", 0)
|
|
load5 = o.get("load_5min", 0)
|
|
load15 = o.get("load_15min", 0)
|
|
cores = o.get("cpu_cores", 1)
|
|
load_col = GREEN if load1 < cores else YELLOW if load1 < cores * 2 else RED
|
|
_kv("Load avg (1/5/15m)", f"{load_col}{load1:.2f} / {load5:.2f} / {load15:.2f}{RESET}")
|
|
_kv("Users logged in", str(o.get("users_logged", 0)))
|
|
|
|
# --- CPU ---
|
|
if "cpu" in data:
|
|
_section_header(f"{order.index('cpu') + 1}. CPU")
|
|
c = data["cpu"]
|
|
phys = c.get("physical_cores", 0)
|
|
threads = c.get("threads", 0)
|
|
_kv("Model", str(c.get("model", "unknown")))
|
|
_kv("Architecture", str(c.get("architecture", "unknown")))
|
|
_kv("Cores / Threads", f"{phys} physical / {threads} logical")
|
|
_kv("Frequency", str(c.get("frequency", "unknown")))
|
|
_kv("Temperature", str(c.get("temperature", "unknown")))
|
|
usage = c.get("usage_pct", 0.0)
|
|
usage_col = GREEN if usage < 80 else YELLOW if usage < 90 else RED
|
|
_kv("CPU usage", f"{usage_col}{usage}%{RESET} {_bar(usage)}")
|
|
|
|
per_core = c.get("per_core", [])
|
|
if per_core:
|
|
print(f" {BOLD}Per-core usage:{RESET}")
|
|
for core in per_core:
|
|
core_usage = core["usage_pct"]
|
|
core_col = GREEN if core_usage < 80 else YELLOW if core_usage < 90 else RED
|
|
print(
|
|
f" cpu{core['cpu']:<4s} {core_col}{core_usage:>5.1f}%{RESET} "
|
|
f"{_bar(core_usage)}"
|
|
)
|
|
|
|
# --- Memory ---
|
|
if "memory" in data:
|
|
_section_header(f"{order.index('memory') + 1}. Memory")
|
|
m = data["memory"]
|
|
_kv("Total", f"{m.get('total_mb', 0)} MiB")
|
|
_kv("Used", f"{m.get('used_mb', 0)} MiB {_bar(m.get('usage_pct', 0))}")
|
|
_kv("Available", f"{m.get('available_mb', 0)} MiB")
|
|
_kv("Buffers", f"{m.get('buffers_mb', 0)} MiB")
|
|
_kv("Cached", f"{m.get('cached_mb', 0)} MiB")
|
|
|
|
swap_total = m.get("swap_total_mb", 0)
|
|
swap_used = m.get("swap_used_mb", 0)
|
|
swap_pct = m.get("swap_usage_pct", 0)
|
|
swap_col = GREEN if swap_pct < 50 else YELLOW if swap_pct < 75 else RED
|
|
_kv("Swap", f"{swap_col}{swap_used} / {swap_total} MiB{RESET} {_bar(swap_pct)}")
|
|
|
|
top_mem = m.get("top_processes", [])
|
|
if top_mem:
|
|
print(f"\n {BOLD}Top 5 memory consumers:{RESET}")
|
|
print(f" {'PID':<8s} {'USER':<10s} {'RSS':>8s} COMMAND")
|
|
for proc in top_mem:
|
|
print(
|
|
f" {proc['pid']:<8d} {proc['user']:<10s} "
|
|
f"{proc['rss_mb']:>6.1f}M {proc['command']}"
|
|
)
|
|
|
|
# --- Disk ---
|
|
if "disk" in data:
|
|
_section_header(f"{order.index('disk') + 1}. Disk")
|
|
d = data["disk"]
|
|
_kv("Total space", f"{d.get('total_gb', 0):.1f} GiB")
|
|
_kv("Used", f"{d.get('used_gb', 0):.1f} GiB {_bar(d.get('usage_pct', 0))}")
|
|
|
|
fss = d.get("filesystems", [])
|
|
if fss:
|
|
print(f"\n {BOLD}Filesystems:{RESET}")
|
|
header = f" {'Device':<20s} {'Mount':<18s} {'Size':>7s} {'Used':>7s} {'Use%'}"
|
|
print(header)
|
|
print(f" {'-' * (len(header) - 2)}")
|
|
for fs in fss:
|
|
use_pct = fs["use_pct"]
|
|
pct_col = GREEN if use_pct < 80 else YELLOW if use_pct < 90 else RED
|
|
dev = fs["device"][:19] if len(fs["device"]) > 19 else fs["device"]
|
|
mnt = fs["mountpoint"][:17] if len(fs["mountpoint"]) > 17 else fs["mountpoint"]
|
|
print(
|
|
f" {dev:<20s} {mnt:<18s} {fs['size_gb']:>6.0f}G "
|
|
f"{fs['used_gb']:>6.0f}G {pct_col}{use_pct:>4.1f}%{RESET}"
|
|
)
|
|
|
|
if not fss and d.get("container"):
|
|
print(f" {DIM}(container detected — overlay root filesystem not shown){RESET}")
|
|
|
|
critical = d.get("critical_fs", [])
|
|
if critical:
|
|
print(f"\n {RED}⚠ Warning: {len(critical)} filesystem(s) above 90% usage!{RESET}")
|
|
|
|
io_stats = d.get("io_stats")
|
|
if io_stats:
|
|
print(f"\n {BOLD}I/O Stats (iostat):{RESET}")
|
|
for dev in io_stats.get("devices", []):
|
|
print(
|
|
f" {dev['device']:<10s} tps: {dev['tps']:>8.1f} "
|
|
f"read: {dev['read_mb_s']:>8.2f} MB/s "
|
|
f"write: {dev['write_mb_s']:>8.2f} MB/s"
|
|
)
|
|
|
|
# --- Network ---
|
|
if "network" in data:
|
|
_section_header(f"{order.index('network') + 1}. Network")
|
|
n = data["network"]
|
|
|
|
ifaces = n.get("interfaces", [])
|
|
for iface in ifaces:
|
|
state_col = GREEN if iface["state"] == "UP" else RED
|
|
v6 = ", ".join(iface["ipv6"]) if iface["ipv6"] else f"{DIM}(none){RESET}"
|
|
v4 = ", ".join(iface["ipv4"]) if iface["ipv4"] else f"{DIM}(none){RESET}"
|
|
print(
|
|
f" {iface['name']:<8s} {state_col}{iface['state']:<5s}{RESET} "
|
|
f"speed: {iface.get('speed', 'unknown')}"
|
|
)
|
|
print(f" IPv6: {v6}")
|
|
print(f" IPv4: {v4}")
|
|
|
|
conns = n.get("connections", {})
|
|
if conns:
|
|
_kv("TCP established", str(conns.get("tcp_established", "?")))
|
|
_kv("Total connections", str(conns.get("total", "?")))
|
|
|
|
routes = n.get("routes", {})
|
|
if routes.get("v6"):
|
|
_kv("Default IPv6 route", str(routes["v6"]))
|
|
if routes.get("v4"):
|
|
_kv("Default IPv4 route", str(routes["v4"]))
|
|
|
|
ext_ip = n.get("external_ip", {})
|
|
if ext_ip.get("v6", "unknown") != "unknown":
|
|
_kv("External IPv6", str(ext_ip["v6"]))
|
|
if ext_ip.get("v4", "unknown") != "unknown":
|
|
_kv("External IPv4", str(ext_ip["v4"]))
|
|
|
|
# --- GPU ---
|
|
if "gpu" in data:
|
|
_section_header(f"{order.index('gpu') + 1}. GPU")
|
|
g = data["gpu"]
|
|
if g and g.get("unavailable"):
|
|
print(f" {DIM}GPU detection unavailable (no nvidia-smi/rocm-smi/lspci){RESET}")
|
|
elif g:
|
|
if g.get("fallback"):
|
|
print(f" {YELLOW}(hardware detected, GPU drivers not loaded){RESET}")
|
|
print(f" Vendor: {g.get('vendor', 'unknown')}")
|
|
for card in g.get("cards", []):
|
|
for k, v in card.items():
|
|
_kv(k.capitalize(), str(v))
|
|
vram = g.get("vram")
|
|
if vram:
|
|
total = vram.get("total_mb", 0)
|
|
used = vram.get("used_mb", 0)
|
|
free = vram.get("free_mb", 0)
|
|
pct = (used / total * 100) if total > 0 else 0
|
|
colour = GREEN if pct < 80 else YELLOW if pct < 90 else RED
|
|
_kv("VRAM Used/Total", f"{colour}{used}/{total} MiB{RESET} ({pct:.0f}%)")
|
|
_kv("VRAM Free", f"{free} MiB")
|
|
else:
|
|
print(f" {DIM}No GPU detected{RESET}")
|
|
|
|
# --- Services ---
|
|
if "services" in data:
|
|
_section_header(f"{order.index('services') + 1}. Services")
|
|
s = data["services"]
|
|
|
|
failed = s.get("failed_units")
|
|
if failed is None:
|
|
print(f" {DIM}Failed units: unavailable (systemctl not installed){RESET}")
|
|
elif failed:
|
|
print(f" {RED}Failed units: {len(failed)}{RESET}")
|
|
for unit in failed:
|
|
print(f" {RED}✗{RESET} {unit['unit']} ({unit['state']})")
|
|
else:
|
|
print(f" {GREEN}No failed units{RESET}")
|
|
|
|
print()
|
|
ks = s.get("key_services", {})
|
|
for name, info in sorted(ks.items()):
|
|
status = info.get("status", "unknown")
|
|
if status == "active":
|
|
print(f" {GREEN}●{RESET} {name:<15s} {GREEN}{status}{RESET}")
|
|
elif status == "inactive":
|
|
print(f" {YELLOW}○{RESET} {name:<15s} {YELLOW}{status}{RESET}")
|
|
elif status == "failed":
|
|
print(f" {RED}✗{RESET} {name:<15s} {RED}{status}{RESET}")
|
|
else:
|
|
print(f" {DIM}─{RESET} {name:<15s} {DIM}{status}{RESET}")
|
|
|
|
# --- Security ---
|
|
if "security" in data:
|
|
_section_header(f"{order.index('security') + 1}. Security")
|
|
sc = data["security"]
|
|
|
|
selinux = sc.get("selinux", "unknown")
|
|
if selinux == "Enforcing":
|
|
sel_col = GREEN
|
|
elif selinux == "Permissive":
|
|
sel_col = YELLOW
|
|
elif selinux == "Disabled":
|
|
sel_col = RED
|
|
else: # "unknown" — SELinux tooling not installed
|
|
sel_col = DIM
|
|
_kv("SELinux", f"{sel_col}{selinux}{RESET}")
|
|
|
|
fw = sc.get("firewalld", {})
|
|
if fw.get("available"):
|
|
_kv_status("Firewalld active", bool(fw.get("active")), "active", "inactive")
|
|
if fw.get("default_zone"):
|
|
_kv("Default zone", str(fw["default_zone"]))
|
|
if fw.get("zones"):
|
|
_kv("Active zones", ", ".join(fw["zones"]))
|
|
else:
|
|
_kv("Firewalld", f"{DIM}not installed{RESET}")
|
|
|
|
_kv("Failed SSH (today)", str(sc.get("failed_ssh_attempts", 0)))
|
|
_kv("Open TCP ports", str(sc.get("open_ports_count", 0)))
|
|
|
|
logins = sc.get("last_logins", [])
|
|
if logins:
|
|
print(f"\n {BOLD}Last logins:{RESET}")
|
|
for login in logins:
|
|
print(f" {login}")
|
|
|
|
# --- Performance ---
|
|
if "performance" in data:
|
|
_section_header(f"{order.index('performance') + 1}. Performance")
|
|
p = data["performance"]
|
|
|
|
_kv("Processes total", str(p.get("processes_total", 0)))
|
|
zombies = p.get("zombies", 0)
|
|
zombie_col = GREEN if zombies == 0 else RED
|
|
_kv("Zombie processes", f"{zombie_col}{zombies}{RESET}")
|
|
|
|
ctxt = p.get("context_switches", {})
|
|
_kv("Ctxt switches (boot)", str(ctxt.get("ctxt", "?")))
|
|
_kv("Interrupts (boot)", str(ctxt.get("intr", "?")))
|
|
|
|
load_ratio = p.get("load_vs_cores_ratio", 0)
|
|
ratio_col = GREEN if load_ratio < 1 else YELLOW if load_ratio < 2 else RED
|
|
_kv("Load / cores ratio", f"{ratio_col}{load_ratio:.2f}{RESET}")
|
|
|
|
top_cpu = p.get("top_cpu_processes", [])
|
|
if top_cpu:
|
|
print(f"\n {BOLD}Top 5 CPU consumers (lifetime average):{RESET}")
|
|
print(f" {'PID':<8s} {'USER':<10s} {'CPU%':>6s} COMMAND")
|
|
for proc in top_cpu:
|
|
print(
|
|
f" {proc['pid']:<8d} {proc['user']:<10s} "
|
|
f"{proc['cpu_pct']:>5.1f}% {proc['command']}"
|
|
)
|
|
|
|
# --- Recent Issues ---
|
|
if "issues" in data:
|
|
_section_header(f"{order.index('issues') + 1}. Recent Issues")
|
|
i = data["issues"]
|
|
|
|
errors = i.get("recent_errors", [])
|
|
if errors:
|
|
print(f" {BOLD}Last {len(errors)} errors from journal:{RESET}")
|
|
for line in errors:
|
|
# Truncate long lines
|
|
display = line[:120] + "…" if len(line) > 120 else line
|
|
print(f" {DIM}{display}{RESET}")
|
|
else:
|
|
print(f" {GREEN}No recent errors in journal{RESET}")
|
|
|
|
oops = i.get("kernel_oops", [])
|
|
if oops:
|
|
print(f"\n {RED}⚠ Kernel OOPS / panic messages:{RESET}")
|
|
for line in oops:
|
|
display = line[:120] + "…" if len(line) > 120 else line
|
|
print(f" {RED}{display}{RESET}")
|
|
else:
|
|
print(f"\n {GREEN}No kernel OOPS detected{RESET}")
|
|
|
|
oom = i.get("oom_events", [])
|
|
if oom:
|
|
print(f"\n {RED}⚠ OOM events:{RESET}")
|
|
for line in oom:
|
|
display = line[:120] + "…" if len(line) > 120 else line
|
|
print(f" {RED}{display}{RESET}")
|
|
else:
|
|
print(f"\n {GREEN}No OOM events in kernel log{RESET}")
|
|
|
|
# --- Health Score ---
|
|
_section_header(f"{len(order) + 1}. Health Score")
|
|
bd = health.get("breakdown", {})
|
|
labels = {
|
|
"cpu_usage": "CPU usage < 80%",
|
|
"memory_usage": "Memory usage < 85%",
|
|
"disk_usage": "No disk > 90%",
|
|
"selinux": "SELinux enforcing",
|
|
"services": "No failed services",
|
|
"load": "Load < cores",
|
|
"zombies": "No zombie processes",
|
|
"swap": "Swap usage < 50%",
|
|
"firewall": "Firewall active",
|
|
}
|
|
maxima = health.get("maxima", {})
|
|
for key, label in labels.items():
|
|
if key not in bd:
|
|
continue # factor could not be assessed in this run
|
|
pts = bd[key]
|
|
max_pts = maxima.get(key, 10)
|
|
bar = "█" * max(0, pts) + "░" * (max_pts - max(0, pts))
|
|
print(f" {label:<24s} {bar} {pts}/{max_pts}")
|
|
skipped = health.get("factors_skipped", [])
|
|
if skipped:
|
|
names = ", ".join(labels.get(key, key) for key in skipped)
|
|
print(f" {DIM}Not assessed (data unavailable — score renormalised):{RESET}")
|
|
print(f" {DIM}{names}{RESET}")
|
|
|
|
# --- Footer ---
|
|
print()
|
|
print(f"{BOLD}{'═' * WIDTH}{RESET}")
|
|
print(
|
|
f" {BOLD}Health Score: {grade_colour}{grade_letter}{RESET} "
|
|
f"({health.get('score', '?')}/{health.get('max_score', 100)})"
|
|
)
|
|
print(f"{BOLD}{'═' * WIDTH}{RESET}")
|
|
print()
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Data collection orchestrator
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def collect_all(sections: list[str], external_ip: bool = False) -> dict[str, Any]:
|
|
"""Collect all diagnostics data, optionally filtered by section."""
|
|
data: dict[str, Any] = {
|
|
"version": VERSION,
|
|
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
|
|
}
|
|
|
|
collectors: dict[str, tuple[Callable[[], Any], str]] = {
|
|
"overview": (collect_overview, "system overview"),
|
|
"cpu": (collect_cpu, "CPU"),
|
|
"memory": (collect_memory, "memory"),
|
|
"disk": (collect_disk, "disk"),
|
|
"network": (lambda: collect_network(include_external=external_ip), "network"),
|
|
"gpu": (collect_gpu, "GPU"),
|
|
"services": (collect_services, "services"),
|
|
"security": (collect_security, "security"),
|
|
"performance": (collect_performance, "performance"),
|
|
"issues": (collect_issues, "recent issues"),
|
|
}
|
|
|
|
with ThreadPoolExecutor(max_workers=6) as executor:
|
|
future_map: dict[Future[Any], str] = {
|
|
executor.submit(collectors[section][0]): section
|
|
for section in dict.fromkeys(sections)
|
|
if section in collectors
|
|
}
|
|
|
|
for future in as_completed(future_map):
|
|
section = future_map[future]
|
|
label = collectors[section][1]
|
|
_status(f"Collecting {label}")
|
|
try:
|
|
result = future.result()
|
|
data[section] = result if result is not None else {}
|
|
_status_done()
|
|
except Exception as exc:
|
|
data[section] = {"error": str(exc)}
|
|
_status_done("error")
|
|
print(f" [{label}] {exc}", file=sys.stderr)
|
|
|
|
return data
|
|
|
|
|
|
def main() -> None:
|
|
"""Parse arguments, collect diagnostics, and print the report."""
|
|
parser = argparse.ArgumentParser(
|
|
description=(
|
|
"System diagnostics — CPU, memory, disk, network, GPU, services, security, performance"
|
|
),
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
)
|
|
parser.add_argument(
|
|
"--section",
|
|
default="",
|
|
help=(
|
|
f"Comma-separated sections to run. Available: {', '.join(ALL_SECTIONS)} [default: all]"
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--external-ip",
|
|
action="store_true",
|
|
help=(
|
|
"Also detect external IP addresses via icanhazip.com "
|
|
"(sends a request to a third-party service)"
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--json",
|
|
action="store_true",
|
|
help="Output machine-readable JSON to stdout",
|
|
)
|
|
parser.add_argument(
|
|
"--version",
|
|
action="version",
|
|
version=f"%(prog)s {VERSION}",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
if sys.platform != "linux":
|
|
print(
|
|
f"{RED}Error: system-diag.py currently supports Linux only "
|
|
f"(detected platform: {sys.platform}).{RESET}",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
|
|
# Determine which sections to run
|
|
if args.section:
|
|
sections = [s.strip().lower() for s in args.section.split(",")]
|
|
invalid = [s for s in sections if s not in ALL_SECTIONS]
|
|
if invalid:
|
|
print(
|
|
f"{RED}Error: unknown section(s): {', '.join(invalid)}{RESET}",
|
|
file=sys.stderr,
|
|
)
|
|
print(
|
|
f"Available: {', '.join(ALL_SECTIONS)}",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
else:
|
|
sections = list(ALL_SECTIONS)
|
|
|
|
# Collect data
|
|
data = collect_all(sections, external_ip=args.external_ip)
|
|
|
|
# Compute health score
|
|
health = compute_health(data)
|
|
data["health"] = health
|
|
|
|
# Output
|
|
if args.json:
|
|
print(json.dumps(data, indent=2, default=str))
|
|
else:
|
|
print_report(data)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|