switch.py

Go to the documentation of this file.

"""Support for Ankuoo RecSwitch MS6126 devices."""
 
from __future__ import annotations
 
import logging
from typing import Any
 
from pyrecswitch import RSNetwork, RSNetworkError
import voluptuous as vol
 
from homeassistant.components.switch import (
    PLATFORM_SCHEMA as SWITCH_PLATFORM_SCHEMA,
    SwitchEntity,
)
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_NAME
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 
_LOGGER = logging.getLogger(__name__)
 
DEFAULT_NAME = "RecSwitch {0}"
 
DATA_RSN = "RSN"
 
PLATFORM_SCHEMA = SWITCH_PLATFORM_SCHEMA.extend(
    {
        vol.Required(CONF_HOST): cv.string,
        vol.Required(CONF_MAC): vol.All(cv.string, vol.Upper),
        vol.Optional(CONF_NAME): cv.string,
    }
)
 
 
async def async_setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    async_add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None,
) -> None:
    """Set up the device."""
 
    host = config[CONF_HOST]
    mac_address = config[CONF_MAC]
    device_name = config.get(CONF_NAME)
 
    if not hass.data.get(DATA_RSN):
        hass.data[DATA_RSN] = RSNetwork()
        job = hass.data[DATA_RSN].create_datagram_endpoint()
        hass.async_create_task(job)
 
    device = hass.data[DATA_RSN].register_device(mac_address, host)
    async_add_entities([RecSwitchSwitch(device, device_name, mac_address)])
 
 
class RecSwitchSwitch(SwitchEntity):
    """Representation of a recswitch device."""
 
    def __init__(self, device, device_name, mac_address):
        """Initialize a recswitch device."""
        self.gpio_stategpio_state = False
        self.devicedevice = device
        self.device_namedevice_name = device_name
        self.mac_addressmac_address = mac_address
        if not self.device_namedevice_name:
            self.device_namedevice_name = DEFAULT_NAME.format(self.mac_addressmac_address)
 
    @property
    def unique_id(self):
        """Return the switch unique ID."""
        return self.mac_addressmac_address
 
    @property
    def name(self):
        """Return the switch name."""
        return self.device_namedevice_name
 
    @property
    def is_on(self):
        """Return true if switch is on."""
        return self.gpio_stategpio_state
 
    async def async_turn_on(self, **kwargs: Any) -> None:
        """Turn on the switch."""
        await self.async_set_gpio_statusasync_set_gpio_status(True)
 
    async def async_turn_off(self, **kwargs: Any) -> None:
        """Turn off the switch."""
        await self.async_set_gpio_statusasync_set_gpio_status(False)
 
    async def async_set_gpio_status(self, status):
        """Set the switch status."""
 
        try:
            ret = await self.devicedevice.set_gpio_status(status)
            self.gpio_stategpio_state = ret.state
        except RSNetworkError as error:
            _LOGGER.error("Setting status to %s: %r", self.namenamename, error)
 
    async def async_update(self) -> None:
        """Update the current switch status."""
 
        try:
            ret = await self.devicedevice.get_gpio_status()
            self.gpio_stategpio_state = ret.state
        except RSNetworkError as error:
            _LOGGER.error("Reading status from %s: %r", self.namenamename, error)