refactor/converse

make converse event based, it was a little odd because it was the only code from MycroftSkill called outside of the class itself

The event based approach also adds support for external skills, eg, Hivemind and OVOSAbstractApp

refactored converse process into smaller methods

added a new converse acknowledge bus message, now converse is only called for skills where converse is implemented.

skills are removed from active skill list if
- skill does not have converse capabilities
- skill was unloaded/disconnected
- skill does not want to converse at this time

Author: JarbasAl

mycroft/skills/intent_service.py

@@ -13,22 +13,21 @@
 # limitations under the License.
 #
 """Mycroft's intent service, providing intent parsing since forever!"""
-from copy import copy
 import time
+from threading import Event
 
 from mycroft.configuration import Configuration, setup_locale
-from mycroft.util.log import LOG
-from mycroft.util.parse import normalize
+from mycroft.messagebus.message import Message
 from mycroft.metrics import report_timing, Stopwatch
+from mycroft.skills.intent_service_interface import open_intent_envelope
 from mycroft.skills.intent_services import (
-    AdaptService, AdaptIntent,
-    FallbackService,
+    AdaptService, FallbackService,
     PadatiousService, PadatiousMatcher,
     IntentMatch
 )
-from mycroft.skills.intent_service_interface import open_intent_envelope
 from mycroft.skills.permissions import ConverseMode, ConverseActivationMode
-from mycroft.messagebus.message import Message
+from mycroft.util.log import LOG
+from mycroft.util.parse import normalize
 
 
 def _get_message_lang(message):
@@ -81,13 +80,13 @@ class IntentService:
     The intent service also provides the internal API for registering and
     querying the intent service.
     """
+
     def __init__(self, bus):
         self.bus = bus
         config = Configuration.get()
 
         # Dictionary for translating a skill id to a name
         self.skill_names = {}
-
         self.adapt_service = AdaptService(config.get('context', {}))
         try:
             self.padatious_service = PadatiousService(bus, config['padatious'])
@@ -211,17 +210,12 @@ def handle_activate_skill_request(self, message):
             return  # skill activation disabled
         elif self._consecutive_activations.get(skill_id, 0) > max_activations:
             return  # skill exceeded authorized consecutive number of
-                    # activations
+            # activations
 
         # activate skill
         self.add_active_skill(skill_id)
         self._consecutive_activations[skill_id] += 1
 
-        # converse handling
-
-    def handle_activate_skill_request(self, message):
-        self.add_active_skill(message.data['skill_id'])
-
     def handle_deactivate_skill_request(self, message):
         # TODO imperfect solution - only a skill can deactivate itself
         # someone can forge this message and emit it raw, but in ovos-core all
@@ -236,8 +230,8 @@ def reset_converse(self, message):
         """Let skills know there was a problem with speech recognition"""
         lang = _get_message_lang(message)
         setup_locale(lang)  # restore default lang
-        for skill in copy(self.active_skills):
-            self.do_converse(None, skill[0], lang, message)
+        for skill_id in self._collect_converse_skills():
+            self.do_converse(None, skill_id, lang, message)
 
     def do_converse(self, utterances, skill_id, lang, message):
         """Call skill and ask if they want to process the utterance.
@@ -249,29 +243,93 @@ def do_converse(self, utterances, skill_id, lang, message):
             lang (str): current language
             message (Message): message containing interaction info.
         """
+        if self._converse_allowed(skill_id):
+            converse_msg = message.reply("skill.converse.request",
+                                         {"skill_id": skill_id,
+                                          "utterances": utterances,
+                                          "lang": lang})
+            result = self.bus.wait_for_response(converse_msg,
+                                                'skill.converse.response')
+            if result and 'error' in result.data:
+                self.handle_converse_error(result)
+                return False
+            elif result is not None:
+                return result.data.get('result', False)
+        return False
+
+    def _converse_allowed(self, skill_id):
+        """ check if skill_id is not blacklisted from using converse """
         opmode = self.converse_config.get("converse_mode",
                                           ConverseMode.ACCEPT_ALL)
-
         if opmode == ConverseMode.BLACKLIST and skill_id in \
                 self.converse_config.get("converse_blacklist", []):
             return False
-
         elif opmode == ConverseMode.WHITELIST and skill_id not in \
                 self.converse_config.get("converse_whitelist", []):
             return False
+        return True
+
+    def _collect_converse_skills(self):
+        """use the messagebus api to determine which skills want to converse
+        This includes all skills and external applications"""
+        skill_ids = []
+        want_converse = []
+        active_skills = self.get_active_skill_ids()
+
+        def handle_ack(message):
+            skill_id = message.data["skill_id"]
+            if message.data.get("can_handle", True):
+                if skill_id in active_skills:
+                    want_converse.append(skill_id)
+            skill_ids.append(skill_id)
+
+        self.bus.on("mycroft.skill.converse.pong", handle_ack)
+
+        # wait for all skills to acknowledge they want to converse
+        self.bus.emit(Message("mycroft.skill.converse.ping"))
+        start = time.time()
+        while not all(s in skill_ids for s in active_skills) \
+                and time.time() - start <= 0.5:
+            time.sleep(0.02)
+
+        # remove any skill that didn't answer the ack from active list
+        for skill_id in active_skills:
+            if skill_id not in want_converse:
+                # - skill does not have converse capabilities
+                # - skill was unloaded/disconnected
+                # - skill does not want to converse at this time
+                self.remove_active_skill(skill_id)
+
+        self.bus.remove("mycroft.skill.converse.pong", handle_ack)
+        return skill_ids
+
+    def _check_converse_timeout(self):
+        """ filter active skill list based on timestamps """
+        timeouts = self.converse_config.get("skill_timeouts") or {}
+        def_timeout = self.converse_config.get("timeout", 300)
+        self.active_skills = [
+            skill for skill in self.active_skills
+            if time.time() - skill[1] <= timeouts.get(skill[0], def_timeout)]
 
-        converse_msg = (message.reply("skill.converse.request", {
-            "skill_id": skill_id, "utterances": utterances, "lang": lang}))
-        result = self.bus.wait_for_response(converse_msg,
-                                            'skill.converse.response')
-        if result and 'error' in result.data:
-            self.handle_converse_error(result)
-            ret = False
-        elif result is not None:
-            ret = result.data.get('result', False)
-        else:
-            ret = False
-        return ret
+    def _converse_with_skills(self, utterances, lang, message):
+        """Give active skills a chance at the utterance
+
+        Args:
+            utterances (list):  list of utterances
+            lang (string):      4 letter ISO language code
+            message (Message):  message to use to generate reply
+
+        Returns:
+            IntentMatch if handled otherwise None.
+        """
+        utterances = [item for tup in utterances for item in tup]
+        # filter allowed skills
+        self._check_converse_timeout()
+        # check if any skill wants to handle utterance
+        for skill_id in self._collect_converse_skills():
+            if self.do_converse(utterances, skill_id, lang, message):
+                return IntentMatch('Converse', None, None, skill_id)
+        return None
 
     def handle_converse_error(self, message):
         """Handle error in converse system.
