ImageProcessor_MainServer/worker/utils_debug.py

108 lines
3.8 KiB
Python

# utils_debug.py
import os, json, cv2, numpy as np
from datetime import datetime
# ── 공통 ─────────────────────────────────────────────────────────
def _truthy(v: str) -> bool:
return str(v).lower() in ("1", "true", "yes", "on")
def _get_debug_dir(default="/app/temp_files/debug") -> str:
return os.getenv("DEBUG_DUMP_DIR", default)
def _debug_enabled() -> bool:
return _truthy(os.getenv("DEBUG_DUMP_ENABLE", "0"))
def _ensure_dir(path: str):
os.makedirs(path, exist_ok=True)
def _dump_one(value, base_no_ext: str):
"""
value 타입에 맞춰 저장:
- np.ndarray(BGR/GRAY): PNG
- dict/list: JSON
- bytes/bytearray: .bin
- 그 외: .txt
반환: 실제 저장된 파일 경로(str)
"""
if isinstance(value, np.ndarray):
path = base_no_ext + ".png"
cv2.imwrite(path, value)
return path
if isinstance(value, (dict, list)):
path = base_no_ext + ".json"
with open(path, "w", encoding="utf-8") as f:
json.dump(value, f, ensure_ascii=False, indent=2)
return path
if isinstance(value, (bytes, bytearray)):
path = base_no_ext + ".bin"
with open(path, "wb") as f:
f.write(value)
return path
path = base_no_ext + ".txt"
with open(path, "w", encoding="utf-8") as f:
f.write(str(value))
return path
# ── 메인 API (시그니처 호환) ─────────────────────────────────────
def save_debug_artifacts(debug_dir: str, guid: str, **artifacts):
"""
기존 호출과 동일한 시그니처:
save_debug_artifacts(debug_dir, guid, ocr=..., mask=..., inpaint=..., final=..., ...)
저장 위치:
{DEBUG_DUMP_DIR or debug_dir}/{guid}/[stage/]key.png …
예약 키워드:
_stage="OCR"{dir}/{guid}/OCR/key.png
_prefix="001" → 파일명 앞에 접두사
환경변수:
DEBUG_DUMP_ENABLE: "1"|"true"|... 이면 저장, 아니면 no-op
DEBUG_DUMP_DIR : 기본 저장 루트 (debug_dir 인자가 None이거나 빈 문자열이면 이를 사용)
반환: {key: saved_path, ...}
"""
# 비활성화면 no-op
if not _debug_enabled():
return {}
# 인자로 온 경로가 빈 값이면 환경변수 사용
base_root = debug_dir or _get_debug_dir()
stage = artifacts.pop("_stage", None)
prefix = artifacts.pop("_prefix", None)
# 폴더 구성
base_dir = os.path.join(base_root, guid, stage) if stage else os.path.join(base_root, guid)
_ensure_dir(base_dir)
# 묶음 타임스탬프
ts_file = os.path.join(base_dir, "_time.txt")
if not os.path.exists(ts_file):
with open(ts_file, "w", encoding="utf-8") as f:
f.write(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
saved = {}
for key, val in artifacts.items():
if val is None:
continue
name = f"{prefix}_{key}" if prefix else key
try:
saved[key] = _dump_one(val, os.path.join(base_dir, name))
except Exception as e:
print(f"[DEBUG-DUMP] save fail: {key}{e}")
return saved
# ── 보조: OCR 시각화는 기존과 동일 ──────────────────────────────
def draw_ocr_overlay(img_bgr, ocr_items, color=(0,255,0)):
vis = img_bgr.copy()
for r in ocr_items:
poly = np.array(r["polygon"], np.int32)
cv2.polylines(vis, [poly], True, color, 2)
x,y,w,h = r["bbox"]
cv2.putText(vis, r["text"], (x, max(0,y-5)),
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0,0,255), 2)
return vis