命令行时钟流工具

0x00. 什么是时钟流

在直播领域，通常会使用一路画面中包含当前精确时间画面的视频流结合推送端（主播）和播放端（观众）的画面来判断流的传播延迟，用于问题定位和调优。这样一路特殊的视频流叫做时钟流。

那么在通常使用的过程中，如果生成一路时钟流呢？最简单的办法是点开windows右下角的时间窗口，捕获整个桌面的画面并使用推流工具推流，就得到了一路时钟流。

0x01. 命令行生成时钟流

如同第一节中所说，可以非常简单的生成一路时钟流，为什么还需要一个命令行工具来生成时钟流呢？

其实关于第一段的描述中，不难发现上述一路时钟流推送虽然非常简单，但同样有一个非常大的限制条件——必须有一个可以捕获画面的图形界面。这个限制在日常工作中是非常让人头疼的。在一些实际场景中，可能会有某个地区的主播反馈视频延迟高，那么为了模拟这个主播实际推流的情况，就需要在主播所在地进行模拟推流测试。不可能每次都麻烦主播进行推流测试。更何况往往问题定位和调优都是需要花费较长时间的，作为开发也希望能更方便地使用这项能力。

因此，在实际的运营中，通常会使用不同地区的服务器充当测试机进行推流测试，然而，线上的服务器都是没有图形界面的Linux服务器，那么就必须要有一款能在命令行界面下正常生成时钟流并且进行推流的工具来协助工作。

需要解决的问题就可以分为两个步骤，其一是如何在命令行下生成一路时钟流，其二是如何将这路时钟流推送出去。

0x02. 时钟流的生成

想要得到一路时钟流，在纯命令行的系统里面，是不能奢望通过捕获的方式获得一路流了。那么就必须通过程序生成的方式获取一路流。

如何生成一路流呢？进一步思考，可以想到，其实一路流的本质就是按照一定的速度（帧率）将画面推送出去，那么只要能按照规定的帧率生成足够数量的单幅画面并且保证画面上包含时间信息就OK了。

那么现在只需要解决生成单幅画面的问题即可。说到生成单幅画面，最容易想到的方式就是用PIL直接绘图即可。为了中间尽可能减少处理过程，只用生成黑色背景白色文字的二值化图像即可。

from PIL import Image, ImageDraw, ImageFont
from datetime import datetime

def gen_clock_image():
     im = Image.new('1', (800, 600))
     draw = ImageDraw.ImageDraw(im)
     draw.text((100, 100), str(datetime.now()), fill='white', font=ImageFont.truetype('arial.ttf', 30))
     return im

im = gen_clock_image()
im.show()

于是就可以得到下面的图片了

那么接下来的工作就是需要按照一定时间来生成图片序列即可。

0X03. 时钟流的推送

命令行推流的工具其实没有什么好纠结的，当然只有大名鼎鼎的ffmpeg。只要在启动的时候进行推流参数的设置即可。需要注意如下的几个参数：

y: 覆盖输出
f: 第一个是输入文件格式，通常会自动根据文件来判断，由于生成的是二值化的图片，不带有任何头信息所以选择rawvideo来指定输入格式。
c:v: 输入视频编码，设置rawvideo
pix_fmt: 像素格式，monob，和PIL里的1一致，都代表一个像素1bit的二值化像素排列。
s: 输入画面大小，800x600。
r: 输入帧率，30fps。
i: 输入数据地址，-，在程序中通过pipe将数据通过stdin输入。所以这里用-。
c:v: 输出视频编码，设置libx264
s: 输入画面大小，800x600。
r: 输入帧率，30fps。
f: 输出格式，flv。
preset: x264的参数主要调节编码速度和质量的平衡，为了中间处理损耗小，选择最快的ultrafast。
pix_fmt: 输出像素格式，yuv420p，使用最简单的yuv420p的排列就好。

参数列表差不多就是这些，其中一部分数据在推送的时候通过程序进行设置即可。

