Изучаем VoIP-движок Mediastreamer2. Часть 10

Материал статьи взят с моего дзен-канала.

В прошлой статье мы сделали дуплексное переговорное устройство, выполняющее обмен звуковым сигналом через дуплексную RTP-сессию. В этой статье мы научимся писать фильтры и добавим сделанный своими руками фильтр в сделанное своими руками переговорное устройство.

Разрабатываем плагин

Плагины в медиастримере, как и во многих других программах, используются для расширения функционала без необходимости перекомпиляции самого медиастримера.

Чтобы использовать плагин в своей программе, вы с помощью include должны подключить заголовочный файл плагина. В теле программы, с помощью функции у ms_filter_register() выполнить регистрацию нового фильтра. Естественно, ваша программа и и исходник плагина должны быть скомпилированы и собраны в одно приложение.

Теперь обратимся к написанию плагина. Все фильтры медиастримера и плагины подчиняются в написании общему канону, что значительно облегчает понимание устройства очередного фильтра, который вы захотели изучить. Поэтому далее, чтобы не размножать сущности, плагины буду называть фильтрами.

Предположим, мы хотим разработать новый фильтр с названием НАШ_ФИЛЬТР (NASH_FILTR). Он будет выполнять элементарную вещь — получать блоки со своего единственного входа и передавать на свои пять выходов. А еще он будет формировать событие, в случае если через него пройдет более пяти блоков с уровнем сигнала ниже заданного порога и если через него пройдет более пяти блоков с уровнем выше порога, он тоже будет формировать событие.

Порог будет задаваться с помощью метода фильтра. Второй и третий методы будут разрешать/воспрещать прохождение блоков на выходы.

Приступаем. Написание фильтра нужно начинать с заголовочного файла. Он в первых строках должен подключить файл msfilter.h, с помощью макроса MS_FILTER_METHOD объявить методы нового фильтра (если они есть), объявить события генерируемые фильтром (если они есть) и объявить экспортируемую структуру типа MSFilterDesc с описанием параметров фильтра. Структура имеет следующий вид:

struct _MSFilterDesc{
    MSFilterId id;    /* Идентификатор типа фильтра заданный в файле allfilters.h или нами самими.  */
    const char *name; /* Имя фильтра (латиницей конечно). */
    const char *text; /** Короткий текст, описывающий фильтр. */
    MSFilterCategory category; /* Категория фильтра, описывающая его роль. */
    const char *enc_fmt; /* sub-mime используемого формата,  должен быть указан для категорий фильтров MS_FILTER_ENCODER или MS_FILTER_DECODER */
    int ninputs; /* Количество входов. */
    int noutputs; /*Количество выходов.  */
    MSFilterFunc init; /* Функция начальной инициализации фильтра. */
    MSFilterFunc preprocess; /* Функция вызываемая однократно перед запуском фильтра в работу. */
    MSFilterFunc process; /**< Функция выполняющая основную работу фильтра, вызываемая на каждый тик  тикера MSTicker. */
    MSFilterFunc postprocess; /* Функция завершения работы фильтра, вызывается однократно после последнего вызова process(), перед удалением фильтра. */
    MSFilterFunc uninit; /**< Функция завершения работы фильтра, выполняет освобождение памяти, которая была занята при создании внутренних структур фильтра. */
    MSFilterMethod *methods; /* Таблица методов фильтра. */
    unsigned int flags; /* Специальные флаги фильтра  описанные в перечислении  MSFilterFlags. */
};

/*
 Структура для описания фильтра.
*/
typedef struct _MSFilterDesc MSFilterDesc;

Типы, используемые структурой можно изучить по файлу msfilter.h. Заголовочный файл нашего фильтра будет иметь вид:

/* Файл nash_filter.h, описывает фильтр-разветвитель и нойзгейт. */

#ifndef myfilter_h
#define myfilter_h

/* Подключаем заголовочный файл с перечислением фильтров медиастримера. */
#include <mediastreamer2/msticker.h>

