Improve SERVICE_TO_METHOD typing (#120474)

.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/*

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_JOIN: ServiceMethodDetails(method="async_join", schema=BS_JOIN_SCHEMA),
-    SERVICE_UNJOIN: {"method": "async_unjoin", "schema": BS_SCHEMA},
+    SERVICE_UNJOIN: ServiceMethodDetails(method="async_unjoin", schema=BS_SCHEMA),
-    SERVICE_SET_TIMER: {"method": "async_increase_timer", "schema": BS_SCHEMA},
+    SERVICE_SET_TIMER: ServiceMethodDetails(
-    SERVICE_CLEAR_TIMER: {"method": "async_clear_timer", "schema": BS_SCHEMA},
+        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
         )
 
 

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_BUTTON: ServiceMethodDetails(method="async_button", schema=BUTTON_SCHEMA),
-    SERVICE_COMMAND: {"method": "async_command", "schema": COMMAND_SCHEMA},
+    SERVICE_COMMAND: ServiceMethodDetails(
-    SERVICE_SELECT_SOUND_OUTPUT: {
+        method="async_command", schema=COMMAND_SCHEMA
-        "method": "async_select_sound_output",
+    ),
-        "schema": SOUND_OUTPUT_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

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_RESET_FILTER: ServiceMethodDetails(method="async_reset_filter"),
-    SERVICE_SET_EXTRA_FEATURES: {
+    SERVICE_SET_EXTRA_FEATURES: ServiceMethodDetails(
-        "method": "async_set_extra_features",
+        method="async_set_extra_features",
-        "schema": SERVICE_SCHEMA_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
         )

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: {
+    SERVICE_SET_DELAYED_TURN_OFF: ServiceMethodDetails(
-        "method": "async_set_delayed_turn_off",
+        method="async_set_delayed_turn_off",
-        "schema": SERVICE_SCHEMA_SET_DELAYED_TURN_OFF,
+        schema=SERVICE_SCHEMA_SET_DELAYED_TURN_OFF,
-    },
+    ),
-    SERVICE_SET_SCENE: {
+    SERVICE_SET_SCENE: ServiceMethodDetails(
-        "method": "async_set_scene",
+        method="async_set_scene",
-        "schema": SERVICE_SCHEMA_SET_SCENE,
+        schema=SERVICE_SCHEMA_SET_SCENE,
-    },
+    ),
-    SERVICE_REMINDER_ON: {"method": "async_reminder_on"},
+    SERVICE_REMINDER_ON: ServiceMethodDetails(method="async_reminder_on"),
-    SERVICE_REMINDER_OFF: {"method": "async_reminder_off"},
+    SERVICE_REMINDER_OFF: ServiceMethodDetails(method="async_reminder_off"),
-    SERVICE_NIGHT_LIGHT_MODE_ON: {"method": "async_night_light_mode_on"},
+    SERVICE_NIGHT_LIGHT_MODE_ON: ServiceMethodDetails(
-    SERVICE_NIGHT_LIGHT_MODE_OFF: {"method": "async_night_light_mode_off"},
+        method="async_night_light_mode_on"
-    SERVICE_EYECARE_MODE_ON: {"method": "async_eyecare_mode_on"},
+    ),
-    SERVICE_EYECARE_MODE_OFF: {"method": "async_eyecare_mode_off"},
+    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
             )

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_ON: ServiceMethodDetails(method="async_set_wifi_led_on"),
-    SERVICE_SET_WIFI_LED_OFF: {"method": "async_set_wifi_led_off"},
+    SERVICE_SET_WIFI_LED_OFF: ServiceMethodDetails(method="async_set_wifi_led_off"),
-    SERVICE_SET_POWER_MODE: {
+    SERVICE_SET_POWER_MODE: ServiceMethodDetails(
-        "method": "async_set_power_mode",
+        method="async_set_power_mode",
-        "schema": SERVICE_SCHEMA_POWER_MODE,
+        schema=SERVICE_SCHEMA_POWER_MODE,
-    },
+    ),
-    SERVICE_SET_POWER_PRICE: {
+    SERVICE_SET_POWER_PRICE: ServiceMethodDetails(
-        "method": "async_set_power_price",
+        method="async_set_power_price",
-        "schema": SERVICE_SCHEMA_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
             )

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