Skip to content

Use ColorMode enum in mqtt #70579

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.

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Merged
bdraco merged 1 commit into from
Apr 24, 2022
Merged

Use ColorMode enum in mqtt #70579

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 42 additions & 51 deletions homeassistant/components/mqtt/light/schema_basic.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,22 +19,13 @@
ATTR_WHITE,
ATTR_WHITE_VALUE,
ATTR_XY_COLOR,
COLOR_MODE_BRIGHTNESS,
COLOR_MODE_COLOR_TEMP,
COLOR_MODE_HS,
COLOR_MODE_ONOFF,
COLOR_MODE_RGB,
COLOR_MODE_RGBW,
COLOR_MODE_RGBWW,
COLOR_MODE_UNKNOWN,
COLOR_MODE_WHITE,
COLOR_MODE_XY,
ENTITY_ID_FORMAT,
SUPPORT_BRIGHTNESS,
SUPPORT_COLOR,
SUPPORT_COLOR_TEMP,
SUPPORT_EFFECT,
SUPPORT_WHITE_VALUE,
ColorMode,
LightEntity,
valid_supported_color_modes,
)
Expand Down Expand Up @@ -384,35 +375,35 @@ def _setup_from_config(self, config):
self._optimistic_xy_color = optimistic or topic[CONF_XY_STATE_TOPIC] is None
supported_color_modes = set()
if topic[CONF_COLOR_TEMP_COMMAND_TOPIC] is not None:
supported_color_modes.add(COLOR_MODE_COLOR_TEMP)
self._color_mode = COLOR_MODE_COLOR_TEMP
supported_color_modes.add(ColorMode.COLOR_TEMP)
self._color_mode = ColorMode.COLOR_TEMP
if topic[CONF_HS_COMMAND_TOPIC] is not None:
supported_color_modes.add(COLOR_MODE_HS)
self._color_mode = COLOR_MODE_HS
supported_color_modes.add(ColorMode.HS)
self._color_mode = ColorMode.HS
if topic[CONF_RGB_COMMAND_TOPIC] is not None:
supported_color_modes.add(COLOR_MODE_RGB)
self._color_mode = COLOR_MODE_RGB
supported_color_modes.add(ColorMode.RGB)
self._color_mode = ColorMode.RGB
if topic[CONF_RGBW_COMMAND_TOPIC] is not None:
supported_color_modes.add(COLOR_MODE_RGBW)
self._color_mode = COLOR_MODE_RGBW
supported_color_modes.add(ColorMode.RGBW)
self._color_mode = ColorMode.RGBW
if topic[CONF_RGBWW_COMMAND_TOPIC] is not None:
supported_color_modes.add(COLOR_MODE_RGBWW)
self._color_mode = COLOR_MODE_RGBWW
supported_color_modes.add(ColorMode.RGBWW)
self._color_mode = ColorMode.RGBWW
if topic[CONF_WHITE_COMMAND_TOPIC] is not None:
supported_color_modes.add(COLOR_MODE_WHITE)
supported_color_modes.add(ColorMode.WHITE)
if topic[CONF_XY_COMMAND_TOPIC] is not None:
supported_color_modes.add(COLOR_MODE_XY)
self._color_mode = COLOR_MODE_XY
supported_color_modes.add(ColorMode.XY)
self._color_mode = ColorMode.XY
if len(supported_color_modes) > 1:
self._color_mode = COLOR_MODE_UNKNOWN
self._color_mode = ColorMode.UNKNOWN

if not supported_color_modes:
if topic[CONF_BRIGHTNESS_COMMAND_TOPIC] is not None:
self._color_mode = COLOR_MODE_BRIGHTNESS
supported_color_modes.add(COLOR_MODE_BRIGHTNESS)
self._color_mode = ColorMode.BRIGHTNESS
supported_color_modes.add(ColorMode.BRIGHTNESS)
else:
self._color_mode = COLOR_MODE_ONOFF
supported_color_modes.add(COLOR_MODE_ONOFF)
self._color_mode = ColorMode.ONOFF
supported_color_modes.add(ColorMode.ONOFF)

