import time import numpy as np import cv2 from typing import List, Tuple, Sequence, Any from numpy import ndarray from multiprocessing import cpu_count from functools import partial import copy import threading from pathlib import Path # Module-level template variables (initialized lazily) _TEMPLATE_KEYPOINTS: list[tuple[int, int]] = None _TEMPLATE_IMAGE: np.ndarray = None # Cached template dimensions for performance (default values) _TEMPLATE_MAX_X: int = 1045 _TEMPLATE_MAX_Y: int = 675 def _initialize_template_variables(template_keypoints=None, template_image=None): """ Initialize module-level template variables. Called once from run_keypoints_post_processing. Args: template_keypoints: Optional template keypoints (pre-loaded) template_image: Optional template image (pre-loaded from miner constructor) """ global _TEMPLATE_KEYPOINTS, _TEMPLATE_IMAGE if _TEMPLATE_KEYPOINTS is None or _TEMPLATE_IMAGE is None: try: from keypoint_evaluation import ( TEMPLATE_KEYPOINTS, ) # Set template keypoints (use provided or use default) if _TEMPLATE_KEYPOINTS is None: if template_keypoints is not None: _TEMPLATE_KEYPOINTS = template_keypoints else: _TEMPLATE_KEYPOINTS = TEMPLATE_KEYPOINTS # Set template image (use provided pre-loaded image) if _TEMPLATE_IMAGE is None: if template_image is not None: # Use pre-loaded template image (from miner constructor) _TEMPLATE_IMAGE = template_image else: print("Warning: Template image not provided, some validation may be skipped") # Cache template dimensions for performance global _TEMPLATE_MAX_X, _TEMPLATE_MAX_Y if _TEMPLATE_KEYPOINTS is not None and len(_TEMPLATE_KEYPOINTS) > 0: valid_template_points = [(x, y) for x, y in _TEMPLATE_KEYPOINTS if x > 0 and y > 0] if len(valid_template_points) > 0: _TEMPLATE_MAX_X = max(x for x, y in valid_template_points) _TEMPLATE_MAX_Y = max(y for x, y in valid_template_points) except ImportError: pass except Exception as e: print(f"Warning: Could not load template: {e}") FOOTBALL_KEYPOINTS: list[tuple[int, int]] = [ (0, 0), # 1 (0, 0), # 2 (0, 0), # 3 (0, 0), # 4 (0, 0), # 5 (0, 0), # 6 (0, 0), # 7 (0, 0), # 8 (0, 0), # 9 (0, 0), # 10 (0, 0), # 11 (0, 0), # 12 (0, 0), # 13 (0, 0), # 14 (527, 283), # 15 (527, 403), # 16 (0, 0), # 17 (0, 0), # 18 (0, 0), # 19 (0, 0), # 20 (0, 0), # 21 (0, 0), # 22 (0, 0), # 23 (0, 0), # 24 (0, 0), # 25 (0, 0), # 26 (0, 0), # 27 (0, 0), # 28 (0, 0), # 29 (0, 0), # 30 (405, 340), # 31 (645, 340), # 32 ] def convert_keypoints_to_val_format(keypoints): return [tuple(int(x) for x in pair) for pair in keypoints] def validate_with_nearby_keypoints( kp_idx: int, kp: tuple[int, int], valid_indices: list[int], result: list[tuple[int, int]], template_keypoints: list[tuple[int, int]], scale_factor: float = None, ) -> float: """ Validate a keypoint by checking distances to nearby keypoints on the same side. Returns validation score (lower is better), or None if validation not possible. """ template_kp = template_keypoints[kp_idx] # Define which keypoints are on the same side # Left side: 10, 11, 12, 13 (indices 9, 10, 11, 12) # Right side: 18, 19, 20, 21, 22, 23, 24, 25-30 (indices 17-29) left_side_indices = [9, 10, 11, 12] # Keypoints 10-13 right_side_indices = list(range(17, 30)) # Keypoints 18-30 # Determine which side this keypoint should be on if kp_idx in left_side_indices: same_side_indices = left_side_indices elif kp_idx in right_side_indices: same_side_indices = right_side_indices else: return None # Can't validate # Find nearby keypoints on the same side that are detected nearby_kps = [] for nearby_idx in same_side_indices: if nearby_idx != kp_idx and nearby_idx in valid_indices: nearby_kp = result[nearby_idx] nearby_template_kp = template_keypoints[nearby_idx] nearby_kps.append((nearby_idx, nearby_kp, nearby_template_kp)) if len(nearby_kps) == 0: return None # No nearby keypoints to validate with # Calculate distance errors to nearby keypoints distance_errors = [] for nearby_idx, nearby_kp, nearby_template_kp in nearby_kps: # Detected distance detected_dist = np.sqrt((kp[0] - nearby_kp[0])**2 + (kp[1] - nearby_kp[1])**2) # Template distance template_dist = np.sqrt((template_kp[0] - nearby_template_kp[0])**2 + (template_kp[1] - nearby_template_kp[1])**2) if template_dist > 0: # Expected detected distance if scale_factor: expected_dist = template_dist * scale_factor else: expected_dist = template_dist if expected_dist > 0: # Normalized error error = abs(detected_dist - expected_dist) / expected_dist distance_errors.append(error) if len(distance_errors) > 0: return np.mean(distance_errors) return None def remove_duplicate_detections( keypoints: list[tuple[int, int]], frame_width: int = None, frame_height: int = None, ) -> list[tuple[int, int]]: """ Remove duplicate/conflicting keypoint detections using distance-based validation. Uses the principle that if two keypoints are detected very close together, but in the template they should be far apart, one of them is likely wrong. Validates each keypoint by checking if its distances to other keypoints match the expected template distances. Args: keypoints: List of 32 keypoints frame_width: Optional frame width for validation frame_height: Optional frame height for validation Returns: Cleaned list of keypoints with duplicates removed """ if len(keypoints) != 32: if len(keypoints) < 32: keypoints = list(keypoints) + [(0, 0)] * (32 - len(keypoints)) else: keypoints = keypoints[:32] result = list(keypoints) try: from keypoint_evaluation import TEMPLATE_KEYPOINTS template_available = True except ImportError: template_available = False if not template_available: return result # Get all valid detected keypoints valid_indices = [] for i in range(32): if result[i][0] > 0 and result[i][1] > 0: valid_indices.append(i) if len(valid_indices) < 2: return result # Calculate scale factor from detected keypoints to template # Use pairs of keypoints that are far apart in template to estimate scale scale_factor = None if len(valid_indices) >= 2: max_template_dist = 0 max_detected_dist = 0 for i in range(len(valid_indices)): for j in range(i + 1, len(valid_indices)): idx_i = valid_indices[i] idx_j = valid_indices[j] template_i = TEMPLATE_KEYPOINTS[idx_i] template_j = TEMPLATE_KEYPOINTS[idx_j] template_dist = np.sqrt((template_i[0] - template_j[0])**2 + (template_i[1] - template_j[1])**2) kp_i = result[idx_i] kp_j = result[idx_j] detected_dist = np.sqrt((kp_i[0] - kp_j[0])**2 + (kp_i[1] - kp_j[1])**2) if template_dist > max_template_dist and detected_dist > 0: max_template_dist = template_dist max_detected_dist = detected_dist if max_template_dist > 0 and max_detected_dist > 0: scale_factor = max_detected_dist / max_template_dist # For each keypoint, validate it by checking distances to other keypoints keypoint_scores = {} for idx in valid_indices: kp = result[idx] template_kp = TEMPLATE_KEYPOINTS[idx] # Calculate how well this keypoint's distances match template distances distance_errors = [] num_comparisons = 0 for other_idx in valid_indices: if other_idx == idx: continue other_kp = result[other_idx] other_template_kp = TEMPLATE_KEYPOINTS[other_idx] # Calculate detected distance detected_dist = np.sqrt((kp[0] - other_kp[0])**2 + (kp[1] - other_kp[1])**2) # Calculate template distance template_dist = np.sqrt((template_kp[0] - other_template_kp[0])**2 + (template_kp[1] - other_template_kp[1])**2) if template_dist > 50: # Only check keypoints that should be reasonably far apart num_comparisons += 1 # Expected detected distance (scaled from template) if scale_factor: expected_dist = template_dist * scale_factor else: expected_dist = template_dist # Calculate error (normalized) if expected_dist > 0: error = abs(detected_dist - expected_dist) / expected_dist distance_errors.append(error) # Score: lower is better (smaller distance errors) if num_comparisons > 0: avg_error = np.mean(distance_errors) keypoint_scores[idx] = avg_error else: keypoint_scores[idx] = 0.0 # Find pairs of keypoints that are too close but should be far apart conflicts = [] for i in range(len(valid_indices)): for j in range(i + 1, len(valid_indices)): idx_i = valid_indices[i] idx_j = valid_indices[j] kp_i = result[idx_i] kp_j = result[idx_j] # Calculate detected distance detected_dist = np.sqrt((kp_i[0] - kp_j[0])**2 + (kp_i[1] - kp_j[1])**2) # Calculate template distance template_i = TEMPLATE_KEYPOINTS[idx_i] template_j = TEMPLATE_KEYPOINTS[idx_j] template_dist = np.sqrt((template_i[0] - template_j[0])**2 + (template_i[1] - template_j[1])**2) # If template distance is large but detected distance is small, it's a conflict if template_dist > 100 and detected_dist < 30: # Enhanced validation: use nearby keypoints to determine which is correct # For example, if we have 24 and 29, we can check distances to determine if it's 13 or 21 score_i = keypoint_scores.get(idx_i, 1.0) score_j = keypoint_scores.get(idx_j, 1.0) # Try to validate using nearby keypoints on the same side # Keypoint 13 is on left side, keypoint 21 is on right side # If we have right-side keypoints (like 24, 29), check distances nearby_validation_i = validate_with_nearby_keypoints( idx_i, kp_i, valid_indices, result, TEMPLATE_KEYPOINTS, scale_factor ) nearby_validation_j = validate_with_nearby_keypoints( idx_j, kp_j, valid_indices, result, TEMPLATE_KEYPOINTS, scale_factor ) # Prioritize nearby validation: if one has nearby validation and the other doesn't, # prefer the one with nearby validation (it's more reliable) validation_score_i = score_i validation_score_j = score_j if nearby_validation_i is not None and nearby_validation_j is not None: # Both have nearby validation, use those scores validation_score_i = nearby_validation_i validation_score_j = nearby_validation_j elif nearby_validation_i is not None: # Only i has nearby validation, prefer it (give it much better score) validation_score_i = nearby_validation_i validation_score_j = score_j + 1.0 # Penalize j for not having nearby validation elif nearby_validation_j is not None: # Only j has nearby validation, prefer it validation_score_i = score_i + 1.0 # Penalize i for not having nearby validation validation_score_j = nearby_validation_j # If neither has nearby validation, use general distance scores # Remove the one with worse validation score if validation_score_i > validation_score_j: conflicts.append((idx_i, idx_j, validation_score_i, validation_score_j)) else: conflicts.append((idx_j, idx_i, validation_score_j, validation_score_i)) # Remove conflicting keypoints (keep the one with better score) removed_indices = set() for remove_idx, keep_idx, remove_score, keep_score in conflicts: if remove_idx not in removed_indices: print(f"Removing duplicate detection: keypoint {remove_idx+1} at {result[remove_idx]} conflicts with keypoint {keep_idx+1} at {result[keep_idx]} " f"(detected distance: {np.sqrt((result[remove_idx][0] - result[keep_idx][0])**2 + (result[remove_idx][1] - result[keep_idx][1])**2):.1f}, " f"template distance: {np.sqrt((TEMPLATE_KEYPOINTS[remove_idx][0] - TEMPLATE_KEYPOINTS[keep_idx][0])**2 + (TEMPLATE_KEYPOINTS[remove_idx][1] - TEMPLATE_KEYPOINTS[keep_idx][1])**2):.1f}). " f"Keeping keypoint {keep_idx+1} (score: {keep_score:.3f} vs {remove_score:.3f}).") result[remove_idx] = (0, 0) removed_indices.add(remove_idx) return result def calculate_missing_keypoints( keypoints: list[tuple[int, int]], frame_width: int = None, frame_height: int = None, ) -> list[tuple[int, int]]: """ Calculate missing keypoint coordinates for multiple cases: 1. Given keypoints 14, 15, 16 (and possibly 17), and either 31 or 32, calculate the missing center circle point (32 or 31). 2. Given three or four of keypoints 18, 19, 20, 21 and any of 22-30, calculate missing keypoint positions (like 22 or others) to prevent warping failures. Args: keypoints: List of 32 keypoints (some may be (0,0) if missing) frame_width: Optional frame width for validation frame_height: Optional frame height for validation Returns: Updated list of 32 keypoints with calculated missing keypoints filled in """ if len(keypoints) != 32: # Pad or truncate to 32 if len(keypoints) < 32: keypoints = list(keypoints) + [(0, 0)] * (32 - len(keypoints)) else: keypoints = keypoints[:32] result = list(keypoints) # Helper to get keypoint def get_kp(kp_idx): if kp_idx < 0 or kp_idx >= 32: return None x, y = result[kp_idx] if x == 0 and y == 0: return None return (x, y) # Case 1: Find center x-coordinate from center line keypoints (14, 15, 16, or 17) # Keypoints 14, 15, 16, 17 are on the center vertical line (indices 13, 14, 15, 16) center_x = None for center_kp_idx in [13, 14, 15, 16]: # 14, 15, 16, 17 (0-indexed) kp = get_kp(center_kp_idx) if kp: center_x = kp[0] break # If we have center line, calculate missing center circle point if center_x is not None: # Keypoint 31 is at index 30 (left side of center circle) # Keypoint 32 is at index 31 (right side of center circle) kp_31 = get_kp(30) # Keypoint 31 kp_32 = get_kp(31) # Keypoint 32 if kp_31 and not kp_32: # Given 31, calculate 32 by reflecting across center_x # Formula: x_32 = center_x + (center_x - x_31) = 2*center_x - x_31 # y_32 = y_31 (same y-coordinate, both on center horizontal line) dx = center_x - kp_31[0] result[31] = (int(round(center_x + dx)), kp_31[1]) elif kp_32 and not kp_31: # Given 32, calculate 31 by reflecting across center_x # Formula: x_31 = center_x - (x_32 - center_x) = 2*center_x - x_32 # y_31 = y_32 (same y-coordinate, both on center horizontal line) dx = kp_32[0] - center_x result[30] = (int(round(center_x - dx)), kp_32[1]) # Case 1.5: Unified handling of left side keypoints (1-13) # Three parallel vertical lines on left side: # - Line 1-6: keypoints 1, 2, 3, 4, 5, 6 (indices 0-5) # - Line 7-8: keypoints 7, 8 (indices 6-7) # - Line 10-13: keypoints 10, 11, 12, 13 (indices 9-12) # Keypoint 9 (index 8) is between line 1-6 and line 10-13 # Collect all left-side keypoints (1-13, indices 0-12, excluding 9 which is center) left_side_all = [] line_1_6_points = [] # Indices 0-5 line_7_8_points = [] # Indices 6-7 line_10_13_points = [] # Indices 9-12 for idx in range(0, 13): # Keypoints 1-13 (indices 0-12) if idx == 8: # Skip keypoint 9 (index 8) - it's a center point continue kp = get_kp(idx) if kp: left_side_all.append((idx, kp)) if 0 <= idx <= 5: # Line 1-6 line_1_6_points.append((idx, kp)) elif 6 <= idx <= 7: # Line 7-8 line_7_8_points.append((idx, kp)) elif 9 <= idx <= 12: # Line 10-13 line_10_13_points.append((idx, kp)) kp_9 = get_kp(8) # Keypoint 9 if kp_9: left_side_all.append((8, kp_9)) total_left_side_count = len(left_side_all) # If we have 6 or more points, no need to calculate more if total_left_side_count >= 6: pass # Don't calculate more points elif total_left_side_count == 5: # Check if 4 points are on one line and 1 on another line counts_per_line = [ len(line_1_6_points), len(line_7_8_points), len(line_10_13_points) ] if max(counts_per_line) == 4 and sum(counts_per_line) == 4: # 4 points on one line, need to calculate 1 more point on another line # Determine which line has 4 points and calculate on a different line if len(line_1_6_points) == 4: # All 4 on line 1-6, calculate on line 10-13 or 7-8 # Prefer line 10-13 (right edge of left side) if len(line_10_13_points) == 0: # Calculate a point on line 10-13 # Fit line through 1-6 points points_1_6 = np.array([[kp[0], kp[1]] for _, kp in line_1_6_points]) x_coords = points_1_6[:, 0] y_coords = points_1_6[:, 1] A = np.vstack([x_coords, np.ones(len(x_coords))]).T m_1_6, b_1_6 = np.linalg.lstsq(A, y_coords, rcond=None)[0] # Calculate a point on line 10-13 (parallel to 1-6) # Use template y-coordinate for one of 10-13 points template_ys_10_13 = [140, 270, 410, 540] # Template y for 10-13 template_indices_10_13 = [9, 10, 11, 12] # Use median y from 1-6 points to estimate scale median_y = np.median(y_coords) # Calculate x using parallel line geometry # In template: line 10-13 is at x=165, line 1-6 is at x=5 # Ratio: 165/5 = 33 if abs(m_1_6) > 1e-6: x_on_line_1_6 = (median_y - b_1_6) / m_1_6 x_new = int(round(x_on_line_1_6 * 33)) else: x_new = int(round(np.median(x_coords) * 33)) # Find first missing index in 10-13 range for template_y, idx in zip(template_ys_10_13, template_indices_10_13): if result[idx] is None: result[idx] = (x_new, int(round(median_y))) break elif len(line_10_13_points) == 4: # All 4 on line 10-13, calculate on line 1-6 # Similar logic but in reverse points_10_13 = np.array([[kp[0], kp[1]] for _, kp in line_10_13_points]) x_coords = points_10_13[:, 0] y_coords = points_10_13[:, 1] A = np.vstack([x_coords, np.ones(len(x_coords))]).T m_10_13, b_10_13 = np.linalg.lstsq(A, y_coords, rcond=None)[0] # Calculate a point on line 1-6 template_ys_1_6 = [5, 140, 250, 430, 540, 675] # Template y for 1-6 template_indices_1_6 = [0, 1, 2, 3, 4, 5] median_y = np.median(y_coords) # Calculate x using parallel line geometry # Ratio: 5/165 ≈ 0.0303 if abs(m_10_13) > 1e-6: x_on_line_10_13 = (median_y - b_10_13) / m_10_13 x_new = int(round(x_on_line_10_13 * 0.0303)) else: x_new = int(round(np.median(x_coords) * 0.0303)) for template_y, idx in zip(template_ys_1_6, template_indices_1_6): if result[idx] is None: result[idx] = (x_new, int(round(median_y))) break elif total_left_side_count < 5: # Need to calculate missing keypoints to get exactly 5 points # Requirements: # 1. Must have keypoint 9 (if possible) # 2. 4 points shouldn't be all on one line (need distribution) # Template coordinates for reference template_coords_left = { 0: (5, 5), # 1 1: (5, 140), # 2 2: (5, 250), # 3 3: (5, 430), # 4 4: (5, 540), # 5 5: (5, 675), # 6 6: (55, 250), # 7 7: (55, 430), # 8 8: (110, 340), # 9 (what we're calculating) 9: (165, 140), # 10 10: (165, 270), # 11 11: (165, 410), # 12 12: (165, 540), # 13 } # Define line groups (vertical and horizontal lines) # Vertical lines: 1-6, 7-8, 10-13 # Horizontal lines: 2-10, 3-7, 4-8, 5-13 line_groups_left = { '1-6': ([0, 1, 2, 3, 4, 5], 'vertical'), # indices: 1, 2, 3, 4, 5, 6 '7-8': ([6, 7], 'vertical'), # indices: 7, 8 '10-13': ([9, 10, 11, 12], 'vertical'), # indices: 10, 11, 12, 13 '2-10': ([1, 9], 'horizontal'), # indices: 2, 10 '3-7': ([2, 6], 'horizontal'), # indices: 3, 7 '4-8': ([3, 7], 'horizontal'), # indices: 4, 8 '5-13': ([4, 12], 'horizontal'), # indices: 5, 13 } # Collect all available points with their indices all_available_points_left = {} for idx, kp in line_1_6_points: all_available_points_left[idx] = kp for idx, kp in line_7_8_points: all_available_points_left[idx] = kp for idx, kp in line_10_13_points: all_available_points_left[idx] = kp # Step 1: Find the best vertical line and best horizontal line separately best_vertical_line_name_left = None best_vertical_line_points_left = [] max_vertical_points_left = 1 best_horizontal_line_name_left = None best_horizontal_line_points_left = [] max_horizontal_points_left = 1 for line_name, (indices, line_type) in line_groups_left.items(): line_points = [(idx, all_available_points_left[idx]) for idx in indices if idx in all_available_points_left] if line_type == 'vertical' and len(line_points) > max_vertical_points_left: max_vertical_points_left = len(line_points) best_vertical_line_name_left = line_name best_vertical_line_points_left = line_points elif line_type == 'horizontal' and len(line_points) > max_horizontal_points_left: max_horizontal_points_left = len(line_points) best_horizontal_line_name_left = line_name best_horizontal_line_points_left = line_points # Check and calculate missing points on detected lines # For vertical lines if best_vertical_line_name_left is not None: expected_indices = line_groups_left[best_vertical_line_name_left][0] detected_indices = {idx for idx, _ in best_vertical_line_points_left} missing_indices = [idx for idx in expected_indices if idx not in detected_indices] if len(missing_indices) > 0: # Calculate missing points using template ratios template_start = template_coords_left[best_vertical_line_points_left[0][0]] template_end = template_coords_left[best_vertical_line_points_left[-1][0]] frame_start = best_vertical_line_points_left[0][1] frame_end = best_vertical_line_points_left[-1][1] for missing_idx in missing_indices: template_missing = template_coords_left[missing_idx] # Calculate ratio along the line based on y-coordinate (vertical line) template_y_start = template_start[1] template_y_end = template_end[1] template_y_missing = template_missing[1] if abs(template_y_end - template_y_start) > 1e-6: ratio = (template_y_missing - template_y_start) / (template_y_end - template_y_start) else: ratio = 0.5 # Calculate frame coordinates x_new = frame_start[0] + (frame_end[0] - frame_start[0]) * ratio y_new = frame_start[1] + (frame_end[1] - frame_start[1]) * ratio new_point = (int(round(x_new)), int(round(y_new))) # Add to result and update collections result[missing_idx] = new_point best_vertical_line_points_left.append((missing_idx, new_point)) all_available_points_left[missing_idx] = new_point total_left_side_count += 1 max_vertical_points_left = len(best_vertical_line_points_left) # Sort by index to maintain order best_vertical_line_points_left.sort(key=lambda x: x[0]) # Check if we can now form a horizontal line with the newly calculated points for line_name, (indices, line_type) in line_groups_left.items(): if line_type == 'horizontal': line_points = [(idx, all_available_points_left[idx]) for idx in indices if idx in all_available_points_left] if len(line_points) > max_horizontal_points_left: max_horizontal_points_left = len(line_points) best_horizontal_line_name_left = line_name best_horizontal_line_points_left = line_points # For horizontal lines if best_horizontal_line_name_left is not None: expected_indices = line_groups_left[best_horizontal_line_name_left][0] detected_indices = {idx for idx, _ in best_horizontal_line_points_left} missing_indices = [idx for idx in expected_indices if idx not in detected_indices] if len(missing_indices) > 0: # Calculate missing points using template ratios template_start = template_coords_left[best_horizontal_line_points_left[0][0]] template_end = template_coords_left[best_horizontal_line_points_left[-1][0]] frame_start = best_horizontal_line_points_left[0][1] frame_end = best_horizontal_line_points_left[-1][1] for missing_idx in missing_indices: template_missing = template_coords_left[missing_idx] # Calculate ratio along the line based on x-coordinate (horizontal line) template_x_start = template_start[0] template_x_end = template_end[0] template_x_missing = template_missing[0] if abs(template_x_end - template_x_start) > 1e-6: ratio = (template_x_missing - template_x_start) / (template_x_end - template_x_start) else: ratio = 0.5 # Calculate frame coordinates x_new = frame_start[0] + (frame_end[0] - frame_start[0]) * ratio y_new = frame_start[1] + (frame_end[1] - frame_start[1]) * ratio new_point = (int(round(x_new)), int(round(y_new))) # Add to result and update collections result[missing_idx] = new_point best_horizontal_line_points_left.append((missing_idx, new_point)) all_available_points_left[missing_idx] = new_point total_left_side_count += 1 max_horizontal_points_left = len(best_horizontal_line_points_left) # Sort by index to maintain order best_horizontal_line_points_left.sort(key=lambda x: x[0]) # Check if we can now form a vertical line with the newly calculated points for line_name, (indices, line_type) in line_groups_left.items(): if line_type == 'vertical': line_points = [(idx, all_available_points_left[idx]) for idx in indices if idx in all_available_points_left] if len(line_points) > max_vertical_points_left: max_vertical_points_left = len(line_points) best_vertical_line_name_left = line_name best_vertical_line_points_left = line_points # If we only have one direction, try to calculate the other direction line # Similar logic to right side, adapted for left side structure if best_vertical_line_name_left is not None and best_horizontal_line_name_left is None: # We have vertical line but no horizontal line # Find an off-line point (not on the vertical line) off_line_point = None off_line_idx = None vertical_line_indices = line_groups_left[best_vertical_line_name_left][0] for idx, kp in all_available_points_left.items(): if idx not in vertical_line_indices: off_line_point = kp off_line_idx = idx break if off_line_point is not None: # Convert off_line_point to numpy array for arithmetic operations off_line_point = np.array(off_line_point) # Project off_line_point onto vertical line template_off_line = template_coords_left[off_line_idx] template_vertical_start_index = best_vertical_line_points_left[0][0] template_vertical_end_index = best_vertical_line_points_left[-1][0] template_vertical_start = template_coords_left[template_vertical_start_index] template_vertical_end = template_coords_left[template_vertical_end_index] # Project at same y as off_line_point template_y_off = template_off_line[1] template_y_vertical_start = template_vertical_start[1] template_y_vertical_end = template_vertical_end[1] if abs(template_y_vertical_end - template_y_vertical_start) > 1e-6: ratio_proj = (template_y_off - template_y_vertical_start) / (template_y_vertical_end - template_y_vertical_start) else: ratio_proj = 0.5 frame_vertical_start = best_vertical_line_points_left[0][1] frame_vertical_end = best_vertical_line_points_left[-1][1] proj_x = frame_vertical_start[0] + (frame_vertical_end[0] - frame_vertical_start[0]) * ratio_proj proj_y = frame_vertical_start[1] + (frame_vertical_end[1] - frame_vertical_start[1]) * ratio_proj proj_point = np.array([proj_x, proj_y]) # Calculate horizontal line points based on which vertical line we have if best_vertical_line_name_left == '10-13': # Line 10-13: can calculate points on horizontal lines 2-10, 5-13 if off_line_idx == 1: # Point 2 (index 1) is off-line, calculate point 10 (index 9) kp_10 = np.array(best_vertical_line_points_left[0][1]) # 10 point kp_2 = off_line_point + (kp_10 - proj_point) result[1] = tuple(kp_2.astype(int)) total_left_side_count += 1 all_available_points_left[1] = tuple(kp_2.astype(int)) elif off_line_idx == 4: # Point 5 (index 4) is off-line, calculate point 13 (index 12) kp_13 = np.array(best_vertical_line_points_left[-1][1]) # 13 point kp_5 = off_line_point + (kp_13 - proj_point) result[4] = tuple(kp_5.astype(int)) total_left_side_count += 1 all_available_points_left[4] = tuple(kp_5.astype(int)) elif best_vertical_line_name_left == '1-6': # Line 1-6: can calculate points on horizontal lines 2-10, 3-7, 4-8, 5-13 if off_line_idx == 6 or off_line_idx == 7: # Point 7 or 8 is off-line, calculate point 3 or 4 template_off = template_coords_left[off_line_idx] template_3 = template_coords_left[2] # 3 point, index 2 template_4 = template_coords_left[3] # 4 point, index 3 template_7 = template_coords_left[6] # 7 point, index 6 template_8 = template_coords_left[7] # 8 point, index 7 if off_line_idx == 6: # Point 7, calculate point 3 ratio = (template_3[0] - template_7[0]) / (template_7[0] - template_off[0]) if abs(template_7[0] - template_off[0]) > 1e-6 else 0.5 kp_3 = proj_point + (off_line_point - proj_point) * ratio result[2] = tuple(kp_3.astype(int)) total_left_side_count += 1 all_available_points_left[2] = tuple(kp_3.astype(int)) else: # Point 8, calculate point 4 ratio = (template_4[0] - template_8[0]) / (template_8[0] - template_off[0]) if abs(template_8[0] - template_off[0]) > 1e-6 else 0.5 kp_4 = proj_point + (off_line_point - proj_point) * ratio result[3] = tuple(kp_4.astype(int)) total_left_side_count += 1 all_available_points_left[3] = tuple(kp_4.astype(int)) elif off_line_idx == 9 or off_line_idx == 12: # Point 10 or 13 is off-line, calculate point 2 or 5 if off_line_idx == 9: # Point 10, calculate point 2 kp_2 = off_line_point + (np.array(best_vertical_line_points_left[1][1]) - proj_point) result[1] = tuple(kp_2.astype(int)) total_left_side_count += 1 all_available_points_left[1] = tuple(kp_2.astype(int)) else: # Point 13, calculate point 5 kp_5 = off_line_point + (np.array(best_vertical_line_points_left[4][1]) - proj_point) result[4] = tuple(kp_5.astype(int)) total_left_side_count += 1 all_available_points_left[4] = tuple(kp_5.astype(int)) elif best_vertical_line_name_left == '7-8': # Line 7-8: can calculate points on horizontal lines 3-7, 4-8 if off_line_idx == 2 or off_line_idx == 3: # Point 3 or 4 is off-line, calculate point 7 or 8 if off_line_idx == 2: # Point 3, calculate point 7 kp_7 = off_line_point + (np.array(best_vertical_line_points_left[0][1]) - proj_point) result[6] = tuple(kp_7.astype(int)) total_left_side_count += 1 all_available_points_left[6] = tuple(kp_7.astype(int)) else: # Point 4, calculate point 8 kp_8 = off_line_point + (np.array(best_vertical_line_points_left[-1][1]) - proj_point) result[7] = tuple(kp_8.astype(int)) total_left_side_count += 1 all_available_points_left[7] = tuple(kp_8.astype(int)) # Check if we can now form a horizontal line with the newly calculated points for line_name, (indices, line_type) in line_groups_left.items(): if line_type == 'horizontal': line_points = [(idx, all_available_points_left[idx]) for idx in indices if idx in all_available_points_left] if len(line_points) > max_horizontal_points_left: max_horizontal_points_left = len(line_points) best_horizontal_line_name_left = line_name best_horizontal_line_points_left = line_points elif best_horizontal_line_name_left is not None and best_vertical_line_name_left is None: # We have horizontal line but no vertical line # Find an off-line point (not on the horizontal line) off_line_point = None off_line_idx = None horizontal_line_indices = line_groups_left[best_horizontal_line_name_left][0] for idx, kp in all_available_points_left.items(): if idx not in horizontal_line_indices: off_line_point = kp off_line_idx = idx break if off_line_point is not None: # Project off_line_point onto horizontal line template_off_line = template_coords_left[off_line_idx] template_horizontal_start = template_coords_left[best_horizontal_line_points_left[0][0]] template_horizontal_end = template_coords_left[best_horizontal_line_points_left[-1][0]] # Project at same x as off_line_point template_x_off = template_off_line[0] template_x_horizontal_start = template_horizontal_start[0] template_x_horizontal_end = template_horizontal_end[0] if abs(template_x_horizontal_end - template_x_horizontal_start) > 1e-6: ratio_proj = (template_x_off - template_x_horizontal_start) / (template_x_horizontal_end - template_x_horizontal_start) else: ratio_proj = 0.5 frame_horizontal_start = best_horizontal_line_points_left[0][1] frame_horizontal_end = best_horizontal_line_points_left[-1][1] proj_x = frame_horizontal_start[0] + (frame_horizontal_end[0] - frame_horizontal_start[0]) * ratio_proj proj_y = frame_horizontal_start[1] + (frame_horizontal_end[1] - frame_horizontal_start[1]) * ratio_proj proj_point = np.array([proj_x, proj_y]) off_line_point = np.array(off_line_point) # Calculate vertical line points based on which horizontal line we have if best_horizontal_line_name_left == '2-10': # Line 2-10: can calculate points on vertical lines 1-6, 10-13 if off_line_idx == 0 or off_line_idx == 5: # Point 1 or 6 is off-line, calculate point 2 kp_2 = off_line_point + (np.array(best_horizontal_line_points_left[0][1]) - proj_point) result[1] = tuple(kp_2.astype(int)) total_left_side_count += 1 all_available_points_left[1] = tuple(kp_2.astype(int)) elif off_line_idx == 9 or off_line_idx == 12: # Point 10 or 13 is off-line, calculate point 10 kp_10 = off_line_point + (np.array(best_horizontal_line_points_left[-1][1]) - proj_point) result[9] = tuple(kp_10.astype(int)) total_left_side_count += 1 all_available_points_left[9] = tuple(kp_10.astype(int)) elif best_horizontal_line_name_left == '3-7': # Line 3-7: can calculate points on vertical lines 1-6, 7-8 if off_line_idx == 0 or off_line_idx == 5: # Point 1 or 6 is off-line, calculate point 3 kp_3 = off_line_point + (np.array(best_horizontal_line_points_left[0][1]) - proj_point) result[2] = tuple(kp_3.astype(int)) total_left_side_count += 1 all_available_points_left[2] = tuple(kp_3.astype(int)) elif off_line_idx == 6 or off_line_idx == 7: # Point 7 or 8 is off-line, calculate point 7 kp_7 = off_line_point + (np.array(best_horizontal_line_points_left[-1][1]) - proj_point) result[6] = tuple(kp_7.astype(int)) total_left_side_count += 1 all_available_points_left[6] = tuple(kp_7.astype(int)) elif best_horizontal_line_name_left == '4-8': # Line 4-8: can calculate points on vertical lines 1-6, 7-8 if off_line_idx == 0 or off_line_idx == 5: # Point 1 or 6 is off-line, calculate point 4 kp_4 = off_line_point + (np.array(best_horizontal_line_points_left[0][1]) - proj_point) result[3] = tuple(kp_4.astype(int)) total_left_side_count += 1 all_available_points_left[3] = tuple(kp_4.astype(int)) elif off_line_idx == 6 or off_line_idx == 7: # Point 7 or 8 is off-line, calculate point 8 kp_8 = off_line_point + (np.array(best_horizontal_line_points_left[-1][1]) - proj_point) result[7] = tuple(kp_8.astype(int)) total_left_side_count += 1 all_available_points_left[7] = tuple(kp_8.astype(int)) elif best_horizontal_line_name_left == '5-13': # Line 5-13: can calculate points on vertical lines 1-6, 10-13 if off_line_idx == 0 or off_line_idx == 5: # Point 1 or 6 is off-line, calculate point 5 kp_5 = off_line_point + (np.array(best_horizontal_line_points_left[0][1]) - proj_point) result[4] = tuple(kp_5.astype(int)) total_left_side_count += 1 all_available_points_left[4] = tuple(kp_5.astype(int)) elif off_line_idx == 9 or off_line_idx == 12: # Point 10 or 13 is off-line, calculate point 13 kp_13 = off_line_point + (np.array(best_horizontal_line_points_left[-1][1]) - proj_point) result[12] = tuple(kp_13.astype(int)) total_left_side_count += 1 all_available_points_left[12] = tuple(kp_13.astype(int)) # Check if we can now form a vertical line with the newly calculated points for line_name, (indices, line_type) in line_groups_left.items(): if line_type == 'vertical': line_points = [(idx, all_available_points_left[idx]) for idx in indices if idx in all_available_points_left] if len(line_points) > max_vertical_points_left: max_vertical_points_left = len(line_points) best_vertical_line_name_left = line_name best_vertical_line_points_left = line_points # Calculate keypoint 9 if we have at least one line if best_vertical_line_name_left is not None and best_horizontal_line_name_left is not None: if kp_9 is None: print(f"Calculating keypoint 9 using both vertical and horizontal lines: {best_vertical_line_name_left} and {best_horizontal_line_name_left}") template_x_9 = 110 template_y_9 = 340 # Project keypoint 9 onto vertical line template_vertical_start = template_coords_left[best_vertical_line_points_left[0][0]] template_vertical_end = template_coords_left[best_vertical_line_points_left[-1][0]] # Project at y=340 (same y as keypoint 9) template_y_vertical_start = template_vertical_start[1] template_y_vertical_end = template_vertical_end[1] if abs(template_y_vertical_end - template_y_vertical_start) > 1e-6: ratio_9_vertical = (template_y_9 - template_y_vertical_start) / (template_y_vertical_end - template_y_vertical_start) else: ratio_9_vertical = 0.5 frame_vertical_start = best_vertical_line_points_left[0][1] frame_vertical_end = best_vertical_line_points_left[-1][1] proj_9_on_vertical_x = frame_vertical_start[0] + (frame_vertical_end[0] - frame_vertical_start[0]) * ratio_9_vertical proj_9_on_vertical_y = frame_vertical_start[1] + (frame_vertical_end[1] - frame_vertical_start[1]) * ratio_9_vertical proj_9_on_vertical = (proj_9_on_vertical_x, proj_9_on_vertical_y) # Project keypoint 9 onto horizontal line template_horizontal_start = template_coords_left[best_horizontal_line_points_left[0][0]] template_horizontal_end = template_coords_left[best_horizontal_line_points_left[-1][0]] # Project at x=110 (same x as keypoint 9) template_x_horizontal_start = template_horizontal_start[0] template_x_horizontal_end = template_horizontal_end[0] if abs(template_x_horizontal_end - template_x_horizontal_start) > 1e-6: ratio_9_horizontal = (template_x_9 - template_x_horizontal_start) / (template_x_horizontal_end - template_x_horizontal_start) else: ratio_9_horizontal = 0.5 frame_horizontal_start = best_horizontal_line_points_left[0][1] frame_horizontal_end = best_horizontal_line_points_left[-1][1] proj_9_on_horizontal_x = frame_horizontal_start[0] + (frame_horizontal_end[0] - frame_horizontal_start[0]) * ratio_9_horizontal proj_9_on_horizontal_y = frame_horizontal_start[1] + (frame_horizontal_end[1] - frame_horizontal_start[1]) * ratio_9_horizontal proj_9_on_horizontal = (proj_9_on_horizontal_x, proj_9_on_horizontal_y) # Calculate keypoint 9 as intersection of two lines # Line 1: Passes through proj_9_on_vertical, parallel to best_horizontal_line # Line 2: Passes through proj_9_on_horizontal, parallel to best_vertical_line # Calculate direction vector of best_horizontal_line horizontal_dir_x = frame_horizontal_end[0] - frame_horizontal_start[0] horizontal_dir_y = frame_horizontal_end[1] - frame_horizontal_start[1] horizontal_dir_length = np.sqrt(horizontal_dir_x**2 + horizontal_dir_y**2) # Calculate direction vector of best_vertical_line vertical_dir_x = frame_vertical_end[0] - frame_vertical_start[0] vertical_dir_y = frame_vertical_end[1] - frame_vertical_start[1] vertical_dir_length = np.sqrt(vertical_dir_x**2 + vertical_dir_y**2) if horizontal_dir_length > 1e-6 and vertical_dir_length > 1e-6: # Normalize direction vectors horizontal_dir_x /= horizontal_dir_length horizontal_dir_y /= horizontal_dir_length vertical_dir_x /= vertical_dir_length vertical_dir_y /= vertical_dir_length # Find intersection: proj_9_on_vertical + t * horizontal_dir = proj_9_on_horizontal + s * vertical_dir A = np.array([ [horizontal_dir_x, -vertical_dir_x], [horizontal_dir_y, -vertical_dir_y] ]) b = np.array([ proj_9_on_horizontal[0] - proj_9_on_vertical[0], proj_9_on_horizontal[1] - proj_9_on_vertical[1] ]) try: t, s = np.linalg.solve(A, b) # Calculate intersection point using line 1 x_9 = proj_9_on_vertical[0] + t * horizontal_dir_x y_9 = proj_9_on_vertical[1] + t * horizontal_dir_y result[8] = (int(round(x_9)), int(round(y_9))) total_left_side_count += 1 except np.linalg.LinAlgError: # Lines are parallel or nearly parallel, use simple intersection x_9 = proj_9_on_vertical[0] y_9 = proj_9_on_horizontal[1] result[8] = (int(round(x_9)), int(round(y_9))) total_left_side_count += 1 else: # Fallback: use simple intersection x_9 = proj_9_on_vertical[0] y_9 = proj_9_on_horizontal[1] result[8] = (int(round(x_9)), int(round(y_9))) total_left_side_count += 1 print(f"total_left_side_count: {total_left_side_count}, result: {result}") if total_left_side_count > 5: pass # Continue to right side logic # Calculate m_line and b_line from best vertical or horizontal line for use in calculating other points m_line_left = None b_line_left = None best_line_for_calc_left = None best_line_type_for_calc_left = None if best_vertical_line_name_left is not None and len(best_vertical_line_points_left) >= 2: best_line_for_calc_left = best_vertical_line_points_left best_line_type_for_calc_left = 'vertical' points_array = np.array([[kp[0], kp[1]] for _, kp in best_vertical_line_points_left]) x_coords = points_array[:, 0] y_coords = points_array[:, 1] A = np.vstack([x_coords, np.ones(len(x_coords))]).T m_line_left, b_line_left = np.linalg.lstsq(A, y_coords, rcond=None)[0] elif best_horizontal_line_name_left is not None and len(best_horizontal_line_points_left) >= 2: best_line_for_calc_left = best_horizontal_line_points_left best_line_type_for_calc_left = 'horizontal' points_array = np.array([[kp[0], kp[1]] for _, kp in best_horizontal_line_points_left]) x_coords = points_array[:, 0] y_coords = points_array[:, 1] A = np.vstack([x_coords, np.ones(len(x_coords))]).T m_line_left, b_line_left = np.linalg.lstsq(A, y_coords, rcond=None)[0] # Calculate missing points to reach exactly 5 points # Ensure 4 points aren't all on one line if total_left_side_count < 5 and (m_line_left is not None or (best_line_for_calc_left is not None and best_line_type_for_calc_left == 'vertical')): # Check current distribution counts_per_line = [ len(line_1_6_points), len(line_7_8_points), len(line_10_13_points) ] # Calculate points on line 1-6 if needed template_ys_1_6 = [5, 140, 250, 430, 540, 675] template_indices_1_6 = [0, 1, 2, 3, 4, 5] if best_vertical_line_name_left == '10-13': # Construct parallel line 1-6 from line 10-13 for template_y, idx in zip(template_ys_1_6, template_indices_1_6): if result[idx] is None and total_left_side_count < 5: # Check if adding this point would put 4 on one line new_counts = counts_per_line.copy() new_counts[0] += 1 # Adding to line 1-6 if max(new_counts) >= 4 and total_left_side_count == 4: # Would have 4 on one line, skip continue # Calculate y using scale from template ref_ys = [kp[1] for _, kp in line_10_13_points] ref_template_ys = [140, 270, 410, 540] ref_indices = [9, 10, 11, 12] matched_template_ys = [] for ref_idx, ref_kp in line_10_13_points: if ref_idx in ref_indices: template_idx = ref_indices.index(ref_idx) matched_template_ys.append((ref_template_ys[template_idx], ref_kp[1])) if len(matched_template_ys) >= 1: ref_template_y, ref_frame_y = matched_template_ys[0] if ref_template_y > 0: scale = ref_frame_y / ref_template_y y_new = int(round(template_y * scale)) else: y_new = ref_frame_y else: y_new = int(round(np.median(ref_ys))) if ref_ys else template_y # Calculate x using parallel line geometry if abs(m_line_left) > 1e-6: x_on_line_10_13 = (y_new - b_line_left) / m_line_left x_new = int(round(x_on_line_10_13 * 0.0303)) # 5/165 else: x_new = int(round(np.median([kp[0] for _, kp in line_10_13_points]) * 0.0303)) result[idx] = (x_new, y_new) total_left_side_count += 1 if total_left_side_count >= 5: break elif best_vertical_line_name_left == '1-6': # Calculate missing points on line 1-6 for template_y, idx in zip(template_ys_1_6, template_indices_1_6): if result[idx] is None and total_left_side_count < 5: # Check if adding this point would put 4 on one line new_counts = counts_per_line.copy() new_counts[0] += 1 # Adding to line 1-6 if max(new_counts) >= 4 and total_left_side_count == 4: # Would have 4 on one line, skip continue # Calculate x on the line if abs(m_line_left) > 1e-6: x_new = (template_y - b_line_left) / m_line_left else: x_new = np.median([kp[0] for _, kp in line_1_6_points]) # Scale y based on available points ref_ys = [kp[1] for _, kp in line_1_6_points] ref_template_ys = [] for ref_idx, _ in line_1_6_points: if ref_idx in template_indices_1_6: template_idx = template_indices_1_6.index(ref_idx) ref_template_ys.append(template_ys_1_6[template_idx]) if len(ref_ys) >= 1 and len(ref_template_ys) >= 1: ref_template_y = ref_template_ys[0] ref_frame_y = ref_ys[0] if ref_template_y > 0: scale = ref_frame_y / ref_template_y y_new = int(round(template_y * scale)) else: y_new = ref_frame_y else: y_new = int(round(np.median(ref_ys))) if ref_ys else template_y result[idx] = (int(round(x_new)), y_new) total_left_side_count += 1 if total_left_side_count >= 5: break print(f"total_left_side_count: {total_left_side_count}, result: {result}") # Case 2: Unified handling of right side keypoints (18-30) # Three parallel lines on right side: # - Line 18-21: keypoints 18, 19, 20, 21 (indices 17-20) # - Line 23-24: keypoints 23, 24 (indices 22-23) # - Line 25-30: keypoints 25, 26, 27, 28, 29, 30 (indices 24-29) # Keypoint 22 (index 21) is between line 18-21 and line 25-30 # Collect all right-side keypoints (18-30, indices 17-29) right_side_all = [] line_18_21_points = [] # Indices 17-20 line_23_24_points = [] # Indices 22-23 line_25_30_points = [] # Indices 24-29 for idx in range(17, 30): # Keypoints 18-30 (indices 17-29) kp = get_kp(idx) if kp: right_side_all.append((idx, kp)) if 17 <= idx <= 20: # Line 18-21 line_18_21_points.append((idx, kp)) elif 22 <= idx <= 23: # Line 23-24 line_23_24_points.append((idx, kp)) elif 24 <= idx <= 29: # Line 25-30 line_25_30_points.append((idx, kp)) kp_22 = get_kp(21) # Keypoint 22 if kp_22: right_side_all.append((21, kp_22)) total_right_side_count = len(right_side_all) # If we have 6 or more points, no need to calculate more if total_right_side_count >= 6: pass # Don't calculate more points elif total_right_side_count == 5: # Check if 4 points are on one line and 1 on another line counts_per_line = [ len(line_18_21_points), len(line_23_24_points), len(line_25_30_points) ] if max(counts_per_line) == 4 and sum(counts_per_line) == 4: # 4 points on one line, need to calculate 1 more point on another line # Determine which line has 4 points and calculate on a different line if len(line_18_21_points) == 4: # All 4 on line 18-21, calculate on line 25-30 or 23-24 # Prefer line 25-30 (right edge) if len(line_25_30_points) == 0: # Calculate a point on line 25-30 # Fit line through 18-21 points points_18_21 = np.array([[kp[0], kp[1]] for _, kp in line_18_21_points]) x_coords = points_18_21[:, 0] y_coords = points_18_21[:, 1] A = np.vstack([x_coords, np.ones(len(x_coords))]).T m_18_21, b_18_21 = np.linalg.lstsq(A, y_coords, rcond=None)[0] # Calculate a point on line 25-30 (parallel to 18-21) # Use template y-coordinate for one of 25-30 points template_ys_25_30 = [5, 140, 250, 430, 540, 675] # Template y for 25-30 template_indices_25_30 = [24, 25, 26, 27, 28, 29] # Use median y from 18-21 points to estimate scale median_y = np.median(y_coords) # Find closest template y ref_template_y = min(template_ys_25_30, key=lambda ty: abs(ty - np.median([kp[1] for _, kp in line_18_21_points]))) ref_idx = template_ys_25_30.index(ref_template_y) # Calculate y for the new point y_new = int(round(median_y)) # Calculate x using parallel line geometry # In template: line 25-30 is at x=1045, line 18-21 is at x=888 # Ratio: 1045/888 ≈ 1.177 if abs(m_18_21) > 1e-6: x_on_line_18_21 = (y_new - b_18_21) / m_18_21 x_new = int(round(x_on_line_18_21 * 1.177)) else: x_new = int(round(np.median(x_coords) * 1.177)) # Find first missing index in 25-30 range for template_y, idx in zip(template_ys_25_30, template_indices_25_30): if result[idx] is None: result[idx] = (x_new, y_new) break elif len(line_25_30_points) == 4: # All 4 on line 25-30, calculate on line 18-21 # Similar logic but in reverse points_25_30 = np.array([[kp[0], kp[1]] for _, kp in line_25_30_points]) x_coords = points_25_30[:, 0] y_coords = points_25_30[:, 1] A = np.vstack([x_coords, np.ones(len(x_coords))]).T m_25_30, b_25_30 = np.linalg.lstsq(A, y_coords, rcond=None)[0] # Calculate a point on line 18-21 template_ys_18_21 = [140, 270, 410, 540] # Template y for 18-21 template_indices_18_21 = [17, 18, 19, 20] median_y = np.median(y_coords) # Calculate x using parallel line geometry # Ratio: 888/1045 ≈ 0.850 if abs(m_25_30) > 1e-6: x_on_line_25_30 = (median_y - b_25_30) / m_25_30 x_new = int(round(x_on_line_25_30 * 0.850)) else: x_new = int(round(np.median(x_coords) * 0.850)) for template_y, idx in zip(template_ys_18_21, template_indices_18_21): if result[idx] is None: result[idx] = (x_new, int(round(median_y))) break elif total_right_side_count < 5: # Need to calculate missing keypoints to get exactly 5 points # Requirements: # 1. Must have keypoint 22 # 2. 4 points shouldn't be all on one line (need distribution) # Template coordinates for reference template_coords = { 17: (888, 140), # 18 18: (888, 270), # 19 19: (888, 410), # 20 20: (888, 540), # 21 21: (940, 340), # 22 (what we're calculating) 22: (998, 250), # 23 23: (998, 430), # 24 24: (1045, 5), # 25 25: (1045, 140), # 26 26: (1045, 250), # 27 27: (1045, 430), # 28 28: (1045, 540), # 29 29: (1045, 675), # 30 } # Define line groups (vertical and horizontal lines) # Vertical lines: 18-21, 23-24, 25-30 # Horizontal lines: 18-26, 23-27, 24-28, 21-29 line_groups = { '18-21': ([17, 18, 19, 20], 'vertical'), # indices: 18, 19, 20, 21 '23-24': ([22, 23], 'vertical'), # indices: 23, 24 '25-30': ([24, 25, 26, 27, 28, 29], 'vertical'), # indices: 25, 26, 27, 28, 29, 30 '18-26': ([17, 25], 'horizontal'), # indices: 18, 26 '23-27': ([22, 26], 'horizontal'), # indices: 23, 27 '24-28': ([23, 27], 'horizontal'), # indices: 24, 28 '21-29': ([20, 28], 'horizontal'), # indices: 21, 29 } # Collect all available points with their indices all_available_points = {} for idx, kp in line_18_21_points: all_available_points[idx] = kp for idx, kp in line_23_24_points: all_available_points[idx] = kp for idx, kp in line_25_30_points: all_available_points[idx] = kp # Step 1: Find the best vertical line and best horizontal line separately best_vertical_line_name = None best_vertical_line_points = [] max_vertical_points = 1 best_horizontal_line_name = None best_horizontal_line_points = [] max_horizontal_points = 1 for line_name, (indices, line_type) in line_groups.items(): line_points = [(idx, all_available_points[idx]) for idx in indices if idx in all_available_points] if line_type == 'vertical' and len(line_points) > max_vertical_points: max_vertical_points = len(line_points) best_vertical_line_name = line_name best_vertical_line_points = line_points elif line_type == 'horizontal' and len(line_points) > max_horizontal_points: max_horizontal_points = len(line_points) best_horizontal_line_name = line_name best_horizontal_line_points = line_points # Check and calculate missing points on detected lines # For vertical lines if best_vertical_line_name is not None: expected_indices = line_groups[best_vertical_line_name][0] detected_indices = {idx for idx, _ in best_vertical_line_points} missing_indices = [idx for idx in expected_indices if idx not in detected_indices] if len(missing_indices) > 0: # Calculate missing points using template ratios template_start = template_coords[best_vertical_line_points[0][0]] template_end = template_coords[best_vertical_line_points[-1][0]] frame_start = best_vertical_line_points[0][1] frame_end = best_vertical_line_points[-1][1] for missing_idx in missing_indices: template_missing = template_coords[missing_idx] # Calculate ratio along the line based on y-coordinate (vertical line) template_y_start = template_start[1] template_y_end = template_end[1] template_y_missing = template_missing[1] if abs(template_y_end - template_y_start) > 1e-6: ratio = (template_y_missing - template_y_start) / (template_y_end - template_y_start) else: ratio = 0.5 # Calculate frame coordinates x_new = frame_start[0] + (frame_end[0] - frame_start[0]) * ratio y_new = frame_start[1] + (frame_end[1] - frame_start[1]) * ratio new_point = (int(round(x_new)), int(round(y_new))) # Add to result and update collections result[missing_idx] = new_point best_vertical_line_points.append((missing_idx, new_point)) all_available_points[missing_idx] = new_point total_right_side_count += 1 max_vertical_points = len(best_vertical_line_points) # Sort by index to maintain order best_vertical_line_points.sort(key=lambda x: x[0]) # Check if we can now form a horizontal line with the newly calculated points for line_name, (indices, line_type) in line_groups.items(): if line_type == 'horizontal': line_points = [(idx, all_available_points[idx]) for idx in indices if idx in all_available_points] if len(line_points) > max_horizontal_points: max_horizontal_points = len(line_points) best_horizontal_line_name = line_name best_horizontal_line_points = line_points # For horizontal lines if best_horizontal_line_name is not None: expected_indices = line_groups[best_horizontal_line_name][0] detected_indices = {idx for idx, _ in best_horizontal_line_points} missing_indices = [idx for idx in expected_indices if idx not in detected_indices] if len(missing_indices) > 0: # Calculate missing points using template ratios template_start = template_coords[best_horizontal_line_points[0][0]] template_end = template_coords[best_horizontal_line_points[-1][0]] frame_start = best_horizontal_line_points[0][1] frame_end = best_horizontal_line_points[-1][1] for missing_idx in missing_indices: template_missing = template_coords[missing_idx] # Calculate ratio along the line based on x-coordinate (horizontal line) template_x_start = template_start[0] template_x_end = template_end[0] template_x_missing = template_missing[0] if abs(template_x_end - template_x_start) > 1e-6: ratio = (template_x_missing - template_x_start) / (template_x_end - template_x_start) else: ratio = 0.5 # Calculate frame coordinates x_new = frame_start[0] + (frame_end[0] - frame_start[0]) * ratio y_new = frame_start[1] + (frame_end[1] - frame_start[1]) * ratio new_point = (int(round(x_new)), int(round(y_new))) # Add to result and update collections result[missing_idx] = new_point best_horizontal_line_points.append((missing_idx, new_point)) all_available_points[missing_idx] = new_point total_right_side_count += 1 max_horizontal_points = len(best_horizontal_line_points) # Sort by index to maintain order best_horizontal_line_points.sort(key=lambda x: x[0]) # Check if we can now form a vertical line with the newly calculated points for line_name, (indices, line_type) in line_groups.items(): if line_type == 'vertical': line_points = [(idx, all_available_points[idx]) for idx in indices if idx in all_available_points] if len(line_points) > max_vertical_points: max_vertical_points = len(line_points) best_vertical_line_name = line_name best_vertical_line_points = line_points # If we only have one direction, try to calculate the other direction line if best_vertical_line_name is not None and best_horizontal_line_name is None: # possible cases: # line is 25-30 and off line point is 19, then we can calculate 18 so get horizontal line 18-26 # line is 25-30 and off line point is 20, then we can calculate 18 so get horizontal line 18-26 # line is 18-21 and off line point is 23, then we can calculate 27 so get horizontal line 23-27 # line is 18-21 and off line point is 24, then we can calculate 28 so get horizontal line 24-28 # line is 18-21 and off line point is 25, then we can calculate 26 so get horizontal line 18-26 # line is 18-21 and off line point is 27, then we can calculate 26 so get horizontal line 18-26 # line is 18-21 and off line point is 28, then we can calculate 29 so get horizontal line 21-29 # line is 18-21 and off line point is 30, then we can calculate 29 so get horizontal line 21-29 # line is 23-24 and off line point is 18, then we can calculate 26 so get horizontal line 18-26 # line is 23-24 and off line point is 19, then we can calculate 18 so get horizontal line 18-26 # line is 23-24 and off line point is 20, then we can calculate 21 so get horizontal line 21-29 # line is 23-24 and off line point is 21, then we can calculate 29 so get horizontal line 21-29 # line is 23-24 and off line point is 25, then we can calculate 27 so get horizontal line 23-27 # line is 23-24 and off line point is 26, then we can calculate 27 so get horizontal line 23-27 # line is 23-24 and off line point is 29, then we can calculate 28 so get horizontal line 24-28 # line is 23-24 and off line point is 30, then we can calculate 28 so get horizontal line 24-28 # We have vertical line but no horizontal line # Find an off-line point (not on the vertical line) off_line_point = None off_line_idx = None vertical_line_indices = line_groups[best_vertical_line_name][0] for idx, kp in all_available_points.items(): if idx not in vertical_line_indices: off_line_point = kp off_line_idx = idx break if off_line_point is not None: # Convert off_line_point to numpy array for arithmetic operations off_line_point = np.array(off_line_point) # Project off_line_point onto vertical line template_off_line = template_coords[off_line_idx] template_vertical_start_index = best_vertical_line_points[0][0] template_vertical_end_index = best_vertical_line_points[-1][0] template_vertical_start = template_coords[template_vertical_start_index] template_vertical_end = template_coords[template_vertical_end_index] # Project at same y as off_line_point template_y_off = template_off_line[1] template_y_vertical_start = template_vertical_start[1] template_y_vertical_end = template_vertical_end[1] if abs(template_y_vertical_end - template_y_vertical_start) > 1e-6: ratio_proj = (template_y_off - template_y_vertical_start) / (template_y_vertical_end - template_y_vertical_start) else: ratio_proj = 0.5 frame_vertical_start = best_vertical_line_points[0][1] frame_vertical_end = best_vertical_line_points[-1][1] proj_x = frame_vertical_start[0] + (frame_vertical_end[0] - frame_vertical_start[0]) * ratio_proj proj_y = frame_vertical_start[1] + (frame_vertical_end[1] - frame_vertical_start[1]) * ratio_proj proj_point = np.array([proj_x, proj_y]) if best_vertical_line_name == '25-30' and len(best_vertical_line_points) == 6: if off_line_idx == 18 or off_line_idx == 19: # 19 or 20 point is off line point, so we can calculate 18 kp_26 = np.array(best_vertical_line_points[1][1]) # 26 point kp_18 = off_line_point + (kp_26 - proj_point) result[17] = tuple(kp_18.astype(int)) total_right_side_count += 1 all_available_points[17] = tuple(kp_18.astype(int)) # 18 point is now available, index is 17 if best_vertical_line_name == '18-21' and len(best_vertical_line_points) == 4: if off_line_idx == 22 or off_line_idx == 23: # 23 or 24 point is off line point, so we can calculate 27 template_19 = template_coords[18] # 19 point, index is 18 template_23 = template_coords[22] # 23 point, index is 22 template_27 = template_coords[26] # 27 point, index is 26 ratio = (template_27[0] - template_19[0]) / (template_23[0] - template_19[0]) # ratio in x coordinates because y coordinates are the same expected_point = proj_point + (off_line_point - proj_point) * ratio if off_line_idx == 22: result[26] = tuple(expected_point.astype(int)) # 27 point, index is 26 total_right_side_count += 1 all_available_points[26] = tuple(expected_point.astype(int)) # 27 point is now available, index is 26 else: result[27] = tuple(expected_point.astype(int)) # 28 point, index is 27 total_right_side_count += 1 all_available_points[27] = tuple(expected_point.astype(int)) # 28 point is now available, index is 27 if off_line_idx == 24 or off_line_idx == 26: # 25 or 27 point is off line point, so we can calculate 26 kp_18 = np.array(best_vertical_line_points[0][1]) # 18 point kp_26 = off_line_point + (kp_18 - proj_point) result[25] = tuple(kp_26.astype(int)) total_right_side_count += 1 all_available_points[25] = tuple(kp_26.astype(int)) # 26 point is now available, index is 25 if off_line_idx == 27 or off_line_idx == 29: # 28 or 30 point is off line point, so we can calculate 29 kp_21 = np.array(best_vertical_line_points[-1][1]) # 21 point kp_29 = off_line_point + (kp_21 - proj_point) result[28] = tuple(kp_29.astype(int)) total_right_side_count += 1 all_available_points[28] = tuple(kp_29.astype(int)) # 29 point is now available, index is 28 if best_vertical_line_name == '23-24' and len(best_vertical_line_points) == 2: if off_line_idx == 17 or off_line_idx == 18 or off_line_idx == 19 or off_line_idx == 20: # 18 or 19 or 20 or 21 point is off line point, so we can calculate 26 template_18 = template_coords[17] # 18 point, index is 17 template_26 = template_coords[25] # 26 point, index is 25 template_23 = template_coords[22] # 23 point, index is 22 ratio_26 = (template_26[0] - template_18[0]) / (template_23[0] - template_18[0]) # ratio in x coordinates because y coordinates are the same kp_18 = None if off_line_idx == 17: kp_18 = off_line_point elif off_line_idx == 18 or off_line_idx == 19 or off_line_idx == 20: template_off_line = template_coords[off_line_idx] ratio = (template_18[1] - template_off_line[1]) / (template_23[1] - template_off_line[1]) kp_18 = off_line_point + (np.array(best_vertical_line_points[0][1]) - proj_point) * ratio if kp_18 is not None: kp_26 = kp_18 + (proj_point - off_line_point) * ratio_26 result[25] = tuple(kp_26.astype(int)) total_right_side_count += 1 all_available_points[25] = tuple(kp_26.astype(int)) # 26 point is now available, index is 25 if off_line_idx == 24 or off_line_idx == 25: # 25 or 26 point is off line point, so we can calculate 27 kp_27 = off_line_point + (np.array(best_vertical_line_points[0][1]) - proj_point) result[26] = tuple(kp_27.astype(int)) total_right_side_count += 1 all_available_points[26] = tuple(kp_27.astype(int)) # 27 point is now available, index is 26 if off_line_idx == 28 or off_line_idx == 29: # 29 or 30 point is off line point, so we can calculate 29 kp_29 = off_line_point + (np.array(best_vertical_line_points[-1][1]) - proj_point) result[28] = tuple(kp_29.astype(int)) total_right_side_count += 1 all_available_points[28] = tuple(kp_29.astype(int)) # 29 point is now available, index is 28 # Check if we can now form a horizontal line with the newly calculated points for line_name, (indices, line_type) in line_groups.items(): if line_type == 'horizontal': line_points = [(idx, all_available_points[idx]) for idx in indices if idx in all_available_points] if len(line_points) > max_horizontal_points: max_horizontal_points = len(line_points) best_horizontal_line_name = line_name best_horizontal_line_points = line_points elif best_horizontal_line_name is not None and best_vertical_line_name is None: # possible cases: # line is 18-26 and off line point is 23, then we can calculate 27 so get vertical line 25-30 # line is 18-26 and off line point is 24, then we can calculate 28 so get vertical line 25-30 # line is 23-27 and off line point is 18, then we can calculate 26 so get vertical line 25-30 # line is 23-27 and off line point is 19, then we can calculate 18 so get vertical line 18-21 # line is 23-27 and off line point is 20, then we can calculate 18 so get vertical line 18-21 # line is 23-27 and off line point is 21, then we can calculate 29 so get vertical line 25-30 # line is 24-28 and off line point is 18, then we can calculate 26 so get vertical line 25-30 # line is 24-28 and off line point is 19, then we can calculate 21 so get vertical line 18-21 # line is 24-28 and off line point is 20, then we can calculate 21 so get vertical line 18-21 # line is 24-28 and off line point is 21, then we can calculate 29 so get vertical line 25-30 # line is 21-29 and off line point is 23, then we can calculate 27 so get vertical line 25-30 # line is 21-29 and off line point is 24, then we can calculate 28 so get vertical line 25-30 # We have horizontal line but no vertical line # Find an off-line point (not on the horizontal line) off_line_point = None off_line_idx = None horizontal_line_indices = line_groups[best_horizontal_line_name][0] for idx, kp in all_available_points.items(): if idx not in horizontal_line_indices: off_line_point = kp off_line_idx = idx break if off_line_point is not None: # Project off_line_point onto horizontal line template_off_line = template_coords[off_line_idx] template_horizontal_start = template_coords[best_horizontal_line_points[0][0]] template_horizontal_end = template_coords[best_horizontal_line_points[-1][0]] # Project at same x as off_line_point template_x_off = template_off_line[0] template_x_horizontal_start = template_horizontal_start[0] template_x_horizontal_end = template_horizontal_end[0] if abs(template_x_horizontal_end - template_x_horizontal_start) > 1e-6: ratio_proj = (template_x_off - template_x_horizontal_start) / (template_x_horizontal_end - template_x_horizontal_start) else: ratio_proj = 0.5 frame_horizontal_start = best_horizontal_line_points[0][1] frame_horizontal_end = best_horizontal_line_points[-1][1] proj_x = frame_horizontal_start[0] + (frame_horizontal_end[0] - frame_horizontal_start[0]) * ratio_proj proj_y = frame_horizontal_start[1] + (frame_horizontal_end[1] - frame_horizontal_start[1]) * ratio_proj proj_point = np.array([proj_x, proj_y]) if best_horizontal_line_name == '18-26': if off_line_idx == 22 or off_line_idx == 23: # 23 or 24 point is off line point, so we can calculate 27 or 28 template_18 = template_coords[best_horizontal_line_points[0][0]] # 18 point, index is 17 template_26 = template_coords[best_horizontal_line_points[-1][0]] # 26 point, index is 25 template_23 = template_coords[off_line_idx] # 23 or 24 point, index is 22 or 23 ratio_26 = (template_26[0] - template_23[0]) / (template_26[0] - template_18[0]) # ratio in x coordinates because y coordinates are the same detected_point = off_line_point + (np.array(best_horizontal_line_points[-1][1]) - np.array(best_horizontal_line_points[0][1])) * ratio_26 if off_line_idx == 22: result[26] = tuple(detected_point.astype(int)) total_right_side_count += 1 all_available_points[26] = tuple(detected_point.astype(int)) # 26 point is now available, index is 26 else: result[27] = tuple(detected_point.astype(int)) total_right_side_count += 1 all_available_points[27] = tuple(detected_point.astype(int)) # 27 point is now available, index is 27 if best_horizontal_line_name == '23-27': if off_line_idx == 17 or off_line_idx == 20: template_18 = template_coords[17] # 18 point, index is 17 template_26 = template_coords[25] # 26 point, index is 25 template_23 = template_coords[best_horizontal_line_points[0][0]] # 23 , index is 22 ratio_26 = (template_26[0] - template_18[0]) / (template_26[0] - template_23[0]) # ratio in x coordinates because y coordinates are the same detected_point = off_line_point + (np.array(best_horizontal_line_points[-1][1]) - np.array(best_horizontal_line_points[0][1])) * ratio_26 if off_line_idx == 17: result[25] = tuple(detected_point.astype(int)) total_right_side_count += 1 all_available_points[25] = tuple(detected_point.astype(int)) # 26 point is now available, index is 25 else: result[28] = tuple(detected_point.astype(int)) total_right_side_count += 1 all_available_points[28] = tuple(detected_point.astype(int)) # 29 point is now available, index is 28 if off_line_idx == 18 or off_line_idx == 19: # 19 or 20 point is off line point, so we can calculate 18 template_18 = template_coords[17] # 18 point, index is 17 template_off_line = template_coords[off_line_idx] template_23 = template_coords[best_horizontal_line_points[0][0]] # 23 point, index is 22 ratio = (template_off_line[1] - template_18[1]) / (template_off_line[1] - template_23[1]) kp_18 = off_line_point + (proj_point - off_line_point) * ratio result[17] = tuple(kp_18.astype(int)) total_right_side_count += 1 all_available_points[17] = tuple(kp_18.astype(int)) # 18 point is now available, index is 17 if best_horizontal_line_name == '24-28': if off_line_idx == 17 or off_line_idx == 20: template_18 = template_coords[17] # 18 point, index is 17 template_26 = template_coords[25] # 26 point, index is 25 template_24 = template_coords[best_horizontal_line_points[0][0]] # 24 , index is 23 ratio_26 = (template_26[0] - template_18[0]) / (template_26[0] - template_24[0]) # ratio in x coordinates because y coordinates are the same detected_point = off_line_point + (np.array(best_horizontal_line_points[-1][1]) - np.array(best_horizontal_line_points[0][1])) * ratio_26 if off_line_idx == 17: result[25] = tuple(detected_point.astype(int)) total_right_side_count += 1 all_available_points[25] = tuple(detected_point.astype(int)) # 26 point is now available, index is 25 else: result[28] = tuple(detected_point.astype(int)) total_right_side_count += 1 all_available_points[28] = tuple(detected_point.astype(int)) # 29 point is now available, index is 28 if off_line_idx == 18 or off_line_idx == 19: # 19 or 20 point is off line point, so we can calculate 18 template_21 = template_coords[20] # 21 point, index is 20 template_off_line = template_coords[off_line_idx] template_24 = template_coords[best_horizontal_line_points[0][0]] # 24 point, index is 23 ratio = (template_21[1] - template_off_line[1]) / (template_24[1] - template_off_line[1]) kp_21 = off_line_point + (proj_point - off_line_point) * ratio result[20] = tuple(kp_18.astype(int)) total_right_side_count += 1 all_available_points[20] = tuple(kp_18.astype(int)) # 21 point is now available, index is 20 if best_horizontal_line_name == '21-29': if off_line_idx == 22 or off_line_idx == 23: # 23 or 24 point is off line point, so we can calculate 27 or 28 template_21 = template_coords[best_horizontal_line_points[0][0]] # 21 point, index is 20 template_29 = template_coords[best_horizontal_line_points[-1][0]] # 29 point, index is 28 template_23 = template_coords[off_line_idx] # 23 or 24 point, index is 22 or 23 ratio_29 = (template_29[0] - template_23[0]) / (template_29[0] - template_21[0]) # ratio in x coordinates because y coordinates are the same detected_point = off_line_point + (np.array(best_horizontal_line_points[-1][1]) - np.array(best_horizontal_line_points[0][1])) * ratio_29 if off_line_idx == 22: result[26] = tuple(detected_point.astype(int)) total_right_side_count += 1 all_available_points[26] = tuple(detected_point.astype(int)) # 26 point is now available, index is 26 else: result[27] = tuple(detected_point.astype(int)) total_right_side_count += 1 all_available_points[27] = tuple(detected_point.astype(int)) # 27 point is now available, index is 27 # Check if we can now form a vertical line with the newly calculated points for line_name, (indices, line_type) in line_groups.items(): if line_type == 'vertical': line_points = [(idx, all_available_points[idx]) for idx in indices if idx in all_available_points] if len(line_points) > max_vertical_points: max_vertical_points = len(line_points) best_vertical_line_name = line_name best_vertical_line_points = line_points # Calculate keypoint 22 if we have at least one line if best_vertical_line_name is not None and best_horizontal_line_name is not None: if kp_22 is None: print(f"Calculating keypoint 22 using both vertical and horizontal lines: {best_vertical_line_name} and {best_horizontal_line_name}") template_x_22 = 940 template_y_22 = 340 # Step 2: Project keypoint 22 onto vertical line (if available) template_vertical_start = template_coords[best_vertical_line_points[0][0]] template_vertical_end = template_coords[best_vertical_line_points[-1][0]] # Project at y=340 (same y as keypoint 22) template_y_vertical_start = template_vertical_start[1] template_y_vertical_end = template_vertical_end[1] if abs(template_y_vertical_end - template_y_vertical_start) > 1e-6: ratio_22_vertical = (template_y_22 - template_y_vertical_start) / (template_y_vertical_end - template_y_vertical_start) else: ratio_22_vertical = 0.5 frame_vertical_start = best_vertical_line_points[0][1] frame_vertical_end = best_vertical_line_points[-1][1] proj_22_on_vertical_x = frame_vertical_start[0] + (frame_vertical_end[0] - frame_vertical_start[0]) * ratio_22_vertical proj_22_on_vertical_y = frame_vertical_start[1] + (frame_vertical_end[1] - frame_vertical_start[1]) * ratio_22_vertical proj_22_on_vertical = (proj_22_on_vertical_x, proj_22_on_vertical_y) # Step 3: Project keypoint 22 onto horizontal line (if available) template_horizontal_start = template_coords[best_horizontal_line_points[0][0]] template_horizontal_end = template_coords[best_horizontal_line_points[-1][0]] # Project at x=940 (same x as keypoint 22) template_x_horizontal_start = template_horizontal_start[0] template_x_horizontal_end = template_horizontal_end[0] if abs(template_x_horizontal_end - template_x_horizontal_start) > 1e-6: ratio_22_horizontal = (template_x_22 - template_x_horizontal_start) / (template_x_horizontal_end - template_x_horizontal_start) else: ratio_22_horizontal = 0.5 frame_horizontal_start = best_horizontal_line_points[0][1] frame_horizontal_end = best_horizontal_line_points[-1][1] proj_22_on_horizontal_x = frame_horizontal_start[0] + (frame_horizontal_end[0] - frame_horizontal_start[0]) * ratio_22_horizontal proj_22_on_horizontal_y = frame_horizontal_start[1] + (frame_horizontal_end[1] - frame_horizontal_start[1]) * ratio_22_horizontal proj_22_on_horizontal = (proj_22_on_horizontal_x, proj_22_on_horizontal_y) # Step 4: Calculate keypoint 22 as intersection of two lines # Line 1: Passes through proj_22_on_vertical, parallel to best_horizontal_line # Line 2: Passes through proj_22_on_horizontal, parallel to best_vertical_line # Calculate direction vector of best_horizontal_line horizontal_dir_x = frame_horizontal_end[0] - frame_horizontal_start[0] horizontal_dir_y = frame_horizontal_end[1] - frame_horizontal_start[1] horizontal_dir_length = np.sqrt(horizontal_dir_x**2 + horizontal_dir_y**2) # Calculate direction vector of best_vertical_line vertical_dir_x = frame_vertical_end[0] - frame_vertical_start[0] vertical_dir_y = frame_vertical_end[1] - frame_vertical_start[1] vertical_dir_length = np.sqrt(vertical_dir_x**2 + vertical_dir_y**2) if horizontal_dir_length > 1e-6 and vertical_dir_length > 1e-6: # Normalize direction vectors horizontal_dir_x /= horizontal_dir_length horizontal_dir_y /= horizontal_dir_length vertical_dir_x /= vertical_dir_length vertical_dir_y /= vertical_dir_length # Line 1: passes through proj_22_on_vertical with direction of best_horizontal_line # Parametric: p1 = proj_22_on_vertical + t * horizontal_dir # Line 2: passes through proj_22_on_horizontal with direction of best_vertical_line # Parametric: p2 = proj_22_on_horizontal + s * vertical_dir # Find intersection: proj_22_on_vertical + t * horizontal_dir = proj_22_on_horizontal + s * vertical_dir # This gives us: # proj_22_on_vertical[0] + t * horizontal_dir_x = proj_22_on_horizontal[0] + s * vertical_dir_x # proj_22_on_vertical[1] + t * horizontal_dir_y = proj_22_on_horizontal[1] + s * vertical_dir_y # Rearranging: # t * horizontal_dir_x - s * vertical_dir_x = proj_22_on_horizontal[0] - proj_22_on_vertical[0] # t * horizontal_dir_y - s * vertical_dir_y = proj_22_on_horizontal[1] - proj_22_on_vertical[1] # Solve for t and s using linear algebra A = np.array([ [horizontal_dir_x, -vertical_dir_x], [horizontal_dir_y, -vertical_dir_y] ]) b = np.array([ proj_22_on_horizontal[0] - proj_22_on_vertical[0], proj_22_on_horizontal[1] - proj_22_on_vertical[1] ]) try: t, s = np.linalg.solve(A, b) # Calculate intersection point using line 1 x_22 = proj_22_on_vertical[0] + t * horizontal_dir_x y_22 = proj_22_on_vertical[1] + t * horizontal_dir_y result[21] = (int(round(x_22)), int(round(y_22))) total_right_side_count += 1 except np.linalg.LinAlgError: # Lines are parallel or nearly parallel, use simple intersection # If lines are parallel, use the projection points directly x_22 = proj_22_on_vertical[0] y_22 = proj_22_on_horizontal[1] result[21] = (int(round(x_22)), int(round(y_22))) total_right_side_count += 1 else: # Fallback: use simple intersection x_22 = proj_22_on_vertical[0] y_22 = proj_22_on_horizontal[1] result[21] = (int(round(x_22)), int(round(y_22))) total_right_side_count += 1 print(f"total_right_side_count: {total_right_side_count}, result: {result}") if total_right_side_count > 5: return result # Calculate m_line and b_line from best vertical or horizontal line for use in calculating other points m_line = None b_line = None best_line_for_calc = None best_line_type_for_calc = None if best_vertical_line_name is not None and len(best_vertical_line_points) >= 2: best_line_for_calc = best_vertical_line_points best_line_type_for_calc = 'vertical' points_array = np.array([[kp[0], kp[1]] for _, kp in best_vertical_line_points]) x_coords = points_array[:, 0] y_coords = points_array[:, 1] A = np.vstack([x_coords, np.ones(len(x_coords))]).T m_line, b_line = np.linalg.lstsq(A, y_coords, rcond=None)[0] elif best_horizontal_line_name is not None and len(best_horizontal_line_points) >= 2: best_line_for_calc = best_horizontal_line_points best_line_type_for_calc = 'horizontal' points_array = np.array([[kp[0], kp[1]] for _, kp in best_horizontal_line_points]) x_coords = points_array[:, 0] y_coords = points_array[:, 1] A = np.vstack([x_coords, np.ones(len(x_coords))]).T m_line, b_line = np.linalg.lstsq(A, y_coords, rcond=None)[0] # Calculate missing points to reach exactly 5 points # Ensure 4 points aren't all on one line if total_right_side_count < 5 and (m_line is not None or (best_line_for_calc is not None and best_line_type_for_calc == 'vertical')): # Check current distribution counts_per_line = [ len(line_18_21_points), len(line_23_24_points), len(line_25_30_points) ] # Calculate points on line 18-21 if needed template_ys_18_21 = [140, 270, 410, 540] template_indices_18_21 = [17, 18, 19, 20] if best_vertical_line_name == '25-30': # Construct parallel line 18-21 from line 25-30 for template_y, idx in zip(template_ys_18_21, template_indices_18_21): if result[idx] is None and total_right_side_count < 5: # Check if adding this point would put 4 on one line new_counts = counts_per_line.copy() new_counts[0] += 1 # Adding to line 18-21 if max(new_counts) >= 4 and total_right_side_count == 4: # Would have 4 on one line, skip continue # Calculate y using scale from template ref_ys = [kp[1] for _, kp in line_25_30_points] ref_template_ys = [5, 140, 250, 430, 540, 675] ref_indices = [24, 25, 26, 27, 28, 29] matched_template_ys = [] for ref_idx, ref_kp in line_25_30_points: if ref_idx in ref_indices: template_idx = ref_indices.index(ref_idx) matched_template_ys.append((ref_template_ys[template_idx], ref_kp[1])) if len(matched_template_ys) >= 1: ref_template_y, ref_frame_y = matched_template_ys[0] if ref_template_y > 0: scale = ref_frame_y / ref_template_y y_new = int(round(template_y * scale)) else: y_new = ref_frame_y else: y_new = int(round(np.median(ref_ys))) if ref_ys else template_y # Calculate x using parallel line geometry if abs(m_line) > 1e-6: x_on_line_25_30 = (y_new - b_line) / m_line x_new = int(round(x_on_line_25_30 * 0.850)) else: x_new = int(round(np.median([kp[0] for _, kp in line_25_30_points]) * 0.850)) result[idx] = (x_new, y_new) total_right_side_count += 1 if total_right_side_count >= 5: break elif best_vertical_line_name == '18-21': # Calculate missing points on line 18-21 for template_y, idx in zip(template_ys_18_21, template_indices_18_21): if result[idx] is None and total_right_side_count < 5: # Check if adding this point would put 4 on one line new_counts = counts_per_line.copy() new_counts[0] += 1 # Adding to line 18-21 if max(new_counts) >= 4 and total_right_side_count == 4: # Would have 4 on one line, skip continue # Calculate x on the line if abs(m_line) > 1e-6: x_new = (template_y - b_line) / m_line else: x_new = np.median([kp[0] for _, kp in line_18_21_points]) # Scale y based on available points ref_ys = [kp[1] for _, kp in line_18_21_points] ref_template_ys = [] for ref_idx, _ in line_18_21_points: if ref_idx in template_indices_18_21: template_idx = template_indices_18_21.index(ref_idx) ref_template_ys.append(template_ys_18_21[template_idx]) if len(ref_ys) >= 1 and len(ref_template_ys) >= 1: ref_template_y = ref_template_ys[0] ref_frame_y = ref_ys[0] if ref_template_y > 0: scale = ref_frame_y / ref_template_y y_new = int(round(template_y * scale)) else: y_new = ref_frame_y else: y_new = int(round(np.median(ref_ys))) if ref_ys else template_y result[idx] = (int(round(x_new)), y_new) total_right_side_count += 1 if total_right_side_count >= 5: break # Note: The unified approach above handles all cases (2a and 2b combined) # Legacy code removed - all logic is now in the unified case 2 above return result def check_keypoints_would_cause_invalid_mask( frame_keypoints: list[tuple[int, int]], template_keypoints: list[tuple[int, int]] = None, frame: np.ndarray = None, floor_markings_template: np.ndarray = None, return_warped_data: bool = False, ) -> tuple[bool, str] | tuple[bool, str, tuple]: """ Check if keypoints would cause InvalidMask errors during evaluation. Args: frame_keypoints: Frame keypoints to check template_keypoints: Template keypoints (defaults to TEMPLATE_KEYPOINTS) frame: Optional frame image for full validation floor_markings_template: Optional template image for full validation Returns: Tuple of (would_cause_error, error_message) """ try: from keypoint_evaluation import ( validate_projected_corners, TEMPLATE_KEYPOINTS, INDEX_KEYPOINT_CORNER_BOTTOM_LEFT, INDEX_KEYPOINT_CORNER_BOTTOM_RIGHT, INDEX_KEYPOINT_CORNER_TOP_LEFT, INDEX_KEYPOINT_CORNER_TOP_RIGHT, findHomography, InvalidMask, ) if template_keypoints is None: template_keypoints = TEMPLATE_KEYPOINTS # Filter valid keypoints filtered_template = [] filtered_frame = [] for i, (t_kp, f_kp) in enumerate(zip(template_keypoints, frame_keypoints)): if f_kp[0] > 0 and f_kp[1] > 0: filtered_template.append(t_kp) filtered_frame.append(f_kp) if len(filtered_template) < 4: if return_warped_data: return (True, "Not enough keypoints for homography", None) return (True, "Not enough keypoints for homography") # Compute homography src_pts = np.array(filtered_template, dtype=np.float32) dst_pts = np.array(filtered_frame, dtype=np.float32) result = findHomography(src_pts, dst_pts) if result is None: if return_warped_data: return (True, "Failed to compute homography", None) return (True, "Failed to compute homography") H, _ = result # Check for twisted projection (bowtie) try: validate_projected_corners( source_keypoints=template_keypoints, homography_matrix=H ) except Exception as e: error_msg = "Projection twisted (bowtie)" if "twisted" in str(e).lower() or "Projection twisted" in str(e).lower() else str(e) if return_warped_data: return (True, error_msg, None) return (True, error_msg) # If frame and template are provided, check mask validation if frame is not None and floor_markings_template is not None: try: from keypoint_evaluation import ( project_image_using_keypoints, extract_masks_for_ground_and_lines, InvalidMask, ) # project_image_using_keypoints can raise InvalidMask from validate_projected_corners try: # start_time = time.time() warped_template = project_image_using_keypoints( image=floor_markings_template, source_keypoints=template_keypoints, destination_keypoints=frame_keypoints, destination_width=frame.shape[1], destination_height=frame.shape[0], ) # end_time = time.time() # print(f"project_image_using_keypoints time: {end_time - start_time} seconds") except InvalidMask as e: if return_warped_data: return (True, f"Projection validation failed: {e}", None) return (True, f"Projection validation failed: {e}") except Exception as e: # Other errors (e.g., ValueError from homography failure) if return_warped_data: return (True, f"Projection failed: {e}", None) return (True, f"Projection failed: {e}") # extract_masks_for_ground_and_lines can raise InvalidMask from validation try: mask_ground, mask_lines_expected = extract_masks_for_ground_and_lines( image=warped_template ) except InvalidMask as e: if return_warped_data: return (True, f"Mask extraction validation failed: {e}", None) return (True, f"Mask extraction validation failed: {e}") except Exception as e: if return_warped_data: return (True, f"Mask extraction failed: {e}", None) return (True, f"Mask extraction failed: {e}") # Additional explicit validation (though extract_masks_for_ground_and_lines already validates) from keypoint_evaluation import validate_mask_lines, validate_mask_ground try: validate_mask_lines(mask_lines_expected) except InvalidMask as e: if return_warped_data: return (True, f"Mask lines validation failed: {e}", None) return (True, f"Mask lines validation failed: {e}") except Exception as e: if return_warped_data: return (True, f"Mask lines validation error: {e}", None) return (True, f"Mask lines validation error: {e}") try: validate_mask_ground(mask_ground) except InvalidMask as e: if return_warped_data: return (True, f"Mask ground validation failed: {e}", None) return (True, f"Mask ground validation failed: {e}") except Exception as e: if return_warped_data: return (True, f"Mask ground validation error: {e}", None) return (True, f"Mask ground validation error: {e}") # If return_warped_data is True and validation passed, return the computed data if return_warped_data: return (False, "", (warped_template, mask_ground, mask_lines_expected)) except ImportError: # If keypoint_evaluation is not available, skip validation pass except InvalidMask as e: # Catch any InvalidMask that wasn't caught above if return_warped_data: return (True, f"InvalidMask error: {e}", None) return (True, f"InvalidMask error: {e}") except Exception as e: # If we can't check masks for other reasons, assume it's okay # Don't let exceptions propagate pass # If we get here, keypoints should be valid if return_warped_data: return (False, "", None) # No warped data if frame/template not provided return (False, "") except ImportError: # If keypoint_evaluation is not available, skip validation if return_warped_data: return (False, "", None) return (False, "") except Exception as e: # Any other error - assume it would cause problems if return_warped_data: return (True, f"Validation error: {e}", None) return (True, f"Validation error: {e}") def evaluate_keypoints_with_cached_data( frame: np.ndarray, mask_ground: np.ndarray, mask_lines_expected: np.ndarray, ) -> float: """ Evaluate keypoints using pre-computed warped template and masks. This avoids redundant computation when we already have the warped data from validation. Args: frame: Frame image mask_ground: Pre-computed ground mask from warped template mask_lines_expected: Pre-computed expected lines mask from warped template Returns: Score between 0.0 and 1.0 """ try: from keypoint_evaluation import ( extract_mask_of_ground_lines_in_image, bitwise_and, ) # Only need to extract predicted lines from frame (uses cached mask_ground) mask_lines_predicted = extract_mask_of_ground_lines_in_image( image=frame, ground_mask=mask_ground ) pixels_overlapping = bitwise_and( mask_lines_expected, mask_lines_predicted ).sum() pixels_on_lines = mask_lines_expected.sum() score = pixels_overlapping / (pixels_on_lines + 1e-8) return min(1.0, max(0.0, score)) # Clamp to [0, 1] except Exception as e: print(f'Error in cached keypoint evaluation: {e}') return 0.0 def check_and_evaluate_keypoints( frame_keypoints: list[tuple[int, int]], frame: np.ndarray, ) -> tuple[bool, float, str]: """ Check if keypoints would cause InvalidMask errors and evaluate them in one call. This reuses the warped template and masks computed during validation for evaluation. Args: frame_keypoints: Frame keypoints to check and evaluate frame: Frame image Returns: Tuple of (is_valid, score, error_msg). - If is_valid is True, score is the evaluation score and error_msg is empty string. - If is_valid is False, score is 0.0 and error_msg contains the error message. """ # Check with return_warped_data=True to get cached data # start_time = time.time() check_result = check_keypoints_would_cause_invalid_mask( frame_keypoints, _TEMPLATE_KEYPOINTS, frame, _TEMPLATE_IMAGE, return_warped_data=True ) # end_time = time.time() # print(f"check_keypoints_would_cause_invalid_mask time: {end_time - start_time} seconds") if len(check_result) == 3: would_cause_error, error_msg, warped_data = check_result else: would_cause_error, error_msg = check_result warped_data = None if would_cause_error: return (False, 0.0, error_msg) # If we have cached warped data, use it for fast evaluation if warped_data is not None: _, mask_ground, mask_lines_expected = warped_data try: score = evaluate_keypoints_with_cached_data( frame, mask_ground, mask_lines_expected ) return (True, score, "") except Exception as e: print(f'Error evaluating with cached data: {e}') return (True, 0.0, "") # Fallback to regular evaluation if no cached data try: from keypoint_evaluation import evaluate_keypoints_for_frame score = evaluate_keypoints_for_frame( _TEMPLATE_KEYPOINTS, frame_keypoints, frame, _TEMPLATE_IMAGE ) return (True, score, "") except Exception as e: print(f'Error in regular evaluation: {e}') return (True, 0.0, "") # ============================================================================ # MULTIPROCESSING WORKER FUNCTIONS # ============================================================================ def _evaluate_batch_of_candidates(args): """ Worker function to evaluate a batch of keypoint candidates. Uses threading, so we can share the frame/template without pickling overhead. OpenCV operations are thread-safe for read operations, so no locking needed. """ candidate_batch, frame = args results = [] for test_kps, candidate_metadata in candidate_batch: # Match the exact behavior of sequential evaluation # Only catch exceptions silently like the sequential version does try: if frame is not None and _TEMPLATE_IMAGE is not None: is_valid, score, _ = check_and_evaluate_keypoints( test_kps, frame ) # Only append valid results with positive scores (matching sequential behavior) if is_valid: results.append((is_valid, score, test_kps, candidate_metadata)) except Exception: # Silently skip like the sequential version - don't add invalid results # This matches the original behavior exactly pass return results def evaluate_keypoints_candidates_parallel( candidate_kps_list: List[List[Tuple[int, int]]], candidate_metadata: List[Any], frame: np.ndarray, num_workers: int = None, ) -> Tuple[bool, float, List[Tuple[int, int]], Any]: """ Evaluate multiple keypoint candidates in parallel using threading. Threading is faster than multiprocessing here because: 1. OpenCV releases GIL, so threads can run in parallel 2. No pickling overhead for large arrays (frame, template) 3. Lower overhead than spawning processes """ if len(candidate_kps_list) == 0: return (False, -1.0, None, None) if num_workers is None: # Cap workers to avoid overhead with too many threads # Optimal range is typically 8-32 workers depending on workload # Too many threads cause context switching overhead and contention # Cap at 32 even if CPU count is higher (e.g., cloud servers with 96+ CPUs) max_cpu_workers = min(32, cpu_count()) # Cap at 32 to avoid overhead max_workers = min(max_cpu_workers, len(candidate_kps_list)) num_workers = max(1, max_workers) # For small numbers of candidates, use sequential evaluation # Threading overhead isn't worth it for very small batches # Lowered threshold to ensure we don't miss candidates due to batching issues if len(candidate_kps_list) < 10: best_result = None best_score = -1.0 for test_kps, metadata in zip(candidate_kps_list, candidate_metadata): try: is_valid, score, _ = check_and_evaluate_keypoints( test_kps, frame ) if is_valid and score > best_score: best_score = score best_result = (is_valid, score, test_kps, metadata) except Exception: pass else: # Check if we're on Linux - ThreadPoolExecutor doesn't work well with opencv-python-headless import platform is_linux = platform.system().lower() == 'linux' # Use parallel processing for larger batches if is_linux: # Use ProcessPoolExecutor on Linux (multiprocessing) - works because each process has its own GIL from concurrent.futures import ProcessPoolExecutor, as_completed else: # Use ThreadPoolExecutor on Windows/Other (threading) - OpenCV releases GIL from concurrent.futures import ThreadPoolExecutor, as_completed # Split candidates into batches for each worker # Ensure we process ALL candidates - use ceiling division batch_size = max(1, (len(candidate_kps_list) + num_workers - 1) // num_workers) batches = [] total_candidates_in_batches = 0 for i in range(0, len(candidate_kps_list), batch_size): batch = list(zip( candidate_kps_list[i:i+batch_size], candidate_metadata[i:i+batch_size] )) if len(batch) > 0: # Only add non-empty batches batches.append((batch, frame)) total_candidates_in_batches += len(batch) # Verify we're processing all candidates if total_candidates_in_batches != len(candidate_kps_list): print(f"Warning: Batch mismatch! Expected {len(candidate_kps_list)} candidates, got {total_candidates_in_batches}") best_result = None best_score = -1.0 try: if is_linux: executor_class = ProcessPoolExecutor else: executor_class = ThreadPoolExecutor with executor_class(max_workers=num_workers) as executor: futures = [executor.submit(_evaluate_batch_of_candidates, args) for args in batches] all_results = [] for future in as_completed(futures): try: batch_results = future.result() if batch_results: # Only extend if we have results all_results.extend(batch_results) except Exception as e: # Log but continue processing other batches print(f"Error processing batch result: {e}") import traceback traceback.print_exc() pass # Debug: Check if we got results from all batches if len(all_results) == 0: print(f"Warning: No valid results from parallel evaluation of {len(candidate_kps_list)} candidates") # Process all results and find the best one # This ensures we compare ALL candidates, not just within batches # Match the exact logic from sequential evaluation for result in all_results: if result is not None: is_valid, score, test_kps, metadata = result # Ensure score is numeric for comparison try: score = float(score) if score is not None else 0.0 except (ValueError, TypeError): score = 0.0 # Match sequential evaluation: only update if valid and score is better if is_valid and score > best_score: best_score = score best_result = (is_valid, score, test_kps, metadata) except Exception as e: print(f"Threading evaluation failed: {e}, falling back to sequential") for test_kps, metadata in zip(candidate_kps_list, candidate_metadata): try: is_valid, score, _ = check_and_evaluate_keypoints( test_kps, frame ) if is_valid and score > best_score: best_score = score best_result = (is_valid, score, test_kps, metadata) except Exception: pass if best_result is not None: return best_result return (False, -1.0, None, None) def _process_single_frame_for_prediction(args): """ Worker function to process a single frame for failed index prediction. Returns: (frame_index, score, adjusted_success) - score: evaluation score of the calculated keypoints (0.0 if failed or invalid) - adjusted_success: True if keypoints were successfully adjusted, False otherwise """ frame_index, frame_result, frame_width, frame_height, frame_image, offset = args try: from keypoint_helper_v2_optimized import ( remove_duplicate_detections, calculate_missing_keypoints, adjust_keypoints_to_avoid_invalid_mask, ) frame_keypoints = getattr(frame_result, "keypoints", []) or [] original_count = sum(1 for (x, y) in frame_keypoints if int(x) != 0 and int(y) != 0) cleaned_keypoints = remove_duplicate_detections( frame_keypoints, frame_width, frame_height ) valid_keypoint_indices = [idx for idx, kp in enumerate(cleaned_keypoints) if kp[0] != 0 and kp[1] != 0] if len(valid_keypoint_indices) > 5: calculated_keypoints = cleaned_keypoints else: left_side_indices_range = range(0, 13) right_side_indices_range = range(17, 30) side_check_set = set() if len(valid_keypoint_indices) >= 4: for idx in valid_keypoint_indices: if idx in left_side_indices_range: side_check_set.add("left") elif idx in right_side_indices_range: side_check_set.add("right") else: side_check_set.add("center") if len(side_check_set) > 1: calculated_keypoints = cleaned_keypoints else: calculated_keypoints = calculate_missing_keypoints( cleaned_keypoints, frame_width, frame_height ) original_frame_number = offset + frame_index print(f"Frame {original_frame_number} (index {frame_index}): original_count: {original_count}, cleaned_keypoints: {len([kp for kp in cleaned_keypoints if kp[0] != 0 and kp[1] != 0])}, calculated_keypoints: {len([kp for kp in calculated_keypoints if kp[0] != 0 and kp[1] != 0])}") start_time = time.time() adjusted_success, calculated_keypoints, score = adjust_keypoints_to_avoid_invalid_mask( calculated_keypoints, frame_image ) end_time = time.time() print(f"adjust_keypoints_to_avoid_invalid_mask time: {end_time - start_time} seconds") if not adjusted_success: return (frame_index, 0.0, False) # Failed, score is 0.0 print(f"after adjustment, calculated_keypoints: {calculated_keypoints}, score: {score:.4f}") setattr(frame_result, "keypoints", list(calculated_keypoints)) return (frame_index, score, True) # Success with score except Exception as e: print(f"Error processing frame {frame_index}: {e}") return (frame_index, 0.0, False) # Failed on error, score is 0.0 def _generate_sparse_keypoints_for_frame(args): """ Worker function to generate sparse keypoints for a single frame. Returns: (frame_index, sparse_keypoints) """ frame_index, frame_width, frame_height, frame_image = args try: from keypoint_helper_v2_optimized import ( _generate_sparse_template_keypoints, ) sparse_keypoints = _generate_sparse_template_keypoints( frame_width, frame_height, frame_image=frame_image, ) return (frame_index, sparse_keypoints) except Exception as e: print(f"Error generating sparse keypoints for frame {frame_index}: {e}") # Return empty keypoints on error return (frame_index, [(0, 0)] * 32) def _evaluate_keypoints_for_frame(args): """ Worker function to evaluate both sparse and calculated keypoints for a single frame. Returns: (frame_index, sparse_score, calculated_score, sparse_keypoints, calculated_keypoints) """ frame_index, sparse_keypoints, calculated_keypoints, frame_image, pre_calculated_score = args sparse_score = 0.0 calculated_score = 0.0 # Use pre-calculated score if available (from _process_single_frame_for_prediction) if pre_calculated_score is not None and pre_calculated_score > 0.0: calculated_score = pre_calculated_score print(f"Frame {frame_index}: Using pre-calculated score: {calculated_score:.4f}") else: # Need to evaluate calculated keypoints calculated_score = 0.0 try: from keypoint_evaluation import evaluate_keypoints_for_frame # Evaluate sparse keypoints if frame_image is not None and _TEMPLATE_IMAGE is not None and _TEMPLATE_KEYPOINTS is not None: try: sparse_score = evaluate_keypoints_for_frame( template_keypoints=_TEMPLATE_KEYPOINTS, frame_keypoints=sparse_keypoints, frame=frame_image, floor_markings_template=_TEMPLATE_IMAGE, ) except Exception: sparse_score = 0.0 # Evaluate calculated keypoints only if not pre-calculated if pre_calculated_score is None or pre_calculated_score <= 0.0: calculated_keypoints_valid = len([kp for kp in calculated_keypoints if kp[0] != 0 or kp[1] != 0]) >= 4 if calculated_keypoints_valid: try: calculated_score = evaluate_keypoints_for_frame( template_keypoints=_TEMPLATE_KEYPOINTS, frame_keypoints=calculated_keypoints, frame=frame_image, floor_markings_template=_TEMPLATE_IMAGE, ) except Exception: calculated_score = 0.0 else: calculated_score = -1.0 except Exception as e: print(f"Error evaluating keypoints for frame {frame_index}: {e}") return (frame_index, sparse_score, calculated_score, sparse_keypoints, calculated_keypoints) def _calculate_keypoints_score( keypoints: list[tuple[int, int]], frame: np.ndarray, ) -> float: """ Helper function to calculate score for keypoints. Returns 0.0 if evaluation fails or keypoints are invalid. """ score = 0.0 try: from keypoint_evaluation import evaluate_keypoints_for_frame # Check if keypoints are valid (at least 4 non-zero keypoints) keypoints_valid = len([kp for kp in keypoints if kp[0] != 0 or kp[1] != 0]) >= 4 if keypoints_valid and frame is not None and _TEMPLATE_IMAGE is not None and _TEMPLATE_KEYPOINTS is not None: try: score = evaluate_keypoints_for_frame( template_keypoints=_TEMPLATE_KEYPOINTS, frame_keypoints=keypoints, frame=frame, floor_markings_template=_TEMPLATE_IMAGE, ) except Exception: score = 0.0 except Exception: score = 0.0 return score def adjust_keypoints_to_avoid_invalid_mask( frame_keypoints: list[tuple[int, int]], frame: np.ndarray = None, max_iterations: int = 5, num_workers: int = None, ) -> tuple[bool, list[tuple[int, int]], float]: """ Adjust keypoints to avoid InvalidMask errors. This function tries to fix common issues: 1. Twisted projection (bowtie) - adjusts corner keypoints 2. Ground covers too much - shrinks projected area by moving corners inward 3. Other mask validation issues - adjusts keypoints to improve projection Args: frame_keypoints: Frame keypoints to adjust frame: Optional frame image for validation max_iterations: Maximum number of adjustment iterations num_workers: Number of workers for parallel evaluation Returns: Tuple of (success, adjusted_keypoints, score): - success: True if keypoints were successfully adjusted, False otherwise - adjusted_keypoints: Adjusted keypoints - score: Evaluation score of the adjusted keypoints (0.0 if failed or invalid) """ adjusted = list(frame_keypoints) # Check if adjustment is needed and evaluate score in one call # This reuses warped data from validation for efficient evaluation error_msg = "" would_cause_error = False is_valid, score, error_msg = check_and_evaluate_keypoints( adjusted, frame ) if is_valid: return (True, adjusted, score) # error_msg is already available from check_and_evaluate_keypoints would_cause_error = True # Keypoints are invalid print(f"Would cause error: {would_cause_error}, error_msg: {error_msg}") # Try to fix twisted projection (most common issue) if "twisted" in error_msg.lower() or "bowtie" in error_msg.lower() or "Projection twisted" in error_msg.lower(): # Use the existing _adjust_keypoints_to_pass_validation function adjusted = _adjust_keypoints_to_pass_validation( adjusted, frame.shape[1] if frame is not None else None, frame.shape[0] if frame is not None else None ) # Check again after adjustment and evaluate score if frame is not None and _TEMPLATE_IMAGE is not None and _TEMPLATE_KEYPOINTS is not None: is_valid, score, error_msg = check_and_evaluate_keypoints( adjusted, frame ) if is_valid: return (True, adjusted, score) # error_msg is already available from check_and_evaluate_keypoints else: would_cause_error, error_msg = check_keypoints_would_cause_invalid_mask( adjusted, _TEMPLATE_KEYPOINTS, frame, _TEMPLATE_IMAGE ) if not would_cause_error: score = 0.0 return (True, adjusted, score) # Handle "a projected line is too wide" error # This happens when projected lines are too thick/wide (aspect ratio too high) if "too wide" in error_msg.lower() or "wide line" in error_msg.lower(): print(f"Adjusting keypoints to fix 'a projected line is too wide' error") try: # This error usually means the projection is creating lines that are too thick # Strategy: Adjust keypoints to reduce projection distortion valid_keypoints = [] for idx in range(len(adjusted)): x, y = adjusted[idx] if x == 0 and y == 0: continue valid_keypoints.append((idx, x, y)) if len(valid_keypoints) >= 4: # Calculate center and spread of keypoints center_x = sum(x for _, x, y in valid_keypoints) / len(valid_keypoints) center_y = sum(y for _, x, y in valid_keypoints) / len(valid_keypoints) # Calculate distances from center distances = [] for idx, x, y in valid_keypoints: dist = np.sqrt((x - center_x)**2 + (y - center_y)**2) distances.append((idx, x, y, dist)) # Sort by distance distances.sort(key=lambda d: d[3], reverse=True) # Strategy 1: Try moving keypoints slightly outward to reduce compression # This can help if keypoints are too close together causing wide lines best_wide_kps = None best_wide_score = -1.0 # Collect all candidate keypoints first, then evaluate in parallel candidate_kps_list = [] candidate_metadata = [] # Strategy 1: Try expanding keypoints slightly (opposite of shrinking) # Reduced from 4 to 2 candidates for faster computation for expand_factor in [1.05, 1.10]: test_kps = list(adjusted) for idx, x, y, dist in distances: new_x = int(round(center_x + (x - center_x) * expand_factor)) new_y = int(round(center_y + (y - center_y) * expand_factor)) test_kps[idx] = (new_x, new_y) # Add directly - validation and evaluation will be done in parallel candidate_kps_list.append(test_kps) candidate_metadata.append(('expand', expand_factor)) # Strategy 2: Try adjusting individual keypoints (only top 2 farthest, reduced adjustments) # Reduced from 6x6=36 per keypoint to 3x3=9, and only test top 2 keypoints for idx, x, y, dist in distances[:2]: for adjust_x in [-2, 0, 2]: for adjust_y in [-2, 0, 2]: if adjust_x == 0 and adjust_y == 0: continue # Skip no-op test_kps = list(adjusted) test_kps[idx] = (x + adjust_x, y + adjust_y) # Add directly - validation and evaluation will be done in parallel candidate_kps_list.append(test_kps) candidate_metadata.append(('perturb', idx, adjust_x, adjust_y)) # Strategy 3: Try slight shrinking (opposite approach - reduce projection area) # Reduced from 3 to 2 candidates for shrink_factor in [0.96, 0.94]: test_kps = list(adjusted) for idx, x, y, dist in distances: new_x = int(round(center_x + (x - center_x) * shrink_factor)) new_y = int(round(center_y + (y - center_y) * shrink_factor)) test_kps[idx] = (new_x, new_y) # Add directly - validation and evaluation will be done in parallel candidate_kps_list.append(test_kps) candidate_metadata.append(('shrink', shrink_factor)) # Evaluate all candidates in parallel if len(candidate_kps_list) > 0: print(f"Evaluating {len(candidate_kps_list)} wide-line candidates in parallel...") eval_start = time.time() is_valid, score, best_kps, best_meta = evaluate_keypoints_candidates_parallel( candidate_kps_list, candidate_metadata, frame, num_workers ) eval_time = time.time() - eval_start print(f"Parallel evaluation took {eval_time:.2f} seconds for {len(candidate_kps_list)} candidates") if is_valid and score > best_wide_score: best_wide_score = score best_wide_kps = best_kps print(f"Found best wide-line adjustment: {best_meta}, score: {score:.4f}") if best_wide_kps is not None: # Score is already calculated in evaluate_keypoints_candidates_parallel return (True, best_wide_kps, best_wide_score) except Exception as e: print(f"Error in wide line adjustment: {e}") pass # Handle "projected ground should be a single object" error # This happens when the ground mask has multiple disconnected regions if "should be a single" in error_msg.lower() or "single object" in error_msg.lower() or "distinct regions" in error_msg.lower(): print(f"Adjusting keypoints to fix 'projected ground should be a single object' error (optimized)") try: valid_keypoints = [] for idx in range(len(adjusted)): x, y = adjusted[idx] if x == 0 and y == 0: continue valid_keypoints.append((idx, x, y)) if len(valid_keypoints) >= 4: center_x = sum(x for _, x, y in valid_keypoints) / len(valid_keypoints) center_y = sum(y for _, x, y in valid_keypoints) / len(valid_keypoints) candidate_kps_list = [] candidate_metadata = [] # Strategy 1: Move keypoints closer to center # Reduced from 5 to 3 candidates for shrink_factor in [0.96, 0.92, 0.90]: test_kps = list(adjusted) for idx, x, y in valid_keypoints: new_x = int(round(center_x + (x - center_x) * shrink_factor)) new_y = int(round(center_y + (y - center_y) * shrink_factor)) test_kps[idx] = (new_x, new_y) # Add directly - validation and evaluation will be done in parallel candidate_kps_list.append(test_kps) candidate_metadata.append(('shrink', shrink_factor)) # Strategy 2: Adjust boundary keypoints distances = [] for idx, x, y in valid_keypoints: dist = np.sqrt((x - center_x)**2 + (y - center_y)**2) distances.append((idx, x, y, dist)) distances.sort(key=lambda d: d[3], reverse=True) # Reduced from 3 to 2 candidates for shrink_factor in [0.90, 0.85]: test_kps = list(adjusted) boundary_count = max(1, len(distances) // 4) for idx, x, y, dist in distances[:boundary_count]: new_x = int(round(center_x + (x - center_x) * shrink_factor)) new_y = int(round(center_y + (y - center_y) * shrink_factor)) test_kps[idx] = (new_x, new_y) # Add directly - validation and evaluation will be done in parallel candidate_kps_list.append(test_kps) candidate_metadata.append(('boundary', shrink_factor)) # Evaluate all candidates in parallel if len(candidate_kps_list) > 0: print(f"Evaluating {len(candidate_kps_list)} single-object candidates in parallel...") eval_start = time.time() is_valid, score, best_kps, best_meta = evaluate_keypoints_candidates_parallel( candidate_kps_list, candidate_metadata, frame, num_workers ) eval_time = time.time() - eval_start print(f"Parallel evaluation took {eval_time:.2f} seconds for {len(candidate_kps_list)} candidates") if is_valid: print(f"Found best single-object adjustment: {best_meta}, score: {score:.4f}") # Score is already calculated in evaluate_keypoints_candidates_parallel return (True, best_kps, score) except Exception as e: print(f"Error in optimized single object adjustment: {e}") pass # Handle "ground covers too much" error by shrinking the projected area if "ground covers" in error_msg.lower() or "covers more than" in error_msg.lower(): print(f"Adjusting keypoints to avoid 'ground covers too much' error") try: from keypoint_evaluation import ( INDEX_KEYPOINT_CORNER_BOTTOM_LEFT, INDEX_KEYPOINT_CORNER_BOTTOM_RIGHT, INDEX_KEYPOINT_CORNER_TOP_LEFT, INDEX_KEYPOINT_CORNER_TOP_RIGHT, ) # First, try adjusting corners if available corner_indices = [ INDEX_KEYPOINT_CORNER_TOP_LEFT, INDEX_KEYPOINT_CORNER_TOP_RIGHT, INDEX_KEYPOINT_CORNER_BOTTOM_RIGHT, INDEX_KEYPOINT_CORNER_BOTTOM_LEFT, ] # Get corner keypoints corners = [] center_x, center_y = 0, 0 valid_corners = 0 for corner_idx in corner_indices: if corner_idx < len(adjusted): x, y = adjusted[corner_idx] if x == 0 and y == 0: continue corners.append((corner_idx, x, y)) center_x += x center_y += y valid_corners += 1 if valid_corners >= 4: center_x /= valid_corners center_y /= valid_corners candidate_kps_list = [] candidate_metadata = [] # Move corners inward # Reduced from 7 to 4 candidates for faster computation for shrink_factor in [0.90, 0.85, 0.75, 0.65]: test_kps = list(adjusted) for corner_idx, x, y in corners: new_x = int(round(center_x + (x - center_x) * shrink_factor)) new_y = int(round(center_y + (y - center_y) * shrink_factor)) test_kps[corner_idx] = (new_x, new_y) # Add directly - validation and evaluation will be done in parallel candidate_kps_list.append(test_kps) candidate_metadata.append(('corner', shrink_factor)) # Evaluate all candidates in parallel if len(candidate_kps_list) > 0: print(f"Evaluating {len(candidate_kps_list)} corner adjustment candidates in parallel...") eval_start = time.time() is_valid, score, best_kps, best_meta = evaluate_keypoints_candidates_parallel( candidate_kps_list, candidate_metadata, frame, num_workers ) eval_time = time.time() - eval_start print(f"Parallel evaluation took {eval_time:.2f} seconds for {len(candidate_kps_list)} candidates") if is_valid: print(f"Found best corner adjustment: {best_meta}, score: {score:.4f}") # Score is already calculated in evaluate_keypoints_candidates_parallel return (True, best_kps, score) # If corners adjustment didn't work or we don't have enough corners, # try adjusting individual keypoints one at a time # This handles cases where non-corner keypoints (like 15, 16, 17, 31, 32) are causing the issue valid_keypoints = [] all_center_x, all_center_y = 0, 0 valid_count = 0 for idx in range(len(adjusted)): x, y = adjusted[idx] if x == 0 and y == 0: continue valid_keypoints.append((idx, x, y)) all_center_x += x all_center_y += y valid_count += 1 if valid_count >= 4: all_center_x /= valid_count all_center_y /= valid_count # Calculate distances from center for each keypoint # Try adjusting keypoints farthest from center first (most likely to cause coverage issues) distances = [] for idx, x, y in valid_keypoints: dist = np.sqrt((x - all_center_x)**2 + (y - all_center_y)**2) distances.append((idx, x, y, dist)) # Sort by distance (farthest first) - these are most likely causing the coverage issue distances.sort(key=lambda d: d[3], reverse=True) # Collect all candidate keypoints for parallel evaluation candidate_kps_list = [] candidate_metadata = [] # Try adjusting each keypoint individually # Reduced: only test top 3 farthest keypoints, and reduce shrink factors from 9 to 4 for idx, x, y, dist in distances[:3]: for shrink_factor in [0.95, 0.90, 0.80, 0.70]: test_kps = list(adjusted) new_x = int(round(all_center_x + (x - all_center_x) * shrink_factor)) new_y = int(round(all_center_y + (y - all_center_y) * shrink_factor)) test_kps[idx] = (new_x, new_y) # Add directly - validation and evaluation will be done in parallel candidate_kps_list.append(test_kps) candidate_metadata.append(('individual', idx, shrink_factor)) # Try adjusting pairs # Reduced from 6 to 3 candidates if valid_count >= 6: for shrink_factor in [0.90, 0.80, 0.70]: test_kps = list(adjusted) for idx, x, y, dist in distances[:2]: new_x = int(round(all_center_x + (x - all_center_x) * shrink_factor)) new_y = int(round(all_center_y + (y - all_center_y) * shrink_factor)) test_kps[idx] = (new_x, new_y) # Add directly - validation and evaluation will be done in parallel candidate_kps_list.append(test_kps) candidate_metadata.append(('pair', shrink_factor)) # Evaluate all candidates in parallel if len(candidate_kps_list) > 0: print(f"Evaluating {len(candidate_kps_list)} ground-coverage candidates in parallel...") eval_start = time.time() is_valid, score, best_kps, best_meta = evaluate_keypoints_candidates_parallel( candidate_kps_list, candidate_metadata, frame, num_workers ) eval_time = time.time() - eval_start print(f"Parallel evaluation took {eval_time:.2f} seconds for {len(candidate_kps_list)} candidates") if is_valid: print(f"Found best ground-coverage adjustment: {best_meta}, score: {score:.4f}") # Score is already calculated in evaluate_keypoints_candidates_parallel return (True, best_kps, score) except Exception as e: print(f"Error in ground coverage adjustment: {e}") pass # If still causing errors, try small perturbations to corner keypoints # This helps with mask validation issues if would_cause_error and max_iterations > 0: try: from keypoint_evaluation import ( INDEX_KEYPOINT_CORNER_BOTTOM_LEFT, INDEX_KEYPOINT_CORNER_BOTTOM_RIGHT, INDEX_KEYPOINT_CORNER_TOP_LEFT, INDEX_KEYPOINT_CORNER_TOP_RIGHT, ) corner_indices = [ INDEX_KEYPOINT_CORNER_TOP_LEFT, INDEX_KEYPOINT_CORNER_TOP_RIGHT, INDEX_KEYPOINT_CORNER_BOTTOM_RIGHT, INDEX_KEYPOINT_CORNER_BOTTOM_LEFT, ] # Collect all corner perturbation candidates candidate_kps_list = [] candidate_metadata = [] # Reduced corner perturbations: from 6x6=36 per corner to 3x3=9 per corner # Also skip (0,0) to avoid no-op for corner_idx in corner_indices: if corner_idx < len(adjusted): x, y = adjusted[corner_idx] if x == 0 and y == 0: continue for dx in [-3, 0, 3]: for dy in [-3, 0, 3]: if dx == 0 and dy == 0: continue # Skip no-op test_kps = list(adjusted) test_kps[corner_idx] = (x + dx, y + dy) # Add directly - validation and evaluation will be done in parallel candidate_kps_list.append(test_kps) candidate_metadata.append(('corner_perturb', corner_idx, dx, dy)) # Evaluate all candidates in parallel if len(candidate_kps_list) > 0: print(f"Evaluating {len(candidate_kps_list)} corner perturbation candidates in parallel...") eval_start = time.time() is_valid, score, best_kps, best_meta = evaluate_keypoints_candidates_parallel( candidate_kps_list, candidate_metadata, frame, num_workers ) eval_time = time.time() - eval_start print(f"Parallel evaluation took {eval_time:.2f} seconds for {len(candidate_kps_list)} candidates") if is_valid: print(f"Found best corner perturbation: {best_meta}, score: {score:.4f}") # Score is already calculated in evaluate_keypoints_candidates_parallel return (True, best_kps, score) except Exception: pass # If we can't fix it, return adjusted (best effort) with score 0.0 score = _calculate_keypoints_score(adjusted, frame) return (False, adjusted, score) def _validate_keypoints_corners( frame_keypoints: list[tuple[int, int]], template_keypoints: list[tuple[int, int]] = None, ) -> bool: """ Validate that frame keypoints can form a valid homography with template keypoints (corners don't create twisted projection). Returns True if validation passes, False otherwise. """ try: from keypoint_evaluation import ( validate_projected_corners, TEMPLATE_KEYPOINTS, INDEX_KEYPOINT_CORNER_BOTTOM_LEFT, INDEX_KEYPOINT_CORNER_BOTTOM_RIGHT, INDEX_KEYPOINT_CORNER_TOP_LEFT, INDEX_KEYPOINT_CORNER_TOP_RIGHT, ) # Use provided template_keypoints or default TEMPLATE_KEYPOINTS if template_keypoints is None: template_keypoints = TEMPLATE_KEYPOINTS # Filter valid keypoints (non-zero) filtered_template = [] filtered_frame = [] for i, (t_kp, f_kp) in enumerate(zip(template_keypoints, frame_keypoints)): if f_kp[0] > 0 and f_kp[1] > 0: # Frame keypoint is valid filtered_template.append(t_kp) filtered_frame.append(f_kp) if len(filtered_template) < 4: return False # Not enough keypoints for homography # Compute homography from template to frame src_pts = np.array(filtered_template, dtype=np.float32) dst_pts = np.array(filtered_frame, dtype=np.float32) H, mask = cv2.findHomography(src_pts, dst_pts) if H is None: return False # Homography computation failed # Validate corners using the homography try: validate_projected_corners( source_keypoints=template_keypoints, homography_matrix=H ) return True # Validation passed except Exception: return False # Validation failed (twisted projection) except ImportError: # If keypoint_evaluation is not available, skip validation return True except Exception: # Any other error - assume invalid return False def calculate_and_adjust_keypoints( results_frames: Sequence[Any], frame_width: int = None, frame_height: int = None, frames: List[np.ndarray] = None, offset: int = 0, num_workers: int = None, ) -> list[tuple[int, float, bool]]: """ Calculate missing keypoints, adjust them to avoid invalid masks, and evaluate scores. Processes frames in parallel using threading. For each frame: 1. Calculates missing keypoints if needed 2. Adjusts keypoints to avoid InvalidMask errors 3. Evaluates the adjusted keypoints and calculates a score Args: results_frames: Sequence of frame results with keypoints frame_width: Frame width frame_height: Frame height frames: Optional list of frame images for validation offset: Frame offset for tracking num_workers: Number of worker threads (defaults to cpu_count()) Returns: List of tuples (frame_index, score, adjusted_success) for all frames: - frame_index: Index of the frame - score: Evaluation score of the adjusted keypoints (0.0 if failed) - adjusted_success: True if keypoints were successfully adjusted, False otherwise """ max_frames = len(results_frames) if max_frames == 0: return [] if num_workers is None: # Cap workers to avoid overhead with too many threads # Optimal range is typically 8-32 workers depending on workload # Too many threads cause context switching overhead and contention # Cap at 32 even if CPU count is higher (e.g., cloud servers with 96+ CPUs) max_cpu_workers = min(32, cpu_count()) # Cap at 32 to avoid overhead max_workers = min(max_cpu_workers, max_frames) num_workers = max(1, max_workers) # Prepare arguments for each frame # Note: With spawn method, each worker will pickle/unpickle the data anyway # So we pass references - copying here would be redundant args_list = [] for frame_index, frame_result in enumerate(results_frames): frame_image = None if frames is not None and frame_index < len(frames): frame_image = frames[frame_index] # Pass reference, will be copied by pickle args_list.append(( frame_index, frame_result, frame_width, frame_height, frame_image, offset )) results = [] # Check if we're on Linux - ThreadPoolExecutor doesn't work well with opencv-python-headless # OpenCV headless doesn't release GIL properly on Linux, so use ProcessPoolExecutor instead import platform is_linux = platform.system().lower() == 'linux' # Use parallel processing for larger batches if max_frames >= 4 and num_workers > 1: try: if is_linux: # Use ProcessPoolExecutor on Linux (multiprocessing) - works because each process has its own GIL from concurrent.futures import ProcessPoolExecutor, as_completed print(f"Linux detected: Processing {max_frames} frames in parallel using {num_workers} processes (ProcessPoolExecutor)...") with ProcessPoolExecutor(max_workers=num_workers) as executor: futures = {executor.submit(_process_single_frame_for_prediction, args): args for args in args_list} for future in as_completed(futures): try: frame_index, score, adjusted_success = future.result() results.append((frame_index, score, adjusted_success)) except Exception as e: print(f"Error getting result from worker: {e}") # If we can't get the result, mark as failed with score 0.0 args = futures[future] frame_index = args[0] results.append((frame_index, 0.0, False)) else: # Use ThreadPoolExecutor on Windows/Other (threading) - OpenCV releases GIL from concurrent.futures import ThreadPoolExecutor, as_completed print(f"Processing {max_frames} frames in parallel using {num_workers} workers (ThreadPoolExecutor)...") with ThreadPoolExecutor(max_workers=num_workers) as executor: futures = {executor.submit(_process_single_frame_for_prediction, args): args for args in args_list} for future in as_completed(futures): try: frame_index, score, adjusted_success = future.result() results.append((frame_index, score, adjusted_success)) except Exception as e: print(f"Error getting result from worker: {e}") # If we can't get the result, mark as failed with score 0.0 args = futures[future] frame_index = args[0] results.append((frame_index, 0.0, False)) except Exception as e: print(f"Parallel processing failed: {e}, falling back to sequential") # Fallback to sequential for args in args_list: try: frame_index, score, adjusted_success = _process_single_frame_for_prediction(args) results.append((frame_index, score, adjusted_success)) except Exception as e: print(f"Error processing frame: {e}") # Mark as failed if exception occurs frame_index = args[0] results.append((frame_index, 0.0, False)) else: # Sequential processing for small batches for args in args_list: try: frame_index, score, adjusted_success = _process_single_frame_for_prediction(args) results.append((frame_index, score, adjusted_success)) except Exception as e: print(f"Error processing frame: {e}") # Mark as failed if exception occurs frame_index = args[0] results.append((frame_index, 0.0, False)) # Sort results by frame_index to ensure consistent ordering results.sort(key=lambda x: x[0]) return results def _generate_sparse_template_keypoints( frame_width: int, frame_height: int, frame_image: np.ndarray = None, ) -> list[tuple[int, int]]: # Use cached template dimensions for performance template_max_x = _TEMPLATE_MAX_X template_max_y = _TEMPLATE_MAX_Y # Calculate scaling factors for both dimensions sx = float(frame_width) / float(template_max_x if template_max_x != 0 else 1) sy = float(frame_height) / float(template_max_y if template_max_y != 0 else 1) # Always use uniform scaling to preserve pitch geometry and aspect ratio # This prevents distortion that creates square contours (like 3x3, 4x4) which fail the wide line check # Uniform scaling ensures the pitch maintains its shape and avoids twisted projections uniform_scale = min(sx, sy) # Scale down significantly to create a much smaller pitch in the warped template # Use a small fraction of the uniform scale to make the pitch as small as possible # This creates a small pitch centered in the frame, avoiding edge artifacts scale_factor = 0.15 # Use 15% of the frame-filling scale to make pitch much smaller uniform_scale = uniform_scale * scale_factor # Ensure minimum scale to avoid keypoints being too close together # Very small scales cause warping artifacts that create square contours (1x1, 2x2 pixels) # These single-pixel artifacts trigger the "too wide" error # Use a fixed minimum scale based on template dimensions to ensure keypoints are spaced properly # This prevents warping artifacts regardless of frame size # Template is 1045x675, need sufficient scale to avoid 1x1 pixel artifacts from warping # Higher minimum scale ensures warped template doesn't create tiny square artifacts min_scale_absolute = 0.3 # Fixed minimum 30% of template size to avoid 1x1 pixel warping artifacts # Higher scale is necessary to prevent warping interpolation from creating single-pixel squares uniform_scale = max(uniform_scale, min_scale_absolute) # Use only corner keypoints for sparse template # Get corner indices from keypoint_evaluation try: from keypoint_evaluation import ( INDEX_KEYPOINT_CORNER_TOP_LEFT, INDEX_KEYPOINT_CORNER_TOP_RIGHT, INDEX_KEYPOINT_CORNER_BOTTOM_RIGHT, INDEX_KEYPOINT_CORNER_BOTTOM_LEFT, ) selected_keypoint_indices = set([ INDEX_KEYPOINT_CORNER_TOP_LEFT, INDEX_KEYPOINT_CORNER_TOP_RIGHT, INDEX_KEYPOINT_CORNER_BOTTOM_RIGHT, INDEX_KEYPOINT_CORNER_BOTTOM_LEFT, ]) except ImportError: # Fallback to default corner indices if import fails # Based on typical template: top-left=0, top-right=24, bottom-right=29, bottom-left=5 selected_keypoint_indices = set([0, 24, 29, 5]) # Default corner indices line_distribution = None # Will store: (total_count, best_region_center, max_density) # If we have line distribution analysis, select appropriate keypoints if frame_image is not None and _TEMPLATE_IMAGE is not None and _TEMPLATE_KEYPOINTS is not None: try: from keypoint_evaluation import ( project_image_using_keypoints, extract_masks_for_ground_and_lines_no_validation, extract_mask_of_ground_lines_in_image ) # Generate initial keypoints for analysis using EXACT FITTING (full frame coverage) # This ensures we get correct line distribution analysis # Use non-uniform scaling to fit exactly to frame dimensions # Optimized with NumPy for better performance initial_sx = float(frame_width) / float(template_max_x if template_max_x != 0 else 1) initial_sy = float(frame_height) / float(template_max_y if template_max_y != 0 else 1) num_template_kps = len(_TEMPLATE_KEYPOINTS) if _TEMPLATE_KEYPOINTS is not None else 32 num_kps = max(32, num_template_kps) # Ensure we have at least 32 keypoints # Use NumPy for vectorized scaling if _TEMPLATE_KEYPOINTS is not None and len(_TEMPLATE_KEYPOINTS) >= num_kps: template_array = np.array(_TEMPLATE_KEYPOINTS[:num_kps], dtype=np.float32) else: # Pad with zeros if needed template_array = np.zeros((num_kps, 2), dtype=np.float32) if _TEMPLATE_KEYPOINTS is not None: template_array[:len(_TEMPLATE_KEYPOINTS)] = _TEMPLATE_KEYPOINTS # Vectorized scaling and clamping scaled_array = template_array.copy() scaled_array[:, 0] = np.clip(np.round(template_array[:, 0] * initial_sx), 0, frame_width - 1) scaled_array[:, 1] = np.clip(np.round(template_array[:, 1] * initial_sy), 0, frame_height - 1) # Set zero keypoints to (0, 0) mask = (template_array[:, 0] <= 0) | (template_array[:, 1] <= 0) scaled_array[mask] = 0 # Convert to list of tuples initial_scaled = [(int(x), int(y)) for x, y in scaled_array] # With exact fitting, keypoints already fill the frame, no centering needed initial_centered = initial_scaled if len(initial_scaled) > 0: try: warped_template = project_image_using_keypoints( image=_TEMPLATE_IMAGE, source_keypoints=_TEMPLATE_KEYPOINTS, destination_keypoints=initial_centered, destination_width=frame_width, destination_height=frame_height, ) # Use non-validating version for line distribution analysis # Exact fitting might create invalid masks, but we still want to analyze line distribution mask_ground, mask_lines = extract_masks_for_ground_and_lines_no_validation(image=warped_template) mask_lines_predicted = extract_mask_of_ground_lines_in_image( image=frame_image, ground_mask=mask_ground ) h, w = mask_lines_predicted.shape # Density-based region analysis: Find arbitrary region with highest line density # Optimized with NumPy convolution for much faster computation region_size_ratio = 0.35 # Region will be 35% of frame size region_w = max(50, int(w * region_size_ratio)) region_h = max(50, int(h * region_size_ratio)) # Use larger step size for faster computation (less precise but much faster) # Increase step size significantly to reduce iterations step_size = max(20, min(region_w // 3, region_h // 3, w // 10, h // 10)) # Optimized sliding window using NumPy max_density = 0.0 best_region_center = None # Pre-compute valid regions to avoid repeated calculations y_starts = list(range(0, h - region_h + 1, step_size)) x_starts = list(range(0, w - region_w + 1, step_size)) # Use vectorized operations where possible for y_start in y_starts: y_end = min(y_start + region_h, h) for x_start in x_starts: x_end = min(x_start + region_w, w) # Extract region and compute density in one operation region_mask = mask_lines_predicted[y_start:y_end, x_start:x_end] region_area = (x_end - x_start) * (y_end - y_start) if region_area == 0: continue # Vectorized line count and density calculation line_count = np.count_nonzero(region_mask) density = float(line_count) / float(region_area) # Track region with highest density if density > max_density: max_density = density best_region_center = ((x_start + x_end) // 2, (y_start + y_end) // 2) # If no region found, use frame center as fallback if best_region_center is None: best_region_center = (w // 2, h // 2) max_density = 0.0 # Calculate total line count for validation total_line_count = np.sum(mask_lines_predicted > 0) line_distribution = (total_line_count, best_region_center, max_density) print(f"Density-based region analysis: center={best_region_center}, density={max_density:.4f}, total_lines={total_line_count}") except Exception: pass # Use default keypoints if analysis fails except Exception: pass # Use default keypoints if analysis fails # Generate scaled keypoints only for selected indices # Use _TEMPLATE_KEYPOINTS if available, otherwise fall back to FOOTBALL_KEYPOINTS source_keypoints = _TEMPLATE_KEYPOINTS if _TEMPLATE_KEYPOINTS is not None else FOOTBALL_KEYPOINTS num_keypoints = len(source_keypoints) if source_keypoints is not None else 32 scaled: list[tuple[int, int]] = [] for i in range(num_keypoints): if i in selected_keypoint_indices and i < len(source_keypoints): tx, ty = source_keypoints[i] if tx > 0 and ty > 0: # Only scale non-zero keypoints x_scaled = int(round(tx * uniform_scale)) y_scaled = int(round(ty * uniform_scale)) scaled.append((x_scaled, y_scaled)) else: scaled.append((0, 0)) else: scaled.append((0, 0)) # Set unselected keypoints to (0, 0) # Ensure minimum spacing between keypoints to avoid warping artifacts # Very close keypoints can create single-pixel square contours during warping # Check if any keypoints are too close and adjust scale if needed # Optimized with NumPy for better performance min_spacing = 5 # Minimum 5 pixels between keypoints to avoid 1x1 artifacts min_spacing_sq = min_spacing * min_spacing # Use squared distance to avoid sqrt # Extract valid keypoints (non-zero) for distance checking valid_kps = np.array([(x, y) for x, y in scaled if x != 0 or y != 0], dtype=np.float32) needs_adjustment = False if len(valid_kps) > 1: # Use NumPy broadcasting for efficient distance calculation # Compute pairwise squared distances diff = valid_kps[:, None, :] - valid_kps[None, :, :] # Shape: (n, n, 2) dist_sq = np.sum(diff ** 2, axis=2) # Shape: (n, n) # Set diagonal to large value to ignore self-distances np.fill_diagonal(dist_sq, min_spacing_sq + 1) # Check if any distance is below threshold if np.any(dist_sq < min_spacing_sq): needs_adjustment = True # If keypoints are too close, slightly increase scale to maintain minimum spacing if needs_adjustment and uniform_scale < 0.25: uniform_scale = uniform_scale * 1.2 # Increase by 20% to ensure spacing uniform_scale = min(uniform_scale, 0.25) # Cap at 25% to keep it small # Recalculate selected keypoints with adjusted scale using NumPy source_array = np.array(source_keypoints[:num_keypoints] if len(source_keypoints) >= num_keypoints else source_keypoints + [(0, 0)] * (num_keypoints - len(source_keypoints)), dtype=np.float32) # Create mask for selected indices selected_mask = np.array([i in selected_keypoint_indices for i in range(num_keypoints)], dtype=bool) valid_mask = (source_array[:, 0] > 0) & (source_array[:, 1] > 0) final_mask = selected_mask & valid_mask # Vectorized scaling scaled_array = source_array.copy() scaled_array[final_mask, 0] = np.round(source_array[final_mask, 0] * uniform_scale) scaled_array[final_mask, 1] = np.round(source_array[final_mask, 1] * uniform_scale) scaled_array[~final_mask] = 0 # Convert to list of tuples scaled = [(int(x), int(y)) for x, y in scaled_array] # Use line distribution analysis (already computed above) to determine optimal pitch placement offset_x = 0 offset_y = 0 if line_distribution is not None: # Extract line distribution data (new format: total_count, best_region_center, max_density) if len(line_distribution) >= 3: total_line_count, best_region_center, max_density = line_distribution else: # Fallback if format is unexpected total_line_count = line_distribution[0] if len(line_distribution) > 0 else 0 best_region_center = None max_density = 0.0 # Adjust keypoint placement based on line distribution valid_points = [(x, y) for x, y in scaled if x > 0 and y > 0] if len(valid_points) > 0: scaled_width = max(x for x, y in valid_points) scaled_height = max(y for x, y in valid_points) margin = 5 # Only use line distribution analysis if we detected a reasonable number of lines and found a good region # Otherwise fall back to default centering if total_line_count > 100 and best_region_center is not None and max_density > 0.01: # Minimum threshold to trust the analysis # Use density-based region analysis: center sparse template on the region with highest density target_center_x, target_center_y = best_region_center # Calculate offset to center the scaled template on the target region center # The template center should align with the target region center scaled_center_x = scaled_width // 2 scaled_center_y = scaled_height // 2 offset_x = target_center_x - scaled_center_x offset_y = target_center_y - scaled_center_y # Ensure template stays within frame bounds offset_x = max(margin, min(offset_x, frame_width - scaled_width - margin)) offset_y = max(margin, min(offset_y, frame_height - scaled_height - margin)) print(f"Positioning sparse template: target_center=({target_center_x}, {target_center_y}), offset=({offset_x}, {offset_y}), scaled_size=({scaled_width}, {scaled_height}), density={max_density:.4f}") else: # Fallback to default centering # Simple center positioning offset_x = max(margin, (frame_width - scaled_width) // 2) offset_y = max(margin, (frame_height - scaled_height) // 2) offset_x = min(offset_x, frame_width - scaled_width - margin) offset_y = min(offset_y, frame_height - scaled_height - margin) offset_x = max(0, offset_x) offset_y = max(0, offset_y) else: # Default centering if no line distribution analysis valid_points = [(x, y) for x, y in scaled if x > 0 and y > 0] if len(valid_points) > 0: scaled_width = max(x for x, y in valid_points) scaled_height = max(y for x, y in valid_points) margin = 5 offset_x = max(margin, (frame_width - scaled_width) // 2) offset_y = max(margin, (frame_height - scaled_height) // 2) offset_x = min(offset_x, frame_width - scaled_width - margin) offset_y = min(offset_y, frame_height - scaled_height - margin) offset_x = max(0, offset_x) offset_y = max(0, offset_y) # Lightweight vertical adjustment: Try small vertical offsets to align top/bottom edge with lines # This improves overlap without much speed penalty if frame_image is not None and _TEMPLATE_IMAGE is not None and _TEMPLATE_KEYPOINTS is not None and line_distribution is not None: try: total_line_count, best_region_center, max_density = line_distribution if total_line_count > 100: # Only adjust if we have enough lines from keypoint_evaluation import ( project_image_using_keypoints, extract_masks_for_ground_and_lines_no_validation, extract_mask_of_ground_lines_in_image ) # Get initial positioned keypoints initial_centered = [] for x, y in scaled: if x == 0 and y == 0: initial_centered.append((0, 0)) else: new_x = x + offset_x new_y = y + offset_y new_x = max(0, min(new_x, frame_width - 1)) initial_centered.append((new_x, new_y)) # Try small vertical adjustments (only 5 positions for speed) best_adjusted_offset_y = offset_y best_overlap = 0.0 # Try vertical offsets: -15, -7, 0, 7, 15 pixels vertical_adjustments = [-15, -7, 0, 7, 15] for adj_y in vertical_adjustments: test_offset_y = offset_y + adj_y # Ensure within bounds test_offset_y = max(margin, min(test_offset_y, frame_height - scaled_height - margin)) # Generate test keypoints with adjusted vertical position test_centered = [] for x, y in scaled: if x == 0 and y == 0: test_centered.append((0, 0)) else: new_x = x + offset_x new_y = y + test_offset_y new_x = max(0, min(new_x, frame_width - 1)) new_y = max(0, min(new_y, frame_height - 1)) test_centered.append((new_x, new_y)) # Quick validation: check spacing test_corners = [test_centered[idx] for idx in sorted(selected_keypoint_indices) if idx < len(test_centered) and test_centered[idx][0] > 0] if len(test_corners) == 4: min_dist = float('inf') for i in range(len(test_corners)): for j in range(i + 1, len(test_corners)): x1, y1 = test_corners[i] x2, y2 = test_corners[j] dist = np.sqrt((x2 - x1)**2 + (y2 - y1)**2) min_dist = min(min_dist, dist) min_required_dist = max(30, min(frame_width, frame_height) * 0.1) if min_dist < min_required_dist: continue # Skip if corners too close # Project and calculate overlap try: warped = project_image_using_keypoints( image=_TEMPLATE_IMAGE, source_keypoints=_TEMPLATE_KEYPOINTS, destination_keypoints=test_centered, destination_width=frame_width, destination_height=frame_height, ) mask_ground_test, mask_lines_expected = extract_masks_for_ground_and_lines_no_validation(image=warped) mask_lines_predicted = extract_mask_of_ground_lines_in_image( image=frame_image, ground_mask=mask_ground_test ) # Calculate overlap overlap_mask = (mask_lines_expected > 0) & (mask_lines_predicted > 0) expected_pixels = np.sum(mask_lines_expected > 0) if expected_pixels > 0: overlap = np.sum(overlap_mask) / float(expected_pixels) if overlap > best_overlap: best_overlap = overlap best_adjusted_offset_y = test_offset_y except Exception: continue # Skip if projection fails # Use the best vertical offset if best_overlap > 0.0: offset_y = best_adjusted_offset_y print(f"Vertical adjustment: best overlap={best_overlap:.4f}, adjusted offset_y={offset_y}") except Exception as e: print(f"Vertical adjustment error: {e}") pass # Continue with original offset if adjustment fails # Apply centering offset centered = [] for x, y in scaled: if x == 0 and y == 0: centered.append((0, 0)) else: new_x = x + offset_x new_y = y + offset_y # Allow negative y coordinates (pitch extends above frame) # But ensure x coordinates are within frame bounds to avoid warping artifacts new_x = max(0, min(new_x, frame_width - 1)) # Allow negative y, but ensure at least some keypoints are in frame # This prevents large square artifacts from warping centered.append((new_x, new_y)) # Ensure at least some keypoints have positive y coordinates (visible in frame) # This prevents warping from creating large square artifacts visible_keypoints = [kp for kp in centered if kp[1] > 0] if len(visible_keypoints) < 4: # Not enough visible keypoints - adjust offset_y to ensure visibility # This prevents warping artifacts that create large squares min_y = min(y for x, y in centered if y != 0) if visible_keypoints else 0 if min_y < 0: adjustment = abs(min_y) + 10 # Push down by at least 10 pixels centered = [] for x, y in scaled: if x == 0 and y == 0: centered.append((0, 0)) else: new_x = x + offset_x new_y = y + offset_y + adjustment new_x = max(0, min(new_x, frame_width - 1)) new_y = max(0, new_y) # Ensure at least some are visible centered.append((new_x, new_y)) return centered # def _generate_sparse_template_keypoints( # frame_width: int, # frame_height: int, # frame_image: np.ndarray = None, # template_image: np.ndarray = None, # template_keypoints: list[tuple[int, int]] = None, # ) -> list[tuple[int, int]]: # """ # Generate sparse template keypoints that fill the frame exactly. # We map the template bounds to the frame bounds (non-uniform scale), # so the warped template covers the full frame without manual shifts. # """ # # Infer template dimensions from provided keypoints if available # if template_keypoints is not None and len(template_keypoints) > 0: # valid_template_points = [(x, y) for x, y in template_keypoints if x > 0 and y > 0] # if len(valid_template_points) > 0: # template_max_x = max(x for x, y in valid_template_points) # template_max_y = max(y for x, y in valid_template_points) # else: # template_max_x, template_max_y = (1045, 675) # else: # template_max_x, template_max_y = (1045, 675) # # Non-uniform scale to fit the frame exactly (may stretch if aspect differs) # sx = float(frame_width) / float(template_max_x if template_max_x != 0 else 1) # sy = float(frame_height) / float(template_max_y if template_max_y != 0 else 1) # source_keypoints = template_keypoints if template_keypoints is not None else FOOTBALL_KEYPOINTS # num_kps = len(source_keypoints) if source_keypoints is not None else 32 # scaled: list[tuple[int, int]] = [] # for i in range(num_kps): # tx, ty = source_keypoints[i] # if tx > 0 and ty > 0: # x_scaled = int(round(tx * sx)) # y_scaled = int(round(ty * sy)) # # Clamp to frame bounds # x_scaled = max(0, min(x_scaled, frame_width - 1)) # y_scaled = max(0, min(y_scaled, frame_height - 1)) # scaled.append((x_scaled, y_scaled)) # else: # scaled.append((0, 0)) # return scaled def _adjust_keypoints_to_pass_validation( keypoints: list[tuple[int, int]], frame_width: int = None, frame_height: int = None, ) -> list[tuple[int, int]]: """ Adjust keypoints to pass validate_projected_corners. If validation fails, try to fix by ensuring corners form a valid quadrilateral. """ if _validate_keypoints_corners(keypoints, _TEMPLATE_KEYPOINTS): return keypoints # Already valid # If validation fails, try to fix by ensuring corner keypoints are in correct order try: from keypoint_evaluation import ( INDEX_KEYPOINT_CORNER_BOTTOM_LEFT, INDEX_KEYPOINT_CORNER_BOTTOM_RIGHT, INDEX_KEYPOINT_CORNER_TOP_LEFT, INDEX_KEYPOINT_CORNER_TOP_RIGHT, ) template_keypoints = _TEMPLATE_KEYPOINTS # Get corner indices corner_indices = [ INDEX_KEYPOINT_CORNER_TOP_LEFT, INDEX_KEYPOINT_CORNER_TOP_RIGHT, INDEX_KEYPOINT_CORNER_BOTTOM_RIGHT, INDEX_KEYPOINT_CORNER_BOTTOM_LEFT, ] # Check if we have all corner keypoints corners = [] for idx in corner_indices: if idx < len(keypoints): x, y = keypoints[idx] if x > 0 and y > 0: corners.append((x, y, idx)) if len(corners) < 4: # Not enough corners - can't fix, return original return keypoints # Extract corner coordinates corner_coords = [(x, y) for x, y, _ in corners] # Check if corners form a bowtie (twisted quadrilateral) # A bowtie occurs when opposite edges intersect def segments_intersect(p1, p2, q1, q2): """Check if line segments p1-p2 and q1-q2 intersect.""" def ccw(a, b, c): return (c[1] - a[1]) * (b[0] - a[0]) > (b[1] - a[1]) * (c[0] - a[0]) return (ccw(p1, q1, q2) != ccw(p2, q1, q2)) and (ccw(p1, p2, q1) != ccw(p1, p2, q2)) # Try different corner orderings to find a valid one # Current order: top-left, top-right, bottom-right, bottom-left # If this creates a bowtie, we need to reorder # Sort corners by position to get proper order # Top row (smaller y values) top_corners = sorted([c for c in corners if c[1] <= np.mean([c[1] for c in corners])], key=lambda c: c[0]) # Bottom row (larger y values) bottom_corners = sorted([c for c in corners if c[1] > np.mean([c[1] for c in corners])], key=lambda c: c[0]) # If we have 2 top and 2 bottom corners, ensure proper ordering if len(top_corners) == 2 and len(bottom_corners) == 2: # Ensure left < right if top_corners[0][0] > top_corners[1][0]: top_corners = top_corners[::-1] if bottom_corners[0][0] > bottom_corners[1][0]: bottom_corners = bottom_corners[::-1] # Reconstruct with proper order: top-left, top-right, bottom-right, bottom-left result = list(keypoints) # Map to corner indices corner_mapping = { INDEX_KEYPOINT_CORNER_TOP_LEFT: top_corners[0], INDEX_KEYPOINT_CORNER_TOP_RIGHT: top_corners[1], INDEX_KEYPOINT_CORNER_BOTTOM_RIGHT: bottom_corners[1], INDEX_KEYPOINT_CORNER_BOTTOM_LEFT: bottom_corners[0], } for corner_idx, (x, y, _) in corner_mapping.items(): if corner_idx < len(result): result[corner_idx] = (x, y) # Validate the adjusted keypoints if _validate_keypoints_corners(result, _TEMPLATE_KEYPOINTS): return result # Alternative: If we can't fix by reordering, try using template-based scaling # for corners only, keeping other keypoints as-is if len(corners) >= 4: # Calculate scale from non-corner keypoints if available non_corner_kps = [(i, keypoints[i]) for i in range(len(keypoints)) if i not in corner_indices and keypoints[i][0] > 0 and keypoints[i][1] > 0] if len(non_corner_kps) >= 2: # Use template scaling approach scales_x = [] scales_y = [] for idx, (x, y) in non_corner_kps: if idx < len(template_keypoints): tx, ty = template_keypoints[idx] if tx > 0: scales_x.append(x / tx) if ty > 0: scales_y.append(y / ty) if scales_x and scales_y: avg_scale_x = sum(scales_x) / len(scales_x) avg_scale_y = sum(scales_y) / len(scales_y) result = list(keypoints) # Recalculate corners using template scaling for corner_idx in corner_indices: if corner_idx < len(template_keypoints): tx, ty = template_keypoints[corner_idx] new_x = int(round(tx * avg_scale_x)) new_y = int(round(ty * avg_scale_y)) if corner_idx < len(result): result[corner_idx] = (new_x, new_y) # Validate again if _validate_keypoints_corners(result, _TEMPLATE_KEYPOINTS): return result except Exception: pass # If we can't fix, return original return keypoints def fix_keypoints( results_frames: Sequence[Any], frame_results: list[tuple[int, float, bool]], frame_width: int, frame_height: int, frames: List[np.ndarray] = None, offset: int = 0, num_workers: int = None, ) -> list[Any]: """ Optimized version using batch-first approach: 1. Generate sparse keypoints for ALL frames first 2. Evaluate both sparse and calculated keypoints for ALL frames 3. Choose the one with bigger score per frame Args: results_frames: Sequence of frame results with keypoints frame_results: List of tuples (frame_index, score, adjusted_success) from calculate_and_adjust_keypoints frame_width: Frame width frame_height: Frame height frames: Optional list of frame images for validation offset: Frame offset for tracking num_workers: Number of worker threads (defaults to cpu_count()) Returns: List of processed frame results """ max_frames = len(results_frames) if max_frames == 0: return list(results_frames) # Create a dictionary mapping frame_index to (score, adjusted_success) for quick lookup frame_results_dict = {frame_index: (score, adjusted_success) for frame_index, score, adjusted_success in frame_results} if num_workers is None: # Cap workers to avoid overhead with too many threads # Optimal range is typically 8-32 workers depending on workload # Too many threads cause context switching overhead and contention # Cap at 32 even if CPU count is higher (e.g., cloud servers with 96+ CPUs) max_cpu_workers = min(32, cpu_count()) # Cap at 32 to avoid overhead max_workers = min(max_cpu_workers, max_frames) num_workers = max(1, max_workers) # Step 1: Extract calculated keypoints and pre-calculated scores from frame_results # (already calculated in calculate_and_adjust_keypoints) from keypoint_helper_v2_optimized import convert_keypoints_to_val_format calculated_keypoints_list = [] pre_calculated_scores = {} last_success_kps = None for frame_index in range(max_frames): frame_result = results_frames[frame_index] current_kps_raw = getattr(frame_result, "keypoints", []) or [] calculated_kps = convert_keypoints_to_val_format(current_kps_raw) # Get pre-calculated score from frame_results (from calculate_and_adjust_keypoints) if frame_index in frame_results_dict: score, adjusted_success = frame_results_dict[frame_index] if adjusted_success: # Only use valid scores pre_calculated_scores[frame_index] = score calculated_keypoints_list.append(calculated_kps) last_success_kps = calculated_kps else: if last_success_kps is not None: calculated_keypoints_list.append(last_success_kps) else: calculated_keypoints_list.append(calculated_kps) else: if last_success_kps is not None: calculated_keypoints_list.append(last_success_kps) else: calculated_keypoints_list.append(calculated_kps) # Step 2: Generate sparse keypoints for ALL frames in parallel print(f"Generating sparse keypoints for {max_frames} frames...") sparse_args_list = [] for frame_index in range(max_frames): frame_for_analysis = None if frames is not None and frame_index < len(frames): frame_for_analysis = frames[frame_index] sparse_args_list.append(( frame_index, frame_width, frame_height, frame_for_analysis )) sparse_keypoints_dict = {} # Check if we're on Linux - use ProcessPoolExecutor instead of ThreadPoolExecutor import platform is_linux = platform.system().lower() == 'linux' if max_frames >= 4 and num_workers > 1: try: if is_linux: from concurrent.futures import ProcessPoolExecutor, as_completed executor_class = ProcessPoolExecutor else: from concurrent.futures import ThreadPoolExecutor, as_completed executor_class = ThreadPoolExecutor with executor_class(max_workers=num_workers) as executor: futures = [executor.submit(_generate_sparse_keypoints_for_frame, args) for args in sparse_args_list] for future in as_completed(futures): try: frame_idx, sparse_kps = future.result() sparse_keypoints_dict[frame_idx] = sparse_kps except Exception as e: print(f"Error generating sparse keypoints: {e}") except Exception as e: print(f"Parallel processing failed for sparse generation: {e}, falling back to sequential") for args in sparse_args_list: try: frame_idx, sparse_kps = _generate_sparse_keypoints_for_frame(args) sparse_keypoints_dict[frame_idx] = sparse_kps except Exception: pass else: # Sequential for small batches for args in sparse_args_list: try: frame_idx, sparse_kps = _generate_sparse_keypoints_for_frame(args) sparse_keypoints_dict[frame_idx] = sparse_kps except Exception: pass # Ensure we have sparse keypoints for all frames for frame_index in range(max_frames): if frame_index not in sparse_keypoints_dict: sparse_keypoints_dict[frame_index] = [(0, 0)] * 32 # Step 3: Evaluate both sparse and calculated keypoints for ALL frames in parallel print(f"Evaluating sparse and calculated keypoints for {max_frames} frames...") eval_args_list = [] for frame_index in range(max_frames): sparse_kps = sparse_keypoints_dict[frame_index] calculated_kps = calculated_keypoints_list[frame_index] frame_for_analysis = None if frames is not None and frame_index < len(frames): frame_for_analysis = frames[frame_index] # Get pre-calculated score if available pre_calculated_score = pre_calculated_scores.get(frame_index, None) eval_args_list.append(( frame_index, sparse_kps, calculated_kps, frame_for_analysis, pre_calculated_score )) evaluation_results = {} if max_frames >= 4 and num_workers > 1: try: if is_linux: from concurrent.futures import ProcessPoolExecutor, as_completed executor_class = ProcessPoolExecutor else: from concurrent.futures import ThreadPoolExecutor, as_completed executor_class = ThreadPoolExecutor with executor_class(max_workers=num_workers) as executor: futures = [executor.submit(_evaluate_keypoints_for_frame, args) for args in eval_args_list] for future in as_completed(futures): try: frame_idx, sparse_score, calculated_score, sparse_kps, calculated_kps = future.result() evaluation_results[frame_idx] = (sparse_score, calculated_score, sparse_kps, calculated_kps) except Exception as e: print(f"Error evaluating keypoints: {e}") except Exception as e: print(f"Threading failed for evaluation: {e}, falling back to sequential") for args in eval_args_list: try: frame_idx, sparse_score, calculated_score, sparse_kps, calculated_kps = _evaluate_keypoints_for_frame(args) evaluation_results[frame_idx] = (sparse_score, calculated_score, sparse_kps, calculated_kps) except Exception: pass else: # Sequential for small batches for args in eval_args_list: try: frame_idx, sparse_score, calculated_score, sparse_kps, calculated_kps = _evaluate_keypoints_for_frame(args) evaluation_results[frame_idx] = (sparse_score, calculated_score, sparse_kps, calculated_kps) except Exception: pass # Step 4: Choose the keypoint set with bigger score per frame print(f"Choosing best keypoints for {max_frames} frames...") for frame_index in range(max_frames): frame_result = results_frames[frame_index] # Get evaluation results for this frame if frame_index in evaluation_results: sparse_score, calculated_score, sparse_kps, calculated_kps = evaluation_results[frame_index] # Choose the one with bigger score if calculated_score > sparse_score: final_keypoints = calculated_kps print(f"Frame {frame_index}: Using calculated keypoints (score: {calculated_score:.4f} > sparse: {sparse_score:.4f})") else: final_keypoints = sparse_kps print(f"Frame {frame_index}: Using sparse keypoints (score: {sparse_score:.4f} >= calculated: {calculated_score:.4f})") else: # Fallback to sparse if evaluation failed final_keypoints = sparse_keypoints_dict.get(frame_index, [(0, 0)] * 32) print(f"Frame {frame_index}: Using sparse keypoints (evaluation failed)") setattr(frame_result, "keypoints", list(convert_keypoints_to_val_format(final_keypoints))) return list(results_frames) def run_keypoints_post_processing( results_frames: Sequence[Any], frame_width: int, frame_height: int, frames: List[np.ndarray] = None, template_keypoints: list[tuple[int, int]] = None, template_image: np.ndarray = None, offset: int = 0, num_workers: int = None, ) -> list[Any]: """ Optimized post-processing with multiprocessing support. Args: results_frames: Sequence of frame results with keypoints frame_width: Frame width frame_height: Frame height frames: Optional list of frame images for validation template_keypoints: Optional template keypoints (defaults to TEMPLATE_KEYPOINTS) template_image: Optional pre-loaded template image (from miner constructor) offset: Frame offset for tracking (defaults to 0) num_workers: Number of worker processes for multiprocessing (defaults to cpu_count()) Returns: List of processed frame results """ # Initialize module-level template variables (use pre-loaded template_image) _initialize_template_variables(template_keypoints, template_image) # Calculate and adjust keypoints for all frames, getting scores and success status frame_results = calculate_and_adjust_keypoints( results_frames, frame_width, frame_height, frames, offset, num_workers ) return fix_keypoints( results_frames, frame_results, frame_width, frame_height, frames, offset, num_workers )