__init__.py

Go to the documentation of this file.

"""The Samsung TV integration."""
 
from __future__ import annotations
 
from collections.abc import Coroutine, Mapping
from functools import partial
from typing import Any
from urllib.parse import urlparse
 
import getmac
 
from homeassistant.components import ssdp
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
    CONF_HOST,
    CONF_MAC,
    CONF_METHOD,
    CONF_MODEL,
    CONF_PORT,
    CONF_TOKEN,
    EVENT_HOMEASSISTANT_STOP,
    Platform,
)
from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.debounce import Debouncer
 
from .bridge import (
    SamsungTVBridge,
    async_get_device_info,
    mac_from_device_info,
    model_requires_encryption,
)
from .const import (
    CONF_SESSION_ID,
    CONF_SSDP_MAIN_TV_AGENT_LOCATION,
    CONF_SSDP_RENDERING_CONTROL_LOCATION,
    ENTRY_RELOAD_COOLDOWN,
    LEGACY_PORT,
    LOGGER,
    METHOD_ENCRYPTED_WEBSOCKET,
    METHOD_LEGACY,
    UPNP_SVC_MAIN_TV_AGENT,
    UPNP_SVC_RENDERING_CONTROL,
)
from .coordinator import SamsungTVDataUpdateCoordinator
 
PLATFORMS = [Platform.MEDIA_PLAYER, Platform.REMOTE]
 
 
SamsungTVConfigEntry = ConfigEntry[SamsungTVDataUpdateCoordinator]
 
 
@callback
def _async_get_device_bridge(
    hass: HomeAssistant, data: dict[str, Any]
) -> SamsungTVBridge:
    """Get device bridge."""
    return SamsungTVBridge.get_bridge(
        hass,
        data[CONF_METHOD],
        data[CONF_HOST],
        data[CONF_PORT],
        data,
    )
 
 
class DebouncedEntryReloader:
    """Reload only after the timer expires."""
 
    def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
        """Init the debounced entry reloader."""
        self.hasshass = hass
        self.entryentry = entry
        self.tokentoken = self.entryentry.data.get(CONF_TOKEN)
        self._debounced_reload: Debouncer[Coroutine[Any, Any, None]] = Debouncer(
            hass,
            LOGGER,
            cooldown=ENTRY_RELOAD_COOLDOWN,
            immediate=False,
            function=self._async_reload_entry_async_reload_entry,
        )
 
    async def async_call(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
        """Start the countdown for a reload."""
        if (new_token := entry.data.get(CONF_TOKEN)) != self.tokentoken:
            LOGGER.debug("Skipping reload as its a token update")
            self.tokentoken = new_token
            return  # Token updates should not trigger a reload
        LOGGER.debug("Calling debouncer to get a reload after cooldown")
        await self._debounced_reload.async_call()
 
    @callback
    def async_shutdown(self) -> None:
        """Cancel any pending reload."""
        self._debounced_reload.async_shutdown()
 
    async def _async_reload_entry(self) -> None:
        """Reload entry."""
        LOGGER.debug("Reloading entry %s", self.entryentry.title)
        await self.hasshass.config_entries.async_reload(self.entryentry.entry_id)
 
 
async def _async_update_ssdp_locations(hass: HomeAssistant, entry: ConfigEntry) -> None:
    """Update ssdp locations from discovery cache."""
    updates = {}
    for ssdp_st, key in (
        (UPNP_SVC_RENDERING_CONTROL, CONF_SSDP_RENDERING_CONTROL_LOCATION),
        (UPNP_SVC_MAIN_TV_AGENT, CONF_SSDP_MAIN_TV_AGENT_LOCATION),
    ):
        for discovery_info in await ssdp.async_get_discovery_info_by_st(hass, ssdp_st):
            location = discovery_info.ssdp_location
            host = urlparse(location).hostname
            if host == entry.data[CONF_HOST]:
                updates[key] = location
                break
 
    if updates:
        hass.config_entries.async_update_entry(entry, data={**entry.data, **updates})
 
 
async def async_setup_entry(hass: HomeAssistant, entry: SamsungTVConfigEntry) -> bool:
    """Set up the Samsung TV platform."""
    # Initialize bridge
    if entry.data.get(CONF_METHOD) == METHOD_ENCRYPTED_WEBSOCKET:
        if not entry.data.get(CONF_TOKEN) or not entry.data.get(CONF_SESSION_ID):
            raise ConfigEntryAuthFailed(
                "Token and session id are required in encrypted mode"
            )
    bridge = await _async_create_bridge_with_updated_data(hass, entry)
 
    @callback
    def _access_denied() -> None:
        """Access denied callback."""
        LOGGER.debug("Access denied in getting remote object")
        entry.async_start_reauth(hass)
 
    bridge.register_reauth_callback(_access_denied)
 
    # Ensure updates get saved against the config_entry
    @callback
    def _update_config_entry(updates: Mapping[str, Any]) -> None:
        """Update config entry with the new token."""
        hass.config_entries.async_update_entry(entry, data={**entry.data, **updates})
 
    bridge.register_update_config_entry_callback(_update_config_entry)
 
    async def stop_bridge(event: Event | None = None) -> None:
        """Stop SamsungTV bridge connection."""
        LOGGER.debug("Stopping SamsungTVBridge %s", bridge.host)
        await bridge.async_close_remote()
 
    entry.async_on_unload(
        hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, stop_bridge)
    )
    entry.async_on_unload(stop_bridge)
 
    await _async_update_ssdp_locations(hass, entry)
 
    # We must not await after we setup the reload or there
    # will be a race where the config flow will see the entry
    # as not loaded and may reload it
    debounced_reloader = DebouncedEntryReloader(hass, entry)
    entry.async_on_unload(debounced_reloader.async_shutdown)
    entry.async_on_unload(entry.add_update_listener(debounced_reloader.async_call))
 
    coordinator = SamsungTVDataUpdateCoordinator(hass, bridge)
    await coordinator.async_config_entry_first_refresh()
    entry.runtime_data = coordinator
    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
 
    return True
 
 
