buffering in parallel, youtube now buffered -> streaming is fallback, currently broken for s3

syng/client.py

@@ -25,7 +25,7 @@ state = {"current": None, "queue": [], "recent": [], "room": None, "server": "",
 @sio.on("skip")
 async def handle_skip():
     logger.info("Skipping current")
-    await state["current"].skip_current()
+    await state["current"].skip_current(state["queue"][0])
 
 
 @sio.on("state")
@@ -55,6 +55,11 @@ async def handle_buffer(data):
     await sio.emit("meta-info", {"uuid": data["uuid"], "meta": meta_info})
 
 
+async def buffer_and_report(entry):
+    meta_info = await sources[entry.source].buffer(entry)
+    await sio.emit("meta-info", {"uuid": entry.uuid, "meta": meta_info})
+
+
 @sio.on("play")
 async def handle_play(data):
     entry = Entry(**data)
@@ -62,9 +67,8 @@ async def handle_play(data):
         f"Playing: {entry.artist} - {entry.title} [{entry.album}] ({entry.source}) for {entry.performer}"
     )
     try:
-        meta_info = await sources[entry.source].buffer(entry)
-        await sio.emit("meta-info", {"uuid": data["uuid"], "meta": meta_info})
         state["current"] = sources[entry.source]
+        asyncio.create_task(buffer_and_report(entry))
         await sources[entry.source].play(entry)
     except Exception:
         print_exc()

syng/sources/common.py

@@ -3,9 +3,9 @@ import asyncio
 
 
 async def play_mpv(
-    video: str, audio: str | None, options
+        video: str, audio: str | None, options: list[str] = list()
 ) -> asyncio.subprocess.Process:
-    args = [*options, video] + ([f"--audio-file={audio}"] if audio else [])
+    args = ["--fullscreen", *options, video] + ([f"--audio-file={audio}"] if audio else [])
 
     mpv_process = asyncio.create_subprocess_exec("mpv", *args)
     return await mpv_process

syng/sources/s3.py

@@ -57,12 +57,12 @@ class S3Source(Source):
         mp3_file = self.downloaded_files[entry.uuid]["mp3"]
 
         self.player = await play_mpv(
-            cdg_file, mp3_file, ["--scale=oversample", "--fullscreen"]
+            cdg_file, mp3_file, ["--scale=oversample"]
         )
 
         await self.player.wait()
 
-    async def skip_current(self) -> None:
+    async def skip_current(self, entry) -> None:
         await self.player.kill()
 
     @async_in_thread

syng/sources/source.py

@@ -29,7 +29,7 @@ class Source:
     async def play(self, entry: Entry) -> None:
         raise NotImplementedError
 
-    async def skip_current(self) -> None:
+    async def skip_current(self, entry: Entry) -> None:
         pass
 
     async def init_server(self) -> None:

syng/sources/youtube.py

@@ -1,6 +1,7 @@
 import asyncio
 import shlex
 from functools import partial
+from threading import Event, Lock
 
 from pytube import YouTube, Search, Channel, innertube
 
@@ -15,12 +16,23 @@ class YoutubeSource(Source):
         super().__init__()
         self.innertube_client = innertube.InnerTube(client="WEB")
         self.channels = config["channels"] if "channels" in config else []
+        self.tmp_dir = config["tmp_dir"] if "tmp_dir" in config else "/tmp/syng"
         self.player: None | asyncio.subprocess.Process = None
+        self.downloaded_files = {}
+        self.masterlock = Lock()
 
     async def get_config(self):
         return {"channels": self.channels}
 
     async def play(self, entry: Entry) -> None:
+
+        if entry.uuid in self.downloaded_files and "video" in self.downloaded_files[entry.uuid]:
+            print("playing locally")
+            video_file = self.downloaded_files[entry.uuid]["video"]
+            audio_file = self.downloaded_files[entry.uuid]["audio"]
+            self.player = await play_mpv(video_file, audio_file)
+        else:
+            print("streaming")
             self.player = await play_mpv(
                 entry.id,
                 None,
@@ -30,9 +42,10 @@ class YoutubeSource(Source):
                     "--fullscreen",
                 ],
             )
+
         await self.player.wait()
 
-    async def skip_current(self) -> None:
+    async def skip_current(self, entry) -> None:        
         await self.player.kill()
 
     @async_in_thread
@@ -121,5 +134,39 @@ class YoutubeSource(Source):
                 pass
         return list_of_videos
 
+    @async_in_thread
+    def buffer(self, entry: Entry) -> dict:
+        print(f"Buffering {entry}")
+        with self.masterlock:
+            if entry.uuid in self.downloaded_files:
+                print(f"Already buffering {entry}")
+                return {}
+            self.downloaded_files[entry.uuid] = {}
+
+        yt = YouTube(entry.id)
+
+        streams = yt.streams
+    
+        video_streams = streams.filter(
+                type="video", 
+                custom_filter_functions=[lambda s: int(s.resolution[:-1]) <= 1080]
+                )
+        audio_streams = streams.filter(only_audio=True)
+
+        best_720_stream = sorted(video_streams, key=lambda s: int(s.resolution[:-1]) + (1 if s.is_progressive else 0))[-1]
+        best_audio_stream = sorted(audio_streams, key=lambda s: int(s.abr[:-4]))[-1]
+
+        print(best_720_stream)
+        print(best_audio_stream)
+
+        if not best_720_stream.is_progressive:
+            self.downloaded_files[entry.uuid]["audio"] = best_audio_stream.download(output_path=self.tmp_dir, filename_prefix=f"{entry.uuid}-audio")
+        else:
+            self.downloaded_files[entry.uuid]["audio"] = None
+
+        self.downloaded_files[entry.uuid]["video"] = best_720_stream.download(output_path=self.tmp_dir, filename_prefix=entry.uuid)
+
+        return {}
+
 
 available_sources["youtube"] = YoutubeSource