RTSP RTP Multicast reflector

rtsp-reflector.py

#!/usr/bin/env python3
#
# If there is an existing h264 rtp multicast stream, but an application requires rtsp,
# but does support rtp multicast over rtsp, then this program will help. It's an
# rtsp server which simply tells the client to use the rtp stream encoded in the rtsp URI.
# For example, with "vlc rtsp://127.0.0.1:8090/rtp/239.1.2.50:5004", it'll tell vlc
# that there is an h264 rtp multicast stream at 239.1.2.50:5004.
#
# SPDX-License-Identifier: MIT-0
# Copyright 2021 Daniel Abrecht 

import re
import socket
import traceback
import threading
import socketserver

session_id_next = 0

code_message = {
  100: "Continue",
  200: "OK",
  201: "Created",
  250: "Low on Storage Space",
  300: "Multiple Choices",
  301: "Moved Permanently",
  302: "Moved Temporarily",
  303: "See Other",
  304: "Not Modified",
  305: "Use Proxy",
  400: "Bad Request",
  401: "Unauthorized",
  402: "Payment Required",
  403: "Forbidden",
  404: "Not Found",
  405: "Method Not Allowed",
  406: "Not Acceptable",
  407: "Proxy Authentication Required",
  408: "Request Time-out",
  410: "Gone",
  411: "Length Required",
  412: "Precondition Failed",
  413: "Request Entity Too Large",
  414: "Request-URI Too Large",
  415: "Unsupported Media Type",
  451: "Parameter Not Understood",
  452: "Conference Not Found",
  453: "Not Enough Bandwidth",
  454: "Session Not Found",
  455: "Method Not Valid in This State",
  456: "Header Field Not Valid for Resource",
  457: "Invalid Range",
  458: "Parameter Is Read-Only",
  459: "Aggregate operation not allowed",
  460: "Only aggregate operation allowed",
  461: "Unsupported transport",
  462: "Destination unreachable",
  500: "Internal Server Error",
  501: "Not Implemented",
  502: "Bad Gateway",
  503: "Service Unavailable",
  504: "Gateway Time-out",
  505: "RTSP Version not supported",
  551: "Option not supported"
}

def parse_rtp_location(loc):
  m = re.match('rtsp://[^/]*/rtp/([^/?]*)(:([0-9]*))', loc)
  if not m:
    return None
  return m.group(1,3)

class ThreadedTCPRequestHandler(socketserver.StreamRequestHandler):
  def handle(self):
    try:
      while True:
        method, location, rtspv = re.match(r' *([^ ]+) +(.+) +([^ ]+) *', self.rfile.readline().decode('utf-8')).group(1,2,3)
        headers = {}
        key = None
        value = ''
        while True:
          line = self.rfile.readline().decode('utf-8').rstrip('\n\r')
          if not line or line == '':
            break
          if re.match(r'\s', line):
            if key:
              value += ' ' + line.lstrip()
              headers[key] = value
          else:
            key, value = line.split(':', 1)
            key = key.rstrip()
            value = value.lstrip()
            headers[key] = value
        content = b'';
        if headers.get('Content-Length'):
          content = self.rfile.read(int(headers['Content-Length']))
        rcode, rheaders, rcontent = getattr(self, 'on_' + method.lower(), lambda a,b,c: (400, {}, b''))(location, headers, content)
        if headers.get('CSeq'):
          rheaders['CSeq'] = headers['CSeq']
        if rcontent != b'':
          rheaders['Content-Length'] = len(rcontent)
        response  = b"";
        response += b"RTSP/1.0 " + str(rcode).encode('utf-8') + b" " + code_message[int(rcode)].encode('utf-8') + b"\r\n" 
        response += b"\r\n".join([str(k).encode('utf-8') + b': ' + str(v).encode('utf-8') for k, v in rheaders.items()]) + b"\r\n"
        response += b"\r\n"
        response += rcontent
        self.request.sendall(response)
    except Exception:
      traceback.print_exc()

  def on_options(self, l, h, c):
    return 200, { "Public": "DESCRIBE, SETUP, PLAY" }, b''
  
  def on_describe(self, l, h, c):
    res = parse_rtp_location(l)
    if not res:
      return 404, {}, b''
    addr, port = res
    return 200, {'Content-Type': 'application/sdp'}, b'''\
c=IN IP4 ''' + str(addr).encode('utf-8') + b'''
m=video ''' + str(port).encode('utf-8') + b''' RTP/AVP 96 
a=rtpmap:96 H264/90000
'''

  def on_setup(self, l, h, c):
    res = parse_rtp_location(l)
    if not res:
      return 404, {}, b''
    addr, port = res
    global session_id_next
    session = ++session_id_next
    return 200, {
      'Session': session,
      'Transport': 'RTP/AVP/UDP;multicast;destination='+ str(addr) +';port='+str(port)
    }, b''

  def on_play(self, l, h, c):
    return 200, {"Session": h['Session']}, b''


class ThreadedTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
  pass

if __name__ == "__main__":
    HOST, PORT = "0.0.0.0", 8090
    socketserver.TCPServer.allow_reuse_address = True
    server = ThreadedTCPServer((HOST, PORT), ThreadedTCPRequestHandler)
    with server:
      server.serve_forever()