diff --git a/src/abbfreeathome/bin/interface.py b/src/abbfreeathome/bin/interface.py index 3d41a93..7986829 100644 --- a/src/abbfreeathome/bin/interface.py +++ b/src/abbfreeathome/bin/interface.py @@ -16,4 +16,3 @@ class Interface(enum.Enum): WIRELESS_RF = "RF" HUE = "hue" SONOS = "sonos" - VIRTUAL_DEVICE = "vdev:installer@busch-jaeger.de" diff --git a/src/abbfreeathome/const.py b/src/abbfreeathome/const.py index 126c408..cdb8a85 100644 --- a/src/abbfreeathome/const.py +++ b/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, diff --git a/src/abbfreeathome/devices/base.py b/src/abbfreeathome/devices/base.py index 3150ce2..8d0cb8c 100644 --- a/src/abbfreeathome/devices/base.py +++ b/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,6 +149,24 @@ 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.""" @@ -149,3 +174,11 @@ 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) diff --git a/src/abbfreeathome/devices/switch_actuator_virtual.py b/src/abbfreeathome/devices/switch_actuator_virtual.py new file mode 100644 index 0000000..66e3a57 --- /dev/null +++ b/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, + ) diff --git a/src/abbfreeathome/devices/window_door_sensor_virtual.py b/src/abbfreeathome/devices/window_door_sensor_virtual.py new file mode 100644 index 0000000..d99d6cb --- /dev/null +++ b/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, + ) diff --git a/src/abbfreeathome/freeathome.py b/src/abbfreeathome/freeathome.py index 575c342..5c77a77 100644 --- a/src/abbfreeathome/freeathome.py +++ b/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 + } + )