feat/legacy_common_play (#457)

* feat/ocp_legacy

restore support for Mycroft CommonPlay skills

enable with
```json
{
  "intents": {
    "legacy_cps": true,
    "pipelines": ["...", "ocp_legacy"]
  }
}
```

* rm debug print

* more representative legacy test skill

.github/workflows/coverage.yml

@@ -33,6 +33,7 @@ jobs:
           pip install ./test/end2end/skill-new-stop
           pip install ./test/end2end/skill-old-stop
           pip install ./test/end2end/skill-fake-fm
+          pip install ./test/end2end/skill-fake-fm-legacy
     - name: Install core repo
       run: |
         pip install -e .[mycroft,deprecated]

.github/workflows/unit_tests.yml

@@ -60,6 +60,7 @@ jobs:
           pip install ./test/end2end/skill-new-stop
           pip install ./test/end2end/skill-old-stop
           pip install ./test/end2end/skill-fake-fm
+          pip install ./test/end2end/skill-fake-fm-legacy
       - name: Install core repo
         run: |
           pip install -e .[mycroft,deprecated]

mycroft/skills/common_play_skill.py

@@ -56,13 +56,13 @@ class CommonPlaySkill(MycroftSkill, ABC):
     is needed.
     """
 
-    def __init__(self, name=None, bus=None):
-        super().__init__(name, bus)
+    def __init__(self, *args, **kwargs):
         self.audioservice = None
+        super().__init__(*args, **kwargs)
         self.play_service_string = None
 
         # "MusicServiceSkill" -> "Music Service"
-        spoken = name or self.__class__.__name__
+        spoken = self.name or self.__class__.__name__
         self.spoken_name = re.sub(r"([a-z])([A-Z])", r"\g<1> \g<2>",
                                   spoken.replace("Skill", ""))
         # NOTE: Derived skills will likely want to override self.spoken_name
@@ -158,7 +158,7 @@ def __handle_play_start(self, message):
         data = message.data.get("callback_data")
 
         # Stop any currently playing audio
-        if self.audioservice.is_playing:
+        if self.audioservice and self.audioservice.is_playing:
             self.audioservice.stop()
         message.context["skill_id"] = self.skill_id
         self.bus.emit(message.forward("mycroft.stop"))
@@ -167,7 +167,7 @@ def __handle_play_start(self, message):
         # "... on the chromecast"
         self.play_service_string = phrase
 
-        self.make_active()
+        self.activate()
 
         # Invoke derived class to provide playback data
         self.CPS_start(phrase, data)
@@ -193,7 +193,7 @@ def CPS_play(self, *args, **kwargs):
 
     def stop(self):
         """Stop anything playing on the audioservice."""
-        if self.audioservice.is_playing:
+        if self.audioservice and self.audioservice.is_playing:
             self.audioservice.stop()
             return True
         else:

ovos_core/intent_services/__init__.py

@@ -252,7 +252,8 @@ def get_pipeline(self, skips=None, session=None):
             matchers.update({
                 "ocp_high": self.ocp.match_high,
                 "ocp_medium": self.ocp.match_medium,
-                "ocp_fallback": self.ocp.match_fallback})
+                "ocp_fallback": self.ocp.match_fallback,
+                "ocp_legacy": self.ocp.match_legacy})
         skips = skips or []
         pipeline = [k for k in session.pipeline if k not in skips]
         if any(k not in matchers for k in pipeline):

ovos_core/intent_services/ocp_service.py

@@ -1,5 +1,6 @@
 import os
 import random
+import time
 from os.path import join, dirname
 from threading import RLock
 from typing import List, Tuple, Optional
@@ -12,6 +13,7 @@
 import ovos_core.intent_services
 from ovos_bus_client.apis.ocp import OCPInterface, OCPQuery, ClassicAudioServiceInterface
 from ovos_bus_client.message import Message
+from ovos_bus_client.util import wait_for_reply
 from ovos_config import Configuration
 from ovos_plugin_manager.ocp import load_stream_extractors
 from ovos_utils import classproperty
@@ -107,6 +109,7 @@ def __init__(self, bus=None, config=None):
 
         self.ocp_api = OCPInterface(self.bus)
         self.legacy_api = ClassicAudioServiceInterface(self.bus)
+        self.mycroft_cps = LegacyCommonPlay(self.bus)
 
         self.config = config or {}
         self.search_lock = RLock()
@@ -204,6 +207,7 @@ def register_ocp_intents(self):
         self.bus.on("ocp:media_stop", self.handle_stop_intent)
         self.bus.on("ocp:search_error", self.handle_search_error_intent)
         self.bus.on("ocp:like_song", self.handle_like_intent)
+        self.bus.on("ocp:legacy_cps", self.handle_legacy_cps)
 
     def handle_get_SEIs(self, message: Message):
         """report available StreamExtractorIds
@@ -919,3 +923,144 @@ def _handle_legacy_audio_end(self, message: Message):
         if self.use_legacy_audio:
             self.player_state = PlayerState.STOPPED
             self.media_state = MediaState.END_OF_MEDIA
+
+    ############
+    # Legacy Mycroft CommonPlay skills
+
+    def match_legacy(self, utterances: List[str], lang: str, message: Message = None):
+        """ match legacy mycroft common play skills  (must import from deprecated mycroft module)
+        not recommended, legacy support only
+
+        legacy base class at mycroft/skills/common_play_skill.py marked for removal in ovos-core 0.1.0
+        """
+        if not self.config.get("legacy_cps"):
+            # needs to be explicitly enabled in pipeline config
+            return None
+
+        utterance = utterances[0].lower()
+
+        match = self.intent_matchers[lang].calc_intent(utterance)
+
+        if match["name"] is None:
+            return None
+        if match["name"] == "play":
+            LOG.info(f"Legacy Mycroft CommonPlay match: {match}")
+            # we dont call self.activate , the skill itself is activated on selection
+            # playback is happening outside of OCP
+            utterance = match["entities"].pop("query")
+            return ovos_core.intent_services.IntentMatch(intent_service="OCP_media",
+                                                         intent_type=f"ocp:legacy_cps",
+                                                         intent_data={"query": utterance,
+                                                                      "conf": 0.7},
+                                                         skill_id=OCP_ID,
+                                                         utterance=utterance)
+
+    def handle_legacy_cps(self, message: Message):
+        """intent handler for legacy CPS matches"""
+        utt = message.data["query"]
+        res = self.mycroft_cps.search(utt)
+        if res:
+            best = self.select_best([r[0] for r in res])
+            if best:
+                callback = [r[1] for r in res if r[0].uri == best.uri][0]
+                self.mycroft_cps.skill_play(skill_id=best.skill_id,
+                                            callback_data=callback,
+                                            phrase=utt,
+                                            message=message)
+                return
+        self.bus.emit(message.forward("mycroft.audio.play_sound",
+                                      {"uri": "snd/error.mp3"}))
+
+
+class LegacyCommonPlay:
+    """ interface for mycroft common play
+    1 - emit 'play:query'
+    2 - gather 'play:query.response' from legacy skills
+    3 - emit 'play:start' for selected skill
+
+    legacy base class at mycroft/skills/common_play_skill.py
+    marked for removal in ovos-core 0.1.0
+    """
+
+    def __init__(self, bus):
+        self.bus = bus
+        self.query_replies = {}
+        self.query_extensions = {}
+        self.waiting = False
+        self.start_ts = 0
+        self.bus.on("play:query.response", self.handle_cps_response)
+
+    def skill_play(self, skill_id: str, callback_data: dict,
+                   phrase: Optional[str] = "",
+                   message: Optional[Message] = None):
+        """tell legacy CommonPlaySkills they were selected and should handle playback"""
+        message = message or Message("ocp:legacy_cps")
+        self.bus.emit(message.forward(
+            'play:start',
+            {"skill_id": skill_id,
+             "phrase": phrase,
+             "callback_data": callback_data}
+        ))
+
+    def shutdown(self):
+        self.bus.remove("play:query.response", self.handle_cps_response)
+
+    @property
+    def cps_status(self):
+        return wait_for_reply('play:status.query',
+                              reply_type="play:status.response",
+                              bus=self.bus).data
+
+    def handle_cps_response(self, message):
+        """receive matches from legacy skills"""
+        search_phrase = message.data["phrase"]
+
+        if ("searching" in message.data and
+                search_phrase in self.query_extensions):
+            # Manage requests for time to complete searches
+            skill_id = message.data["skill_id"]
+            if message.data["searching"]:
+                # extend the timeout by N seconds
+                # IGNORED HERE, used in mycroft-playback-control skill
+                if skill_id not in self.query_extensions[search_phrase]:
+                    self.query_extensions[search_phrase].append(skill_id)
+            else:
+                # Search complete, don't wait on this skill any longer
+                if skill_id in self.query_extensions[search_phrase]:
+                    self.query_extensions[search_phrase].remove(skill_id)
+
+        elif search_phrase in self.query_replies:
+            # Collect all replies until the timeout
+            self.query_replies[message.data["phrase"]].append(message.data)
+
+    def send_query(self, phrase):
+        self.query_replies[phrase] = []
+        self.query_extensions[phrase] = []
+        self.bus.emit(Message('play:query',
+                              {"phrase": phrase}))
+
+    def get_results(self, phrase):
+        if self.query_replies.get(phrase):
+            return [self.cps2media(r) for r in self.query_replies[phrase]]
+        return []
+
+    def search(self, phrase, timeout=5):
+        self.send_query(phrase)
+        self.waiting = True
+        start_ts = time.time()
+        while self.waiting and time.time() - start_ts <= timeout:
+            time.sleep(0.2)
+        self.waiting = False
+        return self.get_results(phrase)
+
+    @staticmethod
+    def cps2media(res: dict, media_type=MediaType.GENERIC) -> Tuple[MediaEntry, dict]:
+        """convert a cps result into a modern result"""
+        entry = MediaEntry(title=res["phrase"],
+                           artist=res["skill_id"],
+                           uri=f"callback:{res['skill_id']}",
+                           media_type=media_type,
+                           playback=PlaybackType.SKILL,
+                           match_confidence=res["conf"] * 100,
+                           skill_id=res["skill_id"])
+        return entry, res['callback_data']

test/end2end/session/test_ocp.py

@@ -155,6 +155,9 @@ def wait_for_n_messages(n):
         for idx, m in enumerate(messages):
             self.assertEqual(m.msg_type, expected_messages[idx])
 
+        play = messages[-1]
+        self.assertEqual(play.data["media"]["uri"], "https://fake_4.mp3")
+
     def test_unk_media_match(self):
         self.assertIsNotNone(self.core.intent_service.ocp)
         self.assertFalse(self.core.intent_service.ocp.use_legacy_audio)
@@ -964,3 +967,132 @@ def wait_for_n_messages(n):
 
             for idx, m in enumerate(messages):
                 self.assertEqual(m.msg_type, expected_messages[idx])
+
+    def test_legacy_cps(self):
+        self.assertIsNotNone(self.core.intent_service.ocp)
+
+        self.core.intent_service.ocp.config = {"legacy_cps": True}
+
+        messages = []
+
+        def new_msg(msg):
+            nonlocal messages
+            m = Message.deserialize(msg)
+            if m.msg_type in ["ovos.skills.settings_changed", "gui.status.request"]:
+                return  # skip these, only happen in 1st run
+            messages.append(m)
+            print(len(messages), msg)
+
+        def wait_for_n_messages(n):
+            nonlocal messages
+            t = time.time()
+            while len(messages) < n:
+                sleep(0.1)
+                if time.time() - t > 10:
+                    raise RuntimeError("did not get the number of expected messages under 10 seconds")
+
+        self.core.bus.on("message", new_msg)
+
+        sess = Session("test-session",
+                       pipeline=[
+                           "ocp_legacy"
+                       ])
+        utt = Message("recognizer_loop:utterance",
+                      {"utterances": ["play rammstein"]},
+                      {"session": sess.serialize(),  # explicit
+                       })
+        self.core.bus.emit(utt)
+
+        # confirm all expected messages are sent
+        expected_messages = [
+            "recognizer_loop:utterance",
+            "ocp:legacy_cps",
+            # legacy cps api
+            "play:query",
+            "mycroft.audio.play_sound"  # error -  no results
+        ]
+        wait_for_n_messages(len(expected_messages))
+
+        self.assertEqual(len(expected_messages), len(messages))
+
+        for idx, m in enumerate(messages):
+            self.assertEqual(m.msg_type, expected_messages[idx])
+
+
+class TestLegacyCPSPipeline(TestCase):
+
+    def setUp(self):
+        self.skill_id = "skill-fake-fm-legacy.openvoiceos"
+        self.core = get_minicroft(self.skill_id, ocp=True)
+        self.core.intent_service.ocp.config = {"legacy_cps": True}
+
+    def tearDown(self) -> None:
+        self.core.stop()
+
+    def test_legacy_cps(self):
+        self.assertIsNotNone(self.core.intent_service.ocp)
+
+        messages = []
+
+        def new_msg(msg):
+            nonlocal messages
+            m = Message.deserialize(msg)
+            if m.msg_type in ["ovos.skills.settings_changed", "gui.status.request"]:
+                return  # skip these, only happen in 1st run
+            messages.append(m)
+            print(len(messages), msg)
+
+        def wait_for_n_messages(n):
+            nonlocal messages
+            t = time.time()
+            while len(messages) < n:
+                sleep(0.1)
+                if time.time() - t > 10:
+                    raise RuntimeError("did not get the number of expected messages under 10 seconds")
+
+        self.core.bus.on("message", new_msg)
+
+        sess = Session("test-session",
+                       pipeline=[
+                           "ocp_legacy"
+                       ])
+        utt = Message("recognizer_loop:utterance",
+                      {"utterances": ["play rammstein"]},
+                      {"session": sess.serialize(),  # explicit
+                       })
+        self.core.bus.emit(utt)
+
+        # confirm all expected messages are sent
+        expected_messages = [
+            "recognizer_loop:utterance",
+            "ocp:legacy_cps",
+            # legacy cps api
+            "play:query",
+            "play:query.response", # searching
+            "play:query.response", # report results
+            "play:start",  # skill selected
+            "mycroft.audio.service.track_info",  # check is legacy audio service is playing
+            # global stop signal
+            "mycroft.stop",
+            "ovos.common_play.stop",
+            "ovos.common_play.stop.response",
+            "skill-fake-fm-legacy.openvoiceos.stop",
+            "skill-fake-fm-legacy.openvoiceos.stop.response",
+            "mycroft.audio.service.track_info",  # check is legacy audio service is playing
+            # activate skill
+            "intent.service.skills.activate",
+            "intent.service.skills.activated",
+            f"{self.skill_id}.activate",
+            # skill callback code
+            "mycroft.audio.service.play"
+        ]
+        wait_for_n_messages(len(expected_messages))
+
+        self.assertEqual(len(expected_messages), len(messages))
+
+        for idx, m in enumerate(messages):
+            self.assertEqual(m.msg_type, expected_messages[idx])
+
+        play = messages[-1]
+        self.assertEqual(play.data["tracks"], ["https://fake.mp3"])
+