並列画像処理

Mix.install([
  {:nx, "~> 0.9"},
  {:evision, "~> 0.2"},
  {:flow, "~> 1.2"},
  {:req, "~> 0.5"},
  {:kino, "~> 0.14"},
  {:benchee, "~> 1.3"}
])

画像の取得

# 指定したパスにダウンロードする
img_path = "rwakabay.jpg"

Req.get!(
  "https://www.elixirconf.eu/assets/images/ryo-wakabayashi.jpg",
  output: File.stream!(img_path)
)

img_mat = Evision.imread(img_path)

単体処理

proc = fn mat ->
  mat
  |> Evision.threshold(127, 255, Evision.Constant.cv_THRESH_BINARY())
  |> elem(1)
  |> Evision.rectangle({50, 10}, {125, 60}, {255, 0, 0})
  |> Evision.rectangle({250, 60}, {325, 110}, {0, 255, 0}, thickness: -1)
  |> Evision.rectangle({150, 120}, {225, 320}, {0, 0, 255},
    thickness: 5,
    lineType: Evision.Constant.cv_LINE_4()
  )
  |> Evision.ellipse({300, 300}, {100, 200}, 30, 0, 360, {255, 255, 0}, thickness: 3)
end

proc.(img_mat)

画像をコピー

file_ext = Path.extname(img_path)
file_basename = Path.basename(img_path, file_ext)

{file_basename, file_ext}

# コピー数
copy_count = 128

file_path_list =
  img_mat
  |> List.duplicate(copy_count)
  |> Enum.with_index()
  |> Enum.map(fn {copied_img_mat, index} ->
    filename = "#{file_basename}_p_#{index}#{file_ext}"

    Evision.imwrite(filename, copied_img_mat)

    filename
  end)

# コピーしたファイル先頭6件を読込
file_path_list
|> Enum.slice(0..5)
|> Enum.map(fn filename ->
  img = Evision.imread(filename)

  [filename, img]
  |> Kino.Layout.grid(columns: 1)
end)
|> Kino.Layout.grid(columns: 3)

逐次処理

# 存在するファイルを取得
stream =
  Stream.unfold(0, fn counter -> {counter, counter + 1} end)
  |> Stream.map(&{&1, "#{file_basename}_p_#{&1}#{file_ext}"})
  |> Stream.take_while(fn {_, filename} -> File.exists?(filename) end)

# Enum.map で処理
enum_proc = fn stream ->
  stream
  |> Enum.map(fn {_, filename} ->
    {
      filename,
      filename |> Evision.imread() |> proc.()
    }
  end)
end

imgs_tuple = enum_proc.(stream)

# 先頭6件を表示
imgs_tuple
|> Enum.slice(0..5)
|> Enum.map(fn {filename, img} ->
  [filename, img]
  |> Kino.Layout.grid(columns: 1)
end)
|> Kino.Layout.grid(columns: 3)

並列処理

# Flow.map で処理
flow_proc = fn stream, stages ->
  stream
  |> Flow.from_enumerable(stages: stages, max_demand: 1)
  |> Flow.map(fn {_, filename} ->
    {
      filename,
      filename |> Evision.imread() |> proc.()
    }
  end)
  |> Enum.to_list()
end

imgs_tuple = flow_proc.(stream, 4)

# 先頭6件を表示
imgs_tuple
|> Enum.slice(0..5)
|> Enum.map(fn {filename, img} ->
  [filename, img]
  |> Kino.Layout.grid(columns: 1)
end)
|> Kino.Layout.grid(columns: 3)

速度比較

Benchee.run(%{
  "enum" => fn -> enum_proc.(stream) end,
  "flow 1" => fn -> flow_proc.(stream, 1) end,
  "flow 2" => fn -> flow_proc.(stream, 2) end,
  "flow 4" => fn -> flow_proc.(stream, 4) end,
  "flow 8" => fn -> flow_proc.(stream, 8) end
})