import cv2 import numpy as np from dataclasses import dataclass, field from typing import Optional import base64 from itertools import combinations @dataclass class BubbleResult: question_number: int detected_answer: Optional[str] correct_answer: Optional[str] is_correct: bool confidence: float @dataclass class OMRResult: total_questions: int correct_count: int wrong_count: int empty_count: int score: float percentage: float details: list = field(default_factory=list) processed_image_b64: Optional[str] = None error: Optional[str] = None class OMRProcessor: CHOICE_LABELS = ['A', 'B', 'C', 'D', 'E'] def __init__( self, num_questions: int = 50, num_choices: int = 5, num_column_blocks: int = 5, bubble_fill_threshold: float = 0.38, min_bubble_size_ratio: float = 0.015, max_bubble_size_ratio: float = 0.12, debug: bool = False, min_mark_density: float = 0.35, min_mark_coverage_ratio: float = 0.45 ): self.num_questions = num_questions self.num_choices = num_choices self.num_column_blocks = num_column_blocks self.choice_labels = self.CHOICE_LABELS[:num_choices] self.bubble_fill_threshold = bubble_fill_threshold self.min_bubble_size_ratio = min_bubble_size_ratio self.max_bubble_size_ratio = max_bubble_size_ratio self.debug = debug self.questions_per_block = num_questions // num_column_blocks self.min_mark_density = min_mark_density self.min_mark_coverage_ratio = min_mark_coverage_ratio def process( self, image_input, answer_key: dict, return_preview: bool = True ) -> OMRResult: try: image = self._load_image(image_input) image = self._deskew(image) gray, blurred, thresh = self._preprocess(image) warped, warped_thresh = self._detect_answer_sheet(image, gray, thresh) detected_answers, annotated = self._detect_bubbles( warped, warped_thresh, answer_key, return_preview ) result = self._calculate_score(detected_answers, answer_key) if return_preview and annotated is not None: result.processed_image_b64 = self._encode_image(annotated) return result except SheetNotFoundError as e: return OMRResult( total_questions=self.num_questions, correct_count=0, wrong_count=0, empty_count=self.num_questions, score=0.0, percentage=0.0, error=f"Lembar jawaban tidak terdeteksi: {str(e)}" ) except Exception as e: import traceback return OMRResult( total_questions=self.num_questions, correct_count=0, wrong_count=0, empty_count=self.num_questions, score=0.0, percentage=0.0, error=f"Error: {str(e)}\n{traceback.format_exc()}" ) def process_answer_key_image(self, image_input) -> dict: try: image = self._load_image(image_input) image = self._deskew(image) gray, blurred, thresh = self._preprocess(image) warped, warped_thresh = self._detect_answer_sheet(image, gray, thresh) answer_key, _ = self._detect_bubbles(warped, warped_thresh, {}, False) return answer_key except Exception as e: raise ValueError(f"Gagal membaca kunci jawaban: {str(e)}") def _load_image(self, image_input) -> np.ndarray: if isinstance(image_input, np.ndarray): return image_input elif isinstance(image_input, (bytes, bytearray)): arr = np.frombuffer(image_input, np.uint8) img = cv2.imdecode(arr, cv2.IMREAD_COLOR) if img is None: raise ValueError("Tidak dapat membaca image bytes.") return img elif isinstance(image_input, str): img = cv2.imread(image_input) if img is None: raise ValueError(f"File tidak ditemukan: {image_input}") return img else: raise TypeError(f"Tipe input tidak didukung: {type(image_input)}") def _deskew(self, image: np.ndarray) -> np.ndarray: gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) blurred = cv2.GaussianBlur(gray, (5, 5), 0) angle = self._estimate_angle_hough(blurred) if angle is None: angle = self._estimate_angle_contour(blurred) if angle is None or abs(angle) < 0.5 or abs(angle) > 20.0: if self.debug and angle is not None: print(f"[DESKEW] Sudut diabaikan: {angle:.2f}°") return image if self.debug: print(f"[DESKEW] Koreksi rotasi: {angle:.2f}°") h, w = image.shape[:2] center = (w // 2, h // 2) M = cv2.getRotationMatrix2D(center, angle, 1.0) corrected = cv2.warpAffine( image, M, (w, h), flags=cv2.INTER_CUBIC, borderMode=cv2.BORDER_REPLICATE ) return corrected def _estimate_angle_hough(self, gray_blurred: np.ndarray): edges = cv2.Canny(gray_blurred, 50, 150, apertureSize=3) h, w = gray_blurred.shape[:2] min_line_len = int(w * 0.25) lines = cv2.HoughLinesP( edges, 1, np.pi / 180, threshold=80, minLineLength=min_line_len, maxLineGap=20 ) if lines is None: return None angles = [] for line in lines: x1, y1, x2, y2 = line[0] dx = x2 - x1 dy = y2 - y1 if dx == 0: continue angle = np.degrees(np.arctan2(dy, dx)) if abs(angle) <= 20.0: angles.append(angle) if len(angles) < 5: return None return float(np.median(angles)) def _estimate_angle_contour(self, gray_blurred: np.ndarray): _, thresh = cv2.threshold( gray_blurred, 0, 255, cv2.THRESH_BINARY_INV | cv2.THRESH_OTSU ) contours, _ = cv2.findContours( thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE ) h, w = gray_blurred.shape[:2] img_area = h * w angles = [] for c in contours: area = cv2.contourArea(c) if area < img_area * 0.005: continue rect = cv2.minAreaRect(c) angle = rect[2] if angle < -45: angle += 90 if abs(angle) <= 20.0: angles.append(angle) if len(angles) < 3: return None return float(np.median(angles)) def _preprocess(self, image: np.ndarray): h, w = image.shape[:2] if w > 1600: scale = 1600 / w image = cv2.resize(image, (1600, int(h * scale))) elif w < 800: scale = 800 / w image = cv2.resize(image, (800, int(h * scale))) gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) blurred = cv2.GaussianBlur(gray, (5, 5), 0) thresh = cv2.adaptiveThreshold( blurred, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY_INV, blockSize=11, C=2 ) return gray, blurred, thresh def _detect_answer_sheet(self, image, gray, thresh): contours, _ = cv2.findContours( thresh.copy(), cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE ) contours = sorted(contours, key=cv2.contourArea, reverse=True) sheet_contour = None for c in contours[:10]: peri = cv2.arcLength(c, True) approx = cv2.approxPolyDP(c, 0.02 * peri, True) if len(approx) == 4: area = cv2.contourArea(c) img_area = image.shape[0] * image.shape[1] if area > img_area * 0.15: sheet_contour = approx break if sheet_contour is None: h, w = image.shape[:2] sheet_contour = np.array([ [[0, 0]], [[w - 1, 0]], [[w - 1, h - 1]], [[0, h - 1]] ], dtype=np.int32) warped = self._four_point_transform(image, sheet_contour.reshape(4, 2)) warped_gray = cv2.cvtColor(warped, cv2.COLOR_BGR2GRAY) _, warped_thresh = cv2.threshold( warped_gray, 0, 255, cv2.THRESH_BINARY_INV | cv2.THRESH_OTSU ) return warped, warped_thresh def _four_point_transform(self, image, pts): rect = self._order_points(pts) (tl, tr, br, bl) = rect widthA = np.linalg.norm(br - bl) widthB = np.linalg.norm(tr - tl) maxWidth = max(int(widthA), int(widthB)) heightA = np.linalg.norm(tr - br) heightB = np.linalg.norm(tl - bl) maxHeight = max(int(heightA), int(heightB)) dst = np.array([ [0, 0], [maxWidth - 1, 0], [maxWidth - 1, maxHeight - 1], [0, maxHeight - 1] ], dtype=np.float32) M = cv2.getPerspectiveTransform(rect, dst) return cv2.warpPerspective(image, M, (maxWidth, maxHeight)) def _order_points(self, pts): rect = np.zeros((4, 2), dtype=np.float32) s = pts.sum(axis=1) rect[0] = pts[np.argmin(s)] rect[2] = pts[np.argmax(s)] diff = np.diff(pts, axis=1) rect[1] = pts[np.argmin(diff)] rect[3] = pts[np.argmax(diff)] return rect def _detect_bubbles(self, warped, warped_thresh, answer_key, annotate): h, w = warped.shape[:2] min_dim = min(w, h) contours, _ = cv2.findContours( warped_thresh.copy(), cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE ) bubble_contours = [] for c in contours: (x, y, bw, bh) = cv2.boundingRect(c) area = cv2.contourArea(c) perimeter = cv2.arcLength(c, True) if area <= 0 or perimeter <= 0: continue ar = bw / float(bh) min_b = min(bw, bh) extent = area / float(bw * bh) circularity = (4.0 * np.pi * area) / (perimeter * perimeter) if (0.70 <= ar <= 1.30) and \ (min_dim * self.min_bubble_size_ratio <= min_b <= min_dim * self.max_bubble_size_ratio) and \ (0.45 <= circularity <= 1.25) and \ (extent >= 0.40): cx_center = x + bw // 2 cy_center = y + bh // 2 bubble_contours.append((cx_center, cy_center, bw, bh)) bubble_contours = self._filter_by_dominant_bubble_size(bubble_contours) if self.debug: print(f"[DEBUG] Kandidat bubble: {len(bubble_contours)}") row_tolerance = max(h * 0.03, 25) rows = self._cluster_bubbles_by_y(bubble_contours, row_tolerance) if self.debug: for i, row in enumerate(rows): ys = [b[1] for b in row] print(f"[DEBUG] Row {i+1}: {len(row)} bubbles, Y~{int(np.mean(ys))}") min_bubbles_per_row = int(self.num_choices * self.num_column_blocks * 0.8) question_rows = [r for r in rows if len(r) >= min_bubbles_per_row] if self.debug: print(f"[DEBUG] Question rows (setelah filter header): {len(question_rows)}") if len(question_rows) < self.questions_per_block: if self.debug: print("[DEBUG] Fallback ke grid-based detection") return self._grid_based_detection(warped, warped_thresh, answer_key, annotate) question_rows = question_rows[:self.questions_per_block] all_cx = sorted([b[0] for row in question_rows for b in row]) col_block_boundaries = self._find_column_boundaries(all_cx, self.num_column_blocks) if self.debug: print(f"[DEBUG] Column boundaries: {col_block_boundaries}") detected_answers = {} annotated = warped.copy() if annotate else None for row_i, row in enumerate(question_rows): row_sorted = sorted(row, key=lambda b: b[0]) col_groups = [[] for _ in range(self.num_column_blocks)] for b in row_sorted: for si in range(self.num_column_blocks): if col_block_boundaries[si] <= b[0] < col_block_boundaries[si + 1]: col_groups[si].append(b) break for sec_i, sec_bubbles in enumerate(col_groups): q_num = sec_i * self.questions_per_block + row_i + 1 if q_num > self.num_questions: continue sec_sorted = self._select_choice_bubbles(sec_bubbles) if len(sec_sorted) < self.num_choices: if self.debug: print(f"[DEBUG] Q{q_num}: hanya {len(sec_sorted)} bubble terdeteksi") detected_answers[q_num] = None continue fill_scores = [] fill_densities = [] coverage_ratios = [] for b in sec_sorted: cx, cy, bw, bh = b score, density, coverage_ratio = self._measure_bubble_fill( warped_thresh, cx, cy, bw, bh ) fill_scores.append(score) fill_densities.append(density) coverage_ratios.append(coverage_ratio) max_val = max(fill_scores) max_idx = fill_scores.index(max_val) other_vals = [v for i, v in enumerate(fill_scores) if i != max_idx] other_avg = float(np.mean(other_vals)) if other_vals else 0.0 fill_ratio = max_val / (max_val + other_avg + 1e-6) if ( fill_ratio >= self.bubble_fill_threshold and fill_densities[max_idx] >= self.min_mark_density and coverage_ratios[max_idx] >= self.min_mark_coverage_ratio ): chosen = self.choice_labels[max_idx] confidence = float(fill_ratio) else: chosen = None confidence = 0.0 detected_answers[q_num] = chosen if annotate and annotated is not None: correct_ans = answer_key.get(q_num) for i, b in enumerate(sec_sorted): cx, cy, bw, bh = b radius = min(bw, bh) // 2 + 3 if i == max_idx and chosen is not None: if correct_ans and chosen == correct_ans: color = (0, 200, 0) elif correct_ans: color = (0, 0, 220) else: color = (200, 160, 0) cv2.circle(annotated, (cx, cy), radius, color, 2) if sec_sorted: x0 = sec_sorted[0][0] y0 = sec_sorted[0][1] cv2.putText( annotated, str(q_num), (max(0, x0 - 30), y0 + 5), cv2.FONT_HERSHEY_SIMPLEX, 0.32, (80, 80, 80), 1 ) return detected_answers, annotated def _filter_by_dominant_bubble_size(self, bubbles): """Keep contours close to the dominant printed bubble size.""" if len(bubbles) < self.num_choices: return bubbles sizes = np.array([min(b[2], b[3]) for b in bubbles], dtype=np.float32) median_size = float(np.median(sizes)) tolerance = max(4.0, median_size * 0.35) filtered = [ b for b in bubbles if abs(min(b[2], b[3]) - median_size) <= tolerance ] return filtered if len(filtered) >= self.num_choices else bubbles def _select_choice_bubbles(self, sec_bubbles): candidates = sorted(sec_bubbles, key=lambda b: b[0]) if len(candidates) <= self.num_choices: return candidates if len(candidates) == self.num_choices else [] index_groups = ( combinations(range(len(candidates)), self.num_choices) if len(candidates) <= 12 else (range(i, i + self.num_choices) for i in range(0, len(candidates) - self.num_choices + 1)) ) best_group = None best_score = None all_x_span = candidates[-1][0] - candidates[0][0] + 1e-6 for indexes in index_groups: group = [candidates[i] for i in indexes] xs = np.array([b[0] for b in group], dtype=np.float32) ys = np.array([b[1] for b in group], dtype=np.float32) sizes = np.array([min(b[2], b[3]) for b in group], dtype=np.float32) gaps = np.diff(xs) if len(gaps) == 0 or np.any(gaps <= 0): continue mean_gap = float(np.mean(gaps)) median_size = float(np.median(sizes)) if mean_gap < median_size * 1.05: continue gap_cv = float(np.std(gaps) / (mean_gap + 1e-6)) size_cv = float(np.std(sizes) / (float(np.mean(sizes)) + 1e-6)) y_span = float((np.max(ys) - np.min(ys)) / (median_size + 1e-6)) if gap_cv > 0.55 or size_cv > 0.40 or y_span > 0.90: continue left_skip = float((group[0][0] - candidates[0][0]) / all_x_span) score = gap_cv * 4.0 + size_cv * 1.5 + y_span + max(0.0, 0.08 - left_skip) if best_score is None or score < best_score: best_score = score best_group = group if best_group is not None: return sorted(best_group, key=lambda b: b[0]) return candidates[-self.num_choices:] def _measure_bubble_fill(self, thresh_img, cx, cy, bw, bh): h, w = thresh_img.shape[:2] outer_radius = max(1, min(bw, bh) // 2) inner_radius = max(1, int(outer_radius * 0.72)) x1 = max(0, cx - inner_radius) x2 = min(w, cx + inner_radius + 1) y1 = max(0, cy - inner_radius) y2 = min(h, cy + inner_radius + 1) roi = thresh_img[y1:y2, x1:x2] if roi.size == 0: return 0.0, 0.0, 0.0 mask = np.zeros(roi.shape[:2], dtype=np.uint8) cv2.circle(mask, (cx - x1, cy - y1), inner_radius, 255, -1) mask_area = cv2.countNonZero(mask) if mask_area == 0: return 0.0, 0.0, 0.0 marked = cv2.bitwise_and(roi, roi, mask=mask) density = cv2.countNonZero(marked) / float(mask_area) grid_size = 5 solid_cells = 0 valid_cells = 0 min_cell_area = max(3, mask_area / float(grid_size * grid_size) * 0.35) for gy in range(grid_size): y_start = int(round(gy * roi.shape[0] / grid_size)) y_end = int(round((gy + 1) * roi.shape[0] / grid_size)) for gx in range(grid_size): x_start = int(round(gx * roi.shape[1] / grid_size)) x_end = int(round((gx + 1) * roi.shape[1] / grid_size)) cell_mask = mask[y_start:y_end, x_start:x_end] cell_marked = marked[y_start:y_end, x_start:x_end] cell_area = cv2.countNonZero(cell_mask) if cell_area < min_cell_area: continue valid_cells += 1 cell_density = cv2.countNonZero(cell_marked) / float(cell_area) if cell_density >= 0.35: solid_cells += 1 coverage_ratio = solid_cells / float(valid_cells) if valid_cells else 0.0 score = (density * 0.55) + (coverage_ratio * 0.45) return float(score), float(density), float(coverage_ratio) def _cluster_bubbles_by_y(self, bubbles, tolerance): if not bubbles: return [] xs = np.array([b[0] for b in bubbles], dtype=np.float64) ys = np.array([b[1] for b in bubbles], dtype=np.float64) slope = 0.0 if len(bubbles) >= 10: x_mean = float(np.mean(xs)) y_mean = float(np.mean(ys)) denom = float(np.sum((xs - x_mean) ** 2)) if denom > 1e-6: slope = float(np.sum((xs - x_mean) * (ys - y_mean)) / denom) max_slope = np.tan(np.radians(20)) slope = max(-max_slope, min(max_slope, slope)) if self.debug and abs(slope) > 0.005: angle_deg = np.degrees(np.arctan(slope)) print(f"[CLUSTER] Estimasi kemiringan baris: {angle_deg:.2f}°") x_ref = float(np.mean(xs)) projected_ys = ys - slope * (xs - x_ref) order = np.argsort(projected_ys) bubbles_sorted = [bubbles[i] for i in order] proj_sorted = projected_ys[order] rows = [] current = [bubbles_sorted[0]] current_proj = [proj_sorted[0]] for idx in range(1, len(bubbles_sorted)): b = bubbles_sorted[idx] py = proj_sorted[idx] avg_proj = float(np.mean(current_proj)) if abs(py - avg_proj) <= tolerance: current.append(b) current_proj.append(py) else: rows.append(current) current = [b] current_proj = [py] rows.append(current) return rows def _find_column_boundaries(self, all_cx_sorted, num_blocks): if len(all_cx_sorted) < 2: w_approx = all_cx_sorted[-1] * 2 if all_cx_sorted else 1000 step = w_approx // num_blocks return [i * step for i in range(num_blocks + 1)] col_group_centers = [] cur_group = [all_cx_sorted[0]] for x in all_cx_sorted[1:]: if x - cur_group[-1] <= 35: cur_group.append(x) else: col_group_centers.append(int(np.mean(cur_group))) cur_group = [x] col_group_centers.append(int(np.mean(cur_group))) if len(col_group_centers) < 2: step = (all_cx_sorted[-1] - all_cx_sorted[0]) // num_blocks start = all_cx_sorted[0] return [0] + [start + i * step for i in range(1, num_blocks)] + [all_cx_sorted[-1] + 100] gaps = [] for i in range(1, len(col_group_centers)): gap = col_group_centers[i] - col_group_centers[i - 1] gaps.append((gap, col_group_centers[i - 1], col_group_centers[i])) gaps_sorted = sorted(gaps, reverse=True) separator_xs = sorted([ g[1] + (g[2] - g[1]) // 2 for g in gaps_sorted[:num_blocks - 1] ]) boundaries = [0] + separator_xs + [all_cx_sorted[-1] + 100] return boundaries def _grid_based_detection(self, warped, warped_thresh, answer_key, annotate): h, w = warped.shape[:2] margin_top = int(h * 0.12) margin_bottom = int(h * 0.05) margin_left = int(w * 0.05) margin_right = int(w * 0.05) grid_h = h - margin_top - margin_bottom grid_w = w - margin_left - margin_right block_w = grid_w / self.num_column_blocks row_step = grid_h / self.questions_per_block col_step = block_w / self.num_choices bubble_r = int(min(row_step, col_step) * 0.35) detected_answers = {} annotated = warped.copy() if annotate else None for q in range(self.num_questions): question_num = q + 1 sec_i = (q) // self.questions_per_block row_i = (q) % self.questions_per_block cy_center = int(margin_top + (row_i + 0.5) * row_step) block_start_x = margin_left + sec_i * int(block_w) intensities = [] fill_densities = [] coverage_ratios = [] for c in range(self.num_choices): cx_center = int(block_start_x + (c + 0.5) * col_step) score, density, coverage_ratio = self._measure_bubble_fill( warped_thresh, cx_center, cy_center, bubble_r * 2, bubble_r * 2 ) intensities.append(score) fill_densities.append(density) coverage_ratios.append(coverage_ratio) if max(intensities) <= 0: detected_answers[question_num] = None continue max_idx = intensities.index(max(intensities)) other_avg = np.mean([v for i, v in enumerate(intensities) if i != max_idx]) \ if len(intensities) > 1 else 0 fill_ratio = intensities[max_idx] / (intensities[max_idx] + other_avg + 1e-6) detected_answers[question_num] = ( self.choice_labels[max_idx] if ( fill_ratio >= self.bubble_fill_threshold and fill_densities[max_idx] >= self.min_mark_density and coverage_ratios[max_idx] >= self.min_mark_coverage_ratio ) else None ) return detected_answers, annotated def _calculate_score(self, detected: dict, answer_key: dict) -> OMRResult: correct = 0 wrong = 0 empty = 0 details = [] for q_num in range(1, self.num_questions + 1): detected_ans = detected.get(q_num) correct_ans = answer_key.get(q_num) if detected_ans is None: empty += 1 is_correct = False confidence = 0.0 elif correct_ans is None: wrong += 1 is_correct = False confidence = 1.0 elif detected_ans == correct_ans: correct += 1 is_correct = True confidence = 1.0 else: wrong += 1 is_correct = False confidence = 1.0 details.append(BubbleResult( question_number=q_num, detected_answer=detected_ans, correct_answer=correct_ans, is_correct=is_correct, confidence=confidence )) total_with_key = sum(1 for q in range(1, self.num_questions + 1) if answer_key.get(q)) if total_with_key == 0: total_with_key = self.num_questions score = (correct / total_with_key) * 100 return OMRResult( total_questions=self.num_questions, correct_count=correct, wrong_count=wrong, empty_count=empty, score=round(score, 2), percentage=round(score, 2), details=details ) def _encode_image(self, image: np.ndarray) -> str: _, buffer = cv2.imencode('.png', image) return base64.b64encode(buffer).decode('utf-8') class SheetNotFoundError(Exception): pass if __name__ == "__main__": import sys if len(sys.argv) < 2: print("Usage: python omr_processor.py [answer_key_image_path]") print(" python omr_processor.py --key A,B,C,D,...") sys.exit(1) processor = OMRProcessor( num_questions=50, num_choices=5, num_column_blocks=5, bubble_fill_threshold=0.38, debug=True ) image_path = sys.argv[1] answer_key = {} if len(sys.argv) >= 4 and sys.argv[2] == "--key": labels_str = sys.argv[3].upper().split(',') for i, label in enumerate(labels_str): label = label.strip() if label: answer_key[i + 1] = label print(f"Kunci jawaban dari argumen: {answer_key}") elif len(sys.argv) >= 3 and sys.argv[2] != "--key": key_image_path = sys.argv[2] print(f"Membaca kunci jawaban dari: {key_image_path}") answer_key = processor.process_answer_key_image(key_image_path) print(f"Kunci jawaban terdeteksi: {answer_key}") print(f"\nMemproses: {image_path}") result = processor.process(image_path, answer_key, return_preview=False) if result.error: print(f"\nERROR: {result.error}") else: print(f"\n{'='*40}") print(f"Total Soal : {result.total_questions}") print(f"Benar : {result.correct_count}") print(f"Salah : {result.wrong_count}") print(f"Kosong : {result.empty_count}") print(f"Skor : {result.score:.2f}") print(f"{'='*40}") print("\nDetail per soal:") for d in result.details: status = "✓" if d.is_correct else ("○" if d.detected_answer is None else "✗") print(f" Q{d.question_number:2d}: Jawab={d.detected_answer or '-':>1} " f"Kunci={d.correct_answer or '-':>1} {status}")