# Validate the color_modes configuration
self._supported_color_modes = valid_supported_color_modes(supported_color_modes)
Expand Down Expand Up @@ -503,7 +494,7 @@ def _rgbx_received(msg, template, color_mode, convert_color):
def rgb_received(msg):
"""Handle new MQTT messages for RGB."""
rgb = _rgbx_received(
msg, CONF_RGB_VALUE_TEMPLATE, COLOR_MODE_RGB, lambda *x: x
msg, CONF_RGB_VALUE_TEMPLATE, ColorMode.RGB, lambda *x: x
)
if not rgb:
return
Expand All @@ -522,7 +513,7 @@ def rgbw_received(msg):
rgbw = _rgbx_received(
msg,
CONF_RGBW_VALUE_TEMPLATE,
COLOR_MODE_RGBW,
ColorMode.RGBW,
color_util.color_rgbw_to_rgb,
)
if not rgbw:
Expand All @@ -539,7 +530,7 @@ def rgbww_received(msg):
rgbww = _rgbx_received(
msg,
CONF_RGBWW_VALUE_TEMPLATE,
COLOR_MODE_RGBWW,
ColorMode.RGBWW,
color_util.color_rgbww_to_rgb,
)
if not rgbww:
Expand Down Expand Up @@ -577,7 +568,7 @@ def color_temp_received(msg):
return

if self._optimistic_color_mode:
self._color_mode = COLOR_MODE_COLOR_TEMP
self._color_mode = ColorMode.COLOR_TEMP
self._color_temp = int(payload)
self.async_write_ha_state()

Expand Down Expand Up @@ -610,7 +601,7 @@ def hs_received(msg):
try:
hs_color = tuple(float(val) for val in payload.split(",", 2))
if self._optimistic_color_mode:
self._color_mode = COLOR_MODE_HS
self._color_mode = ColorMode.HS
self._hs_color = hs_color
self.async_write_ha_state()
except ValueError:
Expand Down Expand Up @@ -647,7 +638,7 @@ def xy_received(msg):

xy_color = tuple(float(val) for val in payload.split(","))
if self._optimistic_color_mode:
self._color_mode = COLOR_MODE_XY
self._color_mode = ColorMode.XY
if self._legacy_mode:
self._hs_color = color_util.color_xy_to_hs(*xy_color)
else:
Expand Down Expand Up @@ -863,9 +854,9 @@ def render_rgbx(color, template, color_mode):
"""Render RGBx payload."""
if tpl := self._command_templates[template]:
keys = ["red", "green", "blue"]
if color_mode == COLOR_MODE_RGBW:
if color_mode == ColorMode.RGBW:
keys.append("white")
elif color_mode == COLOR_MODE_RGBWW:
elif color_mode == ColorMode.RGBWW:
keys.extend(["cold_white", "warm_white"])
rgb_color_str = tpl(variables=zip(keys, color))
else:
Expand Down Expand Up @@ -905,15 +896,15 @@ def set_optimistic(attribute, value, color_mode=None, condition_attribute=None):
):
# Legacy mode: Convert HS to RGB
rgb = scale_rgbx(color_util.color_hsv_to_RGB(*hs_color, 100))
rgb_s = render_rgbx(rgb, CONF_RGB_COMMAND_TEMPLATE, COLOR_MODE_RGB)
rgb_s = render_rgbx(rgb, CONF_RGB_COMMAND_TEMPLATE, ColorMode.RGB)
await publish(CONF_RGB_COMMAND_TOPIC, rgb_s)
should_update |= set_optimistic(
ATTR_HS_COLOR, hs_color, condition_attribute=ATTR_RGB_COLOR
)

if hs_color and self._topic[CONF_HS_COMMAND_TOPIC] is not None:
await publish(CONF_HS_COMMAND_TOPIC, f"{hs_color[0]},{hs_color[1]}")
should_update |= set_optimistic(ATTR_HS_COLOR, hs_color, COLOR_MODE_HS)
should_update |= set_optimistic(ATTR_HS_COLOR, hs_color, ColorMode.HS)

if (
hs_color
Expand All @@ -933,37 +924,37 @@ def set_optimistic(attribute, value, color_mode=None, condition_attribute=None):
and not self._legacy_mode
):
scaled = scale_rgbx(rgb)
rgb_s = render_rgbx(scaled, CONF_RGB_COMMAND_TEMPLATE, COLOR_MODE_RGB)
rgb_s = render_rgbx(scaled, CONF_RGB_COMMAND_TEMPLATE, ColorMode.RGB)
await publish(CONF_RGB_COMMAND_TOPIC, rgb_s)
should_update |= set_optimistic(ATTR_RGB_COLOR, rgb, COLOR_MODE_RGB)
should_update |= set_optimistic(ATTR_RGB_COLOR, rgb, ColorMode.RGB)