/* 
   Задаем числовой идентификатор нового типа фильтра.  Это число не должно
   совпадать ни с одним из других типов.  В медиастримере  в файле allfilters.h
   есть соответствующее перечисление enum MSFilterId. К сожалению, непонятно
   как определить максимальное занятое значение, кроме как заглянуть в этот
   файл. Но мы возьмем в качестве id для нашего фильтра заведомо большее
   значение: 4000.  Будем полагать, что разработчики добавляя новые фильтры, не
   скоро доберутся до этого номера.  
   */
#define NASH_FILTER_ID 4000

/* 
   Определяем методы нашего фильтра. Вторым параметром макроса должен
   порядковый номер метода, число от 0.  Третий параметр это тип аргумента
   метода, указатель на который будет передаваться методу при вызове. У методов
   аргументов может и не быть, как показано ниже. 
   */
#define NASH_FILTER_SET_TRESHOLD MS_FILTER_METHOD(NASH_FILTER_ID , 0, float)
#define NASH_FILTER_TUNE_OFF     MS_FILTER_METHOD_NO_ARG(NASH_FILTER_ID ,1)
#define NASH_FILTER_TUNE_ON      MS_FILTER_METHOD_NO_ARG(NASH_FILTER_ID ,2)

/* Теперь определяем структуру, которая будет передаваться вместе с событием. */
struct _NASHFilterEvent
{
    /* Это поле, которое будет выполнять роль флага,
       0 - появились нули, 1 - появился сигнал.*/
    char state; 
    /* Время, когда произошло событие. */
    uint64_t time;
};
typedef struct _NASHFilterEvent NASHFilterEvent;

/* Определяем событие для нашего фильтра. */
#define NASH_FILTER_EVENT MS_FILTER_EVENT(MS_RTP_RECV_ID, 0, NASHFilterEvent)

/* Определяем экспортируемую переменную, которая будет
   хранить характеристики для данного типа фильтров. */
extern MSFilterDesc nash_filter_desc;

#endif /* myfilter_h */

Теперь можно заняться исходным файлом. Исходный код фильтра с комментариями показан ниже. Здесь определены методы фильтра, обязательные функции фильтра. Затем ссылки на методы и функции в определенном порядке помещаются в экспортируемую структуру nash_filter_desc. Которая используется медиастримером для "вживления " фильтров данного типа в рабочий процесс обработки данных.

/* Файл nash_filter.с, описывает фильтр-разветвитель и нойзгейт. */

#include "nash_filter.h"
#include <math.h>

#define NASH_FILTER_NOUTPUTS 5

/* Определяем структуру, которая хранит внутреннее состояние фильтра. */
typedef struct _nash_filterData
{
    bool_t disable_out;  /* Разрешение передачи блоков на выход. */
    int last_state;   /* Текущее состояние переключателя. */
    char zero_count;     /* Счетчик нулевых блоков. */
    char lag;            /* Количество блоков для принятия решения нойзгейтом. */
    char n_count;        /* Счетчик НЕнулевых блоков. */
    float skz_level;     /* Среднеквадратическое значение сигнала внутри
блока, при котором фильтр будет пропускать сигнал. Одновременно это порог
срабатывания, по которому будет формироваться событие.  */

} nash_filterData;

/*----------------------------------------------------------*/
/* Обязательная функция инициализации. */
static void nash_filter_init(MSFilter *f)
{
    nash_filterData *d=ms_new0(nash_filterData, 1);
    d->lag=5;
    f->data=d;
}

/*----------------------------------------------------------*/
/* Обязательная функция финализации работы фильтра,
   освобождается память. */
static void nash_filter_uninit(MSFilter *f)
{
    ms_free(f->data);
}

/*----------------------------------------------------------*/
/* Определяем образцовый массив с нулями, заведомо
   большего размера чем блок. */
char zero_array[1024]={0};

/* Определяем событие фильтра. */
NASHFilterEvent event;

