Adding streaming from youtube sources for additional audio streams

microservice.py

@@ -2,6 +2,7 @@
 
 import SoapySDR as soapy
 from radio import Radio
+from tuuube import Tuuube
 
 from flask import Flask, jsonify
 from flasgger import Swagger
@@ -115,7 +116,7 @@ def configure(radio, frequency):
 def start_stream(radio):
     """Start the radio stream.
     Once the stream has been started, connect to the stream at:
-    rtsp://[host]:8554/radio/[radio]/stream
+    rtsp://[host]:8554/radio/[radio]
     ---
     parameters:
       - name: radio
@@ -169,6 +170,50 @@ def radio_info(radio):
         return str(e), 400
 
 
+# Youtube shit
+tubes = {}
+
+@app.route('/tuuube/<id>/start')
+def start_tuuube_stream(id):
+    """Start streaming from a youtube source.
+    Once the stream has been started, connect to the stream at:
+    rtsp://[host]:8554/tuuube/[id]
+    ---
+    parameters:
+      - name: id
+        description: The youtube video ID. That is the part following the `watch?v=` in the URL. For example, dQw4w9WgXcQ.
+        in: path
+        type: string
+        required: true
+    """
+    if id not in tubes:
+        tubes[id] = Tuuube(id)
+
+    try:
+        tubes[id].start_stream()
+        return "", 200
+    except Exception as e:
+        return str(e), 400
+
+@app.route('/tuuube/<id>/end')
+def end_tuuube_stream(id):
+    """Terminate the youtube stream.
+    ---
+    parameters:
+      - name: id
+        description: The youtube video ID.
+        in: path
+        type: string
+        required: true
+    """
+    try:
+        tubes[id].end_stream()
+        return "", 200
+    except Exception as e:
+        return str(e), 400
+
+
+
 if __name__ == '__main__':
     import subprocess
 
@@ -193,6 +238,11 @@ if __name__ == '__main__':
             radios[radio].end_stream()
     radios = None
 
+    for tube in tubes:
+        if tubes[tube].is_streaming():
+            tubes[tube].end_stream()
+    tubes = None
+
     print('Killing RTSP relay...')
     rtsp_relay.kill()
     rtsp_relay.wait() # Necessary?

streamer.py

@@ -0,0 +1,28 @@
+from threading import Thread
+
+def is_alive(subprocess):
+    return (subprocess.poll() is None)
+
+class Streamer:
+    def __init__(self):
+        self.run = False
+        self.thread = None
+
+    def is_streaming(self):
+        return True if (self.thread and self.thread.is_alive()) else False
+
+    def start_stream(self):
+        if self.is_streaming():
+            raise RuntimeError('Stream thread is already running')
+
+        self.run = True
+        self.thread = Thread(target=self._stream_thread, daemon=True, args=())
+        self.thread.start()
+
+    def end_stream(self):
+        if self.thread is None:
+            raise RuntimeError('No stream thread to terminate')
+
+        self.run = False
+        self.thread.join()
+        self.thread = None

tuuube.py

@@ -0,0 +1,86 @@
+#! /usr/bin/env python3
+
+"""
+# Install issues?
+https://stackoverflow.com/a/27481870
+
+https://stackoverflow.com/a/66024755
+sudo apt purge youtube-dl
+sudo pip3 install youtube-dl
+
+https://stackoverflow.com/a/75504772
+python3 -m pip install --force-reinstall https://github.com/yt-dlp/yt-dlp/archive/master.tar.gz
+"""
+
+#import youtube_dl
+import yt_dlp as youtube_dl
+from streamer import Streamer, is_alive
+import subprocess
+import time
+import sys
+
+
+class Tuuube(Streamer):
+    PORT = 8554
+
+    def __init__(self, id):
+        self.id = id
+
+        super().__init__()
+
+    def _stream_path(self):
+        return f":{Tuuube.PORT}/tuuube/{self.id}"
+
+    def _stream_thread(self):
+        ydl_opts = {
+            'format': 'bestaudio/best',
+            'outtmpl': f'/tmp/{self.id}.%(ext)s',
+            'postprocessors': [{
+                'key': 'FFmpegExtractAudio',
+                'preferredcodec': 'mp3',
+                'preferredquality': '192',
+            }],
+        }
+
+        try:
+            with youtube_dl.YoutubeDL(ydl_opts) as ydl:
+                ydl.download([f'https://www.youtube.com/watch?v={self.id}'])
+        except Exception as e:
+            print(f'File sourcing failed, aborting stream. {e}', file=sys.stderr)
+            self.run = False
+            return
+
+        self.playback = subprocess.Popen(
+            [
+                '/usr/bin/ffmpeg', '-re', '-stream_loop', '-1', '-i',
+                f"/tmp/{self.id}.mp3", '-c', 'copy',
+                '-f', 'rtsp', f"rtsp://localhost{self._stream_path()}"
+            ],
+            stdin=subprocess.PIPE,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL
+        )
+
+        while self.run:
+            if not is_alive(self.playback):
+                print('Playback failed, aborting stream.', file=sys.stderr)
+                break
+            time.sleep(0.1)
+
+        self.run = False
+
+        self.playback.kill()
+        self.playback.wait()
+        self.playback = None
+
+
+
+if __name__ == '__main__':
+    #tube = Tuuube('dQw4w9WgXcQ')
+    tube = Tuuube('BaW_jenozKc')
+
+    tube.start_stream()
+
+    while True:
+        print(tube.is_streaming())
+        time.sleep(1)