在 MacBook 上使用 Python 作实况视讯串流

本文是山姆锅在学习实况视讯串流 (live video streaming) 过程，用来验证概念(proof of concept) 的纪录。透过 MacBook 内建的镜头作为视讯源，并借由 HTTP Live Streaming (HLS) 协定作实况串流。 虽说是实况，但因为采用 HLS 协定，先天上就会有延迟的。实验的结果不算太成功，本来只能使用桌面环境的 Safari 浏览器来观看视讯，经过高手指正后，现在手机版的也可以了。

何谓 HTTP Live Streaming (HLS)?

HLS 是苹果公司制定，以 HTTP 协定为基础的媒体串流协定，可以支持随选 (Video-on-Demand; VOD) 以及 实况 (live) 模式。其它同样使用 HTTP 作为基础的串流协定，主要的有：

• Adobe HTTP Dynamic Streaming (HDS)
• Microsoft Smooth Streaming (MSS)
• MPEG-DASH

本文选择使用 HLS 纯粹是因为山姆锅比较熟悉。

测试环境

• 主机: MacBook Pro
• OS: OSX 10.10
• CPU: X86-64
• Python: 2.7.10

运行流程

程序共分成发布端 (publisher)、串流端(streamer) 以及回放端 (player) 三个部分，回放端使用的是 OSX 内建的 Safari 浏览器，所以我们只需要有发布端跟串流端即可。

基本流程说明如下：

1. 发布端即时从镜头撷取影像，转码 (encode) 成串流需要的编码与格式(MPEG2 TS)后通知串流端有新的区块(segment)；
2. 串流端根据收到的视讯区块动态产生串流中介数据档(metadata)；
3. 回放端则依照中介数据档来决定该回放的区块。

串流端

串流端在正式系统需要使用其它的伺服软件，如 Nginx。因为只是验证，这里山姆锅使用 gevent + bottle 来作为串流端的技术推叠(technology stack)。

为了要完成 HLS 串流工作，串流端需提供两种数据给回放端：

1. 串流中介数据

   HLS 的中介数据以 m3u8 格式，content type 为: application/x-mpegURL

2. 媒体区段数据

   HLS 的区段须以 MPEG2 TS 格式存放，每个区段一个文件，通常副文件名为 .ts， content type: video/mp2t

底下简单说明串流中介数据，首先看一段实际的内容：

#EXTM3U<br />#EXT-X-VERSION:3<br />#EXT-X-TARGETDURATION:3<br />#EXT-X-MEDIA-SEQUENCE:28<br />#EXTINF:2.250000,<br />http://127.0.0.1:8080/live/out028.ts<br />#EXTINF:1.500000,<br />http://127.0.0.1:8080/live/out029.ts<br />

其中，

• #EXTM3U

  让回放端知道中介数据是以扩充版的 M3U 格式撰写。

• #EXT-X-VERSION:3

  指定此中介数据格式的版本，不支持此版本的回放端无法解读。

• #EXT-X-TARGETDURATION:3

  指定串流中，此叙述之后的视讯区段最长的秒数。本文每个区段接近 2 秒，所以这里指定 3 秒。

• #EXT-X-MEDIA-SEQUENCE:58

  指定中介数据中的第一个区块在整个串流中的序号，没有这个叙述则默认为 0。 因为是实况串流，区块会不断持续产生，如果保留所有过往的区块数据，除了浪费带宽跟性能外， 最终也会导致程序挂点。所以，需要以滚动窗口(rolling window)的方式，只保留最近的区块。

• #EXTINF:1.500000

  每个区块之前都需要有这个声明，其中 1.50000 是此区块的时间长度(以秒为单位)。 这个声明之后的下一行必须是区块文件的 URL 位址，让回放端知道要如何以及去何处撷取区块数据。

• #EXT-X-ENDLIST

  如果是实况串流，了解以上的声明就足够，但对于随选视讯，需要这个声明让回放端知道中介数据结束。 也就是说，只要这个声明没有出现，回放端会假设是实况串流。

关于 HLS 的近一步资讯可以参考 规格文档 。

底下是串流端主要的程序内容(已删减)：

streamer.pyview raw

