Add port devices and enable toggles

custom_components/ha_knx_bridge/switch.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.components.switch import SwitchEntity
+from homeassistant.helpers import device_registry as dr
+
+from .const import CONF_PORTS, CONF_PORT_ENABLED, DOMAIN
+from .bridge import iter_ports
+
+
+@dataclass(frozen=True)
+class PortDescriptor:
+    port_id: str
+    port_type: str
+    title: str
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities
+) -> None:
+    device_registry = dr.async_get(hass)
+    device_registry.async_get_or_create(
+        config_entry_id=entry.entry_id,
+        identifiers={(DOMAIN, entry.entry_id)},
+        name="HA KNX Bridge",
+        manufacturer="HA KNX Bridge",
+        model="Integration",
+    )
+    entities: list[SwitchEntity] = []
+    for port in iter_ports(entry):
+        entities.append(
+            PortEnabledSwitch(entry, port.port_id, port.port_type, port.title)
+        )
+    async_add_entities(entities, update_before_add=True)
+
+
+class PortEnabledSwitch(SwitchEntity):
+    _attr_has_entity_name = True
+
+    def __init__(self, entry: ConfigEntry, port_id: str, port_type: str, title: str):
+        self._entry = entry
+        self._port_id = port_id
+        self._port_type = port_type
+        self._title = title
+        self._attr_unique_id = f"{entry.entry_id}_{port_id}_enabled"
+
+    @property
+    def name(self) -> str:
+        return f"{self._title} Enabled"
+
+    @property
+    def device_info(self) -> DeviceInfo:
+        return DeviceInfo(
+            identifiers={(DOMAIN, self._port_id)},
+            name=f"{self._title} Port",
+            manufacturer="HA KNX Bridge",
+            model=self._port_type,
+            via_device=(DOMAIN, self._entry.entry_id),
+        )
+
+    @property
+    def is_on(self) -> bool:
+        overrides = self._entry.options.get(CONF_PORT_ENABLED, {})
+        if self._port_id in overrides:
+            return bool(overrides[self._port_id])
+        return True
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        await self._async_set_enabled(True)
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        await self._async_set_enabled(False)
+
+    async def _async_set_enabled(self, enabled: bool) -> None:
+        overrides = dict(self._entry.options.get(CONF_PORT_ENABLED, {}))
+        overrides[self._port_id] = enabled
+        self._entry.async_update_options(
+            {**self._entry.options, CONF_PORT_ENABLED: overrides}
+        )
+        await self.hass.config_entries.async_reload(self._entry.entry_id)