from __future__ import annotations import copy import json import itertools import math import os import re import shutil import subprocess import sys import tempfile import threading import time import uuid from collections import Counter from pathlib import Path from typing import Any import cv2 import gradio as gr import pandas as pd import numpy as np import fastapi import starlette print( "【系統版本】" f"gradio={gr.__version__}, " f"fastapi={fastapi.__version__}, " f"starlette={starlette.__version__}" ) # ======================================================= # Gradio 4.44.x OpenAPI schema compatibility patch # ======================================================= try: import gradio_client.utils _orig_get_type = gradio_client.utils.get_type _orig_json_schema_to_python_type = ( gradio_client.utils._json_schema_to_python_type ) def _patched_get_type(schema: Any, *args: Any, **kwargs: Any) -> Any: if isinstance(schema, bool): return "Any" return _orig_get_type(schema, *args, **kwargs) def _patched_json_schema_to_python_type( schema: Any, *args: Any, **kwargs: Any ) -> Any: if isinstance(schema, bool): return "Any" return _orig_json_schema_to_python_type(schema, *args, **kwargs) gradio_client.utils.get_type = _patched_get_type gradio_client.utils._json_schema_to_python_type = ( _patched_json_schema_to_python_type ) print("【系統提示】Gradio Schema 容錯補丁已套用。") except Exception as patch_error: print(f"【系統提示】Gradio 補丁未套用：{patch_error}") # ======================================================= # Paths and repository settings # ======================================================= ROOT_DIR = Path(__file__).resolve().parent OMR_DIR = ROOT_DIR / "OMRChecker" MOCK_DIR = ROOT_DIR / "mock_libs" GENERATED_DIR = ROOT_DIR / "generated" MARKER_CONFIG_PATH = ROOT_DIR / "marker_config.json" MARKER_REFERENCE_PATH = ROOT_DIR / "marker_reference.png" OMR_REPO_URL = "https://github.com/Udayraj123/OMRChecker.git" OMR_REPO_BRANCH = "master" # OMRChecker is CPU-heavy and writes multiple files. Serialize requests so # two students cannot overwrite each other's temporary outputs. PROCESS_LOCK = threading.Lock() REPO_LOCK = threading.Lock() ALLOWED_ANSWERS = {"A", "B", "C", "D"} SENSITIVITY_PROFILES = [ { "name": "Level 1：標準掃描", "shrink_w": 0, "shrink_h": 0, "levels_high": 0.90, }, { "name": "Level 2：增強靈敏度", "shrink_w": 12, "shrink_h": 8, "levels_high": 0.75, }, { "name": "Level 3：淡筆跡模式", "shrink_w": 20, "shrink_h": 14, "levels_high": 0.55, }, ] def _prepare_headless_helpers() -> None: """Create modules injected into the OMRChecker subprocess. OMRChecker imports screeninfo and may call OpenCV GUI functions. Hugging Face Spaces is headless, so both behaviours need safe fallbacks. """ MOCK_DIR.mkdir(parents=True, exist_ok=True) (MOCK_DIR / "screeninfo.py").write_text( """ class ScreenInfoError(Exception): pass class FakeMonitor: width = 1920 height = 1080 def get_monitors(): return [FakeMonitor()] """.strip() + "\n", encoding="utf-8", ) # Python imports sitecustomize automatically when it is available on # PYTHONPATH. This prevents cv2.imshow/namedWindow from crashing in the # headless Space even if a debug path is accidentally reached. (MOCK_DIR / "sitecustomize.py").write_text( """ try: import cv2 cv2.imshow = lambda *args, **kwargs: None cv2.namedWindow = lambda *args, **kwargs: None cv2.moveWindow = lambda *args, **kwargs: None cv2.destroyAllWindows = lambda *args, **kwargs: None cv2.getWindowProperty = lambda *args, **kwargs: 1.0 cv2.waitKey = lambda *args, **kwargs: ord('q') except Exception: pass """.strip() + "\n", encoding="utf-8", ) _prepare_headless_helpers() GENERATED_DIR.mkdir(parents=True, exist_ok=True) def _valid_omr_checkout(path: Path) -> bool: return ( path.is_dir() and (path / "main.py").is_file() and (path / "src").is_dir() ) def ensure_omrchecker() -> None: """Clone OMRChecker lazily and recover from incomplete checkouts. Dependencies are intentionally *not* installed here. Hugging Face installs requirements.txt during the build stage; runtime pip installation caused the original Space crash. """ if _valid_omr_checkout(OMR_DIR): return with REPO_LOCK: if _valid_omr_checkout(OMR_DIR): return if OMR_DIR.exists(): shutil.rmtree(OMR_DIR, ignore_errors=True) temporary_checkout = ROOT_DIR / f"OMRChecker.clone.{uuid.uuid4().hex}" command = [ "git", "clone", "--depth", "1", "--branch", OMR_REPO_BRANCH, OMR_REPO_URL, str(temporary_checkout), ] result = subprocess.run( command, cwd=ROOT_DIR, capture_output=True, text=True, timeout=180, ) if result.returncode != 0: shutil.rmtree(temporary_checkout, ignore_errors=True) raise RuntimeError( "未能下載 OMRChecker 核心。\n" + (result.stderr or result.stdout or "Git clone failed.")[-2000:] ) if not _valid_omr_checkout(temporary_checkout): shutil.rmtree(temporary_checkout, ignore_errors=True) raise RuntimeError( "OMRChecker 下載完成，但缺少 main.py 或 src 目錄。" ) temporary_checkout.replace(OMR_DIR) print("【系統提示】OMRChecker 核心已準備完成。") def _find_local_file(prefix: str, extensions: set[str]) -> Path | None: for path in sorted(ROOT_DIR.iterdir()): if ( path.is_file() and path.name.lower().startswith(prefix.lower()) and path.suffix.lower() in extensions ): return path return None def _load_template_content(template_content: str | None) -> tuple[dict[str, Any], Path]: if template_content and str(template_content).strip(): try: return json.loads(str(template_content)), ROOT_DIR except json.JSONDecodeError as error: raise ValueError(f"Template JSON 格式不正確：{error}") from error template_path = _find_local_file("template", {".json"}) if template_path is None: raise FileNotFoundError( "找不到 template.json。請把 Template JSON 放在 Space 根目錄。" ) try: return json.loads(template_path.read_text(encoding="utf-8")), template_path.parent except json.JSONDecodeError as error: raise ValueError(f"{template_path.name} JSON 格式不正確：{error}") from error def _load_marker_config() -> dict[str, Any] | None: """Load the optional four-corner marker configuration. The Space can still run without marker_config.json, in which case the original OMRChecker FeatureBasedAlignment path is used as a fallback. """ if not MARKER_CONFIG_PATH.is_file(): return None try: data = json.loads(MARKER_CONFIG_PATH.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: raise ValueError(f"marker_config.json 無法讀取：{error}") from error dimensions = data.get("pageDimensions") markers = data.get("largeCornerMarkers") if ( not isinstance(dimensions, list) or len(dimensions) != 2 or not isinstance(markers, dict) ): raise ValueError("marker_config.json 缺少 pageDimensions 或 largeCornerMarkers。") required = {"top_left", "top_right", "bottom_left", "bottom_right"} if not required.issubset(markers): raise ValueError("marker_config.json 的四個大型定位點不完整。") return data def _resize_for_detection(image: np.ndarray, max_dimension: int = 1800) -> tuple[np.ndarray, float]: """Downscale only for marker detection; return scale back to original.""" height, width = image.shape[:2] longest = max(height, width) if longest <= max_dimension: return image.copy(), 1.0 ratio = max_dimension / float(longest) resized = cv2.resize( image, (max(1, int(round(width * ratio))), max(1, int(round(height * ratio)))), interpolation=cv2.INTER_AREA, ) return resized, 1.0 / ratio class MarkerDetectionError(RuntimeError): """Marker failure carrying an optional candidate-debug image.""" def __init__(self, message: str, debug_image: np.ndarray | None = None): super().__init__(message) self.debug_image = debug_image def _hierarchy_max_depth(hierarchy_row: np.ndarray | None, index: int) -> int: if hierarchy_row is None: return 0 def walk(node: int, visited: set[int]) -> int: if node < 0 or node in visited: return 0 visited = set(visited) visited.add(node) child = int(hierarchy_row[node][2]) best = 0 while child >= 0: best = max(best, 1 + walk(child, visited)) child = int(hierarchy_row[child][0]) return best return walk(index, set()) def _square_marker_candidates( image: np.ndarray, marker_config: dict[str, Any] ) -> list[dict[str, Any]]: """Find nested square candidates and reject small five-question marks. V3 relies on the contour hierarchy of the five-layer marker. Text glyphs, answer fills and the smaller group markers should not survive the combined depth, size and aspect-ratio filters. """ detection_image, scale_back = _resize_for_detection(image) gray = cv2.cvtColor(detection_image, cv2.COLOR_BGR2GRAY) gray = cv2.createCLAHE(clipLimit=2.2, tileGridSize=(8, 8)).apply(gray) blurred = cv2.GaussianBlur(gray, (5, 5), 0) _, binary = cv2.threshold( blurred, 0, 255, cv2.THRESH_BINARY_INV + cv2.THRESH_OTSU ) contours, hierarchy = cv2.findContours( binary, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE ) hierarchy_row = hierarchy[0] if hierarchy is not None else None height, width = gray.shape[:2] page_width, page_height = map(float, marker_config['pageDimensions']) outer_size = float(marker_config.get('largeMarkerOuterSize', 144)) options = marker_config.get('detection') or {} min_depth = int(options.get('minNestedDepth', 3)) side_range = options.get('candidateSideScaleRange', [0.38, 2.50]) scale_x = width / page_width scale_y = height / page_height expected_side = outer_size * math.sqrt(max(scale_x * scale_y, 1e-9)) min_side = max(12.0, expected_side * float(side_range[0])) max_side = expected_side * float(side_range[1]) raw: list[dict[str, Any]] = [] for index, contour in enumerate(contours): x, y, box_width, box_height = cv2.boundingRect(contour) side = (box_width + box_height) / 2.0 if side < min_side or side > max_side: continue aspect = box_width / max(float(box_height), 1.0) if not 0.76 <= aspect <= 1.24: continue perimeter = cv2.arcLength(contour, True) if perimeter <= 0: continue polygon = cv2.approxPolyDP(contour, 0.035 * perimeter, True) if len(polygon) < 4 or len(polygon) > 8: continue depth = _hierarchy_max_depth(hierarchy_row, index) if depth < min_depth: continue area = float(cv2.contourArea(contour)) fill_ratio = area / max(float(box_width * box_height), 1.0) if fill_ratio < 0.55: continue center_x = (x + box_width / 2.0) * scale_back center_y = (y + box_height / 2.0) * scale_back raw.append({ 'id': index, 'center': (center_x, center_y), 'side': side * scale_back, 'area': area * scale_back * scale_back, 'depth': depth, 'aspect': aspect, 'rect': ( int(round(x * scale_back)), int(round(y * scale_back)), int(round(box_width * scale_back)), int(round(box_height * scale_back)), ), }) # The same physical marker can create several nested candidate contours. # Keep the largest candidate around each center. raw.sort(key=lambda item: (-item['side'], -item['depth'])) candidates: list[dict[str, Any]] = [] for candidate in raw: cx, cy = candidate['center'] duplicate = False for kept in candidates: kx, ky = kept['center'] distance = math.hypot(cx - kx, cy - ky) if distance < 0.22 * max(candidate['side'], kept['side']): duplicate = True break if not duplicate: candidates.append(candidate) return candidates def _marker_candidate_debug( image: np.ndarray, candidates: list[dict[str, Any]] ) -> np.ndarray: debug = image.copy() for candidate in candidates: x, y, w, h = candidate['rect'] cv2.rectangle(debug, (x, y), (x + w, y + h), (255, 80, 0), 3) cv2.putText( debug, f"d{candidate['depth']} s{int(candidate['side'])}", (x, max(18, y - 7)), cv2.FONT_HERSHEY_SIMPLEX, 0.52, (180, 40, 0), 2, cv2.LINE_AA, ) return debug def _quadrilateral_geometry_score( points: np.ndarray, marker_sides: list[float], image_width: int, image_height: int, options: dict[str, Any], ) -> float | None: tl, tr, br, bl = points if not (tl[0] < tr[0] and bl[0] < br[0] and tl[1] < bl[1] and tr[1] < br[1]): return None if not cv2.isContourConvex(points.reshape(-1, 1, 2).astype(np.float32)): return None def distance(a: np.ndarray, b: np.ndarray) -> float: return float(np.linalg.norm(a - b)) top = distance(tl, tr) bottom = distance(bl, br) left = distance(tl, bl) right = distance(tr, br) diag1 = distance(tl, br) diag2 = distance(tr, bl) polygon_area = abs(float(cv2.contourArea(points.reshape(-1, 1, 2)))) area_ratio = polygon_area / max(float(image_width * image_height), 1.0) if area_ratio < float(options.get('minQuadrilateralAreaRatio', 0.20)): return None if min(top, bottom) < image_width * float(options.get('minHorizontalSpanRatio', 0.48)): return None if min(left, right) < image_height * float(options.get('minVerticalSpanRatio', 0.48)): return None def in_range(value: float, range_value: list[float]) -> bool: return float(range_value[0]) <= value <= float(range_value[1]) width_ratio = top / max(bottom, 1.0) height_ratio = left / max(right, 1.0) diagonal_ratio = diag1 / max(diag2, 1.0) if not in_range(width_ratio, options.get('topBottomWidthRatioRange', [0.48, 1.85])): return None if not in_range(height_ratio, options.get('leftRightHeightRatioRange', [0.48, 1.85])): return None if not in_range(diagonal_ratio, options.get('diagonalRatioRange', [0.55, 1.80])): return None size_mean = float(np.mean(marker_sides)) size_cv = float(np.std(marker_sides) / max(size_mean, 1e-6)) if size_cv > float(options.get('maxMarkerSizeCv', 0.42)): return None # Lower is better. Log penalties treat reciprocal distortion symmetrically. return ( abs(math.log(max(width_ratio, 1e-6))) + abs(math.log(max(height_ratio, 1e-6))) + 0.55 * abs(math.log(max(diagonal_ratio, 1e-6))) + 1.4 * size_cv - 0.20 * area_ratio ) def _corner_regions() -> dict[str, Any]: """Normalized safe-search regions for the four main markers.""" return { 'top_left': lambda nx, ny: 0.08 <= nx <= 0.48 and 0.18 <= ny <= 0.53, 'top_right': lambda nx, ny: 0.68 <= nx <= 1.02 and 0.18 <= ny <= 0.53, 'bottom_right': lambda nx, ny: 0.68 <= nx <= 1.02 and 0.72 <= ny <= 1.02, 'bottom_left': lambda nx, ny: 0.08 <= nx <= 0.48 and 0.72 <= ny <= 1.02, } def _build_corner_pools( candidates: list[dict[str, Any]], image_width: int, image_height: int, marker_config: dict[str, Any], ) -> dict[str, list[tuple[float, dict[str, Any]]]]: page_width, page_height = map(float, marker_config['pageDimensions']) expected_raw = marker_config['largeCornerMarkers'] options = marker_config.get('detection') or {} max_per_corner = int(options.get('maxCandidatesPerCorner', 8)) expected = { name: (point[0] / page_width, point[1] / page_height) for name, point in expected_raw.items() } regions = _corner_regions() expected_side = float(marker_config.get('largeMarkerOuterSize', 144)) * math.sqrt( max((image_width / page_width) * (image_height / page_height), 1e-9) ) pools: dict[str, list[tuple[float, dict[str, Any]]]] = {} for marker_name in ('top_left', 'top_right', 'bottom_right', 'bottom_left'): ex, ey = expected[marker_name] scored: list[tuple[float, dict[str, Any]]] = [] for candidate in candidates: cx, cy = candidate['center'] nx, ny = cx / image_width, cy / image_height if not regions[marker_name](nx, ny): continue distance_score = math.hypot(nx - ex, ny - ey) size_score = abs( math.log(max(candidate['side'] / max(expected_side, 1e-6), 1e-6)) ) depth_bonus = -0.015 * max(0, int(candidate.get('depth', 3)) - 3) rescue_penalty = 0.025 if candidate.get('source') == 'local_rescue' else 0.0 match_bonus = -0.035 * max(0.0, float(candidate.get('match_score', 0.0)) - 0.40) scored.append(( distance_score + 0.20 * size_score + depth_bonus + rescue_penalty + match_bonus, candidate, )) scored.sort(key=lambda item: item[0]) pools[marker_name] = scored[:max_per_corner] return pools def _select_best_marker_set( pools: dict[str, list[tuple[float, dict[str, Any]]]], image_width: int, image_height: int, options: dict[str, Any], ) -> tuple[float, tuple[dict[str, Any], ...]] | None: ordered_names = ('top_left', 'top_right', 'bottom_right', 'bottom_left') if any(not pools.get(name) for name in ordered_names): return None best: tuple[float, tuple[dict[str, Any], ...]] | None = None for combination in itertools.product(*(pools[name] for name in ordered_names)): local_scores = [item[0] for item in combination] selected = tuple(item[1] for item in combination) if len({str(item['id']) for item in selected}) != 4: continue points = np.array([item['center'] for item in selected], dtype=np.float32) geometry_score = _quadrilateral_geometry_score( points, [float(item['side']) for item in selected], image_width, image_height, options, ) if geometry_score is None: continue total_score = sum(local_scores) + geometry_score if best is None or total_score < best[0]: best = (total_score, selected) return best def _load_marker_reference() -> np.ndarray: reference = cv2.imread(str(MARKER_REFERENCE_PATH), cv2.IMREAD_GRAYSCALE) if reference is None: raise MarkerDetectionError( 'V3.1 局部補救需要 marker_reference.png，但專案根目錄找不到或無法讀取。' ) return reference def _local_template_rescue_candidate( image: np.ndarray, marker_name: str, marker_config: dict[str, Any], synthetic_id: str, ) -> dict[str, Any] | None: """Search one corner ROI using local contrast, multi-threshold and edge matching. This path is deliberately only used after the strict V3 contour detector cannot produce a valid four-marker set. It handles the common case where a phone/hand shadow darkens the white rings and collapses contour hierarchy. """ local_options = marker_config.get('localRescue') or {} page_width, page_height = map(float, marker_config['pageDimensions']) image_height, image_width = image.shape[:2] expected_point = marker_config['largeCornerMarkers'][marker_name] expected_x = expected_point[0] / page_width * image_width expected_y = expected_point[1] / page_height * image_height half_width = image_width * float(local_options.get('roiHalfWidthRatio', 0.22)) half_height = image_height * float(local_options.get('roiHalfHeightRatio', 0.20)) x0 = max(0, int(round(expected_x - half_width))) x1 = min(image_width, int(round(expected_x + half_width))) y0 = max(0, int(round(expected_y - half_height))) y1 = min(image_height, int(round(expected_y + half_height))) if x1 - x0 < 40 or y1 - y0 < 40: return None roi = image[y0:y1, x0:x1] gray = cv2.cvtColor(roi, cv2.COLOR_BGR2GRAY) gray = cv2.createCLAHE(clipLimit=2.5, tileGridSize=(8, 8)).apply(gray) gray = cv2.GaussianBlur(gray, (3, 3), 0) _, local_otsu = cv2.threshold( gray, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU ) block_size = int(local_options.get('adaptiveBlockSize', 31)) if block_size % 2 == 0: block_size += 1 block_size = max(15, block_size) adaptive = cv2.adaptiveThreshold( gray, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY, block_size, float(local_options.get('adaptiveC', 7)), ) edges = cv2.Canny(gray, 50, 150) reference = _load_marker_reference() expected_side = float(marker_config.get('largeMarkerOuterSize', 144)) * math.sqrt( max((image_width / page_width) * (image_height / page_height), 1e-9) ) scales = local_options.get( 'templateMatchScales', [0.68, 0.78, 0.88, 0.98, 1.08, 1.18, 1.28] ) angles = local_options.get('templateMatchAngles', [-10, 0, 10]) threshold = float(local_options.get('templateMatchThreshold', 0.40)) best: tuple[float, dict[str, Any]] | None = None for scale in scales: side = max(24, int(round(expected_side * float(scale)))) if side >= roi.shape[0] or side >= roi.shape[1]: continue interpolation = cv2.INTER_AREA if side < reference.shape[0] else cv2.INTER_CUBIC resized_reference = cv2.resize(reference, (side, side), interpolation=interpolation) for angle in angles: matrix = cv2.getRotationMatrix2D((side / 2.0, side / 2.0), float(angle), 1.0) template = cv2.warpAffine( resized_reference, matrix, (side, side), flags=cv2.INTER_LINEAR, borderMode=cv2.BORDER_CONSTANT, borderValue=255, ) if gray.shape[0] < side or gray.shape[1] < side: continue gray_result = cv2.matchTemplate(gray, template, cv2.TM_CCOEFF_NORMED) _, gray_score, _, location = cv2.minMaxLoc(gray_result) lx, ly = location def score_at(source: np.ndarray, target: np.ndarray) -> float: result = cv2.matchTemplate(source, target, cv2.TM_CCOEFF_NORMED) if ly >= result.shape[0] or lx >= result.shape[1]: return -1.0 return float(result[ly, lx]) otsu_score = score_at(local_otsu, template) adaptive_score = score_at(adaptive, template) edge_template = cv2.Canny(template, 50, 150) edge_score = score_at(edges, edge_template) combined_score = ( 0.62 * float(gray_score) + 0.16 * otsu_score + 0.12 * adaptive_score + 0.10 * edge_score ) if best is None or combined_score > best[0]: center_x = x0 + lx + side / 2.0 center_y = y0 + ly + side / 2.0 candidate = { 'id': synthetic_id, 'center': (center_x, center_y), # Use the physically expected outer size for cross-corner # size consistency; the matched inner/outer crop may be smaller. 'side': expected_side, 'area': expected_side * expected_side, 'depth': int(marker_config.get('largeMarkerNestedLevels', 5)), 'aspect': 1.0, 'rect': ( int(round(center_x - side / 2.0)), int(round(center_y - side / 2.0)), side, side, ), 'source': 'local_rescue', 'match_score': combined_score, 'gray_score': float(gray_score), 'otsu_score': otsu_score, 'adaptive_score': adaptive_score, 'edge_score': edge_score, } best = (combined_score, candidate) if best is None or best[0] < threshold: return None return best[1] def _merge_marker_candidates( base: list[dict[str, Any]], additions: list[dict[str, Any]] ) -> list[dict[str, Any]]: merged = list(base) for candidate in additions: cx, cy = candidate['center'] duplicate = False for kept in merged: kx, ky = kept['center'] if math.hypot(cx - kx, cy - ky) < 0.28 * max( float(candidate['side']), float(kept['side']) ): duplicate = True break if not duplicate: merged.append(candidate) return merged def _detect_large_corner_markers( image: np.ndarray, marker_config: dict[str, Any] ) -> tuple[np.ndarray, np.ndarray, bool]: """Detect V3 markers, then use V3.1 local rescue only when necessary.""" candidates = _square_marker_candidates(image, marker_config) image_height, image_width = image.shape[:2] options = marker_config.get('detection') or {} pools = _build_corner_pools(candidates, image_width, image_height, marker_config) best = _select_best_marker_set( pools, image_width, image_height, options ) rescue_used = False local_options = marker_config.get('localRescue') or {} rescue_enabled = bool(local_options.get('enabled', True)) minimum_global = int(local_options.get('minimumGlobalCandidates', 2)) if best is None and rescue_enabled and len(candidates) >= minimum_global: missing_names = [ name for name in ('top_left', 'top_right', 'bottom_right', 'bottom_left') if not pools.get(name) ] # If each region has a candidate but the set fails geometry, retest all # four ROIs. Otherwise only search the empty corner(s), limiting work. search_names = missing_names or [ 'top_left', 'top_right', 'bottom_right', 'bottom_left' ] max_rescue_corners = int(local_options.get('maxRescueCorners', 2)) if len(search_names) <= max_rescue_corners or not missing_names: rescued: list[dict[str, Any]] = [] for index, marker_name in enumerate(search_names): candidate = _local_template_rescue_candidate( image, marker_name, marker_config, synthetic_id=f"rescue-{marker_name}-{index}", ) if candidate is not None: rescued.append(candidate) if rescued: candidates = _merge_marker_candidates(candidates, rescued) pools = _build_corner_pools( candidates, image_width, image_height, marker_config ) best = _select_best_marker_set( pools, image_width, image_height, options ) rescue_used = best is not None and any( item.get('source') == 'local_rescue' for item in best[1] ) candidate_debug = _marker_candidate_debug(image, candidates) if best is None: empty_regions = [ name for name in ('top_left', 'top_right', 'bottom_right', 'bottom_left') if not pools.get(name) ] if len(candidates) < 4: message = ( f"只找到 {len(candidates)} 個符合 V3/V3.1 結構、尺寸或局部相似度的候選。" ) elif empty_regions: message = "以下角點搜尋區沒有可靠標記：" + ", ".join(empty_regions) else: message = ( '候選標記存在，但沒有任何四點組合通過面積、跨度、比例及尺寸一致性檢查。' ) raise MarkerDetectionError(message, candidate_debug) selected = best[1] points = np.array([item['center'] for item in selected], dtype=np.float32) debug_image = candidate_debug labels = ('TL', 'TR', 'BR', 'BL') for label, candidate, (point_x, point_y) in zip(labels, selected, points): x, y, w, h = candidate['rect'] color = (0, 215, 255) if candidate.get('source') == 'local_rescue' else (0, 255, 0) cv2.rectangle(debug_image, (x, y), (x + w, y + h), color, 6) position = (int(round(point_x)), int(round(point_y))) cv2.circle(debug_image, position, 18, color, 5) suffix = '*' if candidate.get('source') == 'local_rescue' else '' cv2.putText( debug_image, label + suffix, (position[0] + 18, position[1] - 18), cv2.FONT_HERSHEY_SIMPLEX, 1.0, color, 3, cv2.LINE_AA, ) if rescue_used: cv2.putText( debug_image, 'V3.1 LOCAL SHADOW RESCUE', (28, 48), cv2.FONT_HERSHEY_SIMPLEX, 1.05, (0, 165, 255), 3, cv2.LINE_AA, ) return points, debug_image, rescue_used def _warp_with_large_markers( image: np.ndarray, marker_config: dict[str, Any] ) -> tuple[np.ndarray, np.ndarray, bool]: source_points, debug_image, rescue_used = _detect_large_corner_markers( image, marker_config ) page_width, page_height = map(int, marker_config['pageDimensions']) marker_points = marker_config['largeCornerMarkers'] destination_points = np.array( [ marker_points['top_left'], marker_points['top_right'], marker_points['bottom_right'], marker_points['bottom_left'], ], dtype=np.float32, ) transform = cv2.getPerspectiveTransform(source_points, destination_points) warped = cv2.warpPerspective( image, transform, (page_width, page_height), flags=cv2.INTER_CUBIC, borderMode=cv2.BORDER_CONSTANT, borderValue=(255, 255, 255), ) return warped, debug_image, rescue_used def _expand_question_label(label: Any) -> list[int]: text = str(label).strip() range_match = re.fullmatch(r"q(\d+)\.\.(\d+)", text, flags=re.IGNORECASE) if range_match: start, end = map(int, range_match.groups()) step = 1 if end >= start else -1 return list(range(start, end + step, step)) single_match = re.fullmatch(r"q0*(\d+)", text, flags=re.IGNORECASE) if single_match: return [int(single_match.group(1))] return [] def _question_numbers_from_template(template: dict[str, Any]) -> list[int]: numbers: set[int] = set() for block in (template.get("fieldBlocks") or {}).values(): if not isinstance(block, dict): continue for label in block.get("fieldLabels") or []: numbers.update(_expand_question_label(label)) if not numbers: raise ValueError( "Template JSON 沒有可辨識的 q1、q2 或 q1..100 題號。" ) return sorted(numbers) def _normalize_answer(value: Any) -> str: if value is None or (isinstance(value, float) and pd.isna(value)): return "—" answer = str(value).strip().upper() return answer if answer in ALLOWED_ANSWERS else "—" def _apply_profile(template: dict[str, Any], profile: dict[str, Any]) -> dict[str, Any]: current = copy.deepcopy(template) base_width, base_height = current.get("bubbleDimensions", [60, 30]) current["bubbleDimensions"] = [ max(10, int(base_width) - int(profile["shrink_w"])), max(10, int(base_height) - int(profile["shrink_h"])), ] for processor in current.get("preProcessors") or []: if not isinstance(processor, dict): continue if processor.get("name") == "Levels": processor.setdefault("options", {})["high"] = profile["levels_high"] return current def _remove_feature_alignment(template: dict[str, Any]) -> dict[str, Any]: """Remove FeatureBasedAlignment after successful marker homography. Once the four large markers have already mapped the photo into the fixed 3093 x 4374 template coordinate system, a second feature-based warp is both unnecessary and potentially harmful on a sheet containing many repeated A/B/C/D boxes and grid lines. """ current = copy.deepcopy(template) current["preProcessors"] = [ processor for processor in (current.get("preProcessors") or []) if not ( isinstance(processor, dict) and processor.get("name") == "FeatureBasedAlignment" ) ] return current def _crop_answer_region( image: np.ndarray, marker_config: dict[str, Any], padding: int = 24 ) -> np.ndarray: """Crop the corrected answer area for a clearer UI preview.""" bounds = marker_config.get("answerRegionBounds") or {} height, width = image.shape[:2] left = max(0, int(bounds.get("left", 0)) - padding) top = max(0, int(bounds.get("top", 0)) - padding) right = min(width, int(bounds.get("right", width)) + padding) bottom = min(height, int(bounds.get("bottom", height)) + padding) if right <= left or bottom <= top: return image return image[top:bottom, left:right] def _alignment_reference_names(template: dict[str, Any]) -> set[str]: references: set[str] = set() for processor in template.get("preProcessors") or []: if not isinstance(processor, dict): continue options = processor.get("options") or {} for key in ("reference", "relativePath"): value = options.get(key) if isinstance(value, str) and value.strip(): references.add(Path(value).name) return references def _copy_alignment_assets(template: dict[str, Any], destination: Path) -> None: reference_names = _alignment_reference_names(template) # The provided DSE template uses template.jpg. Also support any explicit # reference filename that exists in the Space root. for reference_name in reference_names: direct = ROOT_DIR / reference_name source = direct if direct.is_file() else None if source is None and reference_name.lower().startswith("template"): source = _find_local_file("template", {".jpg", ".jpeg", ".png"}) if source is None: raise FileNotFoundError( f"Template 需要對齊參考圖 {reference_name}，但 Space 根目錄找不到它。" ) shutil.copy2(source, destination / reference_name) def _write_headless_config(destination: Path) -> None: """Write a deliberately minimal OMRChecker config. OMRChecker's master branch has changed its JSON schema over time. Keeping only the two long-standing image-level keys avoids schema failures caused by optional output keys that are present in some releases but absent in others. If even this minimal config is rejected, the scan loop retries once without a local config.json so OMRChecker can use its own defaults. """ config = { "outputs": { "show_image_level": 0, "save_image_level": 3, } } (destination / "config.json").write_text( json.dumps(config, ensure_ascii=False, indent=2), encoding="utf-8", ) def _find_output_file(output_dir: Path, suffixes: set[str], prefer: str = "") -> Path | None: candidates = [ path for path in output_dir.rglob("*") if path.is_file() and path.suffix.lower() in suffixes ] if not candidates: return None if prefer: preferred = [path for path in candidates if prefer.lower() in str(path).lower()] if preferred: candidates = preferred return max(candidates, key=lambda path: (path.stat().st_mtime, path.stat().st_size)) def _parse_csv_answers(csv_path: Path | None, question_numbers: list[int]) -> dict[int, str]: answers = {number: "—" for number in question_numbers} if csv_path is None or not csv_path.is_file(): return answers try: dataframe = pd.read_csv(csv_path) except Exception: return answers if dataframe.empty: return answers normalized_columns = { str(column).strip().lower(): column for column in dataframe.columns } for number in question_numbers: column = normalized_columns.get(f"q{number}") if column is not None: answers[number] = _normalize_answer(dataframe.iloc[0][column]) return answers def _parse_log_answers(text: str, question_numbers: list[int]) -> dict[int, str]: allowed_numbers = set(question_numbers) answers = {number: "—" for number in question_numbers} patterns = [ r"(?i)\bq(?:uestion)?\s*0*(\d+)\s*[:=]\s*([A-D])\b", r"(?i)\b0*(\d+)\s*[|:]\s*([A-D])\b", ] for pattern in patterns: for question_text, answer in re.findall(pattern, text or ""): number = int(question_text) if number in allowed_numbers: answers[number] = answer.upper() return answers def _merge_answer_sources(primary: dict[int, str], fallback: dict[int, str]) -> dict[int, str]: merged = dict(primary) for number, answer in fallback.items(): if merged.get(number, "—") == "—" and answer != "—": merged[number] = answer return merged def _cleanup_old_generated_files(max_age_seconds: int = 7200) -> None: now = time.time() for path in GENERATED_DIR.iterdir(): try: if now - path.stat().st_mtime > max_age_seconds: if path.is_dir(): shutil.rmtree(path, ignore_errors=True) else: path.unlink(missing_ok=True) except OSError: pass def _format_answer_summary(final_answers: dict[int, str]) -> str: lines = [ "### 📊 多重採樣投票結果", "```text", ] row: list[str] = [] for number in sorted(final_answers): row.append(f"q{number:02d}: {final_answers[number]}") if len(row) == 4: lines.append(" | ".join(row)) row = [] if row: lines.append(" | ".join(row)) lines.extend( [ "```", "💡 請在前端逐題人工核對；投票結果不是免校對的正式答案。", ] ) return "\n".join(lines) def process_omr(image_file: str | None, template_content: str | None = None): if image_file is None: return None, None, None, None, "錯誤：請先上傳答案卡圖片。" with PROCESS_LOCK: request_root: Path | None = None try: ensure_omrchecker() _cleanup_old_generated_files() image = cv2.imread(str(image_file)) if image is None: raise ValueError("OpenCV 無法讀取上傳圖片。") image_height, image_width = image.shape[:2] dimension_info = f"【圖片載入成功】解析度：{image_width} × {image_height} px" base_template, _ = _load_template_content(template_content) question_numbers = _question_numbers_from_template(base_template) question_count = len(question_numbers) request_id = uuid.uuid4().hex request_root = Path( tempfile.mkdtemp(prefix=f"hf_omr_{request_id}_", dir=str(ROOT_DIR)) ) result_dir = GENERATED_DIR / request_id result_dir.mkdir(parents=True, exist_ok=True) logs = [ dimension_info, f"【Template】題號範圍：q{question_numbers[0]}–q{question_numbers[-1]}，共 {question_count} 題", ] # ------------------------------------------------------- # Stage 1: marker-based perspective correction. # ------------------------------------------------------- omr_input_path = Path(image_file) marker_debug_path: Path | None = None corrected_preview: Path | None = None marker_config = _load_marker_config() marker_warp_succeeded = False if marker_config is not None: try: warped, marker_debug, rescue_used = _warp_with_large_markers( image, marker_config ) warped_full_path = result_dir / "01_perspective_corrected_full.jpg" corrected_preview = result_dir / "02_corrected_answer_region.jpg" marker_debug_path = result_dir / "00_detected_markers.jpg" cv2.imwrite( str(warped_full_path), warped, [int(cv2.IMWRITE_JPEG_QUALITY), 95], ) cv2.imwrite( str(corrected_preview), _crop_answer_region(warped, marker_config), [int(cv2.IMWRITE_JPEG_QUALITY), 95], ) cv2.imwrite( str(marker_debug_path), marker_debug, [int(cv2.IMWRITE_JPEG_QUALITY), 92], ) omr_input_path = warped_full_path marker_warp_succeeded = True if rescue_used: logs.append("【四角校正】V3.1 局部補救成功：陰影中的缺失角點已恢復，四點幾何檢查通過。") else: logs.append("【四角校正】V3 嚴格模式成功偵測四個大型定位點。") logs.append("【四角校正】答案區已映射到固定座標。") logs.append("【預覽說明】介面只顯示校正後答案區；上半頁不參與答案座標判定。") except Exception as marker_error: fallback_path = result_dir / "01_original_marker_failure.jpg" shutil.copy2(image_file, fallback_path) corrected_preview = fallback_path if isinstance(marker_error, MarkerDetectionError) and marker_error.debug_image is not None: marker_debug_path = result_dir / "00_marker_candidates_failed.jpg" cv2.imwrite( str(marker_debug_path), marker_error.debug_image, [int(cv2.IMWRITE_JPEG_QUALITY), 92], ) strict_mode = bool(marker_config.get("strictMarkerMode", True)) logs.append( "【四角校正】失敗：" f"{type(marker_error).__name__}: {marker_error}" ) if strict_mode: failure_log = ( "### ❌ 四角定位未通過 V3.1 安全檢查\n\n" "系統沒有把不可靠的校正圖交給 OMRChecker。請確保四個大型定位標記完整入鏡、" "相片清晰，並避免手機影子完全遮蔽標記後重新拍攝。\n\n" "========================================\n" "⚙️ 執行摘要\n" "========================================\n" + "\n".join(logs) ) return ( None, str(marker_debug_path) if marker_debug_path else None, str(corrected_preview) if corrected_preview else None, None, failure_log, ) logs.append("【回退模式】strictMarkerMode=false，改用 FeatureBasedAlignment。") else: fallback_path = result_dir / "01_original_no_marker_config.jpg" shutil.copy2(image_file, fallback_path) corrected_preview = fallback_path logs.append("【四角校正】沒有 marker_config.json，使用原始對齊流程。") scans: list[dict[str, Any]] = [] logs.append("--- 開始三輪多重採樣 ---") for index, profile in enumerate(SENSITIVITY_PROFILES, start=1): profile_dir = request_root / f"profile_{index}" images_dir = profile_dir / "images" output_dir = request_root / f"output_{index}" images_dir.mkdir(parents=True, exist_ok=True) output_dir.mkdir(parents=True, exist_ok=True) current_template = _apply_profile(base_template, profile) if marker_warp_succeeded: current_template = _remove_feature_alignment(current_template) (profile_dir / "template.json").write_text( json.dumps(current_template, ensure_ascii=False, indent=2), encoding="utf-8", ) _write_headless_config(profile_dir) _copy_alignment_assets(current_template, profile_dir) shutil.copy2(omr_input_path, images_dir / "sheet.jpg") environment = os.environ.copy() environment["PYTHONPATH"] = ( str(MOCK_DIR) + os.pathsep + environment.get("PYTHONPATH", "") ) environment["OMR_CHECKER_CONTAINER"] = "true" environment["MPLBACKEND"] = "Agg" command = [ sys.executable, "main.py", "-i", str(profile_dir), "-o", str(output_dir), ] result = subprocess.run( command, cwd=OMR_DIR, capture_output=True, text=True, env=environment, timeout=180, ) full_log = (result.stdout or "") + "\n" + (result.stderr or "") # Runtime compatibility fallback: if the checked-out master # branch rejects our minimal config schema, remove config.json # and let OMRChecker load its own built-in defaults. if ( result.returncode != 0 and "Provided config JSON is Invalid" in full_log ): (profile_dir / "config.json").unlink(missing_ok=True) logs.append( f"{profile['name']}：偵測到 OMRChecker config schema 差異，" "已移除本地 config.json 並自動重試。" ) result = subprocess.run( command, cwd=OMR_DIR, capture_output=True, text=True, env=environment, timeout=180, ) full_log = (result.stdout or "") + "\n" + (result.stderr or "") csv_path = _find_output_file(output_dir, {".csv"}) image_path = _find_output_file( output_dir, {".jpg", ".jpeg", ".png"}, prefer="CheckedOMRs", ) csv_answers = _parse_csv_answers(csv_path, question_numbers) log_answers = _parse_log_answers(full_log, question_numbers) answers = _merge_answer_sources(csv_answers, log_answers) valid_count = sum(answer != "—" for answer in answers.values()) logs.append( f"{profile['name']}：returncode={result.returncode}，辨識 {valid_count}/{question_count} 題" ) if result.returncode != 0: logs.append("本輪錯誤摘要：" + full_log[-1800:].replace("\n", " | ")) scans.append( { "profile": profile["name"], "answers": answers, "valid_count": valid_count, "csv": csv_path, "image": image_path, "returncode": result.returncode, } ) successful_scans = [scan for scan in scans if scan["returncode"] == 0] if not successful_scans: failure_log = ( "### ❌ OMRChecker 三輪均未能執行\n\n" "本次沒有產生答案 CSV；這不是 100 題全部留空，而是辨識核心未成功啟動。\n\n" "========================================\n" "⚙️ 執行摘要\n" "========================================\n" + "\n".join(logs) ) return ( None, str(marker_debug_path) if marker_debug_path else None, str(corrected_preview) if corrected_preview else None, None, failure_log, ) final_answers: dict[int, str] = {} for number in question_numbers: votes = [ scan["answers"][number] for scan in successful_scans if scan["answers"][number] != "—" ] if not votes: final_answers[number] = "—" continue vote_counts = Counter(votes) final_answer = vote_counts.most_common(1)[0][0] final_answers[number] = final_answer if len(vote_counts) > 1: logs.append(f"q{number} 分歧：{dict(vote_counts)} → {final_answer}") final_valid_count = sum(answer != "—" for answer in final_answers.values()) logs.append( f"融合後辨識：{final_valid_count}/{question_count} 題；空白 {question_count - final_valid_count} 題。" ) fused_csv = result_dir / "omr_fused_answers.csv" pd.DataFrame( [{f"q{number}": final_answers[number] for number in question_numbers}] ).to_csv(fused_csv, index=False, encoding="utf-8-sig") best_scan = max(successful_scans, key=lambda item: item["valid_count"]) checked_image: Path | None = None if best_scan["image"] and Path(best_scan["image"]).is_file(): source_image = Path(best_scan["image"]) checked_image = result_dir / ("checked_omr" + source_image.suffix.lower()) shutil.copy2(source_image, checked_image) combined_log = ( _format_answer_summary(final_answers) + "\n\n========================================\n" + "⚙️ 執行摘要\n" + "========================================\n" + "\n".join(logs) ) return ( str(fused_csv), str(marker_debug_path) if marker_debug_path else None, str(corrected_preview) if corrected_preview else None, str(checked_image) if checked_image else None, combined_log, ) except subprocess.TimeoutExpired: return None, None, None, None, "錯誤：OMRChecker 單輪處理超過 180 秒，已中止。" except Exception as error: return None, None, None, None, f"錯誤：{type(error).__name__}: {error}" finally: if request_root is not None: shutil.rmtree(request_root, ignore_errors=True) interface = gr.Interface( fn=process_omr, inputs=[ gr.Image(type="filepath", label="1. 上傳已填寫的 OMR 答案卡相片 (JPG/PNG)"), gr.Textbox( lines=6, label="2. [選填] Template JSON 配置", placeholder="留空時自動使用 Space 根目錄的 template.json", ), ], outputs=[ gr.File(label="下載融合辨識結果 (CSV)"), gr.Image(label="四個大型定位點偵測結果"), gr.Image(label="透視校正後答案區"), gr.Image(label="OMRChecker 視覺化劃記檢視"), gr.Textbox(label="辨識答案與執行摘要", lines=22), ], title="DSE OMR V3.1 手機陰影局部補救辨識系統", description=( "先以 V3 嚴格模式偵測四個 144px 五層巢狀大型定位標記；" "若手機陰影令一至兩角輪廓失真，才啟動 V3.1 局部多閾值與模板補救。" "四點通過幾何安全檢查後會停用第二次 FeatureBasedAlignment，" "再以 OMRChecker 執行三輪靈敏度掃描及融合。" "定位未通過幾何安全檢查時會直接拒絕處理；正式提交前仍建議學生核對辨識結果。" ), ) if __name__ == "__main__": interface.launch( server_name=os.getenv("GRADIO_SERVER_NAME", "0.0.0.0"), server_port=int(os.getenv("GRADIO_SERVER_PORT", "7860")), show_error=True, )