feat:pipeline plugin factory

ovos_core/intent_services/converse_service.py

@@ -1,24 +1,28 @@
 import time
 from threading import Event
-from typing import Optional, List
+from typing import Optional, Dict, List, Union
 
+from ovos_bus_client.client import MessageBusClient
 from ovos_bus_client.message import Message
 from ovos_bus_client.session import SessionManager, UtteranceState, Session
 from ovos_bus_client.util import get_message_lang
 from ovos_config.config import Configuration
 from ovos_config.locale import setup_locale
-from ovos_plugin_manager.templates.pipeline import PipelineMatch, PipelinePlugin
+from ovos_plugin_manager.templates.pipeline import PipelineMatch, PipelineStageMatcher
 from ovos_utils import flatten_list
+from ovos_utils.fakebus import FakeBus
 from ovos_utils.lang import standardize_lang_tag
-from ovos_utils.log import LOG
+from ovos_utils.log import LOG, deprecated
 from ovos_workshop.permissions import ConverseMode, ConverseActivationMode
 
 
-class ConverseService(PipelinePlugin):
+class ConverseService(PipelineStageMatcher):
     """Intent Service handling conversational skills."""
 
-    def __init__(self, bus):
-        self.bus = bus
+    def __init__(self, bus: Optional[Union[MessageBusClient, FakeBus]] = None,
+                 config: Optional[Dict] = None):
+        config = config or Configuration().get("skills", {}).get("converse", {})
+        super().__init__(bus, config)
         self._consecutive_activations = {}
         self.bus.on('mycroft.speech.recognition.unknown', self.reset_converse)
         self.bus.on('intent.service.skills.deactivate', self.handle_deactivate_skill_request)
@@ -27,7 +31,6 @@ def __init__(self, bus):
         self.bus.on('intent.service.active_skills.get', self.handle_get_active_skills)
         self.bus.on("skill.converse.get_response.enable", self.handle_get_response_enable)
         self.bus.on("skill.converse.get_response.disable", self.handle_get_response_disable)
-        super().__init__(config=Configuration().get("skills", {}).get("converse") or {})
 
     @property
     def active_skills(self):
@@ -312,7 +315,7 @@ def converse(self, utterances: List[str], skill_id: str, lang: str, message: Mes
                             f'increasing "max_skill_runtime" in mycroft.conf might help alleviate this issue')
         return False
 
-    def converse_with_skills(self, utterances: List[str], lang: str, message: Message) -> Optional[PipelineMatch]:
+    def match(self, utterances: List[str], lang: str, message: Message) -> Optional[PipelineMatch]:
         """Give active skills a chance at the utterance
 
         Args:
@@ -398,6 +401,10 @@ def handle_get_active_skills(self, message: Message):
         self.bus.emit(message.reply("intent.service.active_skills.reply",
                                     {"skills": self.get_active_skills(message)}))
 
+    @deprecated("'converse_with_skills' has been renamed to 'match'", "2.0.0")
+    def converse_with_skills(self, utterances: List[str], lang: str, message: Message = None) -> Optional[PipelineMatch]:
+        return self.match(utterances, lang, message)
+
     def shutdown(self):
         self.bus.remove('mycroft.speech.recognition.unknown', self.reset_converse)
         self.bus.remove('intent.service.skills.deactivate', self.handle_deactivate_skill_request)

ovos_core/intent_services/fallback_service.py

@@ -12,41 +12,43 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-"""Intent service for Mycroft's fallback system."""
+"""Intent service for OVOS's fallback system."""
 import operator
 import time
 from collections import namedtuple
-from typing import Optional, List
+from typing import Optional, Dict, List, Union
 
+from ovos_bus_client.client import MessageBusClient
 from ovos_bus_client.message import Message
 from ovos_bus_client.session import SessionManager
 from ovos_config import Configuration
-from ovos_plugin_manager.templates.pipeline import PipelineMatch, PipelinePlugin
+from ovos_plugin_manager.templates.pipeline import PipelineMatch, PipelineStageConfidenceMatcher
 from ovos_utils import flatten_list
+from ovos_utils.fakebus import FakeBus
 from ovos_utils.lang import standardize_lang_tag
-from ovos_utils.log import LOG
+from ovos_utils.log import LOG, deprecated, log_deprecation
 from ovos_workshop.permissions import FallbackMode
 
 FallbackRange = namedtuple('FallbackRange', ['start', 'stop'])
 
 
-class FallbackService(PipelinePlugin):
+class FallbackService(PipelineStageConfidenceMatcher):
     """Intent Service handling fallback skills."""
 
