# -*- 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: """ """ 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): """ """ 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): """ """ 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): """ """ interface_name = "com.revolutionpi.ios1.InpInt" class InterfaceOutInt(InterfaceInpInt): """ """ interface_name = "com.revolutionpi.ios1.OutInt"