if (
(rgbw := kwargs.get(ATTR_RGBW_COLOR))
and self._topic[CONF_RGBW_COMMAND_TOPIC] is not None
and not self._legacy_mode
):
scaled = scale_rgbx(rgbw)
rgbw_s = render_rgbx(scaled, CONF_RGBW_COMMAND_TEMPLATE, COLOR_MODE_RGBW)
rgbw_s = render_rgbx(scaled, CONF_RGBW_COMMAND_TEMPLATE, ColorMode.RGBW)
await publish(CONF_RGBW_COMMAND_TOPIC, rgbw_s)
should_update |= set_optimistic(ATTR_RGBW_COLOR, rgbw, COLOR_MODE_RGBW)
should_update |= set_optimistic(ATTR_RGBW_COLOR, rgbw, ColorMode.RGBW)

if (
(rgbww := kwargs.get(ATTR_RGBWW_COLOR))
and self._topic[CONF_RGBWW_COMMAND_TOPIC] is not None
and not self._legacy_mode
):
scaled = scale_rgbx(rgbww)
rgbww_s = render_rgbx(scaled, CONF_RGBWW_COMMAND_TEMPLATE, COLOR_MODE_RGBWW)
rgbww_s = render_rgbx(scaled, CONF_RGBWW_COMMAND_TEMPLATE, ColorMode.RGBWW)
await publish(CONF_RGBWW_COMMAND_TOPIC, rgbww_s)
should_update |= set_optimistic(ATTR_RGBWW_COLOR, rgbww, COLOR_MODE_RGBWW)
should_update |= set_optimistic(ATTR_RGBWW_COLOR, rgbww, ColorMode.RGBWW)

if (
(xy_color := kwargs.get(ATTR_XY_COLOR))
and self._topic[CONF_XY_COMMAND_TOPIC] is not None
and not self._legacy_mode
):
await publish(CONF_XY_COMMAND_TOPIC, f"{xy_color[0]},{xy_color[1]}")
should_update |= set_optimistic(ATTR_XY_COLOR, xy_color, COLOR_MODE_XY)
should_update |= set_optimistic(ATTR_XY_COLOR, xy_color, ColorMode.XY)

