#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ============================================================================= THẬP NGŨ NIÊN (THE FIFTEEN SPRINGS) - ZERO-WASTE CONTACT SHEET TILING ENGINE ============================================================================= Module: contact_sheet_engine.py Author: Nguyễn Sĩ Sơn / AI Cinema Autonomous Engineering Reference: FlowKit Architectural Specification (Section 1) Engine ghép khung hình liên hoàn không khoảng chết (Zero-Waste Contact Sheet): - Tính toán ước số chính xác (Exact Divisor Math) cho cols_eff và rows_eff đảm bảo cols_eff * rows_eff == n_frames. - Triệt tiêu 100% các ô đen rác (unfilled black cells), ngăn chặn hoàn toàn hiện tượng Vision LLM (Gemini 3.8 / Flash / Claude) bị ảo giác "black frame defect". - Đóng dấu timestamp badge trực quan [T=X.Xs | Frame N] độ tương phản cao. - Hỗ trợ xuất mảng NumPy, lưu file JPEG chất lượng cao và trích xuất raw JPEG bytes. ============================================================================= """ import os import sys import math from pathlib import Path from typing import Tuple, List, Optional, Union import cv2 import numpy as np # Reconfigure console UTF-8 on Windows if sys.platform == "win32": try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except Exception: pass BASE_DIR = Path(__file__).resolve().parent.parent CRITIC_QUEUE_DIR = BASE_DIR / "04_Assets" / "critic_queue" def compute_zero_waste_grid(n_frames: int) -> Tuple[int, int]: """ Tính toán lưới (cols_eff, rows_eff) sao cho cols_eff * rows_eff == n_frames với tiêu chí: 1. Zero black cells (không ô rác). 2. Tỉ lệ khung hình của lưới gần với tỉ lệ chuẩn 16:9 nhất có thể (cols >= rows). Nếu n_frames là số nguyên tố hoặc không có ước số đẹp, tự động điều chỉnh n_frames về số chia tối ưu gần nhất (ví dụ: 5 -> 6 hoặc 4; 7 -> 6 hoặc 8). """ if n_frames <= 1: return (1, 1) if n_frames == 2: return (2, 1) if n_frames == 3: return (3, 1) # Tìm các cặp ước số (cols, rows) sao cho cols * rows == n_frames candidates: List[Tuple[int, int]] = [] for r in range(1, int(math.isqrt(n_frames)) + 1): if n_frames % r == 0: c = n_frames // r candidates.append((c, r)) if not candidates: # Fallback an toàn nếu có vấn đề toán học return (n_frames, 1) # Chọn cặp có cols >= rows và tỉ lệ (cols / rows) gần với 16:9 (~1.77) hoặc 4:3 (~1.33) target_ratio = 16.0 / 9.0 best_pair = candidates[0] min_diff = float("inf") for c, r in candidates: ratio = float(c) / float(r) diff = abs(ratio - target_ratio) if diff < min_diff: min_diff = diff best_pair = (c, r) return best_pair def normalize_frame_count(requested_frames: int) -> int: """ Chuẩn hóa số lượng frame kiểm duyệt về các giá trị có ước số tối ưu cho màn hình 16:9 (4, 6, 8, 9, 10, 12, 16) để triệt tiêu hoàn toàn ô đen rác. """ if requested_frames <= 3: return max(1, requested_frames) if requested_frames in (4, 6, 8, 9, 10, 12, 16): return requested_frames # Tập hợp các số chia vàng golden_counts = [4, 6, 8, 9, 10, 12, 16] closest = min(golden_counts, key=lambda g: abs(g - requested_frames)) return closest class ZeroWasteContactSheet: """ Động cơ tạo Contact Sheet không ô đen rác cho quy trình kiểm duyệt AI Cinema. """ @staticmethod def extract_uniform_frames( video_path: Union[str, Path], num_frames: int = 6 ) -> List[Tuple[float, int, np.ndarray]]: """ Trích xuất num_frames khung hình phân bố đều trên toàn bộ thời lượng video. Trả về danh sách các tuple: (timestamp_sec, frame_idx, frame_bgr). """ v_path = Path(video_path) if not v_path.exists() or v_path.stat().st_size == 0: return [] cap = cv2.VideoCapture(str(v_path)) if not cap.isOpened(): return [] total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) fps = cap.get(cv2.CAP_PROP_FPS) or 24.0 if total_frames <= 0: cap.release() return [] eff_frames = normalize_frame_count(num_frames) if total_frames <= 1: indices = [0] * eff_frames if total_frames == 1 else [] elif total_frames <= eff_frames: # Lấy mẫu lặp trải đều để luôn đủ eff_frames và duy trì lưới zero-waste indices = [int(i * (total_frames - 1) / (eff_frames - 1)) for i in range(eff_frames)] else: # Lấy đều từ 0 đến total_frames - 1 indices = [int(i * (total_frames - 1) / (eff_frames - 1)) for i in range(eff_frames)] extracted: List[Tuple[float, int, np.ndarray]] = [] for idx in indices: cap.set(cv2.CAP_PROP_POS_FRAMES, idx) ret, frame = cap.read() if ret and frame is not None: t_sec = round(idx / fps, 2) extracted.append((t_sec, idx, frame)) cap.release() return extracted @classmethod def create_contact_sheet( cls, video_path: Union[str, Path], num_frames: int = 6, output_path: Optional[Union[str, Path]] = None, target_size: Tuple[int, int] = (1920, 1080), add_timestamp_labels: bool = True, preserve_aspect_ratio: bool = False ) -> Tuple[Optional[np.ndarray], Optional[Path]]: """ Tạo contact sheet dạng lưới zero-waste: - target_size: (width, height) tổng thể của contact sheet. - add_timestamp_labels: gắn nhãn thời gian và số thứ tự frame. - preserve_aspect_ratio: duy trì tỉ lệ gốc của frame tránh méo mặt/dị dạng khi xếp lưới. - Trả về (contact_sheet_bgr, output_file_path). """ extracted = cls.extract_uniform_frames(video_path, num_frames=num_frames) if not extracted: return None, None actual_n = len(extracted) cols, rows = compute_zero_waste_grid(actual_n) # Tính toán kích thước cho từng ô cell tot_w, tot_h = target_size if preserve_aspect_ratio and extracted[0][2] is not None: first_frame = extracted[0][2] ar = float(first_frame.shape[1]) / float(first_frame.shape[0]) cell_w = tot_w // cols cell_h = int(round(cell_w / ar)) if cell_h * rows > tot_h: cell_h = tot_h // rows cell_w = int(round(cell_h * ar)) else: cell_w = tot_w // cols cell_h = tot_h // rows # Khởi tạo canvas với kích thước chính xác tuyệt đối sheet = np.zeros((cell_h * rows, cell_w * cols, 3), dtype=np.uint8) for i, (t_sec, frame_idx, frame) in enumerate(extracted): r = i // cols c = i % cols # Resize frame khớp chính xác với ô cell cell_img = cv2.resize(frame, (cell_w, cell_h), interpolation=cv2.INTER_AREA) # Đóng dấu timestamp badge nếu được yêu cầu if add_timestamp_labels: label = f"T={t_sec:.1f}s | #{frame_idx}" # Tạo nền badge bán trong suốt màu đen cv2.rectangle(cell_img, (8, 8), (170, 36), (20, 20, 20), -1) cv2.putText( cell_img, label, (14, 28), cv2.FONT_HERSHEY_SIMPLEX, 0.55, (0, 240, 255), # Màu vàng ngọc sang trọng 1, cv2.LINE_AA ) y_start = r * cell_h y_end = y_start + cell_h x_start = c * cell_w x_end = x_start + cell_w sheet[y_start:y_end, x_start:x_end] = cell_img out_path: Optional[Path] = None if output_path: out_path = Path(output_path) out_path.parent.mkdir(parents=True, exist_ok=True) cv2.imwrite(str(out_path), sheet, [int(cv2.IMWRITE_JPEG_QUALITY), 92]) else: # Mặc định lưu vào critic queue shot_stem = Path(video_path).stem.replace("_10s", "") out_path = CRITIC_QUEUE_DIR / f"{shot_stem}_contact_sheet.jpg" out_path.parent.mkdir(parents=True, exist_ok=True) cv2.imwrite(str(out_path), sheet, [int(cv2.IMWRITE_JPEG_QUALITY), 92]) return sheet, out_path @classmethod def extract_contact_sheet_bytes( cls, video_path: Union[str, Path], num_frames: int = 6, target_size: Tuple[int, int] = (1920, 1080) ) -> Optional[bytes]: """ Trích xuất trực tiếp contact sheet dưới dạng JPEG bytes để nạp vào nội dung Multimodal Vision LLM (Google Antigravity SDK hoặc Google GenAI Client). """ sheet, _ = cls.create_contact_sheet( video_path, num_frames=num_frames, output_path=None, target_size=target_size, add_timestamp_labels=True ) if sheet is None: return None success, buf = cv2.imencode(".jpg", sheet, [int(cv2.IMWRITE_JPEG_QUALITY), 88]) if success: return buf.tobytes() return None # ============================================================================= # CONVENIENCE EXPORTS # ============================================================================= def generate_zero_waste_contact_sheet( video_path: Union[str, Path], num_frames: int = 6, output_path: Optional[Union[str, Path]] = None, target_size: Tuple[int, int] = (1920, 1080) ) -> Tuple[Optional[np.ndarray], Optional[Path]]: """Hàm tiện ích tạo contact sheet không ô đen rác.""" return ZeroWasteContactSheet.create_contact_sheet( video_path=video_path, num_frames=num_frames, output_path=output_path, target_size=target_size, add_timestamp_labels=True ) def extract_contact_sheet_bytes( video_path: Union[str, Path], num_frames: int = 6 ) -> Optional[bytes]: """Hàm tiện ích trích xuất JPEG bytes của contact sheet.""" return ZeroWasteContactSheet.extract_contact_sheet_bytes( video_path=video_path, num_frames=num_frames ) if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Zero-Waste Contact Sheet Tiling Engine") parser.add_argument("--video", required=True, help="Đường dẫn file video đầu vào") parser.add_argument("--frames", type=int, default=6, help="Số khung hình cần lấy mẫu (mặc định: 6)") parser.add_argument("--output", default=None, help="Đường dẫn file ảnh đầu ra (.jpg)") args = parser.parse_args() s, p = generate_zero_waste_contact_sheet(args.video, num_frames=args.frames, output_path=args.output) if p: print(f"[✓] Đã tạo Contact Sheet thành công: {p}") print(f" Kích thước: {s.shape[1]}x{s.shape[0]}") else: print("[!] Không thể tạo Contact Sheet từ video chỉ định.")