diff --git a/.coveragerc b/.coveragerc index da3b7b91ecee..1952297eb5fc 100644 --- a/.coveragerc +++ b/.coveragerc @@ -1651,6 +1651,7 @@ omit = homeassistant/components/xiaomi_miio/remote.py homeassistant/components/xiaomi_miio/sensor.py homeassistant/components/xiaomi_miio/switch.py + homeassistant/components/xiaomi_miio/typing.py homeassistant/components/xiaomi_tv/media_player.py homeassistant/components/xmpp/notify.py homeassistant/components/xs1/* diff --git a/homeassistant/components/bluesound/media_player.py b/homeassistant/components/bluesound/media_player.py index 7be5a823bf84..73ce963d481f 100644 --- a/homeassistant/components/bluesound/media_player.py +++ b/homeassistant/components/bluesound/media_player.py @@ -7,7 +7,7 @@ from asyncio import CancelledError, timeout from datetime import timedelta from http import HTTPStatus import logging -from typing import Any +from typing import Any, NamedTuple from urllib import parse import aiohttp @@ -85,15 +85,27 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend( } ) + +class ServiceMethodDetails(NamedTuple): + """Details for SERVICE_TO_METHOD mapping.""" + + method: str + schema: vol.Schema + + BS_SCHEMA = vol.Schema({vol.Optional(ATTR_ENTITY_ID): cv.entity_ids}) BS_JOIN_SCHEMA = BS_SCHEMA.extend({vol.Required(ATTR_MASTER): cv.entity_id}) SERVICE_TO_METHOD = { - SERVICE_JOIN: {"method": "async_join", "schema": BS_JOIN_SCHEMA}, - SERVICE_UNJOIN: {"method": "async_unjoin", "schema": BS_SCHEMA}, - SERVICE_SET_TIMER: {"method": "async_increase_timer", "schema": BS_SCHEMA}, - SERVICE_CLEAR_TIMER: {"method": "async_clear_timer", "schema": BS_SCHEMA}, + SERVICE_JOIN: ServiceMethodDetails(method="async_join", schema=BS_JOIN_SCHEMA), + SERVICE_UNJOIN: ServiceMethodDetails(method="async_unjoin", schema=BS_SCHEMA), + SERVICE_SET_TIMER: ServiceMethodDetails( + method="async_increase_timer", schema=BS_SCHEMA + ), + SERVICE_CLEAR_TIMER: ServiceMethodDetails( + method="async_clear_timer", schema=BS_SCHEMA + ), } @@ -188,12 +200,11 @@ async def async_setup_platform( target_players = hass.data[DATA_BLUESOUND] for player in target_players: - await getattr(player, method["method"])(**params) + await getattr(player, method.method)(**params) for service, method in SERVICE_TO_METHOD.items(): - schema = method["schema"] hass.services.async_register( - DOMAIN, service, async_service_handler, schema=schema + DOMAIN, service, async_service_handler, schema=method.schema ) diff --git a/homeassistant/components/webostv/__init__.py b/homeassistant/components/webostv/__init__.py index 479407c31992..36950b0e02a4 100644 --- a/homeassistant/components/webostv/__init__.py +++ b/homeassistant/components/webostv/__init__.py @@ -4,6 +4,7 @@ from __future__ import annotations from contextlib import suppress import logging +from typing import NamedTuple from aiowebostv import WebOsClient, WebOsTvPairError import voluptuous as vol @@ -43,6 +44,14 @@ CONFIG_SCHEMA = cv.removed(DOMAIN, raise_if_present=False) CALL_SCHEMA = vol.Schema({vol.Required(ATTR_ENTITY_ID): cv.comp_entity_ids}) + +class ServiceMethodDetails(NamedTuple): + """Details for SERVICE_TO_METHOD mapping.""" + + method: str + schema: vol.Schema + + BUTTON_SCHEMA = CALL_SCHEMA.extend({vol.Required(ATTR_BUTTON): cv.string}) COMMAND_SCHEMA = CALL_SCHEMA.extend( @@ -52,12 +61,14 @@ COMMAND_SCHEMA = CALL_SCHEMA.extend( SOUND_OUTPUT_SCHEMA = CALL_SCHEMA.extend({vol.Required(ATTR_SOUND_OUTPUT): cv.string}) SERVICE_TO_METHOD = { - SERVICE_BUTTON: {"method": "async_button", "schema": BUTTON_SCHEMA}, - SERVICE_COMMAND: {"method": "async_command", "schema": COMMAND_SCHEMA}, - SERVICE_SELECT_SOUND_OUTPUT: { - "method": "async_select_sound_output", - "schema": SOUND_OUTPUT_SCHEMA, - }, + SERVICE_BUTTON: ServiceMethodDetails(method="async_button", schema=BUTTON_SCHEMA), + SERVICE_COMMAND: ServiceMethodDetails( + method="async_command", schema=COMMAND_SCHEMA + ), + SERVICE_SELECT_SOUND_OUTPUT: ServiceMethodDetails( + method="async_select_sound_output", + schema=SOUND_OUTPUT_SCHEMA, + ), } _LOGGER = logging.getLogger(__name__) @@ -92,13 +103,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_service_handler(service: ServiceCall) -> None: method = SERVICE_TO_METHOD[service.service] data = service.data.copy() - data["method"] = method["method"] + data["method"] = method.method async_dispatcher_send(hass, DOMAIN, data) for service, method in SERVICE_TO_METHOD.items(): - schema = method["schema"] hass.services.async_register( - DOMAIN, service, async_service_handler, schema=schema + DOMAIN, service, async_service_handler, schema=method.schema ) hass.data[DOMAIN][DATA_CONFIG_ENTRY][entry.entry_id] = client diff --git a/homeassistant/components/xiaomi_miio/fan.py b/homeassistant/components/xiaomi_miio/fan.py index 75533513b5e8..4e0e271b0719 100644 --- a/homeassistant/components/xiaomi_miio/fan.py +++ b/homeassistant/components/xiaomi_miio/fan.py @@ -92,6 +92,7 @@ from .const import ( SERVICE_SET_EXTRA_FEATURES, ) from .device import XiaomiCoordinatedMiioEntity +from .typing import ServiceMethodDetails _LOGGER = logging.getLogger(__name__) @@ -182,11 +183,11 @@ SERVICE_SCHEMA_EXTRA_FEATURES = AIRPURIFIER_SERVICE_SCHEMA.extend( ) SERVICE_TO_METHOD = { - SERVICE_RESET_FILTER: {"method": "async_reset_filter"}, - SERVICE_SET_EXTRA_FEATURES: { - "method": "async_set_extra_features", - "schema": SERVICE_SCHEMA_EXTRA_FEATURES, - }, + SERVICE_RESET_FILTER: ServiceMethodDetails(method="async_reset_filter"), + SERVICE_SET_EXTRA_FEATURES: ServiceMethodDetails( + method="async_set_extra_features", + schema=SERVICE_SCHEMA_EXTRA_FEATURES, + ), } FAN_DIRECTIONS_MAP = { @@ -271,7 +272,7 @@ async def async_setup_entry( update_tasks = [] for entity in filtered_entities: - entity_method = getattr(entity, method["method"], None) + entity_method = getattr(entity, method.method, None) if not entity_method: continue await entity_method(**params) @@ -281,7 +282,7 @@ async def async_setup_entry( await asyncio.wait(update_tasks) for air_purifier_service, method in SERVICE_TO_METHOD.items(): - schema = method.get("schema", AIRPURIFIER_SERVICE_SCHEMA) + schema = method.schema or AIRPURIFIER_SERVICE_SCHEMA hass.services.async_register( DOMAIN, air_purifier_service, async_service_handler, schema=schema ) diff --git a/homeassistant/components/xiaomi_miio/light.py b/homeassistant/components/xiaomi_miio/light.py index 96f9595e0e8e..35537e82b2ef 100644 --- a/homeassistant/components/xiaomi_miio/light.py +++ b/homeassistant/components/xiaomi_miio/light.py @@ -68,6 +68,7 @@ from .const import ( ) from .device import XiaomiMiioEntity from .gateway import XiaomiGatewayDevice +from .typing import ServiceMethodDetails _LOGGER = logging.getLogger(__name__) @@ -108,20 +109,24 @@ SERVICE_SCHEMA_SET_DELAYED_TURN_OFF = XIAOMI_MIIO_SERVICE_SCHEMA.extend( ) SERVICE_TO_METHOD = { - SERVICE_SET_DELAYED_TURN_OFF: { - "method": "async_set_delayed_turn_off", - "schema": SERVICE_SCHEMA_SET_DELAYED_TURN_OFF, - }, - SERVICE_SET_SCENE: { - "method": "async_set_scene", - "schema": SERVICE_SCHEMA_SET_SCENE, - }, - SERVICE_REMINDER_ON: {"method": "async_reminder_on"}, - SERVICE_REMINDER_OFF: {"method": "async_reminder_off"}, - SERVICE_NIGHT_LIGHT_MODE_ON: {"method": "async_night_light_mode_on"}, - SERVICE_NIGHT_LIGHT_MODE_OFF: {"method": "async_night_light_mode_off"}, - SERVICE_EYECARE_MODE_ON: {"method": "async_eyecare_mode_on"}, - SERVICE_EYECARE_MODE_OFF: {"method": "async_eyecare_mode_off"}, + SERVICE_SET_DELAYED_TURN_OFF: ServiceMethodDetails( + method="async_set_delayed_turn_off", + schema=SERVICE_SCHEMA_SET_DELAYED_TURN_OFF, + ), + SERVICE_SET_SCENE: ServiceMethodDetails( + method="async_set_scene", + schema=SERVICE_SCHEMA_SET_SCENE, + ), + SERVICE_REMINDER_ON: ServiceMethodDetails(method="async_reminder_on"), + SERVICE_REMINDER_OFF: ServiceMethodDetails(method="async_reminder_off"), + SERVICE_NIGHT_LIGHT_MODE_ON: ServiceMethodDetails( + method="async_night_light_mode_on" + ), + SERVICE_NIGHT_LIGHT_MODE_OFF: ServiceMethodDetails( + method="async_night_light_mode_off" + ), + SERVICE_EYECARE_MODE_ON: ServiceMethodDetails(method="async_eyecare_mode_on"), + SERVICE_EYECARE_MODE_OFF: ServiceMethodDetails(method="async_eyecare_mode_off"), } @@ -232,9 +237,9 @@ async def async_setup_entry( update_tasks = [] for target_device in target_devices: - if not hasattr(target_device, method["method"]): + if not hasattr(target_device, method.method): continue - await getattr(target_device, method["method"])(**params) + await getattr(target_device, method.method)(**params) update_tasks.append( asyncio.create_task(target_device.async_update_ha_state(True)) ) @@ -243,7 +248,7 @@ async def async_setup_entry( await asyncio.wait(update_tasks) for xiaomi_miio_service, method in SERVICE_TO_METHOD.items(): - schema = method.get("schema", XIAOMI_MIIO_SERVICE_SCHEMA) + schema = method.schema or XIAOMI_MIIO_SERVICE_SCHEMA hass.services.async_register( DOMAIN, xiaomi_miio_service, async_service_handler, schema=schema ) diff --git a/homeassistant/components/xiaomi_miio/switch.py b/homeassistant/components/xiaomi_miio/switch.py index 34ebb9addf51..797a98d9fa1c 100644 --- a/homeassistant/components/xiaomi_miio/switch.py +++ b/homeassistant/components/xiaomi_miio/switch.py @@ -115,6 +115,7 @@ from .const import ( ) from .device import XiaomiCoordinatedMiioEntity, XiaomiMiioEntity from .gateway import XiaomiGatewayDevice +from .typing import ServiceMethodDetails _LOGGER = logging.getLogger(__name__) @@ -176,16 +177,16 @@ SERVICE_SCHEMA_POWER_PRICE = SERVICE_SCHEMA.extend( ) SERVICE_TO_METHOD = { - SERVICE_SET_WIFI_LED_ON: {"method": "async_set_wifi_led_on"}, - SERVICE_SET_WIFI_LED_OFF: {"method": "async_set_wifi_led_off"}, - SERVICE_SET_POWER_MODE: { - "method": "async_set_power_mode", - "schema": SERVICE_SCHEMA_POWER_MODE, - }, - SERVICE_SET_POWER_PRICE: { - "method": "async_set_power_price", - "schema": SERVICE_SCHEMA_POWER_PRICE, - }, + SERVICE_SET_WIFI_LED_ON: ServiceMethodDetails(method="async_set_wifi_led_on"), + SERVICE_SET_WIFI_LED_OFF: ServiceMethodDetails(method="async_set_wifi_led_off"), + SERVICE_SET_POWER_MODE: ServiceMethodDetails( + method="async_set_power_mode", + schema=SERVICE_SCHEMA_POWER_MODE, + ), + SERVICE_SET_POWER_PRICE: ServiceMethodDetails( + method="async_set_power_price", + schema=SERVICE_SCHEMA_POWER_PRICE, + ), } MODEL_TO_FEATURES_MAP = { @@ -488,9 +489,9 @@ async def async_setup_other_entry(hass, config_entry, async_add_entities): update_tasks = [] for device in devices: - if not hasattr(device, method["method"]): + if not hasattr(device, method.method): continue - await getattr(device, method["method"])(**params) + await getattr(device, method.method)(**params) update_tasks.append( asyncio.create_task(device.async_update_ha_state(True)) ) @@ -499,7 +500,7 @@ async def async_setup_other_entry(hass, config_entry, async_add_entities): await asyncio.wait(update_tasks) for plug_service, method in SERVICE_TO_METHOD.items(): - schema = method.get("schema", SERVICE_SCHEMA) + schema = method.schema or SERVICE_SCHEMA hass.services.async_register( DOMAIN, plug_service, async_service_handler, schema=schema ) diff --git a/homeassistant/components/xiaomi_miio/typing.py b/homeassistant/components/xiaomi_miio/typing.py new file mode 100644 index 000000000000..8fbb8e3d83f5 --- /dev/null +++ b/homeassistant/components/xiaomi_miio/typing.py @@ -0,0 +1,12 @@ +"""Typings for the xiaomi_miio integration.""" + +from typing import NamedTuple + +import voluptuous as vol + + +class ServiceMethodDetails(NamedTuple): + """Details for SERVICE_TO_METHOD mapping.""" + + method: str + schema: vol.Schema | None = None