1648 lines
62 KiB
Python
Executable File
1648 lines
62 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
probefetch.py -- Compact system information collector.
|
|
|
|
Outputs single-line system information strings. Supports ANSI 256 / truecolor
|
|
themes and a stealth mode that avoids spawning subprocesses.
|
|
|
|
Default mode: hostname, OS, distro, CPU, GPU, processes, uptime, users,
|
|
load average, battery, memory usage, disk usage, network traffic.
|
|
|
|
Special modes (each produces its own compact line):
|
|
--devel Languages, compilers, build tools, and package managers.
|
|
--admin Administration panels, databases, monitoring, web servers.
|
|
--devops Containers, orchestration, IaC, CI/CD, cloud CLIs.
|
|
--kernel Kernel version, compiler, security modules, module count.
|
|
--pkgs Installed package count per package manager.
|
|
--security Firewall, hardening tools, access-control status.
|
|
--net Interfaces, IPs, gateway, DNS, active connections.
|
|
|
|
Flags:
|
|
--sleuth / --stealth File reads only -- no subprocess spawning.
|
|
--theme=NAME Color theme: auto, dark, light, solarized, dracula,
|
|
gruvbox, nord, mono.
|
|
|
|
All tools referenced in probe lists are free / open-source software.
|
|
No proprietary products are included.
|
|
|
|
Author: Jeremy Anderson
|
|
Website: https://git.dcos.net/dcosnet/probefetch
|
|
|
|
Based on sysinfo.pl by:
|
|
David Rudie <d.rudie@gmail.com>
|
|
Travis Morgan <imbezol@criticaldamage.com>
|
|
Nils Goers <weechatter@arcor.de>
|
|
|
|
SPDX-License-Identifier: MIT
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import platform
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Module state
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_SLEUTH_MODE: bool = False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ANSI 256 / Truecolor theme support
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_ESC = "\033"
|
|
_RESET = f"{_ESC}[0m"
|
|
_BOLD = f"{_ESC}[1m"
|
|
_DIM = f"{_ESC}[2m"
|
|
|
|
|
|
def _fg_256(n: int) -> str:
|
|
"""Return an ANSI escape for xterm-256 foreground color *n*."""
|
|
return f"{_ESC}[38;5;{n}m"
|
|
|
|
|
|
def _fg_rgb(r: int, g: int, b: int) -> str:
|
|
"""Return an ANSI escape for 24-bit truecolor foreground."""
|
|
return f"{_ESC}[38;2;{r};{g};{b}m"
|
|
|
|
|
|
def _rgb_to_256(r: int, g: int, b: int) -> int:
|
|
"""Map an RGB triplet (0-255) to the nearest xterm-256 color-cube index."""
|
|
ri = min(5, round(r / 255 * 5))
|
|
gi = min(5, round(g / 255 * 5))
|
|
bi = min(5, round(b / 255 * 5))
|
|
return 16 + 36 * ri + 6 * gi + bi
|
|
|
|
|
|
def _detect_color_mode() -> str:
|
|
"""Probe the terminal to determine the best available color mode."""
|
|
if not sys.stdout.isatty():
|
|
return "none"
|
|
ct = os.environ.get("COLORTERM", "").lower()
|
|
if ct in ("truecolor", "24bit", "yes"):
|
|
return "truecolor"
|
|
term = os.environ.get("TERM", "")
|
|
if "256color" in term:
|
|
return "256color"
|
|
if "color" in term or "ansi" in term:
|
|
return "16color"
|
|
return "none"
|
|
|
|
|
|
# Theme definitions: (label_rgb, value_rgb, separator_rgb, header_rgb).
|
|
_THEME_DEFS: dict[str, tuple] = {
|
|
"dark": ((100, 149, 237), (211, 215, 207), (88, 88, 88), (0, 255, 136)),
|
|
"light": ((30, 60, 150), (50, 50, 50), (170, 170, 170), (180, 40, 40)),
|
|
"solarized": ((38, 139, 210), (131, 148, 150), (88, 110, 117), (133, 153, 0)),
|
|
"dracula": ((189, 147, 249), (248, 248, 242), (98, 114, 164), (80, 250, 123)),
|
|
"gruvbox": ((214, 137, 16), (235, 219, 178), (146, 131, 116),(152, 195, 121)),
|
|
"nord": ((136, 192, 208), (216, 222, 233), (76, 86, 106), (163, 190, 140)),
|
|
"mono": (None, None, None, None),
|
|
}
|
|
|
|
_VALID_THEMES: frozenset[str] = frozenset(_THEME_DEFS) | frozenset({"auto"})
|
|
|
|
|
|
def _resolve_theme(name: str) -> dict[str, str]:
|
|
"""Resolve *name* into a dict of ANSI escape strings for themed output."""
|
|
if name not in _THEME_DEFS:
|
|
name = "dark"
|
|
spec = _THEME_DEFS[name]
|
|
if spec[0] is None:
|
|
return {"label": _BOLD, "value": "", "sep": _DIM, "header": _BOLD}
|
|
mode = _detect_color_mode()
|
|
if mode == "none":
|
|
return {}
|
|
label_rgb, value_rgb, sep_rgb, header_rgb = spec
|
|
|
|
def _c(rgb: tuple) -> str:
|
|
return _fg_rgb(*rgb) if mode == "truecolor" else _fg_256(_rgb_to_256(*rgb))
|
|
|
|
return {
|
|
"label": _c(label_rgb),
|
|
"value": _c(value_rgb),
|
|
"sep": _c(sep_rgb),
|
|
"header": _c(header_rgb),
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Constants
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_SHELL_TIMEOUT: int = 10
|
|
|
|
_CMD_PS: str = "ps -eo pid= | wc -l"
|
|
|
|
_RE_VERSION_NUM: re.Pattern = re.compile(r"([0-9][0-9.]*)")
|
|
_RE_UPTIME_DAYS: re.Pattern = re.compile(r"(\d+)\s+day[s]*,?\s*(\d+):(\d+)")
|
|
_RE_UPTIME_MINS: re.Pattern = re.compile(r"(\d+)\s+min")
|
|
_RE_UPTIME_HMS: re.Pattern = re.compile(r"(\d+):(\d+)")
|
|
_RE_USERS: re.Pattern = re.compile(r"(\d+)\s+user")
|
|
_RE_BOOT_SEC: re.Pattern = re.compile(r"sec\s*=\s*(\d+)")
|
|
_RE_APM_PCT: re.Pattern = re.compile(r"(\d+)%")
|
|
_RE_BATT_FULL: re.Pattern = re.compile(r"^last full capacity:\s+(\d+)")
|
|
_RE_BATT_CUR: re.Pattern = re.compile(r"^remaining capacity:\s+(\d+)")
|
|
|
|
_CLEANUP_PATTERNS: list[tuple[re.Pattern, str]] = [
|
|
(re.compile(r"\s*@\s*[\d.]*\s*GHz"), ""),
|
|
(re.compile(r"\(R\)"), ""),
|
|
(re.compile(r"\(tm\)"), ""),
|
|
(re.compile(r"\([^)]*GenuineIntel[^)]*\)"), ""),
|
|
(re.compile(r"\s*processor", re.IGNORECASE), ""),
|
|
(re.compile(r"\s*CPU", re.IGNORECASE), ""),
|
|
(re.compile(r" +"), " "),
|
|
]
|
|
|
|
# PCI vendor IDs for GPU vendors.
|
|
_GPU_VENDOR_NAMES: dict[str, str] = {
|
|
"0x10de": "NVIDIA",
|
|
"0x1002": "AMD",
|
|
"0x8086": "Intel",
|
|
"0x1a03": "ASPEED",
|
|
}
|
|
|
|
# PCI class codes that indicate a GPU.
|
|
_GPU_PCI_CLASSES: frozenset[str] = frozenset({
|
|
"0x030000", # VGA compatible controller
|
|
"0x030100", # XGA compatible controller
|
|
"0x030200", # 3D controller
|
|
})
|
|
|
|
# Ordered by specificity. First match wins.
|
|
# Strategies: keyval, num, line -- see detect_distro() docstring.
|
|
_DISTRO_TABLE: list[tuple[str, str, str]] = [
|
|
# -- Embedded / router / appliance ----------------------------------
|
|
("OpenWrt", "/etc/openwrt_release", "keyval"),
|
|
("LEDE", "/etc/lede_release", "keyval"),
|
|
("OpenWrt", "/etc/openwrt_version", "line"),
|
|
("DD-WRT", "/etc/ddwrt_release", "line"),
|
|
("Tomato", "/etc/tomato_version", "line"),
|
|
("pfSense", "/etc/pfSense-version", "line"),
|
|
("OPNsense", "/etc/opnsense-version", "line"),
|
|
("TrueNAS", "/etc/truenas_version", "line"),
|
|
("Proxmox VE", "/etc/pve/.version", "line"),
|
|
("VyOS", "/etc/vyos-version", "line"),
|
|
# -- Source-based / niche --------------------------------------------
|
|
("Source Mage", "/etc/sourcemage_version", "num"),
|
|
("Lunar Linux", "/etc/lunar-release", "line"),
|
|
("Sorcerer", "/etc/sorcerer-release", "line"),
|
|
("Gobo Linux", "/etc/gobo-release", "line"),
|
|
("LFS", "/etc/lfs-release", "line"),
|
|
("CRUX", "/etc/crux-release", "line"),
|
|
("Exherbo", "/etc/exherbo-release", "line"),
|
|
# -- Arch family ----------------------------------------------------
|
|
("Arch Linux", "/etc/arch-release", "line"),
|
|
("Manjaro Linux", "/etc/manjaro-release", "line"),
|
|
("EndeavourOS", "/etc/endeavouros-release","line"),
|
|
("Garuda Linux", "/etc/garuda-release", "line"),
|
|
# -- Gentoo family --------------------------------------------------
|
|
("Gentoo Linux", "/etc/gentoo-release", "line"),
|
|
("Funtoo Linux", "/etc/funtoo-release", "line"),
|
|
("Sabayon", "/etc/sabayon-release", "line"),
|
|
("Pentoo", "/etc/pentoo-release", "line"),
|
|
# -- Red Hat family -------------------------------------------------
|
|
("Rocky Linux", "/etc/rocky-release", "line"),
|
|
("Alma Linux", "/etc/almalinux-release", "line"),
|
|
("Oracle Linux", "/etc/oracle-release", "line"),
|
|
("Amazon Linux", "/etc/system-release", "line"),
|
|
("CentOS", "/etc/centos-release", "line"),
|
|
("Red Hat", "/etc/redhat-release", "line"),
|
|
("Scientific", "/etc/scientific-release", "line"),
|
|
# -- SUSE family ----------------------------------------------------
|
|
("SUSE", "/etc/SuSE-release", "line"),
|
|
# -- Fedora family --------------------------------------------------
|
|
("Fedora", "/etc/fedora-release", "line"),
|
|
# -- Mandriva / Mageia family ---------------------------------------
|
|
("Mageia", "/etc/mageia-release", "line"),
|
|
("OpenMandriva", "/etc/openmandriva-release","line"),
|
|
("ROSA", "/etc/ros-release", "line"),
|
|
# -- Slackware family -----------------------------------------------
|
|
("Slackware", "/etc/slackware-version", "line"),
|
|
("Sali", "/etc/sali-release", "line"),
|
|
("Zenwalk", "/etc/zenwalk-version", "line"),
|
|
("VectorLinux", "/etc/vector-version", "line"),
|
|
# -- Debian family --------------------------------------------------
|
|
("Debian", "/etc/debian_version", "num"),
|
|
("Ubuntu", "/etc/lsb-release", "keyval"),
|
|
("Linux Mint", "/etc/linuxmint-release", "line"),
|
|
("Pop!_OS", "/etc/pop-os-release", "line"),
|
|
("elementary OS", "/etc/elementary-release", "line"),
|
|
("Zorin OS", "/etc/zorin-release", "line"),
|
|
# -- Standalone -----------------------------------------------------
|
|
("Alpine Linux", "/etc/alpine-release", "line"),
|
|
("Void Linux", "/etc/void-release", "line"),
|
|
("Solus", "/etc/solus-release", "line"),
|
|
("Clear Linux", "/etc/clearlinux-release", "line"),
|
|
("PCLinuxOS", "/etc/pclinuxos-release", "line"),
|
|
("deepin", "/etc/deepin-version", "line"),
|
|
("Parrot OS", "/etc/parrot-release", "line"),
|
|
("Kali Linux", "/etc/kali-release", "line"),
|
|
("Raspberry Pi OS", "/etc/rpi-issue", "line"),
|
|
("Armbian", "/etc/armbian-release", "line"),
|
|
("MX Linux", "/etc/mx-linux-release", "line"),
|
|
("Nobara Linux", "/etc/nobara-release", "line"),
|
|
("Trisquel", "/etc/trisquel-release", "line"),
|
|
("SparkyLinux", "/etc/sparkylinux-release","line"),
|
|
("Linux Lite", "/etc/linuxlite-release", "line"),
|
|
("antiX", "/etc/antix-release", "line"),
|
|
("Peppermint", "/etc/peppermint-release", "line"),
|
|
]
|
|
|
|
_EMPTY_FILE_DISTROS: frozenset[str] = frozenset({"Arch Linux"})
|
|
|
|
_RH_CONTENT_MAP: list[tuple[str, str]] = [
|
|
("CentOS", "CentOS"),
|
|
("AlmaLinux", "Alma Linux"),
|
|
("Rocky", "Rocky Linux"),
|
|
("Oracle Linux", "Oracle Linux"),
|
|
("Scientific", "Scientific Linux"),
|
|
]
|
|
|
|
_SUSE_CONTENT_MAP: list[tuple[str, str]] = [
|
|
("SUSE Linux Enterprise Server", "SLES"),
|
|
("SUSE Linux Enterprise Desktop", "SLED"),
|
|
("openSUSE", "openSUSE"),
|
|
]
|
|
|
|
_ID_DISPLAY_NAMES: dict[str, str | None] = {
|
|
"opensuse-leap": "openSUSE Leap",
|
|
"opensuse-tumbleweed":"openSUSE Tumbleweed",
|
|
"opensuse-microos": "openSUSE MicroOS",
|
|
"linuxmint": "Linux Mint",
|
|
"pop!_os": "Pop!_OS",
|
|
"elementary os": "elementary OS",
|
|
"kali linux": "Kali Linux",
|
|
"parrot os": "Parrot OS",
|
|
"void": "Void Linux",
|
|
"alpine": "Alpine Linux",
|
|
"nixos": "NixOS",
|
|
"solus": "Solus",
|
|
"clear-linux-os": "Clear Linux OS",
|
|
"arch": "Arch Linux",
|
|
"manjaro-linux": "Manjaro Linux",
|
|
"endeavouros": "EndeavourOS",
|
|
"garuda-linux": "Garuda Linux",
|
|
"arcolinux": "Arch Linux",
|
|
"xubuntu": "Xubuntu",
|
|
"lubuntu": "Lubuntu",
|
|
"kubuntu": "Kubuntu",
|
|
"zorin-os": "Zorin OS",
|
|
"deepin": "deepin",
|
|
"mx-linux": "MX Linux",
|
|
"neon": "KDE neon",
|
|
"raspbian": "Raspbian",
|
|
"raspberry pi os": "Raspberry Pi OS",
|
|
"ubuntu": "Ubuntu",
|
|
"openwrt": "OpenWrt",
|
|
"amazon linux": "Amazon Linux",
|
|
"oracle linux": "Oracle Linux",
|
|
"rocky linux": "Rocky Linux",
|
|
"almalinux": "Alma Linux",
|
|
"almalinux ubi": "Alma Linux",
|
|
"gentoo": "Gentoo Linux",
|
|
"funtoo": "Funtoo Linux",
|
|
"sabayon": "Sabayon",
|
|
"mageia": "Mageia",
|
|
"openmandriva": "OpenMandriva",
|
|
"rosa": "ROSA",
|
|
"slackware": "Slackware",
|
|
"pclinuxos": "PCLinuxOS",
|
|
"vyos": "VyOS",
|
|
"pve": "Proxmox VE",
|
|
"opnsense": "OPNsense",
|
|
"pfsense": "pfSense",
|
|
"truenas": "TrueNAS",
|
|
"source-mage": "Source Mage",
|
|
"lunar": "Lunar Linux",
|
|
"crux": "CRUX",
|
|
"exherbo": "Exherbo",
|
|
"nobara": "Nobara Linux",
|
|
"trisquel": "Trisquel",
|
|
"sparkylinux": "SparkyLinux",
|
|
"linuxlite": "Linux Lite",
|
|
"antix": "antiX",
|
|
"peppermint": "Peppermint",
|
|
"lxle": "LXLE",
|
|
"devuan": "Devuan",
|
|
}
|
|
|
|
# Commands to probe for --devel mode.
|
|
# Each entry: (binary name, --version flag, regex to extract version, display label)
|
|
_DEVEL_PROBES: list[tuple[str, str, str, str]] = [
|
|
("python3", "--version", r"([\d.]+)", "Python"),
|
|
("node", "--version", r"v([\d.]+)", "Node.js"),
|
|
("go", "version", r"go([\d.]+)", "Go"),
|
|
("rustc", "--version", r"rustc ([\d.]+)", "Rust"),
|
|
("java", "--version", r"\"([\d.]+)", "Java"),
|
|
("php", "--version", r"([\d.]+)", "PHP"),
|
|
("ruby", "--version", r"([\d.]+)", "Ruby"),
|
|
("perl", "-v", r"v([\d.]+)", "Perl"),
|
|
("gcc", "--version", r"([\d.]+)", "GCC"),
|
|
("g++", "--version", r"([\d.]+)", "G++"),
|
|
("make", "--version", r"([\d.]+)", "Make"),
|
|
("cmake", "--version", r"([\d.]+)", "CMake"),
|
|
("git", "--version", r"git version ([\d.]+)", "Git"),
|
|
("pip3", "--version", r"pip ([\d.]+)", "pip"),
|
|
("npm", "--version", r"([\d.]+)", "npm"),
|
|
("cargo", "--version", r"cargo ([\d.]+)", "Cargo"),
|
|
]
|
|
|
|
# Commands to probe for --admin mode.
|
|
# Each entry: (binary name, display label).
|
|
# All entries are free/open-source software only.
|
|
_ADMIN_PROBES: list[tuple[str, str]] = [
|
|
("cockpit-ws", "Cockpit"),
|
|
("mariadb", "MariaDB"),
|
|
("mysql", "MySQL"),
|
|
("psql", "PostgreSQL"),
|
|
("phpmyadmin", "phpMyAdmin"),
|
|
("prometheus", "Prometheus"),
|
|
("grafana-server", "Grafana"),
|
|
("node_exporter", "node_exporter"),
|
|
("nginx", "nginx"),
|
|
("apache2", "Apache"),
|
|
("httpd", "Apache"),
|
|
("caddy", "Caddy"),
|
|
("redis-server", "Redis"),
|
|
("rabbitmqctl", "RabbitMQ"),
|
|
("postgres", "PostgreSQL"),
|
|
("lighttpd", "Lighttpd"),
|
|
("traefik", "Traefik"),
|
|
]
|
|
|
|
# Commands to probe for --devops mode.
|
|
# All entries are free/open-source software only.
|
|
_DEVOPS_PROBES: list[tuple[str, str]] = [
|
|
("docker", "Docker"),
|
|
("podman", "Podman"),
|
|
("kubectl", "kubectl"),
|
|
("helm", "Helm"),
|
|
("tofu", "OpenTofu"),
|
|
("ansible", "Ansible"),
|
|
("puppet", "Puppet"),
|
|
("jenkins", "Jenkins"),
|
|
("gitlab-runner", "GitLab Runner"),
|
|
("aws", "AWS CLI"),
|
|
("az", "Azure CLI"),
|
|
("gcloud", "gcloud CLI"),
|
|
("gh", "GitHub CLI"),
|
|
("vagrant", "Vagrant"),
|
|
("skopeo", "Skopeo"),
|
|
("buildah", "Buildah"),
|
|
("nerdctl", "nerdctl"),
|
|
("crictl", "crictl"),
|
|
("lima", "Lima"),
|
|
("colima", "Colima"),
|
|
("fly", "Fly CLI"),
|
|
("doctl", "DigitalOcean CLI"),
|
|
("linode-cli", "Linode CLI"),
|
|
]
|
|
|
|
# Commands to probe for --security mode.
|
|
_SECURITY_PROBES: list[tuple[str, str]] = [
|
|
# Firewalls
|
|
("ufw", "UFW"),
|
|
("firewalld", "firewalld"),
|
|
("nft", "nftables"),
|
|
("iptables", "iptables"),
|
|
# Network security & observability
|
|
("opensnitchd", "OpenSnitch"),
|
|
("cilium", "Cilium"),
|
|
("hubble", "Hubble"),
|
|
# eBPF tooling
|
|
("bpftool", "bpftool"),
|
|
("bpftrace", "bpftrace"),
|
|
("bcc-ls", "BCC"),
|
|
("tcpreplay", "tcpreplay"),
|
|
("trace-cmd", "trace-cmd"),
|
|
# Host IDS / integrity
|
|
("fail2ban-client","Fail2Ban"),
|
|
("auditd", "auditd"),
|
|
("rkhunter", "rkhunter"),
|
|
("lynis", "Lynis"),
|
|
("chkrootkit", "chkrootkit"),
|
|
("ossec-control", "OSSEC"),
|
|
("samhain", "Samhain"),
|
|
("aide", "AIDE"),
|
|
("tripwire", "Tripwire"),
|
|
# Anti-malware
|
|
("clamscan", "ClamAV"),
|
|
("maldet", "Linux MalDetect"),
|
|
# Mandatory access control
|
|
("apparmor_parser","AppArmor"),
|
|
("sestatus", "SELinux"),
|
|
# Authentication & PAM
|
|
("sssd", "SSSD"),
|
|
("pam_tally2", "PAM"),
|
|
("faillock", "faillock"),
|
|
# System hardening
|
|
("tuned", "tuned"),
|
|
("hardened_malloc","hardened_malloc"),
|
|
# Sandbox / isolation
|
|
("firejail", "Firejail"),
|
|
("bubblewrap", "bubblewrap"),
|
|
("sandstorm", "Sandstorm"),
|
|
# DNS-level security
|
|
("dnscrypt-proxy", "dnscrypt-proxy"),
|
|
("pihole-FTL", "Pi-hole"),
|
|
# VPN & tunnel security
|
|
("wireguard", "WireGuard"),
|
|
("wg-quick", "WireGuard"),
|
|
("tunctl", "TUN/TAP"),
|
|
# Key / certificate management
|
|
("certbot", "certbot"),
|
|
("acme.sh", "acme.sh"),
|
|
("gpg", "GPG"),
|
|
# Rootkit hunters already listed above (rkhunter, chkrootkit)
|
|
]
|
|
|
|
# Package manager package-count commands.
|
|
_PKG_COUNTERS: list[tuple[str, str, str]] = [
|
|
("dpkg-query -f '.\n' -W 2>/dev/null | wc -l", "dpkg"),
|
|
("rpm -qa 2>/dev/null | wc -l", "rpm"),
|
|
("pacman -Q 2>/dev/null | wc -l", "pacman"),
|
|
("apk info 2>/dev/null | wc -l", "apk"),
|
|
("equery list '*' 2>/dev/null | wc -l", "emerge"),
|
|
("xbps-query -l 2>/dev/null | wc -l", "xbps"),
|
|
("nix-store -q --requisites /run/current-system 2>/dev/null | wc -l", "nix"),
|
|
("dnf list installed 2>/dev/null | wc -l", "dnf"),
|
|
("zypper search -i '' 2>/dev/null | wc -l", "zypper"),
|
|
]
|
|
|
|
# Map CLI section names to ProbeFetchConfig attribute names.
|
|
_SECTION_TO_ATTR: dict[str, str] = {
|
|
"hostname": "show_hostname",
|
|
"os": "show_os",
|
|
"distro": "show_distro",
|
|
"cpu": "show_cpu",
|
|
"gpu": "show_gpu",
|
|
"processes": "show_processes",
|
|
"uptime": "show_uptime",
|
|
"loadaverage": "show_load_average",
|
|
"battery": "show_battery",
|
|
"memory": "show_memory_usage",
|
|
"disk": "show_disk_usage",
|
|
"network": "show_network_traffic",
|
|
"users": "show_users",
|
|
}
|
|
|
|
_VALID_SECTIONS: frozenset[str] = frozenset(_SECTION_TO_ATTR)
|
|
|
|
# Special mode flags that produce their own output line.
|
|
_SPECIAL_FLAGS: frozenset[str] = frozenset({
|
|
"--devel", "--admin", "--devops", "--kernel", "--pkgs",
|
|
"--security", "--net",
|
|
})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Configuration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@dataclass(frozen=True)
|
|
class ProbeFetchConfig:
|
|
"""Toggle and label each information section."""
|
|
|
|
show_hostname: bool = True
|
|
use_short_hostname: bool = True
|
|
show_distro: bool = True
|
|
show_os: bool = True
|
|
show_users: bool = True
|
|
show_cpu: bool = True
|
|
show_gpu: bool = True
|
|
show_processes: bool = True
|
|
show_uptime: bool = True
|
|
show_load_average: bool = True
|
|
show_battery: bool = False
|
|
show_memory_usage: bool = True
|
|
show_disk_usage: bool = True
|
|
show_network_traffic: bool = False
|
|
|
|
text_hostname: str = "Host: "
|
|
text_distro: str = "Distro: "
|
|
text_os: str = "OS: "
|
|
text_users: str = "Users: "
|
|
text_cpu: str = "CPU: "
|
|
text_gpu: str = "GPU: "
|
|
text_processes: str = "Processes: "
|
|
text_uptime: str = "Uptime: "
|
|
text_load_average: str = "Load Average: "
|
|
text_battery: str = "Battery: "
|
|
text_memory_usage: str = "Memory Usage: "
|
|
text_disk_usage: str = "Disk Usage: "
|
|
text_network: str = "Network: "
|
|
separator: str = " - "
|
|
nic: str = ""
|
|
nicname: str = ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Utility helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _run(cmd: str | list[str], *, shell: bool = False) -> str:
|
|
"""Execute a command and return stripped stdout.
|
|
|
|
Returns empty string on any failure. When ``_SLEUTH_MODE`` is active
|
|
no subprocesses are spawned -- only file-based probes succeed.
|
|
"""
|
|
if _SLEUTH_MODE:
|
|
return ""
|
|
try:
|
|
result = subprocess.run(
|
|
cmd,
|
|
shell=shell,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=_SHELL_TIMEOUT,
|
|
)
|
|
return result.stdout.strip()
|
|
except (subprocess.TimeoutExpired, OSError, ValueError):
|
|
return ""
|
|
|
|
|
|
def _which(name: str) -> str:
|
|
"""Return the full path to *name*, or empty string if not found."""
|
|
return shutil.which(name) or ""
|
|
|
|
|
|
def _read_file_lines(path: str | Path) -> list[str]:
|
|
"""Read a text file into lines. Returns an empty list on any failure."""
|
|
try:
|
|
with open(path, "r", encoding="utf-8", errors="replace") as fh:
|
|
return fh.readlines()
|
|
except OSError:
|
|
return []
|
|
|
|
|
|
def _file_contains_line_with(path: str | Path, prefix: str) -> str:
|
|
"""Return the first line in *path* starting with *prefix*, stripped."""
|
|
for line in _read_file_lines(path):
|
|
stripped = line.strip()
|
|
if stripped.startswith(prefix):
|
|
return stripped
|
|
return ""
|
|
|
|
|
|
def _safe_int(value: str, *, base: int = 10) -> int:
|
|
"""Convert *value* to int. Returns 0 on any conversion failure."""
|
|
try:
|
|
return int(value, base)
|
|
except (ValueError, TypeError):
|
|
return 0
|
|
|
|
|
|
def _safe_int_hex(value: str) -> int:
|
|
"""Convert a hex string like '0x10de' to int. Returns 0 on failure."""
|
|
try:
|
|
return int(value, 16)
|
|
except (ValueError, TypeError):
|
|
return 0
|
|
|
|
|
|
def _safe_float(value: str) -> float:
|
|
"""Convert *value* to float. Returns 0.0 on any conversion failure."""
|
|
try:
|
|
return float(value)
|
|
except (ValueError, TypeError):
|
|
return 0.0
|
|
|
|
|
|
def _extract_version(text: str) -> str:
|
|
"""Extract the first version-like digit sequence from *text*."""
|
|
m = _RE_VERSION_NUM.search(text)
|
|
return m.group(1) if m else ""
|
|
|
|
|
|
def _apply_cleanup(text: str) -> str:
|
|
"""Strip cosmetic artifacts from CPU description strings."""
|
|
for pattern, replacement in _CLEANUP_PATTERNS:
|
|
text = pattern.sub(replacement, text)
|
|
return text.strip()
|
|
|
|
|
|
def _count_processor_lines(lines: list[str]) -> int:
|
|
"""Count lines beginning with ``processor`` in /proc/cpuinfo."""
|
|
return sum(1 for ln in lines if ln.startswith("processor"))
|
|
|
|
|
|
def _probe_version(binary: str, flag: str, pattern: str) -> str:
|
|
"""Run *binary* *flag*, extract version via *pattern*. Returns empty on failure."""
|
|
path = _which(binary)
|
|
if not path:
|
|
return ""
|
|
output = _run([binary, flag] if not flag.startswith("-") else [binary, flag])
|
|
if not output:
|
|
return ""
|
|
m = re.search(pattern, output)
|
|
return m.group(1) if m else ""
|
|
|
|
|
|
def _probe_present(binaries: list[tuple[str, str]]) -> list[str]:
|
|
"""Return display names of all binaries found on PATH."""
|
|
seen: set[str] = set()
|
|
results: list[str] = []
|
|
for binary, label in binaries:
|
|
if _which(binary) and label not in seen:
|
|
seen.add(label)
|
|
results.append(label)
|
|
return results
|
|
|
|
|
|
def _read_sysfs_int(path: str) -> int:
|
|
"""Read a single integer from a sysfs file. Returns 0 on any failure."""
|
|
try:
|
|
with open(path, "r") as fh:
|
|
return int(fh.read().strip())
|
|
except (OSError, ValueError):
|
|
return 0
|
|
|
|
|
|
def _read_sysfs_str(path: str) -> str:
|
|
"""Read a single string from a sysfs file. Returns empty on any failure."""
|
|
try:
|
|
with open(path, "r") as fh:
|
|
return fh.read().strip()
|
|
except OSError:
|
|
return ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Distro detection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _parse_os_release(path: str) -> tuple[str, str, str]:
|
|
"""Parse an os-release file. Returns (name, version, pretty_name)."""
|
|
fields: dict[str, str] = {}
|
|
for line in _read_file_lines(path):
|
|
line = line.strip()
|
|
if "=" not in line:
|
|
continue
|
|
key, _, value = line.partition("=")
|
|
fields[key.strip()] = value.strip().strip('"')
|
|
|
|
name = fields.get("NAME", "")
|
|
version = fields.get("VERSION", "")
|
|
version_id = fields.get("VERSION_ID", "")
|
|
pretty = fields.get("PRETTY_NAME", "")
|
|
|
|
if version_id and (not version or version == pretty):
|
|
version = version_id
|
|
|
|
return name, version, pretty
|
|
|
|
|
|
def _refine_rh_name(raw: str) -> str:
|
|
"""Determine the correct Red Hat derivative name from file content."""
|
|
for needle, display in _RH_CONTENT_MAP:
|
|
if needle in raw:
|
|
return display
|
|
return "Red Hat"
|
|
|
|
|
|
def _refine_suse_name(raw: str) -> tuple[str, str]:
|
|
"""Determine the correct SUSE product name and version from file content."""
|
|
for needle, display in _SUSE_CONTENT_MAP:
|
|
if needle in raw:
|
|
return display, _extract_version(raw)
|
|
return "SUSE", _extract_version(raw)
|
|
|
|
|
|
def _detect_distro_fallback() -> tuple[str, str]:
|
|
"""Identify a distro by probing release files in specificity order."""
|
|
for distro_name, release_file, strategy in _DISTRO_TABLE:
|
|
if not os.path.isfile(release_file):
|
|
continue
|
|
|
|
content = _read_file_lines(release_file)
|
|
raw = content[0].strip() if content else ""
|
|
|
|
if strategy == "keyval":
|
|
fields: dict[str, str] = {}
|
|
for line in content:
|
|
line = line.strip()
|
|
if "=" not in line:
|
|
continue
|
|
key, _, value = line.partition("=")
|
|
fields[key.strip()] = value.strip().strip('"')
|
|
kid = fields.get("DISTRIB_ID", "")
|
|
if kid:
|
|
return distro_name if distro_name not in _EMPTY_FILE_DISTROS else kid, fields.get("DISTRIB_RELEASE", "")
|
|
return distro_name, ""
|
|
|
|
if strategy == "num":
|
|
return distro_name, _extract_version(raw)
|
|
|
|
if distro_name in _EMPTY_FILE_DISTROS:
|
|
return distro_name, ""
|
|
|
|
version = raw
|
|
|
|
if distro_name == "Red Hat":
|
|
distro_name = _refine_rh_name(raw)
|
|
version = _extract_version(raw)
|
|
elif distro_name == "SUSE":
|
|
distro_name, version = _refine_suse_name(raw)
|
|
elif distro_name in ("Amazon Linux", "Fedora"):
|
|
version = _extract_version(raw)
|
|
elif distro_name == "Debian":
|
|
lsb_desc = _file_contains_line_with("/etc/lsb-release", "DISTRIB_DESCRIPTION=")
|
|
if lsb_desc:
|
|
desc_val = lsb_desc.partition("=")[2].strip().strip('"')
|
|
m = re.match(r"^(\S+)\s+(.+)$", desc_val)
|
|
if m:
|
|
return m.group(1), m.group(2)
|
|
issue_lines = _read_file_lines("/etc/issue")
|
|
if issue_lines:
|
|
issue = issue_lines[0].strip()
|
|
if "Raspbian" in issue:
|
|
return "Raspbian", _extract_version(issue)
|
|
if "OSMC" in issue:
|
|
return "OSMC", ""
|
|
version = _extract_version(raw)
|
|
else:
|
|
version = _extract_version(raw)
|
|
|
|
return distro_name, version
|
|
|
|
issue_lines = _read_file_lines("/etc/issue")
|
|
if issue_lines:
|
|
m = re.match(r"^(\S+(?:\s+\S+){0,2})", issue_lines[0].strip())
|
|
if m:
|
|
return m.group(1), ""
|
|
|
|
return "", ""
|
|
|
|
|
|
def detect_distro(is_linux: bool, show: bool) -> str:
|
|
"""Identify the Linux distribution name and version.
|
|
|
|
Detection layers:
|
|
1. /etc/os-release (systemd standard)
|
|
2. /usr/lib/os-release (chroots, containers, WSL)
|
|
3. Release-file table (60+ distros by file fingerprint)
|
|
4. /etc/issue (terminal banner, last resort)
|
|
"""
|
|
if not is_linux or not show:
|
|
return ""
|
|
|
|
name, version, pretty = _parse_os_release("/etc/os-release")
|
|
if not name:
|
|
name, version, pretty = _parse_os_release("/usr/lib/os-release")
|
|
|
|
if name:
|
|
key = name.lower().strip()
|
|
mapped = _ID_DISPLAY_NAMES.get(key)
|
|
if mapped is not None:
|
|
name = mapped
|
|
if not version and pretty:
|
|
m = re.match(r"^(.+?)\s+([0-9][0-9.]*\d.*)$", pretty)
|
|
if m:
|
|
version = m.group(2).strip()
|
|
if not pretty.startswith(name):
|
|
name = m.group(1).strip()
|
|
return f"{name} {version}" if version else name
|
|
|
|
name, version = _detect_distro_fallback()
|
|
return f"{name} {version}" if name and version else (name if name else "")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Data collector
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class ProbeFetchCollector:
|
|
"""Gathers system information. Each public method returns a display-ready
|
|
string, or ``""`` when the value is unavailable or the section is disabled.
|
|
"""
|
|
|
|
def __init__(self, cfg: Optional[ProbeFetchConfig] = None) -> None:
|
|
self.cfg = cfg or ProbeFetchConfig()
|
|
|
|
self.os_name: str = platform.system()
|
|
self.os_release: str = platform.release()
|
|
self.os_machine: str = platform.machine()
|
|
self.uname_str: str = f"{self.os_name} {self.os_release}/{self.os_machine}"
|
|
|
|
hostname_full = platform.node() or "unknown"
|
|
self.hostname: str = (
|
|
hostname_full.split(".")[0]
|
|
if self.cfg.use_short_hostname
|
|
else hostname_full
|
|
)
|
|
|
|
self.is_linux: bool = self.os_name == "Linux"
|
|
self.is_bsd: bool = self.os_name in ("FreeBSD", "DragonFly", "NetBSD", "OpenBSD")
|
|
self.is_darwin: bool = self.os_name == "Darwin"
|
|
self.is_sun: bool = self.os_name == "SunOS"
|
|
|
|
self._cpuinfo_lines: list[str] = []
|
|
self._meminfo_lines: list[str] = []
|
|
self._netdev_lines: list[str] = []
|
|
self._uptime_seconds: float = 0.0
|
|
|
|
if self.is_linux:
|
|
self._cpuinfo_lines = _read_file_lines("/proc/cpuinfo")
|
|
self._meminfo_lines = _read_file_lines("/proc/meminfo")
|
|
self._netdev_lines = _read_file_lines("/proc/net/dev")
|
|
uptime_lines = _read_file_lines("/proc/uptime")
|
|
if uptime_lines:
|
|
self._uptime_seconds = _safe_float(uptime_lines[0].strip().split()[0])
|
|
|
|
# ---- Distro ---------------------------------------------------------
|
|
|
|
def get_distro(self) -> str:
|
|
return detect_distro(self.is_linux, self.cfg.show_distro)
|
|
|
|
# ---- CPU -------------------------------------------------------------
|
|
|
|
def _cpuinfo_field(self, pattern: str) -> str:
|
|
for line in self._cpuinfo_lines:
|
|
m = re.match(rf"^{pattern}\s*:\s*(.+)", line)
|
|
if m:
|
|
return m.group(1).strip()
|
|
return ""
|
|
|
|
def _cpuinfo_int(self, pattern: str) -> int:
|
|
return _safe_int(self._cpuinfo_field(pattern))
|
|
|
|
def _cpuinfo_float(self, pattern: str) -> float:
|
|
return _safe_float(self._cpuinfo_field(pattern))
|
|
|
|
def get_cpu(self) -> str:
|
|
if not self.cfg.show_cpu:
|
|
return ""
|
|
if not self.is_linux:
|
|
return self._get_cpu_nonlinux()
|
|
return self._get_cpu_linux()
|
|
|
|
def _get_cpu_nonlinux(self) -> str:
|
|
if self.is_darwin:
|
|
brand = _run(["sysctl", "-n", "machdep.cpu.brand_string"])
|
|
if not brand:
|
|
return ""
|
|
freq_int = _safe_int(_run(["sysctl", "-n", "hw.cpufrequency"]))
|
|
mhz = f"{freq_int / 1_000_000:.2f} MHz" if freq_int > 0 else ""
|
|
cores = _safe_int(_run(["sysctl", "-n", "hw.ncpu"]))
|
|
cpu_str = f"{brand} ({mhz})" if mhz else brand
|
|
return f"{cores} x {cpu_str}" if cores > 1 else cpu_str
|
|
if self.is_bsd:
|
|
model = _run(["sysctl", "-n", "hw.model"])
|
|
if not model:
|
|
return ""
|
|
cores = _safe_int(_run(["sysctl", "-n", "hw.ncpu"]))
|
|
return f"{cores} x {model}" if cores > 1 else model
|
|
return ""
|
|
|
|
def _get_cpu_linux(self) -> str:
|
|
machine = self.os_machine
|
|
cpu = ""
|
|
mhz = ""
|
|
smp_count = 0
|
|
|
|
if machine in ("i586", "i686", "x86_64", "amd64"):
|
|
cpu = self._cpuinfo_field(r"model name")
|
|
mhz_raw = self._cpuinfo_float(r"cpu MHz")
|
|
mhz = f"{mhz_raw:.2f} MHz" if mhz_raw > 0 else ""
|
|
cpu = f"{cpu} ({mhz})" if cpu and mhz else cpu
|
|
smp_count = _count_processor_lines(self._cpuinfo_lines)
|
|
|
|
elif machine in ("armv6l", "armv7l"):
|
|
cpu = self._cpuinfo_field(r"model name") or self._cpuinfo_field(r"Processor")
|
|
cpu = re.sub(r"-compatible", "", cpu)
|
|
cpu = re.sub(r"^processor\s*", "", cpu)
|
|
freq_path = "/sys/bus/cpu/devices/cpu0/cpufreq/scaling_cur_freq"
|
|
if os.path.isfile(freq_path):
|
|
freq_lines = _read_file_lines(freq_path)
|
|
if freq_lines:
|
|
khz = _safe_int(freq_lines[0].strip())
|
|
if khz > 0:
|
|
cpu = f"{cpu} ({khz / 1000:.2f} MHz)"
|
|
smp_count = _count_processor_lines(self._cpuinfo_lines)
|
|
|
|
elif machine == "arm":
|
|
cpu = self._cpuinfo_field(r"Processor")
|
|
|
|
elif machine == "alpha":
|
|
cpu = self._cpuinfo_field(r"cpu")
|
|
model = self._cpuinfo_field(r"cpu model")
|
|
sys_type = self._cpuinfo_field(r"system type")
|
|
freq_hz = self._cpuinfo_int(r"cycle frequency \[Hz\]")
|
|
mhz = f"{freq_hz / 1_000_000:.2f} MHz" if freq_hz > 0 else ""
|
|
cpu = f"{cpu} {model} ({sys_type}) ({mhz})" if model else (f"{cpu} ({mhz})" if mhz else cpu)
|
|
smp_count = self._cpuinfo_int(r"cpus detected")
|
|
|
|
elif machine == "ia64":
|
|
vendor = self._cpuinfo_field(r"vendor")
|
|
family = self._cpuinfo_field(r"family")
|
|
mhz_raw = self._cpuinfo_float(r"cpu MHz")
|
|
mhz = f"{mhz_raw:.2f} MHz" if mhz_raw > 0 else ""
|
|
cpu = f"{vendor} {family} ({mhz})" if mhz else f"{vendor} {family}"
|
|
smp_count = _count_processor_lines(self._cpuinfo_lines)
|
|
|
|
elif machine == "mips":
|
|
cpu_name = self._cpuinfo_field(r"cpu")
|
|
model = self._cpuinfo_field(r"cpu model")
|
|
cpu = f"{cpu_name} {model}" if model else cpu_name
|
|
|
|
elif machine in ("parisc", "parisc64"):
|
|
cpu_name = self._cpuinfo_field(r"cpu")
|
|
model = self._cpuinfo_field(r"model name")
|
|
mhz_raw = self._cpuinfo_float(r"cpu MHz")
|
|
mhz = f"{mhz_raw:.2f} MHz" if mhz_raw > 0 else ""
|
|
cpu = f"{model} {cpu_name} ({mhz})" if mhz else f"{model} {cpu_name}"
|
|
smp_count = _count_processor_lines(self._cpuinfo_lines)
|
|
|
|
elif machine in ("ppc", "ppc64"):
|
|
cpu_name = self._cpuinfo_field(r"cpu")
|
|
clock = self._cpuinfo_field(r"clock")
|
|
clock = re.sub(r"^(\d+\.\d{3})\d*\s*MHz", r"\1 MHz", clock)
|
|
cpu_name = re.sub(r", altivec supported", "", cpu_name) if cpu_name else cpu_name
|
|
model = (
|
|
"IBM PowerPC G5" if cpu_name and re.match(r"^(PPC)*9\.", cpu_name)
|
|
else "Motorola PowerPC G4" if cpu_name and re.match(r"^74\.", cpu_name)
|
|
else "IBM PowerPC G3"
|
|
)
|
|
cpu = f"{model} {cpu_name} ({clock})" if clock else f"{model} {cpu_name}"
|
|
smp_count = _count_processor_lines(self._cpuinfo_lines)
|
|
|
|
elif machine in ("s390", "s390x"):
|
|
cpu = self._cpuinfo_field(r"vendor_id")
|
|
smp_count = self._cpuinfo_int(r"processors")
|
|
|
|
elif machine.startswith("sh"):
|
|
cpu_family = self._cpuinfo_field(r"cpu family")
|
|
cpu_type = self._cpuinfo_field(r"cpu type")
|
|
clk = self._cpuinfo_field(r"cpu_clk")
|
|
cpu = f"{cpu_family} {cpu_type} ({clk} MHz)" if clk else f"{cpu_family} {cpu_type}"
|
|
|
|
elif machine in ("sparc", "sparc64"):
|
|
cpu_name = self._cpuinfo_field(r"cpu")
|
|
cpu_type = self._cpuinfo_field(r"type")
|
|
clk_raw = self._cpuinfo_field(r"Cpu0ClkTck")
|
|
mhz = ""
|
|
if clk_raw:
|
|
try:
|
|
mhz = f"{int(clk_raw, 16) / 1_000_000:.2f} MHz"
|
|
except ValueError:
|
|
pass
|
|
cpu = f"{cpu_type} {cpu_name} ({mhz})" if mhz else f"{cpu_type} {cpu_name}"
|
|
smp_count = self._cpuinfo_int(r"ncpus active")
|
|
|
|
else:
|
|
cpu = self._cpuinfo_field(r"model name") or self._cpuinfo_field(r"Processor") or ""
|
|
|
|
if smp_count > 1 and cpu:
|
|
cpu = f"{smp_count} x {cpu}"
|
|
if cpu:
|
|
cpu = _apply_cleanup(cpu)
|
|
return cpu
|
|
|
|
# ---- GPU -------------------------------------------------------------
|
|
|
|
def get_gpu(self) -> str:
|
|
"""Detect GPU via nvidia-smi, sysfs, or lspci. Returns compact string."""
|
|
if not self.cfg.show_gpu:
|
|
return ""
|
|
|
|
gpus: list[str] = []
|
|
|
|
# Strategy 1: nvidia-smi (name, VRAM, CUDA cores in one query).
|
|
if _which("nvidia-smi"):
|
|
smi = _run(
|
|
"nvidia-smi --query-gpu=name,memory.total,count_of_cuda_cores"
|
|
" --format=csv,noheader,nounits",
|
|
shell=True,
|
|
)
|
|
if smi:
|
|
for line in smi.splitlines():
|
|
parts = [p.strip() for p in line.split(",")]
|
|
if len(parts) >= 3:
|
|
gpus.append(
|
|
f"{parts[0]} {parts[1]}MB ({parts[2]} CUDA cores)"
|
|
)
|
|
elif len(parts) >= 2:
|
|
gpus.append(f"{parts[0]} ({parts[1]}MB)")
|
|
if gpus:
|
|
return ", ".join(gpus)
|
|
|
|
# Strategy 2: sysfs PCI device tree (file reads, works in sleuth mode).
|
|
if self.is_linux and not gpus:
|
|
gpus = self._gpu_from_sysfs()
|
|
|
|
# Strategy 3: lspci fallback.
|
|
if self.is_linux and not gpus and _which("lspci"):
|
|
lspci_out = _run("lspci -mm | grep -i 'vga\\|3d\\|display'", shell=True)
|
|
if lspci_out:
|
|
for line in lspci_out.splitlines():
|
|
parts = line.split("\t")
|
|
if len(parts) >= 3:
|
|
gpus.append(parts[2].strip('"'))
|
|
|
|
return ", ".join(gpus) if gpus else ""
|
|
|
|
def _gpu_from_sysfs(self) -> list[str]:
|
|
"""Walk /sys/bus/pci/devices for GPU class codes and read VRAM."""
|
|
results: list[str] = []
|
|
pci_base = Path("/sys/bus/pci/devices")
|
|
if not pci_base.is_dir():
|
|
return results
|
|
|
|
for device_dir in pci_base.iterdir():
|
|
class_hex = _read_sysfs_str(str(device_dir / "class"))
|
|
if class_hex not in _GPU_PCI_CLASSES:
|
|
continue
|
|
|
|
vendor_hex = _read_sysfs_str(str(device_dir / "vendor"))
|
|
vendor_name = _GPU_VENDOR_NAMES.get(vendor_hex, "Unknown")
|
|
|
|
vram_str = ""
|
|
if vendor_hex in ("0x10de", "0x1002"):
|
|
vram_bytes = _read_sysfs_int(str(device_dir / "mem_info_vram_total"))
|
|
if vram_bytes > 0:
|
|
vram_mb = vram_bytes // 1048576
|
|
vram_str = f" ({vram_mb}MB VRAM)"
|
|
|
|
results.append(vendor_name + vram_str)
|
|
|
|
return results
|
|
|
|
# ---- Processes -------------------------------------------------------
|
|
|
|
def get_processes(self) -> str:
|
|
if not self.cfg.show_processes:
|
|
return ""
|
|
result = _run(_CMD_PS, shell=True)
|
|
return result if result else "0"
|
|
|
|
# ---- Uptime ----------------------------------------------------------
|
|
|
|
def get_uptime(self) -> str:
|
|
if not self.cfg.show_uptime:
|
|
return ""
|
|
total_seconds = self._uptime_seconds
|
|
|
|
if not total_seconds and self.is_darwin:
|
|
boot_raw = _run(["sysctl", "-n", "kern.boottime"])
|
|
m = _RE_BOOT_SEC.search(boot_raw)
|
|
if m:
|
|
total_seconds = time.time() - _safe_int(m.group(1))
|
|
|
|
if not total_seconds and self.is_bsd:
|
|
boot_raw = _run("sysctl -n kern.boottime | awk '{print $4}'", shell=True)
|
|
total_seconds = time.time() - _safe_int(boot_raw)
|
|
|
|
if not total_seconds and self.is_sun:
|
|
uptime_str = _run("uptime", shell=True)
|
|
m = _RE_UPTIME_DAYS.search(uptime_str)
|
|
if m:
|
|
total_seconds = _safe_int(m.group(1)) * 86400 + _safe_int(m.group(2)) * 3600 + _safe_int(m.group(3)) * 60
|
|
else:
|
|
m2 = _RE_UPTIME_MINS.search(uptime_str)
|
|
if m2:
|
|
total_seconds = _safe_int(m2.group(1)) * 60
|
|
else:
|
|
m3 = _RE_UPTIME_HMS.search(uptime_str)
|
|
if m3:
|
|
total_seconds = _safe_int(m3.group(1)) * 3600 + _safe_int(m3.group(2)) * 60
|
|
|
|
return self._format_uptime(total_seconds)
|
|
|
|
@staticmethod
|
|
def _format_uptime(total_seconds: float) -> str:
|
|
days = int(total_seconds // 86400)
|
|
remainder = total_seconds % 86400
|
|
hours = int(remainder // 3600)
|
|
minutes = int((remainder % 3600) // 60)
|
|
parts: list[str] = []
|
|
if days >= 1:
|
|
parts.append(f"{days}d")
|
|
if hours >= 1:
|
|
parts.append(f"{hours}h")
|
|
if minutes >= 1:
|
|
parts.append(f"{minutes}m")
|
|
return " ".join(parts) if parts else "0m"
|
|
|
|
# ---- Users -----------------------------------------------------------
|
|
|
|
def get_users(self) -> str:
|
|
if not self.cfg.show_users:
|
|
return ""
|
|
uptime_str = _run("uptime", shell=True)
|
|
m = _RE_USERS.search(uptime_str)
|
|
return m.group(1) if m else "0"
|
|
|
|
# ---- Load Average ----------------------------------------------------
|
|
|
|
def get_load_average(self) -> str:
|
|
if not self.cfg.show_load_average:
|
|
return ""
|
|
try:
|
|
load1, _, _ = os.getloadavg()
|
|
return f"{load1:.2f}"
|
|
except OSError:
|
|
return "0.00"
|
|
|
|
# ---- Battery ---------------------------------------------------------
|
|
|
|
def get_battery(self) -> str:
|
|
if not self.cfg.show_battery:
|
|
return ""
|
|
if os.path.isfile("/proc/apm"):
|
|
lines = _read_file_lines("/proc/apm")
|
|
if lines:
|
|
m = _RE_APM_PCT.search(lines[0])
|
|
if m:
|
|
return f"{m.group(1)}%"
|
|
|
|
batt_dir = Path("/proc/acpi/battery")
|
|
if not batt_dir.is_dir():
|
|
return ""
|
|
|
|
results: list[str] = []
|
|
try:
|
|
for entry in sorted(batt_dir.iterdir()):
|
|
if not entry.is_dir() or entry.name.startswith("."):
|
|
continue
|
|
bfull = 0
|
|
bcur = 0
|
|
info_line = _file_contains_line_with(str(entry / "info"), "last full capacity:")
|
|
if info_line:
|
|
m = _RE_BATT_FULL.match(info_line)
|
|
if m:
|
|
bfull = _safe_int(m.group(1))
|
|
state_line = _file_contains_line_with(str(entry / "state"), "remaining capacity:")
|
|
if state_line:
|
|
m = _RE_BATT_CUR.match(state_line)
|
|
if m:
|
|
bcur = _safe_int(m.group(1))
|
|
if bfull > 0:
|
|
results.append(f"{bcur / bfull * 100:.0f}%")
|
|
except OSError:
|
|
pass
|
|
return " ".join(results) if results else ""
|
|
|
|
# ---- Memory Usage ----------------------------------------------------
|
|
|
|
def _meminfo_kib(self, key: str) -> int:
|
|
for line in self._meminfo_lines:
|
|
if line.startswith(key):
|
|
parts = line.split()
|
|
if len(parts) >= 2:
|
|
return _safe_int(parts[1])
|
|
return 0
|
|
|
|
def get_memory_usage(self) -> str:
|
|
if not self.cfg.show_memory_usage:
|
|
return ""
|
|
total_bytes: int = 0
|
|
used_bytes: int = 0
|
|
|
|
if self.is_linux:
|
|
mem_total = self._meminfo_kib("MemTotal:")
|
|
total_bytes = mem_total * 1024
|
|
# Prefer MemAvailable (kernel >= 3.14) for accurate available
|
|
# memory; fall back to the traditional estimation.
|
|
mem_available = self._meminfo_kib("MemAvailable:")
|
|
if mem_available > 0:
|
|
used_bytes = (mem_total - mem_available) * 1024
|
|
else:
|
|
mem_free = self._meminfo_kib("MemFree:")
|
|
buffers = self._meminfo_kib("Buffers:")
|
|
cached = self._meminfo_kib("Cached:")
|
|
used_bytes = (mem_total - mem_free - buffers - cached) * 1024
|
|
elif self.is_darwin:
|
|
total_bytes = _safe_int(_run(["sysctl", "-n", "hw.memsize"]))
|
|
active_pages = _run("vm_stat | grep 'Pages active' | awk '{print $3}'", shell=True).rstrip(":")
|
|
used_bytes = _safe_int(active_pages.replace(",", "")) * 4096
|
|
elif self.is_sun:
|
|
mem_line = _run("/usr/sbin/prtconf | grep Mem", shell=True)
|
|
m = re.search(r"(\d+)", mem_line)
|
|
total_bytes = _safe_int(m.group(1)) * 1048576 if m else 0
|
|
free_pages = _run("vmstat 1 2 | tail -1 | awk '{print $5}'", shell=True)
|
|
used_bytes = total_bytes - _safe_int(free_pages) * 1024
|
|
elif self.is_bsd:
|
|
total_bytes = _safe_int(_run(["sysctl", "-n", "hw.physmem"]))
|
|
active_raw = _run("vmstat -s | grep 'pages active' | awk '{print $1}'", shell=True)
|
|
page_size_raw = _run("vmstat -s | grep 'per page' | awk '{print $1}'", shell=True)
|
|
used_bytes = _safe_int(active_raw) * _safe_int(page_size_raw)
|
|
|
|
if total_bytes == 0:
|
|
return "0MB/0MB (0%)"
|
|
used_mb = used_bytes / 1048576
|
|
total_mb = total_bytes / 1048576
|
|
pct = used_bytes / total_bytes * 100 if total_bytes > 0 else 0.0
|
|
return f"{used_mb:.2f}MB/{total_mb:.2f}MB ({pct:.2f}%)"
|
|
|
|
# ---- Disk Usage ------------------------------------------------------
|
|
|
|
def get_disk_usage(self) -> str:
|
|
if not self.cfg.show_disk_usage:
|
|
return ""
|
|
df_cmd = "df -lkP" if self.is_linux else "df -lk"
|
|
output = _run(df_cmd, shell=True)
|
|
if not output:
|
|
return "0GB/0GB (0%)"
|
|
total_kb = 0
|
|
used_kb = 0
|
|
# On Linux, only aggregate physical block devices; skip loop devices,
|
|
# network mounts (nfs, cifs, etc.), tmpfs, and other virtual fs.
|
|
# On non-Linux, fall back to the original "dev in line" heuristic.
|
|
if self.is_linux:
|
|
_physical_dev = re.compile(
|
|
r"^(/dev/(sd|nvme|vd|md|xvd|mmcblk|dasd|zram|dm-)[^\s]+)"
|
|
)
|
|
for line in output.splitlines():
|
|
parts = line.split()
|
|
if len(parts) < 4:
|
|
continue
|
|
if self.is_linux:
|
|
if not _physical_dev.match(parts[0]):
|
|
continue
|
|
else:
|
|
if "dev" not in line:
|
|
continue
|
|
try:
|
|
total_kb += _safe_int(parts[1])
|
|
used_kb += _safe_int(parts[2])
|
|
except (IndexError, ValueError):
|
|
continue
|
|
if total_kb == 0:
|
|
return "0GB/0GB (0%)"
|
|
total_gb = total_kb / 1048576
|
|
used_gb = used_kb / 1048576
|
|
pct = used_gb / total_gb * 100 if total_gb > 0 else 0.0
|
|
return f"{used_gb:.2f}GB/{total_gb:.2f}GB ({pct:.2f}%)"
|
|
|
|
# ---- Network Traffic -------------------------------------------------
|
|
|
|
def _netdev_column_mb(self, interface: str, col_idx: int) -> str:
|
|
for line in self._netdev_lines:
|
|
if interface in line and ":" in line:
|
|
_, data = line.split(":", 1)
|
|
parts = data.split()
|
|
try:
|
|
return f"{int(parts[col_idx]) / 1048576:.2f}"
|
|
except (IndexError, ValueError):
|
|
return "0.00"
|
|
return "0.00"
|
|
|
|
def get_network_traffic(self) -> str:
|
|
if not self.cfg.show_network_traffic:
|
|
return ""
|
|
if not self.cfg.nic or not self.cfg.nicname:
|
|
return ""
|
|
interfaces = [n.strip() for n in self.cfg.nic.split(",") if n.strip()]
|
|
names = [n.strip() for n in self.cfg.nicname.split(",") if n.strip()]
|
|
parts: list[str] = []
|
|
for idx, iface in enumerate(interfaces):
|
|
friendly = names[idx] if idx < len(names) else iface
|
|
if self.is_linux:
|
|
rx = self._netdev_column_mb(iface, 0)
|
|
tx = self._netdev_column_mb(iface, 8)
|
|
parts.append(f"{friendly} Traffic ({iface}): {rx}MB In/{tx}MB Out")
|
|
else:
|
|
netstat_out = _run("netstat -ibn", shell=True)
|
|
for line in netstat_out.splitlines():
|
|
if iface in line and "Link" in line:
|
|
ns_parts = line.split()
|
|
try:
|
|
rx = int(ns_parts[6]) / 1048576
|
|
tx = int(ns_parts[9]) / 1048576
|
|
parts.append(f"{friendly} Traffic ({iface}): {rx:.2f}MB In/{tx:.2f}MB Out")
|
|
except (IndexError, ValueError):
|
|
pass
|
|
break
|
|
return " - ".join(parts) if parts else ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Special mode collectors
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def collect_devel() -> str:
|
|
"""Languages, compilers, build tools, and package managers."""
|
|
parts: list[str] = []
|
|
for binary, flag, pattern, label in _DEVEL_PROBES:
|
|
ver = _probe_version(binary, flag, pattern)
|
|
if ver:
|
|
parts.append(f"{label} {ver}")
|
|
return " - ".join(parts) if parts else "None detected"
|
|
|
|
|
|
def collect_admin() -> str:
|
|
"""Administration panels, databases, monitoring, and web servers."""
|
|
present = _probe_present(_ADMIN_PROBES)
|
|
return ", ".join(present) if present else "None detected"
|
|
|
|
|
|
def collect_devops() -> str:
|
|
"""Containers, orchestration, IaC, CI/CD, and cloud CLIs."""
|
|
present = _probe_present(_DEVOPS_PROBES)
|
|
return ", ".join(present) if present else "None detected"
|
|
|
|
|
|
def collect_kernel() -> str:
|
|
"""Kernel version, compiler, security, and module count."""
|
|
parts: list[str] = []
|
|
|
|
# Kernel version + compiler from /proc/version.
|
|
proc_ver = _read_file_lines("/proc/version")
|
|
if proc_ver:
|
|
line = proc_ver[0].strip()
|
|
ver_m = re.search(r"Linux version (\S+)", line)
|
|
gcc_m = re.search(r"gcc[^0-9]*(\d+\.\d+\.\d+)", line)
|
|
if ver_m:
|
|
entry = ver_m.group(1)
|
|
if gcc_m:
|
|
entry += f" (gcc {gcc_m.group(1)})"
|
|
parts.append(entry)
|
|
|
|
# Security modules.
|
|
sec: list[str] = []
|
|
selinux = _run("getenforce 2>/dev/null", shell=True)
|
|
if selinux:
|
|
sec.append(f"SELinux:{selinux}")
|
|
|
|
if os.path.isfile("/sys/kernel/security/lsm"):
|
|
lsm = _read_sysfs_str("/sys/kernel/security/lsm")
|
|
if lsm and "apparmor" in lsm.lower():
|
|
sec.append("AppArmor")
|
|
elif os.path.isdir("/etc/apparmor"):
|
|
sec.append("AppArmor")
|
|
|
|
if os.path.exists("/sys/kernel/security/smack"):
|
|
sec.append("Smack")
|
|
if os.path.isfile("/proc/sys/kernel/yama/ptrace_scope"):
|
|
sec.append("Yama")
|
|
if sec:
|
|
parts.append("Security: " + ", ".join(sec))
|
|
|
|
# Module count.
|
|
if os.path.isfile("/proc/modules"):
|
|
mod_count = sum(1 for _ in _read_file_lines("/proc/modules"))
|
|
parts.append(f"Modules: {mod_count}")
|
|
|
|
return " - ".join(parts) if parts else platform.release()
|
|
|
|
|
|
def collect_pkgs() -> str:
|
|
"""Installed package count per detected package manager."""
|
|
parts: list[str] = []
|
|
for cmd, label in _PKG_COUNTERS:
|
|
count_str = _run(cmd, shell=True).strip()
|
|
count = _safe_int(count_str)
|
|
if count > 0:
|
|
parts.append(f"{label}: {count}")
|
|
return " - ".join(parts) if parts else "None detected"
|
|
|
|
|
|
def collect_security() -> str:
|
|
"""Firewall, hardening tools, and access-control status."""
|
|
parts: list[str] = []
|
|
|
|
# Binary probes.
|
|
present = _probe_present(_SECURITY_PROBES)
|
|
if present:
|
|
parts.append("Tools: " + ", ".join(present))
|
|
|
|
# Firewall status from files (works in sleuth mode on Linux).
|
|
if os.path.isfile("/etc/ufw/ufw.conf"):
|
|
ufw_lines = _read_file_lines("/etc/ufw/ufw.conf")
|
|
for line in ufw_lines:
|
|
stripped = line.strip()
|
|
if stripped.startswith("ENABLED="):
|
|
state = stripped.split("=", 1)[1].strip().strip('"')
|
|
parts.append(f"UFW: {'active' if state == 'yes' else 'inactive'}")
|
|
break
|
|
|
|
if os.path.isdir("/etc/fail2ban"):
|
|
parts.append("Fail2Ban: installed")
|
|
|
|
# OpenSnitch status from its config (works in sleuth mode).
|
|
if os.path.isfile("/etc/opensnitch/default-config.json"):
|
|
parts.append("OpenSnitch: installed")
|
|
elif os.path.isdir("/etc/opensnitch"):
|
|
parts.append("OpenSnitch: installed")
|
|
|
|
# Cilium status (works in sleuth mode via /sys/fs/bpf).
|
|
cilium_bpf = Path("/sys/fs/bpf/cilium")
|
|
if cilium_bpf.is_dir():
|
|
parts.append("Cilium: active")
|
|
elif os.path.isdir("/var/run/cilium"):
|
|
parts.append("Cilium: installed")
|
|
|
|
# eBPF support detection from /sys/kernel/debug/tracing (sleuth-safe).
|
|
tracing = Path("/sys/kernel/debug/tracing")
|
|
if tracing.is_dir():
|
|
parts.append("eBPF: supported")
|
|
# eBPF program count from bpffs if mounted.
|
|
bpffs = Path("/sys/fs/bpf")
|
|
if bpffs.is_dir():
|
|
try:
|
|
bpf_count = sum(1 for _ in bpffs.iterdir() if _.is_dir())
|
|
if bpf_count > 0:
|
|
parts.append(f"eBPF programs: {bpf_count}")
|
|
except OSError:
|
|
pass
|
|
|
|
# Kernel hardening.
|
|
if os.path.isfile("/proc/sys/kernel/randomize_va_space"):
|
|
aslr = _read_sysfs_str("/proc/sys/kernel/randomize_va_space")
|
|
aslr_label = "full" if aslr == "2" else ("partial" if aslr == "1" else "off")
|
|
parts.append(f"ASLR: {aslr_label}")
|
|
|
|
if os.path.isfile("/proc/sys/kernel/dmesg_restrict"):
|
|
dmesg = _read_sysfs_str("/proc/sys/kernel/dmesg_restrict")
|
|
parts.append(f"dmesg restricted: {'yes' if dmesg == '1' else 'no'}")
|
|
|
|
if os.path.isfile("/proc/sys/kernel/kptr_restrict"):
|
|
kptr = _read_sysfs_str("/proc/sys/kernel/kptr_restrict")
|
|
parts.append(f"kptr restricted: {'yes' if kptr == '1' else 'no'}")
|
|
|
|
return " - ".join(parts) if parts else "None detected"
|
|
|
|
|
|
def collect_net() -> str:
|
|
"""Network interfaces, IPs, gateway, DNS, and connection counts."""
|
|
parts: list[str] = []
|
|
|
|
# Default interface from /proc/net/route (file read, works in sleuth mode).
|
|
default_iface = ""
|
|
route_lines = _read_file_lines("/proc/net/route")
|
|
for line in route_lines:
|
|
fields = line.split()
|
|
if len(fields) >= 3 and fields[1] == "00000000":
|
|
default_iface = fields[0]
|
|
break
|
|
|
|
# IPv4 address from /sys or /proc (file reads).
|
|
if default_iface:
|
|
# Try /sys first, then fall back to ip command.
|
|
addr_path = f"/sys/class/net/{default_iface}/address"
|
|
if os.path.isfile(addr_path):
|
|
mac = _read_sysfs_str(addr_path)
|
|
if mac:
|
|
parts.append(f"IF: {default_iface} ({mac})")
|
|
else:
|
|
parts.append(f"IF: {default_iface}")
|
|
|
|
# IP address via command (not available in sleuth mode).
|
|
ip_out = _run(f"ip -4 addr show {default_iface} 2>/dev/null", shell=True)
|
|
ip_m = re.search(r"inet\s+([\d.]+)", ip_out)
|
|
if ip_m:
|
|
parts.append(f"IP: {ip_m.group(1)}")
|
|
|
|
# Gateway via command.
|
|
gw_out = _run("ip -4 route show default 2>/dev/null", shell=True)
|
|
gw_m = re.search(r"via\s+([\d.]+)", gw_out)
|
|
if gw_m:
|
|
parts.append(f"GW: {gw_m.group(1)}")
|
|
|
|
# DNS from /etc/resolv.conf (file read).
|
|
dns_lines = _read_file_lines("/etc/resolv.conf")
|
|
dns_servers: list[str] = []
|
|
for line in dns_lines:
|
|
stripped = line.strip()
|
|
if stripped.startswith("nameserver"):
|
|
ns = stripped.split(None, 1)
|
|
if len(ns) >= 2:
|
|
dns_servers.append(ns[1])
|
|
if dns_servers:
|
|
parts.append(f"DNS: {', '.join(dns_servers[:3])}")
|
|
|
|
# Connection counts via ss (command, not in sleuth mode).
|
|
ss_out = _run("ss -tn state established 2>/dev/null | tail -n +2", shell=True)
|
|
if ss_out:
|
|
conn_count = len(ss_out.splitlines())
|
|
parts.append(f"Established: {conn_count}")
|
|
|
|
listen_out = _run("ss -tln 2>/dev/null | tail -n +2", shell=True)
|
|
if listen_out:
|
|
listen_count = len(listen_out.splitlines())
|
|
parts.append(f"Listening: {listen_count}")
|
|
|
|
return " - ".join(parts) if parts else "None detected"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Output assembly
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_OUTPUT_SECTIONS: list[tuple[str, str, str]] = [
|
|
("show_hostname", "text_hostname", "hostname"),
|
|
("show_os", "text_os", "uname_str"),
|
|
("show_distro", "text_distro", "get_distro"),
|
|
("show_cpu", "text_cpu", "get_cpu"),
|
|
("show_gpu", "text_gpu", "get_gpu"),
|
|
("show_processes", "text_processes", "get_processes"),
|
|
("show_uptime", "text_uptime", "get_uptime"),
|
|
("show_users", "text_users", "get_users"),
|
|
("show_load_average", "text_load_average", "get_load_average"),
|
|
("show_battery", "text_battery", "get_battery"),
|
|
("show_memory_usage", "text_memory_usage", "get_memory_usage"),
|
|
("show_disk_usage", "text_disk_usage", "get_disk_usage"),
|
|
("show_network_traffic","text_network", "get_network_traffic"),
|
|
]
|
|
|
|
|
|
def build_output(collector: ProbeFetchCollector, colors: dict[str, str]) -> str:
|
|
"""Assemble the single-line output string from enabled sections."""
|
|
cfg = collector.cfg
|
|
sep = cfg.separator
|
|
parts: list[str] = []
|
|
for toggle_attr, label_attr, value_source in _OUTPUT_SECTIONS:
|
|
if not getattr(cfg, toggle_attr, False):
|
|
continue
|
|
label = getattr(cfg, label_attr, "")
|
|
if value_source.startswith("get_"):
|
|
value = getattr(collector, value_source)()
|
|
else:
|
|
value = getattr(collector, value_source, "")
|
|
if not value:
|
|
continue
|
|
if colors:
|
|
parts.append(f"{colors['label']}{label}{colors['value']}{value}{_RESET}")
|
|
else:
|
|
parts.append(f"{label}{value}")
|
|
if not parts:
|
|
return ""
|
|
if colors:
|
|
sep_str = f"{colors['sep']}{sep}{_RESET}"
|
|
return sep_str.join(parts)
|
|
return sep.join(parts)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI entry point
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def main() -> None:
|
|
global _SLEUTH_MODE
|
|
|
|
args = sys.argv[1:]
|
|
|
|
if "-v" in args or "--version" in args:
|
|
print("probefetch v5.0.0")
|
|
return
|
|
|
|
# Resolve theme.
|
|
theme_name = "auto"
|
|
for arg in args:
|
|
if arg.startswith("--theme="):
|
|
theme_name = arg.split("=", 1)[1]
|
|
colors = _resolve_theme(theme_name)
|
|
|
|
# Stealth mode: suppress all subprocess spawning.
|
|
if "--sleuth" in args or "--stealth" in args:
|
|
_SLEUTH_MODE = True
|
|
|
|
# Special modes produce their own independent output line.
|
|
special_map = {
|
|
"--devel": ("Dev:", collect_devel),
|
|
"--admin": ("Admin:", collect_admin),
|
|
"--devops": ("DevOps:", collect_devops),
|
|
"--kernel": ("Kernel:", collect_kernel),
|
|
"--pkgs": ("Packages:", collect_pkgs),
|
|
"--security": ("Security:", collect_security),
|
|
"--net": ("Network:", collect_net),
|
|
}
|
|
for flag, (header, func) in special_map.items():
|
|
if flag in args:
|
|
output = func()
|
|
if colors:
|
|
print(f"{colors['header']}{header}{_RESET} {colors['value']}{output}{_RESET}")
|
|
else:
|
|
print(f"{header} {output}")
|
|
return
|
|
|
|
cfg = ProbeFetchConfig()
|
|
|
|
# When explicit section names are provided, disable all first,
|
|
# then enable only those requested.
|
|
requested = [a for a in args if a in _VALID_SECTIONS]
|
|
if requested:
|
|
for attr in _SECTION_TO_ATTR.values():
|
|
object.__setattr__(cfg, attr, False)
|
|
for section in requested:
|
|
object.__setattr__(cfg, _SECTION_TO_ATTR[section], True)
|
|
|
|
collector = ProbeFetchCollector(cfg)
|
|
print(build_output(collector, colors))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|