波形の動画をRubyで作成

Written by 北市真 — September 27, 2026

波形の動画をRubyで作成

音声が流れるのに合わせて波形が動く動画、よく見ると思いますよね。あれをRubyで作ってみたいと思います。果たしてRubyにその道具は揃っているのか……？

音声データの取得

素材はWikimedia Commonsで見付けた、槇原敬之の『ANSWER』にしてみましょう。このファイルはCC-BY 3.0で槇原敬之 Official Channelにより配布されています。

require "pathname"
require "open-uri"

src = URI("https://upload.wikimedia.org/wikipedia/commons/c/c5/ANSWER_TIME_TRAVELING_TOUR_2nd_Season.wav")
audio_path = Pathname("answer.wav")

unless audio_path.exist?
  audio_path.write src.read
end

これをAwaazで読み込みます。Awaazを選んだのはデータを数値計算ライブラリーのNumo::NArrayで返してくれるからで、諸々の計算を柔軟にかつ速くできるようにです。

require "numo/narray/alt"
require "awaaz"

waveform, sample_rate = Awaaz.load(audio_path.to_path)
pp waveform # => Numo::DFloat#shape=[1,1287720]

shape の最初が 1 なのでモノラル、そして1287720サンプルあるようです。

AwaazはNumo::NArrayに依存しているのですが、今はNumo::NArrayその物はメンテされていなくて、Numo::NArray Alternativeを使いたいので、先に読み込んでおきます。

波形画像の作成

これを、動画の前に一旦画像にしてみましょう。 Numo::NArray の各要素をプロットすればそれっぽくなるはず。

1287720サンプルは多過ぎるので取り敢えず400サンプルぐらいにしておきます。また、冒頭だとあまり音がないので、適当に10000サンプル目ぐらいにしておきます。

w = 400
color = Numo::UInt8[255, 255, 0]

data = waveform[10000...(10000 + w)]
# DFloat -> UInt8
# そのまま画像にするとy座標が正の場合は下に、負の場合は上にプロットされるけど、逆になって欲しいので1.0 - dataにしている
q_data = Numo::UInt8.cast((1.0 - data) * (255.0 / 2).round)
# q_dataをone-hot化：
# [0, 5, 1, 4, 2, 3]みたいな配列から
# [
#   [1, 0, 0, 0, 0, 0],
#   [0, 0, 0, 0, 0, 1],
#   [0, 1, 0, 0, 0, 0],
#   [0, 0, 0, 0, 1, 0],
#   [0, 0, 1, 0, 0, 0],
#   [0, 0, 0, 1, 0, 0],
# ]
# と、対応箇所だけ1で他は0という配列の配列を作る
# これを画面に見立てて1の所をプロットすれば波形になる筈
eye = Numo::UInt8.eye(256)
image = eye[true, q_data]
# 1だった所を色を持ったピクセルにする
image = image.expand_dims(2) * color

目で見て確認したいので、保存します。Magroを使うと、Numo::NArrayの配列を画像として保存できます。

require "magro"

Magro::IO.imsave "magro.png", image

線が荒いですが、一先ずそれっぽくはなったんではないでしょうか!?

線を繋げる

上は一つのx座標に対して一つのyをプロットしているので、yが連続していない時に飛び飛びになって荒く見えてしまいます。この端の間を補完してみましょう。「吴小林のラインアルゴリズム」なる物を使ってみます。

詳細は各自調べてもらうとして、実装は次のようになります。

