entity.py

Go to the documentation of this file.

"""Homematic base entity."""
 
from __future__ import annotations
 
from abc import abstractmethod
from datetime import timedelta
import logging
 
from pyhomematic import HMConnection
from pyhomematic.devicetypes.generic import HMGeneric
 
from homeassistant.const import ATTR_NAME
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity, EntityDescription
from homeassistant.helpers.event import track_time_interval
 
from .const import (
    ATTR_ADDRESS,
    ATTR_CHANNEL,
    ATTR_INTERFACE,
    ATTR_PARAM,
    ATTR_UNIQUE_ID,
    DATA_HOMEMATIC,
    DOMAIN,
    HM_ATTRIBUTE_SUPPORT,
)
 
_LOGGER = logging.getLogger(__name__)
 
SCAN_INTERVAL_HUB = timedelta(seconds=300)
SCAN_INTERVAL_VARIABLES = timedelta(seconds=30)
 
 
class HMDevice(Entity):
    """The HomeMatic device base object."""
 
    _homematic: HMConnection
    _hmdevice: HMGeneric
    _attr_should_poll = False
 
    def __init__(
        self,
        config: dict[str, str],
        entity_description: EntityDescription | None = None,
    ) -> None:
        """Initialize a generic HomeMatic device."""
        self._name_name = config.get(ATTR_NAME)
        self._address_address = config.get(ATTR_ADDRESS)
        self._interface_interface = config.get(ATTR_INTERFACE)
        self._channel_channel = config.get(ATTR_CHANNEL)
        self._state_state = config.get(ATTR_PARAM)
        self._unique_id_unique_id = config.get(ATTR_UNIQUE_ID)
        self._data: dict[str, str] = {}
        self._connected_connected = False
        self._available_available = False
        self._channel_map: dict[str, str] = {}
 
        if entity_description is not None:
            self.entity_descriptionentity_description = entity_description
 
        # Set parameter to uppercase
        if self._state_state:
            self._state_state = self._state_state.upper()
 
    async def async_added_to_hass(self):
        """Load data init callbacks."""
        self._subscribe_homematic_events_subscribe_homematic_events()
 
    @property
    def unique_id(self):
        """Return unique ID. HomeMatic entity IDs are unique by default."""
        return self._unique_id_unique_id.replace(" ", "_")
 
    @property
    def name(self):
        """Return the name of the device."""
        return self._name_name
 
    @property
    def available(self):
        """Return true if device is available."""
        return self._available_available
 
    @property
    def extra_state_attributes(self):
        """Return device specific state attributes."""
        # Static attributes
        attr = {
            "id": self._hmdevice_hmdevice.ADDRESS,
            "interface": self._interface_interface,
        }
 
        # Generate a dictionary with attributes
        for node, data in HM_ATTRIBUTE_SUPPORT.items():
            # Is an attribute and exists for this object
            if node in self._data:
                value = data[1].get(self._data[node], self._data[node])
                attr[data[0]] = value
 
        return attr
 
    def update(self):
        """Connect to HomeMatic init values."""
        if self._connected_connected:
            return True
 
        # Initialize
        self._homematic_homematic = self.hasshass.data[DATA_HOMEMATIC]
        self._hmdevice_hmdevice = self._homematic_homematic.devices[self._interface_interface][self._address_address]
        self._connected_connected = True
 
        try:
            # Initialize datapoints of this object
            self._init_data_init_data()
            self._load_data_from_hm_load_data_from_hm()
 
            # Link events from pyhomematic
            self._available_available = not self._hmdevice_hmdevice.UNREACH
        except Exception as err:  # noqa: BLE001
            self._connected_connected = False
            _LOGGER.error("Exception while linking %s: %s", self._address_address, str(err))
 
    def _hm_event_callback(self, device, caller, attribute, value):
        """Handle all pyhomematic device events."""
        has_changed = False
 
        # Is data needed for this instance?
        if device.partition(":")[2] == self._channel_map.get(attribute):
            self._data[attribute] = value
            has_changed = True
 
        # Availability has changed
        if self.availableavailableavailable != (not self._hmdevice_hmdevice.UNREACH):
            self._available_available = not self._hmdevice_hmdevice.UNREACH
            has_changed = True
 
        # If it has changed data point, update Home Assistant
        if has_changed:
            self.schedule_update_ha_stateschedule_update_ha_state()
 
    def _subscribe_homematic_events(self):
        """Subscribe all required events to handle job."""
        for metadata in (
            self._hmdevice_hmdevice.ACTIONNODE,
            self._hmdevice_hmdevice.EVENTNODE,
            self._hmdevice_hmdevice.WRITENODE,
            self._hmdevice_hmdevice.ATTRIBUTENODE,
            self._hmdevice_hmdevice.BINARYNODE,
            self._hmdevice_hmdevice.SENSORNODE,
        ):
            for node, channels in metadata.items():
                # Data is needed for this instance
                if node in self._data:
                    # chan is current channel
                    if len(channels) == 1:
                        channel = channels[0]
                    else:
                        channel = self._channel_channel
                    # Remember the channel for this attribute to ignore invalid events later
                    self._channel_map[node] = str(channel)
 
        _LOGGER.debug("Channel map for %s: %s", self._address_address, str(self._channel_map))
 
        # Set callbacks
        self._hmdevice_hmdevice.setEventCallback(callback=self._hm_event_callback_hm_event_callback, bequeath=True)
 
    def _load_data_from_hm(self):
        """Load first value from pyhomematic."""
        if not self._connected_connected:
            return False
 
        # Read data from pyhomematic
        for metadata, funct in (
            (self._hmdevice_hmdevice.ATTRIBUTENODE, self._hmdevice_hmdevice.getAttributeData),
            (self._hmdevice_hmdevice.WRITENODE, self._hmdevice_hmdevice.getWriteData),
            (self._hmdevice_hmdevice.SENSORNODE, self._hmdevice_hmdevice.getSensorData),
            (self._hmdevice_hmdevice.BINARYNODE, self._hmdevice_hmdevice.getBinaryData),
        ):
            for node in metadata:
                if metadata[node] and node in self._data:
                    self._data[node] = funct(name=node, channel=self._channel_channel)
 
        return True
 
    def _hm_set_state(self, value):
        """Set data to main datapoint."""
        if self._state_state in self._data:
            self._data[self._state_state] = value
 
    def _hm_get_state(self):
        """Get data from main datapoint."""
        if self._state_state in self._data:
            return self._data[self._state_state]
        return None
 
    def _init_data(self):
        """Generate a data dict (self._data) from the HomeMatic metadata."""
        # Add all attributes to data dictionary
        for data_note in self._hmdevice_hmdevice.ATTRIBUTENODE:
            self._data.update({data_note: None})
 
        # Initialize device specific data
        self._init_data_struct_init_data_struct()
 
    @abstractmethod
    def _init_data_struct(self):
        """Generate a data dictionary from the HomeMatic device metadata."""
 
 