/*----------------------------------------------------------*/
/* Функция отправки события. */
static void send_event(MSFilter *f, int state)
{
    nash_filterData *d =( nash_filterData* ) f->data;
     d->last_state = state;
    /* Устанавливаем время возникновения события,
       от момента первого тика. Время в миллисекундах. */
    event.time=f -> ticker -> time;
    event.state=state;  
    ms_filter_notify(f, NASH_FILTER_EVENT, &event);
}   

/*----------------------------------------------------------*/
/* Функция вычисляет среднеквадратическое (эффективное) значение сигнала внутри
  блока. */
static float calc_skz(nash_filterData *d, int16_t *signal, int numsamples)
{
    int i;
    float acc = 0;
    for (i=0; i<numsamples; i++)
    {
        int s=signal[i];
        acc = acc + s * s;
    }
    float skz = (float)sqrt(acc / numsamples);
    return skz;
}

/*----------------------------------------------------------*/
/* Обязательная функция основного цикла фильтра,
   вызывается с каждым тиком. */
static void nash_filter_process(MSFilter *f)
{
    nash_filterData *d=(nash_filterData*)f->data;

    /* Указатель на входное сообщение содержащее блок данных. */
    mblk_t *im;
    int i;
    int state;
    /* Вычитываем сообщения из входной очереди
       до полного её опустошения. */
    while((im=ms_queue_get(f->inputs[0]))!=NULL)
    {
        /* Если выходы запрещены, то просто удаляем входное сообщение. */
        if ( d -> disable_out)
        {
          freemsg(im);
          continue;
        }

        /* Измеряем уровень сигнала и принимаем решение об отправке сигнала. */
        float skz = calc_skz(d, (int16_t*)im->b_rptr, msgdsize(im));
        state = (skz > d->skz_level) ? 1 : 0; 
        if (state) 
        {
            d->n_count++;
            d->zero_count = 0;
        }
        else
        {
            d->n_count = 0;
            d->zero_count++;
        }
        if (((d->zero_count > d->lag) || (d->n_count > d->lag))
            &&  (d->last_state != state)) send_event(f, state);

        /* Приступаем к копированию входного сообщения и раскладке по выходам. Но
         * только по тем, к которым подключена нагрузка. Оригинальное сообщение
         * уйдет на выход с индексом 0, а его копии попадут на остальные
         * выходы. */ 
        int output_count = 0;
        mblk_t *outm; /* Указатель на сообщение с выходным блоком данных. */
        for(i=0; i < f->desc->noutputs; i++)
        {
            if (f->outputs[i]!=NULL)
            {
                if (output_count == 0)
                {
                    outm = im;
                }
                else
                {
                    /* Создаем легкую копию сообщения. */       
                    outm = dupmsg(im);
                }
                /* Помещаем копию или оригинал входного сообщения на очередной
                 * выход фильтра. */ 
                ms_queue_put(f->outputs[i], outm);
                output_count++;
            }
        }
    }
}

/*----------------------------------------------------------*/
/* Функция-обработчик вызова метода NASH_FILTER_SET_LAG. */
static int nash_filter_set_treshold(MSFilter *f, void *arg)
{
    nash_filterData *d=(nash_filterData*)f->data;
    d->skz_level=*(float*)arg;
    return 0;
}

/*----------------------------------------------------------*/
/* Функция-обработчик вызова метода NASH_FILTER_TUNE_OFF. */
static int nash_filter_tune_off(MSFilter *f, void *arg)
{
    nash_filterData *d=(nash_filterData*)f->data;
    d->disable_out=TRUE;
    return 0;
}

/*----------------------------------------------------------*/
/* Функция-обработчик вызова метода NASH_FILTER_TUNE_ON. */
static int nash_filter_tune_on(MSFilter *f, void *arg)
{
    nash_filterData *d=(nash_filterData*)f->data;
    d->disable_out=FALSE;
    return 0;
}

/*----------------------------------------------------------*/
/* Заполняем таблицу методов фильтра, сколько методов
   мы определили в заголовочном файле столько ненулевых
   строк. */
static MSFilterMethod nash_filter_methods[]={
    { NASH_FILTER_SET_TRESHOLD, nash_filter_set_treshold },
    { NASH_FILTER_TUNE_OFF, nash_filter_tune_off },
    { NASH_FILTER_TUNE_ON, nash_filter_tune_on },
    { 0 , NULL } /* Маркер конца таблицы. */
};

