Add support for triggers

pyfritzhome/cli.py

@@ -174,6 +174,27 @@ def template_apply(fritz, args):
     fritz.apply_template(args.ain)
 
 
+def list_triggers(fritz, args):
+    """Command that prints all trigger information."""
+    triggers = fritz.get_triggers()
+
+    for trigger in triggers:
+        print("#" * 30)
+        print("name=%s" % trigger.name)
+        print("  ain=%s" % trigger.ain)
+        print("  active=%s" % trigger.active)
+
+
+def trigger_set_active(fritz, args):
+    """Command that enables a trigger."""
+    fritz.set_trigger_active(args.ain)
+
+
+def trigger_set_inactive(fritz, args):
+    """Command that disables a trigger."""
+    fritz.set_trigger_inactive(args.ain)
+
+
 def main(args=None):
     """Enter the main function of the CLI tool."""
     parser = argparse.ArgumentParser(description="Fritz!Box Smarthome CLI tool.")
@@ -338,6 +359,24 @@ def main(args=None):
     subparser.add_argument("ain", type=str, metavar="AIN", help="Actor Identification")
     subparser.set_defaults(func=template_apply)
 
+    # triggers
+    subparser = _sub.add_parser("trigger", help="Trigger commands")
+    _sub_switch = subparser.add_subparsers()
+
+    # list triggers
+    subparser = _sub_switch.add_parser("list", help="List all available triggers")
+    subparser.set_defaults(func=list_triggers)
+
+    # activate triggers
+    subparser = _sub_switch.add_parser("activate", help="Activate a trigger")
+    subparser.add_argument("ain", type=str, metavar="AIN", help="Actor Identification")
+    subparser.set_defaults(func=trigger_set_active)
+
+    # deactivate triggers
+    subparser = _sub_switch.add_parser("deactivate", help="Deactivate a trigger")
+    subparser.add_argument("ain", type=str, metavar="AIN", help="Actor Identification")
+    subparser.set_defaults(func=trigger_set_inactive)
+
     args = parser.parse_args(args)
 
     logging.basicConfig()

pyfritzhome/devicetypes/__init__.py

@@ -12,6 +12,7 @@
 from .fritzhomedevicelightbulb import FritzhomeDeviceLightBulb
 from .fritzhomedeviceblind import FritzhomeDeviceBlind
 from .fritzhometemplate import FritzhomeTemplate
+from .fritzhometrigger import FritzhomeTrigger
 
 
 __all__ = (
@@ -27,4 +28,5 @@
     "FritzhomeDeviceLightBulb",
     "FritzhomeDeviceBlind",
     "FritzhomeTemplate",
+    "FritzhomeTrigger",
 )

pyfritzhome/devicetypes/fritzhomeentitybase.py

@@ -53,7 +53,11 @@ def _update_from_node(self, node):
     @property
     def device_and_unit_id(self):
         """Get the device and possible unit id."""
-        if self.ain.startswith("tmp") or self.ain.startswith("grp"):
+        if (
+            self.ain.startswith("tmp")
+            or self.ain.startswith("grp")
+            or self.ain.startswith("trg")
+        ):
             return (self.ain, None)
         elif self.ain.startswith("Z") and len(self.ain) == 19:
             return (self.ain[0:17], self.ain[17:])

pyfritzhome/devicetypes/fritzhometrigger.py

@@ -0,0 +1,22 @@
+"""The trigger class."""
+# -*- coding: utf-8 -*-
+
+import logging
+from xml.etree import ElementTree
+
+from .fritzhomeentitybase import FritzhomeEntityBase
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class FritzhomeTrigger(FritzhomeEntityBase):
+    """The Fritzhome Trigger class."""
+
+    active = None
+
+    def _update_from_node(self, node):
+        _LOGGER.debug("update trigger")
+        _LOGGER.debug(ElementTree.tostring(node))
+        self.ain = node.attrib["identifier"]
+        self.name = node.findtext("name").strip()
+        self.active = node.attrib["active"] == "1"

pyfritzhome/fritzhome.py

@@ -16,6 +16,7 @@
 from .errors import InvalidError, LoginError, NotLoggedInError
 from .fritzhomedevice import FritzhomeDevice
 from .fritzhomedevice import FritzhomeTemplate
+from .fritzhomedevice import FritzhomeTrigger
 from typing import Dict, Optional
 
 _LOGGER = logging.getLogger(__name__)
@@ -28,6 +29,7 @@ class Fritzhome(object):
     _session = None
     _devices: Optional[Dict[str, FritzhomeDevice]] = None
     _templates: Optional[Dict[str, FritzhomeTemplate]] = None
+    _triggers: Optional[Dict[str, FritzhomeTrigger]] = None
 
     def __init__(self, host, user, password, ssl_verify=True):
         """Create a fritzhome object."""
@@ -519,3 +521,71 @@ def get_template_by_ain(self, ain):
     def apply_template(self, ain):
         """Appliy a template."""
         self._aha_request("applytemplate", ain=ain)