@@ -281,24 +339,26 @@ def handle_converse_error(self, message):
         """
         skill_id = message.data["skill_id"]
         error_msg = message.data['error']
-        LOG.error("{}: {}".format(skill_id, error_msg))
-        if message.data["error"] == "skill id does not exist":
-            self.remove_active_skill(skill_id)
+        LOG.error(f"{skill_id}: {error_msg}")
+
+    def get_active_skill_ids(self):
+        return [skill[0] for skill in self.active_skills]
 
     def remove_active_skill(self, skill_id):
         """Remove a skill from being targetable by converse.
 
         Args:
             skill_id (str): skill to remove
         """
-        for skill in self.active_skills:
-            if skill[0] == skill_id:
-                self.active_skills.remove(skill)
-                self.bus.emit(
-                    Message("intent.service.skills.deactivated",
-                            {"skill_id": skill_id}))
-                if skill_id in self._consecutive_activations:
-                    self._consecutive_activations[skill_id] = 0
+        active_skills = self.get_active_skill_ids()
+        if skill_id in active_skills:
+            idx = active_skills.index(skill_id)
+            self.active_skills.pop(idx)
+            self.bus.emit(
+                Message("intent.service.skills.deactivated",
+                        {"skill_id": skill_id}))
+            if skill_id in self._consecutive_activations:
+                self._consecutive_activations[skill_id] = 0
 
     def add_active_skill(self, skill_id):
         """Add a skill or update the position of an active skill.
@@ -312,10 +372,13 @@ def add_active_skill(self, skill_id):
         # search the list for an existing entry that already contains it
         # and remove that reference
         if skill_id != '':
-            # do not call remove method to not send deactivate bus event!
-            for skill in self.active_skills:
-                if skill[0] == skill_id:
-                    self.active_skills.remove(skill)
+            # NOTE: do not call self.remove_active_skill
+            # do not want to send the deactivation bus event!
+            active_skills = self.get_active_skill_ids()
+            if skill_id in active_skills:
+                idx = active_skills.index(skill_id)
+                self.active_skills.pop(idx)
+
             # add skill with timestamp to start of skill_list
             self.active_skills.insert(0, [skill_id, time.time()])
             self.bus.emit(
@@ -391,7 +454,7 @@ def handle_utterance(self, message):
             # List of functions to use to match the utterance with intent.
             # These are listed in priority order.
             match_funcs = [
-                self._converse, padatious_matcher.match_high,
+                self._converse_with_skills, padatious_matcher.match_high,
                 self.adapt_service.match_intent, self.fallback.high_prio,
                 padatious_matcher.match_medium, self.fallback.medium_prio,
                 padatious_matcher.match_low, self.fallback.low_prio
@@ -427,35 +490,6 @@ def handle_utterance(self, message):
         except Exception as err:
             LOG.exception(err)
 
-    def _converse(self, utterances, lang, message):
-        """Give active skills a chance at the utterance
-
-        Args:
-            utterances (list):  list of utterances
-            lang (string):      4 letter ISO language code
-            message (Message):  message to use to generate reply
-
-        Returns:
-            IntentMatch if handled otherwise None.
-        """
-        utterances = [item for tup in utterances for item in tup]
-
-        # check for conversation time-out
-        skill_timeouts = self.converse_config.get("skill_timeouts") or {}
-        default_timeout = self.converse_config.get("timeout", 300)
-        self.active_skills = [
-            skill for skill in self.active_skills
-            if time.time() - skill[1] <= skill_timeouts.get(skill[0],
-                                                            default_timeout)]
-
-        # check if any skill wants to handle utterance
-        for skill in copy(self.active_skills):
-            if self.do_converse(utterances, skill[0], lang, message):
-                # update timestamp, or there will be a timeout where
-                # intent stops conversing whether its being used or not
-                return IntentMatch('Converse', None, None, skill[0])
-        return None
-
     def send_complete_intent_failure(self, message):
         """Send a message that no skill could handle the utterance.