描述

手头上有几十个 rtsp 链接，需要接受所有的 rtsp 视频流，并进行相应的处理。现有是使用 ffmpeg-python 库来接受 rtsp 流。由于没有什么好的方法，所以只能用 python 的 Process 对象来管理。

代码

接受 RTSP 的代码

class RTSPStream(Process):
    def __init__(self, queue, stream_id, stream_url):
        Process.__init__(self)
        self.q = q
        self.stream_id = stream_id
        self.stream_url = stream_url

    def get_hnw(self, url):
        probe = ffmpeg.probe(url, rtsp_transport='tcp')
        video_stream = next((stream for stream in probe['streams'] if stream['codec_type'] == 'video'), None)
        width = video_stream['width']
        height = video_stream['height']
        return height, width

    def run(self):
            h, w = get_hnw(self.stream_url)
            assert h != 0
            assert w != 0
            out = (
                ffmpeg
                .input(url, rtsp_transport='tcp')
                .output('pipe:', format='rawvideo', pix_fmt='bgr24', loglevel="quiet", r=1)
                .run_async(pipe_stdout=True)
            )
            cnt_empty = 0
            while True:
                in_bytes = out.stdout.read(h*w*3)
                if not in_bytes:
                    cnt_empty += 1
                    if cnt_empty > 10:
                        break
                    sleep(0.1)
                    continue
                    cnt_empty = 0
                frame = np.frombuffer(in_bytes, dtype=np.uint8).reshape(h, w, 3)
                data = {
                    'stream_id': self.stream_id,
                    'origin': frame
                }
                if not queue.full():
                    self.queue.put(data)

处理代码

class DoSth(Process):
    def __init__(self, queue):
        Process.__init__(self)
        self.queue = queue

    def run(self):
        while True:
            if queue.empty():
                sleep(0.05)
                continue
            data = queue.get()
            # do sth to ndarray