entity.py

Go to the documentation of this file.

"""Support for MotionMount sensors."""
 
import logging
import socket
from typing import TYPE_CHECKING
 
import motionmount
 
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_CONNECTIONS, ATTR_IDENTIFIERS
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import DeviceInfo, format_mac
from homeassistant.helpers.entity import Entity
 
from .const import DOMAIN, EMPTY_MAC
 
_LOGGER = logging.getLogger(__name__)
 
 
class MotionMountEntity(Entity):
    """Representation of a MotionMount entity."""
 
    _attr_should_poll = False
    _attr_has_entity_name = True
 
    def __init__(self, mm: motionmount.MotionMount, config_entry: ConfigEntry) -> None:
        """Initialize general MotionMount entity."""
        self.mmmm = mm
        mac = format_mac(mm.mac.hex())
 
        # Create a base unique id
        if mac == EMPTY_MAC:
            self._base_unique_id_base_unique_id = config_entry.entry_id
        else:
            self._base_unique_id_base_unique_id = mac
 
        # Set device info
        self._attr_device_info_attr_device_info = DeviceInfo(
            name=mm.name,
            manufacturer="Vogel's",
            model="MotionMount SIGNATURE Pro",
            model_id="TVM 7675 Pro",
        )
 
        if mac == EMPTY_MAC:
            self._attr_device_info_attr_device_info[ATTR_IDENTIFIERS] = {(DOMAIN, config_entry.entry_id)}
        else:
            self._attr_device_info_attr_device_info[ATTR_CONNECTIONS] = {
                (dr.CONNECTION_NETWORK_MAC, mac)
            }
 
    @property
    def available(self) -> bool:
        """Return True if the MotionMount is available (we're connected)."""
        return self.mmmm.is_connected
 
    def update_name(self) -> None:
        """Update the name of the associated device."""
        if TYPE_CHECKING:
            assert self.device_entrydevice_entry
        # Update the name in the device registry if needed
        if self.device_entrydevice_entry.name != self.mmmm.name:
            device_registry = dr.async_get(self.hasshass)
            device_registry.async_update_device(self.device_entrydevice_entry.id, name=self.mmmm.name)
 
    async def async_added_to_hass(self) -> None:
        """Store register state change callback."""
        self.mmmm.add_listener(self.async_write_ha_stateasync_write_ha_state)
        self.mmmm.add_listener(self.update_nameupdate_name)
        await super().async_added_to_hass()
 
    async def async_will_remove_from_hass(self) -> None:
        """Remove register state change callback."""
        self.mmmm.remove_listener(self.async_write_ha_stateasync_write_ha_state)
        self.mmmm.remove_listener(self.update_nameupdate_name)
        await super().async_will_remove_from_hass()
 
    async def _ensure_connected(self) -> bool:
        """Make sure there is a connection with the MotionMount.
 
        Returns false if the connection failed to be ensured.
        """
 
        if self.mmmm.is_connected:
            return True
        try:
            await self.mmmm.connect()
        except (ConnectionError, TimeoutError, socket.gaierror):
            # We're not interested in exceptions here. In case of a failed connection
            # the try/except from the caller will report it.
            # The purpose of `_ensure_connected()` is only to make sure we try to
            # reconnect, where failures should not be logged each time
            return False
        else:
            _LOGGER.warning("Successfully reconnected to MotionMount")
            return True