如何用Python案例实现批量图像去噪？

我来为您提供几个批量图像去噪的Python案例,从简单到高级的实现方法。

基础实现：使用OpenCV的均值滤波

import cv2
import os
import glob
import numpy as np
from tqdm import tqdm
def batch_denoise_mean(input_dir, output_dir, kernel_size=3):
    """
    使用均值滤波批量去噪
    """
    # 创建输出目录
    os.makedirs(output_dir, exist_ok=True)
    # 获取所有图片文件
    image_extensions = ['*.jpg', '*.jpeg', '*.png', '*.bmp', '*.tiff']
    image_files = []
    for ext in image_extensions:
        image_files.extend(glob.glob(os.path.join(input_dir, ext)))
    print(f"找到 {len(image_files)} 张图片")
    for img_path in tqdm(image_files, desc="处理中"):
        # 读取图像
        img = cv2.imread(img_path)
        if img is None:
            print(f"无法读取: {img_path}")
            continue
        # 应用均值滤波
        denoised = cv2.blur(img, (kernel_size, kernel_size))
        # 保存结果
        filename = os.path.basename(img_path)
        output_path = os.path.join(output_dir, f"denoised_{filename}")
        cv2.imwrite(output_path, denoised)
    print("批量去噪完成！")
# 使用示例
# batch_denoise_mean('input_images/', 'output_images/', kernel_size=5)

使用多种去噪方法的完整实现

import cv2
import os
import numpy as np
from glob import glob
from tqdm import tqdm
import matplotlib.pyplot as plt
class ImageDenoiser:
    """图像去噪器类"""
    def __init__(self):
        self.methods = {
            'mean': self.mean_filter,
            'median': self.median_filter,
            'gaussian': self.gaussian_filter,
            'bilateral': self.bilateral_filter,
            'nlm': self.nlm_denoise
        }
    def mean_filter(self, img, kernel_size=3):
        """均值滤波"""
        return cv2.blur(img, (kernel_size, kernel_size))
    def median_filter(self, img, kernel_size=3):
        """中值滤波"""
        return cv2.medianBlur(img, kernel_size)
    def gaussian_filter(self, img, kernel_size=3, sigma=0):
        """高斯滤波"""
        return cv2.GaussianBlur(img, (kernel_size, kernel_size), sigma)
    def bilateral_filter(self, img, d=9, sigma_color=75, sigma_space=75):
        """双边滤波"""
        return cv2.bilateralFilter(img, d, sigma_color, sigma_space)
    def nlm_denoise(self, img, h=10, template_window_size=7, search_window_size=21):
        """非局部均值去噪"""
        if len(img.shape) == 3:
            return cv2.fastNlMeansDenoisingColored(
                img, None, h, h, 
                template_window_size, search_window_size
            )
        else:
            return cv2.fastNlMeansDenoising(
                img, None, h, 
                template_window_size, search_window_size
            )
    def batch_denoise(self, input_dir, output_dir, method='nlm', **kwargs):
        """
        批量去噪主函数
        """
        os.makedirs(output_dir, exist_ok=True)
        # 获取图片文件
        image_files = []
        for ext in ['*.jpg', '*.jpeg', '*.png', '*.bmp']:
            image_files.extend(glob(os.path.join(input_dir, ext)))
        if not image_files:
            print(f"在 {input_dir} 中没有找到图片文件")
            return
        print(f"找到 {len(image_files)} 张图片，使用 {method} 方法去噪")
        # 统计信息
        results = []
        for img_path in tqdm(image_files, desc="处理进度"):
            try:
                # 读取图像
                img = cv2.imread(img_path)
                if img is None:
                    continue
                # 获取去噪方法
                denoise_func = self.methods.get(method)
                if denoise_func is None:
                    print(f"不支持的方法: {method}")
                    return
                # 执行去噪
                denoised_img = denoise_func(img, **kwargs)
                # 保存结果
                filename = os.path.basename(img_path)
                name, ext = os.path.splitext(filename)
                output_path = os.path.join(output_dir, f"{name}_{method}{ext}")
                cv2.imwrite(output_path, denoised_img)
                # 记录结果
                results.append({
                    'filename': filename,
                    'input_shape': img.shape,
                    'method': method
                })
            except Exception as e:
                print(f"处理 {img_path} 时出错: {str(e)}")
        # 打印统计信息
        print(f"\n处理完成！成功处理 {len(results)} 张图片")
        print(f"结果保存在: {output_dir}")
        return results
