github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Improve type hints MQTT light schema json (#82208) 

* Improve type hints schema json

* Add hint for brightness

* Follow up comments

* Follow up missed comments

* Correct hint on flash arg

* hints on one line
This commit is contained in:
Jan Bouwhuis 2022-11-18 12:12:27 +01:00 committed by GitHub
parent 5453b346dd
commit fb909d694f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 64 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

111
homeassistant/components/mqtt/light/schema_json.py
View File

@ -3,6 +3,7 @@ from __future__ import annotations
from contextlib import suppress
import logging
from typing import Any, cast
import voluptuous as vol
@ -62,6 +63,7 @@ from ..const import (
)
from ..debug_info import log_messages
from ..mixins import MQTT_ENTITY_COMMON_SCHEMA, MqttEntity
from ..models import ReceiveMessage
from ..util import get_mqtt_data, valid_subscribe_topic
from .schema import MQTT_LIGHT_SCHEMA_SCHEMA
from .schema_basic import (
@ -102,7 +104,7 @@ CONF_MIN_MIREDS = "min_mireds"
CONF_WHITE_VALUE = "white_value"
def valid_color_configuration(config):
def valid_color_configuration(config: ConfigType) -> ConfigType:
"""Test color_mode is not combined with deprecated config."""
deprecated = {CONF_COLOR_TEMP, CONF_HS, CONF_RGB, CONF_XY}
if config[CONF_COLOR_MODE] and any(config.get(key) for key in deprecated):
@ -194,21 +196,27 @@ class MqttLightJson(MqttEntity, LightEntity, RestoreEntity):
_entity_id_format = ENTITY_ID_FORMAT
_attributes_extra_blocked = MQTT_LIGHT_ATTRIBUTES_BLOCKED
def __init__(self, hass, config, config_entry, discovery_data):
"""Initialize MQTT JSON light."""
self._topic = None
self._optimistic = False
self._fixed_color_mode = None
self._flash_times = None
_flash_times: dict[str, int | None]
_topic: dict[str, str | None]
_optimistic: bool
def __init__(
self,
hass: HomeAssistant,
config: ConfigType,
config_entry: ConfigEntry,
discovery_data: DiscoveryInfoType | None,
) -> None:
"""Initialize MQTT JSON light."""
self._fixed_color_mode: ColorMode | str | None = None
MqttEntity.__init__(self, hass, config, config_entry, discovery_data)
@staticmethod
def config_schema():
def config_schema() -> vol.Schema:
"""Return the config schema."""
return DISCOVERY_SCHEMA_JSON
def _setup_from_config(self, config):
def _setup_from_config(self, config: ConfigType) -> None:
"""(Re)Setup the entity."""
self._attr_max_mireds = config.get(CONF_MAX_MIREDS, super().max_mireds)
self._attr_min_mireds = config.get(CONF_MIN_MIREDS, super().min_mireds)
@ -217,7 +225,7 @@ class MqttLightJson(MqttEntity, LightEntity, RestoreEntity):
self._topic = {
key: config.get(key) for key in (CONF_STATE_TOPIC, CONF_COMMAND_TOPIC)
}
optimistic = config[CONF_OPTIMISTIC]
optimistic: bool = config[CONF_OPTIMISTIC]
self._optimistic = optimistic or self._topic[CONF_STATE_TOPIC] is None
self._flash_times = {
@ -240,14 +248,14 @@ class MqttLightJson(MqttEntity, LightEntity, RestoreEntity):
if config[CONF_HS] or config[CONF_RGB] or config[CONF_XY]:
color_modes.add(ColorMode.HS)
self._attr_supported_color_modes = filter_supported_color_modes(color_modes)
if len(self.supported_color_modes) == 1:
if self.supported_color_modes and len(self.supported_color_modes) == 1:
self._fixed_color_mode = next(iter(self.supported_color_modes))
else:
self._attr_supported_color_modes = self._config[CONF_SUPPORTED_COLOR_MODES]
if len(self.supported_color_modes) == 1:
if self.supported_color_modes and len(self.supported_color_modes) == 1:
self._attr_color_mode = next(iter(self.supported_color_modes))
def _update_color(self, values):
def _update_color(self, values: dict[str, Any]) -> None:
if not self._config[CONF_COLOR_MODE]:
# Deprecated color handling
try:
@ -287,7 +295,7 @@ class MqttLightJson(MqttEntity, LightEntity, RestoreEntity):
)
return
else:
color_mode = values["color_mode"]
color_mode: str = values["color_mode"]
if not self._supports_color_mode(color_mode):
_LOGGER.warning(
"Invalid color mode received for entity %s", self.entity_id
@ -336,14 +344,14 @@ class MqttLightJson(MqttEntity, LightEntity, RestoreEntity):
self.entity_id,
)
def _prepare_subscribe_topics(self):
def _prepare_subscribe_topics(self) -> None:
"""(Re)Subscribe to topics."""
@callback
@log_messages(self.hass, self.entity_id)
def state_received(msg):
def state_received(msg: ReceiveMessage) -> None:
"""Handle new MQTT messages."""
values = json_loads(msg.payload)
values: dict[str, Any] = json_loads(msg.payload)
if values["state"] == "ON":
self._attr_is_on = True
@ -382,7 +390,8 @@ class MqttLightJson(MqttEntity, LightEntity, RestoreEntity):
)
if (
ColorMode.COLOR_TEMP in self.supported_color_modes
self.supported_color_modes
and ColorMode.COLOR_TEMP in self.supported_color_modes
and not self._config[CONF_COLOR_MODE]
):
# Deprecated color handling
@ -419,7 +428,7 @@ class MqttLightJson(MqttEntity, LightEntity, RestoreEntity):
},
)
async def _subscribe_topics(self):
async def _subscribe_topics(self) -> None:
"""(Re)Subscribe to topics."""
await subscription.async_subscribe_topics(self.hass, self._sub_state)
@ -448,12 +457,12 @@ class MqttLightJson(MqttEntity, LightEntity, RestoreEntity):
self._attr_xy_color = last_attributes.get(ATTR_XY_COLOR, self.xy_color)
@property
def assumed_state(self):
def assumed_state(self) -> bool:
"""Return true if we do optimistic updates."""
return self._optimistic
@property
def color_mode(self):
def color_mode(self) -> ColorMode | str | None:
"""Return current color mode."""
if self._config[CONF_COLOR_MODE]:
return self._attr_color_mode
@ -465,41 +474,49 @@ class MqttLightJson(MqttEntity, LightEntity, RestoreEntity):
return ColorMode.HS
return ColorMode.COLOR_TEMP
def _set_flash_and_transition(self, message, **kwargs):
def _set_flash_and_transition(self, message: dict[str, Any], **kwargs: Any) -> None:
if ATTR_TRANSITION in kwargs:
message["transition"] = kwargs[ATTR_TRANSITION]
if ATTR_FLASH in kwargs:
flash = kwargs.get(ATTR_FLASH)
flash: str = kwargs[ATTR_FLASH]
if flash == FLASH_LONG:
message["flash"] = self._flash_times[CONF_FLASH_TIME_LONG]
elif flash == FLASH_SHORT:
message["flash"] = self._flash_times[CONF_FLASH_TIME_SHORT]
def _scale_rgbxx(self, rgbxx, kwargs):
def _scale_rgbxx(self, rgbxx: tuple[int, ...], kwargs: Any) -> tuple[int, ...]:
# If there's a brightness topic set, we don't want to scale the
# RGBxx values given using the brightness.
brightness: int
if self._config[CONF_BRIGHTNESS]:
brightness = 255
else:
brightness = kwargs.get(ATTR_BRIGHTNESS, 255)
return tuple(round(i / 255 * brightness) for i in rgbxx)
def _supports_color_mode(self, color_mode):
def _supports_color_mode(self, color_mode: ColorMode | str) -> bool:
"""Return True if the light natively supports a color mode."""
return (
self._config[CONF_COLOR_MODE] and color_mode in self.supported_color_modes
self._config[CONF_COLOR_MODE]
and self.supported_color_modes is not None
and color_mode in self.supported_color_modes
)
async def async_turn_on(self, **kwargs): # noqa: C901
async def async_turn_on(self, **kwargs: Any) -> None: # noqa: C901
"""Turn the device on.
This method is a coroutine.
"""
brightness: int
should_update = False
message = {"state": "ON"}
hs_color: tuple[float, float]
message: dict[str, Any] = {"state": "ON"}
rgb: tuple[int, ...]
rgbw: tuple[int, ...]
rgbcw: tuple[int, ...]
xy_color: tuple[float, float]
if ATTR_HS_COLOR in kwargs and (
self._config[CONF_HS] or self._config[CONF_RGB] or self._config[CONF_XY]
@ -546,37 +563,37 @@ class MqttLightJson(MqttEntity, LightEntity, RestoreEntity):
message["color"] = {"r": rgb[0], "g": rgb[1], "b": rgb[2]}
if self._optimistic:
self._attr_color_mode = ColorMode.RGB
self._attr_rgb_color = rgb
self._attr_rgb_color = cast(tuple[int, int, int], rgb)
should_update = True
if ATTR_RGBW_COLOR in kwargs and self._supports_color_mode(ColorMode.RGBW):
rgb = self._scale_rgbxx(kwargs[ATTR_RGBW_COLOR], kwargs)
message["color"] = {"r": rgb[0], "g": rgb[1], "b": rgb[2], "w": rgb[3]}
rgbw = self._scale_rgbxx(kwargs[ATTR_RGBW_COLOR], kwargs)
message["color"] = {"r": rgbw[0], "g": rgbw[1], "b": rgbw[2], "w": rgbw[3]}
if self._optimistic:
self._attr_color_mode = ColorMode.RGBW
self._attr_rgbw_color = rgb
self._attr_rgbw_color = cast(tuple[int, int, int, int], rgbw)
should_update = True
if ATTR_RGBWW_COLOR in kwargs and self._supports_color_mode(ColorMode.RGBWW):
rgb = self._scale_rgbxx(kwargs[ATTR_RGBWW_COLOR], kwargs)
rgbcw = self._scale_rgbxx(kwargs[ATTR_RGBWW_COLOR], kwargs)
message["color"] = {
"r": rgb[0],
"g": rgb[1],
"b": rgb[2],
"c": rgb[3],
"w": rgb[4],
"r": rgbcw[0],
"g": rgbcw[1],
"b": rgbcw[2],
"c": rgbcw[3],
"w": rgbcw[4],
}
if self._optimistic:
self._attr_color_mode = ColorMode.RGBWW
self._attr_rgbww_color = rgb
self._attr_rgbww_color = cast(tuple[int, int, int, int, int], rgbcw)
should_update = True
if ATTR_XY_COLOR in kwargs and self._supports_color_mode(ColorMode.XY):
xy = kwargs[ATTR_XY_COLOR] # pylint: disable=invalid-name
message["color"] = {"x": xy[0], "y": xy[1]}
xy_color = kwargs[ATTR_XY_COLOR]
message["color"] = {"x": xy_color[0], "y": xy_color[1]}
if self._optimistic:
self._attr_color_mode = ColorMode.XY
self._attr_xy_color = xy
self._attr_xy_color = xy_color
should_update = True
self._set_flash_and_transition(message, **kwargs)
@ -625,7 +642,7 @@ class MqttLightJson(MqttEntity, LightEntity, RestoreEntity):
should_update = True
await self.async_publish(
self._topic[CONF_COMMAND_TOPIC],
str(self._topic[CONF_COMMAND_TOPIC]),
json_dumps(message),
self._config[CONF_QOS],
self._config[CONF_RETAIN],
@ -640,17 +657,17 @@ class MqttLightJson(MqttEntity, LightEntity, RestoreEntity):
if should_update:
self.async_write_ha_state()
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the device off.
This method is a coroutine.
"""
message = {"state": "OFF"}
message: dict[str, Any] = {"state": "OFF"}
self._set_flash_and_transition(message, **kwargs)
await self.async_publish(
self._topic[CONF_COMMAND_TOPIC],
str(self._topic[CONF_COMMAND_TOPIC]),
json_dumps(message),
self._config[CONF_QOS],
self._config[CONF_RETAIN],