Update changelog for 0.0.2-1+deb12+1 release

Release version 0.0.2

data/dbus-policy/com.revolutionpi.ios1.conf

@@ -1,22 +0,0 @@
-<!-- /usr/share/dbus-1/system.d/com.revolutionpi.ios1.conf -->
-<busconfig>
-    <!-- Allow full access to root as the bus owner -->
-    <policy user="root">
-        <allow own="com.revolutionpi.ios1"/>
-        <allow send_destination="com.revolutionpi.ios1"/>
-        <allow receive_sender="com.revolutionpi.ios1"/>
-    </policy>
-
-    <!-- System group picontrol -->
-    <policy group="picontrol">
-        <allow send_destination="com.revolutionpi.ios1"
-               send_interface="org.freedesktop.DBus.Introspectable"/>
-        <allow send_destination="com.revolutionpi.ios1"
-               send_interface="com.revolutionpi.middleware1.PiControl"/>
-    </policy>
-
-    <!-- Standard-Policy -->
-    <policy context="default">
-        <deny send_destination="com.revolutionpi.ios1"/>
-    </policy>
-</busconfig>

data/dbus-policy/com.revolutionpi.middleware1.conf

@@ -1,4 +1,4 @@
-<!-- /usr/share/dbus-1/system.d/com.revolutionpi.middleware1.conf -->
+<!-- /etc/dbus-1/system.d/revpi-middleware.conf -->
 <busconfig>
     <!-- Allow full access to root as the bus owner -->
     <policy user="root">
@@ -12,7 +12,7 @@
         <allow send_destination="com.revolutionpi.middleware1"
                send_interface="org.freedesktop.DBus.Introspectable"/>
         <allow send_destination="com.revolutionpi.middleware1"
-               send_interface="com.revolutionpi.middleware1.PiControl"/>
+               send_interface="com.revolutionpi.middleware1.picontrol"/>
     </policy>
 
     <!-- Standard-Policy -->

data/examples/io_controller.py

@@ -1,46 +0,0 @@
-from time import time
-
-from gi.repository import GLib
-from pydbus import SystemBus
-
-detected_signal = False
-timestamp = time()
-
-bus = SystemBus()
-bus_revpi = bus.get("com.revolutionpi.ios1", "/com/revolutionpi/ios1")
-interface = bus_revpi["com.revolutionpi.ios1.IoManager"]
-
-io_o_1_path = interface.Get("O_1")
-io_revpiled_path = interface.Get("RevPiLED")
-
-io_RevPiLED = bus.get("com.revolutionpi.ios1", io_revpiled_path)
-io_O_1 = bus.get("com.revolutionpi.ios1", io_o_1_path)
-
-
-def signal_handler(io_name, io_value):
-    global timestamp
-
-    print(f"Signal received: {io_name} = {io_value}")
-    if io_name == "O_1":
-        timestamp = time()
-        io_RevPiLED.Set("com.revolutionpi.ios1.OutInt", "value", GLib.Variant("i", int(io_value)))
-
-    elif io_name == "I_1":
-        io_RevPiLED.Set("com.revolutionpi.ios1.OutInt", "value", GLib.Variant("i", int(io_value)))
-        io_O_1.Set("com.revolutionpi.ios1.OutBool", "value", GLib.Variant("b", not io_value))
-        print(time() - timestamp)
-
-
-# Start event detection
-interface.ActivateIoEvents()
-
-interface.onIoChanged = signal_handler
-
-try:
-    loop = GLib.MainLoop()
-    loop.run()
-except KeyboardInterrupt:
-    pass
-
-# Stop event detection
-interface.DeactivateIoEvents()

debian/changelog

@@ -0,0 +1,48 @@
+revpi-middleware (0.0.2-1+deb12+1) bookworm; urgency=medium
+
+  * refactor(dbus): Move D-Bus helper functions to a dedicated file
+  * refactor(dbus): Move ResetDriverWatchdog to process_image_helper.py
+  * refactor(dbus): Parameterize `picontrol_device` and `config_rsc`
+  * feat: Add session bus option for local testing and development
+  * feat(dbus): Add import for BusProvider in dbus_middleware1 module
+  * feat(dbus): Add `running` property to `BusProvider`
+  * refactor(cli): D-Bus helpers support session and system bus types
+  * fix(dbus): Add error handling for DBus publishing and main loop
+  * refactor(dbus): piControl driver reset with PiControlIoctl class
+  * test(dbus): Add unit test framework for dbus_middleware1 module
+  * test(dbus): Add unit tests for PiControl D-Bus interface
+  * refactor(dbus): Fix typo and remove unused thread instance
+  * refactor(dbus): D-Bus interface management with cleanup support.
+  * test(dbus): Add support for testing driver reset notification
+  * feat(deb): Add dbus for testing to build dependencies
+  * fix(deb): Skip tests because of missing SystemBus in build container
+
+ -- Sven Sager <s.sager@kunbus.com>  Sat, 19 Apr 2025 16:34:20 +0200
+
+revpi-middleware (0.0.1-1+deb12+1) bookworm; urgency=medium
+
+  * docs: Start git project with python git-ignore and Readme
+  * docs: Use 'reuse' for SPDX Headers and Licenses
+  * feat: Add python base project files
+  * feat: Add proginit application basic module
+  * feat: Add the data directory for additional data files for the project
+  * test: Add tests directory with a dummy test
+  * build: Add all necessary files for the build system
+  * feat: Add dummy main application script
+  * feat: Add systemd file and data to integrate the app as a daemon
+  * chore: Update proginit to 1.4.0
+  * feat(dbus): Add ResetDriverWatchdog helper as global dbus helper
+  * feat(process_image): Add D-Bus interface for piControl driver
+  * feat(dbus): Add initial D-Bus middleware implementation
+  * feat(dbus): Add `extend_interface` function for dynamic interface naming
+  * feat(dbus): Add DBus policy configuration for revpi-middleware
+  * feat: Add MiddlewareDaemon implementation to revpi-middleware
+  * feat: Add daemon mode and signal handling to main application
+  * feat(cli): Add D-Bus helper functions for CLI commands.
+  * feat(cli): Add await_signal function to handle D-Bus signals
+  * feat(cli): Add `await-reset` to wait for piControl reset signal
+  * chore(build): Update requirements for this project
+  * feat(cli): Add new CLI tool entry point for `revpictl`
+  * feat(deb): Start packaging branch
+
+ -- Sven Sager <s.sager@kunbus.com>  Fri, 18 Apr 2025 19:02:20 +0200

debian/control

