refactor: Add helper functions for D-Bus connections and bus handling

Replaced direct D-Bus initialization with
`get_new_system_dbus_connection` and `get_new_session_dbus_connection`
helper functions. This functions will create a new connection to use
separate busses.

Signed-off-by: Sven Sager <s.sager@kunbus.com>

src/revpi_middleware/daemon.py

@@ -6,9 +6,13 @@
 from logging import getLogger
 from threading import Event
 from time import perf_counter
+from typing import Union
+
+from dbus import SystemBus, SessionBus
 
 from . import proginit as pi
 from .dbus_middleware1.bus_provider import BusProvider
+from .helper import get_new_session_dbus_connection, get_new_system_dbus_connection
 from .ios1.bus_provider_io import BusProviderIo
 
 log = getLogger(__name__)
@@ -39,18 +43,24 @@ class MiddlewareDaemon:
 
         log.debug("leave MiddlewareDaemon._configure")
 
+    def _get_bus(self) -> Union[SystemBus, SessionBus]:
+        return (
+            get_new_session_dbus_connection()
+            if pi.pargs.use_session_bus
+            else get_new_system_dbus_connection()
+        )
+
     def dbus_start(self):
         log.debug("enter MiddlewareDaemon.dbus_start")
 
         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 = BusProvider(self._get_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 = BusProviderIo(self._get_bus())
             self.io_bus_provider.start()
 
         log.debug("leave MiddlewareDaemon.dbus_start")
@@ -63,12 +73,14 @@ class MiddlewareDaemon:
             self.bus_provider.join(timeout=10.0)
             if self.bus_provider.is_alive():
                 log.warning("dbus provider thread is still alive")
+            self.bus_provider.bus.con.close()
 
         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")
+            self.io_bus_provider.bus.con.close()
 
         log.debug("leave MiddlewareDaemon.dbus_stop")
 

src/revpi_middleware/dbus_middleware1/bus_provider.py

@@ -5,6 +5,7 @@
 
 from logging import getLogger
 from threading import Thread
+from typing import Union
 
 from gi.repository import GLib
 from pydbus import SessionBus, SystemBus
@@ -20,14 +21,14 @@ class BusProvider(Thread):
 
     def __init__(
         self,
+        dbus: Union[SystemBus, SessionBus],
         picontrol_device="/dev/piControl0",
         config_rsc="/etc/revpi/config.rsc",
-        use_system_bus=True,
     ):
         log.debug("enter BusProvider.__init__")
         super().__init__()
 
-        self._bus = SystemBus() if use_system_bus else SessionBus()
+        self._bus = dbus
         self._loop = GLib.MainLoop()
 
         self.picontrol_device = picontrol_device
@@ -73,6 +74,10 @@ class BusProvider(Thread):
         self._loop.quit()
         log.debug("leave BusProvider.stop")
 
+    @property
+    def bus(self) -> Union[SystemBus, SessionBus]:
+        return self._bus
+
     @property
     def running(self):
         return self._loop.is_running()

src/revpi_middleware/helper.py

@@ -0,0 +1,48 @@
+# -*- coding: utf-8 -*-
+# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
+# SPDX-License-Identifier: GPL-2.0-or-later
+"""Helper functions for the root project."""
+
+from os import environ, getuid
+
+from pydbus import connect
+
+
+def get_new_system_dbus_connection():
+    """
+    Establishes a new connection to the system D-Bus.
+
+    This function retrieves the system D-Bus address from the environment
+    variable `DBUS_SYSTEM_BUS_ADDRESS`, defaulting to
+    `unix:path=/run/dbus/system_bus_socket` if the variable is not set.
+    It then establishes and returns a connection to the D-Bus at that
+    address.
+
+    Raises:
+        An error may be raised if establishing the connection fails.
+
+    Returns:
+        The established system D-Bus connection.
+    """
+    address = environ.get("DBUS_SYSTEM_BUS_ADDRESS", "unix:path=/run/dbus/system_bus_socket")
+    return connect(address)
+
+
+def get_new_session_dbus_connection():
+    """
+    Establishes a new connection to the session D-Bus.
+
+    This function retrieves the session D-Bus address from the environment
+    variable `DBUS_SESSION_BUS_ADDRESS`, defaulting to
+    `unix:path=/run/user/{UID}/bus` if the variable is not set.
+    It then establishes and returns a connection to the D-Bus at that
+    address.
+
+    Raises:
+        An error may be raised if establishing the connection fails.
+
+    Returns:
+        The established session D-Bus connection.
+    """
+    address = environ.get("DBUS_SESSION_BUS_ADDRESS", f"unix:path=/run/user/{getuid()}/bus")
+    return connect(address)

src/revpi_middleware/ios1/bus_provider_io.py

@@ -5,6 +5,7 @@
 
 from logging import getLogger
 from threading import Thread
+from typing import Union
 
 import revpimodio2
 from gi.repository import GLib
@@ -28,14 +29,14 @@ class BusProviderIo(Thread):
 
     def __init__(
         self,
+        dbus: Union[SystemBus, SessionBus],
         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._bus = dbus
         self._loop = GLib.MainLoop()
         self._lst_device_interfaces = []
         self._lst_io_interfaces = []
@@ -106,6 +107,10 @@ class BusProviderIo(Thread):
         self._loop.quit()
         log.debug("leave BusProviderIo.stop")
 
+    @property
+    def bus(self) -> Union[SystemBus, SessionBus]:
+        return self._bus
+
     @property
     def running(self):
         return self._loop.is_running()