def line_brightness(x0, y0, x1, y1)
  brightness = Numo::SFloat.zeros(256, x1.to_i - x0.to_i + 1)
  x0, y0, x1, y1 = x0.to_f, y0.to_f, x1.to_f, y1.to_f
  steep = (y1 - y0).abs > (x1 - x0).abs
  if steep
    x0, y0 = y0, x0
    x1, y1 = y1, x1
  end
  if x0 > x1
    x0, x1 = x1, x0
    y0, y1 = y1, y0
  end
  dx = x1 - x0
  dy = y1 - y0
  grad = dx == 0 ? 0.0 : dy/dx

  xend = x0.round
  yend = y0 + grad * (xend - x0)
  xgap = rfpart(x0 + 0.5)
  xpxl1 = xend.to_i
  ypxl1 = yend.floor.to_i

  if steep
    brightness[xpxl1,     ypxl1] = rfpart(yend) * xgap
    brightness[xpxl1 + 1, ypxl1] =  fpart(yend) * xgap
  else
    brightness[ypxl1,     xpxl1] = rfpart(yend) * xgap
    brightness[ypxl1 + 1, xpxl1] =  fpart(yend) * xgap
  end

  yi = yend + grad
  xend = x1.round
  yend = y1 + grad * (xend - x1)
  xgap = fpart(x1 + 0.5)
  xpxl2 = xend.to_i
  ypxl2 = yend.floor.to_i

  if steep
    brightness[xpxl2,     ypxl2] = rfpart(yend) * xgap
    brightness[xpxl2 + 1, ypxl2] =  fpart(yend) * xgap
  else
    brightness[ypxl2,     xpxl2] = rfpart(yend) * xgap
    brightness[ypxl2 + 1, xpxl2] =  fpart(yend) * xgap
  end

  (xpxl1 + 1).upto xpxl2 - 1 do |x|
    if steep
      brightness[x, yi.floor.to_i]     = rfpart(yi)
      brightness[x, yi.floor.to_i + 1] =  fpart(yi)
    else
      brightness[yi.floor.to_i, x]     = rfpart(yi)
      brightness[yi.floor.to_i + 1, x] =  fpart(yi)
    end
  end

  brightness
end

def fpart(x)
  x - x.floor
end

def rfpart(x)
  1.0 - fpart(x)
end

これを使って同じ箇所の画像を作ってみます。まずはこの関数を使って「明るさの行列」を作り、それに色を掛けてやることで色付きの滑らかな線にします。

y_data = (1.0 - data) * (255.to_f / 2)
coverage = Numo::SFloat.zeros(256, w)
(w - 1).times do |i|
  segment = line_brightness(0, y_data[i], 1, y_data[i + 1])
  coverage[true, i]     += segment[true, 0]
  coverage[true, i + 1] += segment[true, 1]
end
coverage[coverage > 1.0] = 1.0

image = coverage.expand_dims(2) * color
image = Numo::UInt8.cast(image.round)

Magro::IO.imsave "magro-lined.png", image

大分見易いですね！

動画を生成

RMagickに、複数の画像を登録してから保存すると動画にできる機能があるので、それを使います。 Magick::ImageList に画像を登録していって最後に #write を呼びます。

NDAV を使うと Numo::NArray から Magick::Image に変換できるので、上で作ったやり方で画像データを作り、変換して Magick::ImageList に登録します。

コードを書く前にちょっとチラ裏で計算しておきましょう。

  • fpsは25にする（RMagickの制約っぽい）
  • 一フレーム当たりの秒数 = 1 / fps = 0.04
  • 全音声をこの0.04秒間単位で分割して、その範囲を画像化し、それを連続させることで動画にする
  • 全秒数 = サンプル数 / サンプルレート
  • 全フレーム数 = 全秒数 * fps
  • 一フレーム当たりのサンプル数 = サンプルレート / fps
  • i フレーム目のサンプルは i * 一フレーム当たりのサンプル数 （オフセット）から 一フレーム当たりのサンプル数 - 1 まで
  • 幅400ピクセルにするので、↑のサンプルから間引いて400サンプルにする

一つ一つゆっくり確かめれば難しいことは無い筈です。頭の中だけで考えるより、コードを書きながらの方が理解しやすいかも。

require "rmagick"
require "ndav/magick/image"
require "ndav/numo/narray"

include NDAV::Converter

fps = 25
duration = waveform.shape[1].to_f / sample_rate
samples_per_frame = (sample_rate.to_f / fps).round
num_frames = (duration * fps).ceil
image_list = Magick::ImageList.new
num_frames.times do |i|
  start_frame = i * samples_per_frame
  end_frame = start_frame + samples_per_frame - 1
  # linspaceはstart_frameからend_frameまで、w個になるように間引いた際のインデックスの列を作る関数
  indices = Numo::Int32.cast(Numo::DFloat.linspace(start_frame, end_frame, w).round)
  # Numo::NArrayではインデックスの列（配列）でアクセスすると、そのインデックスにある値をまとめて取って来た配列を返す
  data = waveform[indices]
  y_data = (1.0 - data) * (255.to_f / 2)
  coverage = Numo::SFloat.zeros(256, w)
  (w - 1).times do |i|
    segment = line_brightness(0, y_data[i], 1, y_data[i + 1])
    coverage[true, i]     += segment[true, 0]
    coverage[true, i + 1] += segment[true, 1]
  end
  coverage[coverage > 1.0] = 1.0
  image = coverage.expand_dims(2) * color
  image = Numo::UInt8.cast(image.round)

  image_list << MagickImage(image)
