switch.py

Go to the documentation of this file.

"""Platform for Omnilogic switch integration."""
 
import time
from typing import Any
 
from omnilogic import OmniLogicException
import voluptuous as vol
 
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv, entity_platform
from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
from .common import check_guard
from .const import COORDINATOR, DOMAIN, PUMP_TYPES
from .coordinator import OmniLogicUpdateCoordinator
from .entity import OmniLogicEntity
 
SERVICE_SET_SPEED = "set_pump_speed"
OMNILOGIC_SWITCH_OFF = 7
 
 
async def async_setup_entry(
    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
    """Set up the light platform."""
 
    coordinator: OmniLogicUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
        COORDINATOR
    ]
    entities = []
 
    for item_id, item in coordinator.data.items():
        id_len = len(item_id)
        item_kind = item_id[-2]
        entity_settings = SWITCH_TYPES.get((id_len, item_kind))
 
        if not entity_settings:
            continue
 
        for entity_setting in entity_settings:
            entity_classes: dict[str, type] = entity_setting["entity_classes"]
            for state_key, entity_class in entity_classes.items():
                if check_guard(state_key, item, entity_setting):
                    continue
 
                entity = entity_class(
                    coordinator=coordinator,
                    state_key=state_key,
                    name=entity_setting["name"],
                    kind=entity_setting["kind"],
                    item_id=item_id,
                    icon=entity_setting["icon"],
                )
 
                entities.append(entity)
 
    async_add_entities(entities)
 
    # register service
    platform = entity_platform.async_get_current_platform()
 
    platform.async_register_entity_service(
        SERVICE_SET_SPEED,
        {vol.Required("speed"): cv.positive_int},
        "async_set_speed",
    )
 
 
class OmniLogicSwitch(OmniLogicEntity, SwitchEntity):
    """Define an Omnilogic Base Switch entity to be extended."""
 
    def __init__(
        self,
        coordinator: OmniLogicUpdateCoordinator,
        kind: str,
        name: str,
        icon: str,
        item_id: tuple,
        state_key: str,
    ) -> None:
        """Initialize Entities."""
        super().__init__(
            coordinator=coordinator,
            kind=kind,
            name=name,
            item_id=item_id,
            icon=icon,
        )
 
        self._state_key_state_key = state_key
        self._state_state = None
        self._last_action_last_action = 0
        self._state_delay_state_delay = 30
 
    @property
    def is_on(self):
        """Return the on/off state of the switch."""
        state_int = 0
 
        # The Omnilogic API has a significant delay in state reporting after
        # calling for a change. This state delay will ensure that HA keeps an
        # optimistic value of state during this period to improve the user
        # experience and avoid confusion.
        if self._last_action_last_action < (time.time() - self._state_delay_state_delay):
            state_int = int(self.coordinator.data[self._item_id_item_id][self._state_key_state_key])
 
            if self._state_state == OMNILOGIC_SWITCH_OFF:
                state_int = 0
 
        self._state_state = state_int != 0
 
        return self._state_state
 
 
class OmniLogicRelayControl(OmniLogicSwitch):
    """Define the OmniLogic Relay entity."""
 
    async def async_turn_on(self, **kwargs):
        """Turn on the relay."""
        self._state_state_state = True
        self._last_action_last_action_last_action = time.time()
        self.async_write_ha_stateasync_write_ha_state()
 
        await self.coordinator.api.set_relay_valve(
            int(self._item_id_item_id[1]),
            int(self._item_id_item_id[3]),
            int(self._item_id_item_id[-1]),
            1,
        )
 
    async def async_turn_off(self, **kwargs):
        """Turn off the relay."""
        self._state_state_state = False
        self._last_action_last_action_last_action = time.time()
        self.async_write_ha_stateasync_write_ha_state()
 
        await self.coordinator.api.set_relay_valve(
            int(self._item_id_item_id[1]),
            int(self._item_id_item_id[3]),
            int(self._item_id_item_id[-1]),
            0,
        )
 
 