@@ -0,0 +1,38 @@
+Source: revpi-middleware
+Section: python
+Priority: optional
+Maintainer: KUNBUS GmbH <support@kunbus.com>
+Rules-Requires-Root: no
+Homepage: https://revolutionpi.com/
+Vcs-Browser: https://gitlab.com/revolutionpi/revpi-middleware
+Vcs-Git: https://gitlab.com/revolutionpi/revpi-middleware.git -b debian/bookworm
+Build-Depends:
+  dbus,
+  dbus-x11,
+  debhelper-compat (= 13),
+  dh-python,
+  python3-all,
+  python3-gi (>= 3.42.2),
+  python3-pydbus (>= 0.6.0),
+  python3-setuptools,
+Standards-Version: 4.6.2
+
+Package: revpi-middleware
+Architecture: all
+Pre-Depends: ${misc:Pre-Depends}
+Depends:
+  ${python3:Depends},
+  ${misc:Depends}
+Description: Revolution Pi middleware with D-Bus interface
+ The Revolution Pi middleware provides a robust communication interface for
+ Revolution Pi industrial computers. It enables seamless integration between
+ hardware components and applications through a D-Bus interface. The middleware
+ serves as a bridge for data exchange, device configuration, and system
+ monitoring.
+ .
+ Key features:
+  * Hardware abstraction layer for Revolution Pi I/O modules
+  * Real-time data processing and event handling
+  * Simplified API for accessing Revolution Pi hardware features
+  * Extensive configuration options for industrial automation tasks
+  * Built-in monitoring and diagnostic capabilities

debian/copyright