0x04. 代码实现

ClockVideo.py

from PIL import Image, ImageDraw, ImageFont
from datetime import datetime
import time
import subprocess


class ClockVideo:
    def __init__(self, width, height, fps, rtmp_path):
        self.width = width
        self.height = height
        self.fps = fps
        self.font = ImageFont.truetype('arial.ttf', 30)
        self.rtmp_path = rtmp_path

    def get_ffmpeg_proc(self):
        command = ['ffmpeg',
                   '-y',
                   '-f', 'rawvideo',
                   '-vcodec', 'rawvideo',
                   '-pix_fmt', 'monob',
                   '-s', f'{self.width}x{self.height}',
                   '-r', str(self.fps),
                   '-i', '-',
                   '-c:v', 'libx264',
                   '-s', f'{self.width}x{self.height}',
                   '-r', str(self.fps),
                   '-f', 'flv',
                   '-preset', 'ultrafast',
                   '-pix_fmt', 'yuv420p',
                   self.rtmp_path
                   ]
        return subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

    def gen_clock_image(self, dt):
        im = Image.new('1', (self.width, self.height))
        draw = ImageDraw.ImageDraw(im)
        draw.text((100, 100), dt, fill='white', font=self.font)
        return im.tobytes()

    def run(self):
        ffmpeg_proc = self.get_ffmpeg_proc()
        real_count = count = 0
        hz = 10 ** 9 / self.fps
        last_print_time_ns = last_output_time_ns = time.time_ns()
        delay_fps = 0
        change_times = 1
        while True:
            dt = str(datetime.now())
            image = self.gen_clock_image(dt)
            real_count += 1
            now_time = time.time_ns()
            if now_time - last_output_time_ns > hz:
            	# 通过管道向ffmpeg写入数据
                ffmpeg_proc.stdin.write(image)
                count += 1
                last_output_time_ns = time.time_ns()
            # 每一秒输出一次统计信息
            if now_time - last_print_time_ns > 10 ** 9:
                print(f'dt: {dt}, gen {real_count}fps, push {count}fps, hz: {hz:.2f}, delay: {delay_fps:.2f}')
                # fps 矫正
                if count < self.fps:
                    delay_fps += 0.2 * change_times
                    hz = 10 ** 9 / (self.fps + delay_fps)
                elif count > self.fps:
                    delay_fps -= 0.1 * change_times
                    hz = 10 ** 9 / (self.fps + delay_fps)
                else:
                    change_times = change_times * 0.1 if change_times >= 0.001 else change_times
                real_count = count = 0
                last_print_time_ns = now_time

main.py

from ClockVideo import ClockVideo

SIZE = (1920, 1080)
FPS = 30
RTMP_URL = ''


if __name__ == '__main__':
    app = ClockVideo(
        *SIZE,
        FPS,
        RTMP_URL
    )
    app.run()

注意：

在同目录下需要自行拷贝一个字体文件，我选择的是arial.ttf，并且修改初始化时候的字体名称。
推送的机器上需要预装ffmpeg和Python3。
需要安装PIL。
RTMP_URL需要替换为具体的推流地址。

最后进行推流测试，可以看到如下的输出：

通过对比终端中打印的时间和拉到流画面上的时间即可知道推流端到播放端延迟情况。

0x05. 小结

这个小工具是在工作中为了解决某个产品海外主播反馈的延迟问题定位时随手开发的。代码很简单，但是可以解决一个非常有趣的实际问题。

更进一步思考，这个工具除了用作推送时钟流进行问题定位之外，还可以结合反向操作为一个直播的延迟监控工具出现。

如何做一个延迟监控工具呢：

通过ffmpeg拉流
解码为二值化的图像输出到管道
Python程序进行图像裁剪并OCR
对推送时间和接收时间做差，上报延迟曲线，设置阈值告警。

不过因为还有很多重要的事情要做，后续的一部分工作就留给感兴趣的同学自己试试吧。