<br />from __future__ import absolute_import, print_function<br /><br />import os<br />from collections import deque<br /><br />from gevent import monkey; monkey.patch_all()<br />from bottle import route, run, static_file, request, response, hook<br /><br />WEBROOT = os.path.abspath('./webroot')<br /><br />LIVE_MEDIA_FOLDER = os.path.join(WEBROOT, 'live')<br /><br />ROLLING_WINDOW = 10<br />playlist = deque(maxlen=ROLLING_WINDOW)<br /><br />published_segments = 0<br /><br /><br /><br />def ():<br />    response.set_header('Cache-Control', 'no-cache, no-store, must-revalidate')<br />    response.set_header('Pragma', 'no-cache')<br />    response.set_header('Expires', '0')<br /><br /><br />@route('/stream.m3u8')<br />def live_stream_meta():<br />    global playlist<br />    global published_segments<br /><br />    print("Serve playlist")<br />    response.content_type = 'application/x-mpegURL'<br /><br />    result = list()<br />    result.append('#EXTM3Un')<br />    result.append('#EXT-X-VERSION:3n')<br />    <br />    result.append('#EXT-X-TARGETDURATION:3n')<br /><br />    if len(playlist) == 0:<br />        result.append('#EXT-X-MEDIA-SEQUENCE:0n')<br />    else:<br />        sequence = playlist[0][2]<br />        result.append('#EXT-X-MEDIA-SEQUENCE:%dn' % sequence)<br /><br />    for name, duration, sequence in playlist:<br />        result.append('#EXTINF:%s,n' % duration)<br />        result.append('/live/%sn' % name)<br />    # result.append('#EXT-X-ENDLIST')<br />    print(result)<br />    return result<br /><br /><br />@route('/live/<filename>')<br />def live_stream_data(filename):<br />    print("Serve stream data:", filename)<br />    response.content_type = 'video/mp2t'<br />    in_file = os.path.join(LIVE_MEDIA_FOLDER, filename)<br />    with open(in_file) as f:<br />        return f.read()<br /><br /><br />@route('/publish/<filename:path>/<duration>')<br />def publish(filename, duration):<br />    global playlist<br />    global published_segments<br />    playlist.append((filename, duration, published_segments))<br />    print("Published segment:(%s, %s)" % (filename, duration))<br />    published_segments += 1<br /><br /><br /><br />def main():<br />    run(host='0.0.0.0', port=8080, server='gevent')<br /><br /><br />if __name__ == '__main__':<br />    main()<br />

其中，

• live_stream_meta

  用来提供回放端需要的串流中介数据。

• live_stream_data

  用来提供媒体区块数据给回放端。

• publish

  让发布端通知有新的区块产生，发布端须提供文件名以及区块时间长度。

发布端

从实践的角度，发布端其实比较麻烦，由于山姆锅希望使用实况的视讯来源， 自然把脑筋动到 MacBook 内建的镜头身上；另外需要将影像转码成 HLS 串流可以接受的格式 (MPEG2 TS)，一开始还真的不知道如何着手。

针对转码的部分有评估过 GStreamer(因为 Kivy 好像有使用)，但对于要如何组合 pipeline 还真的没有概念，跳过。说到视讯转码，另外的候选当然是鼎鼎大名的 ffmpeg 了！但问题是要使用 哪个 Python 的绑定(binding)？过程就省略，反正最后选择 pyav 这个程序库，如果您有其它更好的选择，请不吝指教。

再来就是影像撷取的问题：一开始还在想 GStreamer, OpenCV 怎么作？后来发现 ffmpeg 就有支持，幸运的是 PyAV 也有提供相关范例：

source = av.open(format='avfoundation', file='0')<br />

其中，`av` 是 PyAV 的套件名称。当然这个只适用在 OSX 环境。

底下是发布端的程序：

publisher.pyview raw

