""" person_mosaic.py — 실사 인물 전신 모자이크 파이프라인 YOLO11로 사람을 탐지하고 전신에 모자이크(픽셀화) 또는 블러를 적용. 이미지(jpg/png)와 영상(mp4) 모두 지원. 의존성 설치: pip install ultralytics opencv-python tqdm 사용법: # 이미지 모자이크 python person_mosaic.py input.jpg -o output.jpg # 영상 모자이크 (기본 모자이크) python person_mosaic.py input.mp4 -o output.mp4 # 블러 모드 python person_mosaic.py input.mp4 -o output.mp4 -m blur # 모자이크 강도 조절 (낮을수록 강함) python person_mosaic.py input.mp4 -o output.mp4 -s 5 # GPU 장치 지정 python person_mosaic.py input.mp4 -o output.mp4 -d mps # Apple Silicon python person_mosaic.py input.mp4 -o output.mp4 -d cuda # NVIDIA 참고 구현체: - Ultralytics solutions.ObjectBlurrer (공식) - faceblur-anti-ai (JUNGBAE009) - blurface (GPU 가속 + tqdm) - PrivacyGuard (laythayache) """ import argparse import os import sys import cv2 import numpy as np from tqdm import tqdm def parse_args(): parser = argparse.ArgumentParser( description="YOLO11 기반 실사 인물 전신 모자이크/블러" ) parser.add_argument("input", help="입력 파일 경로 (이미지 또는 영상)") parser.add_argument("-o", "--output", default=None, help="출력 파일 경로 (기본: input_mosaic.ext)") parser.add_argument("-m", "--method", choices=["pixelate", "blur"], default="pixelate", help="모자이크 방식 (기본: pixelate)") parser.add_argument("-s", "--strength", type=int, default=10, help="모자이크 강도 (pixelate: 블록 크기, blur: 커널 크기, 기본: 10)") parser.add_argument("--model", default="yolo11s.pt", help="YOLO 모델 이름 또는 경로 (기본: yolo11s.pt)") parser.add_argument("--conf", type=float, default=0.5, help="탐지 신뢰도 임계값 (기본: 0.5)") parser.add_argument("-d", "--device", default="", help="추론 장치 (cuda, mps, cpu, 기본: auto)") parser.add_argument("--no-tracking", action="store_true", help="영상 처리 시 트래킹 비활성화 (프레임별 개별 탐지)") parser.add_argument("--keep-audio", action="store_true", help="영상 처리 후 원본 오디오 유지 (ffmpeg 필요)") return parser.parse_args() def apply_pixelate(roi: np.ndarray, block_size: int) -> np.ndarray: """ROI 영역에 모자이크(픽셀화) 적용""" if roi.size == 0 or block_size < 1: return roi h, w = roi.shape[:2] safe_block = max(1, block_size) small_w = max(1, w // safe_block) small_h = max(1, h // safe_block) temp = cv2.resize(roi, (small_w, small_h), interpolation=cv2.INTER_LINEAR) return cv2.resize(temp, (w, h), interpolation=cv2.INTER_NEAREST) def apply_blur(roi: np.ndarray, kernel_size: int) -> np.ndarray: """ROI 영역에 가우시안 블러 적용""" if roi.size == 0: return roi # 커널 크기는 홀수여야 함 k = kernel_size if kernel_size % 2 == 1 else kernel_size + 1 k = max(3, k) return cv2.GaussianBlur(roi, (k, k), 0) def mosaic_person_region(frame: np.ndarray, box, method: str, strength: int): """바운딩 박스 영역에 모자이크/블러 적용 (in-place)""" x1, y1, x2, y2 = map(int, box) # 프레임 경계 내로 클램핑 x1 = max(0, x1) y1 = max(0, y1) x2 = min(frame.shape[1], x2) y2 = min(frame.shape[0], y2) if x1 >= x2 or y1 >= y2: return roi = frame[y1:y2, x1:x2] if method == "blur": processed = apply_blur(roi, strength) else: processed = apply_pixelate(roi, strength) frame[y1:y2, x1:x2] = processed def process_image(input_path: str, output_path: str, model, args: argparse.Namespace): """단일 이미지 처리""" img = cv2.imread(input_path) if img is None: print(f"오류: 이미지를 읽을 수 없습니다 — {input_path}") sys.exit(1) results = model(img, classes=[0], conf=args.conf, device=args.device) total_detections = 0 if results[0].boxes is not None: for box in results[0].boxes.xyxy: mosaic_person_region(img, box, args.method, args.strength) total_detections += 1 cv2.imwrite(output_path, img) print(f"완료: {total_detections}명 모자이크 처리 → {output_path}") def process_video(input_path: str, output_path: str, model, args: argparse.Namespace): """영상 처리 (프레임 단위)""" cap = cv2.VideoCapture(input_path) if not cap.isOpened(): print(f"오류: 영상을 열 수 없습니다 — {input_path}") sys.exit(1) total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) fps = cap.get(cv2.CAP_PROP_FPS) w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) fourcc = cv2.VideoWriter_fourcc(*"mp4v") out = cv2.VideoWriter(output_path, fourcc, fps, (w, h)) pbar = tqdm(total=total_frames, desc="모자이크 처리", unit="frame", ncols=80) total_blurred = 0 frame_count = 0 while True: ret, frame = cap.read() if not ret: break if args.no_tracking: results = model(frame, classes=[0], conf=args.conf, device=args.device) else: results = model.track(frame, persist=True, classes=[0], conf=args.conf, device=args.device) frame_blurred = 0 if results[0].boxes is not None and results[0].boxes.xyxy is not None: for box in results[0].boxes.xyxy: mosaic_person_region(frame, box, args.method, args.strength) frame_blurred += 1 out.write(frame) total_blurred += frame_blurred frame_count += 1 pbar.update(1) pbar.set_postfix({"blurred": total_blurred}) cap.release() out.release() pbar.close() avg_per_frame = total_blurred / max(1, frame_count) print(f"\n완료: {total_blurred}명 / {frame_count}프레임 " f"(프레임당 평균 {avg_per_frame:.1f}명)") # 오디오 유지 (선택사항) if args.keep_audio: _preserve_audio(input_path, output_path) def _preserve_audio(src: str, dst: str): """ffmpeg로 원본 오디오를 출력 영상에 복사""" import subprocess temp = dst + ".tmp.mp4" os.rename(dst, temp) cmd = [ "ffmpeg", "-i", temp, "-i", src, "-c:v", "copy", "-c:a", "aac", "-map", "0:v:0", "-map", "1:a:0?", "-y", dst ] try: subprocess.run(cmd, capture_output=True, check=True) os.remove(temp) print(f"오디오 복사 완료 → {dst}") except (subprocess.CalledProcessError, FileNotFoundError) as e: os.rename(temp, dst) print(f"오디오 복사 실패 (ffmpeg 미설치?): {e}") def main(): args = parse_args() # 출력 파일명 자동 생성 if args.output is None: name, ext = os.path.splitext(args.input) args.output = f"{name}_mosaic{ext}" # YOLO 모델 로드 print(f"모델 로딩 중: {args.model}") from ultralytics import YOLO model = YOLO(args.model) # 입력 타입 판별 img_exts = (".jpg", ".jpeg", ".png", ".bmp", ".tiff", ".webp") is_image = args.input.lower().endswith(img_exts) if is_image: process_image(args.input, args.output, model, args) else: process_video(args.input, args.output, model, args) if __name__ == "__main__": main()