From 9c851790dc5c5d39c3f10ffcbf4ac49c68c0702e Mon Sep 17 00:00:00 2001 From: Corban Mailloux Date: Wed, 10 Aug 2016 02:55:10 -0400 Subject: [PATCH] Add support for new mqtt_json light platform. (#2777) * Add support for new mqtt_json light platform. * Fix W503 errors. * Bring in feedback from @balloob. * Add test coverage for invalid color and brightness data. * Add coverage for transition in turn_off. --- homeassistant/components/light/mqtt_json.py | 233 +++++++++++++ tests/components/light/test_mqtt_json.py | 344 ++++++++++++++++++++ 2 files changed, 577 insertions(+) create mode 100755 homeassistant/components/light/mqtt_json.py create mode 100755 tests/components/light/test_mqtt_json.py diff --git a/homeassistant/components/light/mqtt_json.py b/homeassistant/components/light/mqtt_json.py new file mode 100755 index 00000000000..bd8ea589252 --- /dev/null +++ b/homeassistant/components/light/mqtt_json.py @@ -0,0 +1,233 @@ +""" +Support for MQTT JSON lights. + +For more details about this platform, please refer to the documentation at +https://home-assistant.io/components/light.mqtt_json/ +""" + +import logging +import json +import voluptuous as vol + +import homeassistant.components.mqtt as mqtt +from homeassistant.components.light import ( + ATTR_BRIGHTNESS, ATTR_RGB_COLOR, ATTR_TRANSITION, + ATTR_FLASH, FLASH_LONG, FLASH_SHORT, Light) +from homeassistant.const import CONF_NAME, CONF_OPTIMISTIC, CONF_PLATFORM +from homeassistant.components.mqtt import ( + CONF_STATE_TOPIC, CONF_COMMAND_TOPIC, CONF_QOS, CONF_RETAIN) +import homeassistant.helpers.config_validation as cv + +_LOGGER = logging.getLogger(__name__) + +DOMAIN = "mqtt_json" + +DEPENDENCIES = ["mqtt"] + +DEFAULT_NAME = "MQTT JSON Light" +DEFAULT_OPTIMISTIC = False +DEFAULT_BRIGHTNESS = False +DEFAULT_RGB = False +DEFAULT_FLASH_TIME_SHORT = 2 +DEFAULT_FLASH_TIME_LONG = 10 + +CONF_BRIGHTNESS = "brightness" +CONF_RGB = "rgb" +CONF_FLASH_TIME_SHORT = "flash_time_short" +CONF_FLASH_TIME_LONG = "flash_time_long" + +# Stealing some of these from the base MQTT configs. +PLATFORM_SCHEMA = vol.Schema({ + vol.Required(CONF_PLATFORM): DOMAIN, + vol.Optional(CONF_QOS, default=mqtt.DEFAULT_QOS): + vol.All(vol.Coerce(int), vol.In([0, 1, 2])), + vol.Required(CONF_COMMAND_TOPIC): mqtt.valid_publish_topic, + vol.Optional(CONF_RETAIN, default=mqtt.DEFAULT_RETAIN): cv.boolean, + vol.Optional(CONF_STATE_TOPIC): mqtt.valid_subscribe_topic, + vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string, + vol.Optional(CONF_OPTIMISTIC, default=DEFAULT_OPTIMISTIC): cv.boolean, + vol.Optional(CONF_BRIGHTNESS, default=DEFAULT_BRIGHTNESS): cv.boolean, + vol.Optional(CONF_RGB, default=DEFAULT_RGB): cv.boolean, + vol.Optional(CONF_FLASH_TIME_SHORT, default=DEFAULT_FLASH_TIME_SHORT): + cv.positive_int, + vol.Optional(CONF_FLASH_TIME_LONG, default=DEFAULT_FLASH_TIME_LONG): + cv.positive_int +}) + + +def setup_platform(hass, config, add_devices_callback, discovery_info=None): + """Add MQTT JSON Light.""" + add_devices_callback([MqttJson( + hass, + config[CONF_NAME], + { + key: config.get(key) for key in ( + CONF_STATE_TOPIC, + CONF_COMMAND_TOPIC + ) + }, + config[CONF_QOS], + config[CONF_RETAIN], + config[CONF_OPTIMISTIC], + config[CONF_BRIGHTNESS], + config[CONF_RGB], + { + key: config.get(key) for key in ( + CONF_FLASH_TIME_SHORT, + CONF_FLASH_TIME_LONG + ) + } + )]) + + +class MqttJson(Light): + """MQTT JSON light.""" + + # pylint: disable=too-many-arguments,too-many-instance-attributes + def __init__(self, hass, name, topic, qos, retain, + optimistic, brightness, rgb, flash_times): + """Initialize MQTT JSON light.""" + self._hass = hass + self._name = name + self._topic = topic + self._qos = qos + self._retain = retain + self._optimistic = optimistic or topic["state_topic"] is None + self._state = False + if brightness: + self._brightness = 255 + else: + self._brightness = None + + if rgb: + self._rgb = [0, 0, 0] + else: + self._rgb = None + + self._flash_times = flash_times + + def state_received(topic, payload, qos): + """A new MQTT message has been received.""" + values = json.loads(payload) + + if values["state"] == "ON": + self._state = True + elif values["state"] == "OFF": + self._state = False + + if self._rgb is not None: + try: + red = int(values["color"]["r"]) + green = int(values["color"]["g"]) + blue = int(values["color"]["b"]) + + self._rgb = [red, green, blue] + except KeyError: + pass + except ValueError: + _LOGGER.warning("Invalid color value received.") + + if self._brightness is not None: + try: + self._brightness = int(values["brightness"]) + except KeyError: + pass + except ValueError: + _LOGGER.warning("Invalid brightness value received.") + + self.update_ha_state() + + if self._topic["state_topic"] is not None: + mqtt.subscribe(self._hass, self._topic["state_topic"], + state_received, self._qos) + + @property + def brightness(self): + """Return the brightness of this light between 0..255.""" + return self._brightness + + @property + def rgb_color(self): + """Return the RGB color value.""" + return self._rgb + + @property + def should_poll(self): + """No polling needed for a MQTT light.""" + return False + + @property + def name(self): + """Return the name of the device if any.""" + return self._name + + @property + def is_on(self): + """Return true if device is on.""" + return self._state + + @property + def assumed_state(self): + """Return true if we do optimistic updates.""" + return self._optimistic + + def turn_on(self, **kwargs): + """Turn the device on.""" + should_update = False + + message = {"state": "ON"} + + if ATTR_RGB_COLOR in kwargs: + message["color"] = { + "r": kwargs[ATTR_RGB_COLOR][0], + "g": kwargs[ATTR_RGB_COLOR][1], + "b": kwargs[ATTR_RGB_COLOR][2] + } + + if self._optimistic: + self._rgb = kwargs[ATTR_RGB_COLOR] + should_update = True + + if ATTR_FLASH in kwargs: + flash = kwargs.get(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] + + if ATTR_TRANSITION in kwargs: + message["transition"] = kwargs[ATTR_TRANSITION] + + if ATTR_BRIGHTNESS in kwargs: + message["brightness"] = int(kwargs[ATTR_BRIGHTNESS]) + + if self._optimistic: + self._brightness = kwargs[ATTR_BRIGHTNESS] + should_update = True + + mqtt.publish(self._hass, self._topic["command_topic"], + json.dumps(message), self._qos, self._retain) + + if self._optimistic: + # Optimistically assume that the light has changed state. + self._state = True + should_update = True + + if should_update: + self.update_ha_state() + + def turn_off(self, **kwargs): + """Turn the device off.""" + message = {"state": "OFF"} + + if ATTR_TRANSITION in kwargs: + message["transition"] = kwargs[ATTR_TRANSITION] + + mqtt.publish(self._hass, self._topic["command_topic"], + json.dumps(message), self._qos, self._retain) + + if self._optimistic: + # Optimistically assume that the light has changed state. + self._state = False + self.update_ha_state() diff --git a/tests/components/light/test_mqtt_json.py b/tests/components/light/test_mqtt_json.py new file mode 100755 index 00000000000..d149d2ba04c --- /dev/null +++ b/tests/components/light/test_mqtt_json.py @@ -0,0 +1,344 @@ +"""The tests for the MQTT JSON light platform. + +Configuration for RGB Version with brightness: + +light: + platform: mqtt_json + name: mqtt_json_light_1 + state_topic: "home/rgb1" + command_topic: "home/rgb1/set" + brightness: true + rgb: true + +Config without RGB: + +light: + platform: mqtt_json + name: mqtt_json_light_1 + state_topic: "home/rgb1" + command_topic: "home/rgb1/set" + brightness: true + +Config without RGB and brightness: + +light: + platform: mqtt_json + name: mqtt_json_light_1 + state_topic: "home/rgb1" + command_topic: "home/rgb1/set" +""" +import json +import unittest + +from homeassistant.bootstrap import _setup_component +from homeassistant.const import STATE_ON, STATE_OFF, ATTR_ASSUMED_STATE +import homeassistant.components.light as light +from tests.common import ( + get_test_home_assistant, mock_mqtt_component, fire_mqtt_message) + + +class TestLightMQTTJSON(unittest.TestCase): + """Test the MQTT JSON light.""" + + def setUp(self): # pylint: disable=invalid-name + """Setup things to be run when tests are started.""" + self.hass = get_test_home_assistant() + self.mock_publish = mock_mqtt_component(self.hass) + + def tearDown(self): # pylint: disable=invalid-name + """Stop everything that was started.""" + self.hass.stop() + + def test_fail_setup_if_no_command_topic(self): + """Test if setup fails with no command topic.""" + self.hass.config.components = ['mqtt'] + assert not _setup_component(self.hass, light.DOMAIN, { + light.DOMAIN: { + 'platform': 'mqtt_json', + 'name': 'test', + } + }) + self.assertIsNone(self.hass.states.get('light.test')) + + def test_no_color_or_brightness_if_no_config(self): + """Test if there is no color and brightness if they aren't defined.""" + self.hass.config.components = ['mqtt'] + assert _setup_component(self.hass, light.DOMAIN, { + light.DOMAIN: { + 'platform': 'mqtt_json', + 'name': 'test', + 'state_topic': 'test_light_rgb', + 'command_topic': 'test_light_rgb/set', + } + }) + + state = self.hass.states.get('light.test') + self.assertEqual(STATE_OFF, state.state) + self.assertIsNone(state.attributes.get('rgb_color')) + self.assertIsNone(state.attributes.get('brightness')) + + fire_mqtt_message(self.hass, 'test_light_rgb', '{"state":"ON"}') + self.hass.pool.block_till_done() + + state = self.hass.states.get('light.test') + self.assertEqual(STATE_ON, state.state) + self.assertIsNone(state.attributes.get('rgb_color')) + self.assertIsNone(state.attributes.get('brightness')) + + def test_controlling_state_via_topic(self): + """Test the controlling of the state via topic.""" + self.hass.config.components = ['mqtt'] + assert _setup_component(self.hass, light.DOMAIN, { + light.DOMAIN: { + 'platform': 'mqtt_json', + 'name': 'test', + 'state_topic': 'test_light_rgb', + 'command_topic': 'test_light_rgb/set', + 'brightness': True, + 'rgb': True, + 'qos': '0' + } + }) + + state = self.hass.states.get('light.test') + self.assertEqual(STATE_OFF, state.state) + self.assertIsNone(state.attributes.get('rgb_color')) + self.assertIsNone(state.attributes.get('brightness')) + self.assertIsNone(state.attributes.get(ATTR_ASSUMED_STATE)) + + # Turn on the light, full white + fire_mqtt_message(self.hass, 'test_light_rgb', + '{"state":"ON",' + + '"color":{"r":255,"g":255,"b":255},' + + '"brightness":255}' + ) + self.hass.pool.block_till_done() + + state = self.hass.states.get('light.test') + self.assertEqual(STATE_ON, state.state) + self.assertEqual([255, 255, 255], state.attributes.get('rgb_color')) + self.assertEqual(255, state.attributes.get('brightness')) + + # Turn the light off + fire_mqtt_message(self.hass, 'test_light_rgb', '{"state":"OFF"}') + self.hass.pool.block_till_done() + + state = self.hass.states.get('light.test') + self.assertEqual(STATE_OFF, state.state) + + fire_mqtt_message(self.hass, 'test_light_rgb', + '{"state":"ON",' + + '"brightness":100}' + ) + self.hass.pool.block_till_done() + + light_state = self.hass.states.get('light.test') + self.hass.pool.block_till_done() + self.assertEqual(100, + light_state.attributes['brightness']) + + fire_mqtt_message(self.hass, 'test_light_rgb', + '{"state":"ON",' + + '"color":{"r":125,"g":125,"b":125}}' + ) + self.hass.pool.block_till_done() + + light_state = self.hass.states.get('light.test') + self.assertEqual([125, 125, 125], + light_state.attributes.get('rgb_color')) + + def test_sending_mqtt_commands_and_optimistic(self): + """Test the sending of command in optimistic mode.""" + self.hass.config.components = ['mqtt'] + assert _setup_component(self.hass, light.DOMAIN, { + light.DOMAIN: { + 'platform': 'mqtt_json', + 'name': 'test', + 'command_topic': 'test_light_rgb/set', + 'brightness': True, + 'rgb': True, + 'qos': 2 + } + }) + + state = self.hass.states.get('light.test') + self.assertEqual(STATE_OFF, state.state) + self.assertTrue(state.attributes.get(ATTR_ASSUMED_STATE)) + + light.turn_on(self.hass, 'light.test') + self.hass.pool.block_till_done() + + self.assertEqual(('test_light_rgb/set', '{"state": "ON"}', 2, False), + self.mock_publish.mock_calls[-1][1]) + state = self.hass.states.get('light.test') + self.assertEqual(STATE_ON, state.state) + + light.turn_off(self.hass, 'light.test') + self.hass.pool.block_till_done() + + self.assertEqual(('test_light_rgb/set', '{"state": "OFF"}', 2, False), + self.mock_publish.mock_calls[-1][1]) + state = self.hass.states.get('light.test') + self.assertEqual(STATE_OFF, state.state) + + light.turn_on(self.hass, 'light.test', rgb_color=[75, 75, 75], + brightness=50) + self.hass.pool.block_till_done() + + self.assertEqual('test_light_rgb/set', + self.mock_publish.mock_calls[-1][1][0]) + self.assertEqual(2, self.mock_publish.mock_calls[-1][1][2]) + self.assertEqual(False, self.mock_publish.mock_calls[-1][1][3]) + # Get the sent message + message_json = json.loads(self.mock_publish.mock_calls[-1][1][1]) + self.assertEqual(50, message_json["brightness"]) + self.assertEqual(75, message_json["color"]["r"]) + self.assertEqual(75, message_json["color"]["g"]) + self.assertEqual(75, message_json["color"]["b"]) + self.assertEqual("ON", message_json["state"]) + + state = self.hass.states.get('light.test') + self.assertEqual(STATE_ON, state.state) + self.assertEqual((75, 75, 75), state.attributes['rgb_color']) + self.assertEqual(50, state.attributes['brightness']) + + def test_flash_short_and_long(self): + """Test for flash length being sent when included.""" + self.hass.config.components = ['mqtt'] + assert _setup_component(self.hass, light.DOMAIN, { + light.DOMAIN: { + 'platform': 'mqtt_json', + 'name': 'test', + 'state_topic': 'test_light_rgb', + 'command_topic': 'test_light_rgb/set', + 'flash_time_short': 5, + 'flash_time_long': 15, + 'qos': 0 + } + }) + + state = self.hass.states.get('light.test') + self.assertEqual(STATE_OFF, state.state) + + light.turn_on(self.hass, 'light.test', flash="short") + self.hass.pool.block_till_done() + + self.assertEqual('test_light_rgb/set', + self.mock_publish.mock_calls[-1][1][0]) + self.assertEqual(0, self.mock_publish.mock_calls[-1][1][2]) + self.assertEqual(False, self.mock_publish.mock_calls[-1][1][3]) + # Get the sent message + message_json = json.loads(self.mock_publish.mock_calls[-1][1][1]) + self.assertEqual(5, message_json["flash"]) + self.assertEqual("ON", message_json["state"]) + + light.turn_on(self.hass, 'light.test', flash="long") + self.hass.pool.block_till_done() + + self.assertEqual('test_light_rgb/set', + self.mock_publish.mock_calls[-1][1][0]) + self.assertEqual(0, self.mock_publish.mock_calls[-1][1][2]) + self.assertEqual(False, self.mock_publish.mock_calls[-1][1][3]) + # Get the sent message + message_json = json.loads(self.mock_publish.mock_calls[-1][1][1]) + self.assertEqual(15, message_json["flash"]) + self.assertEqual("ON", message_json["state"]) + + def test_transition(self): + """Test for transition time being sent when included.""" + self.hass.config.components = ['mqtt'] + assert _setup_component(self.hass, light.DOMAIN, { + light.DOMAIN: { + 'platform': 'mqtt_json', + 'name': 'test', + 'state_topic': 'test_light_rgb', + 'command_topic': 'test_light_rgb/set', + 'qos': 0 + } + }) + + state = self.hass.states.get('light.test') + self.assertEqual(STATE_OFF, state.state) + + light.turn_on(self.hass, 'light.test', transition=10) + self.hass.pool.block_till_done() + + self.assertEqual('test_light_rgb/set', + self.mock_publish.mock_calls[-1][1][0]) + self.assertEqual(0, self.mock_publish.mock_calls[-1][1][2]) + self.assertEqual(False, self.mock_publish.mock_calls[-1][1][3]) + # Get the sent message + message_json = json.loads(self.mock_publish.mock_calls[-1][1][1]) + self.assertEqual(10, message_json["transition"]) + self.assertEqual("ON", message_json["state"]) + + # Transition back off + light.turn_off(self.hass, 'light.test', transition=10) + self.hass.pool.block_till_done() + + self.assertEqual('test_light_rgb/set', + self.mock_publish.mock_calls[-1][1][0]) + self.assertEqual(0, self.mock_publish.mock_calls[-1][1][2]) + self.assertEqual(False, self.mock_publish.mock_calls[-1][1][3]) + # Get the sent message + message_json = json.loads(self.mock_publish.mock_calls[-1][1][1]) + self.assertEqual(10, message_json["transition"]) + self.assertEqual("OFF", message_json["state"]) + + def test_invalid_color_and_brightness_values(self): + """Test that invalid color/brightness values are ignored.""" + self.hass.config.components = ['mqtt'] + assert _setup_component(self.hass, light.DOMAIN, { + light.DOMAIN: { + 'platform': 'mqtt_json', + 'name': 'test', + 'state_topic': 'test_light_rgb', + 'command_topic': 'test_light_rgb/set', + 'brightness': True, + 'rgb': True, + 'qos': '0' + } + }) + + state = self.hass.states.get('light.test') + self.assertEqual(STATE_OFF, state.state) + self.assertIsNone(state.attributes.get('rgb_color')) + self.assertIsNone(state.attributes.get('brightness')) + self.assertIsNone(state.attributes.get(ATTR_ASSUMED_STATE)) + + # Turn on the light + fire_mqtt_message(self.hass, 'test_light_rgb', + '{"state":"ON",' + + '"color":{"r":255,"g":255,"b":255},' + + '"brightness": 255}' + ) + self.hass.pool.block_till_done() + + state = self.hass.states.get('light.test') + self.assertEqual(STATE_ON, state.state) + self.assertEqual([255, 255, 255], state.attributes.get('rgb_color')) + self.assertEqual(255, state.attributes.get('brightness')) + + # Bad color values + fire_mqtt_message(self.hass, 'test_light_rgb', + '{"state":"ON",' + + '"color":{"r":"bad","g":"val","b":"test"}}' + ) + self.hass.pool.block_till_done() + + # Color should not have changed + state = self.hass.states.get('light.test') + self.assertEqual(STATE_ON, state.state) + self.assertEqual([255, 255, 255], state.attributes.get('rgb_color')) + + # Bad brightness values + fire_mqtt_message(self.hass, 'test_light_rgb', + '{"state":"ON",' + + '"brightness": "badValue"}' + ) + self.hass.pool.block_till_done() + + # Brightness should not have changed + state = self.hass.states.get('light.test') + self.assertEqual(STATE_ON, state.state) + self.assertEqual(255, state.attributes.get('brightness'))