From 740063222e045e38ad75791c7f586775322363d6 Mon Sep 17 00:00:00 2001 From: yeyeong5016 <20221373@sungshin.ac.kr> Date: Thu, 28 May 2026 22:08:46 +0900 Subject: [PATCH 1/2] =?UTF-8?q?=ED=86=B5=ED=95=A9=20=ED=8C=8C=EC=9D=B4?= =?UTF-8?q?=ED=94=84=EB=9D=BC=EC=9D=B8=20=EA=B5=AC=EC=A1=B0=20=EB=A7=9E?= =?UTF-8?q?=EC=B6=B0=EC=84=9C=20=EA=B0=9D=EC=B2=B4=ED=83=90=EC=A7=80=20?= =?UTF-8?q?=EB=A1=9C=EC=A7=81=20=EB=B6=84=EB=A6=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config.py | 47 + main.py | 58 ++ models/object_detector.py | 1383 +++++++++++++++++++++++++++++ pipeline/pass3_object_detect.py | 1445 ++----------------------------- 4 files changed, 1542 insertions(+), 1391 deletions(-) create mode 100644 config.py create mode 100644 main.py create mode 100644 models/object_detector.py diff --git a/config.py b/config.py new file mode 100644 index 0000000..5bc5490 --- /dev/null +++ b/config.py @@ -0,0 +1,47 @@ +# config.py + +from pathlib import Path +import torch + +# ========================= +# Project Paths +# ========================= +PROJECT_ROOT = Path(__file__).resolve().parent + +OUTPUT_DIR = PROJECT_ROOT / "outputs" +OBJECT_OUTPUT_DIR = OUTPUT_DIR / "object" + +OBJECT_DB_PATH = OBJECT_OUTPUT_DIR / "object_db.json" +OBJECT_DEBUG_CROP_DIR = OBJECT_OUTPUT_DIR / "debug_crops" + +# ========================= +# Device +# ========================= +if torch.backends.mps.is_available(): + DEVICE = "mps" +elif torch.cuda.is_available(): + DEVICE = "cuda" +else: + DEVICE = "cpu" + +# ========================= +# Object Detection / Privacy Reasoning +# ========================= +OBJECT_QWEN_MODEL_NAME = "Qwen/Qwen2-VL-2B-Instruct" +OBJECT_DEFAULT_PROMPT = "내 프라이버시가 유출될 만한 것들을 가려줘." +OBJECT_SAMPLE_FPS = 1.0 +OBJECT_TEXT_DETECTOR_LANG = "korean" + +OBJECT_CROP_MARGIN_RATIO = 0.35 +OBJECT_MAX_NEW_TOKENS_REASON = 48 +OBJECT_TRACK_IOU_THRESHOLD = 0.30 + +OBJECT_MIN_TEXT_BOX_AREA = 16 +OBJECT_MAX_TEXT_BOX_AREA_RATIO = 0.75 +OBJECT_MIN_TEXT_REGION_WIDTH = 3 +OBJECT_MIN_TEXT_REGION_HEIGHT = 3 +OBJECT_MIN_REC_SCORE = 0.30 +OBJECT_MIN_GROUP_ITEMS = 1 +OBJECT_GROUP_X_GAP_RATIO = 1.6 +OBJECT_GROUP_Y_GAP_RATIO = 1.0 +OBJECT_MIN_QWEN_CROP_SIZE = 56 \ No newline at end of file diff --git a/main.py b/main.py new file mode 100644 index 0000000..3da9258 --- /dev/null +++ b/main.py @@ -0,0 +1,58 @@ +# main.py + +from __future__ import annotations + +import argparse +from pathlib import Path + +import config +from pipeline.pass3_object_detect import run_pass3 + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Privacy Blur Pipeline - Object Detection Test Entry" + ) + + parser.add_argument( + "--video", + required=True, + help="입력 영상 경로", + ) + + parser.add_argument( + "--prompt", + default=config.OBJECT_DEFAULT_PROMPT, + help="객체 탐지용 자연어 프롬프트", + ) + + parser.add_argument( + "--sample-fps", + type=float, + default=config.OBJECT_SAMPLE_FPS, + help="객체 탐지 샘플링 FPS", + ) + + return parser.parse_args() + + +def main() -> None: + args = parse_args() + + video_path = Path(args.video).expanduser().resolve() + + if not video_path.exists(): + raise FileNotFoundError(f"입력 영상을 찾을 수 없습니다: {video_path}") + + config.OBJECT_OUTPUT_DIR.mkdir(parents=True, exist_ok=True) + + run_pass3( + video_path=video_path, + user_prompt=args.prompt, + sample_fps=args.sample_fps, + output_dir=config.OBJECT_OUTPUT_DIR, + ) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/models/object_detector.py b/models/object_detector.py new file mode 100644 index 0000000..b80b950 --- /dev/null +++ b/models/object_detector.py @@ -0,0 +1,1383 @@ +import re +import json +import faulthandler +from dataclasses import dataclass, field +from typing import Dict, List, Optional, Sequence, Tuple, Any + +import cv2 +import numpy as np +from PIL import Image + +from transformers import AutoModelForVision2Seq, AutoProcessor +from paddleocr import PaddleOCR + +@dataclass +class OCRTextItem: + frame_index: int + x1: int + y1: int + x2: int + y2: int + text: str + score: float + + @property + def box(self) -> Tuple[int, int, int, int]: + return (self.x1, self.y1, self.x2, self.y2) + + +@dataclass +class TextGroupCandidate: + frame_index: int + x1: int + y1: int + x2: int + y2: int + merged_text: str + avg_score: float + item_count: int + items: List[OCRTextItem] = field(default_factory=list) + + @property + def box(self) -> Tuple[int, int, int, int]: + return (self.x1, self.y1, self.x2, self.y2) + + +@dataclass +class VerifiedDetection: + frame_index: int + x1: int + y1: int + x2: int + y2: int + label: str + detector_phrase: str + detector_score: float + qwen_visible_text: str = "" + qwen_reason: str = "" + track_id: Optional[str] = None + + @property + def box(self) -> Tuple[int, int, int, int]: + return (self.x1, self.y1, self.x2, self.y2) + + +@dataclass +class TrackState: + object_id: str + label: str + start_frame: int + end_frame: int + last_frame: int + last_box: Tuple[int, int, int, int] + history_boxes: List[Tuple[int, int, int, int]] = field(default_factory=list) + visible_text_samples: List[str] = field(default_factory=list) + reason_samples: List[str] = field(default_factory=list) + hit_count: int = 1 + + def update(self, detection: VerifiedDetection) -> None: + self.end_frame = detection.frame_index + self.last_frame = detection.frame_index + self.last_box = detection.box + self.history_boxes.append(detection.box) + if detection.qwen_visible_text: + self.visible_text_samples.append(detection.qwen_visible_text) + if detection.qwen_reason: + self.reason_samples.append(detection.qwen_reason) + self.hit_count += 1 + + def representative_box(self) -> List[int]: + if not self.history_boxes: + return list(self.last_box) + + arr = np.array(self.history_boxes, dtype=np.float32) + med = np.median(arr, axis=0) + return [int(round(v)) for v in med.tolist()] + + def representative_visible_text(self) -> str: + for text in reversed(self.visible_text_samples): + if text and text.upper() != "NONE": + return text + return "" + + +class PrivacyReasoningEngine: + LABEL_DISPLAY_MAP: Dict[str, str] = { + "license plate": "번호판", + "driver's license": "운전면허증", + "credit card": "신용카드", + "receipt": "영수증", + "address document": "주소 문서", + "other private text": "기타 개인정보 텍스트", + "other": "기타", + } + + WEAK_EVIDENCE_WORDS = { + "card", + "paper", + "screen", + "document", + "printed text", + "text", + "visible text", + "license plate", + "label", + "sign", + "none", + } + + def __init__( + self, + qwen_model_name: str = "Qwen/Qwen2-VL-2B-Instruct", + device: Optional[str] = None, + crop_margin_ratio: float = 0.35, + max_new_tokens_reason: int = 48, + verbose: bool = True, + save_verified_crops: bool = True, + debug_crop_dir: str = "debug_crops", + track_iou_threshold: float = 0.30, + min_text_box_area: int = 16, + max_text_box_area_ratio: float = 0.75, + min_text_region_width: int = 3, + min_text_region_height: int = 3, + text_detector_lang: str = "korean", + min_rec_score: float = 0.30, + min_group_items: int = 1, + group_x_gap_ratio: float = 1.6, + group_y_gap_ratio: float = 1.0, + min_qwen_crop_size: int = 56, + ) -> None: + self.verbose = bool(verbose) + self.save_verified_crops = bool(save_verified_crops) + self.debug_crop_dir = Path(debug_crop_dir) + + self.device = device or ("cuda" if torch.cuda.is_available() else "cpu") + self.crop_margin_ratio = float(crop_margin_ratio) + self.max_new_tokens_reason = int(max_new_tokens_reason) + self.track_iou_threshold = float(track_iou_threshold) + + self.min_text_box_area = int(min_text_box_area) + self.max_text_box_area_ratio = float(max_text_box_area_ratio) + self.min_text_region_width = int(min_text_region_width) + self.min_text_region_height = int(min_text_region_height) + self.min_rec_score = float(min_rec_score) + self.min_group_items = int(min_group_items) + self.group_x_gap_ratio = float(group_x_gap_ratio) + self.group_y_gap_ratio = float(group_y_gap_ratio) + self.min_qwen_crop_size = int(min_qwen_crop_size) + + if self.save_verified_crops: + self.debug_crop_dir.mkdir(parents=True, exist_ok=True) + + self._next_track_index = 1 + self.active_tracks: Dict[str, TrackState] = {} + self.finished_tracks: List[TrackState] = [] + self.last_verified_detections: List[VerifiedDetection] = [] + self.last_tracks: List[TrackState] = [] + self.last_json_payload: List[Dict] = [] + + self._log("엔진 초기화 시작") + self._log(f"device = {self.device}") + self._log(f"crop_margin_ratio = {self.crop_margin_ratio}") + self._log(f"track_iou_threshold = {self.track_iou_threshold}") + self._log(f"text_detector_lang = {text_detector_lang}") + self._log(f"min_text_box_area = {self.min_text_box_area}") + self._log(f"min_text_region_width = {self.min_text_region_width}") + self._log(f"min_text_region_height = {self.min_text_region_height}") + self._log(f"min_rec_score = {self.min_rec_score}") + self._log(f"min_qwen_crop_size = {self.min_qwen_crop_size}") + + self.qwen_processor = AutoProcessor.from_pretrained( + qwen_model_name, + trust_remote_code=True, + ) + self._log("Qwen2-VL Processor 로딩 완료") + + self.qwen_model = AutoModelForVision2Seq.from_pretrained( + qwen_model_name, + trust_remote_code=True, + torch_dtype=torch.float16 if self.device == "cuda" else torch.float32, + ).to(self.device) + self.qwen_model.eval() + self._log("Qwen2-VL Model 로딩 완료") + + if hasattr(self.qwen_model, "generation_config") and self.qwen_model.generation_config is not None: + for attr in ("temperature", "top_p", "top_k"): + if hasattr(self.qwen_model.generation_config, attr): + setattr(self.qwen_model.generation_config, attr, None) + if hasattr(self.qwen_model.generation_config, "do_sample"): + self.qwen_model.generation_config.do_sample = False + + self.text_detector = PaddleOCR( + lang=text_detector_lang, + use_doc_orientation_classify=False, + use_doc_unwarping=False, + use_textline_orientation=False, + ) + self._log("PaddleOCR detector+recognizer 로딩 완료") + + # ------------------------------------------------------------------------- + # Logging + # ------------------------------------------------------------------------- + def _log(self, message: str) -> None: + if self.verbose: + print(f"[Info] {message}", flush=True) + + def _log_group_found(self, frame_index: int, count: int) -> None: + print(f"[Frame {frame_index}] Group 후보 {count}개 발견", flush=True) + + def _log_verified(self, detection: VerifiedDetection) -> None: + print( + f"[Frame {detection.frame_index}] 검증 통과: {detection.label} " + f"| box=({detection.x1}, {detection.y1}, {detection.x2}, {detection.y2}) " + f"| id={detection.track_id}", + flush=True, + ) + + # ------------------------------------------------------------------------- + # Prompt helpers + # ------------------------------------------------------------------------- + def build_privacy_reason_prompt( + self, + user_prompt: str, + visible_text: str, + ) -> str: + return ( + "You are deciding whether a text group and its surrounding object should be privacy-blurred.\n" + f"User request: {user_prompt}\n" + f"Visible text from the grouped region:\n{visible_text}\n" + "Decide based on the visible text and obvious visible context only.\n" + "If this grouped region likely belongs to a privacy-sensitive object such as a receipt, waybill(delivery label with name/address/phone number), address label, card, ID document, license plate, or other personal text-bearing object, answer yes. Road signs, subtitles, and brand logos are NOT privacy-sensitive.\n" + "Short noisy fragments such as a single digit or random short token alone are not enough unless the overall grouped region strongly suggests a sensitive object.\n" + "If any part of the grouped region likely contains privacy-sensitive information, treat the whole grouped object as blur-worthy.\n" + "You MUST return exactly three lines in this order, do NOT skip or merge any line:\n" + "Decision: yes or no\n" + "Label: receipt | waybill | credit card | driver's license | license plate | road sign | address document | other private text | other\n" + "Reason: \n" + "Example output:\n" + "Decision: yes\n" + "Label: waybill\n" + "Reason: Contains name, address and phone number typical of a delivery label.\n" + ) + + # ------------------------------------------------------------------------- + # Qwen helpers + # ------------------------------------------------------------------------- + def ask_qwen_reason( + self, + image_pil: Image.Image, + prompt_text: str, + max_new_tokens: int, + ) -> str: + messages = [ + { + "role": "user", + "content": [ + {"type": "image", "image": image_pil}, + {"type": "text", "text": prompt_text}, + ], + } + ] + + text_prompt = self.qwen_processor.apply_chat_template( + messages, + tokenize=False, + add_generation_prompt=True, + ) + + inputs = self.qwen_processor( + text=[text_prompt], + images=[image_pil], + padding=True, + return_tensors="pt", + ) + + for key, value in inputs.items(): + if hasattr(value, "to"): + inputs[key] = value.to(self.device) + + with torch.inference_mode(): + output_ids = self.qwen_model.generate( + **inputs, + max_new_tokens=max_new_tokens, + do_sample=False, + use_cache=True, + ) + + prompt_len = int(inputs["input_ids"].shape[1]) + generated_ids = output_ids[:, prompt_len:] + + answer = self.qwen_processor.batch_decode( + generated_ids, + skip_special_tokens=True, + clean_up_tokenization_spaces=True, + )[0] + + return answer.strip() + + # ------------------------------------------------------------------------- + # Public API + # ------------------------------------------------------------------------- + def process_video( + self, + video_path: str, + user_prompt: str = "내 프라이버시가 유출될 만한 것들을 가려줘.", + sample_fps: float = 1.0, + ) -> List[TrackState]: + print("[Start] process_video 시작", flush=True) + print(f"[Start] video_path = {video_path}", flush=True) + print(f"[Start] user_prompt = {user_prompt}", flush=True) + print(f"[Start] sample_fps = {sample_fps}", flush=True) + + if sample_fps <= 0: + raise ValueError("sample_fps must be > 0.") + + self.active_tracks = {} + self.finished_tracks = [] + self.last_verified_detections = [] + self.last_tracks = [] + self.last_json_payload = [] + self._next_track_index = 1 + + cap = cv2.VideoCapture(video_path) + print(f"[Start] VideoCapture isOpened = {cap.isOpened()}", flush=True) + + if not cap.isOpened(): + raise FileNotFoundError(f"Could not open video: {video_path}") + + total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) + native_fps = float(cap.get(cv2.CAP_PROP_FPS)) + width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) + height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) + + if native_fps <= 0 or not np.isfinite(native_fps): + native_fps = 30.0 + + sample_interval_frames = max(1, int(round(native_fps / sample_fps))) + max_track_gap_frames = max(sample_interval_frames * 2, sample_interval_frames + 1) + + print(f"[Start] total_frames = {total_frames}", flush=True) + print(f"[Start] native_fps = {native_fps}", flush=True) + print(f"[Start] resolution = {width}x{height}", flush=True) + print(f"[Start] sample_interval_frames = {sample_interval_frames}", flush=True) + print(f"[Start] max_track_gap_frames = {max_track_gap_frames}", flush=True) + + frame_index = 0 + sampled_count = 0 + + try: + while True: + ret, frame_bgr = cap.read() + if not ret: + break + + if frame_index % sample_interval_frames != 0: + frame_index += 1 + continue + + sampled_count += 1 + + try: + groups = self.detect_grouped_text_candidates( + frame_bgr=frame_bgr, + frame_index=frame_index, + ) + except Exception as e: + print(f"[Error] Frame {frame_index} detect/group 실패: {e}", flush=True) + traceback.print_exc() + frame_index += 1 + continue + + if len(groups) > 0: + self._log_group_found(frame_index, len(groups)) + + frame_verified: List[VerifiedDetection] = [] + + for group in groups: + try: + verified_detection = self.verify_group_reason_only( + frame_bgr=frame_bgr, + group=group, + user_prompt=user_prompt, + ) + except Exception as e: + print(f"[Error] Frame {frame_index} reason 실패: {e}", flush=True) + traceback.print_exc() + continue + + if verified_detection is not None: + frame_verified.append(verified_detection) + + self.update_tracks_for_frame( + detections=frame_verified, + frame_index=frame_index, + max_track_gap_frames=max_track_gap_frames, + ) + + for det in frame_verified: + self.last_verified_detections.append(det) + self._log_verified(det) + + frame_index += 1 + + finally: + cap.release() + + self.finalize_all_tracks() + self.last_tracks = list(self.finished_tracks) + + print(f"[Done] 샘플링된 프레임 수 = {sampled_count}", flush=True) + print(f"[Done] 총 검증 통과 수 = {len(self.last_verified_detections)}", flush=True) + print(f"[Done] 총 트랙 수 = {len(self.last_tracks)}", flush=True) + + return self.last_tracks + + # ------------------------------------------------------------------------- + # Stage 1: PaddleOCR detection + recognition + grouping + # ------------------------------------------------------------------------- + def detect_grouped_text_candidates( + self, + frame_bgr: np.ndarray, + frame_index: int, + ) -> List[TextGroupCandidate]: + height, width = frame_bgr.shape[:2] + frame_area = float(width * height) + + frame_rgb = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB) + raw_results = self.text_detector.predict(frame_rgb) + + items = self.extract_text_items_from_predict_results(raw_results, frame_index) + + filtered_items: List[OCRTextItem] = [] + for item in items: + x1, y1, x2, y2 = item.box + box_w = max(1, x2 - x1) + box_h = max(1, y2 - y1) + area = float(box_w * box_h) + + if area < self.min_text_box_area: + continue + if area > frame_area * self.max_text_box_area_ratio: + continue + if box_w < self.min_text_region_width or box_h < self.min_text_region_height: + continue + if item.score < self.min_rec_score: + continue + if self.is_missing_or_weak_visible_text(item.text): + continue + if self.visible_text_is_gibberish(item.text): + continue + + filtered_items.append(item) + + print(f"[Debug][Frame {frame_index}] filtered text items = {len(filtered_items)}", flush=True) + + if len(filtered_items) == 0: + return [] + + groups = self.group_text_items(filtered_items, frame_w=width, frame_h=height) + + candidates: List[TextGroupCandidate] = [] + for group_items in groups: + if len(group_items) < self.min_group_items: + continue + + gx1 = min(item.x1 for item in group_items) + gy1 = min(item.y1 for item in group_items) + gx2 = max(item.x2 for item in group_items) + gy2 = max(item.y2 for item in group_items) + + merged_text = self.build_group_text(group_items) + avg_score = float(sum(item.score for item in group_items) / max(1, len(group_items))) + + if self.group_text_is_too_weak(merged_text, len(group_items)): + continue + + candidates.append( + TextGroupCandidate( + frame_index=frame_index, + x1=gx1, + y1=gy1, + x2=gx2, + y2=gy2, + merged_text=merged_text, + avg_score=avg_score, + item_count=len(group_items), + items=group_items, + ) + ) + + print(f"[Debug][Frame {frame_index}] group candidates = {len(candidates)}", flush=True) + for i, cand in enumerate(candidates[:10]): + print( + f"[Debug][Frame {frame_index}] group[{i}] items={cand.item_count} score={cand.avg_score:.3f} text={repr(cand.merged_text)} box={cand.box}", + flush=True, + ) + + return candidates + + def extract_text_items_from_predict_results( + self, + raw_results: Any, + frame_index: int, + ) -> List[OCRTextItem]: + items: List[OCRTextItem] = [] + + if raw_results is None: + return items + + try: + result_list = list(raw_results) + except TypeError: + result_list = [raw_results] + + if frame_index == 0: + print(f"[Debug][Frame {frame_index}] predict result count = {len(result_list)}", flush=True) + + for item in result_list: + data = None + + if hasattr(item, "res"): + data = item.res + elif isinstance(item, dict): + data = item + else: + try: + data = dict(item) + except Exception: + data = None + + if not isinstance(data, dict): + continue + + if frame_index == 0: + print(f"[Debug][Frame {frame_index}] result keys = {list(data.keys())}", flush=True) + + rec_texts = data.get("rec_texts", None) + rec_scores = data.get("rec_scores", None) + rec_boxes = data.get("rec_boxes", None) + rec_polys = data.get("rec_polys", None) + dt_polys = data.get("dt_polys", None) + + if rec_texts is None: + continue + + rec_texts = list(rec_texts) + rec_scores = list(rec_scores) if rec_scores is not None else [1.0] * len(rec_texts) + + box_source = None + if rec_boxes is not None: + box_source = rec_boxes + elif rec_polys is not None: + box_source = rec_polys + elif dt_polys is not None: + box_source = dt_polys + + if box_source is None: + continue + + try: + box_arr = np.asarray(box_source, dtype=np.float32) + except Exception: + continue + + if box_arr.ndim == 2 and box_arr.shape[-1] == 4: + for i, text in enumerate(rec_texts): + if i >= len(box_arr): + break + x1, y1, x2, y2 = box_arr[i].tolist() + items.append( + OCRTextItem( + frame_index=frame_index, + x1=int(round(x1)), + y1=int(round(y1)), + x2=int(round(x2)), + y2=int(round(y2)), + text=str(text).strip(), + score=float(rec_scores[i]) if i < len(rec_scores) else 1.0, + ) + ) + + elif box_arr.ndim == 3 and box_arr.shape[-1] == 2: + for i, text in enumerate(rec_texts): + if i >= len(box_arr): + break + poly = box_arr[i] + xs = poly[:, 0] + ys = poly[:, 1] + items.append( + OCRTextItem( + frame_index=frame_index, + x1=int(round(float(xs.min()))), + y1=int(round(float(ys.min()))), + x2=int(round(float(xs.max()))), + y2=int(round(float(ys.max()))), + text=str(text).strip(), + score=float(rec_scores[i]) if i < len(rec_scores) else 1.0, + ) + ) + + return items + + def group_text_items(self, items: List[OCRTextItem], frame_w: int = 9999, frame_h: int = 9999) -> List[List[OCRTextItem]]: + if len(items) == 0: + return [] + + items = sorted(items, key=lambda it: (it.y1, it.x1)) + + groups: List[List[OCRTextItem]] = [] + used = [False] * len(items) + + for i in range(len(items)): + if used[i]: + continue + + current_group = [items[i]] + used[i] = True + + changed = True + while changed: + changed = False + gx1 = min(it.x1 for it in current_group) + gy1 = min(it.y1 for it in current_group) + gx2 = max(it.x2 for it in current_group) + gy2 = max(it.y2 for it in current_group) + gh = max(1, gy2 - gy1) + gw = max(1, gx2 - gx1) + + for j in range(len(items)): + if used[j]: + continue + + it = items[j] + iou = self.compute_iou((gx1, gy1, gx2, gy2), it.box) + + x_gap = self.box_horizontal_gap((gx1, gy1, gx2, gy2), it.box) + y_gap = self.box_vertical_gap((gx1, gy1, gx2, gy2), it.box) + + close_x = x_gap <= int(round(gh * self.group_x_gap_ratio)) + close_y = y_gap <= int(round(gh * self.group_y_gap_ratio)) + overlaps = iou > 0.0 + + within_band = ( + it.x1 <= gx2 + int(round(gw * 0.3)) + and it.x2 >= gx1 - int(round(gw * 0.3)) + and it.y1 <= gy2 + int(round(gh * 0.8)) + and it.y2 >= gy1 - int(round(gh * 0.8)) + ) + max_group_area = frame_w * frame_h * 0.15 + if (gx2 - gx1) * (gy2 - gy1) > max_group_area: + break + if (close_x and close_y) or overlaps or within_band: + current_group.append(it) + used[j] = True + changed = True + + groups.append(sorted(current_group, key=lambda it: (it.y1, it.x1))) + + return groups + + def build_group_text(self, items: List[OCRTextItem]) -> str: + if len(items) == 0: + return "" + + lines: List[List[OCRTextItem]] = [] + for item in sorted(items, key=lambda it: (it.y1, it.x1)): + placed = False + for line in lines: + ly1 = min(x.y1 for x in line) + ly2 = max(x.y2 for x in line) + lh = max(1, ly2 - ly1) + if abs(item.y1 - ly1) <= int(round(lh * 0.6)): + line.append(item) + placed = True + break + if not placed: + lines.append([item]) + + out_lines: List[str] = [] + for line in lines: + line_sorted = sorted(line, key=lambda it: it.x1) + line_text = " ".join(it.text.strip() for it in line_sorted if it.text.strip()) + if line_text: + out_lines.append(line_text) + + return "\n".join(out_lines) + + def group_text_is_too_weak(self, merged_text: str, item_count: int) -> bool: + text = merged_text.strip() + if text == "": + return True + + norm = self.normalize_text(text) + if norm in ("none", "unknown", "unreadable"): + return True + + has_digits = re.search(r"\d", text) is not None + has_hangul = re.search(r"[가-힣]", text) is not None + has_alpha_word = re.search(r"[A-Za-z]{3,}", text) is not None + has_cjk = re.search(r"[\u4e00-\u9fff]", text) is not None + + if item_count == 1: + compact = re.sub(r"\s+", "", text) + if len(compact) <= 2 and not has_hangul and not has_alpha_word: + return True + + if not (has_digits or has_hangul or has_alpha_word or has_cjk): + return True + + return False + + # ------------------------------------------------------------------------- + # Stage 2: Qwen reasoning only on grouped object + # ------------------------------------------------------------------------- + def verify_group_reason_only( + self, + frame_bgr: np.ndarray, + group: TextGroupCandidate, + user_prompt: str, + ) -> Optional[VerifiedDetection]: + frame_index = group.frame_index + visible_text = group.merged_text.strip() + + if self.group_text_is_too_weak(visible_text, group.item_count): + return None + + crop_rgb = None + try: + crop_rgb = self.crop_with_margin( + frame_bgr=frame_bgr, + x1=group.x1, + y1=group.y1, + x2=group.x2, + y2=group.y2, + margin_ratio=self.crop_margin_ratio, + ) + + if crop_rgb is None: + return None + + crop_rgb = self.ensure_min_crop_size(crop_rgb, self.min_qwen_crop_size) + image_pil = Image.fromarray(crop_rgb) + + print( + f"[Group-REC][Frame {frame_index}] items={group.item_count} score={group.avg_score:.3f} text={repr(visible_text)} box={group.box}", + flush=True, + ) + + reason_prompt = self.build_privacy_reason_prompt( + user_prompt=user_prompt, + visible_text=visible_text, + ) + raw_reason_answer = self.ask_qwen_reason( + image_pil=image_pil, + prompt_text=reason_prompt, + max_new_tokens=self.max_new_tokens_reason, + ) + + decision, normalized_label, reason = self.parse_qwen_reason_answer( + answer=raw_reason_answer, + fallback_label="other private text", + ) + + print( + f"[Qwen-Reason][Frame {frame_index}] decision={decision} label={normalized_label} text={repr(visible_text)} reason={reason} raw_answer={repr(raw_reason_answer)}", + flush=True, + ) + + if not decision: + return None + + det = VerifiedDetection( + frame_index=frame_index, + x1=group.x1, + y1=group.y1, + x2=group.x2, + y2=group.y2, + label=normalized_label, + detector_phrase="grouped text object", + detector_score=float(group.avg_score), + qwen_visible_text=visible_text, + qwen_reason=reason, + ) + + if self.save_verified_crops: + self.save_debug_crop(crop_rgb=crop_rgb, detection=det) + + return det + + except Exception as e: + print(f"[Error] Frame {frame_index} group reason 예외: {e}", flush=True) + traceback.print_exc() + return None + + finally: + if torch.cuda.is_available(): + torch.cuda.empty_cache() + gc.collect() + + def parse_qwen_reason_answer( + self, + answer: str, + fallback_label: str, + ) -> Tuple[bool, str, str]: + text = answer.strip() + + decision_match = re.search(r"decision\s*:\s*(yes|no)", text, flags=re.IGNORECASE) + label_match = re.search( + r"label\s*:\s*(receipt|credit card|driver'?s license|license plate|address document|other private text|other)", + text, + flags=re.IGNORECASE, + ) + reason_match = re.search(r"reason\s*:\s*(.+)", text, flags=re.IGNORECASE | re.DOTALL) + + decision = False + if decision_match is not None: + decision = decision_match.group(1).lower() == "yes" + else: + decision = self.parse_yes_no(text) + + raw_label = fallback_label + if label_match is not None: + raw_label = label_match.group(1).lower().strip() + + reason = "" + if reason_match is not None: + reason = reason_match.group(1).strip().splitlines()[0].strip() + + normalized_label = self.normalize_label_name(raw_label) + return decision, normalized_label, reason + + def is_missing_or_weak_visible_text(self, visible_text: str) -> bool: + if visible_text is None: + return True + + vt = visible_text.strip() + if vt == "": + return True + + vt_norm = self.normalize_text(vt) + if vt_norm in ("none", "n a", "na", "unknown", "not visible", "unreadable"): + return True + + if vt_norm in self.WEAK_EVIDENCE_WORDS: + return True + + has_digits = re.search(r"\d", vt) is not None + has_hangul = re.search(r"[가-힣]", vt) is not None + has_alpha_word = re.search(r"[A-Za-z]{2,}", vt) is not None + has_cjk = re.search(r"[\u4e00-\u9fff]", vt) is not None + + if not (has_digits or has_hangul or has_alpha_word or has_cjk): + return True + + return False + + def visible_text_is_gibberish(self, visible_text: str) -> bool: + vt = visible_text.strip() + + if len(vt) >= 8 and len(set(vt)) == 1: + return True + + if re.fullmatch(r"(\d)\1{7,}", vt): + return True + if re.fullmatch(r"([A-Za-z])\1{7,}", vt): + return True + + digits = re.sub(r"\D", "", vt) + if len(digits) >= 12 and len(set(digits)) <= 2: + return True + + cleaned = re.sub(r"[0-9A-Za-z가-힣\u4e00-\u9fff\s:/\-.(),#]", "", vt) + if len(cleaned) >= 5 and len(vt) >= 10: + return True + + return False + + def normalize_label_name(self, label: str) -> str: + n = self.normalize_text(label) + if n in ("driver s license", "driver's license", "driver license", "driving license", "id card", "identity card"): + return "driver's license" + if n in ("credit card", "debit card", "bank card", "payment card", "card"): + return "credit card" + if n in ("receipt", "bill", "paper receipt", "purchase receipt"): + return "receipt" + if n in ("license plate", "number plate", "plate", "car plate", "vehicle plate"): + return "license plate" + if n in ("address document",): + return "address document" + if n in ("waybill", "delivery label", "shipping label", "invoice"): + return "waybill" + if n in ("road sign", "street sign", "traffic sign", "sign"): + return "road sign" + if n in ("other private text", "other private", "private text"): + return "other private text" + return "other" + + def save_debug_crop(self, crop_rgb: np.ndarray, detection: VerifiedDetection) -> None: + label = str(detection.label).replace(" ", "_").replace("/", "_").replace("'", "") + filename = ( + f"frame_{detection.frame_index}_{label}_" + f"{detection.x1}_{detection.y1}_{detection.x2}_{detection.y2}.jpg" + ) + save_path = self.debug_crop_dir / filename + Image.fromarray(crop_rgb).save(save_path) + + # ------------------------------------------------------------------------- + # Stage 3: Simple ID-based tracking + # ------------------------------------------------------------------------- + def update_tracks_for_frame( + self, + detections: List[VerifiedDetection], + frame_index: int, + max_track_gap_frames: int, + ) -> None: + expired_ids = [ + track_id + for track_id, track in self.active_tracks.items() + if frame_index - track.last_frame > max_track_gap_frames + ] + for track_id in expired_ids: + self.finished_tracks.append(self.active_tracks.pop(track_id)) + + if len(detections) == 0: + return + + track_ids = list(self.active_tracks.keys()) + candidate_pairs: List[Tuple[float, int, str]] = [] + + for det_idx, det in enumerate(detections): + for track_id in track_ids: + track = self.active_tracks[track_id] + # 라벨이 달라도 위치(IoU)가 겹치면 같은 객체로 인정 + # if track.label != det.label: + # continue + + iou = self.compute_iou(track.last_box, det.box) + if iou >= self.track_iou_threshold: + candidate_pairs.append((iou, det_idx, track_id)) + + candidate_pairs.sort(key=lambda x: x[0], reverse=True) + + matched_det_indices = set() + matched_track_ids = set() + + for _, det_idx, track_id in candidate_pairs: + if det_idx in matched_det_indices or track_id in matched_track_ids: + continue + + det = detections[det_idx] + track = self.active_tracks[track_id] + track.update(det) + det.track_id = track.object_id + + matched_det_indices.add(det_idx) + matched_track_ids.add(track_id) + + for det_idx, det in enumerate(detections): + if det_idx in matched_det_indices: + continue + + new_track = self.create_track(det) + self.active_tracks[new_track.object_id] = new_track + det.track_id = new_track.object_id + + def create_track(self, detection: VerifiedDetection) -> TrackState: + object_id = f"object_{self._next_track_index:03d}" + self._next_track_index += 1 + + return TrackState( + object_id=object_id, + label=detection.label, + start_frame=detection.frame_index, + end_frame=detection.frame_index, + last_frame=detection.frame_index, + last_box=detection.box, + history_boxes=[detection.box], + visible_text_samples=[detection.qwen_visible_text] if detection.qwen_visible_text else [], + reason_samples=[detection.qwen_reason] if detection.qwen_reason else [], + hit_count=1, + ) + + def finalize_all_tracks(self) -> None: + for track_id in list(self.active_tracks.keys()): + self.finished_tracks.append(self.active_tracks.pop(track_id)) + + @staticmethod + def compute_iou( + box_a: Tuple[int, int, int, int], + box_b: Tuple[int, int, int, int], + ) -> float: + ax1, ay1, ax2, ay2 = box_a + bx1, by1, bx2, by2 = box_b + + inter_x1 = max(ax1, bx1) + inter_y1 = max(ay1, by1) + inter_x2 = min(ax2, bx2) + inter_y2 = min(ay2, by2) + + inter_w = max(0, inter_x2 - inter_x1) + inter_h = max(0, inter_y2 - inter_y1) + inter_area = inter_w * inter_h + + area_a = max(0, ax2 - ax1) * max(0, ay2 - ay1) + area_b = max(0, bx2 - bx1) * max(0, by2 - by1) + + union_area = area_a + area_b - inter_area + if union_area <= 0: + return 0.0 + + return float(inter_area / union_area) + + @staticmethod + def box_horizontal_gap( + box_a: Tuple[int, int, int, int], + box_b: Tuple[int, int, int, int], + ) -> int: + ax1, _, ax2, _ = box_a + bx1, _, bx2, _ = box_b + if ax2 < bx1: + return bx1 - ax2 + if bx2 < ax1: + return ax1 - bx2 + return 0 + + @staticmethod + def box_vertical_gap( + box_a: Tuple[int, int, int, int], + box_b: Tuple[int, int, int, int], + ) -> int: + _, ay1, _, ay2 = box_a + _, by1, _, by2 = box_b + if ay2 < by1: + return by1 - ay2 + if by2 < ay1: + return ay1 - by2 + return 0 + + # ------------------------------------------------------------------------- + # Export / debug outputs + # ------------------------------------------------------------------------- + def export_to_json( + self, + output_json_path: str, + tracks: Optional[List[TrackState]] = None, + ) -> List[Dict]: + if tracks is None: + tracks = list(self.last_tracks) # 복사본으로 + # 중복 제거 전 디버그 + print(f"[Debug] 중복 제거 전 트랙 수: {len(tracks)}", flush=True) + for t in tracks: + print(f"[RawTrack] {t.object_id} {t.start_frame} ~ {t.end_frame} {t.representative_box()}", flush=True) + # raw 트랙 저장 + import json as _json + raw_payload = [] + for t in tracks: + raw_payload.append({ + "id": t.object_id, + "start_frame": int(t.start_frame), + "end_frame": int(t.end_frame), + "box": t.representative_box(), + "visible_text": t.representative_visible_text(), + }) + with open("raw_tracks.json", "w", encoding="utf-8") as f: + _json.dump(raw_payload, f, ensure_ascii=False, indent=2) + print("[Debug] raw_tracks.json 저장 완료", flush=True) + + # IoU 기반 중복 트랙 제거 + + def compute_iou_box(a, b): + ax1, ay1, ax2, ay2 = a + bx1, by1, bx2, by2 = b + ix1, iy1 = max(ax1, bx1), max(ay1, by1) + ix2, iy2 = min(ax2, bx2), min(ay2, by2) + inter = max(0, ix2-ix1) * max(0, iy2-iy1) + if inter == 0: + return 0.0 + area_a = (ax2-ax1) * (ay2-ay1) + area_b = (bx2-bx1) * (by2-by1) + return inter / (area_a + area_b - inter) + + merged_tracks = [] + used = set() + for i, t1 in enumerate(tracks): + if i in used: + continue + box1 = t1.representative_box() + # 화면 너무 큰 박스 제거 + bw = box1[2] - box1[0] + bh = box1[3] - box1[1] + if bw * bh > 300000: + used.add(i) + continue + best_start = t1.start_frame + best_end = t1.end_frame + for j, t2 in enumerate(tracks): + if i == j or j in used: + continue + box2 = t2.representative_box() + if compute_iou_box(box1, box2) >= 0.5: + best_start = min(best_start, t2.start_frame) + best_end = max(best_end, t2.end_frame) + used.add(j) + t1.start_frame = best_start + t1.end_frame = best_end + merged_tracks.append(t1) + used.add(i) + tracks = merged_tracks + + print(f"[Debug] 중복 제거 후 트랙 수: {len(tracks)}", flush=True) + + payload: List[Dict] = [] + for track in tracks: + if track.start_frame == track.end_frame: + track.end_frame = track.start_frame + 10 + payload.append( + { + "id": track.object_id, + "type": "object", + "label": self.LABEL_DISPLAY_MAP.get(track.label, track.label), + "start_frame": int(track.start_frame), + "end_frame": int(track.end_frame), + "box": track.representative_box(), + "visible_text": track.representative_visible_text(), + } + ) + + out_path = Path(output_json_path) + out_path.parent.mkdir(parents=True, exist_ok=True) + + with open(out_path, "w", encoding="utf-8") as f: + json.dump(payload, f, ensure_ascii=False, indent=2) + + self.last_json_payload = payload + + print(f"[Export] JSON 저장 완료: {output_json_path}", flush=True) + return payload + + def group_detections_by_frame( + self, + detections: Optional[List[VerifiedDetection]] = None, + ) -> Dict[int, List[VerifiedDetection]]: + if detections is None: + detections = self.last_verified_detections + + grouped: Dict[int, List[VerifiedDetection]] = {} + for det in detections: + grouped.setdefault(det.frame_index, []).append(det) + return grouped + + def write_debug_video_fullfps( + self, + input_video_path: str, + output_video_path: str, + detections: Optional[List[VerifiedDetection]] = None, + ) -> None: + cap = cv2.VideoCapture(input_video_path) + if not cap.isOpened(): + raise FileNotFoundError(f"Could not open input video: {input_video_path}") + + fps = float(cap.get(cv2.CAP_PROP_FPS)) + width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) + height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) + + if fps <= 0 or not np.isfinite(fps): + fps = 24.0 + + fourcc = cv2.VideoWriter_fourcc(*"mp4v") + writer = cv2.VideoWriter(output_video_path, fourcc, fps, (width, height)) + + grouped = self.group_detections_by_frame(detections) + + frame_index = 0 + try: + while True: + ret, frame_bgr = cap.read() + if not ret: + break + + frame_dets = grouped.get(frame_index, []) + for det in frame_dets: + self.draw_bbox( + frame_bgr, + det.x1, + det.y1, + det.x2, + det.y2, + f"{det.track_id}:{self.LABEL_DISPLAY_MAP.get(det.label, det.label)}", + ) + + writer.write(frame_bgr) + frame_index += 1 + + finally: + cap.release() + writer.release() + + print(f"[Debug] full-fps bbox 영상 저장 완료: {output_video_path}", flush=True) + + def write_debug_video_sampled( + self, + input_video_path: str, + output_video_path: str, + sample_fps: float, + detections: Optional[List[VerifiedDetection]] = None, + ) -> None: + cap = cv2.VideoCapture(input_video_path) + if not cap.isOpened(): + raise FileNotFoundError(f"Could not open input video: {input_video_path}") + + native_fps = float(cap.get(cv2.CAP_PROP_FPS)) + width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) + height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) + + if native_fps <= 0 or not np.isfinite(native_fps): + native_fps = 24.0 + + sample_interval_frames = max(1, int(round(native_fps / sample_fps))) + + fourcc = cv2.VideoWriter_fourcc(*"mp4v") + writer = cv2.VideoWriter(output_video_path, fourcc, sample_fps, (width, height)) + + grouped = self.group_detections_by_frame(detections) + + frame_index = 0 + try: + while True: + ret, frame_bgr = cap.read() + if not ret: + break + + if frame_index % sample_interval_frames != 0: + frame_index += 1 + continue + + frame_dets = grouped.get(frame_index, []) + for det in frame_dets: + self.draw_bbox( + frame_bgr, + det.x1, + det.y1, + det.x2, + det.y2, + f"{det.track_id}:{self.LABEL_DISPLAY_MAP.get(det.label, det.label)}", + ) + + writer.write(frame_bgr) + frame_index += 1 + + finally: + cap.release() + writer.release() + + print(f"[Debug] sampled bbox 영상 저장 완료: {output_video_path}", flush=True) + + @staticmethod + def draw_bbox(frame_bgr: np.ndarray, x1: int, y1: int, x2: int, y2: int, label: str) -> None: + x1, y1, x2, y2 = int(x1), int(y1), int(x2), int(y2) + cv2.rectangle(frame_bgr, (x1, y1), (x2, y2), (0, 255, 0), 2) + + # PIL로 한글 텍스트 그리기 + from PIL import ImageFont, ImageDraw, Image + img_pil = Image.fromarray(cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB)) + draw = ImageDraw.Draw(img_pil) + try: + font = ImageFont.truetype("/usr/share/fonts/truetype/nanum/NanumGothic.ttf", 20) + except: + font = ImageFont.load_default() + text_y = max(5, y1 - 25) + draw.text((x1, text_y), label, font=font, fill=(0, 255, 0)) + frame_bgr[:] = cv2.cvtColor(np.array(img_pil), cv2.COLOR_RGB2BGR) + + # ------------------------------------------------------------------------- + # Utilities + # ------------------------------------------------------------------------- + def crop_with_margin( + self, + frame_bgr: np.ndarray, + x1: int, + y1: int, + x2: int, + y2: int, + margin_ratio: float = 0.35, + ) -> Optional[np.ndarray]: + height, width = frame_bgr.shape[:2] + + box_w = max(1, x2 - x1) + box_h = max(1, y2 - y1) + + margin_x = int(round(box_w * margin_ratio)) + margin_y = int(round(box_h * margin_ratio)) + + crop_x1 = max(0, x1 - margin_x) + crop_y1 = max(0, y1 - margin_y) + crop_x2 = min(width, x2 + margin_x) + crop_y2 = min(height, y2 + margin_y) + + if crop_x2 <= crop_x1 or crop_y2 <= crop_y1: + return None + + crop_bgr = frame_bgr[crop_y1:crop_y2, crop_x1:crop_x2] + if crop_bgr.size == 0: + return None + + crop_rgb = cv2.cvtColor(crop_bgr, cv2.COLOR_BGR2RGB) + return crop_rgb + + def ensure_min_crop_size(self, crop_rgb: np.ndarray, min_size: int) -> np.ndarray: + h, w = crop_rgb.shape[:2] + new_h = max(h, min_size) + new_w = max(w, min_size) + + if new_h == h and new_w == w: + return crop_rgb + + pad_top = (new_h - h) // 2 + pad_bottom = new_h - h - pad_top + pad_left = (new_w - w) // 2 + pad_right = new_w - w - pad_left + + return cv2.copyMakeBorder( + crop_rgb, + pad_top, + pad_bottom, + pad_left, + pad_right, + borderType=cv2.BORDER_REPLICATE, + ) + + @staticmethod + def parse_yes_no(answer: str) -> bool: + text = answer.lower().strip() + + match = re.search(r"\b(yes|no)\b", text) + if match is not None: + return match.group(1) == "yes" + + if text.startswith("yes"): + return True + if text.startswith("no"): + return False + + return False + + @staticmethod + def normalize_text(text: str) -> str: + text = text.lower().strip() + text = text.replace("_", " ") + text = text.replace("-", " ") + text = re.sub(r"\s+", " ", text) + return text.strip(" .,:;!?") + + @staticmethod + def clamp_box( + box_xyxy: Sequence[float], + width: int, + height: int, + ) -> Tuple[int, int, int, int]: + x1, y1, x2, y2 = box_xyxy + x1 = int(max(0, min(width - 1, round(float(x1))))) + y1 = int(max(0, min(height - 1, round(float(y1))))) + x2 = int(max(0, min(width, round(float(x2))))) + y2 = int(max(0, min(height, round(float(y2))))) + return x1, y1, x2, y2 diff --git a/pipeline/pass3_object_detect.py b/pipeline/pass3_object_detect.py index fe283d6..14b4660 100644 --- a/pipeline/pass3_object_detect.py +++ b/pipeline/pass3_object_detect.py @@ -1,1401 +1,19 @@ -import os -os.environ["CUDA_VISIBLE_DEVICES"] = "0" -os.environ["PADDLE_PDX_DISABLE_MODEL_SOURCE_CHECK"] = "True" - -import re +import argparse import gc import sys -import json -import argparse import traceback -import faulthandler -from dataclasses import dataclass, field from pathlib import Path -from typing import Dict, List, Optional, Sequence, Tuple, Any - -faulthandler.enable() - +from datetime import datetime +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.append(str(PROJECT_ROOT)) +import torch +from models.object_detector import PrivacyReasoningEngine +import os import cv2 import numpy as np from PIL import Image - -import torch - -from transformers import AutoModelForVision2Seq, AutoProcessor -from paddleocr import PaddleOCR - - -@dataclass -class OCRTextItem: - frame_index: int - x1: int - y1: int - x2: int - y2: int - text: str - score: float - - @property - def box(self) -> Tuple[int, int, int, int]: - return (self.x1, self.y1, self.x2, self.y2) - - -@dataclass -class TextGroupCandidate: - frame_index: int - x1: int - y1: int - x2: int - y2: int - merged_text: str - avg_score: float - item_count: int - items: List[OCRTextItem] = field(default_factory=list) - - @property - def box(self) -> Tuple[int, int, int, int]: - return (self.x1, self.y1, self.x2, self.y2) - - -@dataclass -class VerifiedDetection: - frame_index: int - x1: int - y1: int - x2: int - y2: int - label: str - detector_phrase: str - detector_score: float - qwen_visible_text: str = "" - qwen_reason: str = "" - track_id: Optional[str] = None - - @property - def box(self) -> Tuple[int, int, int, int]: - return (self.x1, self.y1, self.x2, self.y2) - - -@dataclass -class TrackState: - object_id: str - label: str - start_frame: int - end_frame: int - last_frame: int - last_box: Tuple[int, int, int, int] - history_boxes: List[Tuple[int, int, int, int]] = field(default_factory=list) - visible_text_samples: List[str] = field(default_factory=list) - reason_samples: List[str] = field(default_factory=list) - hit_count: int = 1 - - def update(self, detection: VerifiedDetection) -> None: - self.end_frame = detection.frame_index - self.last_frame = detection.frame_index - self.last_box = detection.box - self.history_boxes.append(detection.box) - if detection.qwen_visible_text: - self.visible_text_samples.append(detection.qwen_visible_text) - if detection.qwen_reason: - self.reason_samples.append(detection.qwen_reason) - self.hit_count += 1 - - def representative_box(self) -> List[int]: - if not self.history_boxes: - return list(self.last_box) - - arr = np.array(self.history_boxes, dtype=np.float32) - med = np.median(arr, axis=0) - return [int(round(v)) for v in med.tolist()] - - def representative_visible_text(self) -> str: - for text in reversed(self.visible_text_samples): - if text and text.upper() != "NONE": - return text - return "" - - -class PrivacyReasoningEngine: - LABEL_DISPLAY_MAP: Dict[str, str] = { - "license plate": "번호판", - "driver's license": "운전면허증", - "credit card": "신용카드", - "receipt": "영수증", - "address document": "주소 문서", - "other private text": "기타 개인정보 텍스트", - "other": "기타", - } - - WEAK_EVIDENCE_WORDS = { - "card", - "paper", - "screen", - "document", - "printed text", - "text", - "visible text", - "license plate", - "label", - "sign", - "none", - } - - def __init__( - self, - qwen_model_name: str = "Qwen/Qwen2-VL-2B-Instruct", - device: Optional[str] = None, - crop_margin_ratio: float = 0.35, - max_new_tokens_reason: int = 48, - verbose: bool = True, - save_verified_crops: bool = True, - debug_crop_dir: str = "debug_crops", - track_iou_threshold: float = 0.30, - min_text_box_area: int = 16, - max_text_box_area_ratio: float = 0.75, - min_text_region_width: int = 3, - min_text_region_height: int = 3, - text_detector_lang: str = "korean", - min_rec_score: float = 0.30, - min_group_items: int = 1, - group_x_gap_ratio: float = 1.6, - group_y_gap_ratio: float = 1.0, - min_qwen_crop_size: int = 56, - ) -> None: - self.verbose = bool(verbose) - self.save_verified_crops = bool(save_verified_crops) - self.debug_crop_dir = Path(debug_crop_dir) - - self.device = device or ("cuda" if torch.cuda.is_available() else "cpu") - self.crop_margin_ratio = float(crop_margin_ratio) - self.max_new_tokens_reason = int(max_new_tokens_reason) - self.track_iou_threshold = float(track_iou_threshold) - - self.min_text_box_area = int(min_text_box_area) - self.max_text_box_area_ratio = float(max_text_box_area_ratio) - self.min_text_region_width = int(min_text_region_width) - self.min_text_region_height = int(min_text_region_height) - self.min_rec_score = float(min_rec_score) - self.min_group_items = int(min_group_items) - self.group_x_gap_ratio = float(group_x_gap_ratio) - self.group_y_gap_ratio = float(group_y_gap_ratio) - self.min_qwen_crop_size = int(min_qwen_crop_size) - - if self.save_verified_crops: - self.debug_crop_dir.mkdir(parents=True, exist_ok=True) - - self._next_track_index = 1 - self.active_tracks: Dict[str, TrackState] = {} - self.finished_tracks: List[TrackState] = [] - self.last_verified_detections: List[VerifiedDetection] = [] - self.last_tracks: List[TrackState] = [] - self.last_json_payload: List[Dict] = [] - - self._log("엔진 초기화 시작") - self._log(f"device = {self.device}") - self._log(f"crop_margin_ratio = {self.crop_margin_ratio}") - self._log(f"track_iou_threshold = {self.track_iou_threshold}") - self._log(f"text_detector_lang = {text_detector_lang}") - self._log(f"min_text_box_area = {self.min_text_box_area}") - self._log(f"min_text_region_width = {self.min_text_region_width}") - self._log(f"min_text_region_height = {self.min_text_region_height}") - self._log(f"min_rec_score = {self.min_rec_score}") - self._log(f"min_qwen_crop_size = {self.min_qwen_crop_size}") - - self.qwen_processor = AutoProcessor.from_pretrained( - qwen_model_name, - trust_remote_code=True, - ) - self._log("Qwen2-VL Processor 로딩 완료") - - self.qwen_model = AutoModelForVision2Seq.from_pretrained( - qwen_model_name, - trust_remote_code=True, - torch_dtype=torch.float16 if self.device == "cuda" else torch.float32, - ).to(self.device) - self.qwen_model.eval() - self._log("Qwen2-VL Model 로딩 완료") - - if hasattr(self.qwen_model, "generation_config") and self.qwen_model.generation_config is not None: - for attr in ("temperature", "top_p", "top_k"): - if hasattr(self.qwen_model.generation_config, attr): - setattr(self.qwen_model.generation_config, attr, None) - if hasattr(self.qwen_model.generation_config, "do_sample"): - self.qwen_model.generation_config.do_sample = False - - self.text_detector = PaddleOCR( - lang=text_detector_lang, - use_doc_orientation_classify=False, - use_doc_unwarping=False, - use_textline_orientation=False, - ) - self._log("PaddleOCR detector+recognizer 로딩 완료") - - # ------------------------------------------------------------------------- - # Logging - # ------------------------------------------------------------------------- - def _log(self, message: str) -> None: - if self.verbose: - print(f"[Info] {message}", flush=True) - - def _log_group_found(self, frame_index: int, count: int) -> None: - print(f"[Frame {frame_index}] Group 후보 {count}개 발견", flush=True) - - def _log_verified(self, detection: VerifiedDetection) -> None: - print( - f"[Frame {detection.frame_index}] 검증 통과: {detection.label} " - f"| box=({detection.x1}, {detection.y1}, {detection.x2}, {detection.y2}) " - f"| id={detection.track_id}", - flush=True, - ) - - # ------------------------------------------------------------------------- - # Prompt helpers - # ------------------------------------------------------------------------- - def build_privacy_reason_prompt( - self, - user_prompt: str, - visible_text: str, - ) -> str: - return ( - "You are deciding whether a text group and its surrounding object should be privacy-blurred.\n" - f"User request: {user_prompt}\n" - f"Visible text from the grouped region:\n{visible_text}\n" - "Decide based on the visible text and obvious visible context only.\n" - "If this grouped region likely belongs to a privacy-sensitive object such as a receipt, waybill(delivery label with name/address/phone number), address label, card, ID document, license plate, or other personal text-bearing object, answer yes. Road signs, subtitles, and brand logos are NOT privacy-sensitive.\n" - "Short noisy fragments such as a single digit or random short token alone are not enough unless the overall grouped region strongly suggests a sensitive object.\n" - "If any part of the grouped region likely contains privacy-sensitive information, treat the whole grouped object as blur-worthy.\n" - "You MUST return exactly three lines in this order, do NOT skip or merge any line:\n" - "Decision: yes or no\n" - "Label: receipt | waybill | credit card | driver's license | license plate | road sign | address document | other private text | other\n" - "Reason: \n" - "Example output:\n" - "Decision: yes\n" - "Label: waybill\n" - "Reason: Contains name, address and phone number typical of a delivery label.\n" - ) - - # ------------------------------------------------------------------------- - # Qwen helpers - # ------------------------------------------------------------------------- - def ask_qwen_reason( - self, - image_pil: Image.Image, - prompt_text: str, - max_new_tokens: int, - ) -> str: - messages = [ - { - "role": "user", - "content": [ - {"type": "image", "image": image_pil}, - {"type": "text", "text": prompt_text}, - ], - } - ] - - text_prompt = self.qwen_processor.apply_chat_template( - messages, - tokenize=False, - add_generation_prompt=True, - ) - - inputs = self.qwen_processor( - text=[text_prompt], - images=[image_pil], - padding=True, - return_tensors="pt", - ) - - for key, value in inputs.items(): - if hasattr(value, "to"): - inputs[key] = value.to(self.device) - - with torch.inference_mode(): - output_ids = self.qwen_model.generate( - **inputs, - max_new_tokens=max_new_tokens, - do_sample=False, - use_cache=True, - ) - - prompt_len = int(inputs["input_ids"].shape[1]) - generated_ids = output_ids[:, prompt_len:] - - answer = self.qwen_processor.batch_decode( - generated_ids, - skip_special_tokens=True, - clean_up_tokenization_spaces=True, - )[0] - - return answer.strip() - - # ------------------------------------------------------------------------- - # Public API - # ------------------------------------------------------------------------- - def process_video( - self, - video_path: str, - user_prompt: str = "내 프라이버시가 유출될 만한 것들을 가려줘.", - sample_fps: float = 1.0, - ) -> List[TrackState]: - print("[Start] process_video 시작", flush=True) - print(f"[Start] video_path = {video_path}", flush=True) - print(f"[Start] user_prompt = {user_prompt}", flush=True) - print(f"[Start] sample_fps = {sample_fps}", flush=True) - - if sample_fps <= 0: - raise ValueError("sample_fps must be > 0.") - - self.active_tracks = {} - self.finished_tracks = [] - self.last_verified_detections = [] - self.last_tracks = [] - self.last_json_payload = [] - self._next_track_index = 1 - - cap = cv2.VideoCapture(video_path) - print(f"[Start] VideoCapture isOpened = {cap.isOpened()}", flush=True) - - if not cap.isOpened(): - raise FileNotFoundError(f"Could not open video: {video_path}") - - total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) - native_fps = float(cap.get(cv2.CAP_PROP_FPS)) - width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) - height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) - - if native_fps <= 0 or not np.isfinite(native_fps): - native_fps = 30.0 - - sample_interval_frames = max(1, int(round(native_fps / sample_fps))) - max_track_gap_frames = max(sample_interval_frames * 2, sample_interval_frames + 1) - - print(f"[Start] total_frames = {total_frames}", flush=True) - print(f"[Start] native_fps = {native_fps}", flush=True) - print(f"[Start] resolution = {width}x{height}", flush=True) - print(f"[Start] sample_interval_frames = {sample_interval_frames}", flush=True) - print(f"[Start] max_track_gap_frames = {max_track_gap_frames}", flush=True) - - frame_index = 0 - sampled_count = 0 - - try: - while True: - ret, frame_bgr = cap.read() - if not ret: - break - - if frame_index % sample_interval_frames != 0: - frame_index += 1 - continue - - sampled_count += 1 - - try: - groups = self.detect_grouped_text_candidates( - frame_bgr=frame_bgr, - frame_index=frame_index, - ) - except Exception as e: - print(f"[Error] Frame {frame_index} detect/group 실패: {e}", flush=True) - traceback.print_exc() - frame_index += 1 - continue - - if len(groups) > 0: - self._log_group_found(frame_index, len(groups)) - - frame_verified: List[VerifiedDetection] = [] - - for group in groups: - try: - verified_detection = self.verify_group_reason_only( - frame_bgr=frame_bgr, - group=group, - user_prompt=user_prompt, - ) - except Exception as e: - print(f"[Error] Frame {frame_index} reason 실패: {e}", flush=True) - traceback.print_exc() - continue - - if verified_detection is not None: - frame_verified.append(verified_detection) - - self.update_tracks_for_frame( - detections=frame_verified, - frame_index=frame_index, - max_track_gap_frames=max_track_gap_frames, - ) - - for det in frame_verified: - self.last_verified_detections.append(det) - self._log_verified(det) - - frame_index += 1 - - finally: - cap.release() - - self.finalize_all_tracks() - self.last_tracks = list(self.finished_tracks) - - print(f"[Done] 샘플링된 프레임 수 = {sampled_count}", flush=True) - print(f"[Done] 총 검증 통과 수 = {len(self.last_verified_detections)}", flush=True) - print(f"[Done] 총 트랙 수 = {len(self.last_tracks)}", flush=True) - - return self.last_tracks - - # ------------------------------------------------------------------------- - # Stage 1: PaddleOCR detection + recognition + grouping - # ------------------------------------------------------------------------- - def detect_grouped_text_candidates( - self, - frame_bgr: np.ndarray, - frame_index: int, - ) -> List[TextGroupCandidate]: - height, width = frame_bgr.shape[:2] - frame_area = float(width * height) - - frame_rgb = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB) - raw_results = self.text_detector.predict(frame_rgb) - - items = self.extract_text_items_from_predict_results(raw_results, frame_index) - - filtered_items: List[OCRTextItem] = [] - for item in items: - x1, y1, x2, y2 = item.box - box_w = max(1, x2 - x1) - box_h = max(1, y2 - y1) - area = float(box_w * box_h) - - if area < self.min_text_box_area: - continue - if area > frame_area * self.max_text_box_area_ratio: - continue - if box_w < self.min_text_region_width or box_h < self.min_text_region_height: - continue - if item.score < self.min_rec_score: - continue - if self.is_missing_or_weak_visible_text(item.text): - continue - if self.visible_text_is_gibberish(item.text): - continue - - filtered_items.append(item) - - print(f"[Debug][Frame {frame_index}] filtered text items = {len(filtered_items)}", flush=True) - - if len(filtered_items) == 0: - return [] - - groups = self.group_text_items(filtered_items, frame_w=width, frame_h=height) - - candidates: List[TextGroupCandidate] = [] - for group_items in groups: - if len(group_items) < self.min_group_items: - continue - - gx1 = min(item.x1 for item in group_items) - gy1 = min(item.y1 for item in group_items) - gx2 = max(item.x2 for item in group_items) - gy2 = max(item.y2 for item in group_items) - - merged_text = self.build_group_text(group_items) - avg_score = float(sum(item.score for item in group_items) / max(1, len(group_items))) - - if self.group_text_is_too_weak(merged_text, len(group_items)): - continue - - candidates.append( - TextGroupCandidate( - frame_index=frame_index, - x1=gx1, - y1=gy1, - x2=gx2, - y2=gy2, - merged_text=merged_text, - avg_score=avg_score, - item_count=len(group_items), - items=group_items, - ) - ) - - print(f"[Debug][Frame {frame_index}] group candidates = {len(candidates)}", flush=True) - for i, cand in enumerate(candidates[:10]): - print( - f"[Debug][Frame {frame_index}] group[{i}] items={cand.item_count} score={cand.avg_score:.3f} text={repr(cand.merged_text)} box={cand.box}", - flush=True, - ) - - return candidates - - def extract_text_items_from_predict_results( - self, - raw_results: Any, - frame_index: int, - ) -> List[OCRTextItem]: - items: List[OCRTextItem] = [] - - if raw_results is None: - return items - - try: - result_list = list(raw_results) - except TypeError: - result_list = [raw_results] - - if frame_index == 0: - print(f"[Debug][Frame {frame_index}] predict result count = {len(result_list)}", flush=True) - - for item in result_list: - data = None - - if hasattr(item, "res"): - data = item.res - elif isinstance(item, dict): - data = item - else: - try: - data = dict(item) - except Exception: - data = None - - if not isinstance(data, dict): - continue - - if frame_index == 0: - print(f"[Debug][Frame {frame_index}] result keys = {list(data.keys())}", flush=True) - - rec_texts = data.get("rec_texts", None) - rec_scores = data.get("rec_scores", None) - rec_boxes = data.get("rec_boxes", None) - rec_polys = data.get("rec_polys", None) - dt_polys = data.get("dt_polys", None) - - if rec_texts is None: - continue - - rec_texts = list(rec_texts) - rec_scores = list(rec_scores) if rec_scores is not None else [1.0] * len(rec_texts) - - box_source = None - if rec_boxes is not None: - box_source = rec_boxes - elif rec_polys is not None: - box_source = rec_polys - elif dt_polys is not None: - box_source = dt_polys - - if box_source is None: - continue - - try: - box_arr = np.asarray(box_source, dtype=np.float32) - except Exception: - continue - - if box_arr.ndim == 2 and box_arr.shape[-1] == 4: - for i, text in enumerate(rec_texts): - if i >= len(box_arr): - break - x1, y1, x2, y2 = box_arr[i].tolist() - items.append( - OCRTextItem( - frame_index=frame_index, - x1=int(round(x1)), - y1=int(round(y1)), - x2=int(round(x2)), - y2=int(round(y2)), - text=str(text).strip(), - score=float(rec_scores[i]) if i < len(rec_scores) else 1.0, - ) - ) - - elif box_arr.ndim == 3 and box_arr.shape[-1] == 2: - for i, text in enumerate(rec_texts): - if i >= len(box_arr): - break - poly = box_arr[i] - xs = poly[:, 0] - ys = poly[:, 1] - items.append( - OCRTextItem( - frame_index=frame_index, - x1=int(round(float(xs.min()))), - y1=int(round(float(ys.min()))), - x2=int(round(float(xs.max()))), - y2=int(round(float(ys.max()))), - text=str(text).strip(), - score=float(rec_scores[i]) if i < len(rec_scores) else 1.0, - ) - ) - - return items - - def group_text_items(self, items: List[OCRTextItem], frame_w: int = 9999, frame_h: int = 9999) -> List[List[OCRTextItem]]: - if len(items) == 0: - return [] - - items = sorted(items, key=lambda it: (it.y1, it.x1)) - - groups: List[List[OCRTextItem]] = [] - used = [False] * len(items) - - for i in range(len(items)): - if used[i]: - continue - - current_group = [items[i]] - used[i] = True - - changed = True - while changed: - changed = False - gx1 = min(it.x1 for it in current_group) - gy1 = min(it.y1 for it in current_group) - gx2 = max(it.x2 for it in current_group) - gy2 = max(it.y2 for it in current_group) - gh = max(1, gy2 - gy1) - gw = max(1, gx2 - gx1) - - for j in range(len(items)): - if used[j]: - continue - - it = items[j] - iou = self.compute_iou((gx1, gy1, gx2, gy2), it.box) - - x_gap = self.box_horizontal_gap((gx1, gy1, gx2, gy2), it.box) - y_gap = self.box_vertical_gap((gx1, gy1, gx2, gy2), it.box) - - close_x = x_gap <= int(round(gh * self.group_x_gap_ratio)) - close_y = y_gap <= int(round(gh * self.group_y_gap_ratio)) - overlaps = iou > 0.0 - - within_band = ( - it.x1 <= gx2 + int(round(gw * 0.3)) - and it.x2 >= gx1 - int(round(gw * 0.3)) - and it.y1 <= gy2 + int(round(gh * 0.8)) - and it.y2 >= gy1 - int(round(gh * 0.8)) - ) - max_group_area = frame_w * frame_h * 0.15 - if (gx2 - gx1) * (gy2 - gy1) > max_group_area: - break - if (close_x and close_y) or overlaps or within_band: - current_group.append(it) - used[j] = True - changed = True - - groups.append(sorted(current_group, key=lambda it: (it.y1, it.x1))) - - return groups - - def build_group_text(self, items: List[OCRTextItem]) -> str: - if len(items) == 0: - return "" - - lines: List[List[OCRTextItem]] = [] - for item in sorted(items, key=lambda it: (it.y1, it.x1)): - placed = False - for line in lines: - ly1 = min(x.y1 for x in line) - ly2 = max(x.y2 for x in line) - lh = max(1, ly2 - ly1) - if abs(item.y1 - ly1) <= int(round(lh * 0.6)): - line.append(item) - placed = True - break - if not placed: - lines.append([item]) - - out_lines: List[str] = [] - for line in lines: - line_sorted = sorted(line, key=lambda it: it.x1) - line_text = " ".join(it.text.strip() for it in line_sorted if it.text.strip()) - if line_text: - out_lines.append(line_text) - - return "\n".join(out_lines) - - def group_text_is_too_weak(self, merged_text: str, item_count: int) -> bool: - text = merged_text.strip() - if text == "": - return True - - norm = self.normalize_text(text) - if norm in ("none", "unknown", "unreadable"): - return True - - has_digits = re.search(r"\d", text) is not None - has_hangul = re.search(r"[가-힣]", text) is not None - has_alpha_word = re.search(r"[A-Za-z]{3,}", text) is not None - has_cjk = re.search(r"[\u4e00-\u9fff]", text) is not None - - if item_count == 1: - compact = re.sub(r"\s+", "", text) - if len(compact) <= 2 and not has_hangul and not has_alpha_word: - return True - - if not (has_digits or has_hangul or has_alpha_word or has_cjk): - return True - - return False - - # ------------------------------------------------------------------------- - # Stage 2: Qwen reasoning only on grouped object - # ------------------------------------------------------------------------- - def verify_group_reason_only( - self, - frame_bgr: np.ndarray, - group: TextGroupCandidate, - user_prompt: str, - ) -> Optional[VerifiedDetection]: - frame_index = group.frame_index - visible_text = group.merged_text.strip() - - if self.group_text_is_too_weak(visible_text, group.item_count): - return None - - crop_rgb = None - try: - crop_rgb = self.crop_with_margin( - frame_bgr=frame_bgr, - x1=group.x1, - y1=group.y1, - x2=group.x2, - y2=group.y2, - margin_ratio=self.crop_margin_ratio, - ) - - if crop_rgb is None: - return None - - crop_rgb = self.ensure_min_crop_size(crop_rgb, self.min_qwen_crop_size) - image_pil = Image.fromarray(crop_rgb) - - print( - f"[Group-REC][Frame {frame_index}] items={group.item_count} score={group.avg_score:.3f} text={repr(visible_text)} box={group.box}", - flush=True, - ) - - reason_prompt = self.build_privacy_reason_prompt( - user_prompt=user_prompt, - visible_text=visible_text, - ) - raw_reason_answer = self.ask_qwen_reason( - image_pil=image_pil, - prompt_text=reason_prompt, - max_new_tokens=self.max_new_tokens_reason, - ) - - decision, normalized_label, reason = self.parse_qwen_reason_answer( - answer=raw_reason_answer, - fallback_label="other private text", - ) - - print( - f"[Qwen-Reason][Frame {frame_index}] decision={decision} label={normalized_label} text={repr(visible_text)} reason={reason} raw_answer={repr(raw_reason_answer)}", - flush=True, - ) - - if not decision: - return None - - det = VerifiedDetection( - frame_index=frame_index, - x1=group.x1, - y1=group.y1, - x2=group.x2, - y2=group.y2, - label=normalized_label, - detector_phrase="grouped text object", - detector_score=float(group.avg_score), - qwen_visible_text=visible_text, - qwen_reason=reason, - ) - - if self.save_verified_crops: - self.save_debug_crop(crop_rgb=crop_rgb, detection=det) - - return det - - except Exception as e: - print(f"[Error] Frame {frame_index} group reason 예외: {e}", flush=True) - traceback.print_exc() - return None - - finally: - if torch.cuda.is_available(): - torch.cuda.empty_cache() - gc.collect() - - def parse_qwen_reason_answer( - self, - answer: str, - fallback_label: str, - ) -> Tuple[bool, str, str]: - text = answer.strip() - - decision_match = re.search(r"decision\s*:\s*(yes|no)", text, flags=re.IGNORECASE) - label_match = re.search( - r"label\s*:\s*(receipt|credit card|driver'?s license|license plate|address document|other private text|other)", - text, - flags=re.IGNORECASE, - ) - reason_match = re.search(r"reason\s*:\s*(.+)", text, flags=re.IGNORECASE | re.DOTALL) - - decision = False - if decision_match is not None: - decision = decision_match.group(1).lower() == "yes" - else: - decision = self.parse_yes_no(text) - - raw_label = fallback_label - if label_match is not None: - raw_label = label_match.group(1).lower().strip() - - reason = "" - if reason_match is not None: - reason = reason_match.group(1).strip().splitlines()[0].strip() - - normalized_label = self.normalize_label_name(raw_label) - return decision, normalized_label, reason - - def is_missing_or_weak_visible_text(self, visible_text: str) -> bool: - if visible_text is None: - return True - - vt = visible_text.strip() - if vt == "": - return True - - vt_norm = self.normalize_text(vt) - if vt_norm in ("none", "n a", "na", "unknown", "not visible", "unreadable"): - return True - - if vt_norm in self.WEAK_EVIDENCE_WORDS: - return True - - has_digits = re.search(r"\d", vt) is not None - has_hangul = re.search(r"[가-힣]", vt) is not None - has_alpha_word = re.search(r"[A-Za-z]{2,}", vt) is not None - has_cjk = re.search(r"[\u4e00-\u9fff]", vt) is not None - - if not (has_digits or has_hangul or has_alpha_word or has_cjk): - return True - - return False - - def visible_text_is_gibberish(self, visible_text: str) -> bool: - vt = visible_text.strip() - - if len(vt) >= 8 and len(set(vt)) == 1: - return True - - if re.fullmatch(r"(\d)\1{7,}", vt): - return True - if re.fullmatch(r"([A-Za-z])\1{7,}", vt): - return True - - digits = re.sub(r"\D", "", vt) - if len(digits) >= 12 and len(set(digits)) <= 2: - return True - - cleaned = re.sub(r"[0-9A-Za-z가-힣\u4e00-\u9fff\s:/\-.(),#]", "", vt) - if len(cleaned) >= 5 and len(vt) >= 10: - return True - - return False - - def normalize_label_name(self, label: str) -> str: - n = self.normalize_text(label) - if n in ("driver s license", "driver's license", "driver license", "driving license", "id card", "identity card"): - return "driver's license" - if n in ("credit card", "debit card", "bank card", "payment card", "card"): - return "credit card" - if n in ("receipt", "bill", "paper receipt", "purchase receipt"): - return "receipt" - if n in ("license plate", "number plate", "plate", "car plate", "vehicle plate"): - return "license plate" - if n in ("address document",): - return "address document" - if n in ("waybill", "delivery label", "shipping label", "invoice"): - return "waybill" - if n in ("road sign", "street sign", "traffic sign", "sign"): - return "road sign" - if n in ("other private text", "other private", "private text"): - return "other private text" - return "other" - - def save_debug_crop(self, crop_rgb: np.ndarray, detection: VerifiedDetection) -> None: - label = str(detection.label).replace(" ", "_").replace("/", "_").replace("'", "") - filename = ( - f"frame_{detection.frame_index}_{label}_" - f"{detection.x1}_{detection.y1}_{detection.x2}_{detection.y2}.jpg" - ) - save_path = self.debug_crop_dir / filename - Image.fromarray(crop_rgb).save(save_path) - - # ------------------------------------------------------------------------- - # Stage 3: Simple ID-based tracking - # ------------------------------------------------------------------------- - def update_tracks_for_frame( - self, - detections: List[VerifiedDetection], - frame_index: int, - max_track_gap_frames: int, - ) -> None: - expired_ids = [ - track_id - for track_id, track in self.active_tracks.items() - if frame_index - track.last_frame > max_track_gap_frames - ] - for track_id in expired_ids: - self.finished_tracks.append(self.active_tracks.pop(track_id)) - - if len(detections) == 0: - return - - track_ids = list(self.active_tracks.keys()) - candidate_pairs: List[Tuple[float, int, str]] = [] - - for det_idx, det in enumerate(detections): - for track_id in track_ids: - track = self.active_tracks[track_id] - # 라벨이 달라도 위치(IoU)가 겹치면 같은 객체로 인정 - # if track.label != det.label: - # continue - - iou = self.compute_iou(track.last_box, det.box) - if iou >= self.track_iou_threshold: - candidate_pairs.append((iou, det_idx, track_id)) - - candidate_pairs.sort(key=lambda x: x[0], reverse=True) - - matched_det_indices = set() - matched_track_ids = set() - - for _, det_idx, track_id in candidate_pairs: - if det_idx in matched_det_indices or track_id in matched_track_ids: - continue - - det = detections[det_idx] - track = self.active_tracks[track_id] - track.update(det) - det.track_id = track.object_id - - matched_det_indices.add(det_idx) - matched_track_ids.add(track_id) - - for det_idx, det in enumerate(detections): - if det_idx in matched_det_indices: - continue - - new_track = self.create_track(det) - self.active_tracks[new_track.object_id] = new_track - det.track_id = new_track.object_id - - def create_track(self, detection: VerifiedDetection) -> TrackState: - object_id = f"object_{self._next_track_index:03d}" - self._next_track_index += 1 - - return TrackState( - object_id=object_id, - label=detection.label, - start_frame=detection.frame_index, - end_frame=detection.frame_index, - last_frame=detection.frame_index, - last_box=detection.box, - history_boxes=[detection.box], - visible_text_samples=[detection.qwen_visible_text] if detection.qwen_visible_text else [], - reason_samples=[detection.qwen_reason] if detection.qwen_reason else [], - hit_count=1, - ) - - def finalize_all_tracks(self) -> None: - for track_id in list(self.active_tracks.keys()): - self.finished_tracks.append(self.active_tracks.pop(track_id)) - - @staticmethod - def compute_iou( - box_a: Tuple[int, int, int, int], - box_b: Tuple[int, int, int, int], - ) -> float: - ax1, ay1, ax2, ay2 = box_a - bx1, by1, bx2, by2 = box_b - - inter_x1 = max(ax1, bx1) - inter_y1 = max(ay1, by1) - inter_x2 = min(ax2, bx2) - inter_y2 = min(ay2, by2) - - inter_w = max(0, inter_x2 - inter_x1) - inter_h = max(0, inter_y2 - inter_y1) - inter_area = inter_w * inter_h - - area_a = max(0, ax2 - ax1) * max(0, ay2 - ay1) - area_b = max(0, bx2 - bx1) * max(0, by2 - by1) - - union_area = area_a + area_b - inter_area - if union_area <= 0: - return 0.0 - - return float(inter_area / union_area) - - @staticmethod - def box_horizontal_gap( - box_a: Tuple[int, int, int, int], - box_b: Tuple[int, int, int, int], - ) -> int: - ax1, _, ax2, _ = box_a - bx1, _, bx2, _ = box_b - if ax2 < bx1: - return bx1 - ax2 - if bx2 < ax1: - return ax1 - bx2 - return 0 - - @staticmethod - def box_vertical_gap( - box_a: Tuple[int, int, int, int], - box_b: Tuple[int, int, int, int], - ) -> int: - _, ay1, _, ay2 = box_a - _, by1, _, by2 = box_b - if ay2 < by1: - return by1 - ay2 - if by2 < ay1: - return ay1 - by2 - return 0 - - # ------------------------------------------------------------------------- - # Export / debug outputs - # ------------------------------------------------------------------------- - def export_to_json( - self, - output_json_path: str, - tracks: Optional[List[TrackState]] = None, - ) -> List[Dict]: - if tracks is None: - tracks = list(self.last_tracks) # 복사본으로 - # 중복 제거 전 디버그 - print(f"[Debug] 중복 제거 전 트랙 수: {len(tracks)}", flush=True) - for t in tracks: - print(f"[RawTrack] {t.object_id} {t.start_frame} ~ {t.end_frame} {t.representative_box()}", flush=True) - # raw 트랙 저장 - import json as _json - raw_payload = [] - for t in tracks: - raw_payload.append({ - "id": t.object_id, - "start_frame": int(t.start_frame), - "end_frame": int(t.end_frame), - "box": t.representative_box(), - "visible_text": t.representative_visible_text(), - }) - with open("raw_tracks.json", "w", encoding="utf-8") as f: - _json.dump(raw_payload, f, ensure_ascii=False, indent=2) - print("[Debug] raw_tracks.json 저장 완료", flush=True) - - # IoU 기반 중복 트랙 제거 - - def compute_iou_box(a, b): - ax1, ay1, ax2, ay2 = a - bx1, by1, bx2, by2 = b - ix1, iy1 = max(ax1, bx1), max(ay1, by1) - ix2, iy2 = min(ax2, bx2), min(ay2, by2) - inter = max(0, ix2-ix1) * max(0, iy2-iy1) - if inter == 0: - return 0.0 - area_a = (ax2-ax1) * (ay2-ay1) - area_b = (bx2-bx1) * (by2-by1) - return inter / (area_a + area_b - inter) - - merged_tracks = [] - used = set() - for i, t1 in enumerate(tracks): - if i in used: - continue - box1 = t1.representative_box() - # 화면 너무 큰 박스 제거 - bw = box1[2] - box1[0] - bh = box1[3] - box1[1] - if bw * bh > 300000: - used.add(i) - continue - best_start = t1.start_frame - best_end = t1.end_frame - for j, t2 in enumerate(tracks): - if i == j or j in used: - continue - box2 = t2.representative_box() - if compute_iou_box(box1, box2) >= 0.5: - best_start = min(best_start, t2.start_frame) - best_end = max(best_end, t2.end_frame) - used.add(j) - t1.start_frame = best_start - t1.end_frame = best_end - merged_tracks.append(t1) - used.add(i) - tracks = merged_tracks - - print(f"[Debug] 중복 제거 후 트랙 수: {len(tracks)}", flush=True) - - payload: List[Dict] = [] - for track in tracks: - if track.start_frame == track.end_frame: - track.end_frame = track.start_frame + 10 - payload.append( - { - "id": track.object_id, - "type": "object", - "label": self.LABEL_DISPLAY_MAP.get(track.label, track.label), - "start_frame": int(track.start_frame), - "end_frame": int(track.end_frame), - "box": track.representative_box(), - "visible_text": track.representative_visible_text(), - } - ) - - out_path = Path(output_json_path) - out_path.parent.mkdir(parents=True, exist_ok=True) - - with open(out_path, "w", encoding="utf-8") as f: - json.dump(payload, f, ensure_ascii=False, indent=2) - - self.last_json_payload = payload - - print(f"[Export] JSON 저장 완료: {output_json_path}", flush=True) - return payload - - def group_detections_by_frame( - self, - detections: Optional[List[VerifiedDetection]] = None, - ) -> Dict[int, List[VerifiedDetection]]: - if detections is None: - detections = self.last_verified_detections - - grouped: Dict[int, List[VerifiedDetection]] = {} - for det in detections: - grouped.setdefault(det.frame_index, []).append(det) - return grouped - - def write_debug_video_fullfps( - self, - input_video_path: str, - output_video_path: str, - detections: Optional[List[VerifiedDetection]] = None, - ) -> None: - cap = cv2.VideoCapture(input_video_path) - if not cap.isOpened(): - raise FileNotFoundError(f"Could not open input video: {input_video_path}") - - fps = float(cap.get(cv2.CAP_PROP_FPS)) - width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) - height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) - - if fps <= 0 or not np.isfinite(fps): - fps = 24.0 - - fourcc = cv2.VideoWriter_fourcc(*"mp4v") - writer = cv2.VideoWriter(output_video_path, fourcc, fps, (width, height)) - - grouped = self.group_detections_by_frame(detections) - - frame_index = 0 - try: - while True: - ret, frame_bgr = cap.read() - if not ret: - break - - frame_dets = grouped.get(frame_index, []) - for det in frame_dets: - self.draw_bbox( - frame_bgr, - det.x1, - det.y1, - det.x2, - det.y2, - f"{det.track_id}:{self.LABEL_DISPLAY_MAP.get(det.label, det.label)}", - ) - - writer.write(frame_bgr) - frame_index += 1 - - finally: - cap.release() - writer.release() - - print(f"[Debug] full-fps bbox 영상 저장 완료: {output_video_path}", flush=True) - - def write_debug_video_sampled( - self, - input_video_path: str, - output_video_path: str, - sample_fps: float, - detections: Optional[List[VerifiedDetection]] = None, - ) -> None: - cap = cv2.VideoCapture(input_video_path) - if not cap.isOpened(): - raise FileNotFoundError(f"Could not open input video: {input_video_path}") - - native_fps = float(cap.get(cv2.CAP_PROP_FPS)) - width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) - height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) - - if native_fps <= 0 or not np.isfinite(native_fps): - native_fps = 24.0 - - sample_interval_frames = max(1, int(round(native_fps / sample_fps))) - - fourcc = cv2.VideoWriter_fourcc(*"mp4v") - writer = cv2.VideoWriter(output_video_path, fourcc, sample_fps, (width, height)) - - grouped = self.group_detections_by_frame(detections) - - frame_index = 0 - try: - while True: - ret, frame_bgr = cap.read() - if not ret: - break - - if frame_index % sample_interval_frames != 0: - frame_index += 1 - continue - - frame_dets = grouped.get(frame_index, []) - for det in frame_dets: - self.draw_bbox( - frame_bgr, - det.x1, - det.y1, - det.x2, - det.y2, - f"{det.track_id}:{self.LABEL_DISPLAY_MAP.get(det.label, det.label)}", - ) - - writer.write(frame_bgr) - frame_index += 1 - - finally: - cap.release() - writer.release() - - print(f"[Debug] sampled bbox 영상 저장 완료: {output_video_path}", flush=True) - - @staticmethod - def draw_bbox(frame_bgr: np.ndarray, x1: int, y1: int, x2: int, y2: int, label: str) -> None: - x1, y1, x2, y2 = int(x1), int(y1), int(x2), int(y2) - cv2.rectangle(frame_bgr, (x1, y1), (x2, y2), (0, 255, 0), 2) - - # PIL로 한글 텍스트 그리기 - from PIL import ImageFont, ImageDraw, Image - img_pil = Image.fromarray(cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB)) - draw = ImageDraw.Draw(img_pil) - try: - font = ImageFont.truetype("/usr/share/fonts/truetype/nanum/NanumGothic.ttf", 20) - except: - font = ImageFont.load_default() - text_y = max(5, y1 - 25) - draw.text((x1, text_y), label, font=font, fill=(0, 255, 0)) - frame_bgr[:] = cv2.cvtColor(np.array(img_pil), cv2.COLOR_RGB2BGR) - - # ------------------------------------------------------------------------- - # Utilities - # ------------------------------------------------------------------------- - def crop_with_margin( - self, - frame_bgr: np.ndarray, - x1: int, - y1: int, - x2: int, - y2: int, - margin_ratio: float = 0.35, - ) -> Optional[np.ndarray]: - height, width = frame_bgr.shape[:2] - - box_w = max(1, x2 - x1) - box_h = max(1, y2 - y1) - - margin_x = int(round(box_w * margin_ratio)) - margin_y = int(round(box_h * margin_ratio)) - - crop_x1 = max(0, x1 - margin_x) - crop_y1 = max(0, y1 - margin_y) - crop_x2 = min(width, x2 + margin_x) - crop_y2 = min(height, y2 + margin_y) - - if crop_x2 <= crop_x1 or crop_y2 <= crop_y1: - return None - - crop_bgr = frame_bgr[crop_y1:crop_y2, crop_x1:crop_x2] - if crop_bgr.size == 0: - return None - - crop_rgb = cv2.cvtColor(crop_bgr, cv2.COLOR_BGR2RGB) - return crop_rgb - - def ensure_min_crop_size(self, crop_rgb: np.ndarray, min_size: int) -> np.ndarray: - h, w = crop_rgb.shape[:2] - new_h = max(h, min_size) - new_w = max(w, min_size) - - if new_h == h and new_w == w: - return crop_rgb - - pad_top = (new_h - h) // 2 - pad_bottom = new_h - h - pad_top - pad_left = (new_w - w) // 2 - pad_right = new_w - w - pad_left - - return cv2.copyMakeBorder( - crop_rgb, - pad_top, - pad_bottom, - pad_left, - pad_right, - borderType=cv2.BORDER_REPLICATE, - ) - - @staticmethod - def parse_yes_no(answer: str) -> bool: - text = answer.lower().strip() - - match = re.search(r"\b(yes|no)\b", text) - if match is not None: - return match.group(1) == "yes" - - if text.startswith("yes"): - return True - if text.startswith("no"): - return False - - return False - - @staticmethod - def normalize_text(text: str) -> str: - text = text.lower().strip() - text = text.replace("_", " ") - text = text.replace("-", " ") - text = re.sub(r"\s+", " ", text) - return text.strip(" .,:;!?") - - @staticmethod - def clamp_box( - box_xyxy: Sequence[float], - width: int, - height: int, - ) -> Tuple[int, int, int, int]: - x1, y1, x2, y2 = box_xyxy - x1 = int(max(0, min(width - 1, round(float(x1))))) - y1 = int(max(0, min(height - 1, round(float(y1))))) - x2 = int(max(0, min(width, round(float(x2))))) - y2 = int(max(0, min(height, round(float(y2))))) - return x1, y1, x2, y2 - +import config def parse_args(project_root: Path) -> argparse.Namespace: parser = argparse.ArgumentParser(description="PrivacyReasoningEngine grouped-object video runner") @@ -1431,6 +49,51 @@ def parse_args(project_root: Path) -> argparse.Namespace: ) return parser.parse_args() +def run_pass3( + video_path, + user_prompt, + sample_fps, + output_dir, +): + output_dir = Path(output_dir) + output_dir.mkdir(parents=True, exist_ok=True) + + debug_crop_dir = output_dir / "debug_crops" + json_output_path = output_dir / "object_db.json" + + engine = PrivacyReasoningEngine( + qwen_model_name=config.OBJECT_QWEN_MODEL_NAME, + device=config.DEVICE, + crop_margin_ratio=config.OBJECT_CROP_MARGIN_RATIO, + max_new_tokens_reason=config.OBJECT_MAX_NEW_TOKENS_REASON, + verbose=True, + save_verified_crops=True, + debug_crop_dir=str(debug_crop_dir), + track_iou_threshold=config.OBJECT_TRACK_IOU_THRESHOLD, + min_text_box_area=config.OBJECT_MIN_TEXT_BOX_AREA, + max_text_box_area_ratio=config.OBJECT_MAX_TEXT_BOX_AREA_RATIO, + min_text_region_width=config.OBJECT_MIN_TEXT_REGION_WIDTH, + min_text_region_height=config.OBJECT_MIN_TEXT_REGION_HEIGHT, + text_detector_lang=config.OBJECT_TEXT_DETECTOR_LANG, + min_rec_score=config.OBJECT_MIN_REC_SCORE, + min_group_items=config.OBJECT_MIN_GROUP_ITEMS, + group_x_gap_ratio=config.OBJECT_GROUP_X_GAP_RATIO, + group_y_gap_ratio=config.OBJECT_GROUP_Y_GAP_RATIO, + min_qwen_crop_size=config.OBJECT_MIN_QWEN_CROP_SIZE, + ) + + tracks = engine.process_video( + video_path=str(video_path), + user_prompt=user_prompt, + sample_fps=sample_fps, + ) + + payload = engine.export_to_json( + output_json_path=str(json_output_path), + tracks=tracks, + ) + + return payload if __name__ == "__main__": print("[Main] 프로그램 시작", flush=True) From b56e2ed33bca7714877be8a049db6b653d926b87 Mon Sep 17 00:00:00 2001 From: yeyeong5016 <20221373@sungshin.ac.kr> Date: Thu, 28 May 2026 22:19:13 +0900 Subject: [PATCH 2/2] =?UTF-8?q?=ED=8C=8C=EC=9D=BC=20=EC=88=98=EC=A0=95?= =?UTF-8?q?=ED=95=9C=EA=B1=B0=20=EC=A0=80=EC=9E=A5^^:?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pipeline/pass3_object_detect.py | 1445 +++++++++++++++++++++++++++++-- 1 file changed, 1391 insertions(+), 54 deletions(-) diff --git a/pipeline/pass3_object_detect.py b/pipeline/pass3_object_detect.py index 14b4660..fe283d6 100644 --- a/pipeline/pass3_object_detect.py +++ b/pipeline/pass3_object_detect.py @@ -1,19 +1,1401 @@ -import argparse +import os +os.environ["CUDA_VISIBLE_DEVICES"] = "0" +os.environ["PADDLE_PDX_DISABLE_MODEL_SOURCE_CHECK"] = "True" + +import re import gc import sys +import json +import argparse import traceback +import faulthandler +from dataclasses import dataclass, field from pathlib import Path -from datetime import datetime -PROJECT_ROOT = Path(__file__).resolve().parents[1] -if str(PROJECT_ROOT) not in sys.path: - sys.path.append(str(PROJECT_ROOT)) -import torch -from models.object_detector import PrivacyReasoningEngine -import os +from typing import Dict, List, Optional, Sequence, Tuple, Any + +faulthandler.enable() + import cv2 import numpy as np from PIL import Image -import config + +import torch + +from transformers import AutoModelForVision2Seq, AutoProcessor +from paddleocr import PaddleOCR + + +@dataclass +class OCRTextItem: + frame_index: int + x1: int + y1: int + x2: int + y2: int + text: str + score: float + + @property + def box(self) -> Tuple[int, int, int, int]: + return (self.x1, self.y1, self.x2, self.y2) + + +@dataclass +class TextGroupCandidate: + frame_index: int + x1: int + y1: int + x2: int + y2: int + merged_text: str + avg_score: float + item_count: int + items: List[OCRTextItem] = field(default_factory=list) + + @property + def box(self) -> Tuple[int, int, int, int]: + return (self.x1, self.y1, self.x2, self.y2) + + +@dataclass +class VerifiedDetection: + frame_index: int + x1: int + y1: int + x2: int + y2: int + label: str + detector_phrase: str + detector_score: float + qwen_visible_text: str = "" + qwen_reason: str = "" + track_id: Optional[str] = None + + @property + def box(self) -> Tuple[int, int, int, int]: + return (self.x1, self.y1, self.x2, self.y2) + + +@dataclass +class TrackState: + object_id: str + label: str + start_frame: int + end_frame: int + last_frame: int + last_box: Tuple[int, int, int, int] + history_boxes: List[Tuple[int, int, int, int]] = field(default_factory=list) + visible_text_samples: List[str] = field(default_factory=list) + reason_samples: List[str] = field(default_factory=list) + hit_count: int = 1 + + def update(self, detection: VerifiedDetection) -> None: + self.end_frame = detection.frame_index + self.last_frame = detection.frame_index + self.last_box = detection.box + self.history_boxes.append(detection.box) + if detection.qwen_visible_text: + self.visible_text_samples.append(detection.qwen_visible_text) + if detection.qwen_reason: + self.reason_samples.append(detection.qwen_reason) + self.hit_count += 1 + + def representative_box(self) -> List[int]: + if not self.history_boxes: + return list(self.last_box) + + arr = np.array(self.history_boxes, dtype=np.float32) + med = np.median(arr, axis=0) + return [int(round(v)) for v in med.tolist()] + + def representative_visible_text(self) -> str: + for text in reversed(self.visible_text_samples): + if text and text.upper() != "NONE": + return text + return "" + + +class PrivacyReasoningEngine: + LABEL_DISPLAY_MAP: Dict[str, str] = { + "license plate": "번호판", + "driver's license": "운전면허증", + "credit card": "신용카드", + "receipt": "영수증", + "address document": "주소 문서", + "other private text": "기타 개인정보 텍스트", + "other": "기타", + } + + WEAK_EVIDENCE_WORDS = { + "card", + "paper", + "screen", + "document", + "printed text", + "text", + "visible text", + "license plate", + "label", + "sign", + "none", + } + + def __init__( + self, + qwen_model_name: str = "Qwen/Qwen2-VL-2B-Instruct", + device: Optional[str] = None, + crop_margin_ratio: float = 0.35, + max_new_tokens_reason: int = 48, + verbose: bool = True, + save_verified_crops: bool = True, + debug_crop_dir: str = "debug_crops", + track_iou_threshold: float = 0.30, + min_text_box_area: int = 16, + max_text_box_area_ratio: float = 0.75, + min_text_region_width: int = 3, + min_text_region_height: int = 3, + text_detector_lang: str = "korean", + min_rec_score: float = 0.30, + min_group_items: int = 1, + group_x_gap_ratio: float = 1.6, + group_y_gap_ratio: float = 1.0, + min_qwen_crop_size: int = 56, + ) -> None: + self.verbose = bool(verbose) + self.save_verified_crops = bool(save_verified_crops) + self.debug_crop_dir = Path(debug_crop_dir) + + self.device = device or ("cuda" if torch.cuda.is_available() else "cpu") + self.crop_margin_ratio = float(crop_margin_ratio) + self.max_new_tokens_reason = int(max_new_tokens_reason) + self.track_iou_threshold = float(track_iou_threshold) + + self.min_text_box_area = int(min_text_box_area) + self.max_text_box_area_ratio = float(max_text_box_area_ratio) + self.min_text_region_width = int(min_text_region_width) + self.min_text_region_height = int(min_text_region_height) + self.min_rec_score = float(min_rec_score) + self.min_group_items = int(min_group_items) + self.group_x_gap_ratio = float(group_x_gap_ratio) + self.group_y_gap_ratio = float(group_y_gap_ratio) + self.min_qwen_crop_size = int(min_qwen_crop_size) + + if self.save_verified_crops: + self.debug_crop_dir.mkdir(parents=True, exist_ok=True) + + self._next_track_index = 1 + self.active_tracks: Dict[str, TrackState] = {} + self.finished_tracks: List[TrackState] = [] + self.last_verified_detections: List[VerifiedDetection] = [] + self.last_tracks: List[TrackState] = [] + self.last_json_payload: List[Dict] = [] + + self._log("엔진 초기화 시작") + self._log(f"device = {self.device}") + self._log(f"crop_margin_ratio = {self.crop_margin_ratio}") + self._log(f"track_iou_threshold = {self.track_iou_threshold}") + self._log(f"text_detector_lang = {text_detector_lang}") + self._log(f"min_text_box_area = {self.min_text_box_area}") + self._log(f"min_text_region_width = {self.min_text_region_width}") + self._log(f"min_text_region_height = {self.min_text_region_height}") + self._log(f"min_rec_score = {self.min_rec_score}") + self._log(f"min_qwen_crop_size = {self.min_qwen_crop_size}") + + self.qwen_processor = AutoProcessor.from_pretrained( + qwen_model_name, + trust_remote_code=True, + ) + self._log("Qwen2-VL Processor 로딩 완료") + + self.qwen_model = AutoModelForVision2Seq.from_pretrained( + qwen_model_name, + trust_remote_code=True, + torch_dtype=torch.float16 if self.device == "cuda" else torch.float32, + ).to(self.device) + self.qwen_model.eval() + self._log("Qwen2-VL Model 로딩 완료") + + if hasattr(self.qwen_model, "generation_config") and self.qwen_model.generation_config is not None: + for attr in ("temperature", "top_p", "top_k"): + if hasattr(self.qwen_model.generation_config, attr): + setattr(self.qwen_model.generation_config, attr, None) + if hasattr(self.qwen_model.generation_config, "do_sample"): + self.qwen_model.generation_config.do_sample = False + + self.text_detector = PaddleOCR( + lang=text_detector_lang, + use_doc_orientation_classify=False, + use_doc_unwarping=False, + use_textline_orientation=False, + ) + self._log("PaddleOCR detector+recognizer 로딩 완료") + + # ------------------------------------------------------------------------- + # Logging + # ------------------------------------------------------------------------- + def _log(self, message: str) -> None: + if self.verbose: + print(f"[Info] {message}", flush=True) + + def _log_group_found(self, frame_index: int, count: int) -> None: + print(f"[Frame {frame_index}] Group 후보 {count}개 발견", flush=True) + + def _log_verified(self, detection: VerifiedDetection) -> None: + print( + f"[Frame {detection.frame_index}] 검증 통과: {detection.label} " + f"| box=({detection.x1}, {detection.y1}, {detection.x2}, {detection.y2}) " + f"| id={detection.track_id}", + flush=True, + ) + + # ------------------------------------------------------------------------- + # Prompt helpers + # ------------------------------------------------------------------------- + def build_privacy_reason_prompt( + self, + user_prompt: str, + visible_text: str, + ) -> str: + return ( + "You are deciding whether a text group and its surrounding object should be privacy-blurred.\n" + f"User request: {user_prompt}\n" + f"Visible text from the grouped region:\n{visible_text}\n" + "Decide based on the visible text and obvious visible context only.\n" + "If this grouped region likely belongs to a privacy-sensitive object such as a receipt, waybill(delivery label with name/address/phone number), address label, card, ID document, license plate, or other personal text-bearing object, answer yes. Road signs, subtitles, and brand logos are NOT privacy-sensitive.\n" + "Short noisy fragments such as a single digit or random short token alone are not enough unless the overall grouped region strongly suggests a sensitive object.\n" + "If any part of the grouped region likely contains privacy-sensitive information, treat the whole grouped object as blur-worthy.\n" + "You MUST return exactly three lines in this order, do NOT skip or merge any line:\n" + "Decision: yes or no\n" + "Label: receipt | waybill | credit card | driver's license | license plate | road sign | address document | other private text | other\n" + "Reason: \n" + "Example output:\n" + "Decision: yes\n" + "Label: waybill\n" + "Reason: Contains name, address and phone number typical of a delivery label.\n" + ) + + # ------------------------------------------------------------------------- + # Qwen helpers + # ------------------------------------------------------------------------- + def ask_qwen_reason( + self, + image_pil: Image.Image, + prompt_text: str, + max_new_tokens: int, + ) -> str: + messages = [ + { + "role": "user", + "content": [ + {"type": "image", "image": image_pil}, + {"type": "text", "text": prompt_text}, + ], + } + ] + + text_prompt = self.qwen_processor.apply_chat_template( + messages, + tokenize=False, + add_generation_prompt=True, + ) + + inputs = self.qwen_processor( + text=[text_prompt], + images=[image_pil], + padding=True, + return_tensors="pt", + ) + + for key, value in inputs.items(): + if hasattr(value, "to"): + inputs[key] = value.to(self.device) + + with torch.inference_mode(): + output_ids = self.qwen_model.generate( + **inputs, + max_new_tokens=max_new_tokens, + do_sample=False, + use_cache=True, + ) + + prompt_len = int(inputs["input_ids"].shape[1]) + generated_ids = output_ids[:, prompt_len:] + + answer = self.qwen_processor.batch_decode( + generated_ids, + skip_special_tokens=True, + clean_up_tokenization_spaces=True, + )[0] + + return answer.strip() + + # ------------------------------------------------------------------------- + # Public API + # ------------------------------------------------------------------------- + def process_video( + self, + video_path: str, + user_prompt: str = "내 프라이버시가 유출될 만한 것들을 가려줘.", + sample_fps: float = 1.0, + ) -> List[TrackState]: + print("[Start] process_video 시작", flush=True) + print(f"[Start] video_path = {video_path}", flush=True) + print(f"[Start] user_prompt = {user_prompt}", flush=True) + print(f"[Start] sample_fps = {sample_fps}", flush=True) + + if sample_fps <= 0: + raise ValueError("sample_fps must be > 0.") + + self.active_tracks = {} + self.finished_tracks = [] + self.last_verified_detections = [] + self.last_tracks = [] + self.last_json_payload = [] + self._next_track_index = 1 + + cap = cv2.VideoCapture(video_path) + print(f"[Start] VideoCapture isOpened = {cap.isOpened()}", flush=True) + + if not cap.isOpened(): + raise FileNotFoundError(f"Could not open video: {video_path}") + + total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) + native_fps = float(cap.get(cv2.CAP_PROP_FPS)) + width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) + height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) + + if native_fps <= 0 or not np.isfinite(native_fps): + native_fps = 30.0 + + sample_interval_frames = max(1, int(round(native_fps / sample_fps))) + max_track_gap_frames = max(sample_interval_frames * 2, sample_interval_frames + 1) + + print(f"[Start] total_frames = {total_frames}", flush=True) + print(f"[Start] native_fps = {native_fps}", flush=True) + print(f"[Start] resolution = {width}x{height}", flush=True) + print(f"[Start] sample_interval_frames = {sample_interval_frames}", flush=True) + print(f"[Start] max_track_gap_frames = {max_track_gap_frames}", flush=True) + + frame_index = 0 + sampled_count = 0 + + try: + while True: + ret, frame_bgr = cap.read() + if not ret: + break + + if frame_index % sample_interval_frames != 0: + frame_index += 1 + continue + + sampled_count += 1 + + try: + groups = self.detect_grouped_text_candidates( + frame_bgr=frame_bgr, + frame_index=frame_index, + ) + except Exception as e: + print(f"[Error] Frame {frame_index} detect/group 실패: {e}", flush=True) + traceback.print_exc() + frame_index += 1 + continue + + if len(groups) > 0: + self._log_group_found(frame_index, len(groups)) + + frame_verified: List[VerifiedDetection] = [] + + for group in groups: + try: + verified_detection = self.verify_group_reason_only( + frame_bgr=frame_bgr, + group=group, + user_prompt=user_prompt, + ) + except Exception as e: + print(f"[Error] Frame {frame_index} reason 실패: {e}", flush=True) + traceback.print_exc() + continue + + if verified_detection is not None: + frame_verified.append(verified_detection) + + self.update_tracks_for_frame( + detections=frame_verified, + frame_index=frame_index, + max_track_gap_frames=max_track_gap_frames, + ) + + for det in frame_verified: + self.last_verified_detections.append(det) + self._log_verified(det) + + frame_index += 1 + + finally: + cap.release() + + self.finalize_all_tracks() + self.last_tracks = list(self.finished_tracks) + + print(f"[Done] 샘플링된 프레임 수 = {sampled_count}", flush=True) + print(f"[Done] 총 검증 통과 수 = {len(self.last_verified_detections)}", flush=True) + print(f"[Done] 총 트랙 수 = {len(self.last_tracks)}", flush=True) + + return self.last_tracks + + # ------------------------------------------------------------------------- + # Stage 1: PaddleOCR detection + recognition + grouping + # ------------------------------------------------------------------------- + def detect_grouped_text_candidates( + self, + frame_bgr: np.ndarray, + frame_index: int, + ) -> List[TextGroupCandidate]: + height, width = frame_bgr.shape[:2] + frame_area = float(width * height) + + frame_rgb = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB) + raw_results = self.text_detector.predict(frame_rgb) + + items = self.extract_text_items_from_predict_results(raw_results, frame_index) + + filtered_items: List[OCRTextItem] = [] + for item in items: + x1, y1, x2, y2 = item.box + box_w = max(1, x2 - x1) + box_h = max(1, y2 - y1) + area = float(box_w * box_h) + + if area < self.min_text_box_area: + continue + if area > frame_area * self.max_text_box_area_ratio: + continue + if box_w < self.min_text_region_width or box_h < self.min_text_region_height: + continue + if item.score < self.min_rec_score: + continue + if self.is_missing_or_weak_visible_text(item.text): + continue + if self.visible_text_is_gibberish(item.text): + continue + + filtered_items.append(item) + + print(f"[Debug][Frame {frame_index}] filtered text items = {len(filtered_items)}", flush=True) + + if len(filtered_items) == 0: + return [] + + groups = self.group_text_items(filtered_items, frame_w=width, frame_h=height) + + candidates: List[TextGroupCandidate] = [] + for group_items in groups: + if len(group_items) < self.min_group_items: + continue + + gx1 = min(item.x1 for item in group_items) + gy1 = min(item.y1 for item in group_items) + gx2 = max(item.x2 for item in group_items) + gy2 = max(item.y2 for item in group_items) + + merged_text = self.build_group_text(group_items) + avg_score = float(sum(item.score for item in group_items) / max(1, len(group_items))) + + if self.group_text_is_too_weak(merged_text, len(group_items)): + continue + + candidates.append( + TextGroupCandidate( + frame_index=frame_index, + x1=gx1, + y1=gy1, + x2=gx2, + y2=gy2, + merged_text=merged_text, + avg_score=avg_score, + item_count=len(group_items), + items=group_items, + ) + ) + + print(f"[Debug][Frame {frame_index}] group candidates = {len(candidates)}", flush=True) + for i, cand in enumerate(candidates[:10]): + print( + f"[Debug][Frame {frame_index}] group[{i}] items={cand.item_count} score={cand.avg_score:.3f} text={repr(cand.merged_text)} box={cand.box}", + flush=True, + ) + + return candidates + + def extract_text_items_from_predict_results( + self, + raw_results: Any, + frame_index: int, + ) -> List[OCRTextItem]: + items: List[OCRTextItem] = [] + + if raw_results is None: + return items + + try: + result_list = list(raw_results) + except TypeError: + result_list = [raw_results] + + if frame_index == 0: + print(f"[Debug][Frame {frame_index}] predict result count = {len(result_list)}", flush=True) + + for item in result_list: + data = None + + if hasattr(item, "res"): + data = item.res + elif isinstance(item, dict): + data = item + else: + try: + data = dict(item) + except Exception: + data = None + + if not isinstance(data, dict): + continue + + if frame_index == 0: + print(f"[Debug][Frame {frame_index}] result keys = {list(data.keys())}", flush=True) + + rec_texts = data.get("rec_texts", None) + rec_scores = data.get("rec_scores", None) + rec_boxes = data.get("rec_boxes", None) + rec_polys = data.get("rec_polys", None) + dt_polys = data.get("dt_polys", None) + + if rec_texts is None: + continue + + rec_texts = list(rec_texts) + rec_scores = list(rec_scores) if rec_scores is not None else [1.0] * len(rec_texts) + + box_source = None + if rec_boxes is not None: + box_source = rec_boxes + elif rec_polys is not None: + box_source = rec_polys + elif dt_polys is not None: + box_source = dt_polys + + if box_source is None: + continue + + try: + box_arr = np.asarray(box_source, dtype=np.float32) + except Exception: + continue + + if box_arr.ndim == 2 and box_arr.shape[-1] == 4: + for i, text in enumerate(rec_texts): + if i >= len(box_arr): + break + x1, y1, x2, y2 = box_arr[i].tolist() + items.append( + OCRTextItem( + frame_index=frame_index, + x1=int(round(x1)), + y1=int(round(y1)), + x2=int(round(x2)), + y2=int(round(y2)), + text=str(text).strip(), + score=float(rec_scores[i]) if i < len(rec_scores) else 1.0, + ) + ) + + elif box_arr.ndim == 3 and box_arr.shape[-1] == 2: + for i, text in enumerate(rec_texts): + if i >= len(box_arr): + break + poly = box_arr[i] + xs = poly[:, 0] + ys = poly[:, 1] + items.append( + OCRTextItem( + frame_index=frame_index, + x1=int(round(float(xs.min()))), + y1=int(round(float(ys.min()))), + x2=int(round(float(xs.max()))), + y2=int(round(float(ys.max()))), + text=str(text).strip(), + score=float(rec_scores[i]) if i < len(rec_scores) else 1.0, + ) + ) + + return items + + def group_text_items(self, items: List[OCRTextItem], frame_w: int = 9999, frame_h: int = 9999) -> List[List[OCRTextItem]]: + if len(items) == 0: + return [] + + items = sorted(items, key=lambda it: (it.y1, it.x1)) + + groups: List[List[OCRTextItem]] = [] + used = [False] * len(items) + + for i in range(len(items)): + if used[i]: + continue + + current_group = [items[i]] + used[i] = True + + changed = True + while changed: + changed = False + gx1 = min(it.x1 for it in current_group) + gy1 = min(it.y1 for it in current_group) + gx2 = max(it.x2 for it in current_group) + gy2 = max(it.y2 for it in current_group) + gh = max(1, gy2 - gy1) + gw = max(1, gx2 - gx1) + + for j in range(len(items)): + if used[j]: + continue + + it = items[j] + iou = self.compute_iou((gx1, gy1, gx2, gy2), it.box) + + x_gap = self.box_horizontal_gap((gx1, gy1, gx2, gy2), it.box) + y_gap = self.box_vertical_gap((gx1, gy1, gx2, gy2), it.box) + + close_x = x_gap <= int(round(gh * self.group_x_gap_ratio)) + close_y = y_gap <= int(round(gh * self.group_y_gap_ratio)) + overlaps = iou > 0.0 + + within_band = ( + it.x1 <= gx2 + int(round(gw * 0.3)) + and it.x2 >= gx1 - int(round(gw * 0.3)) + and it.y1 <= gy2 + int(round(gh * 0.8)) + and it.y2 >= gy1 - int(round(gh * 0.8)) + ) + max_group_area = frame_w * frame_h * 0.15 + if (gx2 - gx1) * (gy2 - gy1) > max_group_area: + break + if (close_x and close_y) or overlaps or within_band: + current_group.append(it) + used[j] = True + changed = True + + groups.append(sorted(current_group, key=lambda it: (it.y1, it.x1))) + + return groups + + def build_group_text(self, items: List[OCRTextItem]) -> str: + if len(items) == 0: + return "" + + lines: List[List[OCRTextItem]] = [] + for item in sorted(items, key=lambda it: (it.y1, it.x1)): + placed = False + for line in lines: + ly1 = min(x.y1 for x in line) + ly2 = max(x.y2 for x in line) + lh = max(1, ly2 - ly1) + if abs(item.y1 - ly1) <= int(round(lh * 0.6)): + line.append(item) + placed = True + break + if not placed: + lines.append([item]) + + out_lines: List[str] = [] + for line in lines: + line_sorted = sorted(line, key=lambda it: it.x1) + line_text = " ".join(it.text.strip() for it in line_sorted if it.text.strip()) + if line_text: + out_lines.append(line_text) + + return "\n".join(out_lines) + + def group_text_is_too_weak(self, merged_text: str, item_count: int) -> bool: + text = merged_text.strip() + if text == "": + return True + + norm = self.normalize_text(text) + if norm in ("none", "unknown", "unreadable"): + return True + + has_digits = re.search(r"\d", text) is not None + has_hangul = re.search(r"[가-힣]", text) is not None + has_alpha_word = re.search(r"[A-Za-z]{3,}", text) is not None + has_cjk = re.search(r"[\u4e00-\u9fff]", text) is not None + + if item_count == 1: + compact = re.sub(r"\s+", "", text) + if len(compact) <= 2 and not has_hangul and not has_alpha_word: + return True + + if not (has_digits or has_hangul or has_alpha_word or has_cjk): + return True + + return False + + # ------------------------------------------------------------------------- + # Stage 2: Qwen reasoning only on grouped object + # ------------------------------------------------------------------------- + def verify_group_reason_only( + self, + frame_bgr: np.ndarray, + group: TextGroupCandidate, + user_prompt: str, + ) -> Optional[VerifiedDetection]: + frame_index = group.frame_index + visible_text = group.merged_text.strip() + + if self.group_text_is_too_weak(visible_text, group.item_count): + return None + + crop_rgb = None + try: + crop_rgb = self.crop_with_margin( + frame_bgr=frame_bgr, + x1=group.x1, + y1=group.y1, + x2=group.x2, + y2=group.y2, + margin_ratio=self.crop_margin_ratio, + ) + + if crop_rgb is None: + return None + + crop_rgb = self.ensure_min_crop_size(crop_rgb, self.min_qwen_crop_size) + image_pil = Image.fromarray(crop_rgb) + + print( + f"[Group-REC][Frame {frame_index}] items={group.item_count} score={group.avg_score:.3f} text={repr(visible_text)} box={group.box}", + flush=True, + ) + + reason_prompt = self.build_privacy_reason_prompt( + user_prompt=user_prompt, + visible_text=visible_text, + ) + raw_reason_answer = self.ask_qwen_reason( + image_pil=image_pil, + prompt_text=reason_prompt, + max_new_tokens=self.max_new_tokens_reason, + ) + + decision, normalized_label, reason = self.parse_qwen_reason_answer( + answer=raw_reason_answer, + fallback_label="other private text", + ) + + print( + f"[Qwen-Reason][Frame {frame_index}] decision={decision} label={normalized_label} text={repr(visible_text)} reason={reason} raw_answer={repr(raw_reason_answer)}", + flush=True, + ) + + if not decision: + return None + + det = VerifiedDetection( + frame_index=frame_index, + x1=group.x1, + y1=group.y1, + x2=group.x2, + y2=group.y2, + label=normalized_label, + detector_phrase="grouped text object", + detector_score=float(group.avg_score), + qwen_visible_text=visible_text, + qwen_reason=reason, + ) + + if self.save_verified_crops: + self.save_debug_crop(crop_rgb=crop_rgb, detection=det) + + return det + + except Exception as e: + print(f"[Error] Frame {frame_index} group reason 예외: {e}", flush=True) + traceback.print_exc() + return None + + finally: + if torch.cuda.is_available(): + torch.cuda.empty_cache() + gc.collect() + + def parse_qwen_reason_answer( + self, + answer: str, + fallback_label: str, + ) -> Tuple[bool, str, str]: + text = answer.strip() + + decision_match = re.search(r"decision\s*:\s*(yes|no)", text, flags=re.IGNORECASE) + label_match = re.search( + r"label\s*:\s*(receipt|credit card|driver'?s license|license plate|address document|other private text|other)", + text, + flags=re.IGNORECASE, + ) + reason_match = re.search(r"reason\s*:\s*(.+)", text, flags=re.IGNORECASE | re.DOTALL) + + decision = False + if decision_match is not None: + decision = decision_match.group(1).lower() == "yes" + else: + decision = self.parse_yes_no(text) + + raw_label = fallback_label + if label_match is not None: + raw_label = label_match.group(1).lower().strip() + + reason = "" + if reason_match is not None: + reason = reason_match.group(1).strip().splitlines()[0].strip() + + normalized_label = self.normalize_label_name(raw_label) + return decision, normalized_label, reason + + def is_missing_or_weak_visible_text(self, visible_text: str) -> bool: + if visible_text is None: + return True + + vt = visible_text.strip() + if vt == "": + return True + + vt_norm = self.normalize_text(vt) + if vt_norm in ("none", "n a", "na", "unknown", "not visible", "unreadable"): + return True + + if vt_norm in self.WEAK_EVIDENCE_WORDS: + return True + + has_digits = re.search(r"\d", vt) is not None + has_hangul = re.search(r"[가-힣]", vt) is not None + has_alpha_word = re.search(r"[A-Za-z]{2,}", vt) is not None + has_cjk = re.search(r"[\u4e00-\u9fff]", vt) is not None + + if not (has_digits or has_hangul or has_alpha_word or has_cjk): + return True + + return False + + def visible_text_is_gibberish(self, visible_text: str) -> bool: + vt = visible_text.strip() + + if len(vt) >= 8 and len(set(vt)) == 1: + return True + + if re.fullmatch(r"(\d)\1{7,}", vt): + return True + if re.fullmatch(r"([A-Za-z])\1{7,}", vt): + return True + + digits = re.sub(r"\D", "", vt) + if len(digits) >= 12 and len(set(digits)) <= 2: + return True + + cleaned = re.sub(r"[0-9A-Za-z가-힣\u4e00-\u9fff\s:/\-.(),#]", "", vt) + if len(cleaned) >= 5 and len(vt) >= 10: + return True + + return False + + def normalize_label_name(self, label: str) -> str: + n = self.normalize_text(label) + if n in ("driver s license", "driver's license", "driver license", "driving license", "id card", "identity card"): + return "driver's license" + if n in ("credit card", "debit card", "bank card", "payment card", "card"): + return "credit card" + if n in ("receipt", "bill", "paper receipt", "purchase receipt"): + return "receipt" + if n in ("license plate", "number plate", "plate", "car plate", "vehicle plate"): + return "license plate" + if n in ("address document",): + return "address document" + if n in ("waybill", "delivery label", "shipping label", "invoice"): + return "waybill" + if n in ("road sign", "street sign", "traffic sign", "sign"): + return "road sign" + if n in ("other private text", "other private", "private text"): + return "other private text" + return "other" + + def save_debug_crop(self, crop_rgb: np.ndarray, detection: VerifiedDetection) -> None: + label = str(detection.label).replace(" ", "_").replace("/", "_").replace("'", "") + filename = ( + f"frame_{detection.frame_index}_{label}_" + f"{detection.x1}_{detection.y1}_{detection.x2}_{detection.y2}.jpg" + ) + save_path = self.debug_crop_dir / filename + Image.fromarray(crop_rgb).save(save_path) + + # ------------------------------------------------------------------------- + # Stage 3: Simple ID-based tracking + # ------------------------------------------------------------------------- + def update_tracks_for_frame( + self, + detections: List[VerifiedDetection], + frame_index: int, + max_track_gap_frames: int, + ) -> None: + expired_ids = [ + track_id + for track_id, track in self.active_tracks.items() + if frame_index - track.last_frame > max_track_gap_frames + ] + for track_id in expired_ids: + self.finished_tracks.append(self.active_tracks.pop(track_id)) + + if len(detections) == 0: + return + + track_ids = list(self.active_tracks.keys()) + candidate_pairs: List[Tuple[float, int, str]] = [] + + for det_idx, det in enumerate(detections): + for track_id in track_ids: + track = self.active_tracks[track_id] + # 라벨이 달라도 위치(IoU)가 겹치면 같은 객체로 인정 + # if track.label != det.label: + # continue + + iou = self.compute_iou(track.last_box, det.box) + if iou >= self.track_iou_threshold: + candidate_pairs.append((iou, det_idx, track_id)) + + candidate_pairs.sort(key=lambda x: x[0], reverse=True) + + matched_det_indices = set() + matched_track_ids = set() + + for _, det_idx, track_id in candidate_pairs: + if det_idx in matched_det_indices or track_id in matched_track_ids: + continue + + det = detections[det_idx] + track = self.active_tracks[track_id] + track.update(det) + det.track_id = track.object_id + + matched_det_indices.add(det_idx) + matched_track_ids.add(track_id) + + for det_idx, det in enumerate(detections): + if det_idx in matched_det_indices: + continue + + new_track = self.create_track(det) + self.active_tracks[new_track.object_id] = new_track + det.track_id = new_track.object_id + + def create_track(self, detection: VerifiedDetection) -> TrackState: + object_id = f"object_{self._next_track_index:03d}" + self._next_track_index += 1 + + return TrackState( + object_id=object_id, + label=detection.label, + start_frame=detection.frame_index, + end_frame=detection.frame_index, + last_frame=detection.frame_index, + last_box=detection.box, + history_boxes=[detection.box], + visible_text_samples=[detection.qwen_visible_text] if detection.qwen_visible_text else [], + reason_samples=[detection.qwen_reason] if detection.qwen_reason else [], + hit_count=1, + ) + + def finalize_all_tracks(self) -> None: + for track_id in list(self.active_tracks.keys()): + self.finished_tracks.append(self.active_tracks.pop(track_id)) + + @staticmethod + def compute_iou( + box_a: Tuple[int, int, int, int], + box_b: Tuple[int, int, int, int], + ) -> float: + ax1, ay1, ax2, ay2 = box_a + bx1, by1, bx2, by2 = box_b + + inter_x1 = max(ax1, bx1) + inter_y1 = max(ay1, by1) + inter_x2 = min(ax2, bx2) + inter_y2 = min(ay2, by2) + + inter_w = max(0, inter_x2 - inter_x1) + inter_h = max(0, inter_y2 - inter_y1) + inter_area = inter_w * inter_h + + area_a = max(0, ax2 - ax1) * max(0, ay2 - ay1) + area_b = max(0, bx2 - bx1) * max(0, by2 - by1) + + union_area = area_a + area_b - inter_area + if union_area <= 0: + return 0.0 + + return float(inter_area / union_area) + + @staticmethod + def box_horizontal_gap( + box_a: Tuple[int, int, int, int], + box_b: Tuple[int, int, int, int], + ) -> int: + ax1, _, ax2, _ = box_a + bx1, _, bx2, _ = box_b + if ax2 < bx1: + return bx1 - ax2 + if bx2 < ax1: + return ax1 - bx2 + return 0 + + @staticmethod + def box_vertical_gap( + box_a: Tuple[int, int, int, int], + box_b: Tuple[int, int, int, int], + ) -> int: + _, ay1, _, ay2 = box_a + _, by1, _, by2 = box_b + if ay2 < by1: + return by1 - ay2 + if by2 < ay1: + return ay1 - by2 + return 0 + + # ------------------------------------------------------------------------- + # Export / debug outputs + # ------------------------------------------------------------------------- + def export_to_json( + self, + output_json_path: str, + tracks: Optional[List[TrackState]] = None, + ) -> List[Dict]: + if tracks is None: + tracks = list(self.last_tracks) # 복사본으로 + # 중복 제거 전 디버그 + print(f"[Debug] 중복 제거 전 트랙 수: {len(tracks)}", flush=True) + for t in tracks: + print(f"[RawTrack] {t.object_id} {t.start_frame} ~ {t.end_frame} {t.representative_box()}", flush=True) + # raw 트랙 저장 + import json as _json + raw_payload = [] + for t in tracks: + raw_payload.append({ + "id": t.object_id, + "start_frame": int(t.start_frame), + "end_frame": int(t.end_frame), + "box": t.representative_box(), + "visible_text": t.representative_visible_text(), + }) + with open("raw_tracks.json", "w", encoding="utf-8") as f: + _json.dump(raw_payload, f, ensure_ascii=False, indent=2) + print("[Debug] raw_tracks.json 저장 완료", flush=True) + + # IoU 기반 중복 트랙 제거 + + def compute_iou_box(a, b): + ax1, ay1, ax2, ay2 = a + bx1, by1, bx2, by2 = b + ix1, iy1 = max(ax1, bx1), max(ay1, by1) + ix2, iy2 = min(ax2, bx2), min(ay2, by2) + inter = max(0, ix2-ix1) * max(0, iy2-iy1) + if inter == 0: + return 0.0 + area_a = (ax2-ax1) * (ay2-ay1) + area_b = (bx2-bx1) * (by2-by1) + return inter / (area_a + area_b - inter) + + merged_tracks = [] + used = set() + for i, t1 in enumerate(tracks): + if i in used: + continue + box1 = t1.representative_box() + # 화면 너무 큰 박스 제거 + bw = box1[2] - box1[0] + bh = box1[3] - box1[1] + if bw * bh > 300000: + used.add(i) + continue + best_start = t1.start_frame + best_end = t1.end_frame + for j, t2 in enumerate(tracks): + if i == j or j in used: + continue + box2 = t2.representative_box() + if compute_iou_box(box1, box2) >= 0.5: + best_start = min(best_start, t2.start_frame) + best_end = max(best_end, t2.end_frame) + used.add(j) + t1.start_frame = best_start + t1.end_frame = best_end + merged_tracks.append(t1) + used.add(i) + tracks = merged_tracks + + print(f"[Debug] 중복 제거 후 트랙 수: {len(tracks)}", flush=True) + + payload: List[Dict] = [] + for track in tracks: + if track.start_frame == track.end_frame: + track.end_frame = track.start_frame + 10 + payload.append( + { + "id": track.object_id, + "type": "object", + "label": self.LABEL_DISPLAY_MAP.get(track.label, track.label), + "start_frame": int(track.start_frame), + "end_frame": int(track.end_frame), + "box": track.representative_box(), + "visible_text": track.representative_visible_text(), + } + ) + + out_path = Path(output_json_path) + out_path.parent.mkdir(parents=True, exist_ok=True) + + with open(out_path, "w", encoding="utf-8") as f: + json.dump(payload, f, ensure_ascii=False, indent=2) + + self.last_json_payload = payload + + print(f"[Export] JSON 저장 완료: {output_json_path}", flush=True) + return payload + + def group_detections_by_frame( + self, + detections: Optional[List[VerifiedDetection]] = None, + ) -> Dict[int, List[VerifiedDetection]]: + if detections is None: + detections = self.last_verified_detections + + grouped: Dict[int, List[VerifiedDetection]] = {} + for det in detections: + grouped.setdefault(det.frame_index, []).append(det) + return grouped + + def write_debug_video_fullfps( + self, + input_video_path: str, + output_video_path: str, + detections: Optional[List[VerifiedDetection]] = None, + ) -> None: + cap = cv2.VideoCapture(input_video_path) + if not cap.isOpened(): + raise FileNotFoundError(f"Could not open input video: {input_video_path}") + + fps = float(cap.get(cv2.CAP_PROP_FPS)) + width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) + height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) + + if fps <= 0 or not np.isfinite(fps): + fps = 24.0 + + fourcc = cv2.VideoWriter_fourcc(*"mp4v") + writer = cv2.VideoWriter(output_video_path, fourcc, fps, (width, height)) + + grouped = self.group_detections_by_frame(detections) + + frame_index = 0 + try: + while True: + ret, frame_bgr = cap.read() + if not ret: + break + + frame_dets = grouped.get(frame_index, []) + for det in frame_dets: + self.draw_bbox( + frame_bgr, + det.x1, + det.y1, + det.x2, + det.y2, + f"{det.track_id}:{self.LABEL_DISPLAY_MAP.get(det.label, det.label)}", + ) + + writer.write(frame_bgr) + frame_index += 1 + + finally: + cap.release() + writer.release() + + print(f"[Debug] full-fps bbox 영상 저장 완료: {output_video_path}", flush=True) + + def write_debug_video_sampled( + self, + input_video_path: str, + output_video_path: str, + sample_fps: float, + detections: Optional[List[VerifiedDetection]] = None, + ) -> None: + cap = cv2.VideoCapture(input_video_path) + if not cap.isOpened(): + raise FileNotFoundError(f"Could not open input video: {input_video_path}") + + native_fps = float(cap.get(cv2.CAP_PROP_FPS)) + width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) + height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) + + if native_fps <= 0 or not np.isfinite(native_fps): + native_fps = 24.0 + + sample_interval_frames = max(1, int(round(native_fps / sample_fps))) + + fourcc = cv2.VideoWriter_fourcc(*"mp4v") + writer = cv2.VideoWriter(output_video_path, fourcc, sample_fps, (width, height)) + + grouped = self.group_detections_by_frame(detections) + + frame_index = 0 + try: + while True: + ret, frame_bgr = cap.read() + if not ret: + break + + if frame_index % sample_interval_frames != 0: + frame_index += 1 + continue + + frame_dets = grouped.get(frame_index, []) + for det in frame_dets: + self.draw_bbox( + frame_bgr, + det.x1, + det.y1, + det.x2, + det.y2, + f"{det.track_id}:{self.LABEL_DISPLAY_MAP.get(det.label, det.label)}", + ) + + writer.write(frame_bgr) + frame_index += 1 + + finally: + cap.release() + writer.release() + + print(f"[Debug] sampled bbox 영상 저장 완료: {output_video_path}", flush=True) + + @staticmethod + def draw_bbox(frame_bgr: np.ndarray, x1: int, y1: int, x2: int, y2: int, label: str) -> None: + x1, y1, x2, y2 = int(x1), int(y1), int(x2), int(y2) + cv2.rectangle(frame_bgr, (x1, y1), (x2, y2), (0, 255, 0), 2) + + # PIL로 한글 텍스트 그리기 + from PIL import ImageFont, ImageDraw, Image + img_pil = Image.fromarray(cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB)) + draw = ImageDraw.Draw(img_pil) + try: + font = ImageFont.truetype("/usr/share/fonts/truetype/nanum/NanumGothic.ttf", 20) + except: + font = ImageFont.load_default() + text_y = max(5, y1 - 25) + draw.text((x1, text_y), label, font=font, fill=(0, 255, 0)) + frame_bgr[:] = cv2.cvtColor(np.array(img_pil), cv2.COLOR_RGB2BGR) + + # ------------------------------------------------------------------------- + # Utilities + # ------------------------------------------------------------------------- + def crop_with_margin( + self, + frame_bgr: np.ndarray, + x1: int, + y1: int, + x2: int, + y2: int, + margin_ratio: float = 0.35, + ) -> Optional[np.ndarray]: + height, width = frame_bgr.shape[:2] + + box_w = max(1, x2 - x1) + box_h = max(1, y2 - y1) + + margin_x = int(round(box_w * margin_ratio)) + margin_y = int(round(box_h * margin_ratio)) + + crop_x1 = max(0, x1 - margin_x) + crop_y1 = max(0, y1 - margin_y) + crop_x2 = min(width, x2 + margin_x) + crop_y2 = min(height, y2 + margin_y) + + if crop_x2 <= crop_x1 or crop_y2 <= crop_y1: + return None + + crop_bgr = frame_bgr[crop_y1:crop_y2, crop_x1:crop_x2] + if crop_bgr.size == 0: + return None + + crop_rgb = cv2.cvtColor(crop_bgr, cv2.COLOR_BGR2RGB) + return crop_rgb + + def ensure_min_crop_size(self, crop_rgb: np.ndarray, min_size: int) -> np.ndarray: + h, w = crop_rgb.shape[:2] + new_h = max(h, min_size) + new_w = max(w, min_size) + + if new_h == h and new_w == w: + return crop_rgb + + pad_top = (new_h - h) // 2 + pad_bottom = new_h - h - pad_top + pad_left = (new_w - w) // 2 + pad_right = new_w - w - pad_left + + return cv2.copyMakeBorder( + crop_rgb, + pad_top, + pad_bottom, + pad_left, + pad_right, + borderType=cv2.BORDER_REPLICATE, + ) + + @staticmethod + def parse_yes_no(answer: str) -> bool: + text = answer.lower().strip() + + match = re.search(r"\b(yes|no)\b", text) + if match is not None: + return match.group(1) == "yes" + + if text.startswith("yes"): + return True + if text.startswith("no"): + return False + + return False + + @staticmethod + def normalize_text(text: str) -> str: + text = text.lower().strip() + text = text.replace("_", " ") + text = text.replace("-", " ") + text = re.sub(r"\s+", " ", text) + return text.strip(" .,:;!?") + + @staticmethod + def clamp_box( + box_xyxy: Sequence[float], + width: int, + height: int, + ) -> Tuple[int, int, int, int]: + x1, y1, x2, y2 = box_xyxy + x1 = int(max(0, min(width - 1, round(float(x1))))) + y1 = int(max(0, min(height - 1, round(float(y1))))) + x2 = int(max(0, min(width, round(float(x2))))) + y2 = int(max(0, min(height, round(float(y2))))) + return x1, y1, x2, y2 + def parse_args(project_root: Path) -> argparse.Namespace: parser = argparse.ArgumentParser(description="PrivacyReasoningEngine grouped-object video runner") @@ -49,51 +1431,6 @@ def parse_args(project_root: Path) -> argparse.Namespace: ) return parser.parse_args() -def run_pass3( - video_path, - user_prompt, - sample_fps, - output_dir, -): - output_dir = Path(output_dir) - output_dir.mkdir(parents=True, exist_ok=True) - - debug_crop_dir = output_dir / "debug_crops" - json_output_path = output_dir / "object_db.json" - - engine = PrivacyReasoningEngine( - qwen_model_name=config.OBJECT_QWEN_MODEL_NAME, - device=config.DEVICE, - crop_margin_ratio=config.OBJECT_CROP_MARGIN_RATIO, - max_new_tokens_reason=config.OBJECT_MAX_NEW_TOKENS_REASON, - verbose=True, - save_verified_crops=True, - debug_crop_dir=str(debug_crop_dir), - track_iou_threshold=config.OBJECT_TRACK_IOU_THRESHOLD, - min_text_box_area=config.OBJECT_MIN_TEXT_BOX_AREA, - max_text_box_area_ratio=config.OBJECT_MAX_TEXT_BOX_AREA_RATIO, - min_text_region_width=config.OBJECT_MIN_TEXT_REGION_WIDTH, - min_text_region_height=config.OBJECT_MIN_TEXT_REGION_HEIGHT, - text_detector_lang=config.OBJECT_TEXT_DETECTOR_LANG, - min_rec_score=config.OBJECT_MIN_REC_SCORE, - min_group_items=config.OBJECT_MIN_GROUP_ITEMS, - group_x_gap_ratio=config.OBJECT_GROUP_X_GAP_RATIO, - group_y_gap_ratio=config.OBJECT_GROUP_Y_GAP_RATIO, - min_qwen_crop_size=config.OBJECT_MIN_QWEN_CROP_SIZE, - ) - - tracks = engine.process_video( - video_path=str(video_path), - user_prompt=user_prompt, - sample_fps=sample_fps, - ) - - payload = engine.export_to_json( - output_json_path=str(json_output_path), - tracks=tracks, - ) - - return payload if __name__ == "__main__": print("[Main] 프로그램 시작", flush=True)