From 38f495e864a2e21d49f4bfdf39a17080323d49f3 Mon Sep 17 00:00:00 2001 From: Sven Sager Date: Sun, 20 Apr 2025 15:42:05 +0200 Subject: [PATCH] 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. --- .../system_config/interface_config.py | 71 +------------------ .../system_config/revpi_config.py | 71 ++++++++++++++++++- 2 files changed, 71 insertions(+), 71 deletions(-) diff --git a/src/revpi_middleware/dbus_middleware1/system_config/interface_config.py b/src/revpi_middleware/dbus_middleware1/system_config/interface_config.py index 76c7933..b1aa68c 100644 --- a/src/revpi_middleware/dbus_middleware1/system_config/interface_config.py +++ b/src/revpi_middleware/dbus_middleware1/system_config/interface_config.py @@ -3,12 +3,9 @@ # 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_gui, simple_systemd from ..dbus_helper import DbusInterface log = getLogger(__name__) @@ -16,13 +13,6 @@ log = getLogger(__name__) FeatureFunction = namedtuple("FeatureFunction", ["function", "args"]) -class ConfigActions(Enum): - ENABLE = "enable" - DISABLE = "disable" - STATUS = "status" - AVAILABLE = "available" - - class InterfaceRevpiConfig(DbusInterface): """ @@ -71,28 +61,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,43 +70,6 @@ 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, diff --git a/src/revpi_middleware/dbus_middleware1/system_config/revpi_config.py b/src/revpi_middleware/dbus_middleware1/system_config/revpi_config.py index 12ef5ee..40b57a7 100644 --- a/src/revpi_middleware/dbus_middleware1/system_config/revpi_config.py +++ b/src/revpi_middleware/dbus_middleware1/system_config/revpi_config.py @@ -1,8 +1,11 @@ # -*- coding: utf-8 -*- # SPDX-FileCopyrightText: 2025 KUNBUS GmbH # SPDX-License-Identifier: GPL-2.0-or-later -from enum import IntEnum +from enum import Enum, IntEnum from logging import getLogger +from os import X_OK, access + +from pydbus import SystemBus from ..dbus_helper import grep @@ -18,6 +21,13 @@ class ComputeModuleTypes(IntEnum): CM5 = 24 +class ConfigActions(Enum): + ENABLE = "enable" + DISABLE = "disable" + STATUS = "status" + AVAILABLE = "available" + + class RevPiConfig: def __init__(self): @@ -83,6 +93,65 @@ class RevPiConfig: 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__": rc = RevPiConfig() print("Model:", rc.model)