Code for How to Get Hardware and System Information in Python Tutorial

system_info_report.py

from __future__ import annotations

import argparse
import csv
import json
import os
import platform
import shutil
import socket
import subprocess
import sys
import time
import warnings
from datetime import datetime
from pathlib import Path
from typing import Any

import psutil

try:
    import cpuinfo  # pip install py-cpuinfo
except ImportError:  # py-cpuinfo is optional
    cpuinfo = None


JsonDict = dict[str, Any]


def bytes_to_human(value: int | float | None, suffix: str = "B") -> str:
    """Convert a byte value to a human-readable string."""
    if value is None:
        return "N/A"
    value = float(value)
    for unit in ("", "K", "M", "G", "T", "P"):
        if abs(value) < 1024:
            return f"{value:.2f}{unit}{suffix}"
        value /= 1024
    return f"{value:.2f}EB"


def seconds_to_human(seconds: int | float | None) -> str:
    """Convert seconds to a compact days/hours/minutes string."""
    if seconds is None or seconds < 0:
        return "N/A"
    seconds = int(seconds)
    days, remainder = divmod(seconds, 86_400)
    hours, remainder = divmod(remainder, 3_600)
    minutes, seconds = divmod(remainder, 60)
    parts = []
    if days:
        parts.append(f"{days}d")
    if hours:
        parts.append(f"{hours}h")
    if minutes:
        parts.append(f"{minutes}m")
    if seconds or not parts:
        parts.append(f"{seconds}s")
    return " ".join(parts)


def timestamp_to_iso(timestamp: float) -> str:
    """Return a local ISO timestamp without microseconds."""
    return datetime.fromtimestamp(timestamp).astimezone().isoformat(timespec="seconds")


def cpu_brand() -> str:
    """Return a friendly CPU name when available."""
    if cpuinfo is not None:
        try:
            info = cpuinfo.get_cpu_info()
            brand = info.get("brand_raw") or info.get("arch_string_raw")
            if brand and brand.lower() != "unknown":
                return brand
        except Exception:
            pass
    return platform.processor() or platform.machine() or "Unknown"


def get_linux_distribution() -> JsonDict | None:
    """Return Linux distribution metadata when platform.freedesktop_os_release exists."""
    if platform.system().lower() != "linux":
        return None
    try:
        release = platform.freedesktop_os_release()
    except Exception:
        return None
    return {
        "name": release.get("NAME"),
        "version": release.get("VERSION"),
        "pretty_name": release.get("PRETTY_NAME"),
    }


def collect_system() -> JsonDict:
    """Collect OS, Python, architecture and boot-time information."""
    uname = platform.uname()
    boot_time = psutil.boot_time()
    return {
        "system": uname.system,
        "node": uname.node,
        "release": uname.release,
        "version": uname.version,
        "machine": uname.machine,
        "processor": uname.processor,
        "architecture": platform.architecture()[0],
        "python_version": platform.python_version(),
        "python_executable": sys.executable,
        "boot_time": timestamp_to_iso(boot_time),
        "uptime": seconds_to_human(time.time() - boot_time),
        "linux_distribution": get_linux_distribution(),
    }


def collect_cpu(interval: float = 0.5) -> JsonDict:
    """Collect CPU model, counts, frequency and utilization."""
    freq = psutil.cpu_freq()
    load_avg = None
    if hasattr(os, "getloadavg"):
        try:
            load_avg = os.getloadavg()
        except OSError:
            load_avg = None

    per_core = psutil.cpu_percent(interval=interval, percpu=True)
    return {
        "brand": cpu_brand(),
        "physical_cores": psutil.cpu_count(logical=False),
        "logical_cores": psutil.cpu_count(logical=True),
        "max_frequency_mhz": round(freq.max, 2) if freq else None,
        "min_frequency_mhz": round(freq.min, 2) if freq else None,
        "current_frequency_mhz": round(freq.current, 2) if freq else None,
        "total_usage_percent": psutil.cpu_percent(interval=None),
        "per_core_usage_percent": per_core,
        "load_average": load_avg,
    }


def memory_block(mem: Any) -> JsonDict:
    """Convert a psutil memory namedtuple to raw and human-readable fields."""
    data = mem._asdict()
    return {
        key: value for key, value in data.items()
    } | {
        f"{key}_human": bytes_to_human(value)
        for key, value in data.items()
        if isinstance(value, int) and key != "percent"
    }


