
如何进行网站调试,青岛网站,wordpress 中文,上海南建设培训执业中心网站python 2 大疆照片rtk 改变分辨率重新拷贝GPS信息
给参数default_input_dir = rD:\0ubuntu20\1slam\数据\2RTK\City4_outskirts2\map_cloudy_0303_15pm_135m\images_sourcedefault_output_dir = rD…
python 2 大疆照片rtk 改变分辨率重新拷贝GPS信息给参数default_input_dir = r"D:\0ubuntu20\1slam\数据\2RTK\City4_outskirts2\map_cloudy_0303_15pm_135m\images_source"default_output_dir = r"D:\0ubuntu20\1slam\数据\2RTK\City4_outskirts2\map_cloudy_0303_15pm_135m\images"default_resolution = (1805, 1203)default_start_id = 0"""
楂樿川閲忓浘鐗囩缉鏀捐剼鏈紙鏀寔鑷畾涔夌紪鍙烽噸鍛藉悕锛?鍔熻兘锛?- OpenCV 楂樿川閲忕缉鏀?- 瀹屾暣淇濈暀 EXIF
- 澶氳繘绋嬪姞閫?- 杈撳嚭鏂囦欢鍚嶏細PREFIX_XXXXX.jpg
"""# pip install "numpy2" opencv-python-headless pillow piexif tqdmimport cv2
import piexif
import numpy as np
from PIL import Image
import argparse
from pathlib import Path
import multiprocessing as mp
from tqdm import tqdm
import warnings
from typing import Optional
import jsonclass HighQualityImageResizer:"""High quality image resizer with sequential output names."""def __init__(self,input_dir: str,output_dir: str,max_size: int = 1920,target_width: Optional[int] = None,target_height: Optional[int] = None,quality: int = 100,interpolation: str = 'lanczos',preserve_metadata: bool = True,num_workers: int = 4,format: str = 'jpg',progressive: bool = True,chroma_subsampling: int = 0,optimize: bool = True,start_id: int = 110,prefix: str = "DJI_",id_width: int = 5):self.input_dir = Path(input_dir)self.output_dir = Path(output_dir)self.max_size = max_sizeself.target_width = target_widthself.target_height = target_heightself.quality = min(max(quality, 1), 100)self.preserve_metadata = preserve_metadataself.num_workers = num_workersself.format = format.lower()self.progressive = progressiveself.chroma_subsampling = chroma_subsamplingself.optimize = optimize# 鉁?缂栧彿绯荤粺self.start_id = start_idself.prefix = prefixself.id_width = id_width# 鉁?澶氳繘绋嬪畨鍏ㄨ鏁板櫒self.id_lock = mp.Lock()self.current_id = mp.Value('i', start_id)self.interpolation_map = {'nearest': cv2.INTER_NEAREST,'linear': cv2.INTER_LINEAR,'cubic': cv2.INTER_CUBIC,'area': cv2.INTER_AREA,'lanczos': cv2.INTER_LANCZOS4}self.interpolation = self.interpolation_map.get(interpolation, cv2.INTER_LANCZOS4)self.output_dir.mkdir(parents=True, exist_ok=True)self.log_file = self.output_dir / 'resize_log.json'self.stats = {'processed': 0,'skipped': 0,'failed': 0,'files': []}# -------------------------# 缂栧彿鐢熸垚锛堝杩涚▼瀹夊叏锛? # -------------------------def _generate_output_name(self) - str:with self.id_lock:current = self.current_id.valueself.current_id.value += 1return f"{self.prefix}{current:0{self.id_width}d}.{self.format}"# -------------------------# EXIF 璇诲彇# -------------------------def _get_exif_metadata(self, image_path: Path):metadata = {}exif_bytes = Nonetry:with Image.open(image_path) as img:if 'exif' in img.info:exif_dict = piexif.load(img.info['exif'])exif_bytes = piexif.dump(exif_dict)if '0th' in exif_dict:metadata['camera_make'] = exif_dict['0th'].get(piexif.ImageIFD.Make, b'').decode(errors='ignore')metadata['camera_model'] = exif_dict['0th'].get(piexif.ImageIFD.Model, b'').decode(errors='ignore')metadata['date_time'] = exif_dict['0th'].get(piexif.ImageIFD.DateTime, b'').decode(errors='ignore')metadata['format'] = img.formatmetadata['mode'] = img.modemetadata['size'] = img.sizeexcept Exception as e:warnings.warn(f"璇诲彇 {image_path} 鍏冩暟鎹け璐? {e}")return exif_bytes, metadatadef _get_xmp_app1_segments(self, image_path: Path):"""Return JPEG APP1 XMP segments, including DJI RTK metadata."""try:data = image_path.read_bytes()except Exception as e:warnings.warn(f"璇诲彇 XMP 澶辫触 {image_path}: {e}")return []if not data.startswith(b"\xff\xd8"):return []segments = []pos = 2while pos + 4 = len(data):if data[pos] != 0xFF:breakwhile pos len(data) and data[pos] == 0xFF:pos += 1if pos = len(data):breakmarker = data[pos]pos += 1if marker in (0xD9, 0xDA):breakif marker == 0x01 or 0xD0 = marker = 0xD7:continueif pos + 2 len(data):breaklength = int.from_bytes(data[pos:pos + 2], "big")if length 2 or pos + length len(data):breakpayload = data[pos + 2:pos + length]segment = data[pos - 2:pos + length]is_xmp = (marker == 0xE1and (payload.startswith(b"http://ns.adobe.com/xap/1.0/\x00")or b"drone-dji:" in payloador b"RtkFlag" in payloador b"RtkStdLat" in payload))if is_xmp:segments.append(segment)pos += lengthreturn segmentsdef _insert_jpeg_app_segments(self, output_path: Path, segments):if not segments:returntry:data = output_path.read_bytes()if not data.startswith(b"\xff\xd8"):returnif b"drone-dji:" in data or b"RtkFlag" in data:returnoutput_path.write_bytes(data[:2] + b"".join(segments) + data[2:])except Exception as e:warnings.warn(f"鍐欏叆 XMP 澶辫触 {output_path}: {e}")# -------------------------# 灏哄璁$畻# -------------------------def _calculate_target_size(self, width: int, height: int):if self.target_width is not None and self.target_height is not None:new_w = self.target_widthnew_h = self.target_heightreturn new_w, new_hif self.target_width is not None:scale = self.target_width / widthnew_w = self.target_widthnew_h = int(height * scale)# if new_w % 2 != 0:# new_w += 1# if new_h % 2 != 0:# new_h += 1return new_w, new_hif width = self.max_size and height = self.max_size:return width, heightif width height:scale = self.max_size / widthelse:scale = self.max_size / heightnew_w = int(width * scale)new_h = int(height * scale)# if new_w % 2 != 0:# new_w += 1# if new_h % 2 != 0:# new_h += 1return new_w, new_h# -------------------------# 閿愬寲# -------------------------def _sharpen_image(self, image: np.ndarray, strength: float = 0.2):if strength = 0:return imagekernel = np.array([[-strength, -strength, -strength],[-strength, 1 + 8 * strength, -strength],[-strength, -strength, -strength]])sharpened = cv2.filter2D(image, -1, kernel)return np.clip(sharpened, 0, 255).astype(np.uint8)# -------------------------# 淇濆瓨鍥惧儚# -------------------------def _encode_image(self, image: np.ndarray, output_path: Path, exif_bytes, xmp_segments=None):try:if len(image.shape) == 3 and image.shape[2] == 3:image_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)elif len(image.shape) == 3 and image.shape[2] == 4:image_rgb = cv2.cvtColor(image, cv2.COLOR_BGRA2RGBA)else:image_rgb = imagepil_image = Image.fromarray(image_rgb)if self.format in ('jpg', 'jpeg'):attempts = [{'quality': self.quality,'optimize': self.optimize,'progressive': self.progressive,'subsampling': self.chroma_subsampling}]if exif_bytes and self.preserve_metadata:attempts[0]['exif'] = exif_bytes# Fallback for images that fail with optimize/progressive metadata saving.attempts.append({'quality': self.quality,'optimize': False,'progressive': False,'subsampling': self.chroma_subsampling,**({'exif': exif_bytes} if exif_bytes and self.preserve_metadata else {})})last_error = Nonefor save_kwargs in attempts:try:pil_image.save(output_path, **save_kwargs)if self.preserve_metadata:self._insert_jpeg_app_segments(output_path, xmp_segments or [])return True, Noneexcept Exception as e:last_error = str(e)return False, last_errorif self.format == 'png':pil_image.save(output_path, compress_level=0)return True, Noneif self.format == 'webp':pil_image.save(output_path,quality=self.quality,method=6,lossless=(self.quality == 100))return True, Nonepil_image.save(output_path)return True, Noneexcept Exception as e:return False, str(e)# -------------------------# 鍗曞紶澶勭悊# -------------------------def _process_single_image(self, input_path: Path, output_path: Path):result = {'input': str(input_path),'output': str(output_path),'success': False}try:exif_bytes, metadata = self._get_exif_metadata(input_path)xmp_segments = self._get_xmp_app1_segments(input_path) if self.preserve_metadata else []result['xmp_segments'] = len(xmp_segments)# cv2.imread often fails on Windows paths containing Chinese characters.image_data = np.fromfile(str(input_path), dtype=np.uint8)image = cv2.imdecode(image_data, cv2.IMREAD_UNCHANGED)if image is None:result['error'] = '鏃犳硶璇诲彇鍥惧儚'return resulth, w = image.shape[:2]new_w, new_h = self._calculate_target_size(w, h)if (w, h) != (new_w, new_h):image = cv2.resize(image, (new_w, new_h), interpolation=self.interpolation)image = self._sharpen_image(image)success, save_error = self._encode_image(image, output_path, exif_bytes, xmp_segments)result['success'] = successif save_error:result['error'] = save_errorwarnings.warn(f"淇濆瓨澶辫触 {output_path}: {save_error}")except Exception as e:result['error'] = str(e)return result# -------------------------# Worker# -------------------------def _process_worker(self, file_queue: mp.Queue, result_queue: mp.Queue):while True:task = file_queue.get()if task is None:breakinput_path, output_path = taskresult = self._process_single_image(input_path, output_path)result_queue.put(result)# -------------------------# 鏂囦欢鎵弿# -------------------------def _get_image_files(self):exts = {'.jpg', '.jpeg', '.png', '.webp', '.tiff', '.tif', '.bmp'}files = {}for ext in exts:for path in self.input_dir.glob(f'*{ext}'):files[str(path).lower()] = pathfor path in self.input_dir.glob(f'*{ext.upper()}'):files[str(path).lower()] = pathreturn sorted(files.values(), key=lambda path: path.name.lower())# -------------------------# 涓绘祦绋? # -------------------------def process(self):image_files = self._get_image_files()if not image_files:print("No image files found")returnprint(f"Found {len(image_files)} images")file_queue = mp.Queue()result_queue = mp.Queue()for img_file in image_files:output_name = self._generate_output_name()output_path = self.output_dir / output_namefile_queue.put((img_file, output_path))for _ in range(self.num_workers):file_queue.put(None)processes = []for _ in range(self.num_workers):p = mp.Process(target=self._process_worker, args=(file_queue, result_queue))p.start()processes.append(p)with tqdm(total=len(image_files), desc="澶勭悊杩涘害") as pbar:for _ in range(len(image_files)):result = result_queue.get()self.stats['files'].append(result)if result['success']:self.stats['processed'] += 1else:self.stats['failed'] += 1pbar.update(1)for p in processes:p.join()with open(self.log_file, 'w', encoding='utf-8') as f:json.dump(self.stats, f, indent=2, ensure_ascii=False)print("\n鉁?瀹屾垚")print(f"鎴愬姛: {self.stats['processed']}")print(f"澶辫触: {self.stats['failed']}")# ============================
# CLI
# Default run:
# python tool1_Resize_img_SetID.py
# Optional override:
# python tool1_Resize_img_SetID.py INPUT_DIR OUTPUT_DIR --resolution 1805 1203 --start-id 0
# ============================
def main():default_input_dir = r"D:\0ubuntu20\1slam\数据\2RTK\City4_outskirts2\map_cloudy_0303_15pm_135m\images_source"default_output_dir = r"D:\0ubuntu20\1slam\数据\2RTK\City4_outskirts2\map_cloudy_0303_15pm_135m\images"default_resolution = (1805, 1203)default_start_id = 0parser = argparse.ArgumentParser(description="High quality image resize and sequential rename")parser.add_argument("input_dir", nargs="?", default=default_input_dir)parser.add_argument("output_dir", nargs="?", default=default_output_dir)parser.add_argument("-s", "--size", type=int, default=1805,help="Limit by longest side, default 1805")parser.add_argument("--width", type=int, default=None,help="Set output width only; height follows original aspect ratio")parser.add_argument("--resolution", nargs=2, type=int, metavar=("WIDTH", "HEIGHT"), default=default_resolution,help="Set exact output resolution, for example --resolution 1920 1080")parser.add_argument("-q", "--quality", type=int, default=100)parser.add_argument("-w", "--workers", type=int, default=8)parser.add_argument("--start-id", type=int, default=default_start_id)parser.add_argument("--prefix", type=str, default="DJI_")parser.add_argument("--id-width", type=int, default=5)args = parser.parse_args()if args.width is not None and args.width = 0:parser.error("--width 蹇呴』澶т簬 0")if args.resolution is not None:if args.width is not None:parser.error("--resolution 鍜?--width 涓嶈兘鍚屾椂浣跨敤")if args.resolution[0] = 0 or args.resolution[1] = 0:parser.error("--resolution 鐨勫鍜岄珮閮藉繀椤诲ぇ浜?0")target_width = Nonetarget_height = Noneif args.resolution is not None:target_width, target_height = args.resolutionelif args.width is not None:target_width = args.widthresizer = HighQualityImageResizer(input_dir=args.input_dir,output_dir=args.output_dir,max_size=args.size,target_width=target_width,target_height=target_height,quality=args.quality,num_workers=args.workers,start_id=args.start_id,prefix=args.prefix,id_width=args.id_width)resizer.process()if __name__ == "__main__":main()