@@ -0,0 +1,27 @@
+Format: https://www.debian.org/doc/packaging-manuals/copyright-format/1.0/
+Source: https://gitlab.com/revolutionpi/opcua-revpi-server
+
+Files: *
+Copyright: 2025 KUNBUS GmbH
+License: GPL-2+
+
+Files: debian/*
+Copyright: 2025 KUNBUS GmbH
+License: GPL-2+
+
+License: GPL-2+
+ This package is free software; you can redistribute it and/or modify
+ it under the terms of the GNU General Public License as published by
+ the Free Software Foundation; either version 2 of the License, or
+ (at your option) any later version.
+ .
+ This package is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ GNU General Public License for more details.
+ .
+ You should have received a copy of the GNU General Public License
+ along with this program. If not, see <https://www.gnu.org/licenses/>
+ .
+ On Debian systems, the complete text of the GNU General
+ Public License version 2 can be found in "/usr/share/common-licenses/GPL-2".

debian/gbp.conf

@@ -0,0 +1,7 @@
+[DEFAULT]
+upstream-branch = main
+upstream-tag = v%(version)s
+debian-branch=debian/bookworm
+debian-tag = debian/%(version)s
+debian-tag-msg = %(pkg)s Debian release %(version)s
+pristine-tar = True

debian/revpi-middleware.install

@@ -0,0 +1,4 @@
+data/dbus-policy/com.revolutionpi.middleware1.conf      /usr/share/dbus-1/system.d
+data/etc/default/revpi-middleware                       /etc/default/
+data/etc/revpi-middleware/revpi-middleware.conf         /etc/revpi-middleware/
+data/systemd/before_253/revpi-middleware.service        /lib/systemd/system/

debian/revpi-middleware.links

@@ -0,0 +1,2 @@
+/usr/share/revpi-middleware/revpi-middleware	/usr/sbin/revpi-middleware
+/usr/share/revpi-middleware/revpicli	/usr/bin/revpicli

debian/rules

@@ -0,0 +1,12 @@
+#!/usr/bin/make -f
+
+export PYBUILD_NAME=revpi-middleware
+export PYBUILD_INSTALL_ARGS=--install-lib=/usr/share/$(PYBUILD_NAME)/ --install-scripts=/usr/share/$(PYBUILD_NAME)/
+
+%:
+	dh $@ --with python3 --buildsystem=pybuild
+
+override_dh_auto_test:
+	# Currently, the tests have to be skipped, because no SystemBus is
+	# available in the Docker container.
+	@echo "Skipped tests"

debian/source/format

@@ -0,0 +1 @@
+3.0 (quilt)

src/revpi_middleware/cli_commands/cli_base.py

@@ -5,11 +5,12 @@
 This module provides the foundation for the RevPi middleware CLI commands
 and argument parsing setup.
 """
+from argparse import ArgumentParser
 from logging import getLogger
 
-from . import cli_config, cli_picontrol
+from revpi_middleware.cli_commands import cli_picontrol
+from revpi_middleware.proginit import StdLogOutput
 from .. import proginit as pi
-from ..proginit import StdLogOutput
 
 log = getLogger(__name__)
 
@@ -29,11 +30,6 @@ def setup_command_line_arguments():
         help="RevPi PiControl object",
     )
     cli_picontrol.add_subparsers(obj_picontrol)
-    obj_config = rpictl_obj.add_parser(
-        "config",
-        help="RevPi configuration object (revpi-config)",
-    )
-    cli_config.add_subparsers(obj_config)
 
 
 def main() -> int:
@@ -44,9 +40,6 @@ def main() -> int:
     if obj == "picontrol":
         rc = cli_picontrol.main()
 
-    elif obj == "config":
-        rc = cli_config.main()
-
     else:
         log.error(f"Unknown object: {obj}")
         rc = 1

src/revpi_middleware/cli_commands/cli_config.py

@@ -1,93 +0,0 @@
-# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
-# SPDX-License-Identifier: GPL-2.0-or-later
-"""Command-Line for the picontrol object of CLI."""
-from argparse import ArgumentParser
-from logging import getLogger
-
-from .dbus_helper import BusType, get_properties, simple_call
-from .. import proginit as pi
-from ..dbus_middleware1 import extend_interface
-
-log = getLogger(__name__)
-
-
-def add_subparsers(parent_parser: ArgumentParser):
-    parent_parser.add_argument(
-        "action",
-        choices=["enable", "disable", "status", "available", "list-features"],
-        help="Action to be executed: enable, disable, status or available. "
-             "To get all available features, use 'list-features'.",
-    )
-    parent_parser.add_argument(
-        "feature",
-        nargs="?",
-        default="",
-        help="Name of the feature to configer. To list all features use 'list-features' as action.",
-    )
-
-
-def main() -> int:
-    action = pi.pargs.action
-    dbus_value = False
-    try:
-
-        if action == "list-features":
-            dbus_value = get_properties(
-                "available_features",
-                interface=extend_interface("RevpiConfig"),
-                bus_type=BusType.SESSION if pi.pargs.use_session_bus else BusType.SYSTEM,
-            )
-            for feature in dbus_value:
-                print(feature)
-
-            return 0
-
-        # For the following actions, a feature name is required
-        if pi.pargs.feature == "":
-            raise Exception("Feature name is required")
-
-        if action == "enable":
-            simple_call(
-                "Enable",
-                pi.pargs.feature,
-                interface=extend_interface("RevpiConfig"),
-                bus_type=BusType.SESSION if pi.pargs.use_session_bus else BusType.SYSTEM,
-            )
-
-        elif action == "disable":
-            simple_call(
-                "Disable",
-                pi.pargs.feature,
-                interface=extend_interface("RevpiConfig"),
-                bus_type=BusType.SESSION if pi.pargs.use_session_bus else BusType.SYSTEM,
-            )
-
-        elif action == "status":
-            dbus_value = simple_call(
-                "GetStatus",
-                pi.pargs.feature,
-                interface=extend_interface("RevpiConfig"),
-                bus_type=BusType.SESSION if pi.pargs.use_session_bus else BusType.SYSTEM,
-            )
-
-        elif action == "available":
-            dbus_value = simple_call(
-                "GetAvailability",
-                pi.pargs.feature,
-                interface=extend_interface("RevpiConfig"),
-                bus_type=BusType.SESSION if pi.pargs.use_session_bus else BusType.SYSTEM,
-            )
-
-        else:
-            raise Exception(f"Unknown action: {action}")
-
-    except Exception as e:
-        log.error(f"Error: {e}")
-        return 1
-
-    log.debug(
-        f"D-Bus call of method {action} for feature {pi.pargs.feature} returned: {dbus_value}"
-    )
-    print(int(dbus_value))
-
-    return 0

src/revpi_middleware/cli_commands/cli_picontrol.py

@@ -38,7 +38,7 @@ def method_reset():
     log.debug("D-Bus call of method ResetDriver")
     simple_call(
         "ResetDriver",
-        interface=extend_interface("PiControl"),
+        interface=extend_interface("picontrol"),
         bus_type=BusType.SESSION if pi.pargs.use_session_bus else BusType.SYSTEM,
     )
     log.info("ResetDriver called via D-Bus")
@@ -48,7 +48,7 @@ def method_await_reset(timeout: int = 0):
     detected_signal = await_signal(
         "NotifyDriverReset",
         timeout,
-        extend_interface("PiControl"),
+        extend_interface("picontrol"),
         bus_type=BusType.SESSION if pi.pargs.use_session_bus else BusType.SYSTEM,
     )
     if detected_signal:

src/revpi_middleware/cli_commands/dbus_helper.py

@@ -17,18 +17,6 @@ class BusType(Enum):
     SYSTEM = "system"
 
 
-def get_properties(
-    property_name: str,
-    interface: str,
-    object_path=REVPI_DBUS_BASE_PATH,
-    bus_type=BusType.SYSTEM,
-):
-    bus = SessionBus() if bus_type is BusType.SESSION else SystemBus()
-    revpi = bus.get(REVPI_DBUS_NAME, object_path)
-    iface = revpi[interface]
-    return getattr(iface, property_name)
-
-
 def simple_call(
     method: str,
     *args,

src/revpi_middleware/daemon.py

@@ -9,7 +9,6 @@ from time import perf_counter
 
 from . import proginit as pi
 from .dbus_middleware1.bus_provider import BusProvider
-from .ios1.bus_provider_io import BusProviderIo
 
 log = getLogger(__name__)
 
@@ -27,7 +26,6 @@ class MiddlewareDaemon:
         self._running = True
 
         self.bus_provider = None
-        self.io_bus_provider = None
 
         self._configure()
         log.debug("leave MiddlewareDaemon.__init__")
@@ -41,16 +39,11 @@ class MiddlewareDaemon:
 
     def dbus_start(self):
         log.debug("enter MiddlewareDaemon.dbus_start")
+        if self.bus_provider and self.bus_provider.is_alive():
+            return
 
-        dbus_middleware_running = self.bus_provider and self.bus_provider.is_alive()
-        if not dbus_middleware_running:
-            self.bus_provider = BusProvider(use_system_bus=not pi.pargs.use_session_bus)
-            self.bus_provider.start()
-
-        dbus_io_middleware_running = self.io_bus_provider and self.io_bus_provider.is_alive()
-        if not dbus_io_middleware_running:
-            self.io_bus_provider = BusProviderIo(use_system_bus=not pi.pargs.use_session_bus)
-            self.io_bus_provider.start()
+        self.bus_provider = BusProvider(use_system_bus=not pi.pargs.use_session_bus)
+        self.bus_provider.start()
 
         log.debug("leave MiddlewareDaemon.dbus_start")
 
@@ -63,12 +56,6 @@ class MiddlewareDaemon:
             if self.bus_provider.is_alive():
                 log.warning("dbus provider thread is still alive")
 
-        if self.io_bus_provider:
-            self.io_bus_provider.stop()
-            self.io_bus_provider.join(timeout=10.0)
-            if self.io_bus_provider.is_alive():
-                log.warning("dbus io provider thread is still alive")
-
         log.debug("leave MiddlewareDaemon.dbus_stop")
 
     def reload_config(self) -> None:

src/revpi_middleware/dbus_middleware1/bus_provider.py

@@ -10,7 +10,6 @@ from pydbus import SessionBus, SystemBus
 
 from . import REVPI_DBUS_NAME
 from .process_image import InterfacePiControl
-from .system_config import InterfaceRevpiConfig
 
 log = getLogger(__name__)
 
@@ -42,7 +41,6 @@ class BusProvider(Thread):
         #   ("Subdir2/Whatever", Example())
         lst_interfaces = [
             InterfacePiControl(self.picontrol_device, self.config_rsc),
-            InterfaceRevpiConfig(),
         ]
 
         try:

src/revpi_middleware/dbus_middleware1/dbus_helper.py

@@ -41,38 +41,3 @@ def extend_interface(*args) -> str:
             the provided segments.
     """
     return ".".join([REVPI_DBUS_NAME, *args])
-
-
-def grep(pattern, filename):
-    """
-    Searches for lines in a file that contain a given pattern and returns them as a list.
-
-    The function reads lines from the specified file and checks whether each line
-    contains the provided pattern. It returns a list of lines that match the
-    pattern. If the file is not found, an empty list is returned. Any other
-    exceptions during the file reading process are caught and logged.
-
-    Args:
-        pattern (str): The substring to search for within the file's lines.
-        filename (str): The path to the file that will be searched.
-
-    Returns:
-        list[str]: A list containing lines that include the provided pattern, with
-        leading and trailing spaces removed.
-
-    Raises:
-        FileNotFoundError: This error is caught if the file specified is not
-        found.
-        Exception: Any unforeseen exception during file operations is caught and
-        logged.
-    """
-    try:
-        with open(filename, "r") as file:
-            # Gibt alle Zeilen zurück, die das Muster enthalten
-            matching_lines = [line.strip() for line in file if pattern in line]
-            return matching_lines
-    except FileNotFoundError:
-        return []
-    except Exception as e:
-        log.error(f"Error reading file: {e}")
-        return []

src/revpi_middleware/dbus_middleware1/process_image/interface_picontrol.py

@@ -15,7 +15,7 @@ log = getLogger(__name__)
 class InterfacePiControl(DbusInterface):
     """
     <node>
-        <interface name='com.revolutionpi.middleware1.PiControl'>
+        <interface name='com.revolutionpi.middleware1.picontrol'>
             <signal name="NotifyDriverReset">
             </signal>
             <method name='ResetDriver'>

src/revpi_middleware/dbus_middleware1/system_config/__init__.py

@@ -1,5 +0,0 @@
-# -*- coding: utf-8 -*-
-# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
-# SPDX-License-Identifier: GPL-2.0-or-later
-"""D-Bus interfaces for system configuration."""
-from .interface_config import InterfaceRevpiConfig

src/revpi_middleware/dbus_middleware1/system_config/interface_config.py

@@ -1,100 +0,0 @@
-# -*- coding: utf-8 -*-
-# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
-# SPDX-License-Identifier: GPL-2.0-or-later
-"""D-Bus interfaces for hardware configuration."""
-from collections import namedtuple
-from logging import getLogger
-
-from .revpi_config import (
-    ConfigActions,
-    configure_avahi_daemon,
-    configure_bluetooth,
-    configure_con_can,
-    configure_dphys_swapfile,
-    configure_external_antenna,
-    configure_gui,
-    configure_wlan,
-    simple_systemd,
-)
-from ..dbus_helper import DbusInterface
-
-log = getLogger(__name__)
-
-FeatureFunction = namedtuple("FeatureFunction", ["function", "args"])
-
-
-class InterfaceRevpiConfig(DbusInterface):
-    """
-    <node>
-        <interface name="com.revolutionpi.middleware1.RevpiConfig">
-            <method name="Disable">
-                <arg name="feature" type="s" direction="in"/>
-            </method>
-            <method name="Enable">
-                <arg name="feature" type="s" direction="in"/>
-            </method>
-            <method name="GetStatus">
-                <arg name="feature" type="s" direction="in"/>
-                <arg name="status" type="b" direction="out"/>
-            </method>
-            <method name="GetAvailability">
-                <arg name="feature" type="s" direction="in"/>
-                <arg name="available" type="b" direction="out"/>
-            </method>
-            <property name="available_features" type="as" access="read"/>
-        </interface>
-    </node>
-    """
-
-    def Disable(self, feature: str) -> None:
-        """Disable the feature."""
-        feature_function = get_feature(feature)
-        feature_function.function(ConfigActions.DISABLE, *feature_function.args)
-
-    def Enable(self, feature: str) -> None:
-        """Enable the feature."""
-        feature_function = get_feature(feature)
-        feature_function.function(ConfigActions.ENABLE, *feature_function.args)
-
-    def GetStatus(self, feature: str) -> bool:
-        """Get feature status."""
-        feature_function = get_feature(feature)
-        return feature_function.function(ConfigActions.STATUS, *feature_function.args)
-
-    def GetAvailability(self, feature: str) -> bool:
-        """Get feature availability on the RevPi."""
-        feature_function = get_feature(feature)
-        return feature_function.function(ConfigActions.AVAILABLE, *feature_function.args)
-
-    @property
-    def available_features(self) -> list[str]:
-        return list(AVAILABLE_FEATURES.keys())
-
-
-def get_feature(feature: str) -> FeatureFunction:
-    if feature not in AVAILABLE_FEATURES:
-        raise ValueError(f"feature {feature} does not exist")
-    feature_function = AVAILABLE_FEATURES[feature]
-    if not feature_function:
-        raise NotImplementedError(f"feature {feature} is not implemented")
-    return feature_function
-
-
-AVAILABLE_FEATURES = {
-    "gui": FeatureFunction(configure_gui, []),
-    "revpi-con-can": FeatureFunction(configure_con_can, []),
-    "dphys-swapfile": FeatureFunction(configure_dphys_swapfile, []),
-    "pimodbus-master": FeatureFunction(simple_systemd, ["pimodbus-master.service"]),
-    "pimodbus-slave": FeatureFunction(simple_systemd, ["pimodbus-slave.service"]),
-    "systemd-timesyncd": FeatureFunction(simple_systemd, ["systemd-timesyncd.service"]),
-    "ssh": FeatureFunction(simple_systemd, ["ssh.service"]),
-    "nodered": FeatureFunction(simple_systemd, ["nodered.service"]),
-    "noderedrevpinodes-server": FeatureFunction(
-        simple_systemd, ["noderedrevpinodes-server.service"]
-    ),
-    "revpipyload": FeatureFunction(simple_systemd, ["revpipyload.service"]),
-    "bluetooth": FeatureFunction(configure_bluetooth, []),
-    "wlan": FeatureFunction(configure_wlan, []),
-    "avahi": FeatureFunction(configure_avahi_daemon, []),
-    "external-antenna": FeatureFunction(configure_external_antenna, []),
-}

src/revpi_middleware/dbus_middleware1/system_config/revpi_config.py

@@ -1,741 +0,0 @@
-# -*- coding: utf-8 -*-
-# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
-# SPDX-License-Identifier: GPL-2.0-or-later
-import re
-import shutil
-import subprocess
-from collections import namedtuple
-from enum import Enum, IntEnum
-from glob import glob
-from logging import getLogger
-from os import X_OK, access
-from os.path import exists, join
-from threading import Thread
-from typing import List, Optional
-
-from pydbus import SystemBus
-
-from ..dbus_helper import grep
-
-log = getLogger(__name__)
-
-ConfigVariable = namedtuple("ConfigVariable", ["name", "value", "line_index"])
-
-LINUX_BT_CLASS_PATH = "/sys/class/bluetooth"
-LINUX_WLAN_CLASS_PATH = "/sys/class/ieee80211"
-CONFIG_TXT_LOCATIONS = ("/boot/firmware/config.txt", "/boot/config.txt")
-
-
-class ComputeModuleTypes(IntEnum):
-    """
-    Enumeration class to represent compute module types.
-
-    This class is an enumeration that defines various types of compute
-    modules and assigns them associated integer values for identifying
-    different module types.
-
-    Attributes:
-        UNKNOWN (int): Represents an unknown or undefined compute module type.
-        CM1 (int): Represents a Compute Module 1.
-        CM3 (int): Represents a Compute Module 3.
-        CM4 (int): Represents a Compute Module 4.
-        CM4S (int): Represents a Compute Module 4S.
-        CM5 (int): Represents a Compute Module 5.
-    """
-    UNKNOWN = 0
-    CM1 = 6
-    CM3 = 10
-    CM4 = 20
-    CM4S = 21
-    CM5 = 24
-
-
-class ConfigActions(Enum):
-    """
-    Enumeration class for defining configuration actions.
-
-    This enumeration provides predefined constants for common configuration
-    actions. It can be used to ensure consistency when working with or defining
-    such actions in a system.
-    """
-    ENABLE = "enable"
-    DISABLE = "disable"
-    STATUS = "status"
-    AVAILABLE = "available"
-
-
-class RevPiConfig:
-    """
-    Represents the configuration and hardware details of a Revolution Pi system.
-
-    This class provides methods and properties to initialize and fetch
-    information related to the Revolution Pi device, such as model, serial
-    number, compute module type, WLAN capability, and the presence of a
-    connection bridge. The class works by parsing system-level files (e.g.,
-    `/proc/cpuinfo`) and using this data to identify hardware characteristics
-    and features.
-
-    Attributes:
-        serial (str): The serial number of the Revolution Pi device.
-        model (str): The model name of the Revolution Pi device.
-    """
-
-    def __init__(self):
-        self._cm_type = ComputeModuleTypes.UNKNOWN
-
-        self._revpi_with_con_bridge = False
-        self._wlan_class_path = ""
-
-        self.serial = ""
-        self.model = ""
-
-        self._init_device_info()
-
-    def _init_device_info(self):
-        """
-        Initialize and retrieve detailed hardware information, including CPU details,
-        device type, WLAN interface, and connectivity features.
-
-        This method gathers information from system files and other sources to
-        initialize device-specific attributes such as model, serial number,
-        compute module type, and optional features like integrated WLAN
-        or ConBridge support. It performs checks specific to the detected
-        module type to accurately populate necessary device details.
-
-        Attributes
-        ----------
-        model : str
-            The model of the CPU based on information from /proc/cpuinfo.
-        serial : str
-            The serial number extracted from /proc/cpuinfo.
-        _cm_type : ComputeModuleTypes, optional
-            The type of the compute module derived from the hardware revision value.
-        _wlan_class_path : str, optional
-            Filesystem path to the detected WLAN interface, if any.
-        _revpi_with_con_bridge : bool
-            Indicates whether the device supports the ConBridge feature.
-        """
-        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 WLAN on CM module
-        could_have_wlan = self._cm_type in (ComputeModuleTypes.CM4, ComputeModuleTypes.CM5)
-        if could_have_wlan:
-            wlan_interface = join(LINUX_WLAN_CLASS_PATH, "phy0")
-            if grep("DRIVER=brcmfmac", join(wlan_interface, "device", "uevent")):
-                self._wlan_class_path = wlan_interface
-
-        # If no build in WLAN on the CM, detect third party WLAN on RevPi Flat
-        if not self._wlan_class_path and grep("revpi-flat", "/proc/device-tree/compatible"):
-            lst_wlan_interfaces = glob("/sys/class/ieee80211/*")
-            for wlan_interface in lst_wlan_interfaces:
-                if grep("DRIVER=mwifiex_sdio", join(wlan_interface, "device", "uevent")):
-                    self._wlan_class_path = wlan_interface
-
-        # 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 class_path_wlan(self) -> str:
-        """
-        Provides access to the WLAN class path.
-
-        This property retrieves the stored WLAN class path, allowing the user to access it when
-        needed.
-
-        Returns:
-            str: The WLAN class path.
-        """
-        return self._wlan_class_path
-
-    @property
-    def cm_type(self) -> ComputeModuleTypes:
-        """
-        Gets the type of the compute module.
-
-        The property provides access to the type of the compute
-        module used. The type is represented as an instance of
-        the `ComputeModuleTypes` class.
-
-        Returns
-        -------
-        ComputeModuleTypes
-            The type of the compute module.
-        """
-        return self._cm_type
-
-    @property
-    def with_con_bridge(self) -> bool:
-        """
-        Indicates if the device is configured with a connection bridge.
-
-        This property checks the internal status and determines whether the device setup
-        includes a connection bridge functionality. It is read-only.
-
-        Returns:
-            bool: True if the connection bridge is configured, False otherwise.
-        """
-        return self._revpi_with_con_bridge
-
-    @property
-    def with_wlan(self) -> bool:
-        """
-        Checks if WLAN is available.
-
-        This property evaluates whether WLAN is enabled or available by checking
-        the presence or value of the internal attribute `_wlan_class_path`.
-
-        Returns:
-            bool: True if WLAN is available, False otherwise.
-        """
-        return bool(self._wlan_class_path)
-
-
-class ConfigTxt:
-    """
-    Configuration file handler for managing 'config.txt'.
-
-    This class provides an interface to read, modify, save, and reload
-    Raspbian's configuration file `config.txt`. It includes functionalities
-    to manipulate specific parameters within the configuration and supports
-    managing dtoverlay and dtparam entries. The primary aim of this class
-    is to abstract file operations and make modifications user-friendly.
-
-    Attributes:
-        _config_txt_path (str): The path to the configuration file `config.txt`.
-        _config_txt_lines (list[str]): Contains all lines of the configuration
-            file as a list of strings, where each string represents a line.
-    """
-    re_name_value = re.compile(r"^\s*(?!#)(?P<name>[^=\s].+?)\s*=\s*(?P<value>\S+)\s*$")
-
-    def __init__(self):
-        self._config_txt_path = ""
-        for path in CONFIG_TXT_LOCATIONS:
-            if exists(path):
-                self._config_txt_path = path
-                break
-
-        if not self._config_txt_path:
-            raise FileNotFoundError("no config.txt found")
-
-        self._config_txt_lines = []
-
-    def _clear_name_values(self, name: str, values: str or list) -> int:
-        """
-        Removes all occurrences of specified name-value pairs from the configuration.
-
-        This method searches for all name-value pairs in the configuration and
-        removes those that match the given name and value(s). It returns the
-        number of occurrences removed.
-
-        Arguments:
-            name: str
-                The name of the configuration variable to search for.
-            values: str or list
-                The value or list of values to match the configuration variable
-                against.
-
-        Returns:
-            int: The number of name-value pairs removed from the configuration.
-
-        """
-        counter = 0
-        if type(values) is str:
-            values = [values]
-
-        for config_var in self._get_all_name_values():
-            if config_var.name == name and config_var.value in values:
-                self._config_txt_lines.pop(config_var.line_index)
-                counter += 1
-
-        return counter
-
-    def _get_all_name_values(self) -> List[ConfigVariable]:
-        """
-        Retrieves all name-value pairs from the configuration text lines.
-
-        This method parses the configuration text lines to extract all name-value
-        pairs. If the configuration text lines are not loaded, it reloads the
-        configuration before processing. Each extracted name-value pair is added to a
-        list as a ConfigVariable object, which also holds the index of the match in
-        the text lines. The method returns the compiled list of these ConfigVariable
-        objects.
-
-        Returns:
-            List[ConfigVariable]: A list of ConfigVariable objects representing the
-            name-value pairs found in the configuration text lines, along with their
-            corresponding indexes.
-        """
-        if not self._config_txt_lines:
-            self.reload_config()
-
-        lst_return = []
-
-        for i in range(len(self._config_txt_lines)):
-            match = self.re_name_value.match(self._config_txt_lines[i])
-            if match:
-                lst_return.append(ConfigVariable(match.group("name"), match.group("value"), i))
-
-        return lst_return
-
-    def reload_config(self):
-        """
-        Reloads the configuration file and updates the list of configuration lines.
-
-        This method reads the content of the configuration file specified by the
-        attribute `_config_txt_path` and updates `_config_txt_lines` with the file
-        contents as a list of strings, where each string represents a line.
-
-        Returns:
-            None
-        """
-        with open(self._config_txt_path, "r") as f:
-            self._config_txt_lines = f.readlines()
-
-    def save_config(self):
-        """
-        Saves the current configuration to a file. The method ensures atomicity by first writing
-        to a temporary file and then moving it to the desired path. After the configuration is
-        saved, the internal list of configuration lines is cleared.
-
-        Raises:
-            OSError: If there is an issue writing to or moving the file.
-        """
-        if not self._config_txt_lines:
-            return
-
-        tmp_path = f"{self._config_txt_path}.tmp"
-        with open(tmp_path, "w") as f:
-            f.writelines(self._config_txt_lines)
-        shutil.move(tmp_path, self._config_txt_path)
-
-        self._config_txt_lines.clear()
-
-    def add_name_value(self, name: str, value: str):
-        """
-        Adds a name-value pair to the configuration if it does not already exist.
-
-        This method checks if the given name-value pair is already present in
-        the configuration. If it is not present, the pair is appended to the
-        configuration text lines.
-
-        Parameters:
-            name (str): The name to be added to the configuration.
-            value (str): The value corresponding to the name to be added.
-
-        Returns:
-            None
-        """
-        # Check weather name and value already exists
-        for config_var in self._get_all_name_values():
-            if config_var.name == name and config_var.value == value:
-                return
-
-        self._config_txt_lines.append(f"{name}={value}\n")
-
-    def clear_dtoverlays(self, dtoverlays: str or list) -> int:
-        """
-        Clears the specified device tree overlays. This method removes one or more
-        device tree overlays by clearing their corresponding name-value pairs.
-
-        Args:
-            dtoverlays (str or list): A device tree overlay name as a string, or a
-            list of such overlay names to be cleared.
-
-        Returns:
-            int: The number of device tree overlay name-value pairs successfully
-            cleared.
-        """
-        return self._clear_name_values("dtoverlay", dtoverlays)
-
-    def clear_dtparams(self, dtparams: str or list) -> int:
-        """
-        Clears the specified device tree parameters.
-
-        This method removes the given device tree parameters by utilizing
-        the underlying `_clear_name_values` function with a predefined
-        parameter type.
-
-        Parameters:
-        dtparams: str or list
-            A string or list of strings specifying the device tree
-            parameters to remove.
-
-        Returns:
-        int
-            The number of parameters cleared.
-        """
-        return self._clear_name_values("dtparam", dtparams)
-
-    def get_values(self, var_name: str) -> list:
-        """
-        Get all values associated with a given variable name.
-
-        This method retrieves a list of values corresponding to the specified
-        variable name by iterating through a collection of configuration
-        variables. Each configuration variable is checked for a matching name,
-        and its value is appended to the resulting list if a match is found.
-
-        Parameters:
-        var_name (str): The name of the variable for which values are to
-                        be retrieved.
-
-        Returns:
-        list: A list of values associated with the specified variable name.
-        """
-        var_values = []
-
-        for config_var in self._get_all_name_values():
-            if config_var.name == var_name:
-                var_values.append(config_var.value)
-
-        return var_values
-
-    @property
-    def config_txt_path(self) -> str:
-        """
-        Get the file path for the configuration text file.
-
-        This property provides access to the private attribute `_config_txt_path` which
-        stores the file path to the configuration text file.
-
-        Returns:
-        str
-            The file path to the configuration text file.
-        """
-        return self._config_txt_path
-
-
-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_bluetooth(action: ConfigActions):
-    hci_device = join(LINUX_BT_CLASS_PATH, "hci0")
-    bt_rfkill_index = get_rfkill_index(hci_device)
-
-    # If the bluetooth device is not present, the device should have been
-    # brought up by revpi-bluetooth's udev rules or vendor magic (devices
-    # based on CM4 and newer). Nothing we can do here, so treat the interface
-    # as disabled.
-
-    if action is ConfigActions.ENABLE:
-        if bt_rfkill_index is not None:
-            with open(f"/sys/class/rfkill/rfkill{bt_rfkill_index}/soft", "w") as f:
-                f.write("0")
-
-    elif action is ConfigActions.DISABLE:
-        if bt_rfkill_index is not None:
-            with open(f"/sys/class/rfkill/rfkill{bt_rfkill_index}/soft", "w") as f:
-                f.write("1")
-
-    elif action is ConfigActions.STATUS:
-        if bt_rfkill_index is None:
-            return False
-
-        with open(f"/sys/class/rfkill/rfkill{bt_rfkill_index}/soft", "r") as f:
-            buffer = f.read().strip()
-            return buffer == "0"
-
-    elif action is ConfigActions.AVAILABLE:
-        return bt_rfkill_index is not None
-
-    else:
-        raise ValueError(f"action {action} not supported")
-
-    return None
-
-
-def configure_con_can(action: ConfigActions):
-    revpi = RevPiConfig()
-    if action is ConfigActions.AVAILABLE:
-        return revpi.with_con_bridge
-
-    dt_overlay = "revpi-con-can"
-    config_txt = ConfigTxt()
-
-    if action is ConfigActions.ENABLE and revpi.with_con_bridge:
-        config_txt.clear_dtoverlays([dt_overlay])
-        config_txt.add_name_value("dtoverlay", dt_overlay)
-        config_txt.save_config()
-        subprocess.call(["/usr/bin/dtoverlay", dt_overlay])
-
-    elif action is ConfigActions.DISABLE and revpi.with_con_bridge:
-        config_txt.clear_dtoverlays([dt_overlay])
-        config_txt.save_config()
-        subprocess.call(["/usr/bin/dtoverlay", "-r", dt_overlay])
-
-    elif action is ConfigActions.STATUS:
-        return revpi.with_con_bridge and dt_overlay in config_txt.get_values("dtparam")
-
-    else:
-        raise ValueError(f"action {action} not supported")
-
-    return None
-
-
-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_external_antenna(action: ConfigActions):
-    revpi = RevPiConfig()
-    if action is ConfigActions.AVAILABLE:
-        return revpi.with_wlan
-
-    config_txt = ConfigTxt()
-
-    if action is ConfigActions.ENABLE and revpi.with_wlan:
-        config_txt.clear_dtparams(["ant1", "ant2"])
-        config_txt.add_name_value("dtparam", "ant2")
-        config_txt.save_config()
-
-    elif action is ConfigActions.DISABLE and revpi.with_wlan:
-        config_txt.clear_dtparams(["ant1", "ant2"])
-        config_txt.save_config()
-
-    elif action is ConfigActions.STATUS:
-        return revpi.with_wlan and "ant2" in config_txt.get_values("dtparam")
-
-    else:
-        raise ValueError(f"action {action} not supported")
-
-    return None
-
-
-def configure_gui(action: ConfigActions):
-    gui_available = access("/usr/bin/startx", X_OK)
-
-    if action is ConfigActions.AVAILABLE:
-        return gui_available
-
-    bus = SystemBus()
-    systemd = bus.get(
-        "org.freedesktop.systemd1",
-        "/org/freedesktop/systemd1",
-    )
-    systemd_manager = systemd["org.freedesktop.systemd1.Manager"]
-
-    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 configure_wlan(action: ConfigActions):
-    revpi = RevPiConfig()
-
-    if action is ConfigActions.ENABLE:
-        if revpi.with_wlan:
-            wlan_rfkill_index = get_rfkill_index(revpi.class_path_wlan)
-            with open(f"/sys/class/rfkill/rfkill{wlan_rfkill_index}/soft", "w") as f:
-                f.write("0")
-
-    elif action is ConfigActions.DISABLE:
-        if revpi.with_wlan:
-            wlan_rfkill_index = get_rfkill_index(revpi.class_path_wlan)
-            with open(f"/sys/class/rfkill/rfkill{wlan_rfkill_index}/soft", "w") as f:
-                f.write("1")
-
-    elif action is ConfigActions.AVAILABLE:
-        return revpi.with_wlan
-
-    elif action is ConfigActions.STATUS:
-        if not revpi.with_wlan:
-            return False
-
-        wlan_rfkill_index = get_rfkill_index(revpi.class_path_wlan)
-        with open(f"/sys/class/rfkill/rfkill{wlan_rfkill_index}/soft", "r") as f:
-            buffer = f.read().strip()
-            return buffer == "0"
-
-    else:
-        raise ValueError(f"action {action} not supported")
-
-    return None
-
-
-def get_rfkill_index(device_class_path: str) -> Optional[int]:
-    """
-    Get the rfkill index for a device under a specific device class path.
-
-    This function searches for and extracts the rfkill index associated with
-    devices located under the given device class path. It uses a regular
-    expression to identify and parse the rfkill index from the paths
-    of matching rfkill device files.
-
-    Parameters:
-        device_class_path: str
-            The path to the device class directory where rfkill entries
-            are located.
-
-    Returns:
-        Optional[int]:
-            The index of the rfkill device if found, otherwise None.
-    """
-    re_rfkill_index = re.compile(r"^/.+/rfkill(?P<index>\d+)$")
-    for rfkill_path in glob(join(device_class_path, "rfkill*")):
-        match_index = re_rfkill_index.match(rfkill_path)
-        if match_index:
-            return int(match_index.group("index"))
-
-    return None
-
-
-def simple_systemd(action: ConfigActions, unit: str):
-    """
-    Performs specified actions on systemd units.
-
-    This function allows interaction with systemd units for various operations
-    such as enabling, disabling, checking the status, and verifying availability.
-    It communicates with the systemd manager via the SystemBus and handles units
-    based on the action specified.
-
-    Parameters:
-        action (ConfigActions): Specifies the action to be performed on the
-                                systemd unit. Supported actions include ENABLE,
-                                DISABLE, STATUS, and AVAILABLE.
-        unit (str): The name of the systemd unit on which the action is to be
-                    performed.
-
-    Returns:
-        bool: For STATUS and AVAILABLE actions, returns True if the corresponding
-              criteria are met (e.g., enabled and active for STATUS, or not found
-              for AVAILABLE). Otherwise, returns False.
-
-    Raises:
-        ValueError: If the specified action is not supported.
-    """
-    bus = SystemBus()
-    systemd = bus.get(
-        "org.freedesktop.systemd1",
-        "/org/freedesktop/systemd1",
-    )
-    systemd_manager = systemd["org.freedesktop.systemd1.Manager"]
-
-    if action is ConfigActions.ENABLE:
-
-        def thread_unit_config():
-            """Change configuration asynchronously."""
-            # Dbus call: UnmaskUnitFiles(in as files, in b runtime, out a(sss) changes
-            lst_change_unmask = systemd_manager.UnmaskUnitFiles([unit], False)
-
-            # Dbus call: EnableUnitFiles(in as files, in b runtime, in b force,
-            #                            out b carries_install_info, out a(sss) changes
-            lst_change_enable = systemd_manager.EnableUnitFiles([unit], False, False)
-            if lst_change_unmask or lst_change_enable:
-                # Reload systemd after modified unit property
-                systemd_manager.Reload()
-
-        Thread(target=thread_unit_config, daemon=True).start()
-
-        # Dbus call: StartUnit(in s name, in s mode, out o job
-        systemd_manager.StartUnit(unit, "replace")
-
-    elif action is ConfigActions.DISABLE:
-
-        def thread_unit_config():
-            """Change configuration asynchronously."""
-            # Dbus call: DisableUnitFiles (in as files, in b runtime, out a(sss) changes)
-            change = systemd_manager.DisableUnitFiles([unit], False)
-            if change:
-                # Reload systemd after modified unit property
-                systemd_manager.Reload()
-
-        Thread(target=thread_unit_config, daemon=True).start()
-
-        # Dbus call: StopUnit(in s name,in s mode, out o job
-        systemd_manager.StopUnit(unit, "replace")
-
-
-    elif action is ConfigActions.STATUS:
-        try:
-            unit_path = systemd_manager.LoadUnit(unit)
-            properties = bus.get("org.freedesktop.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("org.freedesktop.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")
-
-    return None
-
-
-if __name__ == "__main__":
-    rc = RevPiConfig()
-    print("Model:", rc.model)
-    print("Serial:         ", rc.serial)
-    print("CM Type:        ", rc.cm_type.name)
-    print("With WLAN:      ", rc.with_wlan)
-    if rc.with_wlan:
-        print("    class path: ", rc.class_path_wlan)
-        print("  rfkill index: ", get_rfkill_index(rc.class_path_wlan))
-    print("With con-bridge:", rc.with_con_bridge)
-
-    config_txt = ConfigTxt()
-    print("Config file:    ", config_txt.config_txt_path)

src/revpi_middleware/ios1/__init__.py

@@ -1,5 +0,0 @@
-# -*- coding: utf-8 -*-
-# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
-# SPDX-License-Identifier: GPL-2.0-or-later
-"""D-Bus ios version 1 of revpi_middleware."""
-from .ios1_helper import REVPI_DBUS_BASE_PATH, REVPI_DBUS_NAME

src/revpi_middleware/ios1/bus_provider_io.py

@@ -1,100 +0,0 @@
-# -*- coding: utf-8 -*-
-# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
-# SPDX-License-Identifier: GPL-2.0-or-later
-"""D-Bus bus provider for revpi_middleware."""
-from logging import getLogger
-from threading import Thread
-
-import revpimodio2
-from gi.repository import GLib
-from pydbus import SessionBus, SystemBus
-from revpimodio2 import Cycletools
-
-from . import REVPI_DBUS_NAME
-from .interface_ios import (
-    InterfaceIoManager,
-    InterfaceInpBool,
-    InterfaceOutBool,
-    InterfaceInpInt,
-    InterfaceOutInt,
-)
-
-log = getLogger(__name__)
-
-
-class BusProviderIo(Thread):
-
-    def __init__(
-        self,
-        picontrol_device="/dev/piControl0",
-        config_rsc="/etc/revpi/config.rsc",
-        use_system_bus=True,
-    ):
-        log.debug("enter BusProviderIo.__init__")
-        super().__init__()
-
-        self._bus = SystemBus() if use_system_bus else SessionBus()
-        self._dc_io_interfaces = {}
-        self._loop = GLib.MainLoop()
-        self._modio = revpimodio2.RevPiModIO(
-            procimg=picontrol_device,
-            configrsc=config_rsc,
-            shared_procimg=True,
-        )
-
-        self.picontrol_device = picontrol_device
-        self.config_rsc = config_rsc
-
-    def run(self):
-        log.debug("enter BusProviderIo.run")
-
-        self._dc_io_interfaces.clear()
-
-        for io in self._modio.io:
-            interface = None
-            value_type = type(io.value)
-            if value_type is bool:
-                interface = (
-                    InterfaceInpBool(self._bus, io)
-                    if io.type == revpimodio2.INP
-                    else InterfaceOutBool(self._bus, io)
-                )
-            elif value_type is int:
-                interface = (
-                    InterfaceInpInt(self._bus, io)
-                    if io.type == revpimodio2.INP
-                    else InterfaceOutInt(self._bus, io)
-                )
-
-            if interface is not None:
-                self._dc_io_interfaces[io.name] = interface
-
-        lst_interfaces = [
-            (f"io/{io_name}", self._dc_io_interfaces[io_name]) for io_name in self._dc_io_interfaces
-        ]
-        try:
-            self._bus.publish(
-                REVPI_DBUS_NAME,
-                InterfaceIoManager(self._modio, self._dc_io_interfaces),
-                *lst_interfaces,
-            )
-        except Exception as e:
-            log.error(f"can not publish dbus {REVPI_DBUS_NAME}: {e}")
-
-        try:
-            self._loop.run()
-        except Exception as e:
-            log.error(f"can not run dbus mainloop: {e}")
-
-        self._modio.cleanup()
-
-        log.debug("leave BusProviderIo.run")
-
-    def stop(self):
-        log.debug("enter BusProviderIo.stop")
-        self._loop.quit()
-        log.debug("leave BusProviderIo.stop")
-
-    @property
-    def running(self):
-        return self._loop.is_running()

src/revpi_middleware/ios1/interface_ios.py

@@ -1,159 +0,0 @@
-# -*- coding: utf-8 -*-
-# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
-# SPDX-License-Identifier: GPL-2.0-or-later
-"""D-Bus interfaces for IOs."""
-from gi.overrides.GLib import Variant
-from pydbus.generic import signal
-from revpimodio2 import RevPiModIO, Cycletools
-
-from .ios1_helper import REVPI_DBUS_BASE_PATH, DbusInterfaceIo
-
-
-class InterfaceIoManager:
-    """
-    <node>
-        <interface name="com.revolutionpi.ios1.IoManager">
-            <method name="GetAllInputs">
-                <arg type="as" direction="out"/>
-            </method>
-            <method name="GetAllOutputs">
-                <arg type="as" direction="out"/>
-            </method>
-            <method name="Get">
-                <arg name="io_name" type="s" direction="in"/>
-                <arg name="object-path" type="o" direction="out"/>
-            </method>
-            <method name="ActivateIoEvents">
-            </method>
-            <method name="DeactivateIoEvents">
-            </method>
-            <signal name="IoChanged">
-                <arg name="name" type="s" direction="out"/>
-                <arg name="value" type="v" direction="out"/>
-            </signal>
-        </interface>
-    </node>
-    """
-
-    interface_name = "com.revolutionpi.ios1.IoManager"
-    IoChanged = signal()
-
-    def __init__(self, modio: RevPiModIO, io_interfaces: dict[str, DbusInterfaceIo]):
-        self._dc_io_interfaces = io_interfaces
-        self.modio = modio
-
-        self.lst_inp = []
-        for dev in self.modio.device:
-            for io in dev.get_inputs():
-                self.lst_inp.append(self._get_io_path(io.name))
-        self.lst_out = []
-        for dev in self.modio.device:
-            for io in dev.get_outputs():
-                self.lst_out.append(self._get_io_path(io.name))
-
-    def _modio_cycle(self, ct: Cycletools) -> None:
-        for io_name in self._dc_io_interfaces:
-            interface = self._dc_io_interfaces[io_name]
-            if ct.changed(interface.io):
-                interface.emit_io_change()
-                self.IoChanged(interface.io.name, Variant("b", interface.io.value))
-
-    def _get_io_path(self, io_name: str) -> str:
-        return f"{REVPI_DBUS_BASE_PATH}/io/{io_name}"
-
-    def GetAllInputs(self) -> list[str]:
-        return self.lst_inp
-
-    def GetAllOutputs(self) -> list[str]:
-        return self.lst_out
-
-    def Get(self, io_name) -> str:
-        if io_name in self.modio.io:
-            return self._get_io_path(io_name)
-
-        raise KeyError(f"No IO with name '{io_name}' found.")
-
-    def ActivateIoEvents(self) -> None:
-        if not self.modio._looprunning:
-            self.modio.autorefresh_all()
-            self.modio.cycleloop(self._modio_cycle, cycletime=50, blocking=False)
-
-    def DeactivateIoEvents(self) -> None:
-        self.modio.exit(False)
-
-
-class InterfaceInpBool(DbusInterfaceIo):
-    """
-    <node>
-        <interface name="com.revolutionpi.ios1.InpBool">
-            <property name="bmk" type="s" access="read"/>
-            <property name="name" type="s" access="read"/>
-            <property name="value" type="b" access="read"/>
-        </interface>
-    </node>
-    """
-
-    interface_name = "com.revolutionpi.ios1.InpBool"
-
-    @property
-    def bmk(self) -> str:
-        return self.io.bmk
-
-    @property
-    def name(self) -> str:
-        return self.io.name
-
-    @property
-    def value(self) -> bool:
-        return self.io.value
-
-
-class InterfaceOutBool(InterfaceInpBool):
-    """
-    <node>
-        <interface name="com.revolutionpi.ios1.OutBool">
-            <property name="bmk" type="s" access="read"/>
-            <property name="name" type="s" access="read"/>
-            <property name="value" type="b" access="readwrite"/>
-        </interface>
-    </node>
-    """
-
-    interface_name = "com.revolutionpi.ios1.OutBool"
-
-    @property
-    def value(self) -> bool:
-        return super().value
-
-    @value.setter
-    def value(self, value: bool):
-        self.io.value = value
-        self.io._parentdevice._modio.writeprocimg()
-
-
-class InterfaceInpInt(DbusInterfaceIo):
-    """
-    <node>
-        <interface name="com.revolutionpi.ios1.InpInt">
-            <property name="bmk" type="s" access="read"/>
-            <property name="name" type="s" access="read"/>
-            <property name="value" type="i" access="read"/>
-        </interface>
-    </node>
-    """
-
-    interface_name = "com.revolutionpi.ios1.InpInt"
-
-
-class InterfaceOutInt(InterfaceInpInt):
-    """
-    <node>
-        <interface name="com.revolutionpi.ios1.OutInt">
-            <property name="bmk" type="s" access="read"/>
-            <property name="name" type="s" access="read"/>
-            <property name="value" type="i" access="readwrite"/>
-        </interface>
-    </node>
-    """
-
-    interface_name = "com.revolutionpi.ios1.OutInt"

src/revpi_middleware/ios1/ios1_helper.py

@@ -1,37 +0,0 @@
-# -*- coding: utf-8 -*-
-# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
-# SPDX-License-Identifier: GPL-2.0-or-later
-"""Helper for io read and write."""
-from logging import getLogger
-
-from gi.overrides.GLib import Variant
-from pydbus import SessionBus, SystemBus
-from pydbus.generic import signal
-from revpimodio2.io import IOBase
-
-log = getLogger(__name__)
-
-REVPI_DBUS_NAME = "com.revolutionpi.ios1"
-REVPI_DBUS_BASE_PATH = "/com/revolutionpi/ios1"
-
-
-class DbusInterfaceIo:
-    interface_name = ""
-    PropertiesChanged = signal()
-
-    def __init__(self, dbus: SystemBus or SessionBus, io: IOBase):
-        self.dbus = dbus
-        self.io = io
-        self.variant_type = "b" if type(self.io.value) is bool else "i"
-
-    def emit_io_change(self):
-        if not self.interface_name:
-            return
-
-        if self.interface_name:
-            print(type(self.io.value))
-            self.PropertiesChanged(
-                self.interface_name,
-                {"value": int(self.io.value)},
-                [],
-            )

tests/dbus_middleware1/process_image/test_interface_picontrol.py

@@ -18,7 +18,7 @@ class TestObjectPicontrol(TestBusProvider):
     def test_reset_driver(self):
         simple_call(
             "ResetDriver",
-            interface=extend_interface("PiControl"),
+            interface=extend_interface("picontrol"),
             bus_type=BusType.SESSION,
         )
         ioctl_call = IOCTL_QUEUE.get(timeout=2.0)
@@ -37,7 +37,7 @@ class TestObjectPicontrol(TestBusProvider):
         result = await_signal(
             "NotifyDriverReset",
             timeout,
-            extend_interface("PiControl"),
+            extend_interface("picontrol"),
             bus_type=BusType.SESSION,
         )
         self.assertTrue(result)