# 使用示例
if __name__ == "__main__":
    denoiser = ImageDenoiser()
    # 1. 使用NLM方法去噪
    results = denoiser.batch_denoise(
        input_dir='input_images/',
        output_dir='output_images/nlm/',
        method='nlm',
        h=15
    )
    # 2. 使用中值滤波
    results = denoiser.batch_denoise(
        input_dir='input_images/',
        output_dir='output_images/median/',
        method='median',
        kernel_size=5
    )

高级实现：自适应去噪 + 质量评估

import cv2
import os
import numpy as np
from glob import glob
from tqdm import tqdm
from skimage import metrics
from concurrent.futures import ThreadPoolExecutor
import json
class AdvancedImageDenoiser:
    """高级图像去噪器"""
    def __init__(self, num_threads=4):
        self.num_threads = num_threads
    def estimate_noise_level(self, img):
        """估计噪声水平"""
        if len(img.shape) == 3:
            gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
        else:
            gray = img
        # 使用拉普拉斯算子估计噪声
        laplacian = cv2.Laplacian(gray, cv2.CV_64F)
        noise_level = np.std(laplacian)
        return noise_level
    def adaptive_denoise(self, img):
        """自适应去噪"""
        noise_level = self.estimate_noise_level(img)
        if noise_level < 10:
            # 低噪声，轻微处理
            return cv2.GaussianBlur(img, (3, 3), 0.5)
        elif noise_level < 30:
            # 中等噪声
            return cv2.fastNlMeansDenoisingColored(
                img, None, 10, 10, 7, 21
            )
        else:
            # 高噪声，强力去噪
            denoised = cv2.medianBlur(img, 5)
            return cv2.fastNlMeansDenoisingColored(
                denoised, None, 15, 15, 7, 21
            )
    def calculate_psnr(self, img1, img2):
        """计算PSNR"""
        return metrics.peak_signal_noise_ratio(img1, img2)
    def calculate_ssim(self, img1, img2):
        """计算SSIM"""
        if len(img1.shape) == 3:
            img1 = cv2.cvtColor(img1, cv2.COLOR_BGR2GRAY)
            img2 = cv2.cvtColor(img2, cv2.COLOR_BGR2GRAY)
        return metrics.structural_similarity(img1, img2)
    def process_single_image(self, img_path, output_dir):
        """处理单张图片"""
        try:
            # 读取图像
            img = cv2.imread(img_path)
            if img is None:
                return None
            # 自适应去噪
            denoised = self.adaptive_denoise(img)
            # 计算质量指标
            psnr = self.calculate_psnr(img, denoised)
            ssim = self.calculate_ssim(img, denoised)
            noise_level = self.estimate_noise_level(img)
            # 保存结果
            filename = os.path.basename(img_path)
            name, ext = os.path.splitext(filename)
            output_path = os.path.join(output_dir, f"{name}_denoised{ext}")
            cv2.imwrite(output_path, denoised)
            return {
                'filename': filename,
                'input_shape': img.shape,
                'noise_level': noise_level,
                'psnr': psnr,
                'ssim': ssim,
                'output_path': output_path
            }
        except Exception as e:
            print(f"处理 {img_path} 时出错: {str(e)}")
            return None
    def batch_denoise_advanced(self, input_dir, output_dir, save_metadata=True):
        """
        批量去噪（高级版）
        """
        os.makedirs(output_dir, exist_ok=True)
        # 获取图片文件
        image_files = []
        for ext in ['*.jpg', '*.jpeg', '*.png', '*.bmp']:
            image_files.extend(glob(os.path.join(input_dir, ext)))
        if not image_files:
            print(f"在 {input_dir} 中没有找到图片文件")
            return
        print(f"找到 {len(image_files)} 张图片，开始批量去噪...")
        # 多线程处理
        all_results = []
        with ThreadPoolExecutor(max_workers=self.num_threads) as executor:
            futures = []
            for img_path in image_files:
                future = executor.submit(
                    self.process_single_image, 
                    img_path, 
                    output_dir
                )
                futures.append(future)
            # 显示进度
            for future in tqdm(futures, desc="处理进度"):
                result = future.result()
                if result:
                    all_results.append(result)
        # 统计信息
        if all_results:
            avg_noise = np.mean([r['noise_level'] for r in all_results])
            avg_psnr = np.mean([r['psnr'] for r in all_results])
            avg_ssim = np.mean([r['ssim'] for r in all_results])
            print(f"\n处理完成！")
            print(f"成功处理 {len(all_results)}/{len(image_files)} 张图片")
            print(f"平均噪声水平: {avg_noise:.2f}")
            print(f"平均PSNR: {avg_psnr:.2f} dB")
            print(f"平均SSIM: {avg_ssim:.4f}")
            # 保存元数据
            if save_metadata:
                metadata_path = os.path.join(output_dir, 'metadata.json')
                with open(metadata_path, 'w', encoding='utf-8') as f:
                    json.dump({
                        'total_processed': len(all_results),
                        'total_files': len(image_files),
                        'average_metrics': {
                            'noise_level': float(avg_noise),
                            'psnr': float(avg_psnr),
                            'ssim': float(avg_ssim)
                        },
                        'results': all_results
                    }, f, indent=2, ensure_ascii=False)
                print(f"元数据已保存至: {metadata_path}")
        return all_results
    def create_comparison_collage(self, input_dir, output_dir, num_samples=9):
        """
        创建对比图拼贴
        """
        # 获取图片文件
        image_files = glob(os.path.join(input_dir, '*'))
        selected_files = image_files[:num_samples]
        if not selected_files:
            return
        n_cols = 3
        n_rows = (len(selected_files) + n_cols - 1) // n_cols
        fig, axes = plt.subplots(n_rows, n_cols * 2, figsize=(15, 5 * n_rows))
        axes = axes.flatten()
        for i, img_path in enumerate(selected_files):
            # 原图
            img = cv2.imread(img_path)
            img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
            # 去噪后
            denoised = self.adaptive_denoise(img)
            denoised_rgb = cv2.cvtColor(denoised, cv2.COLOR_BGR2RGB)
            # 显示原图
            axes[2*i].imshow(img_rgb)
            axes[2*i].set_title(f'原图 {i+1}')
            axes[2*i].axis('off')
            # 显示去噪后
            axes[2*i+1].imshow(denoised_rgb)
            axes[2*i+1].set_title(f'去噪后 {i+1}')
            axes[2*i+1].axis('off')
        plt.tight_layout()
        plt.savefig(os.path.join(output_dir, 'comparison_collage.png'), dpi=150)
        plt.close()
        print(f"对比图已保存至: {os.path.join(output_dir, 'comparison_collage.png')}")