def collect_memory() -> JsonDict:
    """Collect virtual memory and swap details."""
    with warnings.catch_warnings():
        warnings.simplefilter("ignore", RuntimeWarning)
        swap = psutil.swap_memory()
    return {
        "virtual": memory_block(psutil.virtual_memory()),
        "swap": memory_block(swap),
    }


def disk_usage_item(device: str, mountpoint: str, fstype: str = "", opts: str = "") -> JsonDict:
    """Build one disk usage dictionary for a mountpoint."""
    item: JsonDict = {"device": device, "mountpoint": mountpoint, "fstype": fstype, "opts": opts}
    try:
        usage = psutil.disk_usage(mountpoint)
    except (PermissionError, FileNotFoundError, OSError) as exc:
        item["error"] = str(exc)
    else:
        item.update(
            {
                "total": usage.total,
                "used": usage.used,
                "free": usage.free,
                "percent": usage.percent,
                "total_human": bytes_to_human(usage.total),
                "used_human": bytes_to_human(usage.used),
                "free_human": bytes_to_human(usage.free),
            }
        )
    return item


def collect_disks(all_partitions: bool = False) -> JsonDict:
    """Collect mounted partitions, disk usage and disk I/O counters."""
    partitions: list[JsonDict] = []
    seen_mountpoints: set[str] = set()

    for partition in psutil.disk_partitions(all=all_partitions):
        if partition.mountpoint in seen_mountpoints:
            continue
        seen_mountpoints.add(partition.mountpoint)
        partitions.append(
            disk_usage_item(
                device=partition.device,
                mountpoint=partition.mountpoint,
                fstype=partition.fstype,
                opts=partition.opts,
            )
        )

    if not partitions:
        partitions.append(disk_usage_item(device="root", mountpoint="/", fstype="unknown"))

    io = psutil.disk_io_counters()
    return {
        "partitions": partitions,
        "io": None
        if io is None
        else {
            **io._asdict(),
            "read_bytes_human": bytes_to_human(io.read_bytes),
            "write_bytes_human": bytes_to_human(io.write_bytes),
        },
    }


def address_family_name(family: socket.AddressFamily | int) -> str:
    """Normalize socket/psutil address family values."""
    if family == socket.AF_INET:
        return "IPv4"
    if family == socket.AF_INET6:
        return "IPv6"
    if hasattr(psutil, "AF_LINK") and family == psutil.AF_LINK:
        return "MAC"
    if hasattr(socket, "AF_PACKET") and family == socket.AF_PACKET:
        return "MAC"
    return str(family)


def collect_network() -> JsonDict:
    """Collect network addresses, interface status and traffic counters."""
    addrs = psutil.net_if_addrs()
    stats = psutil.net_if_stats()
    counters = psutil.net_io_counters(pernic=True)
    interfaces: JsonDict = {}

    for name, addresses in addrs.items():
        interfaces[name] = {
            "is_up": stats[name].isup if name in stats else None,
            "speed_mbps": stats[name].speed if name in stats else None,
            "mtu": stats[name].mtu if name in stats else None,
            "addresses": [
                {
                    "family": address_family_name(address.family),
                    "address": address.address,
                    "netmask": address.netmask,
                    "broadcast": address.broadcast,
                }
                for address in addresses
            ],
            "io": None
            if name not in counters
            else {
                **counters[name]._asdict(),
                "bytes_sent_human": bytes_to_human(counters[name].bytes_sent),
                "bytes_recv_human": bytes_to_human(counters[name].bytes_recv),
            },
        }

    total = psutil.net_io_counters()
    return {
        "interfaces": interfaces,
        "total_io": {
            **total._asdict(),
            "bytes_sent_human": bytes_to_human(total.bytes_sent),
            "bytes_recv_human": bytes_to_human(total.bytes_recv),
        },
    }


