rtsp2http

#!/usr/bin/env python
# encoding: utf-8
import time
import sys
import threading
import tornado.web
import tornado.ioloop
import tornado.gen
from functools import partial
from tornado.log import app_log as log
from tornado.options import define, options

import os
import re
import tempfile
import uuid
import subprocess


define('url', type=str, default="")
define('protocol', default='tcp')
define('port', default=8888, type=int)
define('address', default='127.0.0.1')
define('chunk_size', default=100, type=int)
define('no_audio', default=False, type=bool)


class HTTPHandler(tornado.web.RequestHandler):
    CLIENTS = set([])
    INFO = []

    def initialize(self):
        self.CLIENTS.add(self)
        self.lock = True
        self.alive = True
        self.count = 0

    def finish(self, *args, **kwargs):
        self.alive = False
        self.CLIENTS.remove(self)
        return super(HTTPHandler, self).finish(*args, **kwargs)

    def on_connection_close(self, *args, **kwargs):
        self.finish()
        return super(HTTPHandler, self).on_connection_close(*args, **kwargs)

    @tornado.gen.coroutine
    def write_media(self, frame):
        if not self.lock and self.alive:
            self.write(frame)
            yield self.flush()

    @tornado.web.asynchronous
    def get(self):
        codecs = ",".join("{0[codec]}.{0[pid]}".format(i) for i in self.INFO)
        self.set_header('Content-Type', 'video/mp2t;codecs="%s"' % codecs)
        self.flush()
        self.lock = False


def Pusher():
    ioloop = tornado.ioloop.IOLoop.instance()
    while True:
        frame = yield
        try:
            for client in HTTPHandler.CLIENTS:
                ioloop.add_callback(partial(client.write_media, frame))
        except Exception as e:
            log.exception(e)


if __name__ == '__main__':
    options.parse_command_line()

    stream_parser = re.compile(r'\s*Stream\s#\d+\:(?P<pid>\d+)\:\s\w+\:\s(?P<codec>\S+)')

    def worker():
        env = dict()
        env['PATH'] = os.environ.get('PATH', '')
        env['TERM'] = 'vt100'

        while True:
            cmd = ("avconv", "-rtsp_transport", options.protocol, "-i", str(options.url))
            cmd += ("-c:v", "copy")
            if options.no_audio:
                cmd += ("-map", "0:0")
                cmd += ("-f", "mpegts", '-streamid', '0:0', "-")
            else:
                cmd += ("-map", "0", "-c:a", "copy")
                cmd += ("-f", "mpegts", '-streamid', '0:0', '-streamid', '1:1', "-")

            process = subprocess.Popen(
                cmd,
                stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                shell=False, env=env, cwd=tempfile.gettempdir()
            )

            def get_info(process):
                info = ''
                payload = False
                data = process.stderr.readline()
                out = list()

                while data:
                    if "Output " in data:
                        break

                    info += data
                    data = process.stderr.readline()

                buf = ''
                for line in info.split('\n'):
                    if 'Audio' in line and options.no_audio:
                        continue

                    if 'Stream ' in line:
                        m = stream_parser.match(line)
                        if m is not None:
                            out.append(m.groupdict())

                return out

            HTTPHandler.INFO = get_info(process)

            pusher = Pusher()
            pusher.next()
            ioloop = tornado.ioloop.IOLoop.instance()
            buff = ''
            retcode = process.poll()
            while retcode is None:
                retcode = process.poll()
                chunk = process.stdout.read(188 * options.chunk_size)

                if not chunk:
                    continue

                if not chunk[0] == '\x47':
                    print (repr(chunk))

                pusher.send(chunk)

    t = threading.Thread(target=worker)
    t.daemon = True
    t.start()

    application = tornado.web.Application([
        ("/", HTTPHandler),
    ])

    application.listen(options.port, address=options.address)
    tornado.ioloop.IOLoop.instance().start()