# 使用示例
if __name__ == "__main__":
    # 创建高级去噪器
    denoiser = AdvancedImageDenoiser(num_threads=4)
    # 批量去噪
    results = denoiser.batch_denoise_advanced(
        input_dir='input_images/',
        output_dir='output_images/advanced/',
        save_metadata=True
    )
    # 创建对比图
    denoiser.create_comparison_collage(
        input_dir='input_images/',
        output_dir='output_images/advanced/',
        num_samples=6
    )

带GUI的批量去噪工具

import tkinter as tk
from tkinter import ttk, filedialog, messagebox
import cv2
import os
from threading import Thread
class DenoiseApp:
    """批量去噪GUI应用"""
    def __init__(self, root):
        self.root = root
        self.root.title("批量图像去噪工具")
        self.root.geometry("600x500")
        # 变量
        self.input_dir = tk.StringVar()
        self.output_dir = tk.StringVar()
        self.method = tk.StringVar(value="nlm")
        self.kernel_size = tk.IntVar(value=3)
        self.progress = tk.DoubleVar()
        self.setup_ui()
    def setup_ui(self):
        """设置UI界面"""
        # 主框架
        main_frame = ttk.Frame(self.root, padding="10")
        main_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
        # 输入目录
        ttk.Label(main_frame, text="输入目录:").grid(row=0, column=0, sticky=tk.W, pady=5)
        ttk.Entry(main_frame, textvariable=self.input_dir, width=50).grid(row=0, column=1, padx=5)
        ttk.Button(main_frame, text="浏览...", command=self.browse_input).grid(row=0, column=2)
        # 输出目录
        ttk.Label(main_frame, text="输出目录:").grid(row=1, column=0, sticky=tk.W, pady=5)
        ttk.Entry(main_frame, textvariable=self.output_dir, width=50).grid(row=1, column=1, padx=5)
        ttk.Button(main_frame, text="浏览...", command=self.browse_output).grid(row=1, column=2)
        # 去噪方法
        ttk.Label(main_frame, text="去噪方法:").grid(row=2, column=0, sticky=tk.W, pady=5)
        methods = ['mean', 'median', 'gaussian', 'bilateral', 'nlm']
        method_combo = ttk.Combobox(main_frame, textvariable=self.method, values=methods)
        method_combo.grid(row=2, column=1, sticky=tk.W, padx=5)
        # 核大小
        ttk.Label(main_frame, text="核大小:").grid(row=3, column=0, sticky=tk.W, pady=5)
        ttk.Spinbox(main_frame, from_=3, to=15, increment=2, 
                   textvariable=self.kernel_size, width=10).grid(row=3, column=1, sticky=tk.W, padx=5)
        # 进度条
        self.progress_bar = ttk.Progressbar(main_frame, variable=self.progress, 
                                           maximum=100, length=400)
        self.progress_bar.grid(row=4, column=0, columnspan=3, pady=20)
        # 状态标签
        self.status_label = ttk.Label(main_frame, text="就绪")
        self.status_label.grid(row=5, column=0, columnspan=3)
        # 处理按钮
        ttk.Button(main_frame, text="开始处理", command=self.start_processing).grid(row=6, column=1, pady=20)
    def browse_input(self):
        """浏览输入目录"""
        directory = filedialog.askdirectory()
        if directory:
            self.input_dir.set(directory)
    def browse_output(self):
        """浏览输出目录"""
        directory = filedialog.askdirectory()
        if directory:
            self.output_dir.set(directory)
    def start_processing(self):
        """开始处理"""
        if not self.input_dir.get() or not self.output_dir.get():
            messagebox.showerror("错误", "请选择输入和输出目录")
            return
        # 在新线程中处理
        thread = Thread(target=self.process_images)
        thread.daemon = True
        thread.start()
    def process_images(self):
        """处理图像"""
        try:
            denoiser = ImageDenoiser()
            input_dir = self.input_dir.get()
            output_dir = self.output_dir.get()
            # 获取图片文件
            image_files = []
            for ext in ['*.jpg', '*.jpeg', '*.png', '*.bmp']:
                image_files.extend(glob(os.path.join(input_dir, ext)))
            if not image_files:
                self.root.after(0, lambda: messagebox.showwarning("警告", "没有找到图片文件"))
                return
            os.makedirs(output_dir, exist_ok=True)
            for i, img_path in enumerate(image_files):
                # 更新进度
                progress = (i + 1) / len(image_files) * 100
                self.root.after(0, lambda v=progress: self.progress.set(v))
                self.root.after(0, lambda i=i, total=len(image_files): 
                               self.status_label.config(text=f"处理中: {i+1}/{total}"))
                # 处理图像
                img = cv2.imread(img_path)
                if img is None:
                    continue
                # 获取去噪方法
                method = self.method.get()
                if method == 'mean':
                    denoised = cv2.blur(img, (self.kernel_size.get(), self.kernel_size.get()))
                elif method == 'median':
                    denoised = cv2.medianBlur(img, self.kernel_size.get())
                elif method == 'gaussian':
                    denoised = cv2.GaussianBlur(img, (self.kernel_size.get(), self.kernel_size.get()), 0)
                elif method == 'bilateral':
                    denoised = cv2.bilateralFilter(img, 9, 75, 75)
                else:  # nlm
                    denoised = cv2.fastNlMeansDenoisingColored(img, None, 10, 10, 7, 21)
                # 保存结果
                filename = os.path.basename(img_path)
                output_path = os.path.join(output_dir, f"denoised_{filename}")
                cv2.imwrite(output_path, denoised)
            # 处理完成
            self.root.after(0, lambda: self.progress.set(100))
            self.root.after(0, lambda: self.status_label.config(text="处理完成！"))
            self.root.after(0, lambda: messagebox.showinfo("完成", "批量去噪处理完成！"))
        except Exception as e:
            self.root.after(0, lambda: messagebox.showerror("错误", str(e)))
            self.root.after(0, lambda: self.status_label.config(text="处理出错"))
# 运行GUI应用
if __name__ == "__main__":
    root = tk.Tk()
    app = DenoiseApp(root)
    root.mainloop()

使用方法

1. 简单使用：

   # 安装依赖
   pip install opencv-python numpy tqdm scikit-image

运行基本去噪

denoiser = ImageDenoiser() results = denoiser.batch_denoise('input/', 'output/', method='nlm')

2. **高级使用**：
```python
# 自适应去噪 + 质量评估
denoiser = AdvancedImageDenoiser()
results = denoiser.batch_denoise_advanced('input/', 'output/')

3. GUI工具：

   # 运行图形界面
   python denoise_gui.py

这些实现涵盖了从基础到高级的批量图像去噪方案,可以根据具体需求选择合适的实现。