-    def __init__(self, bus):
-        self.bus = bus
-        self.fallback_config = Configuration()["skills"].get("fallbacks", {})
+    def __init__(self, bus: Optional[Union[MessageBusClient, FakeBus]] = None,
+                 config: Optional[Dict] = None):
+        config = config or Configuration().get("skills", {}).get("fallbacks", {})
+        super().__init__(bus, config)
         self.registered_fallbacks = {}  # skill_id: priority
         self.bus.on("ovos.skills.fallback.register", self.handle_register_fallback)
         self.bus.on("ovos.skills.fallback.deregister", self.handle_deregister_fallback)
-        super().__init__(self.fallback_config)
 
     def handle_register_fallback(self, message: Message):
         skill_id = message.data.get("skill_id")
         priority = message.data.get("priority") or 101
 
         # check if .conf is overriding the priority for this skill
-        priority_overrides = self.fallback_config.get("fallback_priorities", {})
+        priority_overrides = self.config.get("fallback_priorities", {})
         if skill_id in priority_overrides:
             new_priority = priority_overrides.get(skill_id)
             LOG.info(f"forcing {skill_id} fallback priority from {priority} to {new_priority}")
@@ -71,12 +73,12 @@ def _fallback_allowed(self, skill_id: str) -> bool:
         Returns:
             permitted (bool): True if skill can fallback
         """
-        opmode = self.fallback_config.get("fallback_mode", FallbackMode.ACCEPT_ALL)
+        opmode = self.config.get("fallback_mode", FallbackMode.ACCEPT_ALL)
         if opmode == FallbackMode.BLACKLIST and skill_id in \
-                self.fallback_config.get("fallback_blacklist", []):
+                self.config.get("fallback_blacklist", []):
             return False
         elif opmode == FallbackMode.WHITELIST and skill_id not in \
-                self.fallback_config.get("fallback_whitelist", []):
+                self.config.get("fallback_whitelist", []):
             return False
         return True
 
@@ -147,7 +149,7 @@ def attempt_fallback(self, utterances: List[str], skill_id: str, lang: str, mess
                                     "lang": lang})
             result = self.bus.wait_for_response(fb_msg,
                                                 f"ovos.skills.fallback.{skill_id}.response",
-                                                timeout=self.fallback_config.get("max_skill_runtime", 10))
+                                                timeout=self.config.get("max_skill_runtime", 10))
             if result and 'error' in result.data:
                 error_msg = result.data['error']
                 LOG.error(f"{skill_id}: {error_msg}")
@@ -202,21 +204,43 @@ def _fallback_range(self, utterances: List[str], lang: str,
                                      utterance=utterances[0])
         return None
 
-    def high_prio(self, utterances: List[str], lang: str, message: Message) -> Optional[PipelineMatch]:
+    def match_high(self, utterances: List[str], lang: str, message: Message) -> Optional[PipelineMatch]:
         """Pre-padatious fallbacks."""
         return self._fallback_range(utterances, lang, message,
                                     FallbackRange(0, 5))
 
-    def medium_prio(self, utterances: List[str], lang: str, message: Message) -> Optional[PipelineMatch]:
+    def match_medium(self, utterances: List[str], lang: str, message: Message) -> Optional[PipelineMatch]:
         """General fallbacks."""
         return self._fallback_range(utterances, lang, message,
                                     FallbackRange(5, 90))
 
-    def low_prio(self, utterances: List[str], lang: str, message: Message) -> Optional[PipelineMatch]:
+    def match_low(self, utterances: List[str], lang: str, message: Message) -> Optional[PipelineMatch]:
         """Low prio fallbacks with general matching such as chat-bot."""
         return self._fallback_range(utterances, lang, message,
                                     FallbackRange(90, 101))
 
+    @deprecated("'low_prio' has been renamed to 'match_low'", "2.0.0")
+    def low_prio(self, utterances: List[str], lang: str, message: Message = None) -> Optional[PipelineMatch]:
+        return self.match_low(utterances, lang, message)
+
+    @deprecated("'medium_prio' has been renamed to 'match_medium'", "2.0.0")
+    def medium_prio(self, utterances: List[str], lang: str, message: Message = None) -> Optional[PipelineMatch]:
+        return self.match_medium(utterances, lang, message)
+
+    @deprecated("'high_prio' has been renamed to 'match_high'", "2.0.0")
+    def high_prio(self, utterances: List[str], lang: str, message: Message = None) -> Optional[PipelineMatch]:
+        return self.match_high(utterances, lang, message)
+
+    @property
+    def fallback_config(self) -> Dict:
+        log_deprecation("'self.fallback_config' is deprecated, access 'self.config' directly instead", "1.0.0")
+        return self.config
+
+    @fallback_config.setter
+    def fallback_config(self, val):
+        log_deprecation("'self.fallback_config' is deprecated, access 'self.config' directly instead", "1.0.0")
+        self.config = val
+
     def shutdown(self):
         self.bus.remove("ovos.skills.fallback.register", self.handle_register_fallback)
         self.bus.remove("ovos.skills.fallback.deregister", self.handle_deregister_fallback)

ovos_core/intent_services/stop_service.py

@@ -2,29 +2,32 @@
 import re
 from os.path import dirname
 from threading import Event
-from typing import Optional, List
+from typing import Optional, Dict, List, Union
 
 from langcodes import closest_match
-
+from ovos_bus_client.client import MessageBusClient
 from ovos_bus_client.message import Message
 from ovos_bus_client.session import SessionManager
+
 from ovos_config.config import Configuration
-from ovos_plugin_manager.templates.pipeline import PipelineMatch, PipelinePlugin
+from ovos_plugin_manager.templates.pipeline import PipelineMatch, PipelineStageConfidenceMatcher
 from ovos_utils import flatten_list
 from ovos_utils.bracket_expansion import expand_options
+from ovos_utils.fakebus import FakeBus
 from ovos_utils.lang import standardize_lang_tag
-from ovos_utils.log import LOG
+from ovos_utils.log import LOG, deprecated
 from ovos_utils.parse import match_one
 
 
-class StopService(PipelinePlugin):
+class StopService(PipelineStageConfidenceMatcher):
     """Intent Service thats handles stopping skills."""
 
-    def __init__(self, bus):
-        self.bus = bus
+    def __init__(self, bus: Optional[Union[MessageBusClient, FakeBus]] = None,
+                 config: Optional[Dict] = None):
+        config = config or Configuration().get("skills", {}).get("stop") or {}
+        super().__init__(config=config, bus=bus)
         self._voc_cache = {}
         self.load_resource_files()
-        super().__init__(config=Configuration().get("skills", {}).get("stop") or {})
 
     def load_resource_files(self):
         base = f"{dirname(__file__)}/locale"
@@ -113,7 +116,7 @@ def stop_skill(self, skill_id: str, message: Message) -> bool:
         elif result is not None:
             return result.data.get('result', False)
 
-    def match_stop_high(self, utterances: List[str], lang: str, message: Message) -> Optional[PipelineMatch]:
+    def match_high(self, utterances: List[str], lang: str, message: Message) -> Optional[PipelineMatch]:
         """If utterance is an exact match for "stop" , run before intent stage
 
         Args:
