feat(dbus): Add dphys-swapfile configuration functionality

Implemented `configure_dphys_swapfile` to manage the dphys-swapfile
service, including automatic swapfile removal when the service is
disabled. Updated feature registry to support dphys-swapfile
configuration.

src/revpi_middleware/dbus_middleware1/system_config/interface_config.py

@@ -3,12 +3,15 @@
 # SPDX-License-Identifier: GPL-2.0-or-later
 """D-Bus interfaces for hardware configuration."""
 from collections import namedtuple
-from enum import Enum
 from logging import getLogger
-from os import X_OK, access
-
-from pydbus import SystemBus
 
+from .revpi_config import (
+    ConfigActions,
+    configure_avahi_daemon,
+    configure_dphys_swapfile,
+    configure_gui,
+    simple_systemd,
+)
 from ..dbus_helper import DbusInterface
 
 log = getLogger(__name__)
@@ -16,13 +19,6 @@ log = getLogger(__name__)
 FeatureFunction = namedtuple("FeatureFunction", ["function", "args"])
 
 
-class ConfigActions(Enum):
-    ENABLE = "enable"
-    DISABLE = "disable"
-    STATUS = "status"
-    AVAILABLE = "available"
-
-
 class InterfaceRevpiConfig(DbusInterface):
     """
     <node>
@@ -71,28 +67,6 @@ class InterfaceRevpiConfig(DbusInterface):
         return list(AVAILABLE_FEATURES.keys())
 
 
-def configure_gui(action: ConfigActions):
-    gui_available = access("/usr/bin/startx", X_OK)
-
-    if action is ConfigActions.AVAILABLE:
-        return gui_available
-
-    bus = SystemBus()
-    systemd_manager = bus.get(".systemd1")
-
-    if action is ConfigActions.ENABLE:
-        systemd_manager.SetDefaultTarget("graphical.target", True)
-
-    elif action is ConfigActions.DISABLE:
-        systemd_manager.SetDefaultTarget("multi-user.target", True)
-
-    elif action is ConfigActions.STATUS:
-        return systemd_manager.GetDefaultTarget() == "graphical.target"
-
-    else:
-        raise ValueError(f"action {action} not supported")
-
-
 def get_feature(feature: str) -> FeatureFunction:
     if feature not in AVAILABLE_FEATURES:
         raise ValueError(f"feature {feature} does not exist")
@@ -102,48 +76,10 @@ def get_feature(feature: str) -> FeatureFunction:
     return feature_function
 
 
-def simple_systemd(action: ConfigActions, unit: str):
-    bus = SystemBus()
-    systemd_manager = bus.get(".systemd1")
-
-    if action is ConfigActions.ENABLE:
-        systemd_manager.UnmaskUnitFiles([unit], False)
-        systemd_manager.EnableUnitFiles([unit], False, False)
-        systemd_manager.StartUnit(unit, "replace")
-
-    elif action is ConfigActions.DISABLE:
-        systemd_manager.StopUnit(unit, "replace")
-        systemd_manager.DisableUnitFiles([unit], False)
-
-    elif action is ConfigActions.STATUS:
-        try:
-            unit_path = systemd_manager.LoadUnit(unit)
-            properties = bus.get(".systemd1", unit_path)
-        except Exception:
-            log.warning(f"could not get systemd unit {unit}")
-            return False
-
-        return properties.UnitFileState == "enabled" and properties.ActiveState == "active"
-
-    elif action is ConfigActions.AVAILABLE:
-        try:
-            unit_path = systemd_manager.LoadUnit(unit)
-            properties = bus.get(".systemd1", unit_path)
-        except Exception:
-            log.warning(f"could not get systemd unit {unit}")
-            return False
-
-        return properties.LoadState != "not-found"
-
-    else:
-        raise ValueError(f"action {action} not supported")
-
-
 AVAILABLE_FEATURES = {
     "gui": FeatureFunction(configure_gui, []),
     "revpi-con-can": False,
-    "var-log.mount": False,
+    "dphys-swapfile": FeatureFunction(configure_dphys_swapfile, []),
-    "dphys-swapfile": False,
     "pimodbus-master": FeatureFunction(simple_systemd, ["pimodbus-master.service"]),
     "pimodbus-slave": FeatureFunction(simple_systemd, ["pimodbus-slave.service"]),
     "systemd-timesyncd": FeatureFunction(simple_systemd, ["systemd-timesyncd.service"]),
@@ -155,6 +91,6 @@ AVAILABLE_FEATURES = {
     "revpipyload": FeatureFunction(simple_systemd, ["revpipyload.service"]),
     "bluetooth": False,
     "ieee80211": False,
-    "avahi": False,
+    "avahi": FeatureFunction(configure_avahi_daemon, []),
     "external-antenna": False,
 }

src/revpi_middleware/dbus_middleware1/system_config/revpi_config.py

@@ -0,0 +1,189 @@
+# -*- coding: utf-8 -*-
+# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
+# SPDX-License-Identifier: GPL-2.0-or-later
+import subprocess
+from enum import Enum, IntEnum
+from logging import getLogger
+from os import X_OK, access
+
+from pydbus import SystemBus
+
+from ..dbus_helper import grep
+
+log = getLogger(__name__)
+
+
+class ComputeModuleTypes(IntEnum):
+    UNKNOWN = 0
+    CM1 = 6
+    CM3 = 10
+    CM4 = 20
+    CM4S = 21
+    CM5 = 24
+
+
+class ConfigActions(Enum):
+    ENABLE = "enable"
+    DISABLE = "disable"
+    STATUS = "status"
+    AVAILABLE = "available"
+
+
+class RevPiConfig:
+
+    def __init__(self):
+        self._cm_type = ComputeModuleTypes.UNKNOWN
+        self._cm_with_wifi = False
+
+        self._revpi_with_con_bridge = False
+
+        self.serial = ""
+        self.model = ""
+
+        self._init_device_info()
+
+    def _init_device_info(self):
+        dc_cpuinfo = {}
+
+        # Extract CPU information
+        with open("/proc/cpuinfo", "r") as f:
+            line = "\n"
+            while line:
+                line = f.readline()
+                if line.startswith(("Revision", "Serial", "Model")):
+                    key, value = line.split(":", 1)
+                    key = key.strip().lower()
+                    value = value.strip()
+                    dc_cpuinfo[key] = value
+
+        self.model = dc_cpuinfo.get("model", "")
+        self.serial = dc_cpuinfo.get("serial", "")
+
+        # Detect Compute Module type
+        revision = dc_cpuinfo.get("revision", "")
+        if revision:
+            revision = int(revision, 16)
+            mask = 4080  # 0xFF0 in dezimal
+            try:
+                self._cm_type = ComputeModuleTypes((revision & mask) >> 4)
+            except ValueError:
+                pass
+
+        # Detect WiFi
+        could_have_wifi = self._cm_type in (ComputeModuleTypes.CM4, ComputeModuleTypes.CM5)
+        if could_have_wifi:
+            lst_grep = grep("DRIVER=brcmfmac", "/sys/class/ieee80211/phy0/device/uevent")
+            self._cm_with_wifi = len(lst_grep) > 0 and self._cm_type in (ComputeModuleTypes)
+
+        # Detect ConBridge
+        could_have_con_bridge = self._cm_type in (ComputeModuleTypes.CM3, ComputeModuleTypes.CM4S)
+        if could_have_con_bridge:
+            lst_grep = grep("kunbus,revpi-connect", "/proc/device-tree/compatible")
+            self._revpi_with_con_bridge = len(lst_grep) > 0
+
+    @property
+    def cm_type(self) -> ComputeModuleTypes:
+        return self._cm_type
+
+    @property
+    def with_con_bridge(self) -> bool:
+        return self._revpi_with_con_bridge
+
+    @property
+    def with_wifi(self) -> bool:
+        return self._cm_with_wifi
+
+
+def configure_avahi_daemon(action: ConfigActions):
+    return_value = simple_systemd(action, "avahi-daemon.service")
+
+    # Post actions for avahi-daemon
+    if action in (ConfigActions.ENABLE, ConfigActions.DISABLE):
+        # Apply the enable/disable action to the avahi socket AFTER the service
+        # unit, because a connected socket could interrupt stop
+        simple_systemd(action, "avahi-daemon.socket")
+
+    return return_value
+
+
+def configure_dphys_swapfile(action: ConfigActions):
+    return_value = simple_systemd(action, "dphys-swapfile.service")
+
+    # Post actions for dphys-swapfile
+    if action is ConfigActions.DISABLE:
+        # Remove swapfile afer disabling the service unit
+        subprocess.call(
+            ["/sbin/dphys-swapfile", "uninstall"],
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+
+    return return_value
+
+
+def configure_gui(action: ConfigActions):
+    gui_available = access("/usr/bin/startx", X_OK)
+
+    if action is ConfigActions.AVAILABLE:
+        return gui_available
+
+    bus = SystemBus()
+    systemd_manager = bus.get(".systemd1")
+
+    if action is ConfigActions.ENABLE:
+        systemd_manager.SetDefaultTarget("graphical.target", True)
+
+    elif action is ConfigActions.DISABLE:
+        systemd_manager.SetDefaultTarget("multi-user.target", True)
+
+    elif action is ConfigActions.STATUS:
+        return systemd_manager.GetDefaultTarget() == "graphical.target"
+
+    else:
+        raise ValueError(f"action {action} not supported")
+
+
+def simple_systemd(action: ConfigActions, unit: str):
+    bus = SystemBus()
+    systemd_manager = bus.get(".systemd1")
+
+    if action is ConfigActions.ENABLE:
+        systemd_manager.UnmaskUnitFiles([unit], False)
+        systemd_manager.EnableUnitFiles([unit], False, False)
+        systemd_manager.StartUnit(unit, "replace")
+
+    elif action is ConfigActions.DISABLE:
+        systemd_manager.StopUnit(unit, "replace")
+        systemd_manager.DisableUnitFiles([unit], False)
+
+    elif action is ConfigActions.STATUS:
+        try:
+            unit_path = systemd_manager.LoadUnit(unit)
+            properties = bus.get(".systemd1", unit_path)
+        except Exception:
+            log.warning(f"could not get systemd unit {unit}")
+            return False
+
+        return properties.UnitFileState == "enabled" and properties.ActiveState == "active"
+
+    elif action is ConfigActions.AVAILABLE:
+        try:
+            unit_path = systemd_manager.LoadUnit(unit)
+            properties = bus.get(".systemd1", unit_path)
+        except Exception:
+            log.warning(f"could not get systemd unit {unit}")
+            return False
+
+        return properties.LoadState != "not-found"
+
+    else:
+        raise ValueError(f"action {action} not supported")
+
+
+if __name__ == "__main__":
+    rc = RevPiConfig()
+    print("Model:", rc.model)
+    print("Serial:         ", rc.serial)
+    print("CM Type:        ", rc.cm_type.name)
+    print("With wifi:      ", rc.with_wifi)
+    print("With con-bridge:", rc.with_con_bridge)