"""
HyperCLOVAXOmniProcessor Integration Test (Data Samples from Preprocessor)
HyperCLOVAXOmniProcessor의 출력이 기존 Preprocessor의 출력과 일치하는지 검증합니다.
- input.pkl 파일을 읽어 HyperCLOVAXOmniProcessor로 처리
- output.pkl의 ground truth와 비교
- 결과를 JSON 파일로 저장 및 Streamlit UI로 시각화
실행: python test_processor_with_samples.py --launch-ui
실행 결과: test_processor_result.txt, test_processor_result.json 및 Streamlit UI
"""
import os
import sys
import pickle
import torch
import numpy as np
from tqdm import tqdm
from transformers import (
AutoTokenizer,
AutoVideoProcessor,
)
import soundfile as sf
from PIL import Image
import re
import difflib
import json
import argparse
import subprocess
current_dir = os.path.dirname(os.path.abspath(__file__))
project_root = os.path.abspath(os.path.join(current_dir, "../../"))
if project_root not in sys.path:
sys.path.append(project_root)
template_path = os.path.abspath(os.path.join(current_dir, "../track_b/chat_template.jinja"))
from processing_hyperclovax_omni import HyperCLOVAXOmniProcessor
from audio.processing_audio import HCXOmniAudioProcessor
from image.processing_image import HCXOmniImageProcessor
# 더미 리소스 경로
DUMMY_IMG_PATH = "./dummy_resources/example_img.png"
DUMMY_AUDIO_PATH = "./dummy_resources/example_audio.wav"
DUMMY_VIDEO_PATH = "./dummy_resources/example_video.mp4"
def create_dummy_resources():
"""테스트용 더미 이미지, 오디오, 비디오 리소스 생성"""
# 상위 디렉토리 생성 (이미지와 오디오가 같은 폴더에 있다고 가정)
os.makedirs(os.path.dirname(DUMMY_IMG_PATH), exist_ok=True)
os.makedirs(os.path.dirname(DUMMY_AUDIO_PATH), exist_ok=True)
# 이미지 생성
if not os.path.exists(DUMMY_IMG_PATH):
print(f"[*] Creating dummy image at {DUMMY_IMG_PATH}")
img = Image.new("RGB", (384, 384), color="blue")
img.save(DUMMY_IMG_PATH)
# 2. 오디오 생성 로직 변경: 단일 파일만 생성 (루프 제거)
if not os.path.exists(DUMMY_AUDIO_PATH):
print(f"[*] Creating dummy audio at {DUMMY_AUDIO_PATH}")
sr = 16000
# 3초 길이의 랜덤 노이즈 생성
data = np.random.uniform(-0.1, 0.1, int(sr * 3.0))
sf.write(DUMMY_AUDIO_PATH, data, sr, format="WAV")
# 3. 더미 비디오 생성 (4프레임, 384x384, 2fps, 오디오 포함)
if not os.path.exists(DUMMY_VIDEO_PATH):
print(f"[*] Creating dummy video with audio at {DUMMY_VIDEO_PATH}")
os.makedirs(os.path.dirname(DUMMY_VIDEO_PATH), exist_ok=True)
try:
import av
container = av.open(DUMMY_VIDEO_PATH, mode="w")
# 비디오 스트림 추가
video_stream = container.add_stream("mpeg4", rate=2)
video_stream.width = 384
video_stream.height = 384
video_stream.pix_fmt = "yuv420p"
# 오디오 스트림 추가 (16kHz, mono, 2초 오디오)
audio_stream = container.add_stream("aac", rate=16000)
# 2초 길이의 440Hz 사인파 생성 (4프레임 / 2fps = 2초)
sample_rate = 16000
duration = 2.0 # 2 seconds
t = np.linspace(0, duration, int(sample_rate * duration), False)
audio_data = np.sin(2 * np.pi * 440 * t) * 0.3 # 440Hz tone, volume 0.3
audio_data = (audio_data * 32767).astype(np.int16)
# 비디오 프레임 인코딩
video_frames = []
for i in range(4):
img = Image.new("RGB", (384, 384), color=(i * 60, 0, 255 - i * 60))
video_frame = av.VideoFrame.from_image(img)
video_frames.append(video_frame)
# 오디오를 작은 청크로 나눔 (각 비디오 프레임당 0.5초 오디오)
samples_per_frame = int(sample_rate * 0.5)
audio_frames = []
for i in range(4):
start_idx = i * samples_per_frame
end_idx = min(start_idx + samples_per_frame, len(audio_data))
chunk = audio_data[start_idx:end_idx]
audio_frame = av.AudioFrame.from_ndarray(chunk.reshape(1, -1), format="s16", layout="mono")
audio_frame.sample_rate = sample_rate
audio_frames.append(audio_frame)
# 비디오와 오디오 프레임을 인터리브하며 mux
for i in range(4):
# 비디오 프레임 인코딩
for packet in video_stream.encode(video_frames[i]):
container.mux(packet)
# 오디오 프레임 인코딩
if i < len(audio_frames):
for packet in audio_stream.encode(audio_frames[i]):
container.mux(packet)
for packet in video_stream.encode():
container.mux(packet)
for packet in audio_stream.encode():
container.mux(packet)
container.close()
print(f" -> Dummy video with audio created successfully")
except Exception as e:
print(f" -> Failed to create video with audio: {e}, trying without audio")
try:
import imageio
frames = [np.full((384, 384, 3), [i * 60, 0, 255 - i * 60], dtype=np.uint8) for i in range(4)]
imageio.mimwrite(DUMMY_VIDEO_PATH, frames, fps=2)
except ImportError:
import cv2
fourcc = cv2.VideoWriter_fourcc(*"mp4v")
writer = cv2.VideoWriter(DUMMY_VIDEO_PATH, fourcc, 2, (384, 384))
for i in range(4):
frame_bgr = np.full((384, 384, 3), [255 - i * 60, 0, i * 60], dtype=np.uint8)
writer.write(frame_bgr)
writer.release()
def replace_sample_media(sample, config, pkl_dir=None):
"""
샘플의 미디어 파일 경로를 검증하고, 존재하지 않으면 더미 파일로 대체
Returns: (sample, patched_count): 수정된 샘플과 대체된 파일 수
Args:
sample: 샘플 데이터
config: 설정 딕셔너리
pkl_dir: pkl 파일이 있는 디렉토리 (상대 경로 해석의 기준)
"""
img_dir = config.get("img_dir", "")
# img_dir이 상대 경로이고 pkl_dir이 있으면, pkl_dir 기준으로 절대 경로로 변환
if img_dir and not os.path.isabs(img_dir) and pkl_dir:
img_dir = os.path.abspath(os.path.join(pkl_dir, img_dir))
messages = sample.get("messages", [])
patched_count = {"img": 0, "audio": 0}
for msg in messages:
if "image_urls" in msg and msg["image_urls"]:
new_urls = []
for url in msg["image_urls"]:
full_path = url if os.path.isabs(url) else os.path.join(img_dir, url)
if os.path.exists(full_path):
new_urls.append(full_path) # 절대 경로 사용
else:
new_urls.append(DUMMY_IMG_PATH)
patched_count["img"] += 1
msg["image_urls"] = new_urls
for audio_key in ["audio_files", "audio_urls", "audios"]:
if audio_key in msg and msg[audio_key]:
new_audios = []
for url in msg[audio_key]:
full_path = url if os.path.isabs(url) else os.path.join(img_dir, url)
if os.path.exists(full_path):
new_audios.append(full_path) # 절대 경로 사용
else:
new_audios.append(DUMMY_AUDIO_PATH)
patched_count["audio"] += 1
msg[audio_key] = new_audios
for video_key in ["video_files", "video_urls", "videos"]:
if video_key in msg and msg[video_key]:
patched_count.setdefault("video", 0)
new_videos = []
for url in msg[video_key]:
full_path = url if os.path.isabs(url) else os.path.join(img_dir, url)
if os.path.exists(full_path):
new_videos.append(full_path) # 절대 경로 사용
else:
new_videos.append(DUMMY_VIDEO_PATH)
patched_count["video"] += 1
msg[video_key] = new_videos
return sample, patched_count
def compare_and_validate(generated, ground_truth):
"""생성된 출력과 ground truth를 비교하여 검증"""
# 미디어 블록을 정규화하여 구조만 비교
patterns = [
(r"<\|image_start\|>.*?<\|image_end\|>", "[IMAGE_BLOCK]"),
(r"<\|discrete_image_start\|>.*?<\|discrete_image_end\|>", "[DISCRETE_IMAGE_BLOCK]"),
(r"<\|audio_start\|>.*?<\|audio_end\|>", "[AUDIO_BLOCK]"),
(r"<\|discrete_audio_start\|>.*?<\|discrete_audio_end\|>", "[DISCRETE_AUDIO_BLOCK]"),
(r"<\|video_start\|>.*?<\|video_end\|>", "[VIDEO_BLOCK]"),
(r"(<\|image_pad\|>)+", "[IMAGE_PADS]"),
(r"(<\|audio_pad\|>)+", "[AUDIO_PADS]"),
]
def normalize(text):
if not text:
return ""
norm_text = text
for pat, repl in patterns:
norm_text = re.sub(pat, repl, norm_text, flags=re.DOTALL)
return norm_text.strip()
gen_norm = normalize(generated)
gt_norm = normalize(ground_truth)
is_match = gen_norm == gt_norm
diff_msg = ""
if not is_match:
diff = difflib.unified_diff(
gt_norm.splitlines(),
gen_norm.splitlines(),
fromfile="Ground Truth (Normalized)",
tofile="Generated (Normalized)",
lineterm="",
)
diff_msg = "\n".join(list(diff))
return is_match, diff_msg
def _get_mime_type_from_filename(filename):
"""
파일 확장자로부터 MIME 타입 추정
"""
ext = os.path.splitext(filename)[1].lower()
mime_map = {
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".png": "image/png",
".gif": "image/gif",
".webp": "image/webp",
".bmp": "image/bmp",
}
return mime_map.get(ext, "image/jpeg")
def convert_to_openai_conversation(sample):
"""
Preprocessor 형식의 샘플을 OpenAI conversation 형식으로 변환
Preprocessor의 messages 형식을 HyperCLOVAXOmniProcessor가 이해할 수 있는
OpenAI messages format으로 변환합니다.
Args:
sample: Preprocessor 형식의 샘플 딕셔너리
Returns:
OpenAI conversation 형식의 리스트
"""
messages = sample.get("messages", [])
conversation = []
# MIME ID용 글로벌 카운터 (chat template의 ns_img/ns_aud/ns_vid와 동기화)
img_counter = 0
for msg in messages:
role = msg.get("role", "user")
if role == "system":
# System 메시지: candidates 또는 content에서 텍스트 추출, 문자열로 전달
text_content = msg.get("content", "")
if "candidates" in msg and msg["candidates"]:
candidates = msg["candidates"]
chosen = candidates[0]
text_content = chosen["content"] if isinstance(chosen, dict) else chosen
if isinstance(text_content, list):
parts = [p.get("text", "") for p in text_content if isinstance(p, dict) and p.get("type") == "text"]
text_content = "\n".join(parts)
msg_dict = {"role": "system", "content": text_content if text_content else ""}
conversation.append(msg_dict)
continue
if role == "user":
# User 메시지는 multimodal을 위해 리스트로 변환
media_list = []
text_list = []
# 이미지
if "image_urls" in msg and msg["image_urls"]:
for img_path in msg["image_urls"]:
entry = {"type": "image", "image": img_path}
entry["filename"] = os.path.basename(img_path)
media_list.append(entry)
# User 이미지는 chat template의 ns_img 카운터와 동기화
img_counter += len(msg["image_urls"])
# 오디오
found_audios = []
for audio_key in ["audio_files", "audio_urls", "audios"]:
if audio_key in msg and msg[audio_key]:
found_audios.extend(msg[audio_key])
for audio_path in found_audios:
entry = {"type": "audio", "audio": audio_path}
entry["filename"] = os.path.basename(audio_path)
if msg.get("audio_duration"):
entry["audio_duration"] = msg["audio_duration"]
media_list.append(entry)
# 비디오
found_videos = []
for video_key in ["video_files", "video_urls", "videos"]:
if video_key in msg and msg[video_key]:
found_videos.extend(msg[video_key])
for video_path in found_videos:
entry = {"type": "video", "video": video_path}
entry["filename"] = os.path.basename(video_path)
if msg.get("video_duration"):
entry["video_duration"] = msg["video_duration"]
media_list.append(entry)
# 텍스트
text_content = msg.get("content", "")
if isinstance(text_content, str):
text_content = text_content.replace("<|audio|>", "").replace("<|image|>", "").replace("<|video|>", "")
text_content = text_content.strip()
if text_content:
text_list.append({"type": "text", "text": text_content})
content_list = media_list + text_list
msg_dict = {"role": role, "content": content_list if content_list else ""}
elif role == "assistant":
text_content = msg.get("content", "")
image_urls = msg.get("image_urls", [])
# 1. t2i_model_generation tool_call을 content에 임베딩
tool_calls_to_keep = []
if msg.get("tool_calls"):
for tool_call in msg["tool_calls"]:
func = tool_call.get("function", {})
func_name = func.get("name", "")
args = func.get("arguments", {})
if isinstance(args, str):
try:
args = json.loads(args)
except (json.JSONDecodeError, TypeError):
args = {}
if (
func_name == "t2i_model_generation"
and "<|t2i_model_generation_target_discrete_image|>" in text_content
):
# tool_call 텍스트 생성 (preprocessor line 1533-1538과 동일)
tool_call_text = f"\n{func_name}\n"
for key, value in args.items():
arg_value = json.dumps(value, ensure_ascii=False) if not isinstance(value, str) else value
tool_call_text += f"{key}\n{arg_value}\n"
tool_call_text += ""
text_content = text_content.replace(
"<|t2i_model_generation_target_discrete_image|>", tool_call_text
)
else:
# t2i가 아닌 tool_call은 chat template에서 렌더링
tool_call_copy = {"function": {"name": func_name, "arguments": args}}
tool_calls_to_keep.append(tool_call_copy)
# 2. Assistant <|image|> 태그를 MIME + discrete + continuous 블록으로 확장
if image_urls:
turn_img_idx = 0
while "<|image|>" in text_content and turn_img_idx < len(image_urls):
filename = os.path.basename(image_urls[turn_img_idx])
mime_id = f"image_{img_counter:02d}"
mime_type = _get_mime_type_from_filename(filename)
mime_json = json.dumps({"id": mime_id, "type": mime_type, "filename": filename}, ensure_ascii=False)
replacement = (
f"\n<|mime_start|>{mime_json}<|mime_end|>\n"
f"<|discrete_image_start|><|DISCRETE_IMAGE_PAD|><|discrete_image_end|>\n"
f"<|image_start|><|IMAGE_PAD|><|image_end|>"
)
text_content = text_content.replace("<|image|>", replacement, 1)
img_counter += 1
turn_img_idx += 1
msg_dict = {"role": role, "content": text_content}
# process_mm_info용 미디어 파일 경로 전달
found_assistant_audios = []
for audio_key in ["audio_files", "audio_urls", "audios"]:
if audio_key in msg and msg[audio_key]:
found_assistant_audios.extend(msg[audio_key])
if found_assistant_audios:
msg_dict["audio_files"] = found_assistant_audios
if image_urls:
msg_dict["image_files"] = image_urls
found_assistant_videos = []
for video_key in ["video_files", "video_urls", "videos"]:
if video_key in msg and msg[video_key]:
found_assistant_videos.extend(msg[video_key])
if found_assistant_videos:
msg_dict["video_files"] = found_assistant_videos
# reasoning_content 추가
if msg.get("reasoning_content"):
msg_dict["reasoning_content"] = msg["reasoning_content"]
# 나머지 tool_calls 추가 (t2i 이외)
if tool_calls_to_keep:
msg_dict["tool_calls"] = tool_calls_to_keep
elif role == "tool":
# Tool 메시지: dict content에서 response 추출, name 전달
content = msg.get("content", "")
name = msg.get("name", "")
if isinstance(content, dict):
if "response" in content:
if not name:
name = content.get("name", "")
content = content["response"]
else:
content = json.dumps(content, ensure_ascii=False)
elif isinstance(content, str):
try:
parsed = json.loads(content)
if isinstance(parsed, dict) and "response" in parsed:
if not name:
name = parsed.get("name", "")
content = parsed["response"]
except (json.JSONDecodeError, TypeError):
pass
msg_dict = {"role": "tool", "content": content, "name": name}
else:
msg_dict = {"role": role, "content": msg.get("content", "")}
conversation.append(msg_dict)
return conversation
def run_integration_test(
data_root_dir,
output_log_path="integration_test_result.txt",
gt_model_path="/mnt/cmlssd004/HyperCLOVA-VLM/251221_8b_track_b_exp4_step4_ablation5/checkpoint-400",
):
"""
HyperCLOVAXOmniProcessor 통합 테스트 실행
데이터 디렉토리의 모든 input.pkl 파일을 처리하고 output.pkl과 비교합니다.
Args:
data_root_dir: 테스트 데이터 루트 디렉토리 (input.pkl 파일들이 있는 위치)
output_log_path: 상세 로그 파일 경로
gt_model_path: Ground truth tokenizer 모델 경로 (output.pkl 디코딩용)
"""
create_dummy_resources()
# Ground truth 디코딩용 tokenizer 로드
gt_tokenizer = None
if gt_model_path and os.path.exists(gt_model_path):
try:
gt_tokenizer = AutoTokenizer.from_pretrained(gt_model_path)
print(f" -> GT tokenizer loaded from: {gt_model_path}")
except Exception as e:
print(f"[Warning] Failed to load GT tokenizer from {gt_model_path}: {e}")
print(" Falling back to processor tokenizer for output.pkl decoding.")
else:
print(f"[Warning] GT model path not found: {gt_model_path}")
print(" Falling back to processor tokenizer for output.pkl decoding.")
# =========================================================================
# 1. Processor 초기화
# =========================================================================
print("[1] Initializing Processor...")
# Audio Processor 초기화
try:
audio_processor = HCXOmniAudioProcessor(
feature_extractor_path="openai/whisper-tiny", sampling_rate=16000, chunk_unit=80, min_chunk_size=1600
)
audio_processor.audio_token = "<|AUDIO_PAD|>"
audio_processor.audio_start_token = "<|audio_start|>"
audio_processor.audio_end_token = "<|audio_end|>"
audio_processor.discrete_audio_token = "<|DISCRETE_AUDIO_PAD|>"
audio_processor.discrete_audio_start_token = "<|discrete_audio_start|>"
audio_processor.discrete_audio_end_token = "<|discrete_audio_end|>"
audio_processor.use_discrete_token = True
except Exception as e:
print(f"[Critical Error] Failed to init HCXOmniAudioProcessor: {e}")
return
# discrete ratio tokens
discrete_image_ratio_tokens = {
(1, 1): "<|vision_ratio_1:1|>",
(1, 2): "<|vision_ratio_1:2|>",
(2, 1): "<|vision_ratio_2:1|>",
(3, 4): "<|vision_ratio_3:4|>",
(4, 3): "<|vision_ratio_4:3|>",
(9, 16): "<|vision_ratio_9:16|>",
(16, 9): "<|vision_ratio_16:9|>",
(1, 3): "<|vision_ratio_1:3|>",
(3, 1): "<|vision_ratio_3:1|>",
}
try:
image_processor = HCXOmniImageProcessor(
patch_size=14,
merge_size=2,
min_pixels=4 * 28 * 28,
max_pixels=16384 * 28 * 28,
discrete_image_size=384,
discrete_image_ratios=discrete_image_ratio_tokens,
)
image_processor.image_token = "<|IMAGE_PAD|>"
image_processor.image_start_token = "<|image_start|>"
image_processor.image_end_token = "<|image_end|>"
image_processor.use_discrete_token = True
image_processor.discrete_image_token = "<|DISCRETE_IMAGE_PAD|>"
image_processor.discrete_image_start_token = "<|discrete_image_start|>"
image_processor.discrete_image_end_token = "<|discrete_image_end|>"
image_processor.vision_eol_token = "<|vision_eol|>"
image_processor.vision_eof_token = "<|vision_eof|>"
except Exception as e:
print(f"[Critical Error] Failed to init HCXOmniImageProcessor: {e}")
import traceback
traceback.print_exc()
return
# Video Processor 초기화 (현재는 video 처리 미사용, placeholder만 필요)
base_model_path = "Qwen/Qwen2.5-VL-7B-Instruct"
try:
video_processor = AutoVideoProcessor.from_pretrained(base_model_path)
except Exception:
print("[Info] Video processor not available, creating dummy processor.")
from types import SimpleNamespace
video_processor = SimpleNamespace()
if not hasattr(video_processor, "video_token"):
video_processor.video_token = "<|VIDEO_PAD|>"
if not hasattr(video_processor, "video_start_token"):
video_processor.video_start_token = "<|video_start|>"
if not hasattr(video_processor, "video_end_token"):
video_processor.video_end_token = "<|video_end|>"
tokenizer = AutoTokenizer.from_pretrained(base_model_path)
omni_tokens = [
"<|image_start|>",
"<|image_end|>",
"<|video_start|>",
"<|video_end|>",
"<|audio_start|>",
"<|audio_end|>",
"<|discrete_image_start|>",
"<|discrete_image_end|>",
"<|discrete_audio_start|>",
"<|discrete_audio_end|>",
"<|vision_eol|>",
"<|vision_eof|>",
"<|IMAGE_PAD|>",
"<|DISCRETE_IMAGE_PAD|>",
"<|VIDEO_PAD|>",
"<|AUDIO_PAD|>",
"<|DISCRETE_AUDIO_PAD|>",
"<|mime_start|>",
"<|mime_end|>",
"<|audio_aux_start|>",
"<|audio_aux_end|>",
"<|video_aux_start|>",
"<|video_aux_end|>",
"<|audio_duration|>",
"<|video_duration|>",
"<|VIDEO_AUDIO_PAD|>",
]
tokenizer.add_tokens(omni_tokens, special_tokens=True)
with open(template_path, "r", encoding="utf-8") as f:
custom_chat_template = f.read().strip()
tokenizer.chat_template = custom_chat_template
try:
processor = HyperCLOVAXOmniProcessor(
audio_processor=audio_processor,
image_processor=image_processor,
video_processor=video_processor,
tokenizer=tokenizer,
chat_template=tokenizer.chat_template,
)
print(" -> Processor instantiated manually!")
except Exception as e:
print(f"[Critical Error] Failed to instantiate processor: {e}")
import traceback
traceback.print_exc()
return
omni_tokens = [
"<|image_start|>",
"<|image_end|>",
"<|video_start|>",
"<|video_end|>",
"<|audio_start|>",
"<|audio_end|>",
"<|discrete_image_start|>",
"<|discrete_image_end|>",
"<|discrete_audio_start|>",
"<|discrete_audio_end|>",
"<|vision_eol|>",
"<|vision_eof|>",
"<|IMAGE_PAD|>",
"<|DISCRETE_IMAGE_PAD|>",
"<|VIDEO_PAD|>",
"<|AUDIO_PAD|>",
"<|DISCRETE_AUDIO_PAD|>",
"<|mime_start|>",
"<|mime_end|>",
"<|audio_aux_start|>",
"<|audio_aux_end|>",
"<|video_aux_start|>",
"<|video_aux_end|>",
"<|audio_duration|>",
"<|video_duration|>",
"<|VIDEO_AUDIO_PAD|>",
]
processor.tokenizer.add_tokens(omni_tokens, special_tokens=True)
# =========================================================================
# 2. 데이터 샘플 탐색
# =========================================================================
print(f"[2] Scanning data directory: {data_root_dir}")
pkl_files = []
for root, _, files in os.walk(data_root_dir):
if "input.pkl" in files:
pkl_files.append(os.path.join(root, "input.pkl"))
print(f" -> Found {len(pkl_files)} input.pkl files.")
# =========================================================================
# 3. 테스트 루프
# =========================================================================
test_results_summary = []
with open(output_log_path, "w", encoding="utf-8") as out_f:
out_f.write(f"Integration Test Report\n")
out_f.write(f"Data Root: {data_root_dir}\n")
out_f.write("=" * 80 + "\n\n")
for pkl_path in tqdm(pkl_files, desc="Processing Files"):
try:
with open(pkl_path, "rb") as f:
sample = pickle.load(f)
except Exception as e:
out_f.write(f"[Error] Failed to load pickle {pkl_path}: {e}\n\n")
continue
out_f.write(f"[File] {pkl_path}\n")
out_f.write("-" * 40 + "\n")
try:
# 샘플 설정 추출
config = sample.get("vlm", {})
pkl_dir = os.path.dirname(pkl_path)
sample, p_count = replace_sample_media(sample, config, pkl_dir)
if p_count["img"] > 0 or p_count["audio"] > 0:
out_f.write(
f" [Info] Patched Missing Files -> Imgs: {p_count['img']}, Audios: {p_count['audio']}\n"
)
conversation = convert_to_openai_conversation(sample)
audios, images, videos = processor.process_mm_info(conversation, use_audio_in_video=True)
# tools 정보 가져오기
tools = sample.get("tools", None)
prompt_text = processor.apply_chat_template(
conversation, tools=tools, tokenize=False, add_generation_prompt=False
)
out_f.write("-" * 40 + "\n")
out_f.write(f"[1] apply_chat_template output (Raw Prompt):\n")
out_f.write(f"[File] {pkl_path}\n")
out_f.write("-" * 40 + "\n")
out_f.write(f"{prompt_text}\n\n")
inputs = processor(
text=[prompt_text],
audios=audios if audios else None,
images=images if images else None,
videos=videos if videos else None,
return_tensors="pt",
padding=True,
videos_kwargs={"use_video_audio": True},
)
input_ids = inputs["input_ids"][0]
decoded_text = processor.tokenizer.decode(input_ids, skip_special_tokens=False)
out_f.write("-" * 40 + "\n")
out_f.write(f"[2] Auto Processor Decoded Output:\n")
out_f.write(f"[File] {pkl_path}\n")
out_f.write("-" * 40 + "\n")
out_f.write(decoded_text + "\n\n")
output_pkl_path = os.path.join(os.path.dirname(pkl_path), "output.pkl")
gt_decoded = None
if os.path.exists(output_pkl_path):
try:
with open(output_pkl_path, "rb") as f:
output_data = pickle.load(f)
gt_input_ids = output_data["input_ids"]
if isinstance(gt_input_ids, torch.Tensor):
gt_input_ids_1d = gt_input_ids.squeeze()
else:
gt_input_ids_1d = torch.tensor(gt_input_ids).squeeze()
decode_tokenizer = gt_tokenizer if gt_tokenizer is not None else processor.tokenizer
gt_decoded = decode_tokenizer.decode(gt_input_ids_1d, skip_special_tokens=False)
out_f.write("-" * 40 + "\n")
out_f.write(f"[3] output.pkl decoded result (input_ids):\n")
out_f.write(f"[File] {pkl_path}\n")
out_f.write("-" * 40 + "\n")
out_f.write(gt_decoded + "\n\n")
except Exception as e:
out_f.write(f"[3/4] output.pkl: Failed to load ({e})\n\n")
# 더미 미디어 사용하므로 비교 시 PAD 토큰 제거
decoded_text = decoded_text.replace("<|DISCRETE_AUDIO_PAD|>", "")
decoded_text = decoded_text.replace("<|AUDIO_PAD|>", "")
decoded_text = decoded_text.replace("<|DISCRETE_IMAGE_PAD|>", "")
decoded_text = decoded_text.replace("<|IMAGE_PAD|>", "")
decoded_text = decoded_text.replace("<|VIDEO_PAD|>", "")
gt_decoded = gt_decoded.replace("<|DISCRETE_AUDIO_PAD|>", "")
gt_decoded = gt_decoded.replace("<|AUDIO_PAD|>", "")
gt_decoded = gt_decoded.replace("<|DISCRETE_IMAGE_PAD|>", "")
gt_decoded = gt_decoded.replace("<|IMAGE_PAD|>", "")
gt_decoded = gt_decoded.replace("<|VIDEO_PAD|>", "")
# decoded_text(HyperCLOVAXOmniProcessor)와 gt_decoded(Preprocessor) 비교
is_valid, diff_msg = compare_and_validate(decoded_text, gt_decoded)
test_results_summary.append(
{
"file_name": os.path.basename(pkl_path),
"full_path": pkl_path,
"generated": decoded_text,
"ground_truth": gt_decoded,
"is_valid": is_valid,
"diff_msg": diff_msg,
}
)
except Exception as e:
import traceback
out_f.write(f"[Error] Pipeline Failed:\n{traceback.format_exc()}\n")
print(f"[Error] Pipeline Failed:\n{traceback.format_exc()}\n")
json_path = "test_processor_result.json"
with open(json_path, "w", encoding="utf-8") as f:
json.dump(test_results_summary, f, indent=4, ensure_ascii=False)
print(f"\n[Done] Test finished. Check '{output_log_path}'")
def launch_streamlit_ui(json_path="test_processor_result.json", port=8504):
"""
Streamlit UI를 실행하여 테스트 결과 시각화
"""
# Streamlit UI 코드를 임시 파일로 생성
streamlit_code = f'''
import streamlit as st
import json
import os
import re
import difflib
# -----------------------------------------------------------------------------
# 1. 설정 및 데이터 로드
# -----------------------------------------------------------------------------
st.set_page_config(layout="wide", page_title="Integration Test Reviewer")
st.markdown("""
""", unsafe_allow_html=True)
@st.cache_data
def load_data(json_path):
if not os.path.exists(json_path):
print(f"File not found: {{json_path}}")
return []
with open(json_path, "r", encoding="utf-8") as f:
return json.load(f)
JSON_FILE = "{json_path}"
data = load_data(JSON_FILE)
# -----------------------------------------------------------------------------
# 2. 사이드바 (샘플 선택)
# -----------------------------------------------------------------------------
st.sidebar.title("Test Samples")
if not data:
st.error(f"'{{JSON_FILE}}' 파일을 찾을 수 없습니다. 테스트를 먼저 실행해주세요.")
st.stop()
# 통계 표시
total = len(data)
passed = sum(1 for d in data if d['is_valid'])
failed = total - passed
st.sidebar.markdown(f"**Total:** {{total}} | **Passed:** {{passed}} | **Failed:** {{failed}}")
# 필터링 옵션
filter_option = st.sidebar.radio("Filter", ["All", "Failed Only", "Passed Only"])
filtered_indices = []
for i, item in enumerate(data):
if filter_option == "All":
filtered_indices.append(i)
elif filter_option == "Failed Only" and not item['is_valid']:
filtered_indices.append(i)
elif filter_option == "Passed Only" and item['is_valid']:
filtered_indices.append(i)
if not filtered_indices:
st.sidebar.warning("조건에 맞는 샘플이 없습니다.")
st.stop()
st.sidebar.write(f"Filtered: {{len(filtered_indices)}}")
selected_idx = st.sidebar.number_input(
"Select Sample Index",
min_value=0,
max_value=len(filtered_indices) - 1,
value=0,
step=1,
format="%d"
)
actual_idx = filtered_indices[selected_idx]
sample = data[actual_idx]
# 사이드바 샘플 리스트
st.sidebar.divider()
for fi, idx in enumerate(filtered_indices):
d = data[idx]
status = "PASS" if d['is_valid'] else "FAIL"
marker = ">> " if fi == selected_idx else " "
st.sidebar.text(f"{{marker}}[{{fi}}] {{status}} {{d['file_name']}}")
# -----------------------------------------------------------------------------
# 3. 메인 화면
# -----------------------------------------------------------------------------
status_str = "PASSED" if sample['is_valid'] else "FAILED"
st.title(f"Auto Processing Test Reviewer")
st.caption(f"Sample {{selected_idx}}/{{len(filtered_indices)-1}} | {{status_str}} | {{sample['full_path']}}")
generated = sample.get('generated', '') or ''
ground_truth = sample.get('ground_truth', '') or ''
diff_msg = sample.get('diff_msg', '') or ''
st.divider()
# -----------------------------------------------------------------------------
# 4. 요약 정보
# -----------------------------------------------------------------------------
if not ground_truth:
st.warning("Ground Truth (gt_decoded)가 없습니다.")
elif sample['is_valid']:
st.success(f"MATCH - Generated output과 Ground Truth가 일치합니다. (len={{len(generated)}})")
else:
# 첫 번째 차이 위치 찾기
first_diff_pos = -1
for ci, (c1, c2) in enumerate(zip(generated, ground_truth)):
if c1 != c2:
first_diff_pos = ci
break
if first_diff_pos == -1 and len(generated) != len(ground_truth):
first_diff_pos = min(len(generated), len(ground_truth))
st.error(
f"MISMATCH - len(generated)={{len(generated)}}, len(gt)={{len(ground_truth)}}, "
f"first diff @ char {{first_diff_pos}}"
)
# -----------------------------------------------------------------------------
# 5. 미디어 블록 정규화 비교 (구조 비교)
# -----------------------------------------------------------------------------
def normalize_for_structure(text):
"""미디어 PAD 토큰을 축약하여 구조만 비교"""
if not text:
return ""
patterns = [
(r"(<\\|IMAGE_PAD\\|>)+", "<|IMAGE_PAD|>x..."),
(r"(<\\|DISCRETE_IMAGE_PAD\\|>)+", "<|DISCRETE_IMAGE_PAD|>x..."),
(r"(<\\|VIDEO_PAD\\|>)+", "<|VIDEO_PAD|>x..."),
(r"(<\\|AUDIO_PAD\\|>)+", "<|AUDIO_PAD|>x..."),
(r"(<\\|DISCRETE_AUDIO_PAD\\|>)+", "<|DISCRETE_AUDIO_PAD|>x..."),
]
result = text
for pat, repl in patterns:
result = re.sub(pat, repl, result)
return result
# -----------------------------------------------------------------------------
# 6. Diff 시각화
# -----------------------------------------------------------------------------
def create_diff_html(text1, text2):
if not text1: text1 = ""
if not text2: text2 = ""
diff = difflib.HtmlDiff(wrapcolumn=100)
html = diff.make_file(
text1.splitlines(),
text2.splitlines(),
fromdesc="Generated (decoded_text)",
todesc="Ground Truth (gt_decoded)",
context=False,
numlines=3
)
return html
def create_context_diff_html(text1, text2, n_context=3):
"""변경된 부분 주변만 보여주는 context diff"""
if not text1: text1 = ""
if not text2: text2 = ""
diff = difflib.HtmlDiff(wrapcolumn=100)
html = diff.make_file(
text1.splitlines(),
text2.splitlines(),
fromdesc="Generated (decoded_text)",
todesc="Ground Truth (gt_decoded)",
context=True,
numlines=n_context
)
return html
def get_inline_diff_lines(text1, text2):
"""줄 단위 diff를 반환하되, 같은 줄은 건너뛰고 다른 줄만 반환"""
lines1 = (text1 or "").splitlines()
lines2 = (text2 or "").splitlines()
sm = difflib.SequenceMatcher(None, lines1, lines2)
result = []
for tag, i1, i2, j1, j2 in sm.get_opcodes():
if tag == 'equal':
continue
elif tag == 'replace':
for li in range(i1, i2):
result.append(("del", li + 1, lines1[li]))
for lj in range(j1, j2):
result.append(("add", lj + 1, lines2[lj]))
elif tag == 'delete':
for li in range(i1, i2):
result.append(("del", li + 1, lines1[li]))
elif tag == 'insert':
for lj in range(j1, j2):
result.append(("add", lj + 1, lines2[lj]))
return result
# 탭 구성
tab1, tab2, tab3, tab4 = st.tabs([
"Context Diff (Changes Only)",
"Full Diff View",
"Structure Diff (Normalized)",
"Raw Text Side-by-Side",
])
with tab1:
if not ground_truth:
st.info("Ground Truth가 없어 비교할 수 없습니다.")
elif sample['is_valid']:
st.success("일치합니다. 차이가 없습니다.")
else:
st.markdown("**변경된 부분 주변만 표시합니다 (context=3 lines).**")
ctx_html = create_context_diff_html(generated, ground_truth, n_context=3)
st.components.v1.html(ctx_html, height=1200, scrolling=True)
# 줄 단위 차이 요약
diff_lines = get_inline_diff_lines(generated, ground_truth)
if diff_lines:
with st.expander(f"Line-level diff summary ({{len(diff_lines)}} changed lines)", expanded=False):
for dtype, lnum, content in diff_lines[:100]:
prefix = "- (gen)" if dtype == "del" else "+ (gt) "
short = content[:200] + "..." if len(content) > 200 else content
st.text(f" {{prefix}} L{{lnum}}: {{short}}")
if len(diff_lines) > 100:
st.text(f" ... and {{len(diff_lines) - 100}} more lines")
with tab2:
if not ground_truth:
st.info("Ground Truth가 없어 비교할 수 없습니다.")
else:
html_diff = create_diff_html(generated, ground_truth)
st.components.v1.html(html_diff, height=1600, scrolling=True)
with tab3:
st.markdown("**미디어 PAD 토큰을 축약하여 구조적 차이만 비교합니다.**")
gen_norm = normalize_for_structure(generated)
gt_norm = normalize_for_structure(ground_truth)
if gen_norm == gt_norm:
st.success("구조적으로 일치합니다 (PAD 토큰 개수 차이만 존재할 수 있음).")
else:
norm_html = create_context_diff_html(gen_norm, gt_norm, n_context=3)
st.components.v1.html(norm_html, height=1200, scrolling=True)
with tab4:
col1, col2 = st.columns(2)
with col1:
st.markdown("**Generated (decoded_text)**")
st.text_area("gen", generated, height=1000, label_visibility="collapsed", key="raw_gen")
with col2:
st.markdown("**Ground Truth (gt_decoded)**")
st.text_area("gt", ground_truth, height=1000, label_visibility="collapsed", key="raw_gt")
# -----------------------------------------------------------------------------
# 7. Diff Message (from compare_and_validate)
# -----------------------------------------------------------------------------
if diff_msg:
with st.expander("Normalized Diff (from compare_and_validate)", expanded=False):
st.code(diff_msg, language="diff")
'''
# Streamlit 앱 파일 생성
temp_app_path = os.path.join(os.path.dirname(__file__), "_temp_streamlit_viewer.py")
with open(temp_app_path, "w", encoding="utf-8") as f:
f.write(streamlit_code)
print(f"\n[Streamlit UI] Launching viewer on port {{port}}...")
print(f"[Streamlit UI] Opening http://localhost:{{port}} in your browser...")
print(f"[Streamlit UI] Press Ctrl+C to stop the server")
try:
# Streamlit 실행
subprocess.run(
[
"streamlit",
"run",
temp_app_path,
"--server.port",
str(port),
"--server.headless",
"true",
"--browser.gatherUsageStats",
"false",
]
)
except KeyboardInterrupt:
print("\n[Streamlit UI] Server stopped by user")
finally:
# 임시 파일 정리
if os.path.exists(temp_app_path):
os.remove(temp_app_path)
print(f"[Streamlit UI] Cleaned up temporary file")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Run integration tests for HyperCLOVAXOmniProcessor")
parser.add_argument(
"--data-dir",
type=str,
default="/mnt/cmlssd004/HyperCLOVA-VLM/260115_sample_data_trackb/data_fix_path",
help="Root directory containing test data (input.pkl files)",
)
parser.add_argument(
"--output-log", type=str, default="test_processor_result.txt", help="Path to save detailed test log"
)
parser.add_argument(
"--json-output", type=str, default="test_processor_result.json", help="Path to save JSON summary"
)
parser.add_argument(
"--gt-model-path",
type=str,
default="/mnt/cmlssd004/HyperCLOVA-VLM/251221_8b_track_b_exp4_step4_ablation5/checkpoint-400",
help="Path to ground truth model for tokenizer",
)
parser.add_argument("--launch-ui", action="store_true", help="Launch Streamlit UI after tests complete")
parser.add_argument("--ui-port", type=int, default=8504, help="Port for Streamlit UI")
args = parser.parse_args()
# Processor 테스트 실행
run_integration_test(data_root_dir=args.data_dir, output_log_path=args.output_log, gt_model_path=args.gt_model_path)
# Streamlit UI 실행 (옵션)
if args.launch_ui:
launch_streamlit_ui(json_path=args.json_output, port=args.ui_port)