avfilter/asubprocess: initial incomplete version of subprocess filter

Signed-off-by: Stefan Westerfeld <stefan@space.twc.de>

libavfilter/Makefile

@@ -110,6 +110,7 @@ OBJS-$(CONFIG_ASR_FILTER)                    += af_asr.o
 OBJS-$(CONFIG_ASTATS_FILTER)                 += af_astats.o
 OBJS-$(CONFIG_ASTREAMSELECT_FILTER)          += f_streamselect.o framesync.o
 OBJS-$(CONFIG_ASUBBOOST_FILTER)              += af_asubboost.o
+OBJS-$(CONFIG_ASUBPROCESS_FILTER)            += af_asubprocess.o
 OBJS-$(CONFIG_ASUBCUT_FILTER)                += af_asupercut.o
 OBJS-$(CONFIG_ASUPERCUT_FILTER)              += af_asupercut.o
 OBJS-$(CONFIG_ASUPERPASS_FILTER)             += af_asupercut.o

libavfilter/af_asubprocess.c

+#include "libavutil/opt.h"
+
+#include "audio.h"
+#include "avfilter.h"
+#include "filters.h"
+#include "internal.h"
+
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <unistd.h>
+
+// =============================================
+#include <sys/types.h>
+#include <sys/wait.h>
+
+#include <stdio.h>
+#include <unistd.h>
+#include <fcntl.h>
+#include <assert.h>
+#include <poll.h>
+#include <stdlib.h>
+#include <stdbool.h>
+#include <string.h>
+
+#define BUFFER_SIZE 4096
+
+typedef struct SPB
+{
+  struct SPB *next;
+  int offset;
+  int count;
+  char data[BUFFER_SIZE];
+} SPB;
+
+typedef struct SP
+{
+  int input_pipe;
+  int output_pipe;
+  SPB *out_buffers;
+  int pid;
+} SP;
+
+static SPB *
+spb_new()
+{
+  SPB *spb = malloc (sizeof (SPB));
+
+  spb->next = NULL;
+  spb->offset = 0;
+  spb->count = 0;
+  return spb;
+}
+
+static SP *
+sp_new (char *command)
+{
+  SP *sp = calloc (1, sizeof (SP));
+  int output_pipe[2];
+  int pok = pipe (output_pipe);
+  if (pok != 0)
+    printf ("pipe() failed\n");
+  int input_pipe[2];
+  pok = pipe (input_pipe);
+  if (pok != 0)
+    printf ("pipe() failed\n");
+  pid_t pid = fork();
+  if (pid < 0)
+    printf ("fork() failed\n");
+  if (pid == 0)
+    {
+      if (dup2 (input_pipe[0], 0) < 0)
+        printf ("dup2() failed\n");
+      if (dup2 (output_pipe[1], 1) < 0)
+        printf ("dup2() failed\n");
+      close (output_pipe[1]);
+      close (output_pipe[0]);
+      close (input_pipe[1]);
+      close (input_pipe[0]);
+      execl ("/bin/sh", "/bin/sh", "-c", command, NULL);
+    }
+  sp->pid = pid;
+  int i;
+  close (input_pipe[0]);
+  close (output_pipe[1]);
+  i = fcntl(output_pipe[0], F_GETFL);
+  assert(i != -1);
+  i |= O_NONBLOCK;
+  fcntl(output_pipe[0], F_SETFL, i);
+  i = fcntl(input_pipe[1], F_GETFL);
+  assert(i != -1);
+  i |= O_NONBLOCK;
+  fcntl(input_pipe[1], F_SETFL, i);
+  sp->input_pipe = input_pipe[1];
+  sp->output_pipe = output_pipe[0];
+  return (sp);
+}
+
+static int
+sp_can_read (SP *sp)
+{
+  int can_read = 0;
+  for (SPB *spb = sp->out_buffers; spb; spb = spb->next)
+    {
+      can_read += spb->count - spb->offset;
+    }
+  return can_read;
+}
+
+static void
+sp_read (SP *sp, char *buffer, int count)
+{
+  SPB *spb = sp->out_buffers;
+  while (spb)
+    {
+      int n = spb->count - spb->offset;
+      if (n > count)
+        n = count;
+
+      memcpy (buffer, spb->data + spb->offset, n);
+      spb->offset += n;
+      count -= n;
+      buffer += n;
+      if (spb->count - spb->offset)
+        return;
+
+      sp->out_buffers = spb->next;
+      free (spb);
+      spb = sp->out_buffers;
+    }
+}
+
+static bool
+sp_write (SP *sp, char *buffer, int count)
+{
+  if (count == 0 && sp->input_pipe >= 0) // eof
+    {
+      close (sp->input_pipe);
+      sp->input_pipe = -1;
+    }
+
+  struct pollfd fds[2] = {
+    { .fd = sp->input_pipe, .events = POLLOUT, .revents = 0 },
+    { .fd = sp->output_pipe, .events = POLLIN, .revents = 0 },
+  };
+  int offset = 0;
+  do
+    {
+      poll (fds, 2, -1);
+      if (fds[0].revents & (POLLOUT | POLLHUP))
+        {
+          int rc = write (sp->input_pipe, &buffer[offset], count);
+          if (rc > 0)
+            {
+              offset += rc;
+              count -= rc;
+            }
+        }
+      if (fds[1].revents & (POLLIN | POLLHUP))
+        {
+          SPB *spb = spb_new();
+          int rc = read (sp->output_pipe, spb->data, sizeof (spb->data));
+          if (rc > 0)
+            {
+              spb->count = rc;
+              if (!sp->out_buffers)
+                sp->out_buffers = spb;
+              else
+                {
+                  SPB *last = sp->out_buffers;
+                  while (last->next)
+                    last = last->next;
+                  last->next = spb;
+                }
+            }
+          else if (rc == 0)
+            {
+              int status;
+              waitpid (sp->pid, &status, 0);
+              printf ("wstatus=%d\n", WEXITSTATUS (status));
+              return true;
+            }
+        }
+    } while (count);
+  return false;
+}
+// =============================================
+
+typedef struct AndioWMarkContext {
+    const AVClass *class;
+
+    double strength;
+
+    int    fd;
+    SP    *sp;
+    int    nb_samples;
+    int    eof;
+} AudioWMarkContext;
+
+#define OFFSET(x) offsetof(AudioWMarkContext, x)
+#define A AV_OPT_FLAG_AUDIO_PARAM|AV_OPT_FLAG_FILTERING_PARAM
+
+static const AVOption asubprocess_options[] = {
+    { "strength",   "set watermarking strength", OFFSET(strength), AV_OPT_TYPE_DOUBLE, {.dbl=10}, 1, 100, A },
+    { NULL },
+};
+
+AVFILTER_DEFINE_CLASS(asubprocess);
+
+static av_cold void uninit(AVFilterContext *ctx)
+{
+    AudioWMarkContext *s = ctx->priv;
+
+    printf ("TODO: uninit\n");
+#if 0
+    if (s->rbs)
+        rubberband_delete(s->rbs);
+#endif
+}
+
+static int config_input(AVFilterLink *inlink)
+{
+  printf ("TODO: config_input\n");
+  printf ("sample_rate = %d, n channels = %d\n", inlink->sample_rate, inlink->ch_layout.nb_channels);
+    AVFilterContext *ctx = inlink->dst;
+    AudioWMarkContext *s = ctx->priv;
+    s->sp = sp_new ("audiowmark add --format raw --raw-rate 44100 - - f0");
+#if 0
+    int opts = s->transients|s->detector|s->phase|s->window|
+               s->smoothing|s->formant|s->opitch|s->channels|
+               RubberBandOptionProcessRealTime;
+
+    if (s->rbs)
+        rubberband_delete(s->rbs);
+    s->rbs = rubberband_new(inlink->sample_rate, inlink->ch_layout.nb_channels, opts, 1. / s->tempo, s->pitch);
+    if (!s->rbs)
+        return AVERROR(ENOMEM);
+
+    s->nb_samples = rubberband_get_samples_required(s->rbs);
+    s->first_pts = AV_NOPTS_VALUE;
+#endif
+    s->nb_samples = 1024;
+
+    return 0;
+}
+
+static int filter_frame(AVFilterLink *inlink, AVFrame *in)
+{
+    AVFilterContext *ctx = inlink->dst;
+    AudioWMarkContext *s = ctx->priv;
+    AVFilterLink *outlink = ctx->outputs[0];
+    AVFrame *out;
+    int ret = 0, nb_samples;
+
+    printf ("filter_frame nb_samples=%d\n", in->nb_samples);
+    for (int i = 0; i < in->nb_samples; i++)
+      {
+        for (int j = 0; j < inlink->ch_layout.nb_channels; j++)
+          {
+            float **xd = (float **)in->extended_data;
+            short ss = xd[j][i]*32000;
+            sp_write (s->sp, &ss, 2);
+          }
+      }
+    int avail = sp_can_read (s->sp) / 2 / inlink->ch_layout.nb_channels;
+    if (avail > 2048)
+      avail = 2048;
+    if (avail > 0)
+    {
+        out = ff_get_audio_buffer(outlink, avail);
+        if (!out) {
+            av_frame_free(&in);
+            return AVERROR(ENOMEM);
+        }
+        short buffer[100000]; // FIXME avail * inlink->ch_layout.nb_channels];
+        sp_read (s->sp, buffer, avail * 2 * inlink->ch_layout.nb_channels);
+        int k = 0;
+        for (int i = 0; i < avail; i++)
+          {
+            for (int j = 0; j < inlink->ch_layout.nb_channels; j++)
+              {
+                float **xd = (float **)out->extended_data;
+                xd[j][i] = buffer[k++] * (1/32000.);
+              }
+          }
+        ret = ff_filter_frame(outlink, out);
+    }
+#if 0
+    if (s->first_pts == AV_NOPTS_VALUE)
+        s->first_pts = in->pts;
+
+    rubberband_process(s->rbs, (const float *const *)in->extended_data, in->nb_samples, s->eof);
+    s->nb_samples_in += in->nb_samples;
+
+    nb_samples = rubberband_available(s->rbs);
+    if (nb_samples > 0) {
+        out = ff_get_audio_buffer(outlink, nb_samples);
+        if (!out) {
+            av_frame_free(&in);
+            return AVERROR(ENOMEM);
+        }
+        out->pts = s->first_pts + av_rescale_q(s->nb_samples_out,
+                     (AVRational){ 1, outlink->sample_rate },
+                     outlink->time_base);
+        s->last_pts = out->pts;
+        nb_samples = rubberband_retrieve(s->rbs, (float *const *)out->extended_data, nb_samples);
+        out->nb_samples = nb_samples;
+        ret = ff_filter_frame(outlink, out);
+        s->nb_samples_out += nb_samples;
+    }
+#endif
+
+    av_frame_free(&in);
+    if (ff_inlink_queued_samples(inlink) >= s->nb_samples)
+        ff_filter_set_ready(ctx, 100);
+    return ret < 0 ? ret : nb_samples;
+}
+
+static int activate(AVFilterContext *ctx)
+{
+    AVFilterLink *inlink = ctx->inputs[0];
+    AVFilterLink *outlink = ctx->outputs[0];
+    AudioWMarkContext *s = ctx->priv;
+    AVFrame *in = NULL;
+    int64_t pts;
+    int status, ret;
+
+    FF_FILTER_FORWARD_STATUS_BACK(outlink, inlink);
+
+    ret = ff_inlink_consume_samples(inlink, s->nb_samples, s->nb_samples, &in);
+
+    if (ff_inlink_acknowledge_status(inlink, &status, &pts))
+        s->eof |= status == AVERROR_EOF;
+
+    if (ret < 0)
+        return ret;
+    if (ret > 0) {
+        ret = filter_frame(inlink, in);
+        if (ret != 0)
+            return ret;
+    }
+
+#if 0
+    if (s->eof) {
+        ff_outlink_set_status(outlink, AVERROR_EOF, s->last_pts);
+        return 0;
+    }
+#endif
+
+    FF_FILTER_FORWARD_WANTED(outlink, inlink);
+
+    return FFERROR_NOT_READY;
+}
+
+static int process_command(AVFilterContext *ctx, const char *cmd, const char *args,
+                           char *res, int res_len, int flags)
+{
+  printf ("TODO: process\n");
+#if 0
+    RubberBandContext *s = ctx->priv;
+    int ret;
+
+    ret = ff_filter_process_command(ctx, cmd, args, res, res_len, flags);
+    if (ret < 0)
+        return ret;
+
+    rubberband_set_time_ratio(s->rbs, 1. / s->tempo);
+    rubberband_set_pitch_scale(s->rbs, s->pitch);
+    s->nb_samples = rubberband_get_samples_required(s->rbs);
+#endif
+    return 0;
+}
+static const AVFilterPad asubprocess_inputs[] = {
+    {
+        .name          = "default",
+        .type          = AVMEDIA_TYPE_AUDIO,
+        .config_props  = config_input,
+    },
+};
+
+const AVFilter ff_af_asubprocess = {
+    .name          = "asubprocess",
+    .description   = NULL_IF_CONFIG_SMALL("Filter audio stream with subprocess."),
+    .priv_size     = sizeof(AudioWMarkContext),
+    .priv_class    = &asubprocess_class,
+    .uninit        = uninit,
+    .activate      = activate,
+    FILTER_INPUTS(asubprocess_inputs),
+    FILTER_OUTPUTS(ff_audio_default_filterpad),
+    FILTER_SINGLE_SAMPLEFMT(AV_SAMPLE_FMT_FLTP),
+    .process_command = process_command,
+};

libavfilter/allfilters.c

@@ -101,6 +101,7 @@ extern const AVFilter ff_af_asubcut;
 extern const AVFilter ff_af_asupercut;
 extern const AVFilter ff_af_asuperpass;
 extern const AVFilter ff_af_asuperstop;
+extern const AVFilter ff_af_asubprocess;
 extern const AVFilter ff_af_atempo;
 extern const AVFilter ff_af_atilt;
 extern const AVFilter ff_af_atrim;