switch.py

Go to the documentation of this file.

"""Support for wired switches attached to a Konnected device."""
 
import logging
from typing import Any
 
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
    ATTR_STATE,
    CONF_DEVICES,
    CONF_NAME,
    CONF_REPEAT,
    CONF_SWITCHES,
    CONF_ZONE,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
from .const import (
    CONF_ACTIVATION,
    CONF_MOMENTARY,
    CONF_PAUSE,
    DOMAIN as KONNECTED_DOMAIN,
    STATE_HIGH,
    STATE_LOW,
)
 
_LOGGER = logging.getLogger(__name__)
 
 
async def async_setup_entry(
    hass: HomeAssistant,
    config_entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up switches attached to a Konnected device from a config entry."""
    data = hass.data[KONNECTED_DOMAIN]
    device_id = config_entry.data["id"]
    switches = [
        KonnectedSwitch(device_id, zone_data.get(CONF_ZONE), zone_data)
        for zone_data in data[CONF_DEVICES][device_id][CONF_SWITCHES]
    ]
    async_add_entities(switches)
 
 
class KonnectedSwitch(SwitchEntity):
    """Representation of a Konnected switch."""
 
    def __init__(self, device_id, zone_num, data):
        """Initialize the Konnected switch."""
        self._data_data = data
        self._device_id_device_id = device_id
        self._zone_num_zone_num = zone_num
        self._activation_activation = self._data_data.get(CONF_ACTIVATION, STATE_HIGH)
        self._momentary_momentary = self._data_data.get(CONF_MOMENTARY)
        self._pause_pause = self._data_data.get(CONF_PAUSE)
        self._repeat_repeat = self._data_data.get(CONF_REPEAT)
        self._attr_is_on_attr_is_on = self._boolean_state_boolean_state(self._data_data.get(ATTR_STATE))
        self._attr_name_attr_name = self._data_data.get(CONF_NAME)
        self._attr_unique_id_attr_unique_id = (
            f"{device_id}-{self._zone_num}-{self._momentary}-"
            f"{self._pause}-{self._repeat}"
        )
        self._attr_device_info_attr_device_info = DeviceInfo(identifiers={(KONNECTED_DOMAIN, device_id)})
 
    @property
    def panel(self):
        """Return the Konnected HTTP client."""
        device_data = self.hasshass.data[KONNECTED_DOMAIN][CONF_DEVICES][self._device_id_device_id]
        return device_data.get("panel")
 
    @property
    def available(self) -> bool:
        """Return whether the panel is available."""
        return self.panelpanel.available
 
    async def async_turn_on(self, **kwargs: Any) -> None:
        """Send a command to turn on the switch."""
        resp = await self.panelpanel.update_switch(
            self._zone_num_zone_num,
            int(self._activation_activation == STATE_HIGH),
            self._momentary_momentary,
            self._repeat_repeat,
            self._pause_pause,
        )
 
        if resp.get(ATTR_STATE) is not None:
            self._set_state_set_state(True)
 
            if self._momentary_momentary and resp.get(ATTR_STATE) != -1:
                # Immediately set the state back off for momentary switches
                self._set_state_set_state(False)
 
    async def async_turn_off(self, **kwargs: Any) -> None:
        """Send a command to turn off the switch."""
        resp = await self.panelpanel.update_switch(
            self._zone_num_zone_num, int(self._activation_activation == STATE_LOW)
        )
 
        if resp.get(ATTR_STATE) is not None:
            self._set_state_set_state(self._boolean_state_boolean_state(resp.get(ATTR_STATE)))
 
    def _boolean_state(self, int_state: int | None) -> bool | None:
        if int_state == 0:
            return self._activation_activation == STATE_LOW
        if int_state == 1:
            return self._activation_activation == STATE_HIGH
        return None
 
    def _set_state(self, state):
        self._attr_is_on_attr_is_on = state
        self.async_write_ha_stateasync_write_ha_state()
        _LOGGER.debug(
            "Setting status of %s actuator zone %s to %s",
            self._device_id_device_id,
            self.namename,
            state,
        )
 
    @callback
    def async_set_state(self, state):
        """Update the switch state."""
        self._set_state_set_state(state)
 
    async def async_added_to_hass(self) -> None:
        """Store entity_id and register state change callback."""
        self._data_data["entity_id"] = self.entity_identity_id
        self.async_on_removeasync_on_remove(
            async_dispatcher_connect(
                self.hasshass, f"konnected.{self.entity_id}.update", self.async_set_stateasync_set_state
            )
        )