如何在几秒内处理 10k 张图像

手动、重复性任务是我们都讨厌的事情，特别是当知道它们可以自动化的时候。想象一下，你需要用相同的裁剪和调整大小操作编辑一堆图像。对于几张图片，你可能只需打开一个图像编辑器并手动执行。但是，如果对数千或数万张图像执行相同的操作呢？让我们看看如何使用Python和OpenCV自动执行这样的图像处理任务，以及如何优化此数据处理管道以在大量数据集上高效运行。

数据集

本文中我们使用录制的海滩随机视频中提取了 10,000 帧，目标是将图像裁剪为围绕中心的正方形宽高比，然后将图像调整为固定尺寸224x224。

先决条件

如果您想继续，请确保安装以下软件包，例如uv 。源码下载链接：

https://github.com/trflorian/image-processing

uv add opencv-python tqdm

数据加载

让我们首先使用OpenCV逐一加载图像。所有图像都在子文件夹中，并将使用pathlib glob 方法查找此文件夹中的所有png文件。为了显示进度，我使用了tqdm库。通过使用sorted方法，我确保路径已排序，并将glob调用返回的生成器转换为列表。这样，tqdm就知道显示进度条的迭代长度。

  1. from pathlib import Path
  2. from tqdm import tqdm
  3.  
  4. img_paths = Path("images").glob("*.png")
  5.  
  6. for img_path in tqdm(sorted(img_paths)): 
  7. pass

现在我们还可以准备输出目录，并确保它存在。这是我们处理后的图像将存储的地方。

  1. output_path = Path("output")
  2. output_path.mkdir(exist_ok=True, parents=True)

图像处理

为了处理图像，我们定义一个函数。它将输入和输出图像路径作为参数。

  1. def process_image(input_path: Path, output_path: Path) -> None: 
  2.   """   
  3. Image processing pipeline: 
  4.   - Center crop to square aspect ratio 
  5.   - Resize to 224x224
  6.  
  7.     Args:       
  8.             input_path (Path): Path to input image       
  9.             output_path (Path): Path to save processed image   
  10. """

为了实现此功能，我们首先需要用OpenCV加载图像。确保在文件开头导入 opencv 包。

  1. ...
  2.  
  3. import cv2
  4.  
  5.  
  6. defprocess_image(input_path: Path, output_path: Path) -> None:   
  7. ... 
  8.  
  9. # Read image
  10.     img = cv2.imread(str(input_path))

要裁剪图像，我们可以直接在 x 轴上对图像数组进行切片。请记住，OpenCV 图像数组以YXC形状存储：X/Y是从左上角开始的图像的 2D 轴，C是颜色通道。因此 x 轴是图像的第二个索引。为简单起见，我假设图像为横向格式，其宽度 > 高度。

  1. height, width, _ = img.shape
  2. img = img[:, (width - height) // 2 : (width + height) // 2, :]

要调整图像大小，我们可以简单地使用OpenCV 中的resize函数。如果我们不指定插值方法，它将使用双线性插值，这对于这个项目来说很好。

  1. target_size = 224
  2. img = cv2.resize(img, (target_size, target_size))

最后，使用imwrite函数将图像保存到输出文件。

  1. cv2.imwrite(str(output_path), img)

现在我们可以简单地在图像路径循环中调用 process_image 函数。

  1. for img_path in tqdm(sorted(img_paths)):
  2.     process_image(input_path=img_path, output_path=output_path / img_path.name)

如果我在我的计算机上运行这个程序，处理 10,000 张图像需要一分钟多一点的时间。

  1.      4%|█████▏ | 441/10000 [00:02<01:01，154.34it/s]

现在，虽然对于这个数据集大小，等待一分钟仍然是可行的，但对于 10 倍大的数据集，您已经等待了 10 分钟。我们可以通过并行化此过程做得更好。如果您在运行当前程序时查看资源使用情况，您会注意到只有一个核心的利用率为100%。该程序仅使用一个核心！

跨多核并行

为了让我们的程序使用更多可用内核，我们需要使用 Python 中的一项功能，称为多处理。由于全局解释器锁 (GIL)，单个 Python 进程无法真正并行运行任务（除非禁用 GIL，这可以在 Python≥3.13 中完成）。我们需要做的是生成由主 Python 程序管理的多个 Python 进程（因此称为多处理）。

为了实现这一点，我们可以使用内置的 Python 模块multiprocessing和parallel。理论上，我们可以手动生成 Python 进程，同时确保提交的进程数不超过我们的核心数。由于我们的进程受 CPU 限制，因此增加进程并不会带来速度提升，因为它们必须等待。事实上，在某一时刻，进程间切换的开销会超过并行化的优势。

为了管理 Python 进程，我们可以使用ProcessPoolexecutor。这将保留一个 Python 进程池，而不是为每个提交的任务完全销毁并重新启动每个进程。默认情况下，它将使用与可用逻辑 CPU数量一样多的池，该数量是从os.process_cpu_count()检索的。因此，默认情况下，它将为我的 CPU 的每个核心生成一个进程，在我的情况下是 20 个。您还可以提供max_workers参数来指定要在池中生成的进程数。

  1. from concurrent.futures import ProcessPoolexecutor
  2.  
  3. ...
  4.  
  5. output_paths = [output_path / img_path.name for img_path in img_paths]
  6.  
  7. with ProcessPoolexecutor() as executor:   
  8. all_processes = executor.map( 
  9.       process_image, 
  10.       img_paths,
  11.         output_paths, 
  12.   )
  13. for _ in tqdm(all_processes, total=len(img_paths)):
  14. pass

我们使用上下文管理器（with语句）创建进程池执行器，这将确保即使在执行过程中发生异常，进程也会被清理。然后我们使用map函数为每个输入img_paths和output_paths创建一个进程。最后通过使用tqdm包装all_processes的迭代，我们可以获得已完成进程的进度条。

  1. 18% |█████|1760/10000[00:00<00:04，1857.23it/s]

现在，如果您运行该程序并再次检查 CPU 利用率，您将看到所有核心都已使用！进度条还显示了我们的迭代速度如何提高。

比较

为了快速检查完整性，我绘制了使用不同数量的并行化处理 1000 张图像的时间图，从单个工作器场景开始，然后将工作器数量增加到我机器内核数量的两倍。下图表明最佳情况接近 CPU 内核数量。从 1 个工作器到多个工作器，性能急剧上升，而当工作器数量超过 CPU 内核数量时，性能略有下降。

完整代码下载：

  1. https://github.com/trflorian/image-processing

 

本文转载自【创视智能】

注：文章版权归原作者所有，本文内容、图片、视频来自网络，仅供交流学习之用，如涉及版权等问题，请您告知，我们将及时处理。