/*----------------------------------------------------------*/
/* Описание фильтра для медиастримера. */
MSFilterDesc nash_filter_desc=
{
    NASH_FILTER_ID,
    "NASH_FILTER",
    "A filter with noise gate that reads from input and copy to it's five outputs.",
    MS_FILTER_OTHER,
    NULL,
    1,
    NASH_FILTER_NOUTPUTS,
    nash_filter_init,
    NULL,
    nash_filter_process,
    NULL,
    nash_filter_uninit,
    nash_filter_methods,
    0
};

MS_FILTER_DESC_EXPORT(nash_filter_desc)

Теперь, не откладывая в долгий ящик, применим наш фильтр в сделанном ранее переговорном устройстве. На заглавной картинке показана схема видоизмененного переговорного устройства.
Наш собственноручный фильтр хотелось изобразить как-то по-особенному ярко. Поэтому вы сразу найдете на схеме наш фильтр.

В схему добавился фильтр-регистратор, который пишет входной сигнал в файл формата wav. По замыслу, наш фильтр позволит не писать в файл паузы в речи. Тем самым сокращая его размер.
В начале статьи мы описали алгоритм действий фильтра. В основном приложении выполняется обработка событий, которые он формирует. Если событие содержит флаг "0", то основное приложение приостанавливает запись. Как только придет событие с флагом "1" запись возобновляется.

К прежним аргументам командной строки добавились еще два: --ng, который задает уровень порога срабатывания фильтра и --rec, который запускает запись в файл с именем record.wav.

/* Файл mstest9.c Имитатор переговорного устройства c регистратором и
 * нойзгейтом. */

#include <mediastreamer2/mssndcard.h>
#include <mediastreamer2/dtmfgen.h>
#include <mediastreamer2/msrtp.h>
#include <mediastreamer2/msfilerec.h>

/* Подключаем наш фильтр. */
#include "nash_filter.h"

/* Подключаем файл общих функций. */
#include "mstest_common.c"

/*----------------------------------------------------------*/
struct _app_vars
{
    int  local_port;              /* Локальный порт. */
    int  remote_port;             /* Порт переговорного устройства на удаленном компьютере. */
    char remote_addr[128];        /* IP-адрес удаленного компьютера. */
    MSDtmfGenCustomTone dtmf_cfg; /* Настройки тестового сигнала генератора. */
    MSFilter* recorder;           /* Указатель на фильтр регистратор. */
    bool_t file_is_open;          /* Флаг того, что файл для записи открыт. */
    /* Порог, при котором прекращается запись принимаемого сигнала в файл. */
    float treshold; 
    bool_t en_rec;                /*Включить запись в файл.*/    
};

typedef struct _app_vars app_vars;

/*----------------------------------------------------------*/
/* Создаем дуплексную RTP-сессию. */
RtpSession* create_duplex_rtp_session(app_vars v)
{
    RtpSession *session = create_rtpsession (v.local_port, v.local_port + 1,
            FALSE, RTP_SESSION_SENDRECV);
    rtp_session_set_remote_addr_and_port(session, v.remote_addr, v.remote_port,
            v.remote_port + 1);
    rtp_session_set_send_payload_type(session, PCMU);
    return session;
}

/*----------------------------------------------------------*/
/* Функция преобразования аргументов командной строки в 
 * настройки программы. */
