Files
Eric Rollei d6dd6470cf Add complete Eric Image Processing Nodes implementation
Features:
- 40+ image processing algorithms
- Traditional denoising: Wavelet, NLM, BM3D (GPU-accelerated)
- AI models: DnCNN, SwinIR, SCUNet, NAFNet, Noise-DA
- DeepInv diffusion denoising integration
- Advanced film grain detection and processing
- Frequency domain processing
- Richardson-Lucy deconvolution
- Wiener filter
- External DeepInv service with FastAPI
- Comprehensive documentation and guides
2025-11-06 23:33:19 -08:00

1174 lines
41 KiB
Python

"""
Wavelet denoising implementation using PyWavelets
Supports multiple wavelet types and thresholding methods
Enhanced with GPU acceleration for CUDA-capable systems
"""
import numpy as np
import pywt
import cv2
from typing import Optional, Union, Tuple
from skimage.restoration import estimate_sigma
# GPU acceleration support
try:
import cupy as cp
from cupyx.scipy import ndimage as cp_ndimage
CUPY_AVAILABLE = True
except ImportError:
CUPY_AVAILABLE = False
# Import local GPU utilities
from .gpu_utils import (
to_gpu, to_cpu, can_use_gpu,
gpu_memory_info, cleanup_gpu_memory,
estimate_gpu_memory_usage
)
def wavelet_denoise(
image: np.ndarray,
wavelet: str = 'db8',
sigma: Optional[float] = None,
mode: str = 'soft',
method: str = 'BayesShrink',
levels: Optional[int] = None,
multichannel: bool = True,
rescale_sigma: bool = True
) -> np.ndarray:
"""
Advanced wavelet denoising with multiple threshold selection methods
Args:
image: Input image as numpy array [H, W, C] with values 0-255
wavelet: Wavelet type ('db8', 'db4', 'bior2.2', 'coif2', 'haar')
- 'db8': Daubechies 8 - Good all-around choice, smooth reconstruction
- 'db4': Daubechies 4 - Faster, good for natural images
- 'bior2.2': Biorthogonal - Good for images with edges
- 'coif2': Coiflets - Good reconstruction properties
- 'haar': Haar - Fastest, good for cartoon-like images
sigma: Noise standard deviation (auto-estimated if None)
- Typical range: 5-25 for moderate noise, 25-50 for heavy noise
mode: Thresholding mode ('soft', 'hard')
- 'soft': Gradual transition, smoother results (recommended)
- 'hard': Sharp cutoff, preserves edges better but may create artifacts
method: Threshold selection method ('BayesShrink', 'VisuShrink', 'manual')
- 'BayesShrink': Adaptive, optimal for most natural images
- 'VisuShrink': Conservative, good for preserving details
- 'manual': Use sigma parameter directly as threshold
levels: Decomposition levels (auto-calculated if None)
- Typical range: 3-6 levels, more levels = more denoising
multichannel: Process color channels independently
rescale_sigma: Whether to rescale sigma estimate (recommended True)
Returns:
Denoised image as numpy array [H, W, C] with values 0-255
"""
# Validate inputs
if image.ndim not in [2, 3]:
raise ValueError("Image must be 2D (grayscale) or 3D (color)")
# Convert to float for processing
img_float = image.astype(np.float32) / 255.0
# Handle grayscale vs color
if img_float.ndim == 2:
return _denoise_single_channel(
img_float, wavelet, sigma, mode, method, levels, rescale_sigma
) * 255.0
# Process color image
if multichannel:
# Process each channel independently
denoised_channels = []
for c in range(img_float.shape[2]):
denoised_channel = _denoise_single_channel(
img_float[:, :, c], wavelet, sigma, mode, method, levels, rescale_sigma
)
denoised_channels.append(denoised_channel)
result = np.stack(denoised_channels, axis=2)
else:
# Convert to grayscale and denoise
gray = np.mean(img_float, axis=2)
denoised_gray = _denoise_single_channel(
gray, wavelet, sigma, mode, method, levels, rescale_sigma
)
# Broadcast back to color
result = np.stack([denoised_gray] * img_float.shape[2], axis=2)
# Convert back to uint8 range
result = np.clip(result * 255.0, 0, 255).astype(np.uint8)
return result
def _denoise_single_channel(
image: np.ndarray,
wavelet: str,
sigma: Optional[float],
mode: str,
method: str,
levels: Optional[int],
rescale_sigma: bool
) -> np.ndarray:
"""Denoise a single channel image using wavelets"""
# Auto-calculate decomposition levels if not specified
if levels is None:
# Calculate max levels manually (compatible with all PyWavelets versions)
min_size = min(image.shape)
# More conservative level calculation for better performance
levels = min(5, max(2, int(np.log2(min_size)) - 2))
else:
# Validate user-provided levels
min_size = min(image.shape)
max_levels = int(np.log2(min_size)) - 1
if levels > max_levels:
print(f"Warning: Requested levels ({levels}) exceeds maximum for image size ({max_levels}). Using {max_levels} instead.")
levels = max_levels
elif levels < 1:
print(f"Warning: Levels must be >= 1. Using 1 instead of {levels}.")
levels = 1
# Estimate noise if sigma not provided
if sigma is None:
sigma = estimate_sigma(image, channel_axis=None)
if rescale_sigma:
# Improved rescaling factor based on wavelet type
if wavelet.startswith('db'):
sigma *= 1.1 # Daubechies wavelets
elif wavelet.startswith('bior'):
sigma *= 0.9 # Biorthogonal wavelets
else:
sigma *= 1.0 # Default
# Perform wavelet decomposition - use correct parameter name
coeffs = pywt.wavedec2(image, wavelet, level=levels, mode='symmetric')
# Apply thresholding based on method
if method == 'BayesShrink':
coeffs_thresh = _bayes_shrink_threshold(coeffs, sigma)
elif method == 'VisuShrink':
threshold = sigma * np.sqrt(2 * np.log(image.size))
coeffs_thresh = _apply_threshold(coeffs, threshold, mode)
elif method == 'SureShrink':
coeffs_thresh = _sure_shrink_threshold(coeffs, sigma, mode)
elif method == 'manual':
# For manual method, use sigma as noise level but calculate appropriate threshold
# This prevents using sigma directly as threshold which is often too aggressive
manual_threshold = sigma * 0.1 # More conservative threshold
coeffs_thresh = _apply_threshold(coeffs, manual_threshold, mode)
else:
raise ValueError(f"Unknown thresholding method: {method}")
# Reconstruct image
denoised = pywt.waverec2(coeffs_thresh, wavelet, mode='symmetric')
# Ensure output is same size as input (handle boundary effects)
if denoised.shape != image.shape:
# Crop to original size
h, w = image.shape
denoised = denoised[:h, :w]
return np.clip(denoised, 0, 1)
def _bayes_shrink_threshold(coeffs, sigma):
"""Apply BayesShrink thresholding to wavelet coefficients"""
coeffs_thresh = [coeffs[0]] # Keep approximation coefficients unchanged
for i in range(1, len(coeffs)):
# For each detail level
detail_coeffs = []
for detail in coeffs[i]: # (cH, cV, cD)
# Estimate signal variance
var_y = np.var(detail)
var_x = max(0, var_y - sigma**2)
if var_x > 0:
# Calculate Bayes threshold
threshold = sigma**2 / np.sqrt(var_x)
# Apply soft thresholding
detail_thresh = pywt.threshold(detail, threshold, mode='soft')
else:
# If no signal estimated, use conservative threshold instead of zeroing
# This prevents over-denoising when noise estimation is poor
conservative_threshold = sigma * 0.5 # Much more conservative
detail_thresh = pywt.threshold(detail, conservative_threshold, mode='soft')
detail_coeffs.append(detail_thresh)
coeffs_thresh.append(tuple(detail_coeffs))
return coeffs_thresh
def _apply_threshold(coeffs, threshold, mode):
"""Apply uniform threshold to all detail coefficients"""
coeffs_thresh = [coeffs[0]] # Keep approximation coefficients
for i in range(1, len(coeffs)):
detail_coeffs = []
for detail in coeffs[i]:
detail_thresh = pywt.threshold(detail, threshold, mode=mode)
detail_coeffs.append(detail_thresh)
coeffs_thresh.append(tuple(detail_coeffs))
return coeffs_thresh
def _sure_shrink_threshold(coeffs, sigma, mode):
"""Apply SureShrink thresholding to wavelet coefficients
SureShrink is a hybrid approach that selects between soft and hard thresholding
based on Stein's Unbiased Risk Estimator (SURE).
"""
coeffs_thresh = [coeffs[0]] # Keep approximation coefficients unchanged
for i in range(1, len(coeffs)):
# For each detail level
detail_coeffs = []
for detail in coeffs[i]: # (cH, cV, cD)
# Calculate threshold using SURE
n = detail.size
sorted_coeffs = np.sort(np.abs(detail.flatten()))
# Calculate risk for each threshold
risks = []
thresholds = sorted_coeffs[::max(1, n//100)] # Sample thresholds
for t in thresholds:
# Number of coefficients below threshold
k = np.sum(np.abs(detail) <= t)
# SURE formula
risk = n - 2*k + np.sum(np.minimum(np.abs(detail), t)**2)
risks.append(risk)
# Find threshold with minimum risk
if risks:
min_idx = np.argmin(risks)
threshold = thresholds[min_idx]
else:
threshold = sigma * np.sqrt(2 * np.log(n)) # Fallback to VisuShrink
# Apply thresholding
detail_thresh = pywt.threshold(detail, threshold, mode=mode)
detail_coeffs.append(detail_thresh)
coeffs_thresh.append(tuple(detail_coeffs))
return coeffs_thresh
def wavelet_denoise_stationary(
image: np.ndarray,
wavelet: str = 'db8',
sigma: Optional[float] = None,
mode: str = 'soft',
method: str = 'BayesShrink',
levels: Optional[int] = None,
multichannel: bool = True,
rescale_sigma: bool = True
) -> np.ndarray:
"""
Stationary (translation-invariant) wavelet denoising
This method uses the Stationary Wavelet Transform (SWT) which is shift-invariant,
producing better results for images with important structures at different positions.
Args:
Same as wavelet_denoise() but uses SWT instead of DWT
Returns:
Denoised image with reduced shift-sensitivity artifacts
"""
# Validate inputs
if image.ndim not in [2, 3]:
raise ValueError("Image must be 2D (grayscale) or 3D (color)")
# Convert to float for processing
img_float = image.astype(np.float32) / 255.0
# Handle grayscale vs color
if img_float.ndim == 2:
result = _denoise_single_channel_stationary(
img_float, wavelet, sigma, mode, method, levels, rescale_sigma
)
else:
if multichannel:
# Process each channel independently
result_channels = []
for c in range(img_float.shape[2]):
denoised_channel = _denoise_single_channel_stationary(
img_float[:, :, c], wavelet, sigma, mode, method, levels, rescale_sigma
)
result_channels.append(denoised_channel)
result = np.stack(result_channels, axis=2)
else:
# Convert to grayscale and denoise
gray = np.mean(img_float, axis=2)
denoised_gray = _denoise_single_channel_stationary(
gray, wavelet, sigma, mode, method, levels, rescale_sigma
)
# Broadcast back to color
result = np.stack([denoised_gray] * img_float.shape[2], axis=2)
# Convert back to uint8 range
result = np.clip(result * 255.0, 0, 255).astype(np.uint8)
return result
def _denoise_single_channel_stationary(
image: np.ndarray,
wavelet: str,
sigma: Optional[float],
mode: str,
method: str,
levels: Optional[int],
rescale_sigma: bool
) -> np.ndarray:
"""Denoise a single channel using Stationary Wavelet Transform"""
# Auto-calculate decomposition levels if not specified
if levels is None:
min_size = min(image.shape)
levels = min(4, max(2, int(np.log2(min_size)) - 3)) # Conservative for SWT
else:
# Validate user-provided levels for stationary wavelet
min_size = min(image.shape)
max_levels = int(np.log2(min_size)) - 2 # Even more conservative for SWT
if levels > max_levels:
print(f"Warning: Requested SWT levels ({levels}) exceeds maximum for image size ({max_levels}). Using {max_levels} instead.")
levels = max_levels
elif levels < 1:
print(f"Warning: SWT levels must be >= 1. Using 1 instead of {levels}.")
levels = 1
# Estimate noise if sigma not provided
if sigma is None:
sigma = estimate_sigma(image, channel_axis=None)
if rescale_sigma:
sigma *= 1.0
# Perform stationary wavelet decomposition
coeffs = pywt.swt2(image, wavelet, level=levels, norm=True)
# Apply thresholding based on method
if method == 'BayesShrink':
coeffs_thresh = _bayes_shrink_threshold_swt(coeffs, sigma)
elif method == 'VisuShrink':
threshold = sigma * np.sqrt(2 * np.log(image.size))
coeffs_thresh = _apply_threshold_swt(coeffs, threshold, mode)
elif method == 'SureShrink':
coeffs_thresh = _sure_shrink_threshold_swt(coeffs, sigma, mode)
elif method == 'manual':
# For manual method, use sigma as noise level but calculate appropriate threshold
manual_threshold = sigma * 0.1 # More conservative threshold
coeffs_thresh = _apply_threshold_swt(coeffs, manual_threshold, mode)
else:
raise ValueError(f"Unknown thresholding method: {method}")
# Reconstruct image
denoised = pywt.iswt2(coeffs_thresh, wavelet, norm=True)
# Ensure output is same size as input
if denoised.shape != image.shape:
h, w = image.shape
denoised = denoised[:h, :w]
return np.clip(denoised, 0, 1)
def _bayes_shrink_threshold_swt(coeffs, sigma):
"""Apply BayesShrink thresholding to SWT coefficients"""
coeffs_thresh = []
for i, (approx, (cH, cV, cD)) in enumerate(coeffs):
if i == 0:
# Keep first level approximation coefficients unchanged
coeffs_thresh.append((approx, (cH, cV, cD)))
else:
# Process detail coefficients
detail_coeffs = []
for detail in [cH, cV, cD]:
# Estimate signal variance
var_y = np.var(detail)
var_x = max(0, var_y - sigma**2)
if var_x > 0:
# Calculate Bayes threshold
threshold = sigma**2 / np.sqrt(var_x)
# Apply soft thresholding
detail_thresh = pywt.threshold(detail, threshold, mode='soft')
else:
# If no signal estimated, use conservative threshold instead of zeroing
conservative_threshold = sigma * 0.5 # Much more conservative
detail_thresh = pywt.threshold(detail, conservative_threshold, mode='soft')
detail_coeffs.append(detail_thresh)
coeffs_thresh.append((approx, tuple(detail_coeffs)))
return coeffs_thresh
def _apply_threshold_swt(coeffs, threshold, mode):
"""Apply uniform threshold to SWT coefficients"""
coeffs_thresh = []
for i, (approx, (cH, cV, cD)) in enumerate(coeffs):
if i == 0:
# Keep first level approximation coefficients unchanged
coeffs_thresh.append((approx, (cH, cV, cD)))
else:
detail_coeffs = []
for detail in [cH, cV, cD]:
detail_thresh = pywt.threshold(detail, threshold, mode=mode)
detail_coeffs.append(detail_thresh)
coeffs_thresh.append((approx, tuple(detail_coeffs)))
return coeffs_thresh
def _sure_shrink_threshold_swt(coeffs, sigma, mode):
"""Apply SureShrink thresholding to SWT coefficients
SureShrink is a hybrid approach that selects between soft and hard thresholding
based on Stein's Unbiased Risk Estimator (SURE).
"""
coeffs_thresh = []
for i, (approx, (cH, cV, cD)) in enumerate(coeffs):
if i == 0:
# Keep first level approximation coefficients unchanged
coeffs_thresh.append((approx, (cH, cV, cD)))
else:
detail_coeffs = []
for detail in [cH, cV, cD]:
# Calculate threshold using SURE
n = detail.size
sorted_coeffs = np.sort(np.abs(detail.flatten()))
# Calculate risk for each threshold
risks = []
thresholds = sorted_coeffs[::max(1, n//50)] # Sample thresholds
for t in thresholds:
# Number of coefficients below threshold
k = np.sum(np.abs(detail) <= t)
# SURE formula
risk = n - 2*k + np.sum(np.minimum(np.abs(detail), t)**2)
risks.append(risk)
# Find threshold with minimum risk
if risks:
min_idx = np.argmin(risks)
threshold = thresholds[min_idx]
else:
threshold = sigma * np.sqrt(2 * np.log(n)) # Fallback to VisuShrink
# Apply thresholding
detail_thresh = pywt.threshold(detail, threshold, mode=mode)
detail_coeffs.append(detail_thresh)
coeffs_thresh.append((approx, tuple(detail_coeffs)))
return coeffs_thresh
def gpu_wavelet_denoise(
image: np.ndarray,
wavelet: str = 'db8',
sigma: Optional[float] = None,
mode: str = 'soft',
method: str = 'BayesShrink',
levels: Optional[int] = None,
multichannel: bool = True,
rescale_sigma: bool = True,
force_cpu: bool = False
) -> np.ndarray:
"""
GPU-accelerated wavelet denoising with automatic fallback to CPU
Args:
image: Input image as numpy array [H, W, C] with values 0-255
wavelet: Wavelet type ('db8', 'db4', 'bior2.2', 'coif2', 'haar')
sigma: Noise standard deviation (auto-estimated if None)
mode: Thresholding mode ('soft', 'hard')
method: Threshold selection method ('BayesShrink', 'VisuShrink', 'SureShrink', 'manual')
levels: Decomposition levels (auto-calculated if None)
multichannel: Process color channels independently
rescale_sigma: Whether to rescale sigma estimate
force_cpu: Force CPU processing even if GPU is available
Returns:
Denoised image as numpy array [H, W, C] with values 0-255
"""
# Check if GPU acceleration is beneficial
use_gpu = (
not force_cpu and
CUPY_AVAILABLE and
can_use_gpu(image.shape) and
np.prod(image.shape) > 512*512 # Only use GPU for larger images
)
if use_gpu:
try:
return _gpu_wavelet_denoise_impl(
image, wavelet, sigma, mode, method, levels, multichannel, rescale_sigma
)
except Exception as e:
print(f"GPU wavelet denoising failed: {e}, falling back to CPU")
cleanup_gpu_memory()
# CPU fallback
return wavelet_denoise(
image, wavelet, sigma, mode, method, levels, multichannel, rescale_sigma
)
def _gpu_wavelet_denoise_impl(
image: np.ndarray,
wavelet: str,
sigma: Optional[float],
mode: str,
method: str,
levels: Optional[int],
multichannel: bool,
rescale_sigma: bool
) -> np.ndarray:
"""GPU implementation of wavelet denoising"""
# Validate inputs
if image.ndim not in [2, 3]:
raise ValueError("Image must be 2D (grayscale) or 3D (color)")
# Convert to float for processing
img_float = image.astype(np.float32) / 255.0
gpu_img = to_gpu(img_float)
# Handle grayscale vs color
if gpu_img.ndim == 2:
result = _gpu_denoise_single_channel(
gpu_img, wavelet, sigma, mode, method, levels, rescale_sigma
)
else:
if multichannel:
# Process each channel independently on GPU
result_channels = []
for c in range(gpu_img.shape[2]):
denoised_channel = _gpu_denoise_single_channel(
gpu_img[:, :, c], wavelet, sigma, mode, method, levels, rescale_sigma
)
result_channels.append(denoised_channel)
result = cp.stack(result_channels, axis=2)
else:
# Convert to grayscale and denoise
gray = cp.mean(gpu_img, axis=2)
denoised_gray = _gpu_denoise_single_channel(
gray, wavelet, sigma, mode, method, levels, rescale_sigma
)
# Broadcast back to color
result = cp.stack([denoised_gray] * gpu_img.shape[2], axis=2)
# Convert back to CPU and uint8 range
result_cpu = to_cpu(result)
result_cpu = np.clip(result_cpu * 255.0, 0, 255).astype(np.uint8)
return result_cpu
def _gpu_denoise_single_channel(
gpu_image,
wavelet: str,
sigma: Optional[float],
mode: str,
method: str,
levels: Optional[int],
rescale_sigma: bool
):
"""GPU-accelerated single channel denoising"""
# Convert to CPU for wavelet transform (PyWavelets doesn't support GPU directly)
cpu_image = to_cpu(gpu_image)
# Auto-calculate decomposition levels if not specified
if levels is None:
min_size = min(cpu_image.shape)
levels = min(5, max(2, int(np.log2(min_size)) - 2))
# Estimate noise if sigma not provided
if sigma is None:
sigma = estimate_sigma(cpu_image, channel_axis=None)
if rescale_sigma:
if wavelet.startswith('db'):
sigma *= 1.1
elif wavelet.startswith('bior'):
sigma *= 0.9
else:
sigma *= 1.0
# Perform wavelet decomposition on CPU
coeffs = pywt.wavedec2(cpu_image, wavelet, level=levels, mode='symmetric')
# Apply thresholding using GPU when possible
if method == 'BayesShrink':
coeffs_thresh = _gpu_bayes_shrink_threshold(coeffs, sigma)
elif method == 'VisuShrink':
threshold = sigma * np.sqrt(2 * np.log(cpu_image.size))
coeffs_thresh = _gpu_apply_threshold(coeffs, threshold, mode)
elif method == 'SureShrink':
coeffs_thresh = _gpu_sure_shrink_threshold(coeffs, sigma, mode)
elif method == 'manual':
coeffs_thresh = _gpu_apply_threshold(coeffs, sigma, mode)
else:
raise ValueError(f"Unknown thresholding method: {method}")
# Reconstruct image on CPU
denoised = pywt.waverec2(coeffs_thresh, wavelet, mode='symmetric')
# Ensure output is same size as input
if denoised.shape != cpu_image.shape:
h, w = cpu_image.shape
denoised = denoised[:h, :w]
# Convert back to GPU for final processing
gpu_result = to_gpu(np.clip(denoised, 0, 1))
return gpu_result
def _gpu_bayes_shrink_threshold(coeffs, sigma):
"""GPU-accelerated BayesShrink thresholding"""
coeffs_thresh = [coeffs[0]] # Keep approximation coefficients unchanged
for i in range(1, len(coeffs)):
detail_coeffs = []
for detail in coeffs[i]:
# Move to GPU for variance calculation
gpu_detail = to_gpu(detail)
# Calculate variance on GPU
var_y = cp.var(gpu_detail)
var_x = cp.maximum(0, var_y - sigma**2)
if float(var_x) > 0:
# Calculate Bayes threshold
threshold = sigma**2 / cp.sqrt(var_x)
# Apply soft thresholding on GPU
gpu_detail_thresh = cp.where(
cp.abs(gpu_detail) > threshold,
cp.sign(gpu_detail) * (cp.abs(gpu_detail) - threshold),
0
)
else:
gpu_detail_thresh = cp.zeros_like(gpu_detail)
# Convert back to CPU for wavelet reconstruction
detail_thresh = to_cpu(gpu_detail_thresh)
detail_coeffs.append(detail_thresh)
coeffs_thresh.append(tuple(detail_coeffs))
return coeffs_thresh
def _gpu_apply_threshold(coeffs, threshold, mode):
"""GPU-accelerated uniform thresholding"""
coeffs_thresh = [coeffs[0]] # Keep approximation coefficients
for i in range(1, len(coeffs)):
detail_coeffs = []
for detail in coeffs[i]:
gpu_detail = to_gpu(detail)
if mode == 'soft':
# Soft thresholding on GPU
gpu_detail_thresh = cp.where(
cp.abs(gpu_detail) > threshold,
cp.sign(gpu_detail) * (cp.abs(gpu_detail) - threshold),
0
)
else: # hard thresholding
gpu_detail_thresh = cp.where(
cp.abs(gpu_detail) > threshold,
gpu_detail,
0
)
detail_thresh = to_cpu(gpu_detail_thresh)
detail_coeffs.append(detail_thresh)
coeffs_thresh.append(tuple(detail_coeffs))
return coeffs_thresh
def _gpu_sure_shrink_threshold(coeffs, sigma, mode):
"""GPU-accelerated SureShrink thresholding"""
coeffs_thresh = [coeffs[0]] # Keep approximation coefficients unchanged
for i in range(1, len(coeffs)):
detail_coeffs = []
for detail in coeffs[i]:
gpu_detail = to_gpu(detail)
n = gpu_detail.size
# Calculate SURE on GPU
sorted_coeffs = cp.sort(cp.abs(gpu_detail.flatten()))
# Sample thresholds for efficiency
sample_size = min(100, n)
indices = cp.linspace(0, n-1, sample_size, dtype=int)
thresholds = sorted_coeffs[indices]
# Calculate risk for each threshold
risks = []
for t in thresholds:
k = cp.sum(cp.abs(gpu_detail) <= t)
risk = n - 2*k + cp.sum(cp.minimum(cp.abs(gpu_detail), t)**2)
risks.append(float(risk))
# Find threshold with minimum risk
if risks:
min_idx = np.argmin(risks)
threshold = float(thresholds[min_idx])
else:
threshold = sigma * np.sqrt(2 * np.log(n)) # Fallback
# Apply thresholding on GPU
if mode == 'soft':
gpu_detail_thresh = cp.where(
cp.abs(gpu_detail) > threshold,
cp.sign(gpu_detail) * (cp.abs(gpu_detail) - threshold),
0
)
else: # hard thresholding
gpu_detail_thresh = cp.where(
cp.abs(gpu_detail) > threshold,
gpu_detail,
0
)
detail_thresh = to_cpu(gpu_detail_thresh)
detail_coeffs.append(detail_thresh)
coeffs_thresh.append(tuple(detail_coeffs))
return coeffs_thresh
def gpu_wavelet_denoise_stationary(
image: np.ndarray,
wavelet: str = 'db8',
sigma: Optional[float] = None,
mode: str = 'soft',
method: str = 'BayesShrink',
levels: Optional[int] = None,
multichannel: bool = True,
rescale_sigma: bool = True,
force_cpu: bool = False
) -> np.ndarray:
"""
GPU-accelerated stationary wavelet denoising
Uses the Stationary Wavelet Transform (SWT) which is shift-invariant,
producing better results for images with important structures at different positions.
Args:
Same as gpu_wavelet_denoise() but uses SWT instead of DWT
Returns:
Denoised image with reduced shift-sensitivity artifacts
"""
# Check if GPU acceleration is beneficial
use_gpu = (
not force_cpu and
CUPY_AVAILABLE and
can_use_gpu(image.shape) and
np.prod(image.shape) > 512*512
)
if use_gpu:
try:
return _gpu_wavelet_denoise_stationary_impl(
image, wavelet, sigma, mode, method, levels, multichannel, rescale_sigma
)
except Exception as e:
print(f"GPU stationary wavelet denoising failed: {e}, falling back to CPU")
cleanup_gpu_memory()
# CPU fallback
return wavelet_denoise_stationary(
image, wavelet, sigma, mode, method, levels, multichannel, rescale_sigma
)
def _gpu_wavelet_denoise_stationary_impl(
image: np.ndarray,
wavelet: str,
sigma: Optional[float],
mode: str,
method: str,
levels: Optional[int],
multichannel: bool,
rescale_sigma: bool
) -> np.ndarray:
"""GPU implementation of stationary wavelet denoising"""
# Validate inputs
if image.ndim not in [2, 3]:
raise ValueError("Image must be 2D (grayscale) or 3D (color)")
# Convert to float for processing
img_float = image.astype(np.float32) / 255.0
gpu_img = to_gpu(img_float)
# Handle grayscale vs color
if gpu_img.ndim == 2:
result = _gpu_denoise_single_channel_stationary(
gpu_img, wavelet, sigma, mode, method, levels, rescale_sigma
)
else:
if multichannel:
# Process each channel independently on GPU
result_channels = []
for c in range(gpu_img.shape[2]):
denoised_channel = _gpu_denoise_single_channel_stationary(
gpu_img[:, :, c], wavelet, sigma, mode, method, levels, rescale_sigma
)
result_channels.append(denoised_channel)
result = cp.stack(result_channels, axis=2)
else:
# Convert to grayscale and denoise
gray = cp.mean(gpu_img, axis=2)
denoised_gray = _gpu_denoise_single_channel_stationary(
gray, wavelet, sigma, mode, method, levels, rescale_sigma
)
# Broadcast back to color
result = cp.stack([denoised_gray] * gpu_img.shape[2], axis=2)
# Convert back to CPU and uint8 range
result_cpu = to_cpu(result)
result_cpu = np.clip(result_cpu * 255.0, 0, 255).astype(np.uint8)
return result_cpu
def _gpu_denoise_single_channel_stationary(
gpu_image,
wavelet: str,
sigma: Optional[float],
mode: str,
method: str,
levels: Optional[int],
rescale_sigma: bool
):
"""GPU-accelerated single channel stationary denoising"""
# Convert to CPU for stationary wavelet transform
cpu_image = to_cpu(gpu_image)
# Auto-calculate decomposition levels if not specified
if levels is None:
min_size = min(cpu_image.shape)
levels = min(4, max(2, int(np.log2(min_size)) - 3)) # More conservative for SWT
# Estimate noise if sigma not provided
if sigma is None:
sigma = estimate_sigma(cpu_image, channel_axis=None)
if rescale_sigma:
if wavelet.startswith('db'):
sigma *= 1.1
elif wavelet.startswith('bior'):
sigma *= 0.9
else:
sigma *= 1.0
# Perform stationary wavelet decomposition on CPU
coeffs = pywt.swt2(cpu_image, wavelet, level=levels, trim_approx=True)
# Apply thresholding using GPU when possible
if method == 'BayesShrink':
coeffs_thresh = _gpu_bayes_shrink_threshold_swt(coeffs, sigma)
elif method == 'VisuShrink':
threshold = sigma * np.sqrt(2 * np.log(cpu_image.size))
coeffs_thresh = _gpu_apply_threshold_swt(coeffs, threshold, mode)
elif method == 'SureShrink':
coeffs_thresh = _gpu_sure_shrink_threshold_swt(coeffs, sigma, mode)
elif method == 'manual':
coeffs_thresh = _gpu_apply_threshold_swt(coeffs, sigma, mode)
else:
raise ValueError(f"Unknown thresholding method: {method}")
# Reconstruct image on CPU
denoised = pywt.iswt2(coeffs_thresh, wavelet)
# Ensure output is same size as input
if denoised.shape != cpu_image.shape:
h, w = cpu_image.shape
denoised = denoised[:h, :w]
# Convert back to GPU for final processing
gpu_result = to_gpu(np.clip(denoised, 0, 1))
return gpu_result
def _gpu_bayes_shrink_threshold_swt(coeffs, sigma):
"""GPU-accelerated BayesShrink thresholding for SWT coefficients"""
coeffs_thresh = []
for i, (cA, (cH, cV, cD)) in enumerate(coeffs):
# For SWT, we threshold all coefficients including approximation at deeper levels
if i == len(coeffs) - 1:
# Keep the final approximation coefficients
cA_thresh = cA
else:
# Threshold approximation coefficients at deeper levels
gpu_cA = to_gpu(cA)
var_y = cp.var(gpu_cA)
var_x = cp.maximum(0, var_y - sigma**2)
if float(var_x) > 0:
threshold = sigma**2 / cp.sqrt(var_x)
cA_thresh = to_cpu(cp.where(
cp.abs(gpu_cA) > threshold,
cp.sign(gpu_cA) * (cp.abs(gpu_cA) - threshold),
0
))
else:
cA_thresh = np.zeros_like(cA)
# Threshold detail coefficients
detail_coeffs = []
for detail in [cH, cV, cD]:
gpu_detail = to_gpu(detail)
var_y = cp.var(gpu_detail)
var_x = cp.maximum(0, var_y - sigma**2)
if float(var_x) > 0:
threshold = sigma**2 / cp.sqrt(var_x)
detail_thresh = to_cpu(cp.where(
cp.abs(gpu_detail) > threshold,
cp.sign(gpu_detail) * (cp.abs(gpu_detail) - threshold),
0
))
else:
detail_thresh = np.zeros_like(detail)
detail_coeffs.append(detail_thresh)
coeffs_thresh.append((cA_thresh, tuple(detail_coeffs)))
return coeffs_thresh
def _gpu_apply_threshold_swt(coeffs, threshold, mode):
"""GPU-accelerated uniform thresholding for SWT coefficients"""
coeffs_thresh = []
for i, (cA, (cH, cV, cD)) in enumerate(coeffs):
# For SWT, we can threshold all coefficients
if i == len(coeffs) - 1:
# Keep the final approximation coefficients
cA_thresh = cA
else:
# Threshold approximation coefficients at deeper levels
gpu_cA = to_gpu(cA)
if mode == 'soft':
cA_thresh = to_cpu(cp.where(
cp.abs(gpu_cA) > threshold,
cp.sign(gpu_cA) * (cp.abs(gpu_cA) - threshold),
0
))
else: # hard thresholding
cA_thresh = to_cpu(cp.where(
cp.abs(gpu_cA) > threshold,
gpu_cA,
0
))
# Threshold detail coefficients
detail_coeffs = []
for detail in [cH, cV, cD]:
gpu_detail = to_gpu(detail)
if mode == 'soft':
detail_thresh = to_cpu(cp.where(
cp.abs(gpu_detail) > threshold,
cp.sign(gpu_detail) * (cp.abs(gpu_detail) - threshold),
0
))
else: # hard thresholding
detail_thresh = to_cpu(cp.where(
cp.abs(gpu_detail) > threshold,
gpu_detail,
0
))
detail_coeffs.append(detail_thresh)
coeffs_thresh.append((cA_thresh, tuple(detail_coeffs)))
return coeffs_thresh
def _gpu_sure_shrink_threshold_swt(coeffs, sigma, mode):
"""GPU-accelerated SureShrink thresholding for SWT coefficients"""
coeffs_thresh = []
for i, (cA, (cH, cV, cD)) in enumerate(coeffs):
# For SWT, we can threshold all coefficients
if i == len(coeffs) - 1:
# Keep the final approximation coefficients
cA_thresh = cA
else:
# Apply SURE to approximation coefficients at deeper levels
cA_thresh = _gpu_sure_threshold_single(cA, sigma, mode)
# Apply SURE to detail coefficients
detail_coeffs = []
for detail in [cH, cV, cD]:
detail_thresh = _gpu_sure_threshold_single(detail, sigma, mode)
detail_coeffs.append(detail_thresh)
coeffs_thresh.append((cA_thresh, tuple(detail_coeffs)))
return coeffs_thresh
def _gpu_sure_threshold_single(detail, sigma, mode):
"""Apply SURE thresholding to a single coefficient array"""
gpu_detail = to_gpu(detail)
n = gpu_detail.size
# Calculate SURE on GPU
sorted_coeffs = cp.sort(cp.abs(gpu_detail.flatten()))
# Sample thresholds for efficiency
sample_size = min(100, n)
indices = cp.linspace(0, n-1, sample_size, dtype=int)
thresholds = sorted_coeffs[indices]
# Calculate risk for each threshold
risks = []
for t in thresholds:
k = cp.sum(cp.abs(gpu_detail) <= t)
risk = n - 2*k + cp.sum(cp.minimum(cp.abs(gpu_detail), t)**2)
risks.append(float(risk))
# Find threshold with minimum risk
if risks:
min_idx = np.argmin(risks)
threshold = float(thresholds[min_idx])
else:
threshold = sigma * np.sqrt(2 * np.log(n)) # Fallback
# Apply thresholding on GPU
if mode == 'soft':
gpu_detail_thresh = cp.where(
cp.abs(gpu_detail) > threshold,
cp.sign(gpu_detail) * (cp.abs(gpu_detail) - threshold),
0
)
else: # hard thresholding
gpu_detail_thresh = cp.where(
cp.abs(gpu_detail) > threshold,
gpu_detail,
0
)
return to_cpu(gpu_detail_thresh)
def get_available_wavelets():
"""Get list of available wavelet families suitable for denoising"""
return {
'db8': 'Daubechies 8 (recommended for natural images)',
'db4': 'Daubechies 4 (faster, good balance)',
'bior2.2': 'Biorthogonal 2.2 (good for edges)',
'coif2': 'Coiflets 2 (symmetric, good reconstruction)',
'haar': 'Haar (fastest, simple images)',
'db1': 'Daubechies 1 (same as Haar)',
'db6': 'Daubechies 6 (smoother than db4)',
'bior4.4': 'Biorthogonal 4.4 (higher order)',
'sym8': 'Symlets 8 (nearly symmetric)'
}
def estimate_noise_level(image: np.ndarray) -> float:
"""Estimate noise level in image using robust methods
Args:
image: Input image with values 0-255 (uint8) or 0-1 (float)
Returns:
Noise level in same scale as input (0-255 if input is uint8, 0-1 if input is float)
"""
# Store original data type
original_dtype = image.dtype
is_uint8 = (original_dtype == np.uint8)
# Convert to grayscale if needed
if image.ndim == 3:
if is_uint8:
gray = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY)
else:
# Manual conversion for float images
gray = 0.299 * image[:, :, 0] + 0.587 * image[:, :, 1] + 0.114 * image[:, :, 2]
# FIXED: Ensure contiguous array after arithmetic operation
gray = np.ascontiguousarray(gray)
else:
gray = image.copy()
# Method 1: Try skimage estimate_sigma (robust but sometimes fails)
try:
if is_uint8:
gray_float = gray.astype(np.float32) / 255.0
else:
gray_float = gray.astype(np.float32)
gray_float = np.clip(gray_float, 0, 1)
sigma_normalized = estimate_sigma(gray_float, channel_axis=None)
# Convert back to original scale
if is_uint8:
return sigma_normalized * 255.0
else:
return sigma_normalized
except Exception:
pass # Fall through to backup method
# Method 2: Laplacian-based noise estimation (more reliable fallback)
if is_uint8:
gray_for_laplacian = gray.astype(np.float32)
else:
gray_for_laplacian = gray * 255.0 # Scale to 0-255 for Laplacian
# FIXED: Ensure contiguous array after arithmetic operation
gray_for_laplacian = np.ascontiguousarray(gray_for_laplacian)
# Calculate Laplacian
laplacian = cv2.Laplacian(gray_for_laplacian, cv2.CV_64F)
# Estimate noise using median absolute deviation of Laplacian
sigma_laplacian = np.median(np.abs(laplacian)) / 0.6745
# Convert back to appropriate scale
if is_uint8:
return sigma_laplacian
else:
return sigma_laplacian / 255.0