Added webhook trigger - mimics event trigger for now

Author: contagon

custom_components/pyscript/__init__.py

@@ -51,6 +51,7 @@
 from .requirements import install_requirements
 from .state import State, StateVal
 from .trigger import TrigTime
+from .webhook import Webhook
 
 _LOGGER = logging.getLogger(LOGGER_PATH)
 
@@ -241,6 +242,7 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
     Mqtt.init(hass)
     TrigTime.init(hass)
     State.init(hass)
+    Webhook.init(hass)
     State.register_functions()
     GlobalContextMgr.init()
 

custom_components/pyscript/eval.py

@@ -50,6 +50,7 @@
     "state_trigger",
     "event_trigger",
     "mqtt_trigger",
+    "webhook_trigger",
     "state_active",
     "time_active",
     "task_unique",
@@ -363,6 +364,7 @@ async def trigger_init(self, trig_ctx, func_name):
             "mqtt_trigger",
             "state_trigger",
             "time_trigger",
+            "webhook_trigger",
         }
         arg_check = {
             "event_trigger": {"arg_cnt": {1, 2, 3}, "rep_ok": True},
@@ -373,6 +375,8 @@ async def trigger_init(self, trig_ctx, func_name):
             "task_unique": {"arg_cnt": {1, 2}},
             "time_active": {"arg_cnt": {"*"}},
             "time_trigger": {"arg_cnt": {0, "*"}, "rep_ok": True},
+            # TODO: Add in functionality to webhook with arguments
+            "webhook_trigger": {"arg_cnt": {1, 2, 3}, "rep_ok": True},
         }
         kwarg_check = {
             "event_trigger": {"kwargs": {dict}},
@@ -388,6 +392,8 @@ async def trigger_init(self, trig_ctx, func_name):
                 "state_hold_false": {int, float},
                 "watch": {set, list},
             },
+            # "webhook_trigger": {"call": {str, list}, "local": bool},
+            "webhook_trigger": {"kwargs": {dict}},
         }
 
         for dec in self.decorators:

custom_components/pyscript/trigger.py

@@ -21,6 +21,7 @@
 from .function import Function
 from .mqtt import Mqtt
 from .state import STATE_VIRTUAL_ATTRS, State
+from .webhook import Webhook
 
 _LOGGER = logging.getLogger(LOGGER_PATH + ".trigger")
 
@@ -222,13 +223,14 @@ async def wait_until(
         time_trigger=None,
         event_trigger=None,
         mqtt_trigger=None,
+        webhook_trigger=None,
         timeout=None,
         state_hold=None,
         state_hold_false=None,
         __test_handshake__=None,
     ):
         """Wait for zero or more triggers, until an optional timeout."""
-        if state_trigger is None and time_trigger is None and event_trigger is None and mqtt_trigger is None:
+        if state_trigger is None and time_trigger is None and event_trigger is None and mqtt_trigger is None and webhook_trigger is None:
             if timeout is not None:
                 await asyncio.sleep(timeout)
                 return {"trigger_type": "timeout"}