# -*- coding: utf-8 -*-<br />from __future__ import absolute_import, print_function<br /><br />import os<br />import time<br />import urllib2<br />import av<br />import threading<br />from Queue import Queue<br /><br />OUTPUT_FOLDER = os.path.abspath('./webroot/live')<br /><br />interrupted = False<br /><br /><br />class SegmentSubmitter(threading.Thread):<br /><br />    def __init__(self):<br />        super(SegmentSubmitter, self).__init__()<br />        self.queue = Queue()<br />        self.setDaemon(True)<br /><br />    def put_item(self, item):<br />        self.queue.put_nowait(item)<br /><br />    def run(self):<br />        print("Submitter started.")<br />        while True:<br />            item = self.queue.get()<br />            if len(item) == 0:<br />                break<br /><br />            print("Submitting %s" % item[0])<br />            url = 'http://127.0.0.1:8080/publish/%s/%f' % item<br />            try:<br />                content = urllib2.urlopen(url=url).read()<br />            except urllib2.URLError:<br />                pass<br /><br /><br />def gen_segment(filename, source, bit_rate=1000000, vcodec='h264',<br />                pix_fmt='yuv420p', frame_rate=20, duration=2):<br />    global interrupted<br /><br />    out_filename = os.path.join(OUTPUT_FOLDER, filename)<br />    output = av.open(out_filename, 'w')<br /><br />    outs = output.add_stream(vcodec, str(frame_rate))<br />    outs.bit_rate = bit_rate<br />    outs.pix_fmt = pix_fmt<br />    outs.width = 640<br />    outs.height = 480<br />    secs_per_frame = 1.0 / frame_rate<br />    frame_count = 0<br />    segment_start_time = time.time()<br /><br />    while True:<br />        start_time = time.time()<br />        packet = source.next()<br /><br />        for frame in packet.decode():<br />            frame.pts = None<br />            out_packet = outs.encode(frame)<br />            frame_count += 1<br />            if out_packet:<br />                output.mux(out_packet)<br /><br />        if (time.time() - segment_start_time) > duration:<br />            break<br /><br />        time_to_wait = start_time + secs_per_frame - time.time()<br />        if time_to_wait > 0:<br />            try:<br />                time.sleep(time_to_wait)<br />            except KeyboardInterrupt:<br />                interrupted = True<br />                break<br /><br />    while True:<br />        out_packet = outs.encode()<br />        if out_packet:<br />            frame_count += 1<br />            output.mux(out_packet)<br />        else:<br />            break<br /><br />    output.close()<br /><br />    segment_duration = time.time() - segment_start_time<br />    return segment_duration, frame_count<br /><br /><br />def publish(source):<br />    global interrupted<br /><br />    num_segments = 0<br />    submitter = SegmentSubmitter()<br />    submitter.start()<br /><br />    stream = next(s for s in source.streams if s.type == 'video')<br />    it = source.demux(stream)<br /><br />    while not interrupted:<br />        filename = 'seg-%d.ts' % num_segments<br />        print("Generating segment: %s" % filename)<br />        num_segments += 1<br />        duration, frame_count = gen_segment(filename, it)<br />        print("Segment generated: (%s, %f, %d)" %<br />              (filename, duration, frame_count))<br />        submitter.put_item((filename, duration))<br /><br /><br />def main():<br />    source = av.open(format='avfoundation', file='0')<br />    #source = av.open(file='movie.mp4', 'r')<br /><br />    print("Number of streams in source: %d" % len(source.streams))<br /><br />    publish(source)<br /><br />if __name__ == '__main__':<br />    main()<br />

• 共有两个线程在运行，其中一个负责影像撷取并产生区块文件，另一个负责通知串流端有新区块产生。
• 不知道是程序写得没有效率还是怎样，source 的 frame rate 最多只能到每秒 20 帧左右。
• 虽然有根据 frame rate, 来调整撷取的时间间隔以避免影像快转，结果有改善，但似乎还要加强。

使用 Flowplayer 让其它浏览器也可以观看 HLS 串流

除了 Apple 自家的 Safari 外，其它浏览器对于 HLS 的支持上不完整，在这些浏览器需要特别处理。 底下是使用 Flowplayer 的范例:

<!doctype html><br /><br /><head><br /><br />    <link rel="stylesheet" href="player/skin/functional.css"><br /><br />    <!-- site specific styling --><br />    <style><br />    body { font: 12px "Myriad Pro", "Lucida Grande", sans-serif; text-align: center; padding-top: 5%; }<br />    .flowplayer { width: 80%; }<br />    </style><br /><br />    <!-- for video tag based installs flowplayer depends on jQuery 1.7.2+ --><br />    <script src="https://code.jquery.com/jquery-1.11.2.min.js"></script><br /><br />    <!-- include flowplayer --><br />    <script src="player/flowplayer.min.js"></script><br /><br /></head><br /><br /><body><br /><br />    <!-- the player --><br />    <div class="flowplayer" data-swf="/player/flowplayer.swf" data-ratio="0.4167"><br />    <video><br />      <source type="application/x-mpegurl"<br />              src="http://127.0.0.1:8080/stream.m3u8"><br />    </video><br />    </div><br /><br /></body><br />

实际使用会很卡，由于使用 Safari 也会稍微卡卡的，应该是我的程序问题。

结语

本文提供的范例还有不少坑，真的希望有哪位高人能够指导一下。在过程中， 最大的收获竟然是发现 Nginx (透过插件) 已经可以支持多种串流协定!

参考数据

_`Bottle`: http://bottlepy.org/docs/dev/index.html

_`Gevent`: http://www.gevent.org/

_`PyAV`: https://github.com/mikeboers/PyAV

原文引用 大专栏  https://www.dazhuanlan.com/2019/08/27/5d64b831bf08a/

【说明】：本文章由站长整理发布，文章内容不代表本站观点，如文中有侵权行为，请与本站客服联系(QQ:254677821)！