Add NVMe lifetime info to API

Dockerfile

@@ -26,6 +26,7 @@ RUN \
         musl \
         openssl \
         yaml \
+        nvme-cli \
     \
     && curl -Lso /usr/bin/cosign "https://github.com/home-assistant/cosign/releases/download/${COSIGN_VERSION}/cosign_${BUILD_ARCH}" \
     && chmod a+x /usr/bin/cosign \

supervisor/api/__init__.py

@@ -198,6 +198,8 @@ def _register_host(self) -> None:
                 web.post("/host/reload", api_host.reload),
                 web.post("/host/options", api_host.options),
                 web.get("/host/services", api_host.services),
+                web.get("/host/nvme/{device}/status", api_host.nvme_device_status),
+                web.get("/host/nvme/status", api_host.nvme_device_status),
             ]
         )
 

supervisor/api/const.py

@@ -16,6 +16,7 @@
 ATTR_AGENT_VERSION = "agent_version"
 ATTR_APPARMOR_VERSION = "apparmor_version"
 ATTR_ATTRIBUTES = "attributes"
+ATTR_AVAILABLE_SPARE = "available_spare"
 ATTR_AVAILABLE_UPDATES = "available_updates"
 ATTR_BACKGROUND = "background"
 ATTR_BOOT_CONFIG = "boot_config"
@@ -28,9 +29,14 @@
 ATTR_BY_ID = "by_id"
 ATTR_CHILDREN = "children"
 ATTR_CONNECTION_BUS = "connection_bus"
+ATTR_CONTROLLER_BUSY_MINUTES = "controller_busy_minutes"
+ATTR_CRITICAL_COMPOSITE_TEMP_MINUTES = "critical_composite_temp_minutes"
+ATTR_CRITICAL_WARNING = "critical_warning"
 ATTR_DATA_DISK = "data_disk"
-ATTR_DEVICE = "device"
+ATTR_DATA_UNITS_READ = "data_units_read"
+ATTR_DATA_UNITS_WRITTEN = "data_units_written"
 ATTR_DEV_PATH = "dev_path"
+ATTR_DEVICE = "device"
 ATTR_DISKS = "disks"
 ATTR_DRIVES = "drives"
 ATTR_DT_SYNCHRONIZED = "dt_synchronized"
@@ -40,6 +46,8 @@
 ATTR_FILESYSTEMS = "filesystems"
 ATTR_FORCE = "force"
 ATTR_GROUP_IDS = "group_ids"
+ATTR_HOST_READ_COMMANDS = "host_read_commands"
+ATTR_HOST_WRITE_COMMANDS = "host_write_commands"
 ATTR_IDENTIFIERS = "identifiers"
 ATTR_IS_ACTIVE = "is_active"
 ATTR_IS_OWNER = "is_owner"
@@ -50,10 +58,16 @@
 ATTR_LOCATION_ATTRIBUTES = "location_attributes"
 ATTR_LOCATIONS = "locations"
 ATTR_MDNS = "mdns"
+ATTR_MEDIA_ERRORS = "media_errors"
 ATTR_MODEL = "model"
-ATTR_MOUNTS = "mounts"
 ATTR_MOUNT_POINTS = "mount_points"
+ATTR_MOUNTS = "mounts"
+ATTR_NUMBER_ERROR_LOG_ENTRIES = "number_error_log_entries"
+ATTR_NVME_DEVICES = "nvme_devices"
 ATTR_PANEL_PATH = "panel_path"
+ATTR_PERCENT_USED = "percent_used"
+ATTR_POWER_CYCLES = "power_cycles"
+ATTR_POWER_ON_HOURS = "power_on_hours"
 ATTR_REMOVABLE = "removable"
 ATTR_REMOVE_CONFIG = "remove_config"
 ATTR_REVISION = "revision"
@@ -65,14 +79,17 @@
 ATTR_SUBSYSTEM = "subsystem"
 ATTR_SYSFS = "sysfs"
 ATTR_SYSTEM_HEALTH_LED = "system_health_led"
+ATTR_TEMPERATURE_KELVIN = "temperature_kelvin"
 ATTR_TIME_DETECTED = "time_detected"
+ATTR_UNSAFE_SHUTDOWNS = "unsafe_shutdowns"
 ATTR_UPDATE_TYPE = "update_type"
 ATTR_USAGE = "usage"
 ATTR_USE_NTP = "use_ntp"
-ATTR_USERS = "users"
 ATTR_USER_PATH = "user_path"
+ATTR_USERS = "users"
 ATTR_VENDOR = "vendor"
 ATTR_VIRTUALIZATION = "virtualization"
+ATTR_WARNING_TEMP_MINUTES = "warning_temp_minutes"
 
 
 class BootSlot(StrEnum):

supervisor/api/host.py

@@ -3,6 +3,7 @@
 import asyncio
 from contextlib import suppress
 import logging
