home_emulation/devices/tradfri.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
"""
tradfri devices
===============

**Author:**

* Dirk Alders <sudo-dirk@mount-mockery.de>

**Description:**

    Emulation of tradfri devices

    Communication (MQTT)

        tradfri_light {
                    |      "state": ["ON" / "OFF" / "TOGGLE"]
                    |      "linkquality": [0...255] lqi
                    |      "brightness": [0...254]
                    |      "color_mode": ["color_temp"]
                    |      "color_temp": ["coolest", "cool", "neutral", "warm", "warmest", 250...454]
                    |      "color_temp_startup": ["coolest", "cool", "neutral", "warm", "warmest", "previous", 250...454]
                    |      "update": []
                    | }
                    +- get {
                    |           "state": ""
                    |      }
                    +- set {
                                "state": ["ON" / "OFF"]
                                "brightness": [0...256]
                                "color_temp": [250...454]
                                "transition": [0...] seconds
                                "brightness_move": [-X...0...X] X/s
                                "brightness_step": [-X...0...X]
                                "color_temp_move": [-X...0...X] X/s
                                "color_temp_step": [-X...0...X]
                            }
"""

from devices.base import base
import json


class sw(base):
    """A tradfri device with switching functionality

    Args:
        mqtt_client (mqtt.mqtt_client): A MQTT Client instance
        topic (str): the base topic for this device
        kwargs (**dict): cd_st=list of devices connected to state
    """
    PROPERTIES = [
        "state",
    ]
    def __init__(self, mqtt_client, topic, **kwargs):
        super().__init__(mqtt_client, topic, **kwargs)
        if getattr(self, 'cd_st', None) is None:
            self.cd_st = []
        self["state"] = "off"
        #
        self.mqtt_client.add_callback(self.topic + '/set', self.__rx_set__)
        self.mqtt_client.add_callback(self.topic + '/get', self.__rx_get__)

    def set_state(self, value):
        self.__set__("state", "on" if value else "off")
        self.send_device_status()

    def __rx_set__(self, client, userdata, message):
        data = json.loads(message.payload)
        self.logger.info("Received set data: %s", repr(data))
        for key in data:
            self.__set__(key, data[key])
        self.send_device_status()
        if "state" in data:
            for d in self.cd_st:
                d.set_state(data["state"].lower() == "on")

    def __rx_get__(self, client, userdata, message):
        self.send_device_status()

    def send_device_status(self):
        data = json.dumps(self)
        self.logger.info("Sending status: %s", repr(data))
        self.mqtt_client.send(self.topic, data)


class sw_br(sw):
    """A tradfri device with switching and brightness functionality

    Args:
        mqtt_client (mqtt.mqtt_client): A MQTT Client instance
        topic (str): the base topic for this device
    """
    PROPERTIES = sw.PROPERTIES + [
        "brightness",
    ]
    def __init__(self, mqtt_client, topic, **kwargs):
        super().__init__(mqtt_client, topic, **kwargs)
        self["brightness"] = 64


class sw_br_ct(sw_br):
    """A tradfri device with switching, brightness and colortemp functionality

    Args:
        mqtt_client (mqtt.mqtt_client): A MQTT Client instance
        topic (str): the base topic for this device
    """
    PROPERTIES = sw_br.PROPERTIES + [
        "color_temp",
    ]
    def __init__(self, mqtt_client, topic, **kwargs):
        super().__init__(mqtt_client, topic, **kwargs)
        self["color_temp"] = 413