Observe host hardware for realtime actions (#1530)

* Observe host hardware for realtime actions

* Better logging

* fix testenv

scripts/test_env.sh

@@ -117,8 +117,11 @@ function init_dbus() {
     mkdir -p /var/lib/dbus
     cp -f /etc/machine-id /var/lib/dbus/machine-id
 
-    # run
+    # cleanups
     mkdir -p /run/dbus
+    rm -f /run/dbus/pid
+
+    # run
     dbus-daemon --system --print-address
 }
 

supervisor/bootstrap.py

@@ -21,6 +21,7 @@ from .dns import CoreDNS
 from .hassos import HassOS
 from .homeassistant import HomeAssistant
 from .host import HostManager
+from .hwmon import HwMonitor
 from .ingress import Ingress
 from .services import ServiceManager
 from .snapshots import SnapshotManager
@@ -57,6 +58,7 @@ async def initialize_coresys():
     coresys.addons = AddonManager(coresys)
     coresys.snapshots = SnapshotManager(coresys)
     coresys.host = HostManager(coresys)
+    coresys.hwmonitor = HwMonitor(coresys)
     coresys.ingress = Ingress(coresys)
     coresys.tasks = Tasks(coresys)
     coresys.services = ServiceManager(coresys)

supervisor/core.py

@@ -142,6 +142,9 @@ class Core(CoreSysAttributes):
             if self.sys_homeassistant.version == "landingpage":
                 self.sys_create_task(self.sys_homeassistant.install())
 
+            # Start observe the host Hardware
+            await self.sys_hwmonitor.load()
+
             _LOGGER.info("Supervisor is up and running")
             self.state = CoreStates.RUNNING
 
@@ -168,6 +171,7 @@ class Core(CoreSysAttributes):
                         self.sys_websession_ssl.close(),
                         self.sys_ingress.unload(),
                         self.sys_dns.unload(),
+                        self.sys_hwmonitor.unload(),
                     ]
                 )
         except asyncio.TimeoutError:

supervisor/coresys.py

@@ -22,6 +22,7 @@ if TYPE_CHECKING:
     from .discovery import Discovery
     from .dns import CoreDNS
     from .hassos import HassOS
+    from .hwmon import HwMonitor
     from .homeassistant import HomeAssistant
     from .host import HostManager
     from .ingress import Ingress
@@ -76,6 +77,7 @@ class CoreSys:
         self._secrets: Optional[SecretsManager] = None
         self._store: Optional[StoreManager] = None
         self._discovery: Optional[Discovery] = None
+        self._hwmonitor: Optional[HwMonitor] = None
 
     @property
     def machine(self) -> str:
@@ -345,6 +347,18 @@ class CoreSys:
             raise RuntimeError("HostManager already set!")
         self._host = value
 
+    @property
+    def hwmonitor(self) -> HwMonitor:
+        """Return HwMonitor object."""
+        return self._hwmonitor
+
+    @hwmonitor.setter
+    def hwmonitor(self, value: HwMonitor):
+        """Set a HwMonitor object."""
+        if self._hwmonitor:
+            raise RuntimeError("HwMonitor already set!")
+        self._hwmonitor = value
+
     @property
     def ingress(self) -> Ingress:
         """Return Ingress object."""
@@ -520,6 +534,11 @@ class CoreSysAttributes:
         """Return HostManager object."""
         return self.coresys.host
 
+    @property
+    def sys_hwmonitor(self) -> HwMonitor:
+        """Return HwMonitor object."""
+        return self.coresys.hwmonitor
+
     @property
     def sys_ingress(self) -> Ingress:
         """Return Ingress object."""

supervisor/hwmon.py

@@ -0,0 +1,57 @@
+"""Supervisor Hardware monitor based on udev."""
+from datetime import timedelta
+import logging
+from pprint import pformat
+from typing import Optional
+
+import pyudev
+
+from .coresys import CoreSysAttributes, CoreSys
+from .utils import AsyncCallFilter
+
+_LOGGER: logging.Logger = logging.getLogger(__name__)
+
+
+class HwMonitor(CoreSysAttributes):
+    """Hardware monitor for supervisor."""
+
+    def __init__(self, coresys: CoreSys):
+        """Initialize Hardware Monitor object."""
+        self.coresys: CoreSys = coresys
+        self.context = pyudev.Context()
+        self.monitor = pyudev.Monitor.from_netlink(self.context)
+        self.observer: Optional[pyudev.MonitorObserver] = None
+
+    async def load(self) -> None:
+        """Start hardware monitor."""
+        self.observer = pyudev.MonitorObserver(self.monitor, self._udev_events)
+        self.observer.start()
+
+        _LOGGER.info("Start Supervisor hardware monitor")
+
+    async def unload(self) -> None:
+        """Shutdown sessions."""
+        if self.observer is None:
+            return
+        self.observer.stop()
+        _LOGGER.info("Stop Supervisor hardware monitor")
+
+    def _udev_events(self, action: str, device: pyudev.Device):
+        """Incomming events from udev.
+
+        This is inside a observe thread and need pass into our eventloop.
+        """
+        _LOGGER.debug("Hardware monitor: %s - %s", action, pformat(device))
+        self.sys_loop.call_soon_threadsafe(self._async_udev_events, action, device)
+
+    def _async_udev_events(self, action: str, device: pyudev.Device):
+        """Incomming events from udev into loop."""
+        # Sound changes
+        if device.subsystem == "sound":
+            self._action_sound(device)
+
+    @AsyncCallFilter(timedelta(seconds=5))
+    def _action_sound(self, device: pyudev.Device):
+        """Process sound actions."""
+        _LOGGER.info("Detect changed audio hardware")
+        self.sys_loop.call_later(5, self.sys_create_task, self.sys_host.sound.update())

supervisor/utils/__init__.py

@@ -36,7 +36,7 @@ def process_lock(method):
 class AsyncThrottle:
     """
     Decorator that prevents a function from being called more than once every
-    time period.
+    time period with blocking.
     """
 
     def __init__(self, delta):
@@ -64,6 +64,32 @@ class AsyncThrottle:
         return wrapper
 
 
+class AsyncCallFilter:
+    """
+    Decorator that prevents a function from being called more than once every
+    time period.
+    """
+
+    def __init__(self, delta):
+        """Initialize async throttle."""
+        self.throttle_period = delta
+        self.time_of_last_call = datetime.min
+
+    def __call__(self, method):
+        """Throttle function"""
+
+        async def wrapper(*args, **kwargs):
+            """Throttle function wrapper"""
+            now = datetime.now()
+            time_since_last_call = now - self.time_of_last_call
+
+            if time_since_last_call > self.throttle_period:
+                self.time_of_last_call = now
+                return await method(*args, **kwargs)
+
+        return wrapper
+
+
 def check_port(address: IPv4Address, port: int) -> bool:
     """Check if port is mapped."""
     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)