if (
ATTR_BRIGHTNESS in kwargs
Expand All @@ -990,7 +981,7 @@ def set_optimistic(attribute, value, color_mode=None, condition_attribute=None):
hs_color = self._hs_color if self._hs_color is not None else (0, 0)
brightness = kwargs[ATTR_BRIGHTNESS]
rgb = scale_rgbx(color_util.color_hsv_to_RGB(*hs_color, 100), brightness)
rgb_s = render_rgbx(rgb, CONF_RGB_COMMAND_TEMPLATE, COLOR_MODE_RGB)
rgb_s = render_rgbx(rgb, CONF_RGB_COMMAND_TEMPLATE, ColorMode.RGB)
await publish(CONF_RGB_COMMAND_TOPIC, rgb_s)
should_update |= set_optimistic(ATTR_BRIGHTNESS, kwargs[ATTR_BRIGHTNESS])
elif (
Expand All @@ -1001,7 +992,7 @@ def set_optimistic(attribute, value, color_mode=None, condition_attribute=None):
):
rgb_color = self._rgb_color if self._rgb_color is not None else (255,) * 3
rgb = scale_rgbx(rgb_color, kwargs[ATTR_BRIGHTNESS])
rgb_s = render_rgbx(rgb, CONF_RGB_COMMAND_TEMPLATE, COLOR_MODE_RGB)
rgb_s = render_rgbx(rgb, CONF_RGB_COMMAND_TEMPLATE, ColorMode.RGB)
await publish(CONF_RGB_COMMAND_TOPIC, rgb_s)
should_update |= set_optimistic(ATTR_BRIGHTNESS, kwargs[ATTR_BRIGHTNESS])
elif (
Expand All @@ -1014,7 +1005,7 @@ def set_optimistic(attribute, value, color_mode=None, condition_attribute=None):
self._rgbw_color if self._rgbw_color is not None else (255,) * 4
)
rgbw = scale_rgbx(rgbw_color, kwargs[ATTR_BRIGHTNESS])
rgbw_s = render_rgbx(rgbw, CONF_RGBW_COMMAND_TEMPLATE, COLOR_MODE_RGBW)
rgbw_s = render_rgbx(rgbw, CONF_RGBW_COMMAND_TEMPLATE, ColorMode.RGBW)
await publish(CONF_RGBW_COMMAND_TOPIC, rgbw_s)
should_update |= set_optimistic(ATTR_BRIGHTNESS, kwargs[ATTR_BRIGHTNESS])
elif (
Expand All @@ -1027,7 +1018,7 @@ def set_optimistic(attribute, value, color_mode=None, condition_attribute=None):
self._rgbww_color if self._rgbww_color is not None else (255,) * 5
)
rgbww = scale_rgbx(rgbww_color, kwargs[ATTR_BRIGHTNESS])
rgbww_s = render_rgbx(rgbww, CONF_RGBWW_COMMAND_TEMPLATE, COLOR_MODE_RGBWW)
rgbww_s = render_rgbx(rgbww, CONF_RGBWW_COMMAND_TEMPLATE, ColorMode.RGBWW)
await publish(CONF_RGBWW_COMMAND_TOPIC, rgbww_s)
should_update |= set_optimistic(ATTR_BRIGHTNESS, kwargs[ATTR_BRIGHTNESS])
if (
Expand All @@ -1040,7 +1031,7 @@ def set_optimistic(attribute, value, color_mode=None, condition_attribute=None):

await publish(CONF_COLOR_TEMP_COMMAND_TOPIC, color_temp)
should_update |= set_optimistic(
ATTR_COLOR_TEMP, kwargs[ATTR_COLOR_TEMP], COLOR_MODE_COLOR_TEMP
ATTR_COLOR_TEMP, kwargs[ATTR_COLOR_TEMP], ColorMode.COLOR_TEMP
)

if ATTR_EFFECT in kwargs and self._topic[CONF_EFFECT_COMMAND_TOPIC] is not None:
Expand All @@ -1059,7 +1050,7 @@ def set_optimistic(attribute, value, color_mode=None, condition_attribute=None):
should_update |= set_optimistic(
ATTR_BRIGHTNESS,
kwargs[ATTR_WHITE],
COLOR_MODE_WHITE,
ColorMode.WHITE,
)

if (
Expand Down
51 changes: 23 additions & 28 deletions homeassistant/components/mqtt/light/schema_json.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,6 @@
ATTR_TRANSITION,
ATTR_WHITE_VALUE,
ATTR_XY_COLOR,
COLOR_MODE_COLOR_TEMP,
COLOR_MODE_HS,
COLOR_MODE_RGB,
COLOR_MODE_RGBW,
COLOR_MODE_RGBWW,
COLOR_MODE_XY,
ENTITY_ID_FORMAT,
FLASH_LONG,
FLASH_SHORT,
Expand All @@ -35,6 +29,7 @@
SUPPORT_TRANSITION,
SUPPORT_WHITE_VALUE,
VALID_COLOR_MODES,
ColorMode,
LightEntity,
legacy_supported_features,
valid_supported_color_modes,
Expand Down Expand Up @@ -267,39 +262,39 @@ def _update_color(self, values):
_LOGGER.warning("Invalid color mode received")
return
try:
if color_mode == COLOR_MODE_COLOR_TEMP:
if color_mode == ColorMode.COLOR_TEMP:
self._color_temp = int(values["color_temp"])
self._color_mode = COLOR_MODE_COLOR_TEMP
elif color_mode == COLOR_MODE_HS:
self._color_mode = ColorMode.COLOR_TEMP
elif color_mode == ColorMode.HS:
hue = float(values["color"]["h"])
saturation = float(values["color"]["s"])
self._color_mode = COLOR_MODE_HS
self._color_mode = ColorMode.HS
self._hs = (hue, saturation)
elif color_mode == COLOR_MODE_RGB:
elif color_mode == ColorMode.RGB:
r = int(values["color"]["r"]) # pylint: disable=invalid-name
g = int(values["color"]["g"]) # pylint: disable=invalid-name
b = int(values["color"]["b"]) # pylint: disable=invalid-name
self._color_mode = COLOR_MODE_RGB
self._color_mode = ColorMode.RGB
self._rgb = (r, g, b)
elif color_mode == COLOR_MODE_RGBW:
elif color_mode == ColorMode.RGBW:
r = int(values["color"]["r"]) # pylint: disable=invalid-name
g = int(values["color"]["g"]) # pylint: disable=invalid-name
b = int(values["color"]["b"]) # pylint: disable=invalid-name
w = int(values["color"]["w"]) # pylint: disable=invalid-name
self._color_mode = COLOR_MODE_RGBW
self._color_mode = ColorMode.RGBW
self._rgbw = (r, g, b, w)
elif color_mode == COLOR_MODE_RGBWW:
elif color_mode == ColorMode.RGBWW:
r = int(values["color"]["r"]) # pylint: disable=invalid-name
g = int(values["color"]["g"]) # pylint: disable=invalid-name
b = int(values["color"]["b"]) # pylint: disable=invalid-name
c = int(values["color"]["c"]) # pylint: disable=invalid-name
w = int(values["color"]["w"]) # pylint: disable=invalid-name
self._color_mode = COLOR_MODE_RGBWW
self._color_mode = ColorMode.RGBWW
self._rgbww = (r, g, b, c, w)
elif color_mode == COLOR_MODE_XY:
elif color_mode == ColorMode.XY:
x = float(values["color"]["x"]) # pylint: disable=invalid-name
y = float(values["color"]["y"]) # pylint: disable=invalid-name
self._color_mode = COLOR_MODE_XY
self._color_mode = ColorMode.XY
self._xy = (x, y)
except (KeyError, ValueError):
_LOGGER.warning("Invalid or incomplete color value received")
Expand Down Expand Up @@ -553,31 +548,31 @@ async def async_turn_on(self, **kwargs): # noqa: C901
self._hs = kwargs[ATTR_HS_COLOR]
should_update = True

if ATTR_HS_COLOR in kwargs and self._supports_color_mode(COLOR_MODE_HS):
if ATTR_HS_COLOR in kwargs and self._supports_color_mode(ColorMode.HS):
hs_color = kwargs[ATTR_HS_COLOR]
message["color"] = {"h": hs_color[0], "s": hs_color[1]}
if self._optimistic:
self._color_mode = COLOR_MODE_HS
self._color_mode = ColorMode.HS
self._hs = hs_color
should_update = True

if ATTR_RGB_COLOR in kwargs and self._supports_color_mode(COLOR_MODE_RGB):
if ATTR_RGB_COLOR in kwargs and self._supports_color_mode(ColorMode.RGB):
rgb = self._scale_rgbxx(kwargs[ATTR_RGB_COLOR], kwargs)
message["color"] = {"r": rgb[0], "g": rgb[1], "b": rgb[2]}
if self._optimistic:
self._color_mode = COLOR_MODE_RGB
self._color_mode = ColorMode.RGB
self._rgb = rgb
should_update = True

if ATTR_RGBW_COLOR in kwargs and self._supports_color_mode(COLOR_MODE_RGBW):
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]}
if self._optimistic:
self._color_mode = COLOR_MODE_RGBW
self._color_mode = ColorMode.RGBW
self._rgbw = rgb
should_update = True

if ATTR_RGBWW_COLOR in kwargs and self._supports_color_mode(COLOR_MODE_RGBWW):
if ATTR_RGBWW_COLOR in kwargs and self._supports_color_mode(ColorMode.RGBWW):
rgb = self._scale_rgbxx(kwargs[ATTR_RGBWW_COLOR], kwargs)
message["color"] = {
"r": rgb[0],
Expand All @@ -587,15 +582,15 @@ async def async_turn_on(self, **kwargs): # noqa: C901
"w": rgb[4],
}
if self._optimistic:
self._color_mode = COLOR_MODE_RGBWW
self._color_mode = ColorMode.RGBWW
self._rgbww = rgb
should_update = True

if ATTR_XY_COLOR in kwargs and self._supports_color_mode(COLOR_MODE_XY):
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]}
if self._optimistic:
self._color_mode = COLOR_MODE_XY
self._color_mode = ColorMode.XY
self._xy = xy
should_update = True

Expand Down