@@ -238,6 +240,7 @@ async def wait_until(
         state_trig_eval = None
         event_trig_expr = None
         mqtt_trig_expr = None
+        webhook_trig_expr = None
         exc = None
         notify_q = asyncio.Queue(0)
 
@@ -349,6 +352,22 @@ async def wait_until(
                         State.notify_del(state_trig_ident, notify_q)
                     raise exc
             await Mqtt.notify_add(mqtt_trigger[0], notify_q)
+        if webhook_trigger is not None:
+            if isinstance(webhook_trigger, str):
+                webhook_trigger = [webhook_trigger]
+            if len(webhook_trigger) > 1:
+                webhook_trig_expr = AstEval(
+                    f"{ast_ctx.name} webhook_trigger",
+                    ast_ctx.get_global_ctx(),
+                    logger_name=ast_ctx.get_logger_name(),
+                )
+                Function.install_ast_funcs(webhook_trig_expr)
+                webhook_trig_expr.parse(webhook_trigger[1], mode="eval")
+                exc = webhook_trig_expr.get_exception_obj()
+                if exc is not None:
+                    if len(state_trig_ident) > 0:
+                        State.notify_del(state_trig_ident, notify_q)
+                    raise exc
         time0 = time.monotonic()
 
         if __test_handshake__:
@@ -394,7 +413,7 @@ async def wait_until(
                     state_trig_timeout = True
                     time_next = now + dt.timedelta(seconds=this_timeout)
             if this_timeout is None:
-                if state_trigger is None and event_trigger is None and mqtt_trigger is None:
+                if state_trigger is None and event_trigger is None and mqtt_trigger is None and webhook_trigger is None:
                     _LOGGER.debug(
                         "trigger %s wait_until no next time - returning with none",
                         ast_ctx.name,
@@ -527,6 +546,17 @@ async def wait_until(
                 if mqtt_trig_ok:
                     ret = notify_info
                     break
+            elif notify_type == "webhook":
+                if webhook_trig_expr is None:
+                    ret = notify_info
+                    break
+                webhook_trig_ok = await webhook_trig_expr.eval(notify_info)
+                exc = webhook_trig_expr.get_exception_obj()
+                if exc is not None:
+                    break
+                if webhook_trig_ok:
+                    ret = notify_info
+                    break
             else:
                 _LOGGER.error(
                     "trigger %s wait_until got unexpected queue message %s",
@@ -540,6 +570,8 @@ async def wait_until(
             Event.notify_del(event_trigger[0], notify_q)
         if mqtt_trigger is not None:
             Mqtt.notify_del(mqtt_trigger[0], notify_q)
+        if webhook_trigger is not None:
+            Webhook.notify_del(webhook_trigger[0], notify_q)
         if exc:
             raise exc
         return ret
@@ -826,6 +858,8 @@ def __init__(
         self.event_trigger_kwargs = trig_cfg.get("event_trigger", {}).get("kwargs", {})
         self.mqtt_trigger = trig_cfg.get("mqtt_trigger", {}).get("args", None)
         self.mqtt_trigger_kwargs = trig_cfg.get("mqtt_trigger", {}).get("kwargs", {})
+        self.webhook_trigger = trig_cfg.get("webhook_trigger", {}).get("args", None)
+        self.webhook_trigger_kwargs = trig_cfg.get("webhook_trigger", {}).get("kwargs", {})
         self.state_active = trig_cfg.get("state_active", {}).get("args", None)
         self.time_active = trig_cfg.get("time_active", {}).get("args", None)
         self.time_active_hold_off = trig_cfg.get("time_active", {}).get("kwargs", {}).get("hold_off", None)
@@ -842,6 +876,7 @@ def __init__(
         self.state_trig_ident_any = set()
         self.event_trig_expr = None
         self.mqtt_trig_expr = None
+        self.webhook_trig_expr = None
         self.have_trigger = False
         self.setup_ok = False
         self.run_on_startup = False
@@ -933,6 +968,21 @@ def __init__(
                     return
             self.have_trigger = True
 
+        if self.webhook_trigger is not None:
+            if len(self.webhook_trigger) == 2:
+                self.webhook_trig_expr = AstEval(
+                    f"{self.name} @webhook_trigger()",
+                    self.global_ctx,
+                    logger_name=self.name,
+                )
+                Function.install_ast_funcs(self.webhook_trig_expr)
+                self.webhook_trig_expr.parse(self.webhook_trigger[1], mode="eval")
+                exc = self.webhook_trig_expr.get_exception_long()
+                if exc is not None:
+                    self.webhook_trig_expr.get_logger().error(exc)
+                    return
+            self.have_trigger = True
+
         self.setup_ok = True
 
     def stop(self):
@@ -945,6 +995,8 @@ def stop(self):
                 Event.notify_del(self.event_trigger[0], self.notify_q)
             if self.mqtt_trigger is not None:
                 Mqtt.notify_del(self.mqtt_trigger[0], self.notify_q)
+            if self.webhook_trigger is not None:
+                Webhook.notify_del(self.webhook_trigger[0], self.notify_q)
             if self.task:
                 Function.reaper_cancel(self.task)
                 self.task = None
@@ -995,6 +1047,9 @@ async def trigger_watch(self):
             if self.mqtt_trigger is not None:
                 _LOGGER.debug("trigger %s adding mqtt_trigger %s", self.name, self.mqtt_trigger[0])
                 await Mqtt.notify_add(self.mqtt_trigger[0], self.notify_q)
+            if self.webhook_trigger is not None:
+                _LOGGER.debug("trigger %s adding webhook_trigger %s", self.name, self.webhook_trigger[0])
+                Webhook.notify_add(self.webhook_trigger[0], self.notify_q)
 
             last_trig_time = None
             last_state_trig_time = None
@@ -1237,6 +1292,8 @@ async def trigger_watch(self):
                 Event.notify_del(self.event_trigger[0], self.notify_q)
             if self.mqtt_trigger is not None:
                 Mqtt.notify_del(self.mqtt_trigger[0], self.notify_q)
+            if self.webhook_trigger is not None:
+                Webhook.notify_del(self.webhook_trigger[0], self.notify_q)
             return
 
     def call_action(self, notify_type, func_args, run_task=True):

custom_components/pyscript/webhook.py

@@ -0,0 +1,76 @@
+"""Handles webhooks and notification."""
+
+import logging
+
+from .const import LOGGER_PATH
+
+_LOGGER = logging.getLogger(LOGGER_PATH + ".webhook")
+
+
+class Webhook:
+    """Define webhook functions."""
+
+    #
+    # Global hass instance
+    #
+    hass = None
+
+    #
+    # notify message queues by webhook type
+    #
+    notify = {}
+    notify_remove = {}
+
+    def __init__(self):
+        """Warn on Webhook instantiation."""
+        _LOGGER.error("Webhook class is not meant to be instantiated")
+
+    @classmethod
+    def init(cls, hass):
+        """Initialize Webhook."""
+
+        cls.hass = hass
+
+    @classmethod
+    async def webhook_listener(cls, event):
+        """Listen callback for given webhook which updates any notifications."""
+
+        func_args = {
+            "trigger_type": "event",
+            "event_type": event.event_type,
+            "context": event.context,
+        }
+        func_args.update(event.data)
+        await cls.update(event.event_type, func_args)
+
+    @classmethod
+    def notify_add(cls, webhook_type, queue):
+        """Register to notify for webhooks of given type to be sent to queue."""
+
+        if webhook_type not in cls.notify:
+            cls.notify[webhook_type] = set()
+            _LOGGER.debug("webhook.notify_add(%s) -> adding webhook listener", webhook_type)
+            cls.notify_remove[webhook_type] = cls.hass.bus.async_listen(webhook_type, cls.webhook_listener)
+        cls.notify[webhook_type].add(queue)
+
+    @classmethod
+    def notify_del(cls, webhook_type, queue):
+        """Unregister to notify for webhooks of given type for given queue."""
+
+        if webhook_type not in cls.notify or queue not in cls.notify[webhook_type]:
+            return
+        cls.notify[webhook_type].discard(queue)
+        if len(cls.notify[webhook_type]) == 0:
+            cls.notify_remove[webhook_type]()
+            _LOGGER.debug("webhook.notify_del(%s) -> removing webhook listener", webhook_type)
+            del cls.notify[webhook_type]
+            del cls.notify_remove[webhook_type]
+
+    @classmethod
+    async def update(cls, webhook_type, func_args):
+        """Deliver all notifications for an webhook of the given type."""
+
+        _LOGGER.debug("webhook.update(%s, %s)", webhook_type, func_args)
+        if webhook_type in cls.notify:
+            for queue in cls.notify[webhook_type]:
+                await queue.put(["webhook", func_args.copy()])