void  scan_args(int argc, char *argv[], app_vars *v)
{
    char i;
    for (i=0; i<argc; i++)
    {
        if (!strcmp(argv[i], "--help"))
        {
            char *p=argv[0]; p=p + 2;
            printf("  %s walkie talkie\n\n", p);
            printf("--help      List of options.\n");
            printf("--version   Version of application.\n");
            printf("--addr      Remote abonent IP address string.\n");
            printf("--port      Remote abonent port number.\n");
            printf("--lport     Local port number.\n");
            printf("--gen       Generator frequency.\n");
            printf("--ng        Noise gate treshold level from 0. to 1.0\n");
            printf("--rec       record to file 'record.wav'.\n");
            exit(0);
        }

        if (!strcmp(argv[i], "--version"))
        {
            printf("0.1\n");
            exit(0);
        }

        if (!strcmp(argv[i], "--addr"))
        {
            strncpy(v->remote_addr, argv[i+1], 16);
            v->remote_addr[16]=0;
            printf("remote addr: %s\n", v->remote_addr);
        }

        if (!strcmp(argv[i], "--port"))
        {
            v->remote_port=atoi(argv[i+1]);
            printf("remote port: %i\n", v->remote_port);
        }

        if (!strcmp(argv[i], "--lport"))
        {
            v->local_port=atoi(argv[i+1]);
            printf("local port : %i\n", v->local_port);
        }

        if (!strcmp(argv[i], "--gen"))
        {
            v -> dtmf_cfg.frequencies[0] = atoi(argv[i+1]);
            printf("gen freq : %i\n", v -> dtmf_cfg.frequencies[0]);
        }

        if (!strcmp(argv[i], "--ng"))
        {
            v -> dtmf_cfg.frequencies[0] = atoi(argv[i+1]);
            printf("noise gate treshold: %f\n", v -> treshold);
        }
         if (!strcmp(argv[i], "--rec"))
        {
            v -> en_rec = TRUE;
            printf("enable recording: %i\n", v -> en_rec);
        }
    }
}

/*----------------------------------------------------------*/
/* Функция обратного вызова, она будет вызвана фильтром, как только он
 * заметит, что наступила тишина или наоборот тишина сменилась звуками. */
static void change_detected_cb(void *data, MSFilter *f, unsigned int event_id,
        NASHFilterEvent *ev)
{
    app_vars *vars = (app_vars*) data;

    /* Если запись не была разрешена, то выходим. */
    if (! vars -> en_rec) return; 

    if (ev -> state)
    {
        /* Возобновляем запись. */
        if(!vars->file_is_open)
        {
            ms_filter_call_method(vars->recorder, MS_FILE_REC_OPEN, "record.wav");
            vars->file_is_open = 1;
        }
        ms_filter_call_method(vars->recorder, MS_FILE_REC_START, 0);
        printf("Recording...\n");
    }
    else
    {
        /* Приостанавливаем запись. */
        ms_filter_call_method(vars->recorder, MS_FILE_REC_STOP, 0);
        printf("Pause...\n");
    }
}

