fix:stream extraction (#97)

* feat:stream extraction

use the stream extractors before delegating to audio service

* feat:stream extraction

use the stream extractors before delegating to audio service

* feat:stream extraction

use the stream extractors before delegating to audio service

* autoformat

* str not dict

* tests

* tests

* tests

* typing

.github/workflows/unit_tests.yml

@@ -50,7 +50,7 @@ jobs:
           python -m pip install build wheel
       - name: Install repo
         run: |
-          pip install .
+          pip install .[extras]
       - name: Install test dependencies
         run: |
           pip install -r test/requirements.txt

ovos_audio/audio.py

@@ -12,18 +12,19 @@
 
 import time
 from threading import Lock
+from typing import List, Tuple, Union, Optional
 
+from ovos_audio.utils import require_native_source
 from ovos_bus_client.message import Message
 from ovos_bus_client.message import dig_for_message
 from ovos_config.config import Configuration
 from ovos_plugin_manager.audio import find_audio_service_plugins, \
     setup_audio_service
+from ovos_plugin_manager.ocp import load_stream_extractors
 from ovos_plugin_manager.templates.audio import RemoteAudioBackend
 from ovos_utils.log import LOG
 from ovos_utils.process_utils import MonotonicEvent
 
-from ovos_audio.utils import require_native_source
-
 try:
     from ovos_utils.ocp import MediaState
 except ImportError:
@@ -385,7 +386,19 @@ def restore_volume(msg=message):
         else:
             LOG.debug("No audio service to restore volume of")
 
-    def play(self, tracks, prefered_service, repeat=False):
+    def _extract(self, tracks: Union[List[str], List[Tuple[str, str]]]) -> List[str]:
+        """convert uris into real streams that can be played, eg. handle youtube urls"""
+        xtracted = []
+        xtract = load_stream_extractors()  # @lru_cache, its a lazy loaded singleton
+        for t in tracks:
+            if isinstance(t, str):
+                xtracted.append(xtract.extract_stream(t, video=False)["uri"])
+            else:  # (uri, mime)
+                xtracted.append(xtract.extract_stream(t[0], video=False)["uri"])
+        return xtracted
+
+    def play(self, tracks: Union[List[str], List[Tuple[str, str]]],
+             prefered_service: Optional[str], repeat: bool =False):
         """
             play starts playing the audio on the prefered service if it
             supports the uri. If not the next best backend is found.
@@ -405,6 +418,8 @@ def play(self, tracks, prefered_service, repeat=False):
 
         LOG.debug(f"track uri type: {uri_type}")
 
+        tracks = self._extract(tracks)  # ensure playable streams
+
         # check if user requested a particular service
         if prefered_service and uri_type in prefered_service.supported_uris():
             selected_service = prefered_service

requirements/extras.txt

@@ -1,5 +1,9 @@
+# TTS Plugins
 ovos-tts-plugin-server>=0.0.2, <1.0.0
+
+# Media Playback plugins
 ovos_audio_plugin_simple>=0.1.0, <1.0.0
+ovos-audio-plugin-mpv>=0.0.1, <1.0.0
 ovos_plugin_common_play>=0.0.7, <1.0.0
 
 # OCP plugins

test/unittests/test_end2end.py

@@ -87,6 +87,69 @@ def wait_for_n_messages(n):
         self.assertEqual(state.data["state"], MediaState.LOADED_MEDIA)
         state = messages[6]
         self.assertEqual(state.data["state"], TrackState.PLAYING_AUDIOSERVICE)
+        # confirm stream has been loaded into plugin
+        self.assertEqual(self.core.current._now_playing, "http://fake.mp3")
+
+    def test_youtube(self):
+
+        url = "https://www.youtube.com/watch?v=zc-R6ahuB-8&pp=ygULT3BlblZvaWNlT1M%3D"
+
+        def x_t(u):
+            extracted = []
+            for t in u:
+                if "youtube.com" in t:
+                    t = f"https://NOT-{t}"
+                extracted.append(t)
+            return extracted
+
+        real_x_t = self.core._extract
+
+        self.core._extract = x_t
+        messages = []
+
+        def new_msg(msg):
+            nonlocal messages
+            m = Message.deserialize(msg)
+            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)
+
+        utt = Message('mycroft.audio.service.play',
+                      {"tracks": [url]},
+                      {})
+        self.core.bus.emit(utt)
+
+        # confirm all expected messages are sent
+        expected_messages = [
+            'mycroft.audio.service.play',
+            "ovos.common_play.media.state",  # LOADING_MEDIA
+            "ovos.common_play.track.state",  # QUEUED_AUDIOSERVICE
+            "ovos.common_play.simple.play",  # call simple plugin
+            "ovos.common_play.player.state",  # PLAYING
+            "ovos.common_play.media.state",  # LOADED_MEDIA
+            "ovos.common_play.track.state",  # PLAYING_AUDIOSERVICE
+        ]
+        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])
+
+        # confirm stream has been extracted
+        self.assertNotEqual(self.core.current._now_playing, url)
+
+
+        self.core._extract = real_x_t
 
     def test_uri_error(self):
 

test/unittests/test_service.py

@@ -17,6 +17,7 @@
 from ovos_bus_client import Message
 from ovos_audio.audio import AudioService
 
+from ovos_utils.fakebus import FakeBus
 from .services.working import WorkingBackend
 """Tests for Audioservice class"""
 
@@ -58,6 +59,17 @@ def setup_mock_backends(mock_load_services, emitter):
     return backend, second_backend
 
 
+@unittest.skip("TODO - implement without using youtube plugin, it is blocked in github actions")
+class TestStreamExtract(unittest.TestCase):
+
+    def test_xtract(self):
+        """Test shutdown of audio backend."""
+        url = "https://www.youtube.com/watch?v=zc-R6ahuB-8&pp=ygULT3BlblZvaWNlT1M%3D"
+        service = AudioService(FakeBus())
+        a = service._extract([url])
+        self.assertNotEqual(a[0], url)
+
+
 @unittest.skip("TODO - the mocks no longer apply, rewrite tests")
 class TestService(unittest.TestCase):
     emitter = MockEmitter()