Use NamedTuple for discovery service details (#56751)

This commit is contained in:
Marc Mueller 2021-09-29 01:59:40 +02:00 committed by GitHub
parent a91fbec198
commit 15aafc8db6
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23

View file

@ -4,6 +4,7 @@ from __future__ import annotations
from datetime import timedelta from datetime import timedelta
import json import json
import logging import logging
from typing import NamedTuple
from netdisco.discovery import NetworkDiscovery from netdisco.discovery import NetworkDiscovery
import voluptuous as vol import voluptuous as vol
@ -46,16 +47,24 @@ CONFIG_ENTRY_HANDLERS = {
"logitech_mediaserver": "squeezebox", "logitech_mediaserver": "squeezebox",
} }
class ServiceDetails(NamedTuple):
"""Store service details."""
component: str
platform: str | None
# These have no config flows # These have no config flows
SERVICE_HANDLERS = { SERVICE_HANDLERS = {
SERVICE_ENIGMA2: ("media_player", "enigma2"), SERVICE_ENIGMA2: ServiceDetails("media_player", "enigma2"),
SERVICE_SABNZBD: ("sabnzbd", None), SERVICE_SABNZBD: ServiceDetails("sabnzbd", None),
"yamaha": ("media_player", "yamaha"), "yamaha": ServiceDetails("media_player", "yamaha"),
"frontier_silicon": ("media_player", "frontier_silicon"), "frontier_silicon": ServiceDetails("media_player", "frontier_silicon"),
"openhome": ("media_player", "openhome"), "openhome": ServiceDetails("media_player", "openhome"),
"bose_soundtouch": ("media_player", "soundtouch"), "bose_soundtouch": ServiceDetails("media_player", "soundtouch"),
"bluesound": ("media_player", "bluesound"), "bluesound": ServiceDetails("media_player", "bluesound"),
"lg_smart_device": ("media_player", "lg_soundbar"), "lg_smart_device": ServiceDetails("media_player", "lg_soundbar"),
} }
OPTIONAL_SERVICE_HANDLERS: dict[str, tuple[str, str | None]] = {} OPTIONAL_SERVICE_HANDLERS: dict[str, tuple[str, str | None]] = {}
@ -172,24 +181,24 @@ async def async_setup(hass, config):
) )
return return
comp_plat = SERVICE_HANDLERS.get(service) service_details = SERVICE_HANDLERS.get(service)
if not comp_plat and service in enabled_platforms: if not service_details and service in enabled_platforms:
comp_plat = OPTIONAL_SERVICE_HANDLERS[service] service_details = OPTIONAL_SERVICE_HANDLERS[service]
# We do not know how to handle this service. # We do not know how to handle this service.
if not comp_plat: if not service_details:
logger.debug("Unknown service discovered: %s %s", service, info) logger.debug("Unknown service discovered: %s %s", service, info)
return return
logger.info("Found new service: %s %s", service, info) logger.info("Found new service: %s %s", service, info)
component, platform = comp_plat if service_details.platform is None:
await async_discover(hass, service, info, service_details.component, config)
if platform is None:
await async_discover(hass, service, info, component, config)
else: else:
await async_load_platform(hass, component, platform, info, config) await async_load_platform(
hass, service_details.component, service_details.platform, info, config
)
async def scan_devices(now): async def scan_devices(now):
"""Scan for devices.""" """Scan for devices."""