async def _async_create_bridge_with_updated_data(
    hass: HomeAssistant, entry: ConfigEntry
) -> SamsungTVBridge:
    """Create a bridge object and update any missing data in the config entry."""
    updated_data: dict[str, str | int] = {}
    host: str = entry.data[CONF_HOST]
    port: int | None = entry.data.get(CONF_PORT)
    method: str | None = entry.data.get(CONF_METHOD)
    load_info_attempted = False
    info: dict[str, Any] | None = None
 
    if not port or not method:
        LOGGER.debug("Attempting to get port or method for %s", host)
        if method == METHOD_LEGACY:
            port = LEGACY_PORT
        else:
            # When we imported from yaml we didn't setup the method
            # because we didn't know it
            _result, port, method, info = await async_get_device_info(hass, host)
            load_info_attempted = True
            if not port or not method:
                raise ConfigEntryNotReady(
                    "Failed to determine connection method, make sure the device is on."
                )
 
        LOGGER.debug("Updated port to %s and method to %s for %s", port, method, host)
        updated_data[CONF_PORT] = port
        updated_data[CONF_METHOD] = method
 
    bridge = _async_get_device_bridge(hass, {**entry.data, **updated_data})
 
    mac: str | None = entry.data.get(CONF_MAC)
    model: str | None = entry.data.get(CONF_MODEL)
    mac_is_incorrectly_formatted = mac and dr.format_mac(mac) != mac
    if (
        not mac or not model or mac_is_incorrectly_formatted
    ) and not load_info_attempted:
        info = await bridge.async_device_info()
 
    if not mac or mac_is_incorrectly_formatted:
        LOGGER.debug("Attempting to get mac for %s", host)
        if info:
            mac = mac_from_device_info(info)
 
        if not mac:
            mac = await hass.async_add_executor_job(
                partial(getmac.get_mac_address, ip=host)
            )
 
        if mac and mac != "none":
            # Samsung sometimes returns a value of "none" for the mac address
            # this should be ignored
            LOGGER.debug("Updated mac to %s for %s", mac, host)
            updated_data[CONF_MAC] = dr.format_mac(mac)
        else:
            LOGGER.warning("Failed to get mac for %s", host)
 
    if not model:
        LOGGER.debug("Attempting to get model for %s", host)
        if info:
            model = info.get("device", {}).get("modelName")
            if model:
                LOGGER.debug("Updated model to %s for %s", model, host)
                updated_data[CONF_MODEL] = model
 
    if model_requires_encryption(model) and method != METHOD_ENCRYPTED_WEBSOCKET:
        LOGGER.debug(
            (
                "Detected model %s for %s. Some televisions from H and J series use "
                "an encrypted protocol but you are using %s which may not be supported"
            ),
            model,
            host,
            method,
        )
 
    if updated_data:
        data = {**entry.data, **updated_data}
        hass.config_entries.async_update_entry(entry, data=data)
 
    return bridge
 
 
async def async_unload_entry(hass: HomeAssistant, entry: SamsungTVConfigEntry) -> bool:
    """Unload a config entry."""
    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
 
 
async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
    """Migrate old entry."""
    version = config_entry.version
    minor_version = config_entry.minor_version
 
    LOGGER.debug("Migrating from version %s.%s", version, minor_version)
 
    # 1 -> 2: Unique ID format changed, so delete and re-import:
    if version == 1:
        dev_reg = dr.async_get(hass)
        dev_reg.async_clear_config_entry(config_entry.entry_id)
 
        en_reg = er.async_get(hass)
        en_reg.async_clear_config_entry(config_entry.entry_id)
 
        version = 2
        hass.config_entries.async_update_entry(config_entry, version=2)
 
    if version == 2:
        if minor_version < 2:
            # Cleanup invalid MAC addresses - see #103512
            # Reverted due to device registry collisions - see #119082 / #119249
 
            minor_version = 2
            hass.config_entries.async_update_entry(config_entry, minor_version=2)
 
    LOGGER.debug("Migration to version %s.%s successful", version, minor_version)
 
    return True