config.py

Go to the documentation of this file.

"""Support for local control of entities by emulating a Philips Hue bridge."""
 
from __future__ import annotations
 
from functools import cache
import logging
 
from homeassistant.components import (
    climate,
    cover,
    fan,
    humidifier,
    light,
    media_player,
    scene,
    script,
)
from homeassistant.const import CONF_ENTITIES, CONF_TYPE
from homeassistant.core import (
    Event,
    EventStateChangedData,
    HomeAssistant,
    State,
    callback,
    split_entity_id,
)
from homeassistant.helpers import storage
from homeassistant.helpers.event import (
    async_track_state_added_domain,
    async_track_state_removed_domain,
)
from homeassistant.helpers.typing import ConfigType
 
SUPPORTED_DOMAINS = {
    climate.DOMAIN,
    cover.DOMAIN,
    fan.DOMAIN,
    humidifier.DOMAIN,
    light.DOMAIN,
    media_player.DOMAIN,
    scene.DOMAIN,
    script.DOMAIN,
}
 
 
TYPE_ALEXA = "alexa"
TYPE_GOOGLE = "google_home"
 
 
NUMBERS_FILE = "emulated_hue_ids.json"
DATA_KEY = "emulated_hue.ids"
DATA_VERSION = "1"
SAVE_DELAY = 60
 
CONF_ADVERTISE_IP = "advertise_ip"
CONF_ADVERTISE_PORT = "advertise_port"
CONF_ENTITY_HIDDEN = "hidden"
CONF_ENTITY_NAME = "name"
CONF_EXPOSE_BY_DEFAULT = "expose_by_default"
CONF_EXPOSED_DOMAINS = "exposed_domains"
CONF_HOST_IP = "host_ip"
CONF_LIGHTS_ALL_DIMMABLE = "lights_all_dimmable"
CONF_LISTEN_PORT = "listen_port"
CONF_OFF_MAPS_TO_ON_DOMAINS = "off_maps_to_on_domains"
CONF_UPNP_BIND_MULTICAST = "upnp_bind_multicast"
 
 
DEFAULT_LIGHTS_ALL_DIMMABLE = False
DEFAULT_LISTEN_PORT = 8300
DEFAULT_UPNP_BIND_MULTICAST = True
DEFAULT_OFF_MAPS_TO_ON_DOMAINS = {"script", "scene"}
DEFAULT_EXPOSE_BY_DEFAULT = True
DEFAULT_EXPOSED_DOMAINS = [
    "switch",
    "light",
    "group",
    "input_boolean",
    "media_player",
    "fan",
]
DEFAULT_TYPE = TYPE_GOOGLE
 
