check disk wearout levels

misc/disks.py

@@ -1,7 +1,7 @@
 import json
 import subprocess
 from dataclasses import dataclass
-from enum import StrEnum
+from enum import Enum, StrEnum
 from typing import Optional, Self
 
 
@@ -132,3 +132,58 @@ class LVAttr:
         json_obj = json.loads(subprocess.run(["lvs", "--reportformat=json", name], capture_output=True).stdout)
         attr_str = json_obj["report"][0]["lv"][0]["lv_attr"]
         return cls.from_str(attr_str, name)
+
+
+class WearoutIndicator(Enum):
+    REALLOCATED_SECTORS = 0
+    SPARE_BLOCKS = 1
+
+
+@dataclass
+class WearoutReading:
+    indicator: WearoutIndicator
+    current_reading: int
+    threshold_reading: int
+
+
+def _get_wearout_reading_from_smartctl_output(smartctl_output: dict) -> WearoutReading:
+    disk_protocol = smartctl_output["device"]["protocol"]
+    rotation_rate = smartctl_output.get("rotation_rate", 0)
+    match rotation_rate:
+        case 0:  # assuming non-rotating media is an SSD
+            indicator = WearoutIndicator.SPARE_BLOCKS
+            match disk_protocol:
+                case "ATA":
+                    attr_table = smartctl_output["ata_smart_attributes"]["table"]
+                    for a in attr_table:
+                        if a["name"] == "Available_Reservd_Space":
+                            value = a["value"]
+                            threshold = a["thresh"]
+                            break
+                    else:
+                        raise Exception(f"no Available_Reservd_Space on ATA SSD")
+                case "NVMe":
+                    value = smartctl_output["nvme_smart_health_information_log"]["available_spare"]
+                    threshold = smartctl_output["nvme_smart_health_information_log"]["available_spare_threshold"]
+        case _:
+            indicator = WearoutIndicator.REALLOCATED_SECTORS
+            match disk_protocol:
+                case "ATA":
+                    attr_table = smartctl_output["ata_smart_attributes"]["table"]
+                    for a in attr_table:
+                        if a["name"] == "Reallocated_Sector_Ct":
+                            value = a["value"]
+                            threshold = a["thresh"]
+                            break
+                    else:
+                        raise Exception(f"no Reallocated_Sector_Ct on ATA HDD")
+                case "NVMe":  # ? NVMe HDDs are very rare, if they even exist
+                    raise NotImplementedError
+
+    return WearoutReading(indicator, current_reading=value, threshold_reading=threshold)
+
+
+def get_wearout_reading(disk: str) -> WearoutReading:
+    smartctl_output = json.loads(subprocess.run(["smartctl", "-ja", disk], capture_output=True).stdout.decode("utf-8"))
+    wearout_reading = _get_wearout_reading_from_smartctl_output(smartctl_output)
+    return wearout_reading