jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 13:17:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use new ServiceInfo location in dlna_dmr (#135691)

Browse Source
This commit is contained in:
epenet 2025-01-16 13:41:09 +01:00 committed by GitHub
parent 27c2f2333e
commit 476935050a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 21 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
homeassistant/components/dlna_dmr/config_flow.py
View File

@ -27,6 +27,14 @@ from homeassistant.const import CONF_DEVICE_ID, CONF_HOST, CONF_MAC, CONF_TYPE,
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import IntegrationError from homeassistant.exceptions import IntegrationError
from homeassistant.helpers import config_validation as cv, device_registry as dr from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_DEVICE_TYPE,
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_MANUFACTURER,
ATTR_UPNP_MODEL_NAME,
ATTR_UPNP_SERVICE_LIST,
SsdpServiceInfo,
)
from homeassistant.helpers.typing import VolDictType from homeassistant.helpers.typing import VolDictType
from .const import ( from .const import (
@ -60,7 +68,7 @@ class DlnaDmrFlowHandler(ConfigFlow, domain=DOMAIN):
def __init__(self) -> None: def __init__(self) -> None:
"""Initialize flow.""" """Initialize flow."""
self._discoveries: dict[str, ssdp.SsdpServiceInfo] = {} self._discoveries: dict[str, SsdpServiceInfo] = {}
self._location: str | None = None self._location: str | None = None
self._udn: str | None = None self._udn: str | None = None
self._device_type: str | None = None self._device_type: str | None = None
@ -98,7 +106,7 @@ class DlnaDmrFlowHandler(ConfigFlow, domain=DOMAIN):
return await self.async_step_manual() return await self.async_step_manual()
self._discoveries = { self._discoveries = {
discovery.upnp.get(ssdp.ATTR_UPNP_FRIENDLY_NAME) discovery.upnp.get(ATTR_UPNP_FRIENDLY_NAME)
or cast(str, urlparse(discovery.ssdp_location).hostname): discovery or cast(str, urlparse(discovery.ssdp_location).hostname): discovery
for discovery in discoveries for discovery in discoveries
} }
@ -131,7 +139,7 @@ class DlnaDmrFlowHandler(ConfigFlow, domain=DOMAIN):
) )
async def async_step_ssdp( async def async_step_ssdp(
self, discovery_info: ssdp.SsdpServiceInfo self, discovery_info: SsdpServiceInfo
) -> ConfigFlowResult: ) -> ConfigFlowResult:
"""Handle a flow initialized by SSDP discovery.""" """Handle a flow initialized by SSDP discovery."""
if LOGGER.isEnabledFor(logging.DEBUG): if LOGGER.isEnabledFor(logging.DEBUG):
@ -267,7 +275,7 @@ class DlnaDmrFlowHandler(ConfigFlow, domain=DOMAIN):
return self.async_create_entry(title=title, data=data, options=self._options) return self.async_create_entry(title=title, data=data, options=self._options)
async def _async_set_info_from_discovery( async def _async_set_info_from_discovery(
self, discovery_info: ssdp.SsdpServiceInfo, abort_if_configured: bool = True self, discovery_info: SsdpServiceInfo, abort_if_configured: bool = True
) -> None: ) -> None:
"""Set information required for a config entry from the SSDP discovery.""" """Set information required for a config entry from the SSDP discovery."""
LOGGER.debug( LOGGER.debug(
@ -285,7 +293,7 @@ class DlnaDmrFlowHandler(ConfigFlow, domain=DOMAIN):
self._device_type = discovery_info.ssdp_nt or discovery_info.ssdp_st self._device_type = discovery_info.ssdp_nt or discovery_info.ssdp_st
self._name = ( self._name = (
discovery_info.upnp.get(ssdp.ATTR_UPNP_FRIENDLY_NAME) discovery_info.upnp.get(ATTR_UPNP_FRIENDLY_NAME)
or urlparse(self._location).hostname or urlparse(self._location).hostname
or DEFAULT_NAME or DEFAULT_NAME
) )
@ -301,12 +309,12 @@ class DlnaDmrFlowHandler(ConfigFlow, domain=DOMAIN):
updates[CONF_MAC] = self._mac updates[CONF_MAC] = self._mac
self._abort_if_unique_id_configured(updates=updates, reload_on_update=False) self._abort_if_unique_id_configured(updates=updates, reload_on_update=False)
async def _async_get_discoveries(self) -> list[ssdp.SsdpServiceInfo]: async def _async_get_discoveries(self) -> list[SsdpServiceInfo]:
"""Get list of unconfigured DLNA devices discovered by SSDP.""" """Get list of unconfigured DLNA devices discovered by SSDP."""
LOGGER.debug("_get_discoveries") LOGGER.debug("_get_discoveries")
# Get all compatible devices from ssdp's cache # Get all compatible devices from ssdp's cache
discoveries: list[ssdp.SsdpServiceInfo] = [] discoveries: list[SsdpServiceInfo] = []
for udn_st in DmrDevice.DEVICE_TYPES: for udn_st in DmrDevice.DEVICE_TYPES:
st_discoveries = await ssdp.async_get_discovery_info_by_st( st_discoveries = await ssdp.async_get_discovery_info_by_st(
self.hass, udn_st self.hass, udn_st
@ -386,7 +394,7 @@ class DlnaDmrOptionsFlowHandler(OptionsFlow):
) )
def _is_ignored_device(discovery_info: ssdp.SsdpServiceInfo) -> bool: def _is_ignored_device(discovery_info: SsdpServiceInfo) -> bool:
"""Return True if this device should be ignored for discovery. """Return True if this device should be ignored for discovery.
These devices are supported better by other integrations, so don't bug These devices are supported better by other integrations, so don't bug
@ -402,17 +410,14 @@ def _is_ignored_device(discovery_info: ssdp.SsdpServiceInfo) -> bool:
return True return True
# Is the root device not a DMR? # Is the root device not a DMR?
if ( if discovery_info.upnp.get(ATTR_UPNP_DEVICE_TYPE) not in DmrDevice.DEVICE_TYPES:
discovery_info.upnp.get(ssdp.ATTR_UPNP_DEVICE_TYPE)
not in DmrDevice.DEVICE_TYPES
):
return True return True
# Special cases for devices with other discovery methods (e.g. mDNS), or # Special cases for devices with other discovery methods (e.g. mDNS), or
# that advertise multiple unrelated (sent in separate discovery packets) # that advertise multiple unrelated (sent in separate discovery packets)
# UPnP devices. # UPnP devices.
manufacturer = (discovery_info.upnp.get(ssdp.ATTR_UPNP_MANUFACTURER) or "").lower() manufacturer = (discovery_info.upnp.get(ATTR_UPNP_MANUFACTURER) or "").lower()
model = (discovery_info.upnp.get(ssdp.ATTR_UPNP_MODEL_NAME) or "").lower() model = (discovery_info.upnp.get(ATTR_UPNP_MODEL_NAME) or "").lower()
if manufacturer.startswith("xbmc") or model == "kodi": if manufacturer.startswith("xbmc") or model == "kodi":
# kodi # kodi
@ -432,14 +437,14 @@ def _is_ignored_device(discovery_info: ssdp.SsdpServiceInfo) -> bool:
return False return False
def _is_dmr_device(discovery_info: ssdp.SsdpServiceInfo) -> bool: def _is_dmr_device(discovery_info: SsdpServiceInfo) -> bool:
"""Determine if discovery is a complete DLNA DMR device. """Determine if discovery is a complete DLNA DMR device.
Use the discovery_info instead of DmrDevice.is_profile_device to avoid Use the discovery_info instead of DmrDevice.is_profile_device to avoid
contacting the device again. contacting the device again.
""" """
# Abort if the device doesn't support all services required for a DmrDevice. # Abort if the device doesn't support all services required for a DmrDevice.
discovery_service_list = discovery_info.upnp.get(ssdp.ATTR_UPNP_SERVICE_LIST) discovery_service_list = discovery_info.upnp.get(ATTR_UPNP_SERVICE_LIST)
if not discovery_service_list: if not discovery_service_list:
return False return False