@@ -161,7 +164,7 @@ def match_stop_high(self, utterances: List[str], lang: str, message: Message) ->
                                          utterance=utterance)
         return None
 
-    def match_stop_medium(self, utterances: List[str], lang: str, message: Message) -> Optional[PipelineMatch]:
+    def match_medium(self, utterances: List[str], lang: str, message: Message) -> Optional[PipelineMatch]:
         """ if "stop" intent is in the utterance,
         but it contains additional words not in .intent files
 
@@ -187,21 +190,9 @@ def match_stop_medium(self, utterances: List[str], lang: str, message: Message)
             if not is_global_stop:
                 return None
 
-        return self.match_stop_low(utterances, lang, message)
+        return self.match_low(utterances, lang, message)
 
-    def _get_closest_lang(self, lang: str) -> Optional[str]:
-        if self._voc_cache:
-            lang = standardize_lang_tag(lang)
-            closest, score = closest_match(lang, list(self._voc_cache.keys()))
-            # https://langcodes-hickford.readthedocs.io/en/sphinx/index.html#distance-values
-            # 0 -> These codes represent the same language, possibly after filling in values and normalizing.
-            # 1- 3 -> These codes indicate a minor regional difference.
-            # 4 - 10 -> These codes indicate a significant but unproblematic regional difference.
-            if score < 10:
-                return closest
-        return None
-
-    def match_stop_low(self, utterances: List[str], lang: str, message: Message) -> Optional[PipelineMatch]:
+    def match_low(self, utterances: List[str], lang: str, message: Message) -> Optional[PipelineMatch]:
         """ before fallback_low , fuzzy match stop intent
 
         Args:
@@ -247,6 +238,18 @@ def match_stop_low(self, utterances: List[str], lang: str, message: Message) ->
                              skill_id=None,
                              utterance=utterance)
 
