health event instead of hw

kvmd/apps/kvmd/info/system.py

@@ -28,6 +28,7 @@ from typing import AsyncGenerator
 
 from ....logging import get_logger
 
+from .... import env
 from .... import aiotools
 from .... import aioproc
 
@@ -38,12 +39,30 @@ from .base import BaseInfoSubmanager
 
 # =====
 class SystemInfoSubmanager(BaseInfoSubmanager):
-    def __init__(self, streamer_cmd: list[str]) -> None:
+    def __init__(
+        self,
+        platform_path: str,
+        streamer_cmd: list[str],
+    ) -> None:
+
+        self.__platform_path = platform_path
         self.__streamer_cmd = streamer_cmd
+
+        self.__dt_cache: dict[str, str] = {}
         self.__notifier = aiotools.AioNotifier()
 
     async def get_state(self) -> dict:
-        streamer_info = await self.__get_streamer_info()
+        (
+            base,
+            serial,
+            pl,
+            streamer_info,
+        ) = await asyncio.gather(
+            self.__read_dt_file("model", upper=False),
+            self.__read_dt_file("serial-number", upper=True),
+            self.__read_platform_file(),
+            self.__get_streamer_info(),
+        )
         uname_info = platform.uname()  # Uname using the internal cache
         return {
             "kvmd": {"version": __version__},
@@ -52,6 +71,12 @@ class SystemInfoSubmanager(BaseInfoSubmanager):
                 field: getattr(uname_info, field)
                 for field in ["system", "release", "version", "machine"]
             },
+            "platform": {
+                "type": "rpi",
+                "base": base,
+                "serial": serial,
+                **pl,  # type: ignore
+            },
         }
 
     async def trigger_state(self) -> None:
@@ -64,6 +89,35 @@ class SystemInfoSubmanager(BaseInfoSubmanager):
 
     # =====
 
+    async def __read_dt_file(self, name: str, upper: bool) -> (str | None):
+        if name not in self.__dt_cache:
+            path = os.path.join(f"{env.PROCFS_PREFIX}/proc/device-tree", name)
+            try:
+                value = (await aiotools.read_file(path)).strip(" \t\r\n\0")
+                self.__dt_cache[name] = (value.upper() if upper else value)
+            except Exception as ex:
+                get_logger(0).error("Can't read DT %s from %s: %s", name, path, ex)
+                return None
+        return self.__dt_cache[name]
+
+    async def __read_platform_file(self) -> dict:
+        try:
+            text = await aiotools.read_file(self.__platform_path)
+            parsed: dict[str, str] = {}
+            for row in text.split("\n"):
+                row = row.strip()
+                if row:
+                    (key, value) = row.split("=", 1)
+                    parsed[key.strip()] = value.strip()
+            return {
+                "model": parsed["PIKVM_MODEL"],
+                "video": parsed["PIKVM_VIDEO"],
+                "board": parsed["PIKVM_BOARD"],
+            }
+        except Exception:
+            get_logger(0).exception("Can't read device model")
+            return {"model": None, "video": None, "board": None}
+
     async def __get_streamer_info(self) -> dict:
         version = ""
         features: dict[str, bool] = {}