Add SSH metrics and watchdog control, moved from the Home Assistant integration

docs/ssh.md

@@ -24,3 +24,21 @@ async with NanoKVMSSH(
   phase.
 - Failures raise `NanoKVMSSHAuthenticationError`, `NanoKVMSSHConnectionError`
   (including host-key errors) or `NanoKVMSSHCommandError`.
+
+## Metrics and watchdog
+
+```python
+metrics = await ssh.get_metrics()
+metrics.cpu_temperature  # 45.8 (°C)
+```
+
+`get_metrics()` returns an `SSHMetrics` with `boot_time` (UTC),
+`cpu_temperature`, `memory_total` and `storage_total` (MiB, root filesystem),
+`memory_used_percent` and `storage_used_percent`, and `watchdog_enabled`. A
+field is `None` when the device output can't be parsed.
+
+The watchdog is the non-Pro firmware's own: `kvm_system` reboots the device
+when `kvm_vision` stops responding for more than 10 seconds, but only while
+`/etc/kvm/watchdog` exists. `get_watchdog_enabled()` returns `None` when the
+firmware has no watchdog (the Pro, or older non-Pro firmware), and
+`set_watchdog_enabled()` raises `NanoKVMNotSupportedError` there.

nanokvm/ssh_client.py

@@ -6,16 +6,22 @@
 from concurrent.futures import ThreadPoolExecutor
 import contextlib
 from dataclasses import dataclass, field
+from datetime import UTC, datetime
 from os import PathLike
 import threading
 from types import TracebackType
 
 import paramiko
 
-from .exceptions import NanoKVMError
+from .exceptions import NanoKVMError, NanoKVMNotSupportedError
 
 DEFAULT_SSH_USERNAME = "root"
 
+# The non-Pro kvm_system reboots the device when kvm_vision stops responding,
+# but only while this file exists. The Pro firmware has no such watchdog.
+_WATCHDOG_FILE = "/etc/kvm/watchdog"
+_WATCHDOG_SUPPORTED = f"grep -qs {_WATCHDOG_FILE} /kvmapp/kvm_system/kvm_system"
+
 
 class NanoKVMSSHError(NanoKVMError):
     """Base exception for SSH client errors."""
@@ -37,6 +43,23 @@ class NanoKVMSSHCommandError(NanoKVMSSHError):
     """Exception for SSH command execution errors."""
 
 
+@dataclass(frozen=True, slots=True)
+class SSHMetrics:
+    """Device metrics read over SSH; a field is None when it can't be read."""
+
+    boot_time: datetime | None = None
+    cpu_temperature: float | None = None
+    """Celsius."""
+    memory_total: float | None = None
+    """MiB."""
+    memory_used_percent: float | None = None
+    storage_total: float | None = None
+    """MiB, of the root filesystem."""
+    storage_used_percent: float | None = None
+    watchdog_enabled: bool | None = None
+    """None when the firmware has no watchdog (the Pro, or older non-Pro firmware)."""
+
+
 @dataclass
 class _CommandExecution:
     """Channel and cancellation state shared with one executor worker."""
@@ -215,6 +238,48 @@ async def run_command(self, command: str, timeout: float = 30) -> str:
             self._invalidate_client(client)
             raise NanoKVMSSHCommandError(f"SSH command failed: {e}") from e
 
+    async def get_metrics(self) -> SSHMetrics:
+        """Read boot time, CPU temperature, memory, storage and watchdog state."""
+        stat = await self.run_command("cat /proc/stat")
+        temperature = await self.run_command(
+            "cat /sys/class/thermal/thermal_zone0/temp 2>/dev/null || true"
+        )
+        meminfo = await self.run_command("cat /proc/meminfo")
+        df = await self.run_command("df -k /")
+        memory_total, memory_used_percent = _parse_meminfo(meminfo)
+        storage_total, storage_used_percent = _parse_df(df)
+        return SSHMetrics(
+            boot_time=_parse_boot_time(stat),
+            cpu_temperature=_parse_temperature(temperature),
+            memory_total=memory_total,
+            memory_used_percent=memory_used_percent,
+            storage_total=storage_total,
+            storage_used_percent=storage_used_percent,
+            watchdog_enabled=await self.get_watchdog_enabled(),
+        )
+
+    async def get_watchdog_enabled(self) -> bool | None:
+        """Return the watchdog state, or None when the firmware has none."""
+        # ponytail: greps the 380 KB kvm_system binary on every call (~0.3 s on
+        # the PCIe); cache per connection if polling gets tight.
+        output = await self.run_command(
+            f"{_WATCHDOG_SUPPORTED} || {{ echo -; exit; }}; "
+            f"test -f {_WATCHDOG_FILE} && echo 1 || echo 0"
+        )
+        return None if output == "-" else output == "1"
+
+    async def set_watchdog_enabled(self, enabled: bool) -> None:
+        """Enable or disable the watchdog.
+
+        Raises ``NanoKVMNotSupportedError`` when the firmware has none.
+        """
+        change = f"touch {_WATCHDOG_FILE}" if enabled else f"rm -f {_WATCHDOG_FILE}"
+        output = await self.run_command(
+            f"{_WATCHDOG_SUPPORTED} || {{ echo -; exit; }}; {change}"
+        )
+        if output == "-":
+            raise NanoKVMNotSupportedError("This NanoKVM firmware has no watchdog")
+
     def _invalidate_client(self, client: paramiko.SSHClient) -> None:
         """Drop a failed client unless a newer connection replaced it."""
         if self.ssh_client is client:
@@ -245,3 +310,51 @@ def _exec_command_sync(
             return output, error, channel.recv_exit_status()
         finally:
             channel.close()
+
+
+def _parse_boot_time(stat: str) -> datetime | None:
+    for line in stat.splitlines():
+        parts = line.split()
+        if len(parts) == 2 and parts[0] == "btime" and parts[1].isdigit():
+            return datetime.fromtimestamp(int(parts[1]), UTC)
+    return None
+
+
+def _parse_temperature(output: str) -> float | None:
+    try:
+        value = float(output)
+    except ValueError:
+        return None
+    # Linux reports millidegrees; some kernels report degrees.
+    if value > 1000:
+        value /= 1000
+    return round(value, 1)
+
+
+def _parse_meminfo(meminfo: str) -> tuple[float | None, float | None]:
+    values: dict[str, int] = {}
+    for line in meminfo.splitlines():
+        parts = line.split()
+        if len(parts) >= 2 and parts[1].isdigit():
+            values[parts[0].rstrip(":")] = int(parts[1])
+    total_kb = values.get("MemTotal")
+    if not total_kb:
+        return None, None
+    available_kb = values.get("MemAvailable")
+    used_percent = (
+        None
+        if available_kb is None
+        else round((total_kb - available_kb) / total_kb * 100, 2)
+    )
+    return round(total_kb / 1024, 2), used_percent
+
+
+def _parse_df(df: str) -> tuple[float | None, float | None]:
+    lines = df.splitlines()
+    if len(lines) < 2:
+        return None, None
+    parts = lines[1].split()
+    try:
+        return round(int(parts[1]) / 1024, 2), float(parts[4].rstrip("%"))
+    except (IndexError, ValueError):
+        return None, None

tests/test_ssh_client.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+from datetime import UTC, datetime
 from pathlib import Path
 import threading
 from typing import Any
@@ -11,12 +12,14 @@
 import paramiko
 import pytest
 
+from nanokvm.exceptions import NanoKVMNotSupportedError
 from nanokvm.ssh_client import (
     NanoKVMSSH,
     NanoKVMSSHAuthenticationError,
     NanoKVMSSHCommandError,
     NanoKVMSSHConnectionError,
     NanoKVMSSHNotConnectedError,
+    SSHMetrics,
 )
 
 
@@ -495,3 +498,54 @@ async def test_async_context_manager_disconnects_client() -> None:
 
     assert client.ssh_client is None
     fake_client.close.assert_called_once_with()
+
+
+_METRICS_OUTPUT = {
+    "cat /proc/stat": "cpu  1 2 3\nbtime 1761600000\nprocesses 42",
+    "cat /sys/class/thermal/thermal_zone0/temp 2>/dev/null || true": "45843",
+    "cat /proc/meminfo": "MemTotal:  262144 kB\nMemAvailable: 65536 kB\n",
+    "df -k /": (
+        "Filesystem 1K-blocks Used Available Use% Mounted on\n"
+        "/dev/mmcblk0p2 7923848 1456520 6115884 19% /"
+    ),
+}
+
+
+async def _metrics(outputs: dict[str, str], watchdog: str) -> Any:
+    ssh = NanoKVMSSH("kvm.local")
+
+    async def run_command(command: str, timeout: float = 30) -> str:
+        return watchdog if "watchdog" in command else outputs[command]
+
+    with patch.object(ssh, "run_command", side_effect=run_command):
+        return await ssh.get_metrics()
+
+
+async def test_get_metrics_parses_device_output() -> None:
+    metrics = await _metrics(_METRICS_OUTPUT, "1")
+
+    assert metrics.boot_time == datetime.fromtimestamp(1761600000, UTC)
+    assert metrics.cpu_temperature == 45.8
+    assert metrics.memory_total == 256
+    assert metrics.memory_used_percent == 75
+    assert metrics.storage_total == 7738.13
+    assert metrics.storage_used_percent == 19
+    assert metrics.watchdog_enabled is True
+
+
+async def test_get_metrics_returns_none_for_unreadable_values() -> None:
+    outputs = dict.fromkeys(_METRICS_OUTPUT, "garbage")
+
+    metrics = await _metrics(outputs, "-")
+
+    assert metrics == SSHMetrics()
+
+
+async def test_set_watchdog_enabled_rejects_firmware_without_watchdog() -> None:
+    ssh = NanoKVMSSH("kvm.local")
+
+    with (
+        patch.object(ssh, "run_command", return_value="-"),
+        pytest.raises(NanoKVMNotSupportedError),
+    ):
+        await ssh.set_watchdog_enabled(True)