def collect_sensors() -> JsonDict:
    """Collect battery, temperature and fan information when the OS exposes it."""
    sensors: JsonDict = {"battery": None, "temperatures": {}, "fans": {}}

    if hasattr(psutil, "sensors_battery"):
        try:
            battery = psutil.sensors_battery()
        except Exception:
            battery = None
        if battery is not None:
            sensors["battery"] = {
                "percent": battery.percent,
                "power_plugged": battery.power_plugged,
                "secs_left": battery.secsleft,
                "time_left": seconds_to_human(battery.secsleft),
            }

    if hasattr(psutil, "sensors_temperatures"):
        try:
            temps = psutil.sensors_temperatures(fahrenheit=False)
            sensors["temperatures"] = {
                name: [entry._asdict() for entry in entries]
                for name, entries in temps.items()
            }
        except Exception:
            pass

    if hasattr(psutil, "sensors_fans"):
        try:
            fans = psutil.sensors_fans()
            sensors["fans"] = {
                name: [entry._asdict() for entry in entries]
                for name, entries in fans.items()
            }
        except Exception:
            pass

    return sensors


def collect_nvidia_gpus() -> list[JsonDict]:
    """Collect NVIDIA GPU information using nvidia-smi when available."""
    if not shutil.which("nvidia-smi"):
        return []

    fields = [
        "index",
        "name",
        "driver_version",
        "memory.total",
        "memory.used",
        "memory.free",
        "utilization.gpu",
        "temperature.gpu",
    ]
    command = [
        "nvidia-smi",
        f"--query-gpu={','.join(fields)}",
        "--format=csv,noheader,nounits",
    ]
    try:
        result = subprocess.run(command, capture_output=True, text=True, check=True, timeout=5)
    except (subprocess.SubprocessError, OSError):
        return []

    gpus: list[JsonDict] = []
    for row in csv.reader(result.stdout.strip().splitlines()):
        if len(row) != len(fields):
            continue
        gpu = {field: value.strip() for field, value in zip(fields, row)}
        for key in ("index", "memory.total", "memory.used", "memory.free", "utilization.gpu", "temperature.gpu"):
            try:
                gpu[key] = int(gpu[key])
            except (TypeError, ValueError):
                pass
        if isinstance(gpu.get("memory.total"), int):
            gpu["memory_total_human"] = bytes_to_human(gpu["memory.total"] * 1024 * 1024)
            gpu["memory_used_human"] = bytes_to_human(gpu["memory.used"] * 1024 * 1024)
            gpu["memory_free_human"] = bytes_to_human(gpu["memory.free"] * 1024 * 1024)
        gpus.append(gpu)
    return gpus


def collect_processes(limit: int = 5, interval: float = 0.2) -> list[JsonDict]:
    """Return top processes by CPU usage over a short sample interval."""
    processes = []
    for proc in psutil.process_iter(["pid", "name", "username", "memory_percent"]):
        try:
            proc.cpu_percent(None)
            processes.append(proc)
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue

    time.sleep(interval)
    rows: list[JsonDict] = []
    for proc in processes:
        try:
            info = proc.info
            rows.append(
                {
                    "pid": info.get("pid"),
                    "name": info.get("name"),
                    "username": info.get("username"),
                    "cpu_percent": proc.cpu_percent(None),
                    "memory_percent": round(info.get("memory_percent") or 0, 2),
                }
            )
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue

    return sorted(rows, key=lambda item: item["cpu_percent"], reverse=True)[:limit]


def build_report(args: argparse.Namespace) -> JsonDict:
    """Build the complete system report."""
    report: JsonDict = {
        "generated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
        "system": collect_system(),
        "cpu": collect_cpu(interval=args.cpu_interval),
        "memory": collect_memory(),
        "disks": collect_disks(all_partitions=args.all_partitions),
        "network": collect_network(),
        "sensors": collect_sensors(),
        "gpus": {"nvidia": collect_nvidia_gpus()},
    }
    if args.processes:
        report["top_processes"] = collect_processes(limit=args.processes)
    return report


def print_heading(title: str) -> None:
    print(f"\n{'=' * 12} {title} {'=' * 12}")


