Make helpers.script async

homeassistant/helpers/script.py

@@ -1,6 +1,6 @@
 """Helpers to execute scripts."""
+import asyncio
 import logging
-import threading
 from itertools import islice
 from typing import Optional, Sequence
 
@@ -10,9 +10,11 @@ from homeassistant.core import HomeAssistant
 from homeassistant.const import CONF_CONDITION
 from homeassistant.helpers import (
     service, condition, template, config_validation as cv)
-from homeassistant.helpers.event import track_point_in_utc_time
+from homeassistant.helpers.event import async_track_point_in_utc_time
 from homeassistant.helpers.typing import ConfigType
 import homeassistant.util.dt as date_util
+from homeassistant.util.async import (
+    run_coroutine_threadsafe, run_callback_threadsafe)
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -47,8 +49,7 @@ class Script():
         self.last_action = None
         self.can_cancel = any(CONF_DELAY in action for action
                               in self.sequence)
-        self._lock = threading.Lock()
+        self._async_unsub_delay_listener = None
-        self._unsub_delay_listener = None
         self._template_cache = {}
 
     @property
@@ -56,26 +57,35 @@ class Script():
         """Return true if script is on."""
         return self._cur != -1
 
-    def run(self, variables: Optional[Sequence]=None) -> None:
+    def run(self, variables=None):
         """Run script."""
-        with self._lock:
+        run_coroutine_threadsafe(
+            self.async_run(variables), self.hass.loop).result()
+
+    @asyncio.coroutine
+    def async_run(self, variables: Optional[Sequence]=None) -> None:
+        """Run script.
+
+        Returns a coroutine.
+        """
         if self._cur == -1:
             self._log('Running script')
             self._cur = 0
 
         # Unregister callback if we were in a delay but turn on is called
         # again. In that case we just continue execution.
-            self._remove_listener()
+        self._async_remove_listener()
 
         for cur, action in islice(enumerate(self.sequence), self._cur,
                                   None):
 
             if CONF_DELAY in action:
                 # Call ourselves in the future to continue work
+                @asyncio.coroutine
                 def script_delay(now):
                     """Called after delay is done."""
-                        self._unsub_delay_listener = None
+                    self._async_unsub_delay_listener = None
-                        self.run(variables)
+                    yield from self.async_run(variables)
 
                 delay = action[CONF_DELAY]
 
@@ -83,67 +93,71 @@ class Script():
                     delay = vol.All(
                         cv.time_period,
                         cv.positive_timedelta)(
-                                delay.render())
+                            delay.async_render())
 
-                    self._unsub_delay_listener = track_point_in_utc_time(
+                self._async_unsub_delay_listener = \
+                    async_track_point_in_utc_time(
                         self.hass, script_delay,
                         date_util.utcnow() + delay)
                 self._cur = cur + 1
-                    if self._change_listener:
+                self._trigger_change_listener()
-                        self._change_listener()
                 return
 
             elif CONF_CONDITION in action:
-                    if not self._check_condition(action, variables):
+                if not self._async_check_condition(action, variables):
                     break
 
             elif CONF_EVENT in action:
-                    self._fire_event(action)
+                self._async_fire_event(action)
 
             else:
-                    self._call_service(action, variables)
+                yield from self._async_call_service(action, variables)
 
         self._cur = -1
         self.last_action = None
-            if self._change_listener:
+        self._trigger_change_listener()
-                self._change_listener()
 
     def stop(self) -> None:
         """Stop running script."""
-        with self._lock:
+        run_callback_threadsafe(self.hass.loop, self.async_stop).result()
+
+    def async_stop(self) -> None:
+        """Stop running script."""
         if self._cur == -1:
             return
 
         self._cur = -1
-            self._remove_listener()
+        self._async_remove_listener()
-            if self._change_listener:
+        self._trigger_change_listener()
-                self._change_listener()
 
-    def _call_service(self, action, variables):
+    @asyncio.coroutine
+    def _async_call_service(self, action, variables):
         """Call the service specified in the action."""
         self.last_action = action.get(CONF_ALIAS, 'call service')
         self._log("Executing step %s" % self.last_action)
-        service.call_from_config(self.hass, action, True, variables,
+        yield from service.async_call_from_config(
-                                 validate_config=False)
+            self.hass, action, True, variables, validate_config=False)
 
-    def _fire_event(self, action):
+    def _async_fire_event(self, action):
         """Fire an event."""
         self.last_action = action.get(CONF_ALIAS, action[CONF_EVENT])
         self._log("Executing step %s" % self.last_action)
-        self.hass.bus.fire(action[CONF_EVENT], action.get(CONF_EVENT_DATA))
+        self.hass.bus.async_fire(action[CONF_EVENT],
+                                 action.get(CONF_EVENT_DATA))
 
-    def _check_condition(self, action, variables):
+    def _async_check_condition(self, action, variables):
         """Test if condition is matching."""
         self.last_action = action.get(CONF_ALIAS, action[CONF_CONDITION])
-        check = condition.from_config(action, False)(self.hass, variables)
+        check = condition.async_from_config(action, False)(
+            self.hass, variables)
         self._log("Test condition {}: {}".format(self.last_action, check))
         return check
 
-    def _remove_listener(self):
+    def _async_remove_listener(self):
         """Remove point in time listener, if any."""
-        if self._unsub_delay_listener:
+        if self._async_unsub_delay_listener:
-            self._unsub_delay_listener()
+            self._async_unsub_delay_listener()
-            self._unsub_delay_listener = None
+            self._async_unsub_delay_listener = None
 
     def _log(self, msg):
         """Logger helper."""
@@ -151,3 +165,10 @@ class Script():
             msg = "Script {}: {}".format(self.name, msg)
 
         _LOGGER.info(msg)
+
+    def _trigger_change_listener(self):
+        """Trigger the change listener."""
+        if not self._change_listener:
+            return
+
+        self.hass.async_add_job(self._change_listener)