class HMHub(Entity):
    """The HomeMatic hub. (CCU2/HomeGear)."""
 
    _attr_should_poll = False
 
    def __init__(self, hass, homematic, name):
        """Initialize HomeMatic hub."""
        self.hasshasshass = hass
        self.entity_identity_identity_id = f"{DOMAIN}.{name.lower()}"
        self._homematic_homematic = homematic
        self._variables_variables = {}
        self._name_name = name
        self._state_state = None
 
        # Load data
        track_time_interval(self.hasshasshass, self._update_hub_update_hub, SCAN_INTERVAL_HUB)
        self.hasshasshass.add_job(self._update_hub_update_hub, None)
 
        track_time_interval(self.hasshasshass, self._update_variables_update_variables, SCAN_INTERVAL_VARIABLES)
        self.hasshasshass.add_job(self._update_variables_update_variables, None)
 
    @property
    def name(self):
        """Return the name of the device."""
        return self._name_name
 
    @property
    def state(self):
        """Return the state of the entity."""
        return self._state_state
 
    @property
    def extra_state_attributes(self):
        """Return the state attributes."""
        return self._variables_variables.copy()
 
    @property
    def icon(self):
        """Return the icon to use in the frontend, if any."""
        return "mdi:gradient-vertical"
 
    def _update_hub(self, now):
        """Retrieve latest state."""
        service_message = self._homematic_homematic.getServiceMessages(self._name_name)
        state = None if service_message is None else len(service_message)
 
        # state have change?
        if self._state_state != state:
            self._state_state = state
            self.schedule_update_ha_stateschedule_update_ha_state()
 
    def _update_variables(self, now):
        """Retrieve all variable data and update hmvariable states."""
        variables = self._homematic_homematic.getAllSystemVariables(self._name_name)
        if variables is None:
            return
 
        state_change = False
        for key, value in variables.items():
            if key in self._variables_variables and value == self._variables_variables[key]:
                continue
 
            state_change = True
            self._variables_variables.update({key: value})
 
        if state_change:
            self.schedule_update_ha_stateschedule_update_ha_state()
 
    def hm_set_variable(self, name, value):
        """Set variable value on CCU/Homegear."""
        if name not in self._variables_variables:
            _LOGGER.error("Variable %s not found on %s", name, self.namenamename)
            return
        old_value = self._variables_variables.get(name)
        if isinstance(old_value, bool):
            value = cv.boolean(value)
        else:
            value = float(value)
        self._homematic_homematic.setSystemVariable(self.namenamename, name, value)
 
        self._variables_variables.update({name: value})
        self.schedule_update_ha_stateschedule_update_ha_state()