module_debug_viz.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345
  1. """
  2. 模块级 Debug 可视化(Layout / OCR)
  3. 用于 ``{output_dir}/debug/{subdir}/`` 下基于 inference_image 的调试图;
  4. 用户审计图由 VisualizationUtils + original_image 负责,不在此模块。
  5. """
  6. from __future__ import annotations
  7. import json
  8. from pathlib import Path
  9. from typing import Any, Dict, List, Optional, Union
  10. import cv2
  11. import numpy as np
  12. from loguru import logger
  13. from PIL import Image
  14. # 各模块 debug_options 默认落盘根目录(相对 pipeline output_dir)
  15. MODULE_DEBUG_ROOT = "debug"
  16. def resolve_module_debug_dir(
  17. output_dir: Union[str, Path],
  18. subdir: str,
  19. *,
  20. debug_root: str = MODULE_DEBUG_ROOT,
  21. ) -> Path:
  22. """``{output_dir}/{debug_root}/{subdir}/``,目录不存在则创建。"""
  23. debug_dir = Path(output_dir) / debug_root / subdir
  24. debug_dir.mkdir(parents=True, exist_ok=True)
  25. return debug_dir
  26. LAYOUT_CATEGORY_COLORS_BGR = {
  27. 'table_body': (0, 0, 255),
  28. 'table_caption': (0, 0, 200),
  29. 'table_footnote': (0, 0, 150),
  30. 'text': (255, 0, 0),
  31. 'title': (0, 255, 255),
  32. 'header': (255, 0, 255),
  33. 'footer': (0, 165, 255),
  34. 'image_body': (0, 255, 0),
  35. 'image_caption': (0, 200, 0),
  36. 'image_footnote': (0, 150, 0),
  37. 'chart': (255, 255, 0), # 青色(BGR 下 B=255,G=255,R=0)
  38. # 注意:OpenCV 为 BGR,(0,255,255) 在屏幕上呈黄色(与 title 相同),勿用于 seal
  39. 'seal': (0, 140, 255), # 亮橙,与红 table / 黄 title / 蓝 text 均易区分
  40. 'abandon': (128, 128, 128),
  41. }
  42. # seal 常与 table 重叠:加粗线宽 + 黑色外描边
  43. LAYOUT_HIGHLIGHT_CATEGORIES = frozenset({'seal'})
  44. LAYOUT_HIGHLIGHT_LINE_THICKNESS = 4
  45. LAYOUT_HIGHLIGHT_OUTLINE_BGR = (0, 0, 0)
  46. LAYOUT_DEFAULT_LINE_THICKNESS = 2
  47. # OCR 框线宽 (不受配色统一影响)
  48. OCR_BOX_LINE_THICKNESS = 2
  49. OCR_BOX_DASH_LENGTH = 8
  50. OCR_BOX_DASH_GAP = 6
  51. def _ocr_box_color_bgr() -> tuple:
  52. """亮蓝 OCR 框 (BGR),派生自 VisualizationUtils.COLOR_MAP['ocr_box']。"""
  53. from ocr_utils.visualization_utils import VisualizationUtils
  54. return VisualizationUtils.rgb_to_bgr(VisualizationUtils.COLOR_MAP['ocr_box'])
  55. def _seal_ocr_box_color_bgr() -> tuple:
  56. """印章 OCR 框 (BGR),派生自 VisualizationUtils.COLOR_MAP['seal_ocr_box']。"""
  57. from ocr_utils.visualization_utils import VisualizationUtils
  58. return VisualizationUtils.rgb_to_bgr(VisualizationUtils.COLOR_MAP['seal_ocr_box'])
  59. def ocr_box_color_rgb() -> tuple:
  60. """OCR 亮蓝 (RGB),供 PIL / Plotly 使用。"""
  61. from ocr_utils.visualization_utils import VisualizationUtils
  62. return VisualizationUtils.COLOR_MAP['ocr_box']
  63. def _to_bgr(image: Union[np.ndarray, Image.Image]) -> np.ndarray:
  64. if isinstance(image, Image.Image):
  65. arr = np.array(image)
  66. else:
  67. arr = image.copy()
  68. if arr.ndim == 2:
  69. return cv2.cvtColor(arr, cv2.COLOR_GRAY2BGR)
  70. if arr.shape[2] == 3:
  71. return cv2.cvtColor(arr, cv2.COLOR_RGB2BGR)
  72. return arr
  73. def draw_layout_boxes_cv2(
  74. image: Union[np.ndarray, Image.Image],
  75. layout_results: List[Dict[str, Any]],
  76. ) -> np.ndarray:
  77. """在 BGR 图像上绘制 layout 检测框,返回新图像。"""
  78. vis = _to_bgr(image)
  79. for result in layout_results:
  80. bbox = result.get('bbox', [])
  81. if not bbox or len(bbox) < 4:
  82. continue
  83. category = result.get('category', 'unknown')
  84. color = LAYOUT_CATEGORY_COLORS_BGR.get(category, (128, 128, 128))
  85. thickness = (
  86. LAYOUT_HIGHLIGHT_LINE_THICKNESS
  87. if category in LAYOUT_HIGHLIGHT_CATEGORIES
  88. else LAYOUT_DEFAULT_LINE_THICKNESS
  89. )
  90. x1, y1, x2, y2 = int(bbox[0]), int(bbox[1]), int(bbox[2]), int(bbox[3])
  91. if category in LAYOUT_HIGHLIGHT_CATEGORIES:
  92. cv2.rectangle(
  93. vis, (x1, y1), (x2, y2),
  94. LAYOUT_HIGHLIGHT_OUTLINE_BGR,
  95. thickness + 2,
  96. )
  97. cv2.rectangle(vis, (x1, y1), (x2, y2), color, thickness)
  98. label = category
  99. confidence = result.get('confidence', result.get('score', 0))
  100. if confidence:
  101. label += f":{float(confidence):.2f}"
  102. font = cv2.FONT_HERSHEY_SIMPLEX
  103. font_scale = 0.5 if category in LAYOUT_HIGHLIGHT_CATEGORIES else 0.4
  104. text_thickness = 1
  105. (text_width, text_height), baseline = cv2.getTextSize(
  106. label, font, font_scale, text_thickness
  107. )
  108. text_y = max(y1 - baseline - 1, text_height + baseline)
  109. cv2.rectangle(
  110. vis,
  111. (x1, text_y - text_height - baseline - 2),
  112. (x1 + text_width, text_y),
  113. color,
  114. -1,
  115. )
  116. cv2.putText(
  117. vis, label, (x1, text_y - baseline - 1),
  118. font, font_scale, (255, 255, 255), text_thickness,
  119. )
  120. return vis
  121. def _draw_dashed_segment(
  122. vis: np.ndarray,
  123. p1: np.ndarray,
  124. p2: np.ndarray,
  125. color: tuple,
  126. thickness: int,
  127. *,
  128. dash_length: int = OCR_BOX_DASH_LENGTH,
  129. gap_length: int = OCR_BOX_DASH_GAP,
  130. ) -> None:
  131. """在 p1→p2 上绘制虚线段。"""
  132. start = p1.astype(np.float64)
  133. end = p2.astype(np.float64)
  134. vec = end - start
  135. length = float(np.linalg.norm(vec))
  136. if length < 1e-6:
  137. return
  138. direction = vec / length
  139. pos = 0.0
  140. draw = True
  141. while pos < length:
  142. seg = float(dash_length if draw else gap_length)
  143. seg_end = min(pos + seg, length)
  144. if draw:
  145. s = (start + direction * pos).astype(np.int32)
  146. e = (start + direction * seg_end).astype(np.int32)
  147. cv2.line(
  148. vis,
  149. (int(s[0]), int(s[1])),
  150. (int(e[0]), int(e[1])),
  151. color,
  152. thickness,
  153. cv2.LINE_AA,
  154. )
  155. pos = seg_end
  156. draw = not draw
  157. def _draw_span_outline(
  158. vis: np.ndarray,
  159. pts: np.ndarray,
  160. color: tuple,
  161. thickness: int,
  162. *,
  163. dashed: bool,
  164. ) -> None:
  165. n = len(pts)
  166. if n < 2:
  167. return
  168. for i in range(n):
  169. p1 = pts[i]
  170. p2 = pts[(i + 1) % n]
  171. if dashed:
  172. _draw_dashed_segment(vis, p1, p2, color, thickness)
  173. else:
  174. cv2.line(
  175. vis,
  176. (int(p1[0]), int(p1[1])),
  177. (int(p2[0]), int(p2[1])),
  178. color,
  179. thickness,
  180. cv2.LINE_AA,
  181. )
  182. def draw_ocr_spans_cv2(
  183. image: Union[np.ndarray, Image.Image],
  184. spans: List[Dict[str, Any]],
  185. *,
  186. max_label_chars: int = 12,
  187. ) -> np.ndarray:
  188. """在 BGR 图像上绘制 OCR span(poly 或 bbox);无文字用虚线框。
  189. span 可带 category='seal' 使用印章专用亮橙色,否则使用亮蓝。
  190. """
  191. vis = _to_bgr(image)
  192. for span in spans:
  193. poly = span.get('poly')
  194. bbox = span.get('bbox', [])
  195. pts = None
  196. if poly and len(poly) >= 4:
  197. pts = np.array(poly, dtype=np.int32).reshape(-1, 2)
  198. elif bbox and len(bbox) >= 4:
  199. x0, y0, x1, y1 = map(int, bbox[:4])
  200. pts = np.array(
  201. [[x0, y0], [x1, y0], [x1, y1], [x0, y1]], dtype=np.int32
  202. )
  203. if pts is not None:
  204. text_raw = str(span.get('text', '') or '').strip()
  205. color = _seal_ocr_box_color_bgr() if span.get('category') == 'seal' else _ocr_box_color_bgr()
  206. _draw_span_outline(
  207. vis,
  208. pts,
  209. color,
  210. OCR_BOX_LINE_THICKNESS,
  211. dashed=not text_raw,
  212. )
  213. text = str(span.get('text', '')).strip()[:max_label_chars]
  214. if text and pts is not None:
  215. color = _seal_ocr_box_color_bgr() if span.get('category') == 'seal' else _ocr_box_color_bgr()
  216. x, y = int(pts[0][0]), int(pts[0][1])
  217. cv2.putText(
  218. vis, text, (x, max(y - 2, 10)),
  219. cv2.FONT_HERSHEY_SIMPLEX, 0.35, color, 1, cv2.LINE_AA,
  220. )
  221. return vis
  222. def save_layout_debug(
  223. image: Union[np.ndarray, Image.Image],
  224. layout_results: List[Dict[str, Any]],
  225. output_dir: Union[str, Path],
  226. page_name: str,
  227. *,
  228. suffix: str = 'raw',
  229. subdir: str = 'layout_detection',
  230. image_format: str = 'jpg',
  231. save_json: bool = True,
  232. ) -> Optional[Dict[str, str]]:
  233. """保存 layout 模块 debug 图与 JSON。"""
  234. if not layout_results or not output_dir:
  235. return None
  236. try:
  237. fmt = (image_format or 'jpg').lstrip('.')
  238. debug_dir = resolve_module_debug_dir(output_dir, subdir)
  239. vis = draw_layout_boxes_cv2(image, layout_results)
  240. img_path = debug_dir / f'{page_name}_layout_{suffix}.{fmt}'
  241. cv2.imwrite(str(img_path), vis)
  242. paths: Dict[str, str] = {'image': str(img_path)}
  243. logger.info(f"Saved layout detection image ({suffix}): {img_path}")
  244. if save_json:
  245. json_data = {
  246. 'page_name': page_name,
  247. 'suffix': suffix,
  248. 'count': len(layout_results),
  249. 'results': [
  250. {
  251. 'category': r.get('category'),
  252. 'bbox': r.get('bbox'),
  253. 'confidence': r.get('confidence', r.get('score', 0.0)),
  254. }
  255. for r in layout_results
  256. ],
  257. }
  258. json_path = debug_dir / f'{page_name}_layout_{suffix}.json'
  259. json_path.write_text(
  260. json.dumps(json_data, ensure_ascii=False, indent=2),
  261. encoding='utf-8',
  262. )
  263. paths['json'] = str(json_path)
  264. logger.info(f"Saved layout detection JSON ({suffix}): {json_path}")
  265. return paths
  266. except Exception as e:
  267. logger.warning(f"Failed to save layout debug ({suffix}): {e}")
  268. return None
  269. def save_ocr_debug(
  270. image: Union[np.ndarray, Image.Image],
  271. spans: List[Dict[str, Any]],
  272. output_dir: Union[str, Path],
  273. page_name: str,
  274. *,
  275. subdir: str = 'ocr_recognition',
  276. image_format: str = 'png',
  277. save_json: bool = True,
  278. ) -> Optional[Dict[str, str]]:
  279. """保存 OCR 模块 debug 图与 JSON。"""
  280. if not output_dir:
  281. return None
  282. try:
  283. fmt = (image_format or 'png').lstrip('.')
  284. debug_dir = resolve_module_debug_dir(output_dir, subdir)
  285. vis = draw_ocr_spans_cv2(image, spans or [])
  286. img_path = debug_dir / f'{page_name}_ocr_spans.{fmt}'
  287. cv2.imwrite(str(img_path), vis)
  288. paths: Dict[str, str] = {'image': str(img_path)}
  289. logger.info(f"Saved OCR debug image: {img_path}")
  290. if save_json:
  291. json_data = {
  292. 'page_name': page_name,
  293. 'count': len(spans or []),
  294. 'spans': [
  295. {
  296. 'bbox': s.get('bbox'),
  297. 'poly': s.get('poly'),
  298. 'text': s.get('text'),
  299. 'confidence': s.get('confidence'),
  300. }
  301. for s in (spans or [])
  302. ],
  303. }
  304. json_path = debug_dir / f'{page_name}_ocr_spans.json'
  305. json_path.write_text(
  306. json.dumps(json_data, ensure_ascii=False, indent=2),
  307. encoding='utf-8',
  308. )
  309. paths['json'] = str(json_path)
  310. logger.info(f"Saved OCR debug JSON: {json_path}")
  311. return paths
  312. except Exception as e:
  313. logger.warning(f"Failed to save OCR debug: {e}")
  314. return None