Summarize SMART data on multi-drive systems

smartsummary.py

#!/usr/bin/env python3
# 
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so.
#
# Written by Michael T. Babcock + Google Antigravity
#
from pathlib import Path
import re
import shutil
import subprocess
import sys

try:
    from rich import box
    from rich.console import Console
    from rich.panel import Panel
    from rich.table import Table
    HAS_RICH = True
except ImportError:
    HAS_RICH = False

class DriveHealthScanner:
    def __init__(self):
        self.smartctl_path = shutil.which("smartctl")
        if not self.smartctl_path:
            print("Error: 'smartctl' (smartmontools) not found in PATH.")
            sys.exit(1)
        self.console = Console() if HAS_RICH else None

    def _run_command(self, cmd):
        """Executes a shell command and returns the output."""
        try:
            result = subprocess.run(
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                universal_newlines=True,
                check=False,
            )
            return result.stdout
        except Exception as e:
            return f"Error executing command: {e}"

    def get_disks(self):
        """Finds all relevant SATA and NVMe disks."""
        diskre = re.compile(r'^(sd[a-z]+|nvme[0-9]+n[0-9]+)$')
        devices = [item.name for item in Path('/sys/block').iterdir() if diskre.match(item.name)]
        return [f"/dev/{name.strip()}" for name in sorted(devices) if name.strip()]

    def parse_smart_attribute(self, output, attr_id):
        """
        Parses a specific SMART attribute ID from smartctl -A output.
        Returns the raw value (usually the last column).
        """
        pattern = rf"^\s*{attr_id}\s+"
        for line in output.splitlines():
            if re.match(pattern, line):
                parts = line.split()
                return parts[-1]
        return "0"

    def check_failed_attributes(self, output):
        """
        Checks if any SMART attribute has failed (WHEN_FAILED column is not '-').
        Returns list of failed attribute labels.
        """
        failed = []
        for line in output.splitlines():
            parts = line.split()
            if len(parts) >= 10 and parts[0].isdigit():
                when_failed = parts[8]
                if when_failed != "-" and when_failed.upper() != "NEVER":
                    failed.append(f"Attr {parts[0]} ({parts[1]})")
        return failed

    def get_nvme_metric(self, output, metric_name):
        """Parses NVMe specific metrics."""
        pattern = rf"{metric_name}:?\s*(.*)"
        match = re.search(pattern, output, re.IGNORECASE)
        if match:
            return match.group(1).strip()
        return "N/A"

    def get_health_status(self, disk, is_nvme=False, output_a=""):
        """Extracts overall health status and checks for failed attributes."""
        output = self._run_command([self.smartctl_path, "-H", disk])
        match = re.search(r"test result:\s*(.*)|health status:\s*(.*)|overall capability:\s*(.*)", output, re.IGNORECASE)
        status = "Unknown"
        if match:
            res = match.group(1) or match.group(2) or match.group(3)
            if res:
                status = res.strip().rstrip("!")

        failed_attrs = []
        if not is_nvme and output_a:
            failed_attrs = self.check_failed_attributes(output_a)
            if failed_attrs and status == "PASSED":
                status = "FAILED"

        return status, failed_attrs

    def get_last_test_result(self, disk):
        """Parses the last self-test result from the log across various smartctl output formats."""
        output = self._run_command([self.smartctl_path, "-l", "self-test", disk])
        if not output or "Error executing command" in output:
            return "No log entry"

        lines = [line.strip() for line in output.splitlines() if line.strip()]

        for line in lines:
            if re.match(r"^#\s*1\s+", line):
                return re.sub(r"\s+", " ", line)

        in_table = False
        for line in lines:
            if re.search(r"\bNum\b|\bTest_Description\b|\bStatus\b", line, re.IGNORECASE):
                in_table = True
                continue
            if in_table and re.match(r"^1\s+", line):
                return re.sub(r"\s+", " ", line)

        for i, line in enumerate(lines):
            if "Self-test execution status:" in line or "previous self-test" in line.lower():
                status_text = line
                if i + 1 < len(lines) and not lines[i + 1].startswith("="):
                    status_text += " " + lines[i + 1]
                return re.sub(r"\s+", " ", status_text)

        if "Permission denied" in output or "Operation not permitted" in output:
            return "Permission Denied (run as root)"

        return "No log entry"

    def scan(self):
        disks = self.get_disks()
        if not disks:
            if HAS_RICH:
                self.console.print("[bold red]No compatible drives (SATA/NVMe) detected.[/bold red]")
            else:
                print("No compatible drives (SATA/NVMe) detected.")
            return

        healthy_drives = []
        unhealthy_drives = []
        unknown_drives = []

        if HAS_RICH:
            self._scan_rich(disks, healthy_drives, unhealthy_drives, unknown_drives)
            self._print_summary_rich(healthy_drives, unhealthy_drives, unknown_drives)
        else:
            self._scan_plain(disks, healthy_drives, unhealthy_drives, unknown_drives)
            self._print_summary_plain(healthy_drives, unhealthy_drives, unknown_drives)

    def get_drive_identity(self, disk):
        """Fetches Model and Serial Number from smartctl -i output."""
        output = self._run_command([self.smartctl_path, "-i", disk])
        model = "Unknown Model"
        serial = "Unknown S/N"

        model_match = re.search(r"(?:Device Model|Model Number):\s*(.*)", output, re.IGNORECASE)
        if model_match:
            model = model_match.group(1).strip()

        serial_match = re.search(r"Serial Number:\s*(.*)", output, re.IGNORECASE)
        if serial_match:
            serial = serial_match.group(1).strip()

        return model, serial

    def _evaluate_disk(self, disk, health, failed_attrs, realloc_spare, pending_val, offline_unc_val, last_test, output_a, healthy_drives, unhealthy_drives, unknown_drives):
        realloc_num = int(realloc_spare) if realloc_spare.isdigit() else 0
        pending_num = int(pending_val) if pending_val.isdigit() else 0
        off_unc_num = int(offline_unc_val) if offline_unc_val.isdigit() else 0
        lt_lower = last_test.lower()

        # Separate Attribute 190 (Airflow_Temperature_Cel) from other failed attributes
        non_temp_failed = [a for a in failed_attrs if "190" not in a and "Airflow_Temperature" not in a]
        temp_only_failed = (len(failed_attrs) > 0) and (len(non_temp_failed) == 0)

        reasons = []
        if len(non_temp_failed) > 0:
            reasons.append(f"SMART attribute failure ({', '.join(non_temp_failed)})")
        elif health.upper() == "FAILED" and not temp_only_failed:
            reasons.append("SMART overall health self-check failure")

        if realloc_num > 0:
            reasons.append(f"{realloc_num} reallocated sectors")
        if pending_num > 0:
            reasons.append(f"{pending_num} pending sectors")
        if off_unc_num > 0:
            reasons.append(f"{off_unc_num} offline uncorrectable sectors")
        if any(k in lt_lower for k in ["failure", "failed", "error"]) and "without error" not in lt_lower:
            reasons.append(f"Last self-test error ({last_test})")

        if len(reasons) > 0:
            model, serial = self.get_drive_identity(disk)
            unhealthy_drives.append((disk, model, serial, reasons))
        elif temp_only_failed:
            raw_temp = self.parse_smart_attribute(output_a, "190")
            temp_str = f"{raw_temp}°C" if raw_temp != "0" else "Attr 190 warning"
            unknown_drives.append(f"{disk} (Attr 190 Temp: {temp_str})")
        elif health.upper() == "PASSED":
            healthy_drives.append(disk)
        else:
            unknown_drives.append(disk)

    def _scan_rich(self, disks, healthy_drives, unhealthy_drives, unknown_drives):
        table = Table(
            title="[bold cyan]SMART Drive Health Report[/bold cyan]",
            caption=f"[dim]Scanned {len(disks)} drive(s)[/dim]",
            box=box.ROUNDED,
            header_style="bold blue",
            show_lines=True,
        )

        table.add_column("Disk", style="bold cyan", justify="left")
        table.add_column("Health Status", justify="left")
        table.add_column("Realloc / Spare", justify="right")
        table.add_column("Pending Sectors", justify="right")
        table.add_column("Offline Uncorr.", justify="right")
        table.add_column("Last Self-Test Result", justify="left")

        for disk in disks:
            is_nvme = "nvme" in disk

            if is_nvme:
                output_a = self._run_command([self.smartctl_path, "-a", disk])
                health, failed_attrs = self.get_health_status(disk, is_nvme=True, output_a=output_a)
                realloc_spare = self.get_nvme_metric(output_a, "available_spare")
                pending_val = "N/A"
                offline_unc_val = "N/A"
            else:
                output_a = self._run_command([self.smartctl_path, "-A", disk])
                health, failed_attrs = self.get_health_status(disk, is_nvme=False, output_a=output_a)
                realloc_spare = self.parse_smart_attribute(output_a, "5")
                pending_val = self.parse_smart_attribute(output_a, "197")
                offline_unc_val = self.parse_smart_attribute(output_a, "198")

            last_test = self.get_last_test_result(disk)

            self._evaluate_disk(disk, health, failed_attrs, realloc_spare, pending_val, offline_unc_val, last_test, output_a, healthy_drives, unhealthy_drives, unknown_drives)

            # Style Health Column
            if health.upper() == "PASSED":
                health_fmt = "[bold green]PASSED[/bold green]"
            elif health.upper() == "FAILED":
                health_fmt = "[bold red]FAILED[/bold red]"
            else:
                health_fmt = f"[bold yellow]{health}[/bold yellow]"

            if failed_attrs:
                for attr_info in failed_attrs:
                    health_fmt += f"\n[dim red]└─ {attr_info}[/dim red]"

            # Style Realloc / Spare
            if realloc_spare.isdigit():
                val = int(realloc_spare)
                realloc_fmt = f"[bold red]{val}[/bold red]" if val > 0 else f"[green]{val}[/green]"
            else:
                realloc_fmt = f"[cyan]{realloc_spare}[/cyan]"

            # Style Pending
            if pending_val.isdigit():
                val = int(pending_val)
                pending_fmt = f"[bold red]{val}[/bold red]" if val > 0 else f"[green]{val}[/green]"
            else:
                pending_fmt = f"[dim]{pending_val}[/dim]"

            # Style Offline Uncorrectable
            if offline_unc_val.isdigit():
                val = int(offline_unc_val)
                offline_unc_fmt = f"[bold red]{val}[/bold red]" if val > 0 else f"[green]{val}[/green]"
            else:
                offline_unc_fmt = f"[dim]{offline_unc_val}[/dim]"

            # Style Last Test
            lt_lower = last_test.lower()
            if "without error" in lt_lower:
                last_test_fmt = f"[green]{last_test}[/green]"
            elif "failure" in lt_lower or "error" in lt_lower or "failed" in lt_lower:
                last_test_fmt = f"[bold red]{last_test}[/bold red]"
            elif "no log entry" in lt_lower:
                last_test_fmt = f"[dim]{last_test}[/dim]"
            else:
                last_test_fmt = last_test

            table.add_row(
                disk,
                health_fmt,
                realloc_fmt,
                pending_fmt,
                offline_unc_fmt,
                last_test_fmt,
            )

        self.console.print(table)

    def _scan_plain(self, disks, healthy_drives, unhealthy_drives, unknown_drives):
        disk_w = 12
        health_w = 15
        realloc_w = 15
        pending_w = 12
        off_unc_w = 14
        health_offset = disk_w + 1

        header = f"{'DISK':<{disk_w}} {'HEALTH':<{health_w}} {'REALLOC/SPARE':<{realloc_w}} {'PENDING':<{pending_w}} {'OFFLINE_UNC':<{off_unc_w}} {'LAST TEST'}"
        print(header)
        print("-" * len(header))

        for disk in disks:
            is_nvme = "nvme" in disk
            if is_nvme:
                output_a = self._run_command([self.smartctl_path, "-a", disk])
                health, failed_attrs = self.get_health_status(disk, is_nvme=True, output_a=output_a)
                realloc_spare = self.get_nvme_metric(output_a, "available_spare")
                pending_val = "N/A"
                offline_unc_val = "N/A"
            else:
                output_a = self._run_command([self.smartctl_path, "-A", disk])
                health, failed_attrs = self.get_health_status(disk, is_nvme=False, output_a=output_a)
                realloc_spare = self.parse_smart_attribute(output_a, "5")
                pending_val = self.parse_smart_attribute(output_a, "197")
                offline_unc_val = self.parse_smart_attribute(output_a, "198")

            last_test = self.get_last_test_result(disk)
            self._evaluate_disk(disk, health, failed_attrs, realloc_spare, pending_val, offline_unc_val, last_test, output_a, healthy_drives, unhealthy_drives, unknown_drives)

            print(f"{disk:<{disk_w}} {health:<{health_w}} {realloc_spare:<{realloc_w}} {pending_val:<{pending_w}} {offline_unc_val:<{off_unc_w}} {last_test}")

            if failed_attrs:
                indent = " " * health_offset
                for attr_info in failed_attrs:
                    print(f"{indent}{attr_info}")

    def _print_summary_rich(self, healthy, unhealthy, unknown):
        summary_text = ""
        if unhealthy:
            summary_text += f"[bold red]🚨 Drives Requiring Replacement / Attention ({len(unhealthy)}):[/bold red]\n"

            disk_w = max([len(d) for d, _, _, _ in unhealthy] + [10])
            model_w = max([len(m) for _, m, _, _ in unhealthy] + [25])
            serial_w = max([len(s) for _, _, s, _ in unhealthy] + [15])

            for disk, model, serial, reasons in unhealthy:
                reason_str = ", ".join(reasons) if reasons else "SMART issues detected"
                summary_text += f"  • [bold red]{disk:<{disk_w}}[/bold red]  [cyan]Model: {model:<{model_w}}[/cyan]  [cyan]S/N: {serial:<{serial_w}}[/cyan]\n"
                indent_spaces = disk_w + 6
                summary_text += f"{' ' * indent_spaces}[dim red]└─ Issues: {reason_str}[/dim red]\n"
            summary_text += "\n"

        if healthy:
            summary_text += f"[bold green]✅ Healthy Drives ({len(healthy)}):[/bold green]\n"
            summary_text += f"  • {', '.join(healthy)}\n\n"

        if unknown:
            summary_text += f"[bold yellow]❓ Drives with Unknown / Inconclusive Status ({len(unknown)}):[/bold yellow]\n"
            summary_text += f"  • {', '.join(unknown)}\n"

        self.console.print()
        self.console.print(Panel(summary_text.strip(), title="[bold white]Executive Summary & Action Plan[/bold white]", border_style="cyan"))

    def _print_summary_plain(self, healthy, unhealthy, unknown):
        print("\n" + "=" * 80)
        print("EXECUTIVE SUMMARY & ACTION PLAN")
        print("=" * 80)
        if unhealthy:
            print(f"🚨 Drives Requiring Replacement / Attention ({len(unhealthy)}):")

            disk_w = max([len(d) for d, _, _, _ in unhealthy] + [10])
            model_w = max([len(m) for _, m, _, _ in unhealthy] + [25])
            serial_w = max([len(s) for _, _, s, _ in unhealthy] + [15])

            for disk, model, serial, reasons in unhealthy:
                reason_str = ", ".join(reasons) if reasons else "SMART issues detected"
                print(f"  • {disk:<{disk_w}}  Model: {model:<{model_w}}  S/N: {serial:<{serial_w}}")
                indent_spaces = disk_w + 6
                print(f"{' ' * indent_spaces}└─ Issues: {reason_str}")
            print()
        if healthy:
            print(f"✅ Healthy Drives ({len(healthy)}):")
            print(f"  - {', '.join(healthy)}\n")
        if unknown:
            print(f"❓ Drives with Unknown / Inconclusive Status ({len(unknown)}):")
            print(f"  - {', '.join(unknown)}")

if __name__ == "__main__":
    scanner = DriveHealthScanner()
    scanner.scan()