+from pathlib import Path
 from typing import Any
 
 from aiohttp import ClientConnectionResetError, ClientPayloadError, web
@@ -21,15 +22,17 @@
     ATTR_DISK_USED,
     ATTR_FEATURES,
     ATTR_HOSTNAME,
+    ATTR_ID,
     ATTR_KERNEL,
     ATTR_NAME,
     ATTR_OPERATING_SYSTEM,
+    ATTR_PATH,
     ATTR_SERVICES,
     ATTR_STATE,
     ATTR_TIMEZONE,
 )
 from ..coresys import CoreSysAttributes
-from ..exceptions import APIDBMigrationInProgress, APIError, HostLogError
+from ..exceptions import APIDBMigrationInProgress, APIError, APINotFound, HostLogError
 from ..host.const import (
     PARAM_BOOT_ID,
     PARAM_FOLLOW,
@@ -38,22 +41,40 @@
     LogFormatter,
 )
 from ..host.logs import SYSTEMD_JOURNAL_GATEWAYD_LINES_MAX
+from ..host.nvme.manager import NvmeDevice
 from ..utils.systemd_journal import journal_logs_reader
 from .const import (
     ATTR_AGENT_VERSION,
     ATTR_APPARMOR_VERSION,
+    ATTR_AVAILABLE_SPARE,
     ATTR_BOOT_TIMESTAMP,
     ATTR_BOOTS,
     ATTR_BROADCAST_LLMNR,
     ATTR_BROADCAST_MDNS,
+    ATTR_CONTROLLER_BUSY_MINUTES,
+    ATTR_CRITICAL_COMPOSITE_TEMP_MINUTES,
+    ATTR_CRITICAL_WARNING,
+    ATTR_DATA_UNITS_READ,
+    ATTR_DATA_UNITS_WRITTEN,
     ATTR_DT_SYNCHRONIZED,
     ATTR_DT_UTC,
     ATTR_FORCE,
+    ATTR_HOST_READ_COMMANDS,
+    ATTR_HOST_WRITE_COMMANDS,
     ATTR_IDENTIFIERS,
     ATTR_LLMNR_HOSTNAME,
+    ATTR_MEDIA_ERRORS,
+    ATTR_NUMBER_ERROR_LOG_ENTRIES,
+    ATTR_NVME_DEVICES,
+    ATTR_PERCENT_USED,
+    ATTR_POWER_CYCLES,
+    ATTR_POWER_ON_HOURS,
     ATTR_STARTUP_TIME,
+    ATTR_TEMPERATURE_KELVIN,
+    ATTR_UNSAFE_SHUTDOWNS,
     ATTR_USE_NTP,
     ATTR_VIRTUALIZATION,
+    ATTR_WARNING_TEMP_MINUTES,
     CONTENT_TYPE_TEXT,
     CONTENT_TYPE_X_LOG,
 )
@@ -117,6 +138,13 @@ async def info(self, request):
             ATTR_BOOT_TIMESTAMP: self.sys_host.info.boot_timestamp,
             ATTR_BROADCAST_LLMNR: self.sys_host.info.broadcast_llmnr,
             ATTR_BROADCAST_MDNS: self.sys_host.info.broadcast_mdns,
+            ATTR_NVME_DEVICES: [
+                {
+                    ATTR_ID: dev.id,
+                    ATTR_PATH: dev.path.as_posix(),
+                }
+                for dev in self.sys_host.nvme.devices.values()
+            ],
         }
 
     @api_process
@@ -289,3 +317,57 @@ async def advanced_logs(
     ) -> web.StreamResponse:
         """Return systemd-journald logs. Wrapped as standard API handler."""
         return await self.advanced_logs_handler(request, identifier, follow)