/*----------------------------------------------------------*/
int main(int argc, char *argv[])
{
    /* Устанавливаем настройки по умолчанию. */
    app_vars vars={5004, 7010, "127.0.0.1", {0}, 0, 0, 0.01, 0};

    /* Устанавливаем настройки настройки программы в 
     * соответствии с аргументами командной строки. */
    scan_args(argc, argv, &vars);

    ms_init();

    /* Создаем экземпляры фильтров передающего тракта. */
    MSSndCard *snd_card =
        ms_snd_card_manager_get_default_card(ms_snd_card_manager_get());
    MSFilter *snd_card_read = ms_snd_card_create_reader(snd_card);
    MSFilter *dtmfgen = ms_filter_new(MS_DTMF_GEN_ID);
    MSFilter *rtpsend = ms_filter_new(MS_RTP_SEND_ID);

    /* Создаем фильтр кодера. */
    MSFilter *encoder = ms_filter_create_encoder("PCMU");

    /* Регистрируем типы нагрузки. */
    register_payloads();

    /* Создаем дуплексную RTP-сессию. */
    RtpSession* rtp_session = create_duplex_rtp_session(vars);
    ms_filter_call_method(rtpsend, MS_RTP_SEND_SET_SESSION, rtp_session);

    /* Соединяем фильтры передатчика. */
    ms_filter_link(snd_card_read, 0, dtmfgen, 0);
    ms_filter_link(dtmfgen, 0, encoder, 0);
    ms_filter_link(encoder, 0, rtpsend, 0);

    /* Создаем фильтры приемного тракта. */
    MSFilter *rtprecv = ms_filter_new(MS_RTP_RECV_ID);
    ms_filter_call_method(rtprecv, MS_RTP_RECV_SET_SESSION, rtp_session);

    /* Создаем фильтр декодера. */
    MSFilter *decoder=ms_filter_create_decoder("PCMU");
    //MS_FILE_REC_ID

    /* Регистрируем наш фильтр. */
    ms_filter_register(&nash_filter_desc);
    MSFilter *nash = ms_filter_new(NASH_FILTER_ID);

    /* Создаем фильтр звуковой карты. */
    MSFilter *snd_card_write = ms_snd_card_create_writer(snd_card);

    /* Создаем фильтр регистратора. */
    MSFilter *recorder=ms_filter_new(MS_FILE_REC_ID);
    vars.recorder = recorder; 

    /* Соединяем фильтры приёмного тракта. */
    ms_filter_link(rtprecv, 0, decoder, 0);
    ms_filter_link(decoder, 0, nash, 0);
    ms_filter_link(nash, 0, snd_card_write, 0);
    ms_filter_link(nash, 1, recorder, 0);

    /* Подключаем к фильтру функцию обратного вызова, и передаем ей в
     * качестве пользовательских данных указатель на структуру с настройками
     * программы, в которой среди прочих есть указать на фильтр
     * регистратора. */
    ms_filter_set_notify_callback(nash,
            (MSFilterNotifyFunc)change_detected_cb, &vars);
    ms_filter_call_method(nash,NASH_FILTER_SET_TRESHOLD, &vars.treshold); 

    /* Создаем источник тактов - тикер. */
    MSTicker *ticker = ms_ticker_new();

    /* Подключаем источник тактов. */
    ms_ticker_attach(ticker, snd_card_read);
    ms_ticker_attach(ticker, rtprecv);

    /* Если настройка частоты генератора отлична от нуля, то запускаем генератор. */   
    if (vars.dtmf_cfg.frequencies[0])
    {
        /* Настраиваем структуру, управляющую выходным сигналом генератора. */
        vars.dtmf_cfg.duration = 10000;
        vars.dtmf_cfg.amplitude = 1.0;
    }

    /* Организуем цикл перезапуска генератора. */
    printf("Press ENTER to exit.\n ");
    char c=getchar();
    while(c != '\n')
    {
        if(vars.dtmf_cfg.frequencies[0])
        {
            /* Включаем звуковой генератор. */
            ms_filter_call_method(dtmfgen, MS_DTMF_GEN_PLAY_CUSTOM,
                    (void*)&vars.dtmf_cfg);
        }
        char c=getchar();
        printf("--\n");
    }
    if (vars.en_rec ) ms_filter_call_method(recorder, MS_FILE_REC_CLOSE, 0);
}

Из-за того, что у нас добавились файлы и была использована библиотека math, командная строка для компиляции, усложнилась, и выглядит так:

$ gcc mstest9.c nash_filter.c -o mstest9   `pkg-config mediastreamer   --libs --cflags`  -lm

После сборки приложения запускаем его, на первом компьютере с такими аргументами:

$ ./mstest9  --lport 7010  --port 8010 --addr <тут адрес второго компьютера> --rec

На втором компьютере запускаем с такими настройками:

$ ./mstest9  --lport 8010  --port 7010 --addr <тут адрес первого компьютера>

После этого первый компьютер начнет записывать все, что вы скажете в микрофон второго. При этом в консоли будет написано слово "Recording...". Как только вы замолчите запись будет поставлена на паузу с выводом сообщения "Pause...". Возможно вам придется поэкспериментировать с уровне порога.

В этой статье мы научились писать фильтры. Как вы могли заметить, в функции nash_filter_process() выполняются манипуляции с блоками данных. Поскольку пример учебный, то там был задействован минимум возможностей медиастримера по манипуляциям с блоками данных.

В следующей статье мы рассмотрим организацию очередей сообщений и функции управления сообщениями. Что в дальнейшем поможет разрабатывать фильтры с более сложной обработкой информации.