+
+    # Trigger-related commands
+
+    def has_triggers(self):
+        """Check if the Fritz!Box supports smarthome triggers."""
+        plain = self._aha_request("gettriggerlistinfos")
+        try:
+            ElementTree.fromstring(plain)
+        except ElementTree.ParseError:
+            return False
+        return True
+
+    def update_triggers(self, ignore_removed=True):
+        """Update the triger."""
+        _LOGGER.info("Updating Trigers ...")
+        if self._triggers is None:
+            self._triggers = {}
+
+        trigger_elements = self.get_trigger_elements()
+        for element in trigger_elements:
+            if element.attrib["identifier"] in self._triggers.keys():
+                _LOGGER.info(
+                    "Updating already existing Trigger " + element.attrib["identifier"]
+                )
+                self._triggers[element.attrib["identifier"]]._update_from_node(element)
+            else:
+                _LOGGER.info("Adding new Trigger " + element.attrib["identifier"])
+                trigger = FritzhomeTrigger(self, node=element)
+                self._triggers[trigger.ain] = trigger
+
+        if not ignore_removed:
+            for identifier in list(self._triggers.keys()):
+                if identifier not in [
+                    element.attrib["identifier"] for element in trigger_elements
+                ]:
+                    _LOGGER.info("Removing no more existing trigger " + identifier)
+                    self._triggers.pop(identifier)
+
+        return True
+
+    def get_trigger_elements(self):
+        """Get the DOM elements for the trigger list."""
+        return self._get_listinfo_elements("trigger")
+
+    def get_triggers(self):
+        """Get the list of all known triggers."""
+        return list(self.get_triggers_as_dict().values())
+
+    def get_triggers_as_dict(self):
+        """Get all known triggers as dictionary."""
+        if self._triggers is None:
+            self.update_triggers()
+        return self._triggers
+
+    def get_trigger_by_ain(self, ain):
+        """Return a trigger specified by the AIN."""
+        return self.get_triggers_as_dict()[ain]
+
+    def _set_trigger_state(self, ain, state):
+        self._aha_request("settriggeractive", ain=ain, param={"active": state})
+
+    def set_trigger_active(self, ain):
+        """Set the trigger to active state."""
+        self._set_trigger_state(ain, "1")
+
+    def set_trigger_inactive(self, ain):
+        """Set the trigger to inactive state."""
+        self._set_trigger_state(ain, "0")

pyfritzhome/fritzhomedevice.py

@@ -3,6 +3,7 @@
 # -*- coding: utf-8 -*-
 
 from .devicetypes import FritzhomeTemplate  # noqa: F401
+from .devicetypes import FritzhomeTrigger  # noqa: F401
 from .devicetypes import (
     FritzhomeDeviceAlarm,
     FritzhomeDeviceBlind,

tests/responses/triggers/trigger_list.xml

@@ -0,0 +1,8 @@
+<triggerlist version="1">
+    <trigger identifier="trg303e4f-41979A02D" active="1">
+        <name>Test Trigger Active</name>
+    </trigger>
+    <trigger identifier="trg405e4a-57AE9A80A" active="0">
+        <name>Test Trigger Inactive</name>
+    </trigger>
+</triggerlist>

tests/responses/triggers/trigger_list_removed_trigger.xml

@@ -0,0 +1,5 @@
+<triggerlist version="1">
+    <trigger identifier="trg303e4f-41979A02D" active="1">
+        <name>Test Trigger Active</name>
+    </trigger>
+</triggerlist>

tests/test_fritzhometrigger.py

@@ -0,0 +1,90 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+from unittest.mock import MagicMock
+
+from pyfritzhome import Fritzhome
+
+from .helper import Helper
+
+
+class TestFritzhomeTrigger(object):
+    def setup_method(self):
+        self.mock = MagicMock()
+        self.fritz = Fritzhome("10.0.0.1", "user", "pass")
+        self.fritz._request = self.mock
+        self.fritz._devices = {}
+        self.fritz._sid = "0000001"
+
+        self.mock.side_effect = [Helper.response("triggers/trigger_list")]
+
+        self.fritz.update_triggers()
+
+    def test_trigger_init(self):
+        trigger = self.fritz.get_trigger_by_ain("trg303e4f-41979A02D")
+
+        assert trigger.ain == "trg303e4f-41979A02D"
+        assert trigger.active
+
+        trigger = self.fritz.get_trigger_by_ain("trg405e4a-57AE9A80A")
+
+        assert trigger.ain == "trg405e4a-57AE9A80A"
+        assert not trigger.active
+
+    def test_trigger_removed(self):
+        self.mock.side_effect = [
+            Helper.response("triggers/trigger_list"),
+            Helper.response("triggers/trigger_list_removed_trigger"),
+            Helper.response("triggers/trigger_list_removed_trigger"),
+        ]
+
+        self.fritz.update_triggers()
+        assert len(self.fritz.get_triggers()) == 2
+        self.fritz.update_triggers()
+        assert len(self.fritz.get_triggers()) == 2
+        self.fritz.update_triggers(ignore_removed=False)
+        assert len(self.fritz.get_triggers()) == 1
+
+    def test_has_trigger(self):
+        self.mock.side_effect = [
+            "invalid_xml",
+            Helper.response("triggers/trigger_list"),
+        ]
+        assert not self.fritz.has_triggers()
+        assert self.fritz.has_triggers()
+
+    def test_set_trigger_active(self):
+        self.mock.side_effect = [
+            Helper.response("triggers/trigger_list"),
+            "1",
+        ]
+        self.fritz.update_triggers()
+
+        self.fritz.set_trigger_active("trg303e4f-41979A02D")
+        self.fritz._request.assert_called_with(
+            "http://10.0.0.1/webservices/homeautoswitch.lua",
+            {
+                "switchcmd": "settriggeractive",
+                "sid": "0000001",
+                "active": "1",
+                "ain": "trg303e4f-41979A02D",
+            },
+        )
+
+    def test_set_trigger_inactive(self):
+        self.mock.side_effect = [
+            Helper.response("triggers/trigger_list"),
+            "1",
+        ]
+        self.fritz.update_triggers()
+
+        self.fritz.set_trigger_inactive("trg303e4f-41979A02D")
+        self.fritz._request.assert_called_with(
+            "http://10.0.0.1/webservices/homeautoswitch.lua",
+            {
+                "switchcmd": "settriggeractive",
+                "sid": "0000001",
+                "active": "0",
+                "ain": "trg303e4f-41979A02D",
+            },
+        )