feat(io): Add dbus bus com.revolutionpi.ios1

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

src/revpi_middleware/ios1/interface_ios.py

@@ -0,0 +1,159 @@
+# -*- 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"