def print_text_report(report: JsonDict) -> None:
    """Print a compact human-readable report."""
    print_heading("System")
    system = report["system"]
    print(f"OS: {system['system']} {system['release']} ({system['machine']})")
    if system.get("linux_distribution"):
        print(f"Distribution: {system['linux_distribution'].get('pretty_name')}")
    print(f"Node: {system['node']}")
    print(f"Python: {system['python_version']} at {system['python_executable']}")
    print(f"Boot time: {system['boot_time']}  |  Uptime: {system['uptime']}")

    print_heading("CPU")
    cpu = report["cpu"]
    print(f"CPU: {cpu['brand']}")
    print(f"Cores: {cpu['physical_cores']} physical / {cpu['logical_cores']} logical")
    if cpu.get("current_frequency_mhz"):
        print(
            f"Frequency: {cpu['current_frequency_mhz']} MHz "
            f"(min={cpu['min_frequency_mhz']}, max={cpu['max_frequency_mhz']})"
        )
    print(f"Total usage: {cpu['total_usage_percent']}%")
    print("Per-core usage:", ", ".join(f"{value}%" for value in cpu["per_core_usage_percent"]))
    if cpu.get("load_average"):
        print("Load average:", ", ".join(f"{x:.2f}" for x in cpu["load_average"]))

    print_heading("Memory")
    mem = report["memory"]["virtual"]
    swap = report["memory"]["swap"]
    print(f"RAM: {mem['used_human']} used / {mem['total_human']} total ({mem['percent']}%)")
    print(f"Available: {mem['available_human']}")
    print(f"Swap: {swap['used_human']} used / {swap['total_human']} total ({swap['percent']}%)")

    print_heading("Disks")
    for part in report["disks"]["partitions"]:
        if "error" in part:
            print(f"{part['mountpoint']} ({part['fstype']}): {part['error']}")
        else:
            print(
                f"{part['mountpoint']} ({part['fstype']}): "
                f"{part['used_human']} used / {part['total_human']} total ({part['percent']}%)"
            )
    disk_io = report["disks"]["io"]
    if disk_io:
        print(f"Disk I/O since boot: read={disk_io['read_bytes_human']}, write={disk_io['write_bytes_human']}")

    print_heading("Network")
    for name, info in report["network"]["interfaces"].items():
        state = "up" if info["is_up"] else "down"
        addresses = [addr["address"] for addr in info["addresses"] if addr["family"] in {"IPv4", "IPv6", "MAC"}]
        print(f"{name}: {state}, speed={info['speed_mbps']} Mbps, addresses={', '.join(addresses) or 'N/A'}")
    total_io = report["network"]["total_io"]
    print(f"Network I/O since boot: sent={total_io['bytes_sent_human']}, received={total_io['bytes_recv_human']}")

    print_heading("Sensors")
    battery = report["sensors"].get("battery")
    print("Battery:", "N/A" if battery is None else f"{battery['percent']}%, plugged={battery['power_plugged']}, left={battery['time_left']}")
    print(f"Temperature sensor groups: {len(report['sensors'].get('temperatures', {}))}")
    print(f"Fan sensor groups: {len(report['sensors'].get('fans', {}))}")

    print_heading("GPU")
    gpus = report["gpus"]["nvidia"]
    if not gpus:
        print("No NVIDIA GPU found via nvidia-smi.")
    for gpu in gpus:
        print(
            f"GPU {gpu.get('index')}: {gpu.get('name')} | "
            f"util={gpu.get('utilization.gpu')}% | "
            f"memory={gpu.get('memory_used_human')}/{gpu.get('memory_total_human')} | "
            f"temp={gpu.get('temperature.gpu')}°C"
        )

    if "top_processes" in report:
        print_heading("Top Processes")
        for proc in report["top_processes"]:
            print(
                f"{proc['pid']:>6} {proc['cpu_percent']:>5.1f}% CPU "
                f"{proc['memory_percent']:>5.2f}% MEM  {proc['name']}"
            )


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Print hardware and system information.")
    parser.add_argument("--json", action="store_true", help="Print the report as JSON instead of text.")
    parser.add_argument("--output", type=Path, help="Optional path to save the JSON report.")
    parser.add_argument("--all-partitions", action="store_true", help="Include pseudo and special filesystems.")
    parser.add_argument("--processes", type=int, default=0, metavar="N", help="Include top N processes by CPU usage.")
    parser.add_argument("--cpu-interval", type=float, default=0.5, help="Seconds to sample CPU usage. Default: 0.5")
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    report = build_report(args)

    if args.output:
        args.output.write_text(json.dumps(report, indent=2), encoding="utf-8")
        print(f"Saved JSON report to {args.output}")

    if args.json:
        print(json.dumps(report, indent=2))
    else:
        print_text_report(report)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())