ATTR_EMULATED_HUE_NAME = "emulated_hue_name"
 
 
_LOGGER = logging.getLogger(__name__)
 
 
class Config:
    """Hold configuration variables for the emulated hue bridge."""
 
    def __init__(self, hass: HomeAssistant, conf: ConfigType, local_ip: str) -> None:
        """Initialize the instance."""
        self.hasshass = hass
        self.typetype = conf.get(CONF_TYPE)
        self.numbersnumbers: dict[str, str] = {}
        self.storestore: storage.Store | None = None
        self.cached_states: dict[str, list] = {}
        self._exposed_cache: dict[str, bool] = {}
 
        if self.typetype == TYPE_ALEXA:
            _LOGGER.warning(
                "Emulated Hue running in legacy mode because type has been "
                "specified. More info at https://goo.gl/M6tgz8"
            )
 
        # Get the IP address that will be passed to the Echo during discovery
        self.host_ip_addr: str = conf.get(CONF_HOST_IP) or local_ip
 
        # Get the port that the Hue bridge will listen on
        self.listen_port: int = conf.get(CONF_LISTEN_PORT) or DEFAULT_LISTEN_PORT
 
        # Get whether or not UPNP binds to multicast address (239.255.255.250)
        # or to the unicast address (host_ip_addr)
        self.upnp_bind_multicast: bool = conf.get(
            CONF_UPNP_BIND_MULTICAST, DEFAULT_UPNP_BIND_MULTICAST
        )
 
        # Get domains that cause both "on" and "off" commands to map to "on"
        # This is primarily useful for things like scenes or scripts, which
        # don't really have a concept of being off
        off_maps_to_on_domains = conf.get(CONF_OFF_MAPS_TO_ON_DOMAINS)
        if isinstance(off_maps_to_on_domains, list):
            self.off_maps_to_on_domainsoff_maps_to_on_domains = set(off_maps_to_on_domains)
        else:
            self.off_maps_to_on_domainsoff_maps_to_on_domains = DEFAULT_OFF_MAPS_TO_ON_DOMAINS
 
        # Get whether or not entities should be exposed by default, or if only
        # explicitly marked ones will be exposed
        self.expose_by_default: bool = conf.get(
            CONF_EXPOSE_BY_DEFAULT, DEFAULT_EXPOSE_BY_DEFAULT
        )
 
        # Get domains that are exposed by default when expose_by_default is
        # True
        self.exposed_domainsexposed_domains = set(
            conf.get(CONF_EXPOSED_DOMAINS, DEFAULT_EXPOSED_DOMAINS)
        )
 
        # Calculated effective advertised IP and port for network isolation
        self.advertise_ip: str = conf.get(CONF_ADVERTISE_IP) or self.host_ip_addr
 
        self.advertise_port: int = conf.get(CONF_ADVERTISE_PORT) or self.listen_port
 
        self.entities: dict[str, dict[str, str]] = conf.get(CONF_ENTITIES, {})
 
        self._entities_with_hidden_attr_in_config_entities_with_hidden_attr_in_config = {}
        for entity_id in self.entities:
            hidden_value = self.entities[entity_id].get(CONF_ENTITY_HIDDEN)
            if hidden_value is not None:
                self._entities_with_hidden_attr_in_config_entities_with_hidden_attr_in_config[entity_id] = hidden_value
 
        # Get whether all non-dimmable lights should be reported as dimmable
        # for compatibility with older installations.
        self.lights_all_dimmable: bool = conf.get(CONF_LIGHTS_ALL_DIMMABLE) or False
 
        if self.expose_by_default:
            self.track_domainstrack_domains = set(self.exposed_domainsexposed_domains) or SUPPORTED_DOMAINS
        else:
            self.track_domainstrack_domains = {
                split_entity_id(entity_id)[0] for entity_id in self.entities
            }
 
    async def async_setup(self) -> None:
        """Set up tracking and migrate to storage."""
        hass = self.hasshass
        self.storestore = storage.Store(hass, DATA_VERSION, DATA_KEY)  # type: ignore[arg-type]
        numbers_path = hass.config.path(NUMBERS_FILE)
        self.numbersnumbers = (
            await storage.async_migrator(hass, numbers_path, self.storestore) or {}
        )
        async_track_state_added_domain(
            hass, self.track_domainstrack_domains, self._clear_exposed_cache_clear_exposed_cache
        )
        async_track_state_removed_domain(
            hass, self.track_domainstrack_domains, self._clear_exposed_cache_clear_exposed_cache
        )
 
    @cache  # pylint: disable=method-cache-max-size-none
    def entity_id_to_number(self, entity_id: str) -> str:
        """Get a unique number for the entity id."""
        if self.typetype == TYPE_ALEXA:
            return entity_id
 
        # Google Home
        for number, ent_id in self.numbersnumbers.items():
            if entity_id == ent_id:
                return number
 
        number = "1"
        if self.numbersnumbers:
            number = str(max(int(k) for k in self.numbersnumbers) + 1)
        self.numbersnumbers[number] = entity_id
        assert self.storestore is not None
        self.storestore.async_delay_save(lambda: self.numbersnumbers, SAVE_DELAY)
        return number
 
    def number_to_entity_id(self, number: str) -> str | None:
        """Convert unique number to entity id."""
        if self.typetype == TYPE_ALEXA:
            return number
 
        # Google Home
        return self.numbersnumbers.get(number)
 
    def get_entity_name(self, state: State) -> str:
        """Get the name of an entity."""
        if (
            state.entity_id in self.entities
            and CONF_ENTITY_NAME in self.entities[state.entity_id]
        ):
            return self.entities[state.entity_id][CONF_ENTITY_NAME]
 
        return state.attributes.get(ATTR_EMULATED_HUE_NAME, state.name)  # type: ignore[no-any-return]
 
    @cache  # pylint: disable=method-cache-max-size-none
    def get_exposed_entity_ids(self) -> list[str]:
        """Return a list of exposed states."""
        state_machine = self.hasshass.states
        if self.expose_by_default:
            return [
                state.entity_id
                for state in state_machine.async_all()
                if self.is_state_exposedis_state_exposed(state)
            ]
        return [
            entity_id
            for entity_id in self.entities
            if (state := state_machine.get(entity_id)) and self.is_state_exposedis_state_exposed(state)
        ]
 
    @callback
    def _clear_exposed_cache(self, event: Event[EventStateChangedData]) -> None:
        """Clear the cache of exposed entity ids."""
        self.get_exposed_entity_idsget_exposed_entity_ids.cache_clear()
 
    def is_state_exposed(self, state: State) -> bool:
        """Cache determine if an entity should be exposed on the emulated bridge."""
        if (exposed := self._exposed_cache.get(state.entity_id)) is not None:
            return exposed
        exposed = self._is_state_exposed_is_state_exposed(state)
        self._exposed_cache[state.entity_id] = exposed
        return exposed
 
    def _is_state_exposed(self, state: State) -> bool:
        """Determine if an entity state should be exposed on the emulated bridge.
 
        Async friendly.
        """
        if state.attributes.get("view") is not None:
            # Ignore entities that are views
            return False
 
        if state.entity_id in self._entities_with_hidden_attr_in_config_entities_with_hidden_attr_in_config:
            return not self._entities_with_hidden_attr_in_config_entities_with_hidden_attr_in_config[state.entity_id]
 
        if not self.expose_by_default:
            return False
        # Expose an entity if the entity's domain is exposed by default and
        # the configuration doesn't explicitly exclude it from being
        # exposed, or if the entity is explicitly exposed
        if state.domain in self.exposed_domainsexposed_domains:
            return True
 
        return False