Warn in the navbar when a hard drive runs hot (#470)

* Warn in the navbar when a hard drive runs hot.

The navbar already warns about high CPU temperature; add the equivalent
for drive temperature, since an overheating disk risks data loss and a
user running an off-grid library won't be watching for it.

- controller/lib/smart.py: build_smart_stats() returns the per-drive list
  plus warning thresholds (55C warn / 65C critical) so the frontend stays
  dumb. Reading SMART wakes a spun-down drive, so drives in standby are
  detected with `smartctl -n standby` (which works through the USB-SATA
  bridges where `hdparm -C` only reports "unknown") and NOT read; their
  last-known reading is reused instead.
- controller/lib/status_worker.py: collect SMART on a slow 60s cadence
  (cached on app.state) rather than the 5s status loop, so the poll does
  not keep USB drives awake. The result flows into the same cached payload
  via a new collect_all_status(smart_stats=...) kwarg.
- Frontend: useDriveTemperature() reports the hottest drive; Nav.js shows
  a yellow/red `hdd` icon with a popup, mirroring the CPU warning.

Verified on hardware: smart_stats reports real temps and SMART runs once
a minute, not every 5s.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* Address Greptile review on HDD temperature warning.

- _drive_is_asleep also treats SLEEP mode as asleep, not just STANDBY.
  smartctl -n standby exits "Device is in SLEEP mode, exit(2)" for the
  deeper power mode; matching only "STANDBY" meant a SLEEP drive was read
  and woken — the exact failure the check prevents.  (TDD: added a failing
  SLEEP test, then fixed.)
- useDriveTemperature uses ?? instead of || for the temperature fallback
  to express "fall back only on null/undefined".

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

controller/lib/smart.py

@@ -2,6 +2,7 @@
 SMART disk health monitoring for WROLPi Controller.
 """
 
+import subprocess
 from typing import Optional
 
 # Import pySMART only if available
@@ -14,6 +15,16 @@ except ImportError:
 
 from controller.lib.config import is_docker_mode
 
+# Warning thresholds for hard-drive temperature, in Celsius.  Unlike CPU
+# temperature, neither SMART nor psutil reports a recommended limit, so we
+# carry sensible spinning-disk defaults in the payload (warn 55, critical 65)
+# and let the frontend stay dumb.
+HDD_HIGH_TEMPERATURE = 55
+HDD_CRITICAL_TEMPERATURE = 65
+
+SMARTCTL_SCAN_TIMEOUT = 10
+SMARTCTL_STANDBY_TIMEOUT = 5
+
 
 def is_smart_available() -> bool:
     """Check if SMART monitoring is available."""
@@ -90,3 +101,108 @@ def _get_attribute(device, name: str) -> Optional[int]:
             except (ValueError, TypeError):
                 return None
     return None
+
+
+def _scan_devices() -> list[tuple[str, Optional[str]]]:
+    """List SMART-capable devices via ``smartctl --scan``.
+
+    ``--scan`` only enumerates devices; it does not read SMART data, so it
+    does not wake a drive that has spun down.  Returns (path, interface)
+    tuples, where interface is the ``-d`` type smartctl detected (e.g.
+    "sat") or None.
+    """
+    try:
+        result = subprocess.run(
+            ["smartctl", "--scan"],
+            capture_output=True, text=True, timeout=SMARTCTL_SCAN_TIMEOUT,
+        )
+    except (subprocess.SubprocessError, OSError):
+        return []
+
+    devices = []
+    for line in result.stdout.splitlines():
+        line = line.strip()
+        if not line or line.startswith("#"):
+            continue
+        # Format: "/dev/sda -d sat # /dev/sda [SAT], ATA device"
+        parts = line.split()
+        path = parts[0]
+        if not path.startswith("/dev/"):
+            continue
+        interface = None
+        if "-d" in parts:
+            idx = parts.index("-d")
+            if idx + 1 < len(parts):
+                interface = parts[idx + 1]
+        devices.append((path, interface))
+    return devices
+
+
+def _drive_is_asleep(path: str) -> bool:
+    """Return True if the drive at ``path`` is in standby/sleep.
+
+    Uses ``smartctl -n standby``, which checks the drive's power mode and
+    exits WITHOUT waking it when it is in standby.  This goes through the
+    same SAT translation smartctl already uses to read these drives, so it
+    works on the USB-SATA bridges where ``hdparm -C`` only reports
+    "unknown".  On any error we return False so the caller falls back to a
+    normal SMART read (better to read an awake drive than to never read it).
+    """
+    try:
+        result = subprocess.run(
+            ["smartctl", "-n", "standby", "-i", path],
+            capture_output=True, text=True, timeout=SMARTCTL_STANDBY_TIMEOUT,
+        )
+    except (subprocess.SubprocessError, OSError):
+        return False
+    # In a low-power mode smartctl prints e.g. "Device is in STANDBY mode,
+    # exit(2)" (or "SLEEP" for the deeper mode) and skips the data read.  An
+    # awake -i read mentions neither.
+    output = result.stdout.upper()
+    return "STANDBY" in output or "SLEEP" in output
+
+
+def build_smart_stats(previous: Optional[dict] = None) -> dict:
+    """Collect SMART status for the navbar/Disk Management warning.
+
+    Reading SMART wakes a sleeping drive, so any drive in standby is NOT
+    read: instead we reuse its last-known data from ``previous`` (matched by
+    path).  A spun-down drive is not hot, so this is safe.  The warning
+    thresholds travel in the payload alongside the per-drive list so the
+    frontend does not hardcode them.
+
+    Returns ``{"drives": [...], "high_temperature": int,
+    "critical_temperature": int}``.  In Docker mode or without pySMART the
+    drive list is empty.
+    """
+    stats = {
+        "drives": [],
+        "high_temperature": HDD_HIGH_TEMPERATURE,
+        "critical_temperature": HDD_CRITICAL_TEMPERATURE,
+    }
+    if not is_smart_available():
+        return stats
+
+    previous_by_path = {}
+    if previous:
+        for drive in previous.get("drives", []):
+            previous_by_path[drive.get("path")] = drive
+
+    try:
+        for path, interface in _scan_devices():
+            if _drive_is_asleep(path):
+                # Don't wake it; reuse the last reading if we have one.
+                prior = previous_by_path.get(path)
+                if prior is not None:
+                    stats["drives"].append(prior)
+                continue
+            try:
+                device = Device(path, interface=interface)
+            except Exception:
+                continue
+            stats["drives"].append(_get_device_smart(device))
+    except Exception:
+        # Never let a status-collection failure blank the whole payload.
+        pass
+
+    return stats

controller/lib/status_worker.py

@@ -13,6 +13,7 @@ from typing import Optional
 
 logger = logging.getLogger(__name__)
 
+from controller.lib.smart import build_smart_stats
 from controller.lib.status import (
     get_cpu_status,
     get_disk_bandwidth_status,
@@ -29,13 +30,21 @@ from controller.lib.status import (
 # Type alias for cached status dict
 CachedStatus = dict
 
+# SMART reads can wake spun-down USB drives, so collect them far less often
+# than the fast (5s) stats.
+SMART_REFRESH_SECONDS = 60
 
-async def collect_all_status() -> CachedStatus:
+
+async def collect_all_status(smart_stats: Optional[dict] = None) -> CachedStatus:
     """
     Collect all status information concurrently.
 
     Returns a dict with all status types. Individual failures return None
     for that stat type rather than failing the entire collection.
+
+    ``smart_stats`` is collected on a slower cadence by the worker loop (see
+    ``status_worker_loop``) and passed in so it lands in the same cached
+    payload as the fast 5s stats.
     """
     # Run all status collections concurrently using asyncio.to_thread
     # since the status functions are synchronous (use psutil)
@@ -65,6 +74,7 @@ async def collect_all_status() -> CachedStatus:
         "processes_stats": processes if not isinstance(processes, Exception) else [],
         "disk_bandwidth_stats": disk_bandwidth if not isinstance(disk_bandwidth, Exception) else {},
         "uptime_stats": uptime if not isinstance(uptime, Exception) else None,
+        "smart_stats": smart_stats,
         "last_status": datetime.now().isoformat(),
     }
 
@@ -101,6 +111,30 @@ def get_adaptive_sleep(load_stats: Optional[dict], base_sleep: float = 5.0) -> f
     return base_sleep
 
 
+async def refresh_smart_stats(app, now: datetime) -> dict:
+    """Collect SMART stats, but no more than once per SMART_REFRESH_SECONDS.
+
+    SMART reads can wake spun-down USB drives, so this runs on a slow
+    cadence independent of the fast status loop.  The result and its
+    timestamp are cached on ``app.state``; between refreshes the cached
+    value is returned unchanged.  ``build_smart_stats`` is given the prior
+    result so it can reuse readings for drives that are now asleep.
+    """
+    last = getattr(app.state, "smart_last_collected", None)
+    cache = getattr(app.state, "smart_cache", None)
+    due = (
+        cache is None
+        or last is None
+        or (now - last).total_seconds() >= SMART_REFRESH_SECONDS
+    )
+    if due:
+        # build_smart_stats shells out to smartctl/hdparm; keep it off the loop.
+        cache = await asyncio.to_thread(build_smart_stats, cache)
+        app.state.smart_cache = cache
+        app.state.smart_last_collected = now
+    return cache
+
+
 async def status_worker_loop(app, base_sleep: float = 5.0):
     """
     Background loop that continuously collects status data.
@@ -116,8 +150,11 @@ async def status_worker_loop(app, base_sleep: float = 5.0):
 
     while True:
         try:
+            # SMART on its own slow cadence; everything else every cycle.
+            smart_stats = await refresh_smart_stats(app, datetime.now())
+
             # Collect all status data
-            status = await collect_all_status()
+            status = await collect_all_status(smart_stats=smart_stats)
 
             # Store in app.state (thread-safe for reading in single-worker uvicorn)
             app.state.cached_status = status