refactor(dbus): Move system configuration methods to revpi_config.py

Centralized `ConfigActions`, `configure_gui`, and `simple_systemd` into
`revpi_config.py` to eliminate duplication and improve maintainability.
Updated `interface_config.py` to import these methods, ensuring
consistent functionality across modules.
This commit is contained in:
2025-04-20 15:42:05 +02:00
parent 85dc6a7e56
commit 38f495e864
2 changed files with 71 additions and 71 deletions

View File

@@ -3,12 +3,9 @@
# SPDX-License-Identifier: GPL-2.0-or-later # SPDX-License-Identifier: GPL-2.0-or-later
"""D-Bus interfaces for hardware configuration.""" """D-Bus interfaces for hardware configuration."""
from collections import namedtuple from collections import namedtuple
from enum import Enum
from logging import getLogger from logging import getLogger
from os import X_OK, access
from pydbus import SystemBus
from .revpi_config import ConfigActions, configure_gui, simple_systemd
from ..dbus_helper import DbusInterface from ..dbus_helper import DbusInterface
log = getLogger(__name__) log = getLogger(__name__)
@@ -16,13 +13,6 @@ log = getLogger(__name__)
FeatureFunction = namedtuple("FeatureFunction", ["function", "args"]) FeatureFunction = namedtuple("FeatureFunction", ["function", "args"])
class ConfigActions(Enum):
ENABLE = "enable"
DISABLE = "disable"
STATUS = "status"
AVAILABLE = "available"
class InterfaceRevpiConfig(DbusInterface): class InterfaceRevpiConfig(DbusInterface):
""" """
<node> <node>
@@ -71,28 +61,6 @@ class InterfaceRevpiConfig(DbusInterface):
return list(AVAILABLE_FEATURES.keys()) 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: def get_feature(feature: str) -> FeatureFunction:
if feature not in AVAILABLE_FEATURES: if feature not in AVAILABLE_FEATURES:
raise ValueError(f"feature {feature} does not exist") raise ValueError(f"feature {feature} does not exist")
@@ -102,43 +70,6 @@ def get_feature(feature: str) -> FeatureFunction:
return feature_function 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 = { AVAILABLE_FEATURES = {
"gui": FeatureFunction(configure_gui, []), "gui": FeatureFunction(configure_gui, []),
"revpi-con-can": False, "revpi-con-can": False,

View File

@@ -1,8 +1,11 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH # SPDX-FileCopyrightText: 2025 KUNBUS GmbH
# SPDX-License-Identifier: GPL-2.0-or-later # SPDX-License-Identifier: GPL-2.0-or-later
from enum import IntEnum from enum import Enum, IntEnum
from logging import getLogger from logging import getLogger
from os import X_OK, access
from pydbus import SystemBus
from ..dbus_helper import grep from ..dbus_helper import grep
@@ -18,6 +21,13 @@ class ComputeModuleTypes(IntEnum):
CM5 = 24 CM5 = 24
class ConfigActions(Enum):
ENABLE = "enable"
DISABLE = "disable"
STATUS = "status"
AVAILABLE = "available"
class RevPiConfig: class RevPiConfig:
def __init__(self): def __init__(self):
@@ -83,6 +93,65 @@ class RevPiConfig:
return self._cm_with_wifi return self._cm_with_wifi
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__": if __name__ == "__main__":
rc = RevPiConfig() rc = RevPiConfig()
print("Model:", rc.model) print("Model:", rc.model)