+
+    def get_nvme_device_for_request(self, request: web.Request) -> NvmeDevice:
+        """Return NVME device, raise an exception if it doesn't exist."""
+        if "device" in request.match_info:
+            device: str = request.match_info["device"]
+            if device in self.sys_host.nvme.devices:
+                return self.sys_host.nvme.devices[device]
+            if device.startswith("/dev") and (
+                nvme_device := self.sys_host.nvme.get_by_path(Path(device))
+            ):
+                return nvme_device
+            raise APINotFound(f"NVME device {device} does not exist")
+
+        if self.sys_os.available:
+            if self.sys_os.datadisk.disk_used and (
+                nvme_device := self.sys_host.nvme.get_by_path(
+                    self.sys_os.datadisk.disk_used.device_path
+                )
+            ):
+                return nvme_device
+            raise APIError(
+                "Data Disk is not an NVME device, an ID for the NVME device is required"
+            )
+
+        raise APIError(
+            "Not using Home Assistant Operating System, an ID for the NVME device is required"
+        )
+
+    @api_process
+    async def nvme_device_status(self, request: web.Request):
+        """Return status on NVME device from smart log.
+
+        User can provide a path to identify device. Identifier can be omitted if using HAOS and data disk is an NVME device.
+        """
+        nvme_device = self.get_nvme_device_for_request(request)
+        smart_log = await nvme_device.get_smart_log()
+        return {
+            ATTR_AVAILABLE_SPARE: smart_log.avail_spare,
+            ATTR_CRITICAL_WARNING: smart_log.critical_warning,
+            ATTR_DATA_UNITS_READ: smart_log.data_units_read,
+            ATTR_DATA_UNITS_WRITTEN: smart_log.data_units_written,
+            ATTR_PERCENT_USED: smart_log.percent_used,
+            ATTR_TEMPERATURE_KELVIN: smart_log.temperature,
+            ATTR_HOST_READ_COMMANDS: smart_log.host_read_commands,
+            ATTR_HOST_WRITE_COMMANDS: smart_log.host_write_commands,
+            ATTR_CONTROLLER_BUSY_MINUTES: smart_log.controller_busy_time,
+            ATTR_POWER_CYCLES: smart_log.power_cycles,
+            ATTR_POWER_ON_HOURS: smart_log.power_on_hours,
+            ATTR_UNSAFE_SHUTDOWNS: smart_log.unsafe_shutdowns,
+            ATTR_MEDIA_ERRORS: smart_log.media_errors,
+            ATTR_NUMBER_ERROR_LOG_ENTRIES: smart_log.num_err_log_entries,
+            ATTR_WARNING_TEMP_MINUTES: smart_log.warning_temp_time,
+            ATTR_CRITICAL_COMPOSITE_TEMP_MINUTES: smart_log.critical_comp_time,
+        }

supervisor/exceptions.py

@@ -307,6 +307,10 @@ class HostLogError(HostError):
     """Internal error with host log."""
 
 
+class HostNvmeError(HostError):
+    """Error accessing nvme device info."""
+
+
 # API
 
 

supervisor/host/const.py

@@ -67,6 +67,7 @@ class HostFeature(StrEnum):
     JOURNAL = "journal"
     MOUNT = "mount"
     NETWORK = "network"
+    NVME = "nvme"
     OS_AGENT = "os_agent"
     REBOOT = "reboot"
     RESOLVED = "resolved"

supervisor/host/manager.py

@@ -9,7 +9,7 @@
 
 from ..const import BusEvent
 from ..coresys import CoreSys, CoreSysAttributes
-from ..exceptions import HassioError, HostLogError, PulseAudioError
+from ..exceptions import HassioError, HostLogError, HostNvmeError, PulseAudioError
 from ..hardware.const import PolicyGroup
 from ..hardware.data import Device
 from .apparmor import AppArmorControl
@@ -18,6 +18,7 @@
 from .info import InfoCenter
 from .logs import LogsControl
 from .network import NetworkManager
+from .nvme.manager import NvmeManager
 from .services import ServiceManager
 from .sound import SoundControl
 
@@ -38,6 +39,7 @@ def __init__(self, coresys: CoreSys):
         self._network: NetworkManager = NetworkManager(coresys)
         self._sound: SoundControl = SoundControl(coresys)
         self._logs: LogsControl = LogsControl(coresys)
+        self._nvme: NvmeManager = NvmeManager()
 
     async def post_init(self) -> Self:
         """Post init actions that must occur in event loop."""
@@ -79,6 +81,11 @@ def logs(self) -> LogsControl:
         """Return host logs handler."""
         return self._logs
 
+    @property
+    def nvme(self) -> NvmeManager:
+        """Return NVME device manager."""
+        return self._nvme
+
     @property
     def features(self) -> list[HostFeature]:
         """Return a list of host features."""
@@ -118,6 +125,9 @@ def supported_features(self) -> list[HostFeature]:
         if self.sys_dbus.udisks2.is_connected:
             features.append(HostFeature.DISK)
 
+        if self.nvme.devices:
+            features.append(HostFeature.NVME)
+
         # Support added in OS10. Propagation mode changed on mount in 10.2 to support this
         if (
             self.sys_dbus.systemd.is_connected
@@ -151,6 +161,9 @@ async def reload(self):
         with suppress(PulseAudioError):
             await self.sound.update()
 
+        with suppress(HostNvmeError):
+            await self.nvme.update()
+
         _LOGGER.info("Host information reload completed")
         self.supported_features.cache_clear()  # pylint: disable=no-member
 
@@ -167,6 +180,7 @@ async def load(self):
                 await self.logs.load()
 
             await self.network.load()
+            await self.nvme.load()
 
         # Register for events
         self.sys_bus.register_event(BusEvent.HARDWARE_NEW_DEVICE, self._hardware_events)

supervisor/host/nvme/__init__.py

@@ -0,0 +1 @@
+"""NVME device management."""