PowerPlugins.py

# Copyright (c) 2020 Aldo Hoeben / fieldOfView
# OctoPrintPlugin is released under the terms of the AGPLv3 or higher.

from collections import OrderedDict
from typing import Any, Tuple, List, Dict

class PowerPlugins():

    def __init__(self) -> None:
        self._available_plugs = OrderedDict()  # type: Dict[str, Any]

    def parsePluginData(self, plugin_data: Dict[str, Any]) -> None:
        self._available_plugs = OrderedDict()  # type: Dict[str, Any]

        # plugins that only support a single plug
        simple_plugins = [
            ("psucontrol", "PSU Control", []),
            ("mystromswitch", "MyStrom Switch", ["ip"]),
            ("ikea_tradfri", "IKEA Trådfri", ["gateway_ip", "selected_outlet"])
        ]  # type: List[Tuple[str, str, List[str]]]
        for (plugin_id, plugin_name, additional_data) in simple_plugins:
            if plugin_id in plugin_data:
                all_config_set = True
                for config_item in additional_data:
                    if config_item not in plugin_data[plugin_id] or not plugin_data[plugin_id][config_item]:
                        all_config_set = False
                        break
                if all_config_set:
                    plug = OrderedDict([
                        ("plugin", plugin_id),
                        ("name", plugin_name)
                    ])
                    self._available_plugs[self._createPlugId(plug)] = plug

        # plugins that have a `label` and `ip` specified in `arrSmartplugs`
        common_api_plugins = [
            ("tplinksmartplug", "TP-Link Smartplug", []), # ip
            ("orvibos20", "Orvibo S20", []), # ip
            ("wemoswitch", "Wemo Switch", []), # ip
            ("tuyasmartplug", "Tuya Smartplug", []), # label
            ("domoticz", "Domoticz", ["idx", "username", "password"]), # ip, idx, username, password
            ("tasmota", "Tasmota", ["idx"]), # ip, idx, username, password, backlog_delay
        ]  # type: List[Tuple[str, str, List[str]]]
        for (plugin_id, plugin_name, additional_data) in common_api_plugins:
            if plugin_id in plugin_data and "arrSmartplugs" in plugin_data[plugin_id]:
                for plug_data in plugin_data[plugin_id]["arrSmartplugs"]:
                    if plug_data["ip"] and plug_data["label"]:
                        plug = OrderedDict([
                            ("plugin", plugin_id),
                            ("name", ("%s (%s)" % (plug_data["label"], plugin_name))),
                            ("label", plug_data["label"]),
                            ("ip", plug_data["ip"])
                        ])
                        for key in additional_data:
                            plug[key] = plug_data.get(key, "")
                        self._available_plugs[self._createPlugId(plug)] = plug

        # `tasmota_mqtt` has a slightly different settings dialect
        if "tasmota_mqtt" in plugin_data:
            plugin_id = "tasmota_mqtt"
            plugin_name = "Tasmota MQTT"
            for plug_data in plugin_data[plugin_id]["arrRelays"]:
                if plug_data["topic"] and plug_data["relayN"] != "":
                    plug = OrderedDict([
                        ("plugin", plugin_id),
                        ("name", "%s/%s (%s)" % (plug_data["topic"], plug_data["relayN"], plugin_name)),
                        ("topic", plug_data["topic"]),
                        ("relayN", plug_data["relayN"])
                    ])
                    self._available_plugs[self._createPlugId(plug)] = plug

    def _createPlugId(self, plug_data: Dict[str, Any]) -> str:
        interesting_bits = [v for (k, v) in plug_data.items() if k != "name"]
        return "/".join(interesting_bits)

    def getAvailablePowerPlugs(self) -> Dict[str, Any]:
        return self._available_plugs

    def getSetStateCommand(self, plug_id: str, state: bool) -> Tuple[str, Dict[str, Any]]:
        if plug_id not in self._available_plugs:
            return ("", {})

        plugin_id = self._available_plugs[plug_id]["plugin"]
        end_point = "plugin/" + plugin_id

        if plugin_id == "psucontrol":
            return (end_point, OrderedDict([("command", "turnPSUOn" if state else "turnPSUOff")]))

        if plugin_id == "mystromswitch":
            return (end_point, OrderedDict([("command", "enableRelais" if state else "disableRelais")]))

        plug_data = self._available_plugs[plug_id]
        command = OrderedDict([("command", "turnOn" if state else "turnOff")])
        arguments = []  # type: List[str]
        if plugin_id in ["tplinksmartplug", "orvibos20", "wemoswitch"]:
            # ip
            arguments = ["ip"]
        elif plugin_id == "domoticz":
            # ip, idx, username, password
            arguments = ["ip", "idx", "username", "password"]
        elif plugin_id == "tasmota_mqtt":
            # topic, relayN
            arguments = ["topic", "relayN"]
        elif plugin_id == "tasmota":
            # ip, idx
            arguments = ["ip", "idx"]
        elif plugin_id == "tuyasmartplug":
            # label
            arguments = ["label"]

        for key in arguments:
            command[key] = plug_data[key]

        return (end_point, command)