"""
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()