+    def _get_closest_lang(self, lang: str) -> Optional[str]:
+        if self._voc_cache:
+            lang = standardize_lang_tag(lang)
+            closest, score = closest_match(lang, list(self._voc_cache.keys()))
+            # https://langcodes-hickford.readthedocs.io/en/sphinx/index.html#distance-values
+            # 0 -> These codes represent the same language, possibly after filling in values and normalizing.
+            # 1- 3 -> These codes indicate a minor regional difference.
+            # 4 - 10 -> These codes indicate a significant but unproblematic regional difference.
+            if score < 10:
+                return closest
+        return None
+
     def voc_match(self, utt: str, voc_filename: str, lang: str,
                   exact: bool = False):
         """
@@ -287,3 +290,15 @@ def voc_match(self, utt: str, voc_filename: str, lang: str,
                 return any([re.match(r'.*\b' + i + r'\b.*', utt)
                             for i in _vocs])
         return False
+
+    @deprecated("'match_stop_low' has been renamed to 'match_low'", "2.0.0")
+    def match_stop_low(self, utterances: List[str], lang: str, message: Message = None) -> Optional[PipelineMatch]:
+        return self.match_low(utterances, lang, message)
+
+    @deprecated("'match_stop_medium' has been renamed to 'match_medium'", "2.0.0")
+    def match_stop_medium(self, utterances: List[str], lang: str, message: Message = None) -> Optional[PipelineMatch]:
+        return self.match_medium(utterances, lang, message)
+
+    @deprecated("'match_stop_high' has been renamed to 'match_high'", "2.0.0")
+    def match_stop_high(self, utterances: List[str], lang: str, message: Message = None) -> Optional[PipelineMatch]:
+        return self.match_high(utterances, lang, message)

test/end2end/minicroft.py

@@ -18,27 +18,13 @@ def __init__(self, skill_ids, *args, **kwargs):
         bus = FakeBus()
         super().__init__(bus, *args, **kwargs)
         self.skill_ids = skill_ids
-        self.intent_service = self._register_intent_services()
+        self.intent_service = IntentService(self.bus)
         self.scheduler = EventScheduler(bus, schedule_file="/tmp/schetest.json")
 
     def load_metadata_transformers(self, cfg):
         self.intent_service.metadata_plugins.config = cfg
         self.intent_service.metadata_plugins.load_plugins()
 
-    def _register_intent_services(self):
-        """Start up the all intent services and connect them as needed.
-
-        Args:
-            bus: messagebus client to register the services on
-        """
-        service = IntentService(self.bus)
-        # Register handler to trigger fallback system
-        self.bus.on(
-            'mycroft.skills.fallback',
-            FallbackSkill.make_intent_failure_handler(self.bus)
-        )
-        return service
-
     def load_plugin_skills(self):
         LOG.info("loading skill plugins")
         plugins = find_skill_plugins()

test/end2end/routing/test_session.py

@@ -3,7 +3,7 @@
 from unittest import TestCase
 from ovos_utils.ocp import PlayerState, MediaState
 from ocp_pipeline.opm import OCPPlayerProxy
-
+from ovos_plugin_manager.pipeline import OVOSPipelineFactory
 from ovos_bus_client.message import Message
 from ovos_bus_client.session import SessionManager, Session
 from ..minicroft import get_minicroft
@@ -93,7 +93,7 @@ def tearDown(self) -> None:
         self.core.stop()
 
     def test_no_session(self):
-        self.assertIsNotNone(self.core.intent_service._ocp)
+        self.assertIsNotNone(OVOSPipelineFactory._CACHE.get("ovos-ocp-pipeline-plugin-high"))
         messages = []
 
         def new_msg(msg):
@@ -121,7 +121,8 @@ def wait_for_n_messages(n):
                            "converse",
                            "ocp_high"
                        ])
-        self.core.intent_service._ocp.ocp_sessions[sess.session_id] = OCPPlayerProxy(
+
+        OVOSPipelineFactory._CACHE["ovos-ocp-pipeline-plugin-high"].ocp_sessions[sess.session_id] = OCPPlayerProxy(
             session_id=sess.session_id, available_extractors=[], ocp_available=True,
             player_state=PlayerState.STOPPED, media_state=MediaState.NO_MEDIA)
         utt = Message("recognizer_loop:utterance",

test/unittests/test_intent_service.py

@@ -94,11 +94,11 @@ def on_msg(m):
 
         msg = Message('register_vocab',
                       {'entity_value': 'test', 'entity_type': 'testKeyword'})
-        self.intent_service._adapt_service.handle_register_vocab(msg)
+        self.intent_service.bus.emit(msg)
 
         intent = IntentBuilder('skill:testIntent').require('testKeyword')
         msg = Message('register_intent', intent.__dict__)
-        self.intent_service._adapt_service.handle_register_intent(msg)
+        self.intent_service.bus.emit(msg)
 
     def test_get_intent_no_match(self):
         """Check that if the intent doesn't match at all None is returned."""