end

image_list.write "rmagick.mp4"

それっぽくなったんではないでしょうか!?

動画と音声を合成

では上の画のみ動画と元のオーディオデータを合成して完成……としたいところですがRubyでそれは難しくて結局 ffmpeg コマンドを呼び出すことになります。勿論実用的にはそれでいいのですが、できればRubyでやりたいところ……GStreamergemを使ってやってみましょう。

パイプライン定義

GStreamerはマルチメディア用のパイプライン構築ができるライブラリーです。コードを見ながら説明します。

require "gstreamer"

fps = 30
duration = waveform.shape[1].to_f / sample_rate
samples_per_frame = (sample_rate.to_f / fps).round
num_frames = (duration * fps).ceil

pipeline = Gst.parse_launch(<<~EOP)
  mp4mux name=mux ! filesink location=gstreamer.mp4
  
  appsrc name=audiosrc format=time caps=audio/x-raw,format=F32LE,rate=#{sample_rate},channels=#{waveform.shape[0]},layout=interleaved
    ! audioconvert ! avenc_aac ! aacparse ! queue ! mux.

  appsrc name=videosrc format=time caps=video/x-raw,format=RGB,width=#{w},height=256,framerate=#{fps}/1
    ! videoconvert ! x264enc ! h264parse ! queue ! mux.
EOP

今回は音声付き動画を作ります。音声作成パイプラインと（画のみの）動画作成パイプラインを作って、二つを合流させます。上では三ブロックあり、最初が合流用パイプライン、次がオーディオ作成パイプライン、最後が動画作成パイプラインになっています。因みに、fpsにRMagickみたいな制約は無いので30に変えています。

appsrc name=audiosrc format=time caps=audio/x-raw,format=F32LE,rate=#{sample_rate},channels=#{waveform.shape[0]},layout=interleaved
    ! audioconvert ! avenc_aac ! aacparse ! queue ! mux.

は appsrc という特別なソース（Rubyコードでデータを生成するソース）から出発して、 ! で繋げた audioconvert （オーディオ変換の調整役）、 avenc_aac （MP4で使うためのAACオーディオにエンコード）、 aacparse （AACオーディオのパース）、 queue (キューに積む)と繋げたパイプラインです。最後に mux. （最後の . が大事）と書くことで、一行目の mp4mux に繋げるという意味になります。

因みに、オーディオファイルは元々あるのだから appsrc から流し込むのではなくてそのまま使えないのか？　と思われると思いますが（僕は思いました）、WAVEファイルはそのままではMP4に入れられないのでAACに変換するパイプラインはどうせ作らないといけません。だったらまあ、どうせRubyでオーディオデータの処理もするのだから、Rubyから流し込もうかなと思った次第です。どちらでもいけます。

動画も同様です。雰囲気で読んでください。

最後が

mp4mux name=mux ! filesink location=gstreamer.mp4

となっており、二つのパイプラインを合流させてMP4動画にまとめます。 gstreamer.mp4 という名前でファイルに保存しています。

ごにょごにょとやってこのパイプラインを動かし、そこに appsrc からオーディオデータと画像データを流し込んでやれば動画にしてくれるというわけです。

パイプラインの開始と終了

GStreamerのパイプラインの動かし方です。Cライブラリーの薄いラッパーで、ちょっとRubyらしくないかも知れません。

# 開始状態にする
pipeline.set_state Gst::State::PLAYING

# オーディオと画像を流し込む処理

# EOSかエラーが届いたら……
pipeline.bus.timed_pop_filtered Gst::CLOCK_TIME_NONE, Gst::MessageType::EOS | Gst::MessageType::ERROR
# 終了状態にする
pipeline.set_state Gst::State::NULL

というのが開始と終了です。この間にデータを流し込む処理を書きます。

オーディオと画像の作成

audio_src = pipeline.get_by_name("audiosrc")
video_src = pipeline.get_by_name("videosrc")

frame_duration = Gst::SECOND / fps

