Parallel Image Processing by Flow

parallel_image_processing.exs

System.put_env("EVISION_PREFER_PRECOMPILED", "true") # Remove if you use a platform on which Evision does not provide a pre-compiled library.

System.put_env("EVISION_PRECOMPILED_CACHE_DIR", "#{System.user_home!()}/.cache")

Mix.install(
  [
    {:nx, "~> 0.3"},
    {:exla, "~> 0.3"},
    {:evision, "~> 0.1.2", github: "cocoa-xu/evision", tag: "v0.1.2"},
    {:flow, "~> 1.2"} # Add Flow
  ],
  config: [
    nx: [default_backend: EXLA.Backend]
  ]
)

src_file = "ZACKY-3000.jpg"
src_file_ext = Path.extname(src_file)
src_file_basename = Path.basename(src_file, src_file_ext)

dst_files =
  Stream.unfold(0, fn counter -> {counter, counter + 1} end)
  |> Stream.map(& "#{src_file_basename}_b_#{&1}#{src_file_ext}")

imgs =
  Stream.unfold(0, fn counter -> {counter, counter + 1} end)  # src_files is generated by Stream
  |> Stream.map(& {&1, "#{src_file_basename}_#{&1}#{src_file_ext}"})
  |> Stream.take_while(fn {_, f} -> File.exists?(f) end)
  |> Flow.from_enumerable() # Make the stream parallel
  |> Flow.map(fn {c, f} -> {c, Evision.imread!(f)} end)
  |> Flow.map(fn {f, img} ->
    {
      f,
      Evision.threshold!(img, 127, 255, Evision.cv_THRESH_BINARY) |> elem(1) # image processing for each
    }
  end)
  |> Enum.sort(fn {f1, _}, {f2, _} -> f1 < f2 end) # sorting by the counter

Enum.zip(imgs, dst_files)
|> Enum.map(fn {{_, img}, dst_file} -> Evision.imwrite!(dst_file, img) end)