228 lines
7.2 KiB
Python
228 lines
7.2 KiB
Python
"""The SSDP integration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable, Coroutine
|
|
from functools import partial
|
|
from typing import Any
|
|
|
|
from homeassistant.core import HassJob, HomeAssistant
|
|
from homeassistant.helpers import config_validation as cv
|
|
from homeassistant.helpers.deprecation import (
|
|
DeprecatedConstant,
|
|
all_with_deprecated_constants,
|
|
check_if_deprecated_constant,
|
|
dir_with_deprecated_constants,
|
|
)
|
|
from homeassistant.helpers.service_info.ssdp import (
|
|
ATTR_NT as _ATTR_NT,
|
|
ATTR_ST as _ATTR_ST,
|
|
ATTR_UPNP_DEVICE_TYPE as _ATTR_UPNP_DEVICE_TYPE,
|
|
ATTR_UPNP_FRIENDLY_NAME as _ATTR_UPNP_FRIENDLY_NAME,
|
|
ATTR_UPNP_MANUFACTURER as _ATTR_UPNP_MANUFACTURER,
|
|
ATTR_UPNP_MANUFACTURER_URL as _ATTR_UPNP_MANUFACTURER_URL,
|
|
ATTR_UPNP_MODEL_DESCRIPTION as _ATTR_UPNP_MODEL_DESCRIPTION,
|
|
ATTR_UPNP_MODEL_NAME as _ATTR_UPNP_MODEL_NAME,
|
|
ATTR_UPNP_MODEL_NUMBER as _ATTR_UPNP_MODEL_NUMBER,
|
|
ATTR_UPNP_MODEL_URL as _ATTR_UPNP_MODEL_URL,
|
|
ATTR_UPNP_PRESENTATION_URL as _ATTR_UPNP_PRESENTATION_URL,
|
|
ATTR_UPNP_SERIAL as _ATTR_UPNP_SERIAL,
|
|
ATTR_UPNP_SERVICE_LIST as _ATTR_UPNP_SERVICE_LIST,
|
|
ATTR_UPNP_UDN as _ATTR_UPNP_UDN,
|
|
ATTR_UPNP_UPC as _ATTR_UPNP_UPC,
|
|
SsdpServiceInfo as _SsdpServiceInfo,
|
|
)
|
|
from homeassistant.helpers.typing import ConfigType
|
|
from homeassistant.loader import async_get_ssdp, bind_hass
|
|
from homeassistant.util.logging import catch_log_exception
|
|
|
|
from . import websocket_api
|
|
from .const import DOMAIN, SSDP_SCANNER, UPNP_SERVER
|
|
from .scanner import (
|
|
IntegrationMatchers,
|
|
Scanner,
|
|
SsdpChange,
|
|
SsdpHassJobCallback, # noqa: F401
|
|
)
|
|
from .server import Server
|
|
|
|
# Attributes for accessing info from SSDP response
|
|
ATTR_SSDP_LOCATION = "ssdp_location"
|
|
ATTR_SSDP_ST = "ssdp_st"
|
|
ATTR_SSDP_NT = "ssdp_nt"
|
|
ATTR_SSDP_UDN = "ssdp_udn"
|
|
ATTR_SSDP_USN = "ssdp_usn"
|
|
ATTR_SSDP_EXT = "ssdp_ext"
|
|
ATTR_SSDP_SERVER = "ssdp_server"
|
|
ATTR_SSDP_BOOTID = "BOOTID.UPNP.ORG"
|
|
ATTR_SSDP_NEXTBOOTID = "NEXTBOOTID.UPNP.ORG"
|
|
# Attributes for accessing info from retrieved UPnP device description
|
|
_DEPRECATED_ATTR_ST = DeprecatedConstant(
|
|
_ATTR_ST,
|
|
"homeassistant.helpers.service_info.ssdp.ATTR_ST",
|
|
"2026.2",
|
|
)
|
|
_DEPRECATED_ATTR_NT = DeprecatedConstant(
|
|
_ATTR_NT,
|
|
"homeassistant.helpers.service_info.ssdp.ATTR_NT",
|
|
"2026.2",
|
|
)
|
|
_DEPRECATED_ATTR_UPNP_DEVICE_TYPE = DeprecatedConstant(
|
|
_ATTR_UPNP_DEVICE_TYPE,
|
|
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_DEVICE_TYPE",
|
|
"2026.2",
|
|
)
|
|
_DEPRECATED_ATTR_UPNP_FRIENDLY_NAME = DeprecatedConstant(
|
|
_ATTR_UPNP_FRIENDLY_NAME,
|
|
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_FRIENDLY_NAME",
|
|
"2026.2",
|
|
)
|
|
_DEPRECATED_ATTR_UPNP_MANUFACTURER = DeprecatedConstant(
|
|
_ATTR_UPNP_MANUFACTURER,
|
|
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_MANUFACTURER",
|
|
"2026.2",
|
|
)
|
|
_DEPRECATED_ATTR_UPNP_MANUFACTURER_URL = DeprecatedConstant(
|
|
_ATTR_UPNP_MANUFACTURER_URL,
|
|
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_MANUFACTURER_URL",
|
|
"2026.2",
|
|
)
|
|
_DEPRECATED_ATTR_UPNP_MODEL_DESCRIPTION = DeprecatedConstant(
|
|
_ATTR_UPNP_MODEL_DESCRIPTION,
|
|
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_MODEL_DESCRIPTION",
|
|
"2026.2",
|
|
)
|
|
_DEPRECATED_ATTR_UPNP_MODEL_NAME = DeprecatedConstant(
|
|
_ATTR_UPNP_MODEL_NAME,
|
|
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_MODEL_NAME",
|
|
"2026.2",
|
|
)
|
|
_DEPRECATED_ATTR_UPNP_MODEL_NUMBER = DeprecatedConstant(
|
|
_ATTR_UPNP_MODEL_NUMBER,
|
|
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_MODEL_NUMBER",
|
|
"2026.2",
|
|
)
|
|
_DEPRECATED_ATTR_UPNP_MODEL_URL = DeprecatedConstant(
|
|
_ATTR_UPNP_MODEL_URL,
|
|
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_MODEL_URL",
|
|
"2026.2",
|
|
)
|
|
_DEPRECATED_ATTR_UPNP_SERIAL = DeprecatedConstant(
|
|
_ATTR_UPNP_SERIAL,
|
|
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_SERIAL",
|
|
"2026.2",
|
|
)
|
|
_DEPRECATED_ATTR_UPNP_SERVICE_LIST = DeprecatedConstant(
|
|
_ATTR_UPNP_SERVICE_LIST,
|
|
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_SERVICE_LIST",
|
|
"2026.2",
|
|
)
|
|
_DEPRECATED_ATTR_UPNP_UDN = DeprecatedConstant(
|
|
_ATTR_UPNP_UDN,
|
|
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_UDN",
|
|
"2026.2",
|
|
)
|
|
_DEPRECATED_ATTR_UPNP_UPC = DeprecatedConstant(
|
|
_ATTR_UPNP_UPC,
|
|
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_UPC",
|
|
"2026.2",
|
|
)
|
|
_DEPRECATED_ATTR_UPNP_PRESENTATION_URL = DeprecatedConstant(
|
|
_ATTR_UPNP_PRESENTATION_URL,
|
|
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_PRESENTATION_URL",
|
|
"2026.2",
|
|
)
|
|
# Attributes for accessing info added by Home Assistant
|
|
ATTR_HA_MATCHING_DOMAINS = "x_homeassistant_matching_domains"
|
|
|
|
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
|
|
|
|
_DEPRECATED_SsdpServiceInfo = DeprecatedConstant(
|
|
_SsdpServiceInfo,
|
|
"homeassistant.helpers.service_info.ssdp.SsdpServiceInfo",
|
|
"2026.2",
|
|
)
|
|
|
|
|
|
def _format_err(name: str, *args: Any) -> str:
|
|
"""Format error message."""
|
|
return f"Exception in SSDP callback {name}: {args}"
|
|
|
|
|
|
@bind_hass
|
|
async def async_register_callback(
|
|
hass: HomeAssistant,
|
|
callback: Callable[
|
|
[_SsdpServiceInfo, SsdpChange], Coroutine[Any, Any, None] | None
|
|
],
|
|
match_dict: dict[str, str] | None = None,
|
|
) -> Callable[[], None]:
|
|
"""Register to receive a callback on ssdp broadcast.
|
|
|
|
Returns a callback that can be used to cancel the registration.
|
|
"""
|
|
scanner: Scanner = hass.data[DOMAIN][SSDP_SCANNER]
|
|
job = HassJob(
|
|
catch_log_exception(
|
|
callback,
|
|
partial(_format_err, str(callback)),
|
|
),
|
|
f"ssdp callback {match_dict}",
|
|
)
|
|
return await scanner.async_register_callback(job, match_dict)
|
|
|
|
|
|
@bind_hass
|
|
async def async_get_discovery_info_by_udn_st(
|
|
hass: HomeAssistant, udn: str, st: str
|
|
) -> _SsdpServiceInfo | None:
|
|
"""Fetch the discovery info cache."""
|
|
scanner: Scanner = hass.data[DOMAIN][SSDP_SCANNER]
|
|
return await scanner.async_get_discovery_info_by_udn_st(udn, st)
|
|
|
|
|
|
@bind_hass
|
|
async def async_get_discovery_info_by_st(
|
|
hass: HomeAssistant, st: str
|
|
) -> list[_SsdpServiceInfo]:
|
|
"""Fetch all the entries matching the st."""
|
|
scanner: Scanner = hass.data[DOMAIN][SSDP_SCANNER]
|
|
return await scanner.async_get_discovery_info_by_st(st)
|
|
|
|
|
|
@bind_hass
|
|
async def async_get_discovery_info_by_udn(
|
|
hass: HomeAssistant, udn: str
|
|
) -> list[_SsdpServiceInfo]:
|
|
"""Fetch all the entries matching the udn."""
|
|
scanner: Scanner = hass.data[DOMAIN][SSDP_SCANNER]
|
|
return await scanner.async_get_discovery_info_by_udn(udn)
|
|
|
|
|
|
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
|
"""Set up the SSDP integration."""
|
|
|
|
integration_matchers = IntegrationMatchers()
|
|
integration_matchers.async_setup(await async_get_ssdp(hass))
|
|
|
|
scanner = Scanner(hass, integration_matchers)
|
|
server = Server(hass)
|
|
hass.data[DOMAIN] = {
|
|
SSDP_SCANNER: scanner,
|
|
UPNP_SERVER: server,
|
|
}
|
|
|
|
await scanner.async_start()
|
|
await server.async_start()
|
|
websocket_api.async_setup(hass)
|
|
|
|
return True
|
|
|
|
|
|
# These can be removed if no deprecated constant are in this module anymore
|
|
__getattr__ = partial(check_if_deprecated_constant, module_globals=globals())
|
|
__dir__ = partial(
|
|
dir_with_deprecated_constants, module_globals_keys=[*globals().keys()]
|
|
)
|
|
__all__ = all_with_deprecated_constants(globals())
|