Created
June 11, 2026 16:23
-
-
Save seantcanavan/7a6d7b0fa523631ec137fc40796b43b6 to your computer and use it in GitHub Desktop.
Print failure analysis information for HDDs and SSDs in TrueNAS
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| from __future__ import annotations | |
| import json | |
| import re | |
| import subprocess | |
| from pathlib import Path | |
| from typing import Any | |
| def run_smartctl_json(device: str) -> dict[str, Any] | None: | |
| try: | |
| result = subprocess.run( | |
| ["smartctl", "-a", "-j", device], | |
| capture_output=True, | |
| text=True, | |
| check=False, | |
| ) | |
| if result.returncode not in (0, 4): | |
| return None | |
| return json.loads(result.stdout) | |
| except Exception: | |
| return None | |
| def run_smartctl_text(device: str) -> str: | |
| try: | |
| result = subprocess.run( | |
| ["smartctl", "-x", device], | |
| capture_output=True, | |
| text=True, | |
| check=False, | |
| ) | |
| return result.stdout | |
| except Exception: | |
| return "" | |
| def get_devices() -> list[str]: | |
| devices: list[str] = [] | |
| for path in sorted(Path("/sys/block").iterdir()): | |
| name = path.name | |
| if name.startswith(("sd", "hd", "nvme")): | |
| if name.startswith("nvme") and "n" not in name: | |
| continue | |
| if name.startswith(("sd", "hd")) and any(c.isdigit() for c in name): | |
| continue | |
| devices.append(f"/dev/{name}") | |
| return devices | |
| def nested_string( | |
| data: dict[str, Any], | |
| *keys: str, | |
| default: str = "Unknown", | |
| ) -> str: | |
| current: Any = data | |
| for key in keys: | |
| if not isinstance(current, dict): | |
| return default | |
| current = current.get(key) | |
| if isinstance(current, str) and current: | |
| return current | |
| return default | |
| def is_in_progress_text(text: str) -> bool: | |
| text_lower = text.lower() | |
| return ( | |
| "in progress" in text_lower | |
| or "self test in progress" in text_lower | |
| or "self-test in progress" in text_lower | |
| or "self-test routine in progress" in text_lower | |
| or "running" in text_lower | |
| or "remaining" in text_lower | |
| ) | |
| def bytes_to_human(value: int) -> str: | |
| units = ["B", "KiB", "MiB", "GiB", "TiB", "PiB"] | |
| size = float(value) | |
| for unit in units: | |
| if size < 1024.0 or unit == units[-1]: | |
| if unit == "B": | |
| return f"{int(size)} {unit}" | |
| return f"{size:.2f} {unit}" | |
| size /= 1024.0 | |
| return f"{value} B" | |
| def nvme_data_units_to_bytes(data_units: int) -> int: | |
| return data_units * 512_000 | |
| def get_protocol(data: dict[str, Any]) -> str: | |
| return ( | |
| data.get("device", {}) | |
| .get("protocol", "") | |
| .upper() | |
| ) | |
| def get_ata_attribute(data: dict[str, Any], attr_id: int) -> dict[str, Any] | None: | |
| table = data.get("ata_smart_attributes", {}).get("table", []) | |
| if not isinstance(table, list): | |
| return None | |
| for attr in table: | |
| if isinstance(attr, dict) and attr.get("id") == attr_id: | |
| return attr | |
| return None | |
| def get_ata_attribute_by_name( | |
| data: dict[str, Any], | |
| *names: str, | |
| ) -> dict[str, Any] | None: | |
| wanted = {name.lower() for name in names} | |
| table = data.get("ata_smart_attributes", {}).get("table", []) | |
| if not isinstance(table, list): | |
| return None | |
| for attr in table: | |
| if not isinstance(attr, dict): | |
| continue | |
| name = attr.get("name") | |
| if isinstance(name, str) and name.lower() in wanted: | |
| return attr | |
| return None | |
| def get_ata_raw_value(attr: dict[str, Any] | None) -> int | str | None: | |
| if not isinstance(attr, dict): | |
| return None | |
| raw = attr.get("raw") | |
| if isinstance(raw, dict): | |
| value = raw.get("value") | |
| if value is not None: | |
| return value | |
| string = raw.get("string") | |
| if string is not None: | |
| return str(string) | |
| return None | |
| def get_ata_normalized_value(attr: dict[str, Any] | None) -> int | None: | |
| if not isinstance(attr, dict): | |
| return None | |
| value = attr.get("value") | |
| if isinstance(value, int): | |
| return value | |
| return None | |
| def format_ata_attr( | |
| data: dict[str, Any], | |
| attr_id: int, | |
| label: str, | |
| include_normalized: bool = True, | |
| ) -> tuple[str, str] | None: | |
| attr = get_ata_attribute(data, attr_id) | |
| if attr is None: | |
| return None | |
| raw = get_ata_raw_value(attr) | |
| normalized = get_ata_normalized_value(attr) | |
| if raw is None and normalized is None: | |
| return None | |
| if include_normalized and normalized is not None and raw is not None: | |
| return label, f"{raw} (normalized {normalized})" | |
| if raw is not None: | |
| return label, str(raw) | |
| return label, str(normalized) | |
| def find_attribute( | |
| data: dict[str, Any], | |
| attr_id: int, | |
| ) -> int | str | None: | |
| return get_ata_raw_value(get_ata_attribute(data, attr_id)) | |
| def get_temperature( | |
| data: dict[str, Any], | |
| smartctl_text: str, | |
| ) -> int | None: | |
| nvme_temp = data.get("nvme_smart_health_information_log", {}).get( | |
| "temperature" | |
| ) | |
| if isinstance(nvme_temp, int): | |
| return nvme_temp | |
| temp = data.get("temperature", {}).get("current") | |
| if isinstance(temp, int): | |
| return temp | |
| match = re.search( | |
| r"Current(?: Drive)? Temperature:\s+(\d+)\s+(?:C|Celsius)", | |
| smartctl_text, | |
| re.MULTILINE, | |
| ) | |
| if match: | |
| return int(match.group(1)) | |
| match = re.search( | |
| r"Temperature:\s+(\d+)\s+Celsius", | |
| smartctl_text, | |
| re.MULTILINE, | |
| ) | |
| if match: | |
| return int(match.group(1)) | |
| return None | |
| def get_max_temperature( | |
| data: dict[str, Any], | |
| smartctl_text: str, | |
| ) -> int | None: | |
| if get_protocol(data) == "NVME": | |
| return None | |
| drive_trip = data.get("temperature", {}).get("drive_trip") | |
| if isinstance(drive_trip, int): | |
| return drive_trip | |
| match = re.search( | |
| r"Lifetime\s+Min/Max Temperature:\s+\d+/(\d+)\s+Celsius", | |
| smartctl_text, | |
| re.MULTILINE, | |
| ) | |
| if match: | |
| return int(match.group(1)) | |
| match = re.search( | |
| r"Drive Trip Temperature:\s+(\d+)\s+C", | |
| smartctl_text, | |
| re.MULTILINE, | |
| ) | |
| if match: | |
| return int(match.group(1)) | |
| return None | |
| def get_health(data: dict[str, Any]) -> str: | |
| smart_status = data.get("smart_status") | |
| if isinstance(smart_status, dict): | |
| passed = smart_status.get("passed") | |
| if passed is True: | |
| return "PASSED" | |
| if passed is False: | |
| return "FAILED" | |
| return "UNKNOWN" | |
| def get_nvme_self_test_status(data: dict[str, Any]) -> str | None: | |
| operation = ( | |
| data.get("nvme_self_test_log", {}) | |
| .get("current_self_test_operation", {}) | |
| ) | |
| if not isinstance(operation, dict): | |
| return None | |
| value = operation.get("value") | |
| text = operation.get("string") | |
| if value == 0: | |
| return "None" | |
| if isinstance(text, str) and text: | |
| return text | |
| return "UNKNOWN" | |
| def get_ata_self_test_status(data: dict[str, Any]) -> str | None: | |
| status = ( | |
| data.get("ata_smart_data", {}) | |
| .get("self_test", {}) | |
| .get("status") | |
| ) | |
| if not isinstance(status, dict): | |
| return None | |
| text = status.get("string") | |
| if isinstance(text, dict): | |
| text = text.get("value") | |
| remaining = status.get("remaining_percent") | |
| if isinstance(remaining, int) and remaining > 0: | |
| if isinstance(text, str) and text: | |
| return f"{text} ({100 - remaining}% complete)" | |
| return f"In progress ({100 - remaining}% complete)" | |
| if isinstance(text, str) and is_in_progress_text(text): | |
| return text | |
| return "None" | |
| def get_scsi_self_test_status(data: dict[str, Any]) -> str | None: | |
| for key, value in data.items(): | |
| if not key.startswith("scsi_self_test_"): | |
| continue | |
| if not isinstance(value, dict): | |
| continue | |
| result = value.get("result", {}) | |
| code = value.get("code", {}) | |
| if not isinstance(result, dict): | |
| continue | |
| result_text = result.get("string") | |
| if not isinstance(result_text, str): | |
| continue | |
| if is_in_progress_text(result_text): | |
| test_type = "Unknown" | |
| if isinstance(code, dict): | |
| test_type = str(code.get("string", "Unknown")) | |
| return f"{test_type} - {result_text}" | |
| return None | |
| def get_self_test_status(data: dict[str, Any]) -> str: | |
| protocol = get_protocol(data) | |
| if protocol == "NVME": | |
| return get_nvme_self_test_status(data) or "None" | |
| if protocol == "SCSI": | |
| return get_scsi_self_test_status(data) or "None" | |
| return ( | |
| get_ata_self_test_status(data) | |
| or get_scsi_self_test_status(data) | |
| or get_nvme_self_test_status(data) | |
| or "None" | |
| ) | |
| def get_nvme_last_test(data: dict[str, Any]) -> str | None: | |
| table = data.get("nvme_self_test_log", {}).get("table", []) | |
| if not isinstance(table, list) or not table: | |
| return None | |
| for entry in table: | |
| if not isinstance(entry, dict): | |
| continue | |
| test_type = nested_string( | |
| entry, | |
| "self_test_code", | |
| "string", | |
| ) | |
| status = nested_string( | |
| entry, | |
| "self_test_result", | |
| "string", | |
| ) | |
| if is_in_progress_text(status): | |
| continue | |
| power_on_hours = entry.get("power_on_hours") | |
| if isinstance(power_on_hours, int): | |
| return f"{test_type} - {status} (POH {power_on_hours})" | |
| return f"{test_type} - {status}" | |
| return None | |
| def get_ata_last_test(data: dict[str, Any]) -> str | None: | |
| table = ( | |
| data.get("ata_smart_self_test_log", {}) | |
| .get("standard", {}) | |
| .get("table", []) | |
| ) | |
| if not isinstance(table, list) or not table: | |
| return None | |
| for entry in table: | |
| if not isinstance(entry, dict): | |
| continue | |
| test_type = nested_string( | |
| entry, | |
| "type", | |
| "string", | |
| ) | |
| status = nested_string( | |
| entry, | |
| "status", | |
| "string", | |
| ) | |
| if is_in_progress_text(status): | |
| continue | |
| lifetime_hours = entry.get("lifetime_hours") | |
| if isinstance(lifetime_hours, int): | |
| return f"{test_type} - {status} (POH {lifetime_hours})" | |
| return f"{test_type} - {status}" | |
| return None | |
| def get_scsi_last_test(data: dict[str, Any]) -> str | None: | |
| scsi_tests: list[tuple[int, dict[str, Any]]] = [] | |
| for key, value in data.items(): | |
| match = re.fullmatch(r"scsi_self_test_(\d+)", key) | |
| if not match: | |
| continue | |
| if isinstance(value, dict): | |
| scsi_tests.append((int(match.group(1)), value)) | |
| if not scsi_tests: | |
| return None | |
| for _, entry in sorted(scsi_tests, key=lambda item: item[0]): | |
| test_type = nested_string( | |
| entry, | |
| "code", | |
| "string", | |
| ) | |
| status = nested_string( | |
| entry, | |
| "result", | |
| "string", | |
| ) | |
| if is_in_progress_text(status): | |
| continue | |
| power_on_hours = entry.get("power_on_time", {}).get( | |
| "hours", | |
| ) | |
| if isinstance(power_on_hours, int): | |
| return f"{test_type} - {status} (POH {power_on_hours})" | |
| return f"{test_type} - {status}" | |
| return None | |
| def get_last_test(data: dict[str, Any]) -> str: | |
| protocol = get_protocol(data) | |
| if protocol == "NVME": | |
| return get_nvme_last_test(data) or "None" | |
| if protocol == "SCSI": | |
| return get_scsi_last_test(data) or "None" | |
| return ( | |
| get_ata_last_test(data) | |
| or get_scsi_last_test(data) | |
| or get_nvme_last_test(data) | |
| or "None" | |
| ) | |
| def get_power_on_hours(data: dict[str, Any]) -> str: | |
| power_on = data.get("power_on_time", {}) | |
| if isinstance(power_on, dict): | |
| hours = power_on.get("hours") | |
| if hours is not None: | |
| return str(hours) | |
| nvme_power_on = data.get("nvme_smart_health_information_log", {}).get( | |
| "power_on_hours" | |
| ) | |
| if nvme_power_on is not None: | |
| return str(nvme_power_on) | |
| return "Unknown" | |
| def get_power_cycles(data: dict[str, Any]) -> str: | |
| count = data.get("power_cycle_count") | |
| if count is not None: | |
| return str(count) | |
| nvme_cycles = data.get("nvme_smart_health_information_log", {}).get( | |
| "power_cycles" | |
| ) | |
| if nvme_cycles is not None: | |
| return str(nvme_cycles) | |
| return "Unknown" | |
| def get_rpm(data: dict[str, Any]) -> str: | |
| rpm = data.get("rotation_rate") | |
| if rpm == 0: | |
| return "SSD/NVMe" | |
| if rpm is None: | |
| protocol = get_protocol(data) | |
| if protocol == "NVME": | |
| return "SSD/NVMe" | |
| return "Unknown" | |
| return str(rpm) | |
| def is_flash_drive(data: dict[str, Any]) -> bool: | |
| protocol = get_protocol(data) | |
| if protocol == "NVME": | |
| return True | |
| rpm = data.get("rotation_rate") | |
| if rpm == 0: | |
| return True | |
| model = get_model(data).lower() | |
| return "ssd" in model or "solid state" in model | |
| def get_model(data: dict[str, Any]) -> str: | |
| model = ( | |
| data.get("model_name") | |
| or data.get("scsi_model_name") | |
| or data.get("model_family") | |
| ) | |
| if isinstance(model, str) and model.strip(): | |
| return re.sub(r"\s+", " ", model).strip() | |
| vendor = data.get("scsi_vendor") | |
| product = data.get("scsi_product") | |
| parts = [ | |
| part.strip() | |
| for part in (vendor, product) | |
| if isinstance(part, str) and part.strip() | |
| ] | |
| if parts: | |
| return re.sub(r"\s+", " ", " ".join(parts)).strip() | |
| return "Unknown" | |
| def get_nvme_flash_wear_indicators(data: dict[str, Any]) -> list[tuple[str, str]]: | |
| log = data.get("nvme_smart_health_information_log", {}) | |
| if not isinstance(log, dict): | |
| return [] | |
| rows: list[tuple[str, str]] = [] | |
| percentage_used = log.get("percentage_used") | |
| if isinstance(percentage_used, int): | |
| rows.append(("Percentage Used", f"{percentage_used}%")) | |
| available_spare = log.get("available_spare") | |
| if isinstance(available_spare, int): | |
| rows.append(("Available Spare", f"{available_spare}%")) | |
| spare_threshold = log.get("available_spare_threshold") | |
| if isinstance(spare_threshold, int): | |
| rows.append(("Spare Threshold", f"{spare_threshold}%")) | |
| data_units_written = log.get("data_units_written") | |
| if isinstance(data_units_written, int): | |
| rows.append( | |
| ( | |
| "Data Written", | |
| bytes_to_human(nvme_data_units_to_bytes(data_units_written)), | |
| ) | |
| ) | |
| data_units_read = log.get("data_units_read") | |
| if isinstance(data_units_read, int): | |
| rows.append( | |
| ( | |
| "Data Read", | |
| bytes_to_human(nvme_data_units_to_bytes(data_units_read)), | |
| ) | |
| ) | |
| media_errors = log.get("media_errors") | |
| if isinstance(media_errors, int): | |
| rows.append(("Media/Data Integrity Errors", str(media_errors))) | |
| error_entries = log.get("num_err_log_entries") | |
| if isinstance(error_entries, int): | |
| rows.append(("Error Log Entries", str(error_entries))) | |
| unsafe_shutdowns = log.get("unsafe_shutdowns") | |
| if isinstance(unsafe_shutdowns, int): | |
| rows.append(("Unsafe Shutdowns", str(unsafe_shutdowns))) | |
| return rows | |
| def get_ata_flash_wear_indicators(data: dict[str, Any]) -> list[tuple[str, str]]: | |
| candidates = [ | |
| (173, "Wear Leveling Count"), | |
| (174, "Unexpected Power Loss"), | |
| (177, "Wear Leveling Count"), | |
| (179, "Used Reserved Block Count"), | |
| (181, "Program Fail Count"), | |
| (182, "Erase Fail Count"), | |
| (183, "Runtime Bad Block"), | |
| (202, "Percent Lifetime Used/Remaining"), | |
| (231, "SSD Life Left / Temperature"), | |
| (232, "Available Reserved Space"), | |
| (233, "Media Wearout Indicator"), | |
| (234, "NAND Writes"), | |
| (241, "Total LBAs Written"), | |
| (242, "Total LBAs Read"), | |
| ] | |
| rows: list[tuple[str, str]] = [] | |
| seen_ids: set[int] = set() | |
| for attr_id, label in candidates: | |
| row = format_ata_attr(data, attr_id, label) | |
| if row is not None: | |
| rows.append(row) | |
| seen_ids.add(attr_id) | |
| table = data.get("ata_smart_attributes", {}).get("table", []) | |
| if isinstance(table, list): | |
| for attr in table: | |
| if not isinstance(attr, dict): | |
| continue | |
| attr_id = attr.get("id") | |
| if not isinstance(attr_id, int) or attr_id in seen_ids: | |
| continue | |
| name = attr.get("name") | |
| if not isinstance(name, str): | |
| continue | |
| name_lower = name.lower() | |
| if not any( | |
| token in name_lower | |
| for token in ( | |
| "wear", | |
| "life", | |
| "nand", | |
| "reserved", | |
| "program_fail", | |
| "erase_fail", | |
| "percent", | |
| ) | |
| ): | |
| continue | |
| raw = get_ata_raw_value(attr) | |
| normalized = get_ata_normalized_value(attr) | |
| if raw is None and normalized is None: | |
| continue | |
| label = name.replace("_", " ") | |
| if normalized is not None and raw is not None: | |
| rows.append((label, f"{raw} (normalized {normalized})")) | |
| elif raw is not None: | |
| rows.append((label, str(raw))) | |
| else: | |
| rows.append((label, str(normalized))) | |
| return rows | |
| def get_flash_wear_indicators(data: dict[str, Any]) -> list[tuple[str, str]]: | |
| protocol = get_protocol(data) | |
| if protocol == "NVME": | |
| return get_nvme_flash_wear_indicators(data) | |
| if is_flash_drive(data): | |
| return get_ata_flash_wear_indicators(data) | |
| return [] | |
| def print_table( | |
| title: str, | |
| rows: list[tuple[str, str]], | |
| ) -> None: | |
| if not rows: | |
| return | |
| width = max(len(label) for label, _ in rows) | |
| print(title) | |
| for label, value in rows: | |
| print(f" {label + ':':<{width + 1}} {value}") | |
| print() | |
| def print_drive( | |
| data: dict[str, Any], | |
| device: str, | |
| smartctl_text: str, | |
| ) -> None: | |
| model = get_model(data) | |
| serial = data.get("serial_number", "Unknown") | |
| temp = get_temperature( | |
| data, | |
| smartctl_text, | |
| ) | |
| max_temp = get_max_temperature( | |
| data, | |
| smartctl_text, | |
| ) | |
| print("=" * 80) | |
| print( | |
| f"{device} | " | |
| f"{model} | " | |
| f"SN: {serial} | " | |
| f"Health: {get_health(data)} | " | |
| f"RPM: {get_rpm(data)} | " | |
| f"Temp: {temp if temp is not None else 'Unknown'}°C" | |
| ) | |
| print( | |
| f"Max Temp: {max_temp if max_temp is not None else 'Unknown'}°C | " | |
| f"POH: {get_power_on_hours(data)} | " | |
| f"Cycles: {get_power_cycles(data)}" | |
| ) | |
| print() | |
| flash_wear = get_flash_wear_indicators(data) | |
| if flash_wear: | |
| print_table( | |
| "Flash Wear Indicators", | |
| flash_wear, | |
| ) | |
| print("Critical SMART Attributes") | |
| print( | |
| f" Reallocated Sectors: " | |
| f"{find_attribute(data, 5) or 0}" | |
| ) | |
| print( | |
| f" Pending Sectors: " | |
| f"{find_attribute(data, 197) or 0}" | |
| ) | |
| print( | |
| f" Offline Uncorrectable: " | |
| f"{find_attribute(data, 198) or 0}" | |
| ) | |
| print( | |
| f" Reported Uncorrectable: " | |
| f"{find_attribute(data, 187) or 0}" | |
| ) | |
| print( | |
| f" CRC Errors: " | |
| f"{find_attribute(data, 199) or 0}" | |
| ) | |
| print() | |
| print( | |
| f"Current Test: " | |
| f"{get_self_test_status(data)}" | |
| ) | |
| print( | |
| f"Last Test: " | |
| f"{get_last_test(data)}" | |
| ) | |
| print() | |
| def main() -> None: | |
| for device in get_devices(): | |
| data = run_smartctl_json(device) | |
| if data is None: | |
| continue | |
| smartctl_text = run_smartctl_text(device) | |
| print_drive( | |
| data, | |
| device, | |
| smartctl_text, | |
| ) | |
| if __name__ == "__main__": | |
| main() |
Author
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
sample output: