FFmpeg配合zmq实现动态处理filter

T-Bagwell

在视频会议场景，或者导播场景，经常会遇到画面的位置动态调整，镜头调整等功能需求，好多人在使用FFmpeg的时候喜欢用API自己写一遍对应的功能，但是实际上ffmpeg的filter部分本身已经支持了zmq的功能，zmq大伙熟知是用来发消息所用，那么FFmpeg在任务启动之后，其实也开了一个zmq的listen端口，可以通过发送消息控制ffmpeg的filter内容，下面看一下基本环境操作：
1. 安装zmq
2. ffmpeg编译的时候带上--enable-libzmq
查看一下确认一下ffmpeg是否已经将zmq编译加再进来：
ffmpeg -h filter=zmq

[AppleScript] 纯文本查看 复制代码

ffmpeg version N-91042-bbs.chinaffmpeg.com悟空专用版 Copyright (c) 2000-2018 the FFmpeg developers
  built with Apple LLVM version 9.1.0 (clang-902.0.39.1)
  configuration: --enable-fontconfig --enable-gpl --enable-libass --enable-libbluray --enable-libfreetype --enable-libmp3lame --enable-libspeex --enable-libx264 --enable-libx265 --enable-libfdk-aac --enable-version3 --cc='ccache gcc' --enable-nonfree --enable-videotoolbox --disable-optimizations --disable-stripping --enable-libzmq ;--enable-bbs.chinaffmpeg.com
  libavutil      56. 18.102 / 56. 18.102
  libavcodec     58. 19.101 / 58. 19.101
  libavformat    58. 13.102 / 58. 13.102
  libavdevice    58.  4.100 / 58.  4.100
  libavfilter     7. 21.101 /  7. 21.101
  libswscale      5.  2.100 /  5.  2.100
  libswresample   3.  2.100 /  3.  2.100
  libpostproc    55.  2.100 / 55.  2.100
Filter zmq
  Receive commands through ZMQ and broker them to filters.
    Inputs:
       #0: default (video)
    Outputs:
       #0: default (video)
zmq AVOptions:
  bind_address      <string>     ..FVA.... set bind address (default "tcp://*:5555")
  b                 <string>     ..FVA.... set bind address (default "tcp://*:5555")

从以上输出内容可以看到已经包含了libzmq.
而ffmpeg的官方文档中提供了对应的操作示例：
参考链接 https://ffmpeg.org/ffmpeg-filters.html#zmq_002c-azmq
内容如下：

[AppleScript] 纯文本查看 复制代码

Receive commands sent through a libzmq client, and forward them to filters in the filtergraph.

zmq and azmq work as a pass-through filters. zmq must be inserted between two video filters, azmq between two audio filters. Both are capable to send messages to any filter type.

To enable these filters you need to install the libzmq library and headers and configure FFmpeg with --enable-libzmq.

For more information about libzmq see: [url=http://www.zeromq.org/]http://www.zeromq.org/[/url]

The zmq and azmq filters work as a libzmq server, which receives messages sent through a network interface defined by the bind_address (or the abbreviation "b") option. Default value of this option is tcp://localhost:5555. You may want to alter this value to your needs, but do not forget to escape any ’:’ signs (see filtergraph escaping).

The received message must be in the form:

TARGET COMMAND [ARG]
TARGET specifies the target of the command, usually the name of the filter class or a specific filter instance name. The default filter instance name uses the pattern ‘Parsed_<filter_name>_<index>’, but you can override this by using the ‘filter_name@id’ syntax (see Filtergraph syntax).

COMMAND specifies the name of the command for the target filter.

ARG is optional and specifies the optional argument list for the given COMMAND.

Upon reception, the message is processed and the corresponding command is injected into the filtergraph. Depending on the result, the filter will send a reply to the client, adopting the format:

ERROR_CODE ERROR_REASON
MESSAGE
MESSAGE is optional.

13.28.1 Examples

Look at tools/zmqsend for an example of a zmq client which can be used to send commands processed by these filters.

Consider the following filtergraph generated by ffplay. In this example the last overlay filter has an instance name. All other filters will have default instance names.

ffplay -dumpgraph 1 -f lavfi "
color=s=100x100:c=red  [l];
color=s=100x100:c=blue [r];
nullsrc=s=200x100, zmq [bg];
[bg][l]   overlay     [bg+l];
[bg+l][r] overlay@my=x=100 "
To change the color of the left side of the video, the following command can be used:

echo Parsed_color_0 c yellow | tools/zmqsend
To change the right side:

echo Parsed_color_1 c pink | tools/zmqsend
To change the position of the right side:

echo overlay@my x 150 | tools/zmqsend

可以根据内容的样例操作一下；

[AppleScript] 纯文本查看 复制代码

ffplay -dumpgraph 1 -f lavfi "
color=s=100x100:c=red [l];
color=s=100x100:c=blue [r];
nullsrc=s=200x100, zmq [bg];
[bg][l] overlay [bg+l];
[bg+l][r] overlay@my=x=100 "

然后可以看到根据dumpgraph输出了filter的信息：



从图中可以看到filter的整体的信息流，好了，示例跑起来了，是一个200x100宽高的视频图像，并且是左边红色右边蓝色：


此时图像已经跑起来，接下来改变图像信息，主要是改变颜色，先将左边的颜色动态改变成黄色：

[AppleScript] 纯文本查看 复制代码

liuqideMacBook-Pro:encryption liuqi$ echo Parsed_color_0 c yellow | tools/zmqsend
0 Undefined error: 0
liuqideMacBook-Pro:encryption liuqi$

看一下效果图：



然后将右边的蓝色改为粉色：

[AppleScript] 纯文本查看 复制代码

liuqideMacBook-Pro:encryption liuqi$ echo Parsed_color_1 c pink | tools/zmqsend
0 Undefined error: 0
liuqideMacBook-Pro:encryption liuqi$

看一下效：



通过这个示例，可以看到ffmpeg可以通过zmq动态改变filter里面的内容，那么接下来改变一下图像坐标，将粉色移动到黄色的上层覆盖掉：

[AppleScript] 纯文本查看 复制代码

liuqideMacBook-Pro:encryption liuqi$ echo overlay@my x 0 | tools/zmqsend
0 Undefined error: 0
liuqideMacBook-Pro:encryption liuqi$

看一下效果：



由于背景色是黑色，所以原来右边的粉色移动到了左边，右边就是黑色背景了

在例子上看到有用到tools/zmqsend，其实tools/zmqsend是ffmpeg项目源代码目录中自带的工具，我将代码copy到这里：

[AppleScript] 纯文本查看 复制代码

/*
 * Copyright (c) 2013 Stefano Sabatini
 *
 * This file is part of FFmpeg.
 *
 * FFmpeg is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * FFmpeg is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with FFmpeg; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 */

#include "config.h"

#include <zmq.h>

#include "libavutil/mem.h"
#include "libavutil/bprint.h"

#if HAVE_UNISTD_H
#include <unistd.h>             /* getopt */
#endif

#if !HAVE_GETOPT
#include "compat/getopt.c"
#endif

/**
 * @file
 * zmq message sender example, meant to be used with the zmq filters
 */

static void usage(void)
{
    printf("send message to ZMQ recipient, to use with the zmq filters\n");
    printf("usage: zmqsend [OPTIONS]\n");
    printf("\n"
           "Options:\n"
           "-b ADDRESS        set bind address\n"
           "-h                print this help\n"
           "-i INFILE         set INFILE as input file, stdin if omitted\n");
}

int main(int argc, char **argv)
{
    AVBPrint src;
    char *src_buf, *recv_buf;
    int c;
    int recv_buf_size, ret = 0;
    void *zmq_ctx, *socket;
    const char *bind_address = "tcp://localhost:5555";
    const char *infilename = NULL;
    FILE *infile = NULL;
    zmq_msg_t msg;

    while ((c = getopt(argc, argv, "b:hi:")) != -1) {
        switch (c) {
        case 'b':
            bind_address = optarg;
            break;
        case 'h':
            usage();
            return 0;
        case 'i':
            infilename = optarg;
            break;
        case '?':
            return 1;
        }
    }

    if (!infilename || !strcmp(infilename, "-")) {
        infilename = "stdin";
        infile = stdin;
    } else {
        infile = fopen(infilename, "r");
    }
    if (!infile) {
        av_log(NULL, AV_LOG_ERROR,
               "Impossible to open input file '%s': %s\n", infilename, strerror(errno));
        return 1;
    }

    zmq_ctx = zmq_ctx_new();
    if (!zmq_ctx) {
        av_log(NULL, AV_LOG_ERROR,
               "Could not create ZMQ context: %s\n", zmq_strerror(errno));
        return 1;
    }

    socket = zmq_socket(zmq_ctx, ZMQ_REQ);
    if (!socket) {
        av_log(NULL, AV_LOG_ERROR,
               "Could not create ZMQ socket: %s\n", zmq_strerror(errno));
        ret = 1;
        goto end;
    }

    if (zmq_connect(socket, bind_address) == -1) {
        av_log(NULL, AV_LOG_ERROR, "Could not bind ZMQ responder to address '%s': %s\n",
               bind_address, zmq_strerror(errno));
        ret = 1;
        goto end;
    }

    /* grab the input and store it in src */
    av_bprint_init(&src, 1, AV_BPRINT_SIZE_UNLIMITED);
    while ((c = fgetc(infile)) != EOF)
        av_bprint_chars(&src, c, 1);
    av_bprint_chars(&src, 0, 1);

    if (!av_bprint_is_complete(&src)) {
        av_log(NULL, AV_LOG_ERROR, "Could not allocate a buffer for the source string\n");
        av_bprint_finalize(&src, NULL);
        ret = 1;
        goto end;
    }
    av_bprint_finalize(&src, &src_buf);

    if (zmq_send(socket, src_buf, strlen(src_buf), 0) == -1) {
        av_log(NULL, AV_LOG_ERROR, "Could not send message: %s\n", zmq_strerror(errno));
        ret = 1;
        goto end;
    }

    if (zmq_msg_init(&msg) == -1) {
        av_log(NULL, AV_LOG_ERROR,
               "Could not initialize receiving message: %s\n", zmq_strerror(errno));
        ret = 1;
        goto end;
    }

    if (zmq_msg_recv(&msg, socket, 0) == -1) {
        av_log(NULL, AV_LOG_ERROR,
               "Could not receive message: %s\n", zmq_strerror(errno));
        zmq_msg_close(&msg);
        ret = 1;
        goto end;
    }

    recv_buf_size = zmq_msg_size(&msg) + 1;
    recv_buf = av_malloc(recv_buf_size);
    if (!recv_buf) {
        av_log(NULL, AV_LOG_ERROR,
               "Could not allocate receiving message buffer\n");
        zmq_msg_close(&msg);
        ret = 1;
        goto end;
    }
    memcpy(recv_buf, zmq_msg_data(&msg), recv_buf_size);
    recv_buf[recv_buf_size-1] = 0;
    printf("%s\n", recv_buf);
    zmq_msg_close(&msg);
    av_free(recv_buf);

end:
    zmq_close(socket);
    zmq_ctx_destroy(zmq_ctx);
    return ret;
}

这个代码编译的方法如下：

[AppleScript] 纯文本查看 复制代码

gcc -Llibavcodec -Llibavdevice -Llibavfilter -Llibavformat -Llibavresample -Llibavutil -Llibpostproc -Llibswscale -Llibswresample  -o tools/zmqsend tools/zmqsend.o -lavfilter -lswscale -lpostproc -lavformat -lavcodec -lswresample -lavutil  -pthread -lm -L/usr/local/lib -lass -L/usr/local/Cellar/zeromq/4.2.5/lib -lzmq  -L/usr/local/Cellar/fontconfig/2.12.6/lib -L/usr/local/opt/freetype/lib -lfontconfig -lfreetype -L/usr/local/opt/freetype/lib -lfreetype -lm -lm -lm -lbz2 -L/usr/local/Cellar/libbluray/1.0.2/lib -lbluray -lz  -liconv -lm -llzma -lz  -L/usr/local/lib -lfdk-aac -lmp3lame -lm -L/usr/local/Cellar/speex/1.2rc1/lib -lspeex -L/usr/local/lib -lx264 -L/usr/local/Cellar/x265/2.5_1/lib -lx265 -pthread  -lm -pthread -lm 

最好还是在FFmpeg源代码编译的时候直接

[AppleScript] 纯文本查看 复制代码

make tools/zmqsend

在FFmpeg的filter源代码有对于filter graph动态处理相关的框架，主要内容大概如下，文件libavfilter/f_zmq.c：

[AppleScript] 纯文本查看 复制代码

static int filter_frame(AVFilterLink *inlink, AVFrame *ref)
{
    AVFilterContext *ctx = inlink->dst;
    ZMQContext *zmq = ctx->priv;

    while (1) {
        char cmd_buf[1024];
        char *recv_buf, *send_buf;
        int recv_buf_size;
        Command cmd = {0};
        int ret;

        /* receive command */
        if (recv_msg(ctx, &recv_buf, &recv_buf_size) < 0)
            break;
        zmq->command_count++;

        /* parse command */
        if (parse_command(&cmd, recv_buf, ctx) < 0) {
            av_log(ctx, AV_LOG_ERROR, "Could not parse command #%d\n", zmq->command_count);
            goto end;
        }

        /* process command */
        av_log(ctx, AV_LOG_VERBOSE,
               "Processing command #%d target:%s command:%s arg:%s\n",
               zmq->command_count, cmd.target, cmd.command, cmd.arg);
        ret = avfilter_graph_send_command(inlink->graph,
                                          cmd.target, cmd.command, cmd.arg,
                                          cmd_buf, sizeof(cmd_buf),
                                          AVFILTER_CMD_FLAG_ONE);
        send_buf = av_asprintf("%d %s%s%s",
                               -ret, av_err2str(ret), cmd_buf[0] ? "\n" : "", cmd_buf);
        if (!send_buf) {
            ret = AVERROR(ENOMEM);
            goto end;
        }
        av_log(ctx, AV_LOG_VERBOSE,
               "Sending command reply for command #%d:\n%s\n",
               zmq->command_count, send_buf);
        if (zmq_send(zmq->responder, send_buf, strlen(send_buf), 0) == -1)
            av_log(ctx, AV_LOG_ERROR, "Failed to send reply for command #%d: %s\n",
                   zmq->command_count, zmq_strerror(ret));

    end:
        av_freep(&send_buf);
        av_freep(&recv_buf);
        recv_buf_size = 0;
        av_freep(&cmd.target);
        av_freep(&cmd.command);
        av_freep(&cmd.arg);
    }

    return ff_filter_frame(ctx->outputs[0], ref);
}

这里调用了接口avfilter_graph_send_command
而avfilter_graph_send_command是是现在libavfilter/avfiltergraph.c中：

[AppleScript] 纯文本查看 复制代码

int avfilter_graph_send_command(AVFilterGraph *graph, const char *target, const char *cmd, const char *arg, char *res, int res_len, int flags)
{
    int i, r = AVERROR(ENOSYS);

    if (!graph)
        return r;

    if ((flags & AVFILTER_CMD_FLAG_ONE) && !(flags & AVFILTER_CMD_FLAG_FAST)) {
        r = avfilter_graph_send_command(graph, target, cmd, arg, res, res_len, flags | AVFILTER_CMD_FLAG_FAST);
        if (r != AVERROR(ENOSYS))
            return r;
    }

    if (res_len && res)
        res[0] = 0;

    for (i = 0; i < graph->nb_filters; i++) {
        AVFilterContext *filter = graph->filters[i];
        if (!strcmp(target, "all") || (filter->name && !strcmp(target, filter->name)) || !strcmp(target, filter->filter->name)) {
            r = avfilter_process_command(filter, cmd, arg, res, res_len, flags);
            if (r != AVERROR(ENOSYS)) {
                if ((flags & AVFILTER_CMD_FLAG_ONE) || r < 0)
                    return r;
            }
        }
    }

    return r;
}

在代码里面可以看到filter处理关键之处在于avfilter_process_command，  文件libavfilter/avfilter.c：

[AppleScript] 纯文本查看 复制代码

int avfilter_process_command(AVFilterContext *filter, const char *cmd, const char *arg, char *res, int res_len, int flags)
{
    if(!strcmp(cmd, "ping")){
        char local_res[256] = {0};

        if (!res) {
            res = local_res;
            res_len = sizeof(local_res);
        }
        av_strlcatf(res, res_len, "pong from:%s %s\n", filter->filter->name, filter->name);
        if (res == local_res)
            av_log(filter, AV_LOG_INFO, "%s", res);
        return 0;
    }else if(!strcmp(cmd, "enable")) {
        return set_enable_expr(filter, arg);
    }else if(filter->filter->process_command) {
        return filter->filter->process_command(filter, cmd, arg, res, res_len, flags);
    }
    return AVERROR(ENOSYS);
}

那么，代码中可以看到，调用的是抽象接口process_command，也就是说，如果要支持zmq的功能，需要具体的filter模块中支持process_command方法，grep看一眼都哪些filter支持process_command：

[AppleScript] 纯文本查看 复制代码

liuqideMacBook-Pro:ffmpeg liuqi$ grep -r ".process_command" libavfilter/
libavfilter//vf_scale.c:static int process_command(AVFilterContext *ctx, const char *cmd, const char *args,
libavfilter//vf_scale.c:    .process_command = process_command,
libavfilter//vf_scale.c:    .process_command = process_command,
libavfilter//vf_spp.c:static int process_command(AVFilterContext *ctx, const char *cmd, const char *args,
libavfilter//vf_spp.c:    .process_command = process_command,
libavfilter//f_streamselect.c:static int process_command(AVFilterContext *ctx, const char *cmd, const char *args,
libavfilter//f_streamselect.c:    .process_command = process_command,
libavfilter//f_streamselect.c:    .process_command = process_command,
libavfilter//vf_zscale.c:static int process_command(AVFilterContext *ctx, const char *cmd, const char *args,
libavfilter//vf_zscale.c:    .process_command = process_command,
libavfilter//af_volume.c:static int process_command(AVFilterContext *ctx, const char *cmd, const char *args,
libavfilter//af_volume.c:    .process_command = process_command,
libavfilter//vsrc_testsrc.c:static int color_process_command(AVFilterContext *ctx, const char *cmd, const char *args,
libavfilter//vsrc_testsrc.c:    .process_command = color_process_command,
libavfilter//vf_crop.c:static int process_command(AVFilterContext *ctx, const char *cmd, const char *args,
libavfilter//vf_crop.c:    .process_command = process_command,
libavfilter//vf_overlay.c:static int process_command(AVFilterContext *ctx, const char *cmd, const char *args,
libavfilter//vf_overlay.c:    .process_command = process_command,
libavfilter//af_anequalizer.c:static int process_command(AVFilterContext *ctx, const char *cmd, const char *args,
libavfilter//af_anequalizer.c:    .process_command = process_command,
libavfilter//af_firequalizer.c:static int process_command(AVFilterContext *ctx, const char *cmd, const char *args,
libavfilter//af_firequalizer.c:    .process_command    = process_command,
libavfilter//avfiltergraph.c:            r = avfilter_process_command(filter, cmd, arg, res, res_len, flags);
libavfilter//af_ladspa.c:static int process_command(AVFilterContext *ctx, const char *cmd, const char *args,
libavfilter//af_ladspa.c:    .process_command = process_command,
libavfilter//avf_concat.c:static int process_command(AVFilterContext *ctx, const char *cmd, const char *args,
libavfilter//avf_concat.c:    .process_command = process_command,
libavfilter//avfilter.c:int avfilter_process_command(AVFilterContext *filter, const char *cmd, const char *arg, char *res, int res_len, int flags)
libavfilter//avfilter.c:    }else if(filter->filter->process_command) {
libavfilter//avfilter.c:        return filter->filter->process_command(filter, cmd, arg, res, res_len, flags);
libavfilter//avfilter.c:    ff_inlink_process_commands(link, frame);
libavfilter//avfilter.c:    ff_inlink_process_commands(link, frame);
libavfilter//avfilter.c:int ff_inlink_process_commands(AVFilterLink *link, const AVFrame *frame)
libavfilter//avfilter.c:        avfilter_process_command(link->dst, cmd->command, cmd->arg, 0, 0, cmd->flags);
libavfilter//vf_drawtext.c:    .process_command = command,
libavfilter//vf_hue.c:static int process_command(AVFilterContext *ctx, const char *cmd, const char *args,
libavfilter//vf_hue.c:    .process_command = process_command,
libavfilter//vf_pp.c:static int pp_process_command(AVFilterContext *ctx, const char *cmd, const char *args,
libavfilter//vf_pp.c:    .process_command = pp_process_command,
libavfilter//vf_eq.c:static int process_command(AVFilterContext *ctx, const char *cmd, const char *args,
libavfilter//vf_eq.c:    .process_command = process_command,
libavfilter//vf_rotate.c:static int process_command(AVFilterContext *ctx, const char *cmd, const char *args,
libavfilter//vf_rotate.c:    .process_command = process_command,
libavfilter//af_rubberband.c:static int process_command(AVFilterContext *ctx, const char *cmd, const char *args,
libavfilter//af_rubberband.c:    .process_command = process_command,
libavfilter//af_biquads.c:static int process_command(AVFilterContext *ctx, const char *cmd, const char *args,
libavfilter//af_biquads.c:    .process_command = process_command,                  \
libavfilter//src_movie.c:static int process_command(AVFilterContext *ctx, const char *cmd, const char *args,
libavfilter//src_movie.c:    .process_command = process_command
libavfilter//src_movie.c:    .process_command = process_command,
libavfilter//avfilter.h:    int (*process_command)(AVFilterContext *, const char *cmd, const char *arg, char *res, int res_len, int flags);
libavfilter//avfilter.h:int avfilter_process_command(AVFilterContext *filter, const char *cmd, const char *arg, char *res, int res_len, int flags);
libavfilter//filters.h: * Commands will trigger the process_command() callback.
libavfilter//filters.h:int ff_inlink_process_commands(AVFilterLink *link, const AVFrame *frame);
libavfilter//filters.h: * @note  May trigger process_command() and/or update is_disabled.
libavfilter//filters.h: * @note  May trigger process_command() and/or update is_disabled.
libavfilter//af_atempo.c:static int process_command(AVFilterContext *ctx,
libavfilter//af_atempo.c:    .process_command = process_command,
liuqideMacBook-Pro:ffmpeg liuqi$

刚才的例子改变了overlay中的x坐标，那么看一下overlay滤镜中处理的process_command,文件libavfilter/vf_overlay.c：

[AppleScript] 纯文本查看 复制代码

static int process_command(AVFilterContext *ctx, const char *cmd, const char *args,
                           char *res, int res_len, int flags)
{
    OverlayContext *s = ctx->priv;
    int ret;

    if      (!strcmp(cmd, "x"))
        ret = set_expr(&s->x_pexpr, args, cmd, ctx);
    else if (!strcmp(cmd, "y"))
        ret = set_expr(&s->y_pexpr, args, cmd, ctx);
    else
        ret = AVERROR(ENOSYS);

    if (ret < 0)
        return ret;

    if (s->eval_mode == EVAL_MODE_INIT) {
        eval_expr(ctx);
        av_log(ctx, AV_LOG_VERBOSE, "x:%f xi:%d y:%f yi:%d\n",
               s->var_values[VAR_X], s->x,
               s->var_values[VAR_Y], s->y);
    }
    return ret;
}

从而可以看到，通过zmq改变x，y在overlay中是可以的，到这里，介绍结束