config_flow.py

Go to the documentation of this file.

"""Config flow for DLNA DMS."""
 
from __future__ import annotations
 
import logging
from pprint import pformat
from typing import TYPE_CHECKING, Any, cast
from urllib.parse import urlparse
 
from async_upnp_client.profiles.dlna import DmsDevice
import voluptuous as vol
 
from homeassistant.components import ssdp
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_DEVICE_ID, CONF_HOST, CONF_URL
from homeassistant.data_entry_flow import AbortFlow
 
from .const import CONF_SOURCE_ID, CONFIG_VERSION, DEFAULT_NAME, DOMAIN
from .util import generate_source_id
 
LOGGER = logging.getLogger(__name__)
 
 
class DlnaDmsFlowHandler(ConfigFlow, domain=DOMAIN):
    """Handle a DLNA DMS config flow.
 
    The Unique Service Name (USN) of the DMS device is used as the unique_id for
    config entries and for entities. This USN may differ from the root USN if
    the DMS is an embedded device.
    """
 
    VERSION = CONFIG_VERSION
 
    def __init__(self) -> None:
        """Initialize flow."""
        self._discoveries_discoveries: dict[str, ssdp.SsdpServiceInfo] = {}
        self._location_location: str | None = None
        self._usn_usn: str | None = None
        self._name_name: str | None = None
 
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle a flow initialized by the user by listing unconfigured devices."""
        LOGGER.debug("async_step_user: user_input: %s", user_input)
 
        if user_input is not None and (host := user_input.get(CONF_HOST)):
            # User has chosen a device
            discovery = self._discoveries_discoveries[host]
            await self._async_parse_discovery_async_parse_discovery(discovery, raise_on_progress=False)
            return self._create_entry_create_entry()
 
        if not (discoveries := await self._async_get_discoveries_async_get_discoveries()):
            # Nothing found, abort configuration
            return self.async_abortasync_abortasync_abort(reason="no_devices_found")
 
        self._discoveries_discoveries = {
            cast(str, urlparse(discovery.ssdp_location).hostname): discovery
            for discovery in discoveries
        }
 
        discovery_choices = {
            host: f"{discovery.upnp.get(ssdp.ATTR_UPNP_FRIENDLY_NAME)} ({host})"
            for host, discovery in self._discoveries_discoveries.items()
        }
        data_schema = vol.Schema({vol.Optional(CONF_HOST): vol.In(discovery_choices)})
        return self.async_show_formasync_show_formasync_show_form(step_id="user", data_schema=data_schema)
 
    async def async_step_ssdp(
        self, discovery_info: ssdp.SsdpServiceInfo
    ) -> ConfigFlowResult:
        """Handle a flow initialized by SSDP discovery."""
        if LOGGER.isEnabledFor(logging.DEBUG):
            LOGGER.debug("async_step_ssdp: discovery_info %s", pformat(discovery_info))
 
        await self._async_parse_discovery_async_parse_discovery(discovery_info)
        if TYPE_CHECKING:
            # _async_parse_discovery unconditionally sets self._name
            assert self._name_name is not None
 
        # Abort if the device doesn't support all services required for a DmsDevice.
        # Use the discovery_info instead of DmsDevice.is_profile_device to avoid
        # contacting the device again.
        discovery_service_list = discovery_info.upnp.get(ssdp.ATTR_UPNP_SERVICE_LIST)
        if not discovery_service_list:
            return self.async_abortasync_abortasync_abort(reason="not_dms")
 
        services = discovery_service_list.get("service")
        if not services:
            discovery_service_ids: set[str] = set()
        elif isinstance(services, list):
            discovery_service_ids = {service.get("serviceId") for service in services}
        else:
            # Only one service defined (etree_to_dict failed to make a list)
            discovery_service_ids = {services.get("serviceId")}
 
        if not DmsDevice.SERVICE_IDS.issubset(discovery_service_ids):
            return self.async_abortasync_abortasync_abort(reason="not_dms")
 
        # Abort if another config entry has the same location, in case the
        # device doesn't have a static and unique UDN (breaking the UPnP spec).
        self._async_abort_entries_match_async_abort_entries_match({CONF_URL: self._location_location})
 
        self.context["title_placeholders"] = {"name": self._name_name}
 
        return await self.async_step_confirmasync_step_confirm()
 
    async def async_step_confirm(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Allow the user to confirm adding the device."""
        if user_input is not None:
            return self._create_entry_create_entry()
 
        self._set_confirm_only_set_confirm_only()
        return self.async_show_formasync_show_formasync_show_form(step_id="confirm")
 
    def _create_entry(self) -> ConfigFlowResult:
        """Create a config entry, assuming all required information is now known."""
        LOGGER.debug(
            "_create_entry: name: %s, location: %s, USN: %s",
            self._name_name,
            self._location_location,
            self._usn_usn,
        )
        assert self._name_name
        assert self._location_location
        assert self._usn_usn
 
        data = {
            CONF_URL: self._location_location,
            CONF_DEVICE_ID: self._usn_usn,
            CONF_SOURCE_ID: generate_source_id(self.hass, self._name_name),
        }
        return self.async_create_entryasync_create_entryasync_create_entry(title=self._name_name, data=data)
 
    async def _async_parse_discovery(
        self, discovery_info: ssdp.SsdpServiceInfo, raise_on_progress: bool = True
    ) -> None:
        """Get required details from an SSDP discovery.
 
        Aborts if a device matching the SSDP USN has already been configured.
        """
        LOGGER.debug(
            "_async_parse_discovery: location: %s, USN: %s",
            discovery_info.ssdp_location,
            discovery_info.ssdp_usn,
        )
 
        if not discovery_info.ssdp_location or not discovery_info.ssdp_usn:
            raise AbortFlow("bad_ssdp")
 
        if not self._location_location:
            self._location_location = discovery_info.ssdp_location
 
        self._usn_usn = discovery_info.ssdp_usn
        await self.async_set_unique_idasync_set_unique_id(self._usn_usn, raise_on_progress=raise_on_progress)
 
        # Abort if already configured, but update the last-known location
        self._abort_if_unique_id_configured_abort_if_unique_id_configured(
            updates={CONF_URL: self._location_location}, reload_on_update=False
        )
 
        self._name_name = (
            discovery_info.upnp.get(ssdp.ATTR_UPNP_FRIENDLY_NAME)
            or urlparse(self._location_location).hostname
            or DEFAULT_NAME
        )
 
    async def _async_get_discoveries(self) -> list[ssdp.SsdpServiceInfo]:
        """Get list of unconfigured DLNA devices discovered by SSDP."""
        # Get all compatible devices from ssdp's cache
        discoveries: list[ssdp.SsdpServiceInfo] = []
        for udn_st in DmsDevice.DEVICE_TYPES:
            st_discoveries = await ssdp.async_get_discovery_info_by_st(
                self.hass, udn_st
            )
            discoveries.extend(st_discoveries)
 
        # Filter out devices already configured
        current_unique_ids = {
            entry.unique_id
            for entry in self._async_current_entries_async_current_entries(include_ignore=False)
        }
        return [disc for disc in discoveries if disc.ssdp_udn not in current_unique_ids]