Implement virtual device support

src/abbfreeathome/bin/interface.py

@@ -16,4 +16,3 @@ class Interface(enum.Enum):
     WIRELESS_RF = "RF"
     HUE = "hue"
     SONOS = "sonos"
-    VIRTUAL_DEVICE = "vdev:[email protected]"

src/abbfreeathome/const.py

@@ -21,11 +21,18 @@
 from .devices.room_temperature_controller import RoomTemperatureController
 from .devices.smoke_detector import SmokeDetector
 from .devices.switch_actuator import SwitchActuator
+from .devices.switch_actuator_virtual import SwitchActuatorVirtual
 from .devices.switch_sensor import DimmingSensor, SwitchSensor
 from .devices.temperature_sensor import TemperatureSensor
 from .devices.trigger import Trigger
 from .devices.wind_sensor import WindSensor
 from .devices.window_door_sensor import WindowDoorSensor
+from .devices.window_door_sensor_virtual import WindowDoorSensorVirtual
+
+FUNCTION_DEVICE_MAPPING_VIRTUAL: dict[Function, Base] = {
+    Function.FID_WINDOW_DOOR_SENSOR: WindowDoorSensorVirtual,
+    Function.FID_SWITCH_ACTUATOR: SwitchActuatorVirtual,
+}
 
 FUNCTION_DEVICE_MAPPING: dict[Function, Base] = {
     Function.FID_ATTIC_WINDOW_ACTUATOR: AtticWindowActuator,

src/abbfreeathome/devices/base.py

@@ -15,6 +15,7 @@ class Base:
     """Free@Home Base Class."""
 
     _state_refresh_output_pairings: list[Pairing] = []
+    _state_refresh_input_pairings: list[Pairing] = []
 
     def __init__(
         self,
@@ -44,6 +45,8 @@ def __init__(
 
         # Set the initial state of the device based on output
         self._refresh_state_from_outputs()
+        # Set the initial state of the device based on input
+        self._refresh_state_from_inputs()
 
     @property
     def device_id(self) -> str:
@@ -110,6 +113,10 @@ def update_device(self, datapoint_key: str, datapoint_value: str):
             self._outputs[_io_key]["value"] = datapoint_value
             _refreshed = self._refresh_state_from_output(output=self._outputs[_io_key])
 
+        if _io_key in self._inputs:
+            self._inputs[_io_key]["value"] = datapoint_value
+            _refreshed = self._refresh_state_from_input(input=self._inputs[_io_key])
+
         if _refreshed and self._callbacks:
             for callback in self._callbacks:
                 callback()
@@ -142,10 +149,36 @@ async def refresh_state(self):
                 }
             )
 
+        for _pairing in self._state_refresh_input_pairings:
+            _input_id, _input_value = self.get_input_by_pairing(pairing=_pairing)
+
+            _datapoint = (
+                await self._api.get_datapoint(
+                    device_id=self.device_id,
+                    channel_id=self.channel_id,
+                    datapoint=_input_id,
+                )
+            )[0]
+
+            self._refresh_state_from_input(
+                input={
+                    "pairingID": _pairing.value,
+                    "value": _datapoint,
+                }
+            )
+
     def _refresh_state_from_output(self, output: dict[str, Any]) -> bool:
         """Refresh the state of the device from a single output."""
 
     def _refresh_state_from_outputs(self):
         """Refresh the state of the device from the _outputs."""
         for _output in self._outputs.values():
             self._refresh_state_from_output(_output)
+
+    def _refresh_state_from_input(self, input: dict[str, Any]) -> bool:
+        """Refresh the state of the device from a single input."""
+
+    def _refresh_state_from_inputs(self):
+        """Refresh the state of the device from the _inputs."""
+        for _input in self._inputs.values():
+            self._refresh_state_from_input(_input)

src/abbfreeathome/devices/switch_actuator_virtual.py

@@ -0,0 +1,82 @@
+"""Free@Home SwitchActuatorVirtual Class."""
+
+from typing import Any
+
+from ..api import FreeAtHomeApi
+from ..bin.pairing import Pairing
+from .base import Base
+
+
+class SwitchActuatorVirtual(Base):
+    """Free@Home SwitchActuatorVirtual Class."""
+
+    _state_refresh_input_pairings: list[Pairing] = [
+        Pairing.AL_SWITCH_ON_OFF,
+    ]
+
+    def __init__(
+        self,
+        device_id: str,
+        device_name: str,
+        channel_id: str,
+        channel_name: str,
+        inputs: dict[str, dict[str, Any]],
+        outputs: dict[str, dict[str, Any]],
+        parameters: dict[str, dict[str, Any]],
+        api: FreeAtHomeApi,
+        floor_name: str | None = None,
+        room_name: str | None = None,
+    ) -> None:
+        """Initialize the Free@Home SwitchActuatorVirtual class."""
+        self._state: bool | None = None
+
+        super().__init__(
+            device_id,
+            device_name,
+            channel_id,
+            channel_name,
+            inputs,
+            outputs,
+            parameters,
+            api,
+            floor_name,
+            room_name,
+        )
+
+    @property
+    def state(self) -> bool | None:
+        """Get the state of the switch."""
+        return self._state
+
+    async def turn_on(self):
+        """Turn on the switch."""
+        await self._set_switching_datapoint("1")
+        self._state = True
+
+    async def turn_off(self):
+        """Turn on the switch."""
+        await self._set_switching_datapoint("0")
+        self._state = False
+
+    def _refresh_state_from_input(self, input: dict[str, Any]) -> bool:
+        """
+        Refresh the state of the device from a given input.
+
+        This will return whether the state was refreshed as a boolean value.
+        """
+        if input.get("pairingID") == Pairing.AL_SWITCH_ON_OFF.value:
+            self._state = input.get("value") == "1"
+            return True
+        return False
+
+    async def _set_switching_datapoint(self, value: str):
+        """Set the switching datapoint on the api."""
+        _output_id, _output_value = self.get_output_by_pairing(
+            pairing=Pairing.AL_INFO_ON_OFF
+        )
+        return await self._api.set_datapoint(
+            device_id=self.device_id,
+            channel_id=self.channel_id,
+            datapoint=_output_id,
+            value=value,
+        )

src/abbfreeathome/devices/window_door_sensor_virtual.py

@@ -0,0 +1,67 @@
+"""Free@Home WindowDoorSensorVirtual Class."""
+
+from typing import Any
+
+from ..api import FreeAtHomeApi
+from ..bin.pairing import Pairing
+from .base import Base
+
+
+class WindowDoorSensorVirtual(Base):
+    """Free@Home WindowDoorSensorVirtual Class."""
+
+    def __init__(
+        self,
+        device_id: str,
+        device_name: str,
+        channel_id: str,
+        channel_name: str,
+        inputs: dict[str, dict[str, Any]],
+        outputs: dict[str, dict[str, Any]],
+        parameters: dict[str, dict[str, Any]],
+        api: FreeAtHomeApi,
+        floor_name: str | None = None,
+        room_name: str | None = None,
+    ) -> None:
+        """Initialize the Free@Home WindowDoorSensorVirtual class."""
+        self._state: bool | None = None
+
+        super().__init__(
+            device_id,
+            device_name,
+            channel_id,
+            channel_name,
+            inputs,
+            outputs,
+            parameters,
+            api,
+            floor_name,
+            room_name,
+        )
+
+    @property
+    def state(self) -> bool | None:
+        """Get the sensor state."""
+        return self._state
+
+    async def open(self):
+        """Open the sensor."""
+        await self._set_switching_datapoint("1")
+        self._state = True
+
+    async def close(self):
+        """Close the sensor."""
+        await self._set_switching_datapoint("0")
+        self._state = False
+
+    async def _set_switching_datapoint(self, value: str):
+        """Set the sensor datapoint on the api."""
+        _output_id, _output_value = self.get_output_by_pairing(
+            pairing=Pairing.AL_WINDOW_DOOR
+        )
+        return await self._api.set_datapoint(
+            device_id=self.device_id,
+            channel_id=self.channel_id,
+            datapoint=_output_id,
+            value=value,
+        )

src/abbfreeathome/freeathome.py

@@ -3,7 +3,7 @@
 from .api import FreeAtHomeApi
 from .bin.function import Function
 from .bin.interface import Interface
-from .const import FUNCTION_DEVICE_MAPPING
+from .const import FUNCTION_DEVICE_MAPPING, FUNCTION_DEVICE_MAPPING_VIRTUAL
 from .devices.base import Base
 
 
@@ -16,6 +16,7 @@ def __init__(
         interfaces: list[Interface] | None = None,
         device_classes: list[Base] | None = None,
         include_orphan_channels: bool = False,
+        include_virtual_devices: bool = True,
     ) -> None:
         """Initialize the FreeAtHome class."""
         self._config: dict | None = None
@@ -26,6 +27,7 @@ def __init__(
         self._interfaces: list[Interface] = interfaces
         self._device_classes: list[Base] = device_classes
         self._include_orphan_channels: bool = include_orphan_channels
+        self._include_virtual_devices: bool = include_virtual_devices
 
     def clear_devices(self):
         """Clear all devices in the device list."""
@@ -60,6 +62,10 @@ async def get_devices_by_function(self, function: Function) -> list[dict]:
             ]:
                 continue
 
+            # Filter out virtual devices
+            if _device_key[0:4] == "6000" and not self._include_virtual_devices:
+                continue
+
             for _channel_key, _channel in _device.get("channels", {}).items():
                 # Filter out any channels not on the Free@Home floorplan
                 if (
@@ -81,6 +87,7 @@ async def get_devices_by_function(self, function: Function) -> list[dict]:
                         {
                             "device_id": _device_key,
                             "device_name": _device.get("displayName"),
+                            "is_virtual": _device_key[0:4] == "6000",
                             "channel_id": _channel_key,
                             "channel_name": _channel_name,
                             "function_id": int(_channel.get("functionID"), 16),
@@ -135,6 +142,13 @@ async def get_room_name(
     async def load_devices(self):
         """Load all of the devices into the devices object."""
         self.clear_devices()
+
+        for (
+            _function,
+            _device_class,
+        ) in self._get_function_to_device_mapping_virtual().items():
+            await self._load_devices_by_function(_function, _device_class)
+
         for _function, _device_class in self._get_function_to_device_mapping().items():
             await self._load_devices_by_function(_function, _device_class)
 
@@ -150,6 +164,15 @@ def unload_device_by_device_serial(self, device_serial: str):
     async def _load_devices_by_function(self, function: Function, device_class: Base):
         _devices = await self.get_devices_by_function(function)
         for _device in _devices:
+            if (
+                device_class in FUNCTION_DEVICE_MAPPING_VIRTUAL.values()
+                and not _device.get("is_virtual")
+            ) or (
+                device_class in FUNCTION_DEVICE_MAPPING.values()
+                and _device.get("is_virtual")
+            ):
+                continue
+
             self._devices[f"{_device.get('device_id')}/{_device.get('channel_id')}"] = (
                 device_class(
                     device_id=_device.get("device_id"),
@@ -193,3 +216,14 @@ def _get_function_to_device_mapping(self) -> dict[Function, Base]:
                 if value in self._device_classes
             }
         )
+
+    def _get_function_to_device_mapping_virtual(self) -> dict[Function]:
+        return (
+            FUNCTION_DEVICE_MAPPING_VIRTUAL
+            if not self._device_classes
+            else {
+                key: value
+                for key, value in FUNCTION_DEVICE_MAPPING_VIRTUAL.items()
+                if value in self._device_classes
+            }
+        )