Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Use NamedTuple in Vallox service_to_method mapping #58361

Merged
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
60 changes: 35 additions & 25 deletions homeassistant/components/vallox/__init__.py
Expand Up @@ -4,7 +4,7 @@
from dataclasses import dataclass, field
import ipaddress
import logging
from typing import Any
from typing import Any, NamedTuple

from vallox_websocket_api import PROFILE as VALLOX_PROFILE, Vallox
from vallox_websocket_api.constants import vlxDevConstants
Expand Down Expand Up @@ -64,28 +64,36 @@
}
)


class ServiceMethodDetails(NamedTuple):
"""Details for SERVICE_TO_METHOD mapping."""

method: str
schema: vol.Schema


SERVICE_SET_PROFILE = "set_profile"
SERVICE_SET_PROFILE_FAN_SPEED_HOME = "set_profile_fan_speed_home"
SERVICE_SET_PROFILE_FAN_SPEED_AWAY = "set_profile_fan_speed_away"
SERVICE_SET_PROFILE_FAN_SPEED_BOOST = "set_profile_fan_speed_boost"

SERVICE_TO_METHOD = {
SERVICE_SET_PROFILE: {
"method": "async_set_profile",
"schema": SERVICE_SCHEMA_SET_PROFILE,
},
SERVICE_SET_PROFILE_FAN_SPEED_HOME: {
"method": "async_set_profile_fan_speed_home",
"schema": SERVICE_SCHEMA_SET_PROFILE_FAN_SPEED,
},
SERVICE_SET_PROFILE_FAN_SPEED_AWAY: {
"method": "async_set_profile_fan_speed_away",
"schema": SERVICE_SCHEMA_SET_PROFILE_FAN_SPEED,
},
SERVICE_SET_PROFILE_FAN_SPEED_BOOST: {
"method": "async_set_profile_fan_speed_boost",
"schema": SERVICE_SCHEMA_SET_PROFILE_FAN_SPEED,
},
SERVICE_SET_PROFILE: ServiceMethodDetails(
method="async_set_profile",
schema=SERVICE_SCHEMA_SET_PROFILE,
),
SERVICE_SET_PROFILE_FAN_SPEED_HOME: ServiceMethodDetails(
method="async_set_profile_fan_speed_home",
schema=SERVICE_SCHEMA_SET_PROFILE_FAN_SPEED,
),
SERVICE_SET_PROFILE_FAN_SPEED_AWAY: ServiceMethodDetails(
method="async_set_profile_fan_speed_away",
schema=SERVICE_SCHEMA_SET_PROFILE_FAN_SPEED,
),
SERVICE_SET_PROFILE_FAN_SPEED_BOOST: ServiceMethodDetails(
method="async_set_profile_fan_speed_boost",
schema=SERVICE_SCHEMA_SET_PROFILE_FAN_SPEED,
),
}


Expand Down Expand Up @@ -153,10 +161,12 @@ async def async_update_data() -> ValloxState:
)

service_handler = ValloxServiceHandler(client, coordinator)
for vallox_service, method in SERVICE_TO_METHOD.items():
schema = method["schema"]
for vallox_service, service_details in SERVICE_TO_METHOD.items():
hass.services.async_register(
DOMAIN, vallox_service, service_handler.async_handle, schema=schema
DOMAIN,
vallox_service,
service_handler.async_handle,
schema=service_details.schema,
)

hass.data[DOMAIN] = {"client": client, "coordinator": coordinator, "name": name}
Expand Down Expand Up @@ -258,17 +268,17 @@ async def async_set_profile_fan_speed_boost(

async def async_handle(self, call: ServiceCall) -> None:
"""Dispatch a service call."""
method = SERVICE_TO_METHOD.get(call.service)
service_details = SERVICE_TO_METHOD.get(call.service)
params = call.data.copy()

if method is None:
if service_details is None:
return

if not hasattr(self, method["method"]):
_LOGGER.error("Service not implemented: %s", method["method"])
if not hasattr(self, service_details.method):
_LOGGER.error("Service not implemented: %s", service_details.method)
return

result = await getattr(self, method["method"])(**params)
result = await getattr(self, service_details.method)(**params)

# This state change affects other entities like sensors. Force an immediate update that can
# be observed by all parties involved.
Expand Down