class OmniLogicPumpControl(OmniLogicSwitch):
    """Define the OmniLogic Pump Switch Entity."""
 
    def __init__(
        self,
        coordinator: OmniLogicUpdateCoordinator,
        kind: str,
        name: str,
        icon: str,
        item_id: tuple,
        state_key: str,
    ) -> None:
        """Initialize entities."""
        super().__init__(
            coordinator=coordinator,
            kind=kind,
            name=name,
            icon=icon,
            item_id=item_id,
            state_key=state_key,
        )
 
        self._max_speed_max_speed = int(coordinator.data[item_id].get("Max-Pump-Speed", 100))
        self._min_speed_min_speed = int(coordinator.data[item_id].get("Min-Pump-Speed", 0))
 
        if "Filter-Type" in coordinator.data[item_id]:
            self._pump_type_pump_type = PUMP_TYPES[coordinator.data[item_id]["Filter-Type"]]
        else:
            self._pump_type_pump_type = PUMP_TYPES[coordinator.data[item_id]["Type"]]
 
        self._last_speed_last_speed = None
 
    async def async_turn_on(self, **kwargs):
        """Turn on the pump."""
        self._state_state_state = True
        self._last_action_last_action_last_action = time.time()
        self.async_write_ha_stateasync_write_ha_state()
 
        on_value = 100
 
        if self._pump_type_pump_type != "SINGLE" and self._last_speed_last_speed:
            on_value = self._last_speed_last_speed
 
        await self.coordinator.api.set_relay_valve(
            int(self._item_id_item_id[1]),
            int(self._item_id_item_id[3]),
            int(self._item_id_item_id[-1]),
            on_value,
        )
 
    async def async_turn_off(self, **kwargs):
        """Turn off the pump."""
        self._state_state_state = False
        self._last_action_last_action_last_action = time.time()
        self.async_write_ha_stateasync_write_ha_state()
 
        if self._pump_type_pump_type != "SINGLE":
            if "filterSpeed" in self.coordinator.data[self._item_id_item_id]:
                self._last_speed_last_speed = self.coordinator.data[self._item_id_item_id]["filterSpeed"]
            else:
                self._last_speed_last_speed = self.coordinator.data[self._item_id_item_id]["pumpSpeed"]
 
        await self.coordinator.api.set_relay_valve(
            int(self._item_id_item_id[1]),
            int(self._item_id_item_id[3]),
            int(self._item_id_item_id[-1]),
            0,
        )
 
    async def async_set_speed(self, speed):
        """Set the switch speed."""
 
        if self._pump_type_pump_type != "SINGLE":
            if self._min_speed_min_speed <= speed <= self._max_speed_max_speed:
                success = await self.coordinator.api.set_relay_valve(
                    int(self._item_id_item_id[1]),
                    int(self._item_id_item_id[3]),
                    int(self._item_id_item_id[-1]),
                    speed,
                )
 
                if success:
                    self.async_write_ha_stateasync_write_ha_state()
 
            else:
                raise OmniLogicException(
                    "Cannot set speed. Speed is outside pump range."
                )
 
        else:
            raise OmniLogicException("Cannot set speed on a non-variable speed pump.")
 
 
SWITCH_TYPES: dict[tuple[int, str], list[dict[str, Any]]] = {
    (4, "Relays"): [
        {
            "entity_classes": {"switchState": OmniLogicRelayControl},
            "name": "",
            "kind": "relay",
            "icon": None,
            "guard_condition": [],
        },
    ],
    (6, "Relays"): [
        {
            "entity_classes": {"switchState": OmniLogicRelayControl},
            "name": "",
            "kind": "relay",
            "icon": None,
            "guard_condition": [],
        }
    ],
    (6, "Pumps"): [
        {
            "entity_classes": {"pumpState": OmniLogicPumpControl},
            "name": "",
            "kind": "pump",
            "icon": None,
            "guard_condition": [],
        }
    ],
    (6, "Filter"): [
        {
            "entity_classes": {"filterState": OmniLogicPumpControl},
            "name": "",
            "kind": "pump",
            "icon": None,
            "guard_condition": [],
        }
    ],
}