#!/usr/bin/env python3 """ DNG to JPG Batch Processor Processes .dng files from an input folder and saves them as .jpg files in organized subfolders with different scaling factors. """ import os import sys import shutil import argparse from pathlib import Path from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Tuple, Optional import threading try: import rawpy import numpy as np import cv2 import piexif except ImportError as e: print(f"Required package not installed: {e}") print("Install with: pip install rawpy pillow piexif") sys.exit(1) def aces_tonemap(x: np.ndarray): a = 2.51 b = 0.03 c = 2.43 d = 0.59 e = 0.14 y = (x * (a * x + b)) / (x * (c * x + d) + e) return np.clip(y, 0.0, 1.0) class DNGProcessor: def __init__(self, input_folder: str, output_folder: str, color_space: str, save_format: str, half_size: bool, max_workers: Optional[int] = None): self.input_folder = Path(input_folder) self.output_folder = Path(output_folder) self.color_space = color_space.lower() self.save_format = save_format.lower() self.half_size = half_size self.max_workers = max_workers or os.cpu_count() self.lock = threading.Lock() self.processed_count = 0 self.total_files = 0 # Validate color space if self.color_space not in ['srgb', 'aces', 'linear-rgb', 'linear-aces']: raise ValueError("Color space must be 'sRGB', 'ACES', 'Linear-RGB', or 'Linear-ACES'") def setup_output_folders(self) -> None: """Setup output folder structure, ensuring it's empty or doesn't exist.""" if self.output_folder.exists(): if any(self.output_folder.iterdir()): response = input(f"Output folder is not empty. " "Delete contents? (y/N): ") if response.lower() != 'y': print("Operation cancelled.") sys.exit(1) shutil.rmtree(self.output_folder) # Create main output folder and subfolders self.output_folder.mkdir(parents=True, exist_ok=True) for subfolder in ['images', 'images_2', 'images_4']: (self.output_folder / subfolder).mkdir(exist_ok=True) def get_dng_files(self) -> list: """Get list of .dng files from input folder.""" dng_files = list(self.input_folder.glob('*.dng')) + \ list(self.input_folder.glob('*.DNG')) return sorted(dng_files) def process_raw_image(self, raw_path: Path) -> Tuple[np.ndarray, dict]: """Process a single RAW image with specified parameters.""" with rawpy.imread(str(raw_path)) as raw: if self.color_space == 'srgb': output_color = rawpy.ColorSpace.sRGB elif self.color_space == 'linear-rgb': output_color = rawpy.ColorSpace.sRGB elif self.color_space == 'aces': output_color = rawpy.ColorSpace.ACES # ACES2065-1 elif self.color_space == 'linear-aces': output_color = rawpy.ColorSpace.ACES else: raise ValueError(f"Unknown color space: {self.color_space}") kwargs = {} if self.color_space.startswith('linear-'): kwargs['gamma'] = (1.0, 1.0) else: # kwargs['gamma'] = (2.222, 4.5) kwargs['gamma'] = (2.4, 12.92) if self.save_format == 'png16': kwargs['output_bps'] = 16 else: kwargs['output_bps'] = 8 # matching https://github.com/GNOME/shotwell/blob/b7c5957b4400664f255a6a8c8a63daf2d72958c6/src/photos/GRaw.vala rgb_array = raw.postprocess( output_color=output_color, half_size=self.half_size, no_auto_bright=True, bright=1.0, auto_bright_thr=0.01, use_camera_wb=True, use_auto_wb=True, # use_camera_matrix=1, highlight_mode=rawpy.HighlightMode.Clip, demosaic_algorithm=rawpy.DemosaicAlgorithm.PPG, # fbdd_noise_reduction=rawpy.FBDDNoiseReductionMode.Full, # noise_thr=100, **kwargs ) # brightness adjustment if False: target_l = 0.25 if False: # linear brightness, can lead to over-exposure lut = target_l * np.arange(2**16) / np.mean(rgb_array) lut = (255*aces_tonemap(lut)).astype(np.uint8) else: # gamma, works well for dark area, may reduce contrast k = np.log(target_l) / (np.mean(np.log(np.clip(rgb_array, min=1))) - np.log(2**16)) lut = (255*np.linspace(0, 1, 2**16)**k).astype(np.uint8) rgb_array = lut[rgb_array] metadata = { 'camera_wb': raw.camera_whitebalance, 'daylight_wb': raw.daylight_whitebalance, 'color_matrix': raw.color_matrix, 'raw_image_sizes': raw.raw_image.shape } return rgb_array, metadata def copy_exif_data(self, source_path: Path, target_path: Path) -> None: """Copy EXIF data from source DNG to target JPG.""" try: # Read EXIF from source with rawpy.imread(str(source_path)) as raw: # Extract basic metadata that can be safely transferred exif_dict = { "0th": {}, "Exif": {}, "GPS": {}, "1st": {}, "thumbnail": None } # Add basic camera information if available try: # These are safe to copy and commonly supported if hasattr(raw, 'color_desc'): exif_dict["0th"][piexif.ImageIFD.Software] = "DNG Processor" # Add processing information exif_dict["0th"][piexif.ImageIFD.ProcessingSoftware] = "rawpy + PIL" except AttributeError: pass # Generate EXIF bytes exif_bytes = piexif.dump(exif_dict) piexif.insert(exif_bytes, str(target_path)) except Exception as e: print(f"Warning: Could not copy EXIF data for {source_path.name}: {e}") # Continue without EXIF data def resize_image(self, image: np.ndarray, scale_factor: int) -> np.ndarray: """Resize image by scale factor using high-quality resampling.""" if scale_factor == 1: return image new_size = (image.shape[1] // scale_factor, image.shape[0] // scale_factor) resized = cv2.resize(image, new_size, interpolation=cv2.INTER_AREA) return resized def save_image_variants(self, image_array: np.ndarray, output_name: str) -> None: """Save image in different scales to appropriate subfolders.""" scale_factors = [1, 2, 4] subfolder_names = ['images', 'images_2', 'images_4'] for scale_factor, subfolder in zip(scale_factors, subfolder_names): scaled_array = self.resize_image(image_array, scale_factor) extension = 'jpg' if self.save_format == 'jpg' else 'png' output_path = self.output_folder / subfolder / f"{output_name}.{extension}" scaled_array = cv2.cvtColor(scaled_array, cv2.COLOR_RGB2BGR) if self.save_format == 'jpg': cv2.imwrite(str(output_path), scaled_array, [cv2.IMWRITE_JPEG_QUALITY, 95]) else: cv2.imwrite(str(output_path), scaled_array) def process_single_file(self, dng_path: Path) -> bool: """Process a single DNG file.""" try: rgb_array, metadata = self.process_raw_image(dng_path) output_name = dng_path.stem self.save_image_variants(rgb_array, output_name) if self.save_format == 'jpg': for subfolder in ['images', 'images_2', 'images_4']: full_size_path = self.output_folder / subfolder / f"{output_name}.jpg" self.copy_exif_data(dng_path, full_size_path) with self.lock: self.processed_count += 1 print(f"Processed {self.processed_count}/{self.total_files}: {dng_path.name}") return True except Exception as e: print(f"Error processing {dng_path.name}: {e}") return False def process_all_files(self) -> None: """Process all DNG files using multithreading.""" dng_files = self.get_dng_files() if not dng_files: print("No .dng files found in input folder.") return self.total_files = len(dng_files) print(f"Found {self.total_files} DNG files to process.") print(f"Using {self.max_workers} threads.") print(f"Color space: {self.color_space.upper()}") print(f"Output folder: {self.output_folder}") self.setup_output_folders() successful = 0 failed = 0 # for dng_file in dng_files: # self.process_single_file(dng_file) # exit(0) with ThreadPoolExecutor(max_workers=self.max_workers) as executor: future_to_file = { executor.submit(self.process_single_file, dng_file): dng_file for dng_file in dng_files } for future in as_completed(future_to_file): if future.result(): successful += 1 else: failed += 1 print(f"\nProcessing complete!") print(f"Successfully processed: {successful}") print(f"Failed: {failed}") def main(): parser = argparse.ArgumentParser(description="Convert DNG files to JPG with multiple scales") parser.add_argument("input_folder", help="Input folder containing .dng files") parser.add_argument("output_folder", help="Output folder for processed images") parser.add_argument("--color-space", choices=['sRGB', 'ACES', 'linear-RGB', 'linear-ACES'], default='sRGB', help="Color space for processing (default: sRGB)") parser.add_argument("--save-format", choices=['jpg', 'png', 'png16'], default='jpg', help="Output image format (default: jpg)") parser.add_argument("--no-half-size", action="store_true", help="Whether to interpolate instead of downscale by 2.") parser.add_argument("--threads", type=int, default=None, help="Number of processing threads (default: CPU count)") args = parser.parse_args() # Validate input folder if not Path(args.input_folder).exists(): print(f"Error: Input folder '{args.input_folder}' does not exist.") sys.exit(1) # Create processor and run processor = DNGProcessor( input_folder=args.input_folder, output_folder=args.output_folder, color_space=args.color_space, save_format=args.save_format, half_size=not args.no_half_size, max_workers=args.threads ) try: processor.process_all_files() except KeyboardInterrupt: print("\nProcessing interrupted by user.") except Exception as e: print(f"Error: {e}") sys.exit(1) if __name__ == "__main__": main()