Upload files to "/"

switch.py

@@ -0,0 +1,172 @@
+"""Switch platform for the Riedel RRCS integration."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from homeassistant.components.switch import SwitchEntity
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .coordinator import RRCSCoordinator
+from .entity import RRCSEntity
+from .models import RRCSConfigEntry
+from .rrcs import GpioAddress, RRCSError
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: RRCSConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up the switch entities."""
+    runtime = entry.runtime_data
+    coordinator = runtime.coordinator
+
+    entities: list[SwitchEntity] = [
+        RRCSLogicSourceSwitch(coordinator, entry.entry_id, object_id)
+        for object_id in (coordinator.data.logic_sources if coordinator.data else {})
+    ]
+    entities.extend(
+        RRCSGpOutputSwitch(
+            coordinator,
+            entry.entry_id,
+            address,
+            runtime.gpio_names.get(address.key),
+        )
+        for address in runtime.gpio_outputs
+    )
+
+    async_add_entities(entities)
+
+    known = {
+        object_id for object_id in (coordinator.data.logic_sources if coordinator.data else {})
+    }
+
+    @callback
+    def _add_new_logic_sources() -> None:
+        """Pick up logic sources added to the Artist configuration later."""
+        if coordinator.data is None:
+            return
+        new = set(coordinator.data.logic_sources) - known
+        if not new:
+            return
+        known.update(new)
+        async_add_entities(
+            RRCSLogicSourceSwitch(coordinator, entry.entry_id, object_id)
+            for object_id in new
+        )
+
+    entry.async_on_unload(coordinator.async_add_listener(_add_new_logic_sources))
+
+
+class RRCSLogicSourceSwitch(RRCSEntity, SwitchEntity):
+    """A logic source, driven through SetLogicSourceState."""
+
+    _attr_icon = "mdi:toggle-switch-outline"
+
+    def __init__(
+        self, coordinator: RRCSCoordinator, entry_id: str, object_id: int
+    ) -> None:
+        """Initialise the switch."""
+        super().__init__(coordinator, entry_id)
+        self._object_id = object_id
+        self._attr_unique_id = f"{entry_id}_logic_source_{object_id}"
+        source = coordinator.data.logic_sources.get(object_id) if coordinator.data else None
+        self._attr_name = (source.long_name or source.label) if source else f"Logic source {object_id}"
+
+    @property
+    def available(self) -> bool:
+        """Return whether the logic source is still in the configuration."""
+        return (
+            super().available
+            and self.coordinator.data is not None
+            and self._object_id in self.coordinator.data.logic_sources
+        )
+
+    @property
+    def is_on(self) -> bool | None:
+        """Return the logic source state."""
+        if self.coordinator.data is None:
+            return None
+        source = self.coordinator.data.logic_sources.get(self._object_id)
+        return source.state if source else None
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any]:
+        """Expose the Artist identifiers for use in automations."""
+        source = (
+            self.coordinator.data.logic_sources.get(self._object_id)
+            if self.coordinator.data
+            else None
+        )
+        return {
+            "object_id": self._object_id,
+            "label": source.label if source else None,
+        }
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Activate the logic source."""
+        await self._async_set(True)
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Deactivate the logic source.
+
+        RRCS only clears logic sources it set itself; one driven from a panel
+        key will not respond here.
+        """
+        await self._async_set(False)
+
+    async def _async_set(self, state: bool) -> None:
+        try:
+            await self.coordinator.client.set_logic_source(self._object_id, state)
+        except RRCSError as err:
+            raise HomeAssistantError(f"Could not set logic source: {err}") from err
+        self.coordinator.apply_logic_source_change(self._object_id, state)
+
+
+class RRCSGpOutputSwitch(RRCSEntity, SwitchEntity):
+    """A general purpose output."""
+
+    _attr_icon = "mdi:electric-switch"
+
+    def __init__(
+        self,
+        coordinator: RRCSCoordinator,
+        entry_id: str,
+        address: GpioAddress,
+        name: str | None,
+    ) -> None:
+        """Initialise the switch."""
+        super().__init__(coordinator, entry_id)
+        self._address = address
+        self._attr_unique_id = f"{entry_id}_gpo_{address.key}"
+        self._attr_name = name or f"GP out {address.node}.{address.slot}.{address.index}"
+
+    @property
+    def is_on(self) -> bool | None:
+        """Return the output state."""
+        if self.coordinator.data is None:
+            return None
+        return self.coordinator.data.gpio_outputs.get(self._address.key)
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any]:
+        """Expose the GPIO address."""
+        return {"address": self._address.pretty}
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Close the output."""
+        await self._async_set(True)
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Open the output."""
+        await self._async_set(False)
+
+    async def _async_set(self, state: bool) -> None:
+        try:
+            await self.coordinator.client.set_gp_output(self._address, state)
+        except RRCSError as err:
+            raise HomeAssistantError(f"Could not switch GP output: {err}") from err
+        self.coordinator.apply_gpio_change(self._address, state)