#! /usr/bin/env python3 import os import sys import requests import SoapySDR as soapy from radio import Radio from tuuube import Tuuube from fileradio import FileRadio from flask import Flask, jsonify from flasgger import Swagger app = Flask(__name__) swag = Swagger(app) radios = {} @app.route('/report') def report(): """Get streams report from the RTSP relay. --- responses: 200: description: JSON """ try: r = requests.get("http://localhost:9997/v1/paths/list") j = r.json() for item in j['items']: del j['items'][item]['conf'] del j['items'][item]['confName'] return jsonify(j) except: return "Shat the bed", 500 @app.route('/radio/report') def radio_report(): """List radio devices available to the system. --- responses: 200: description: A list of radio devices. """ devices = [dict(device) for device in soapy.Device.enumerate()] return jsonify(devices) @app.route('/radio//connect') def connect(radio): """Connect to a radio device, by driver name or serial number. --- parameters: - name: radio description: Radio device driver name, or serial number. in: path type: string required: true responses: 200: description: Successfully connected to a radio. 400: description: No radio device by that name is available. 500: description: Failed to connect to radio. """ if radio in radios: return "Radio device already connected", 400 devices = [dict(device) for device in soapy.Device.enumerate()] for device in devices: if radio in device.values(): try: radios[radio] = Radio(radio, device) return "", 200 except Exception as e: radios.pop(radio) return str(e), 500 return "Radio device not found", 400 @app.route('/radio//disconnect') def disconnect(radio): """Disconnect from a radio device. --- parameters: - name: radio description: Radio device driver name, or serial number. in: path type: string required: true responses: 200: description: Successfully connected to a radio. 400: description: No radio device by that name is available. 500: description: An unknown error occurred. """ if radio in radios: radios.pop(radio) return "", 200 else: return "Radio not connected", 400 @app.route('/radio//configure/') def configure(radio, frequency): """Tune the radio to a frequency. You must connect to the radio before attempting to configure it. The radio must be configured before streaming can be started. --- parameters: - name: radio description: Radio device driver name, or serial number. in: path type: string required: true - name: frequency description: Frequency (in Hz) to tune the radio to. Suffixing is also supported, ie 87.5M, 108.0M, etc. in: path type: string required: true responses: 200: description: JSON 400: description: The specified radio is not connected. """ if radio in radios: return jsonify(radios[radio].configure(frequency)) else: return "Radio not connected", 400 @app.route('/radio//start') def start_stream(radio): """Start the radio stream. Once the stream has been started, connect to the stream at: rtsp://[host]:8554/radio/[radio] --- parameters: - name: radio description: Radio device driver name, or serial number. in: path type: string required: true """ try: radios[radio].start_stream() return "", 200 except Exception as e: return str(e), 400 @app.route('/radio//end') def end_stream(radio): """Terminate the radio stream. --- parameters: - name: radio description: Radio device driver name, or serial number. in: path type: string required: true """ try: radios[radio].end_stream() return "", 200 except Exception as e: return str(e), 400 @app.route('/radio//info') def radio_info(radio): """Get information about a radio. --- parameters: - name: radio description: Radio device driver name, or serial number. in: path type: string required: true responses: 200: description: JSON 400: description: The specified radio is not connected. """ try: return jsonify(radios[radio].get_info()) except Exception as e: return str(e), 400 tubes = {} @app.route('/tuuube//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`.\n Other good options are - \n `BaW_jenozKc`, yt_dlp package test video.\n `b2je8uBlgFM`, stereo audio test.\n `LDU_Txk06tM`, crab rave, a commonly used audio fidelity test.\n `sPT_epMLkwQ`, Kilsyth CFA major factory fire dispatch radio traffic. 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//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 rtsp_relay = subprocess.Popen( [ './dependencies/mediamtx/mediamtx', './mediamtx.yml' ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL ) for path, _, files in os.walk('./data/sampleaudio'): for file in files: name,ext = os.path.splitext(file) if ext == '.mp3': tubes[name] = FileRadio(f"{path}/{file}") tubes[name].start_stream() app.run( host='0.0.0.0', threaded=True, debug=False ) print('Stopping any currently streaming radios...') for radio in radios: if radios[radio].is_streaming(): 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?