num_frames.times do |i|
  start_frame = i * samples_per_frame
  end_frame = start_frame + samples_per_frame - 1
  indices = Numo::Int32.cast(Numo::DFloat.linspace(start_frame, end_frame, w).round)
  data = Numo::SFloat.cast(waveform[indices])
  y_data = (1.0 - data) * (255.to_f / 2)
  coverage = Numo::SFloat.zeros(256, w)
  (w - 1).times do |i|
    segment = line_brightness(0, y_data[i], 1, y_data[i + 1])
    coverage[true, i]     += segment[true, 0]
    coverage[true, i + 1] += segment[true, 1]
  end
  coverage[coverage > 1.0] = 1.0
  image = coverage.expand_dims(2) * color
  image = Numo::UInt8.cast(image.round)

  pts = i * frame_duration

  samples = Numo::SFloat.cast(waveform[0, start_frame..end_frame])
  audio_buf = Gst::Buffer.new(nil, samples.byte_size, nil)
  audio_buf.fill 0, samples.to_binary
  audio_buf.pts = pts
  audio_buf.duration = frame_duration
  audio_src.push_buffer audio_buf

  video_buf = Gst::Buffer.new(nil, image.byte_size, nil)
  video_buf.fill 0, image.to_binary
  video_buf.pts = pts
  video_buf.duration = frame_duration
  video_src.push_buffer video_buf
end

audio_src.end_of_stream
video_src.end_of_stream

前半はさっきやった画像の生成で、元のサンプルから、フレームごとに該当する秒数の箇所を使っています。

後半はオーディオと画像それぞれで Numo::NArray#to_binary で String にして、それを元にGStreamerのバッファーを組み立て、パイプラインに流し込んでいます。Numo::NArray -> GStreamerも簡単にできるといいのだけれど……。

これらをまとめるとこうなります。

require "gstreamer"

fps = 30
duration = waveform.shape[1].to_f / sample_rate
samples_per_frame = (sample_rate.to_f / fps).round
num_frames = (duration * fps).ceil

pipeline = Gst.parse_launch(<<~EOP)
  mp4mux name=mux ! filesink location=gstreamer.mp4
  
  appsrc name=audiosrc format=time caps=audio/x-raw,format=F32LE,rate=#{sample_rate},channels=#{waveform.shape[0]},layout=interleaved
    ! audioconvert ! avenc_aac ! aacparse ! queue ! mux.

  appsrc name=videosrc format=time caps=video/x-raw,format=RGB,width=#{w},height=256,framerate=#{fps}/1
    ! videoconvert ! x264enc ! h264parse ! queue ! mux.
EOP

pipeline.set_state Gst::State::PLAYING

audio_src = pipeline.get_by_name("audiosrc")
video_src = pipeline.get_by_name("videosrc")

frame_duration = Gst::SECOND / fps

num_frames.times do |i|
  start_frame = i * samples_per_frame
  end_frame = start_frame + samples_per_frame - 1
  indices = Numo::Int32.cast(Numo::DFloat.linspace(start_frame, end_frame, w).round)
  data = Numo::SFloat.cast(waveform[indices])
  y_data = (1.0 - data) * (255.to_f / 2)
  coverage = Numo::SFloat.zeros(256, w)
  (w - 1).times do |i|
    segment = line_brightness(0, y_data[i], 1, y_data[i + 1])
    coverage[true, i]     += segment[true, 0]
    coverage[true, i + 1] += segment[true, 1]
  end
  coverage[coverage > 1.0] = 1.0
  image = coverage.expand_dims(2) * color
  image = Numo::UInt8.cast(image.round)

  pts = i * frame_duration

  samples = Numo::SFloat.cast(waveform[0, start_frame..end_frame])
  audio_buf = Gst::Buffer.new(nil, samples.byte_size, nil)
  audio_buf.fill 0, samples.to_binary
  audio_buf.pts = pts
  audio_buf.duration = frame_duration
  audio_src.push_buffer audio_buf

  video_buf = Gst::Buffer.new(nil, image.byte_size, nil)
  video_buf.fill 0, image.to_binary
  video_buf.pts = pts
  video_buf.duration = frame_duration
  video_src.push_buffer video_buf
end

audio_src.end_of_stream
video_src.end_of_stream

pipeline.bus.timed_pop_filtered Gst::CLOCK_TIME_NONE, Gst::MessageType::EOS | Gst::MessageType::ERROR
pipeline.set_state Gst::State::NULL
（音声入りです。）

おお、それっぽい！

This article is under the CC-BY-SA license.

Log in, or use your Fediverse account to interact with this article

Comments

No comments yet. Be the first to react!