"""Floor direction analysis helpers for the FastAPI room editor backend.""" from dataclasses import dataclass import math from typing import Any import cv2 import numpy as np @dataclass(frozen=True) class FloorPolygon: points: list[tuple[float, float]] bbox: tuple[float, float, float, float] | None = None @dataclass(frozen=True) class FloorSegmentation: label: str confidence: float polygons: list[FloorPolygon] source_path: str = "root" raw_keys: list[str] | None = None warnings: list[str] | None = None source_width: int | None = None source_height: int | None = None def parse_floor_segmentation(payload: Any) -> FloorSegmentation: root_payload = payload payload, source_path, parse_warnings = find_segmentation_payload(payload) raw_keys = sorted(payload.keys()) if isinstance(payload, dict) else [] if not isinstance(payload, dict): return FloorSegmentation( label="unknown", confidence=0.0, polygons=[], source_path=source_path, raw_keys=raw_keys, warnings=parse_warnings + ["segmentation_response_not_object"], ) source_width = extract_int_value(payload.get("width")) or extract_int_value( root_payload.get("width") if isinstance(root_payload, dict) else None ) source_height = extract_int_value(payload.get("height")) or extract_int_value( root_payload.get("height") if isinstance(root_payload, dict) else None ) label = str(payload.get("label") or payload.get("class") or payload.get("name") or "unknown") try: confidence = float(payload.get("confidence", payload.get("score", 0.0))) except (TypeError, ValueError): confidence = 0.0 polygons: list[FloorPolygon] = [] for polygon_payload in extract_polygon_payloads(payload): if not isinstance(polygon_payload, dict): continue raw_points = polygon_payload.get("points") or polygon_payload.get("polygon") or [] points: list[tuple[float, float]] = [] for point in raw_points: if not isinstance(point, (list, tuple)) or len(point) < 2: continue try: points.append((float(point[0]), float(point[1]))) except (TypeError, ValueError): continue raw_bbox = polygon_payload.get("bbox") bbox = None if isinstance(raw_bbox, (list, tuple)) and len(raw_bbox) >= 4: try: bbox = ( float(raw_bbox[0]), float(raw_bbox[1]), float(raw_bbox[2]), float(raw_bbox[3]), ) except (TypeError, ValueError): bbox = None if len(points) >= 3: polygons.append(FloorPolygon(points=points, bbox=bbox)) return FloorSegmentation( label=label, confidence=max(0.0, min(1.0, confidence)), polygons=polygons, source_path=source_path, raw_keys=raw_keys, warnings=parse_warnings, source_width=source_width, source_height=source_height, ) def extract_int_value(value: Any) -> int | None: try: parsed = int(round(float(value))) except (TypeError, ValueError): return None return parsed if parsed > 0 else None def find_segmentation_payload(payload: Any) -> tuple[Any, str, list[str]]: warnings: list[str] = [] if isinstance(payload, list): floor_item = find_floor_item(payload) if floor_item is not None: return floor_item, "root[floor]", warnings if any(isinstance(item, dict) and has_polygon_data(item) for item in payload): warnings.append("segmentation_response_list_without_floor_label") return {"label": "unknown", "polygons": []}, "root", warnings return payload, "root", warnings if not isinstance(payload, dict): return payload, "root", warnings if has_polygon_data(payload): return payload, "root", warnings for key in ( "floor", "data", "result", "results", "prediction", "predictions", "segments", "response", "output", ): nested = payload.get(key) if nested is None: continue if isinstance(nested, dict) and has_polygon_data(nested): return nested, f"root.{key}", warnings if isinstance(nested, list): floor_item = find_floor_item(nested) if floor_item is not None: return floor_item, f"root.{key}[floor]", warnings if any(isinstance(item, dict) and has_polygon_data(item) for item in nested): warnings.append("segmentation_response_list_without_floor_label") warnings.append("segmentation_response_missing_polygons") return payload, "root", warnings def has_polygon_data(payload: dict[str, Any]) -> bool: return bool(extract_polygon_payloads(payload)) def find_floor_item(items: list[Any]) -> dict[str, Any] | None: for item in items: if not isinstance(item, dict): continue label = str(item.get("label") or item.get("class") or item.get("name") or "").lower() if label == "floor" and has_polygon_data(item): return item return None def extract_polygon_payloads(payload: dict[str, Any]) -> list[Any]: raw_polygons = payload.get("polygons") if isinstance(raw_polygons, list): return raw_polygons raw_polygon = payload.get("polygon") if isinstance(raw_polygon, list): if raw_polygon and all(isinstance(point, (list, tuple)) for point in raw_polygon): return [{"points": raw_polygon}] return raw_polygon raw_points = payload.get("points") if isinstance(raw_points, list): return [{"points": raw_points}] return [] @dataclass(frozen=True) class DetectedLine: x1: int y1: int x2: int y2: int angle_degrees: float length: float @dataclass(frozen=True) class TextureOrientation: angle_degrees: float confidence: float support: int concentration: float @dataclass(frozen=True) class LineFamilyCluster: line_indices: list[int] angle_degrees: float weight: float concentration: float distinct_offset_count: int score: float @dataclass(frozen=True) class DirectionResult: angle_degrees: float | None secondary_angle_degrees: float | None direction_label: str confidence: float floor_area_ratio: float line_count: int dominant_line_count: int grid_pattern_detected: bool dominant_lines: list[DetectedLine] warnings: list[str] overlay_bgr: np.ndarray | None = None render_angle_degrees: float | None = None def analyze_floor_direction( image_bgr: np.ndarray, segmentation: FloorSegmentation, *, include_overlay: bool = False, render_transform: list[float] | np.ndarray | None = None, surface_uv: np.ndarray | None = None, surface_indices: np.ndarray | None = None, surface_plane_ids: np.ndarray | None = None, ) -> DirectionResult: height, width = image_bgr.shape[:2] warnings: list[str] = list(segmentation.warnings or []) warnings.extend(uploaded_image_quality_warnings(image_bgr)) mask = build_floor_mask( width=width, height=height, polygons=segmentation.polygons, source_width=segmentation.source_width, source_height=segmentation.source_height, ) floor_pixels = int(np.count_nonzero(mask)) total_pixels = height * width floor_area_ratio = floor_pixels / total_pixels if total_pixels else 0.0 if segmentation.confidence < 0.75: warnings.append("low_segmentation_confidence") if segmentation_dimensions_differ(segmentation.source_width, segmentation.source_height, width, height): warnings.append("segmentation_coordinate_scale_applied") if not segmentation.polygons or floor_pixels == 0: warnings.append("no_floor_polygon_detected") return DirectionResult( angle_degrees=None, secondary_angle_degrees=None, direction_label="unknown", confidence=0.0, floor_area_ratio=0.0, line_count=0, dominant_line_count=0, grid_pattern_detected=False, dominant_lines=[], warnings=warnings, overlay_bgr=create_overlay(image_bgr, mask, [], None) if include_overlay else None, ) if floor_area_ratio < 0.03: warnings.append("floor_area_too_small") inner_mask = erode_mask(mask) if int(np.count_nonzero(inner_mask)) == 0: inner_mask = mask lines = detect_candidate_lines(image_bgr, inner_mask) hough_orientation = dominant_orientation(lines, image_shape=inner_mask.shape[:2]) texture_orientation = estimate_texture_orientation(image_bgr, inner_mask) selected_orientation = select_orientation(hough_orientation, texture_orientation, warnings) if selected_orientation is None: warnings.append("no_strong_floor_lines_detected") return DirectionResult( angle_degrees=None, secondary_angle_degrees=None, direction_label="unknown", confidence=0.0, floor_area_ratio=floor_area_ratio, line_count=0, dominant_line_count=0, grid_pattern_detected=False, dominant_lines=[], warnings=warnings, overlay_bgr=create_overlay(image_bgr, mask, [], None) if include_overlay else None, ) angle_degrees, orientation_source = selected_orientation render_angle_degrees = estimate_surface_uv_grout_rotation( lines, image_bgr=image_bgr, surface_uv=surface_uv, surface_indices=surface_indices, surface_plane_ids=surface_plane_ids, image_shape=(height, width), ) if render_angle_degrees is not None: warnings.append("grout_direction_projected_to_surface_uv") else: render_angle_degrees = estimate_rectified_grout_rotation( lines, render_transform=render_transform, ) if render_angle_degrees is not None: warnings.append("grout_direction_rectified_for_renderer") dominant_lines: list[DetectedLine] = [] peak_weight_ratio = 0.0 concentration = 0.0 orthogonal_ratio = 0.0 if hough_orientation is not None: ( _hough_angle, dominant_lines, peak_weight_ratio, concentration, orthogonal_ratio, ) = hough_orientation secondary_angle_degrees = detect_secondary_grid_angle( lines=lines, primary_angle_degrees=angle_degrees, primary_line_count=len(dominant_lines), ) grid_pattern_detected = secondary_angle_degrees is not None if len(dominant_lines) < 5: warnings.append("weak_line_support") if orthogonal_ratio > 0.65: warnings.append("possible_square_tile_grid_or_ambiguous_perpendicular_lines") if grid_pattern_detected: warnings.append("grid_pattern_detected") if orientation_source == "texture": warnings.append("texture_orientation_used") elif orientation_source == "combined": warnings.append("hough_texture_orientation_combined") hough_confidence = score_confidence( segmentation_confidence=segmentation.confidence, floor_area_ratio=floor_area_ratio, line_count=len(lines), dominant_line_count=len(dominant_lines), peak_weight_ratio=peak_weight_ratio, concentration=concentration, ) texture_confidence = 0.0 if texture_orientation is not None: texture_confidence = score_texture_confidence( segmentation_confidence=segmentation.confidence, floor_area_ratio=floor_area_ratio, texture_confidence=texture_orientation.confidence, ) confidence = final_orientation_confidence( source=orientation_source, hough_confidence=hough_confidence, texture_confidence=texture_confidence, ) direction_label = label_direction(angle_degrees) overlay = ( create_overlay(image_bgr, mask, lines, angle_degrees, secondary_angle_degrees) if include_overlay else None ) return DirectionResult( angle_degrees=round(angle_degrees, 2), secondary_angle_degrees=round(secondary_angle_degrees, 2) if secondary_angle_degrees is not None else None, direction_label=direction_label, confidence=round(confidence, 3), floor_area_ratio=round(floor_area_ratio, 4), line_count=len(lines), dominant_line_count=len(dominant_lines), grid_pattern_detected=grid_pattern_detected, dominant_lines=dominant_lines[:30], warnings=warnings, overlay_bgr=overlay, render_angle_degrees=render_angle_degrees, ) def build_floor_mask( *, width: int, height: int, polygons: list[FloorPolygon], source_width: int | None = None, source_height: int | None = None, ) -> np.ndarray: mask = np.zeros((height, width), dtype=np.uint8) scale_x = width / source_width if source_width and source_width > 0 else 1.0 scale_y = height / source_height if source_height and source_height > 0 else 1.0 for polygon in polygons: points = np.array(polygon.points, dtype=np.float32) if points.shape[0] < 3: continue points[:, 0] *= scale_x points[:, 1] *= scale_y points[:, 0] = np.clip(points[:, 0], 0, width - 1) points[:, 1] = np.clip(points[:, 1], 0, height - 1) int_points = np.round(points).astype(np.int32) cv2.fillPoly(mask, [int_points], 255) kernel_size = max(3, int(round(min(width, height) * 0.008))) if kernel_size % 2 == 0: kernel_size += 1 kernel = np.ones((kernel_size, kernel_size), dtype=np.uint8) return cv2.morphologyEx(mask, cv2.MORPH_CLOSE, kernel) def segmentation_dimensions_differ( source_width: int | None, source_height: int | None, image_width: int, image_height: int, ) -> bool: if not source_width or not source_height: return False return abs(source_width - image_width) > 2 or abs(source_height - image_height) > 2 def uploaded_image_quality_warnings(image_bgr: np.ndarray) -> list[str]: if largest_flat_row_region_ratio(image_bgr) >= 0.35: return ["uploaded_image_contains_large_flat_filled_region"] return [] def largest_flat_row_region_ratio(image_bgr: np.ndarray) -> float: height, width = image_bgr.shape[:2] if height == 0 or width == 0: return 0.0 gray = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2GRAY) row_std = np.std(gray, axis=1) flat_rows = row_std < 4.0 longest_run = 0 current_run = 0 for is_flat in flat_rows: if bool(is_flat): current_run += 1 longest_run = max(longest_run, current_run) else: current_run = 0 return longest_run / height def erode_mask(mask: np.ndarray) -> np.ndarray: height, width = mask.shape[:2] kernel_size = max(5, int(round(min(width, height) * 0.015))) if kernel_size % 2 == 0: kernel_size += 1 kernel = np.ones((kernel_size, kernel_size), dtype=np.uint8) return cv2.erode(mask, kernel, iterations=1) def detect_candidate_lines(image_bgr: np.ndarray, mask: np.ndarray) -> list[DetectedLine]: height, width = image_bgr.shape[:2] floor_pixels = max(1, int(np.count_nonzero(mask))) gray = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2GRAY) inside_pixels = gray[mask > 0] fill_value = int(np.median(inside_pixels)) if inside_pixels.size else int(np.median(gray)) prepared = gray.copy() prepared[mask == 0] = fill_value clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8)) enhanced = clahe.apply(prepared) blurred = cv2.GaussianBlur(enhanced, (5, 5), 0) min_dimension = min(width, height) min_line_length = max(28, int(min_dimension * 0.055)) max_line_gap = max(8, int(min_dimension * 0.025)) hough_threshold = max(18, int(math.sqrt(floor_pixels) * 0.035)) grout_edges = detect_grout_edges(enhanced, mask) grout_lines = detect_hough_lines_from_edges( grout_edges, min_line_length=max(20, int(min_dimension * 0.045)), max_line_gap=max(10, int(min_dimension * 0.03)), thresholds=[ max(10, int(math.sqrt(floor_pixels) * 0.02)), max(8, int(math.sqrt(floor_pixels) * 0.015)), ], ) grout_lines = deduplicate_lines(grout_lines) if has_usable_grout_line_support(grout_lines, image_shape=(height, width)): return grout_lines attempts = [ (50, 150, hough_threshold, min_line_length), (30, 100, max(12, int(hough_threshold * 0.75)), max(20, int(min_line_length * 0.8))), (80, 220, hough_threshold, min_line_length), ] best_lines: list[DetectedLine] = [] for canny_low, canny_high, threshold, min_length in attempts: edges = cv2.Canny(blurred, canny_low, canny_high) edges = cv2.bitwise_and(edges, mask) raw_lines = cv2.HoughLinesP( edges, rho=1, theta=np.pi / 180, threshold=threshold, minLineLength=min_length, maxLineGap=max_line_gap, ) lines = normalize_hough_lines(raw_lines, min_length=min_length) if len(lines) > len(best_lines): best_lines = lines if len(best_lines) >= 12: break return deduplicate_lines(best_lines) def detect_grout_edges(enhanced_gray: np.ndarray, mask: np.ndarray) -> np.ndarray: height, width = enhanced_gray.shape[:2] min_dimension = min(width, height) if min_dimension <= 0 or int(np.count_nonzero(mask)) < 100: return np.zeros_like(enhanced_gray) kernel_size = max(7, int(round(min_dimension * 0.018))) if kernel_size % 2 == 0: kernel_size += 1 kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (kernel_size, kernel_size)) dark_response = cv2.morphologyEx(enhanced_gray, cv2.MORPH_BLACKHAT, kernel) light_response = cv2.morphologyEx(enhanced_gray, cv2.MORPH_TOPHAT, kernel) response = cv2.max(dark_response, light_response) response = cv2.GaussianBlur(response, (3, 3), 0) response = cv2.bitwise_and(response, mask) valid_response = response[mask > 0] valid_response = valid_response[valid_response > 0] if valid_response.size < 80: return np.zeros_like(enhanced_gray) percentile_threshold = float(np.percentile(valid_response, 84)) spread_threshold = float(np.mean(valid_response) + np.std(valid_response) * 0.35) threshold = max(5.0, min(percentile_threshold, spread_threshold)) binary = np.zeros_like(response) binary[response >= threshold] = 255 cleanup_kernel = np.ones((3, 3), dtype=np.uint8) binary = cv2.morphologyEx(binary, cv2.MORPH_OPEN, cleanup_kernel) binary = cv2.morphologyEx(binary, cv2.MORPH_CLOSE, cleanup_kernel) low_threshold = max(3, int(round(threshold * 0.45))) high_threshold = max(low_threshold + 3, int(round(threshold * 1.35))) edges = cv2.Canny(response, low_threshold, high_threshold) support = cv2.dilate(binary, cleanup_kernel, iterations=1) edges = cv2.bitwise_and(edges, support) return cv2.bitwise_and(edges, mask) def detect_hough_lines_from_edges( edges: np.ndarray, *, min_line_length: int, max_line_gap: int, thresholds: list[int], ) -> list[DetectedLine]: if int(np.count_nonzero(edges)) < max(24, min_line_length): return [] best_lines: list[DetectedLine] = [] for threshold in thresholds: raw_lines = cv2.HoughLinesP( edges, rho=1, theta=np.pi / 180, threshold=threshold, minLineLength=min_line_length, maxLineGap=max_line_gap, ) lines = normalize_hough_lines(raw_lines, min_length=min_line_length) if len(lines) > len(best_lines): best_lines = lines return best_lines def has_usable_grout_line_support( lines: list[DetectedLine], *, image_shape: tuple[int, int], ) -> bool: if len(lines) < 3: return False clusters = cluster_line_families(lines, image_shape=image_shape) if not clusters: return False best_cluster = max(clusters, key=lambda cluster: cluster.score) if best_cluster.concentration < 0.70: return False return ( len(best_cluster.line_indices) >= 4 and best_cluster.distinct_offset_count >= 2 ) or ( len(best_cluster.line_indices) >= 3 and best_cluster.distinct_offset_count >= 3 ) def normalize_hough_lines(raw_lines: np.ndarray | None, *, min_length: int) -> list[DetectedLine]: if raw_lines is None: return [] lines: list[DetectedLine] = [] for raw_line in raw_lines.reshape(-1, 4): x1, y1, x2, y2 = [int(value) for value in raw_line] dx = x2 - x1 dy = y2 - y1 length = math.hypot(dx, dy) if length < min_length: continue angle = normalized_line_angle(dx, dy) lines.append( DetectedLine( x1=x1, y1=y1, x2=x2, y2=y2, angle_degrees=round(angle, 2), length=round(length, 2), ) ) return lines def deduplicate_lines(lines: list[DetectedLine]) -> list[DetectedLine]: seen: set[tuple[int, int, int, int, int]] = set() unique: list[DetectedLine] = [] for line in sorted(lines, key=lambda item: item.length, reverse=True): angle_bin = int(round(line.angle_degrees / 3.0)) midpoint_x = int(round(((line.x1 + line.x2) / 2) / 8.0)) midpoint_y = int(round(((line.y1 + line.y2) / 2) / 8.0)) length_bin = int(round(line.length / 12.0)) key = (angle_bin, midpoint_x, midpoint_y, length_bin, int(line.length > 80)) if key in seen: continue seen.add(key) unique.append(line) return unique[:200] def dominant_orientation( lines: list[DetectedLine], image_shape: tuple[int, int] | None = None, ) -> tuple[float, list[DetectedLine], float, float, float] | None: if not lines: return None angles = np.array([line.angle_degrees for line in lines], dtype=np.float64) weights = np.array([max(1.0, line.length) for line in lines], dtype=np.float64) total_weight = float(np.sum(weights)) if total_weight <= 0: return None clusters = cluster_line_families(lines, image_shape=image_shape) if not clusters: return None dominant_cluster = max(clusters, key=lambda cluster: cluster.score) dominant_angle = dominant_cluster.angle_degrees peak_weight_ratio = dominant_cluster.weight / total_weight concentration = dominant_cluster.concentration orthogonal_angle = (dominant_angle + 90.0) % 180.0 orthogonal_weight = float( np.sum( [ weight for angle, weight in zip(angles, weights, strict=False) if angular_distance_degrees(float(angle), orthogonal_angle) <= 12.0 ] ) ) orthogonal_ratio = orthogonal_weight / dominant_cluster.weight if dominant_cluster.weight else 0.0 dominant_lines = [lines[index] for index in dominant_cluster.line_indices] dominant_lines = sorted(dominant_lines, key=lambda item: item.length, reverse=True) return dominant_angle, dominant_lines, peak_weight_ratio, concentration, orthogonal_ratio def estimate_rectified_grout_rotation( lines: list[DetectedLine], *, render_transform: list[float] | np.ndarray | None, ) -> float | None: """Return a cardinal tile rotation in the renderer's floor-plane coordinates.""" transform = normalize_render_transform(render_transform) if transform is None or len(lines) < 4: return None rectified_lines = transform_detected_lines(lines, transform) orientation = dominant_orientation(rectified_lines) if orientation is None: return None angle_degrees, dominant_lines, _peak_weight_ratio, concentration, _orthogonal_ratio = orientation if len(dominant_lines) < 4 or concentration < 0.76: return None return snap_to_tile_axis(angle_degrees) def estimate_surface_uv_grout_rotation( lines: list[DetectedLine], *, image_bgr: np.ndarray | None = None, surface_uv: np.ndarray | None, surface_indices: np.ndarray | None, surface_plane_ids: np.ndarray | None, image_shape: tuple[int, int], ) -> float | None: """Measure grout in the same surface-UV plane sampled by the renderer.""" uv_grid, plane_grid = build_surface_uv_grids( surface_uv=surface_uv, surface_indices=surface_indices, surface_plane_ids=surface_plane_ids, image_shape=image_shape, ) if uv_grid is None: return None # Image-space Hough families are unreliable for oblique views: parallel floor # seams converge, while furniture edges remain parallel. Rectifying the source # into surface UV makes the actual grout parallel before it is scored. if image_bgr is not None: rectified = rectify_floor_image_to_surface_uv(image_bgr, uv_grid, plane_grid) if rectified is not None: rectified_image, rectified_mask = rectified rectified_lines = detect_candidate_lines(rectified_image, rectified_mask) rectified_angle = estimate_repeated_grout_orientation( rectified_lines, image_shape=rectified_mask.shape[:2], ) if rectified_angle is not None: return rectified_angle if len(lines) < 4: return None uv_lines = project_lines_to_surface_uv(lines, uv_grid, plane_grid) return estimate_repeated_grout_orientation(uv_lines) def rectify_floor_image_to_surface_uv( image_bgr: np.ndarray, uv_grid: np.ndarray, plane_grid: np.ndarray | None, ) -> tuple[np.ndarray, np.ndarray] | None: """Splat source pixels into a uniform, metric surface-UV image. The output uses one scale for U and V so line angles are directly usable as renderer rotations. When the floor contains multiple surfaces, retain only the largest assigned plane: their local UV bases are intentionally separate. """ if image_bgr.shape[:2] != uv_grid.shape[:2]: return None valid = np.isfinite(uv_grid).all(axis=2) if plane_grid is not None: plane_values = plane_grid[valid & (plane_grid != 255)] if plane_values.size: dominant_plane = int(np.bincount(plane_values).argmax()) valid &= plane_grid == dominant_plane if int(valid.sum()) < 600: return None uv = uv_grid[valid].astype(np.float64) lower = np.percentile(uv, 0.5, axis=0) upper = np.percentile(uv, 99.5, axis=0) span = upper - lower if not np.isfinite(span).all() or float(np.min(span)) <= 1e-5: return None # Keep the rectified image detailed enough for grout, without making this # CPU-only preprocessing grow with the uploaded image resolution. aspect = float(span[0] / span[1]) target_long_side = int( np.clip( round(math.sqrt(int(valid.sum()) * max(aspect, 1.0 / aspect))), 256, 1024, ) ) scale = target_long_side / float(np.max(span)) rectified_width = int(np.clip(round(float(span[0]) * scale) + 1, 96, 1024)) rectified_height = int(np.clip(round(float(span[1]) * scale) + 1, 96, 1024)) if rectified_width < 96 or rectified_height < 96: return None source_gray = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2GRAY) source_values = source_gray[valid].astype(np.float64) local_u = uv[:, 0] - lower[0] local_v = uv[:, 1] - lower[1] xs = np.clip(np.rint(local_u * scale).astype(np.int32), 0, rectified_width - 1) ys = np.clip(np.rint(local_v * scale).astype(np.int32), 0, rectified_height - 1) flat_indices = ys * rectified_width + xs pixel_count = rectified_width * rectified_height counts = np.bincount(flat_indices, minlength=pixel_count) values = np.bincount(flat_indices, weights=source_values, minlength=pixel_count) covered = counts > 0 if int(covered.sum()) < max(500, int(pixel_count * 0.08)): return None fill_value = int(np.median(source_values)) if source_values.size else 0 rectified_gray = np.full(pixel_count, fill_value, dtype=np.uint8) rectified_gray[covered] = np.clip(values[covered] / counts[covered], 0, 255).astype(np.uint8) rectified_mask = (covered.reshape(rectified_height, rectified_width).astype(np.uint8) * 255) rectified_gray = rectified_gray.reshape(rectified_height, rectified_width) return cv2.cvtColor(rectified_gray, cv2.COLOR_GRAY2BGR), rectified_mask def estimate_repeated_grout_orientation( lines: list[DetectedLine], *, image_shape: tuple[int, int] | None = None, ) -> float | None: """Select a direction only when it has repeated, regularly spaced seams.""" if len(lines) < 4: return None candidates: list[tuple[float, float]] = [] for cluster in cluster_line_families(lines, image_shape=image_shape): if len(cluster.line_indices) < 4 or cluster.concentration < 0.76: continue offsets, offset_weights, median_length = consolidated_line_offsets( lines, cluster.line_indices, angle_degrees=cluster.angle_degrees, image_shape=image_shape, ) if len(offsets) < 3: continue periodicity = repeated_offset_periodicity(offsets, offset_weights) if periodicity < 0.76: continue support = min(1.0, len(cluster.line_indices) / 12.0) offset_support = min(1.0, (len(offsets) - 1) / 6.0) length_support = min(1.0, median_length / max(1.0, offset_span_reference(offsets, image_shape))) quality = ( 0.28 * periodicity + 0.24 * cluster.concentration + 0.22 * support + 0.18 * offset_support + 0.08 * length_support ) candidates.append((quality, cluster.angle_degrees)) if not candidates: return None quality, angle_degrees = max(candidates, key=lambda candidate: candidate[0]) return angle_degrees if quality >= 0.68 else None def consolidated_line_offsets( lines: list[DetectedLine], line_indices: list[int], *, angle_degrees: float, image_shape: tuple[int, int] | None, ) -> tuple[np.ndarray, np.ndarray, float]: center_x, center_y, _min_dimension = hough_offset_reference_frame( lines, line_indices, image_shape=image_shape, ) radians = math.radians(angle_degrees) normal_x = -math.sin(radians) normal_y = math.cos(radians) values = [] for index in line_indices: line = lines[index] midpoint_x = (line.x1 + line.x2) * 0.5 midpoint_y = (line.y1 + line.y2) * 0.5 offset = (midpoint_x - center_x) * normal_x + (midpoint_y - center_y) * normal_y values.append((offset, max(1.0, line.length))) if not values: return np.empty(0), np.empty(0), 0.0 lengths = np.array([weight for _offset, weight in values], dtype=np.float64) median_length = float(np.median(lengths)) tolerance = max(1.5, median_length * 0.035) if image_shape is not None: # Canny commonly returns the two sides of one grout seam as separate # Hough segments after rectification. Merge that pair before testing # inter-seam periodicity, while keeping adjacent physical seams apart. tolerance = max(tolerance, min(image_shape) * 0.024) groups: list[list[float]] = [] for offset, weight in sorted(values, key=lambda item: item[0]): if not groups or abs(offset - groups[-1][0]) > tolerance: groups.append([offset, weight]) continue previous_offset, previous_weight = groups[-1] total_weight = previous_weight + weight groups[-1][0] = (previous_offset * previous_weight + offset * weight) / total_weight groups[-1][1] = total_weight return ( np.array([group[0] for group in groups], dtype=np.float64), np.array([group[1] for group in groups], dtype=np.float64), median_length, ) def repeated_offset_periodicity(offsets: np.ndarray, weights: np.ndarray) -> float: """Return how well seam offsets fit one regularly repeating lattice.""" if len(offsets) < 3: return 0.0 span = float(offsets[-1] - offsets[0]) if span <= 1e-6: return 0.0 pairwise = offsets[np.newaxis, :] - offsets[:, np.newaxis] distances = pairwise[np.triu_indices(len(offsets), k=1)] distances = distances[distances > 1e-6] if not len(distances): return 0.0 minimum_spacing = span / max(40.0, float(len(offsets) * 4)) candidates = [] for distance in distances: for divisor in range(1, 7): spacing = float(distance / divisor) if minimum_spacing <= spacing <= span * 0.85: candidates.append(spacing) if not candidates: return 0.0 best = 0.0 weights = np.maximum(np.asarray(weights, dtype=np.float64), 1.0) for spacing in candidates: normalized = (offsets - offsets[0]) / spacing residual = np.abs(normalized - np.rint(normalized)) alignment = np.exp(-((residual / 0.10) ** 2)) support = float(np.average(alignment, weights=weights)) occupied_steps = len(np.unique(np.rint(normalized))) coverage = min(1.0, occupied_steps / max(3, len(offsets))) best = max(best, support * (0.82 + 0.18 * coverage)) return best def offset_span_reference( offsets: np.ndarray, image_shape: tuple[int, int] | None, ) -> float: if image_shape is not None: return float(min(image_shape)) * 0.20 return max(1e-6, float(offsets[-1] - offsets[0])) def build_surface_uv_grids( *, surface_uv: np.ndarray | None, surface_indices: np.ndarray | None, surface_plane_ids: np.ndarray | None, image_shape: tuple[int, int], ) -> tuple[np.ndarray | None, np.ndarray | None]: if surface_uv is None or surface_indices is None: return None, None height, width = image_shape uv_values = np.asarray(surface_uv, dtype=np.float32) indices = np.asarray(surface_indices, dtype=np.int64).reshape(-1) if uv_values.ndim != 2 or uv_values.shape[1] != 2 or len(uv_values) != len(indices): return None, None if not len(indices) or np.any(indices < 0) or np.any(indices >= height * width): return None, None uv_grid = np.full((height * width, 2), np.nan, dtype=np.float32) uv_grid[indices] = uv_values uv_grid = uv_grid.reshape(height, width, 2) plane_grid = None if surface_plane_ids is not None: plane_ids = np.asarray(surface_plane_ids, dtype=np.uint8).reshape(-1) if len(plane_ids) == len(indices): plane_grid = np.full(height * width, 255, dtype=np.uint8) plane_grid[indices] = plane_ids plane_grid = plane_grid.reshape(height, width) return uv_grid, plane_grid def project_lines_to_surface_uv( lines: list[DetectedLine], uv_grid: np.ndarray, plane_grid: np.ndarray | None, ) -> list[DetectedLine]: height, width = uv_grid.shape[:2] projected_lines: list[DetectedLine] = [] for line in lines: sample_count = int(np.clip(round(line.length / 8.0) + 1, 9, 64)) fractions = np.linspace(0.0, 1.0, sample_count, dtype=np.float32) xs = np.clip( np.rint(line.x1 + (line.x2 - line.x1) * fractions).astype(np.int32), 0, width - 1, ) ys = np.clip( np.rint(line.y1 + (line.y2 - line.y1) * fractions).astype(np.int32), 0, height - 1, ) points = uv_grid[ys, xs] valid = np.isfinite(points).all(axis=1) if plane_grid is not None: sampled_planes = plane_grid[ys, xs] valid_planes = sampled_planes[valid & (sampled_planes != 255)] if valid_planes.size: dominant_plane = int(np.bincount(valid_planes).argmax()) valid &= sampled_planes == dominant_plane if int(valid.sum()) < max(6, int(math.ceil(sample_count * 0.55))): continue sampled_points = points[valid].astype(np.float64) center = np.mean(sampled_points, axis=0) _unused, _singular_values, basis = np.linalg.svd(sampled_points - center, full_matrices=False) direction = basis[0] distances = (sampled_points - center) @ direction start = center + direction * float(np.min(distances)) end = center + direction * float(np.max(distances)) dx = float(end[0] - start[0]) dy = float(end[1] - start[1]) length = math.hypot(dx, dy) if length <= 1e-5: continue projected_lines.append( DetectedLine( x1=float(start[0]), y1=float(start[1]), x2=float(end[0]), y2=float(end[1]), angle_degrees=normalized_line_angle(dx, dy), length=length, ) ) return projected_lines def normalize_render_transform( value: list[float] | np.ndarray | None, ) -> np.ndarray | None: if value is None: return None transform = np.asarray(value, dtype=np.float64) if transform.size != 9: return None transform = transform.reshape(3, 3) if not np.isfinite(transform).all() or abs(float(np.linalg.det(transform))) < 1e-10: return None return transform.astype(np.float32) def transform_detected_lines( lines: list[DetectedLine], transform: np.ndarray, ) -> list[DetectedLine]: rectified_lines: list[DetectedLine] = [] for line in lines: source_points = np.array([[[line.x1, line.y1], [line.x2, line.y2]]], dtype=np.float32) mapped_points = cv2.perspectiveTransform(source_points, transform)[0] if not np.isfinite(mapped_points).all(): continue start, end = mapped_points dx = float(end[0] - start[0]) dy = float(end[1] - start[1]) length = math.hypot(dx, dy) if length <= 1e-5: continue rectified_lines.append( DetectedLine( x1=float(start[0]), y1=float(start[1]), x2=float(end[0]), y2=float(end[1]), angle_degrees=normalized_line_angle(dx, dy), length=length, ) ) return rectified_lines def snap_to_tile_axis(angle_degrees: float) -> float | None: normalized_angle = angle_degrees % 180.0 nearest_axis = 0.0 if angular_distance_degrees(normalized_angle, 0.0) <= 45.0 else 90.0 if angular_distance_degrees(normalized_angle, nearest_axis) > 28.0: return None return nearest_axis def cluster_line_families( lines: list[DetectedLine], *, image_shape: tuple[int, int] | None = None, ) -> list[LineFamilyCluster]: raw_clusters: list[list[int]] = [] for line_index in sorted(range(len(lines)), key=lambda index: lines[index].length, reverse=True): line_angle = lines[line_index].angle_degrees best_cluster_index = None best_distance = float("inf") for cluster_index, cluster_indices in enumerate(raw_clusters): cluster_angle, _concentration, _weight = line_family_stats(lines, cluster_indices) distance = angular_distance_degrees(line_angle, cluster_angle) if distance <= 10.0 and distance < best_distance: best_cluster_index = cluster_index best_distance = distance if best_cluster_index is None: raw_clusters.append([line_index]) else: raw_clusters[best_cluster_index].append(line_index) raw_clusters = merge_nearby_line_family_clusters(lines, raw_clusters) return [ build_line_family_cluster(lines, cluster_indices, image_shape=image_shape) for cluster_indices in raw_clusters if cluster_indices ] def merge_nearby_line_family_clusters( lines: list[DetectedLine], raw_clusters: list[list[int]], ) -> list[list[int]]: merged_clusters = [list(cluster) for cluster in raw_clusters] changed = True while changed and len(merged_clusters) > 1: changed = False for first_index in range(len(merged_clusters)): first_angle, _first_concentration, _first_weight = line_family_stats( lines, merged_clusters[first_index], ) for second_index in range(first_index + 1, len(merged_clusters)): second_angle, _second_concentration, _second_weight = line_family_stats( lines, merged_clusters[second_index], ) if angular_distance_degrees(first_angle, second_angle) > 8.0: continue merged_clusters[first_index].extend(merged_clusters[second_index]) del merged_clusters[second_index] changed = True break if changed: break return merged_clusters def build_line_family_cluster( lines: list[DetectedLine], line_indices: list[int], *, image_shape: tuple[int, int] | None = None, ) -> LineFamilyCluster: angle, concentration, weight = line_family_stats(lines, line_indices) distinct_offset_count = count_distinct_line_offsets( lines, line_indices, angle_degrees=angle, image_shape=image_shape, ) score = line_family_selection_score( weight=weight, concentration=concentration, line_count=len(line_indices), distinct_offset_count=distinct_offset_count, ) return LineFamilyCluster( line_indices=list(line_indices), angle_degrees=angle, weight=weight, concentration=concentration, distinct_offset_count=distinct_offset_count, score=score, ) def line_family_stats( lines: list[DetectedLine], line_indices: list[int], ) -> tuple[float, float, float]: angles = np.array([lines[index].angle_degrees for index in line_indices], dtype=np.float64) weights = np.array([max(1.0, lines[index].length) for index in line_indices], dtype=np.float64) total_weight = float(np.sum(weights)) if total_weight <= 0.0: return 0.0, 0.0, 0.0 angle, concentration = circular_mean_with_concentration(angles, weights) return angle, concentration, total_weight def count_distinct_line_offsets( lines: list[DetectedLine], line_indices: list[int], *, angle_degrees: float, image_shape: tuple[int, int] | None = None, ) -> int: if not line_indices: return 0 center_x, center_y, min_dimension = hough_offset_reference_frame( lines, line_indices, image_shape=image_shape, ) tolerance = max(6.0, min_dimension * 0.018) radians = math.radians(angle_degrees) normal_x = -math.sin(radians) normal_y = math.cos(radians) offsets = [] for index in line_indices: line = lines[index] midpoint_x = (line.x1 + line.x2) * 0.5 midpoint_y = (line.y1 + line.y2) * 0.5 rho = (midpoint_x - center_x) * normal_x + (midpoint_y - center_y) * normal_y offsets.append((rho, max(1.0, line.length))) offset_groups: list[list[float]] = [] for rho, weight in sorted(offsets, key=lambda item: item[0]): if not offset_groups or abs(rho - offset_groups[-1][0]) > tolerance: offset_groups.append([rho, weight]) continue previous_rho, previous_weight = offset_groups[-1] combined_weight = previous_weight + weight offset_groups[-1][0] = (previous_rho * previous_weight + rho * weight) / combined_weight offset_groups[-1][1] = combined_weight return len(offset_groups) def hough_offset_reference_frame( lines: list[DetectedLine], line_indices: list[int], *, image_shape: tuple[int, int] | None = None, ) -> tuple[float, float, float]: if image_shape is not None: height, width = image_shape min_dimension = max(1.0, float(min(width, height))) return (width - 1) * 0.5, (height - 1) * 0.5, min_dimension xs: list[float] = [] ys: list[float] = [] for index in line_indices: line = lines[index] xs.extend([float(line.x1), float(line.x2)]) ys.extend([float(line.y1), float(line.y2)]) min_x = min(xs) if xs else 0.0 max_x = max(xs) if xs else 1.0 min_y = min(ys) if ys else 0.0 max_y = max(ys) if ys else 1.0 width = max(1.0, max_x - min_x) height = max(1.0, max_y - min_y) return min_x + width * 0.5, min_y + height * 0.5, min(width, height) def line_family_selection_score( *, weight: float, concentration: float, line_count: int, distinct_offset_count: int, ) -> float: distinct_bonus = min(6, max(0, distinct_offset_count - 1)) line_bonus = min(8, max(0, line_count - 1)) repeated_line_multiplier = 1.0 + 0.18 * distinct_bonus + 0.05 * line_bonus return weight * max(0.35, concentration) * repeated_line_multiplier def detect_secondary_grid_angle( *, lines: list[DetectedLine], primary_angle_degrees: float, primary_line_count: int, ) -> float | None: if len(lines) < 8 or primary_line_count < 4: return None target_angle = (primary_angle_degrees + 90.0) % 180.0 secondary_lines = [ line for line in lines if angular_distance_degrees(line.angle_degrees, target_angle) <= 14.0 ] if len(secondary_lines) < max(4, int(round(primary_line_count * 0.35))): return None primary_weight = sum( line.length for line in lines if angular_distance_degrees(line.angle_degrees, primary_angle_degrees) <= 14.0 ) secondary_weight = sum(line.length for line in secondary_lines) if secondary_weight < max(120.0, primary_weight * 0.24): return None angle, concentration = circular_mean_with_concentration( np.array([line.angle_degrees for line in secondary_lines], dtype=np.float64), np.array([line.length for line in secondary_lines], dtype=np.float64), ) if concentration < 0.76: return None return angle def estimate_texture_orientation( image_bgr: np.ndarray, mask: np.ndarray, ) -> TextureOrientation | None: component_mask = largest_mask_component(mask) texture_mask = erode_mask(component_mask) if int(np.count_nonzero(texture_mask)) == 0: texture_mask = component_mask height, width = image_bgr.shape[:2] floor_pixels = int(np.count_nonzero(texture_mask)) if floor_pixels < max(500, int(height * width * 0.015)): return None gray = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2GRAY) inside_pixels = gray[texture_mask > 0] fill_value = int(np.median(inside_pixels)) if inside_pixels.size else int(np.median(gray)) prepared = gray.copy() prepared[texture_mask == 0] = fill_value clahe = cv2.createCLAHE(clipLimit=1.7, tileGridSize=(8, 8)) enhanced = clahe.apply(prepared) blurred = cv2.GaussianBlur(enhanced, (3, 3), 0) gradient_x = cv2.Scharr(blurred, cv2.CV_32F, 1, 0) gradient_y = cv2.Scharr(blurred, cv2.CV_32F, 0, 1) gradient_magnitude = cv2.magnitude(gradient_x, gradient_y) valid_magnitudes = gradient_magnitude[texture_mask > 0] valid_magnitudes = valid_magnitudes[valid_magnitudes > 0] if valid_magnitudes.size < 200: return None low_threshold = max(3.0, float(np.percentile(valid_magnitudes, 45))) high_threshold = float(np.percentile(valid_magnitudes, 97)) if high_threshold <= low_threshold: high_threshold = float(np.max(valid_magnitudes)) valid_texture = ( (texture_mask > 0) & (gradient_magnitude >= low_threshold) & (gradient_magnitude <= high_threshold) ) cell_size = max(28, int(round(min(width, height) * 0.07))) angles: list[float] = [] weights: list[float] = [] for y in range(0, height, cell_size): for x in range(0, width, cell_size): cell_valid = valid_texture[y : y + cell_size, x : x + cell_size] valid_count = int(np.count_nonzero(cell_valid)) if valid_count < max(40, int(cell_size * cell_size * 0.08)): continue cell_gx = gradient_x[y : y + cell_size, x : x + cell_size][cell_valid] cell_gy = gradient_y[y : y + cell_size, x : x + cell_size][cell_valid] cell_mag = gradient_magnitude[y : y + cell_size, x : x + cell_size][cell_valid] tensor_xx = float(np.sum(cell_gx * cell_gx)) tensor_yy = float(np.sum(cell_gy * cell_gy)) tensor_xy = float(np.sum(cell_gx * cell_gy)) energy = tensor_xx + tensor_yy if energy <= 0.0: continue coherence = math.sqrt((tensor_xx - tensor_yy) ** 2 + 4.0 * tensor_xy**2) / energy if coherence < 0.12: continue gradient_angle = math.degrees( 0.5 * math.atan2(2.0 * tensor_xy, tensor_xx - tensor_yy) ) line_angle = (gradient_angle + 90.0) % 180.0 median_mag = float(np.median(cell_mag)) weights.append(coherence * math.log1p(median_mag) * math.sqrt(valid_count)) angles.append(line_angle) if len(angles) < 4: return None angle_array = np.array(angles, dtype=np.float64) weight_array = np.array(weights, dtype=np.float64) if float(np.sum(weight_array)) <= 0.0: return None cap = float(np.percentile(weight_array, 85)) if cap > 0: weight_array = np.minimum(weight_array, cap) angle, concentration = circular_mean_with_concentration(angle_array, weight_array) support_score = min(1.0, len(angles) / 18.0) confidence = max(0.0, min(1.0, (0.65 * concentration + 0.35 * support_score))) confidence *= support_score if confidence < 0.22 or concentration < 0.22: return None return TextureOrientation( angle_degrees=angle, confidence=confidence, support=len(angles), concentration=concentration, ) def select_orientation( hough_orientation: tuple[float, list[DetectedLine], float, float, float] | None, texture_orientation: TextureOrientation | None, warnings: list[str], ) -> tuple[float, str] | None: if hough_orientation is None and texture_orientation is None: return None if hough_orientation is None: return texture_orientation.angle_degrees, "texture" if texture_orientation else None hough_angle, dominant_lines, peak_weight_ratio, concentration, orthogonal_ratio = hough_orientation if texture_orientation is None: return hough_angle, "hough" hough_quality = hough_orientation_quality( dominant_line_count=len(dominant_lines), peak_weight_ratio=peak_weight_ratio, concentration=concentration, orthogonal_ratio=orthogonal_ratio, ) texture_quality = texture_orientation.confidence distance = angular_distance_degrees(hough_angle, texture_orientation.angle_degrees) if is_strong_hough_grout_orientation( dominant_line_count=len(dominant_lines), peak_weight_ratio=peak_weight_ratio, concentration=concentration, orthogonal_ratio=orthogonal_ratio, ): if distance > 10.0: warnings.append("strong_grout_lines_used_over_texture") return hough_angle, "hough" if distance <= 10.0: angle = circular_mean_degrees( [hough_angle, texture_orientation.angle_degrees], [max(hough_quality, 0.1), max(texture_quality, 0.1)], ) return angle, "combined" if ( texture_quality >= 0.42 and texture_orientation.support >= 6 and ( hough_quality < 0.62 or len(dominant_lines) < 8 or peak_weight_ratio < 0.56 or orthogonal_ratio > 0.45 ) ): warnings.append("hough_texture_disagreement_using_texture") return texture_orientation.angle_degrees, "texture" if texture_quality >= hough_quality + 0.16 and texture_orientation.support >= 8: warnings.append("hough_texture_disagreement_using_texture") return texture_orientation.angle_degrees, "texture" if hough_quality >= texture_quality + 0.18: warnings.append("hough_texture_disagreement_using_hough") return hough_angle, "hough" angle = circular_mean_degrees( [hough_angle, texture_orientation.angle_degrees], [max(hough_quality, 0.1), max(texture_quality, 0.1)], ) warnings.append("hough_texture_orientation_combined_after_disagreement") return angle, "combined" def hough_orientation_quality( *, dominant_line_count: int, peak_weight_ratio: float, concentration: float, orthogonal_ratio: float, ) -> float: line_support = min(1.0, dominant_line_count / 12.0) orthogonal_penalty = max(0.0, 1.0 - min(1.0, orthogonal_ratio)) return max( 0.0, min( 1.0, 0.28 * line_support + 0.32 * max(0.0, min(1.0, peak_weight_ratio)) + 0.30 * max(0.0, min(1.0, concentration)) + 0.10 * orthogonal_penalty, ), ) def is_strong_hough_grout_orientation( *, dominant_line_count: int, peak_weight_ratio: float, concentration: float, orthogonal_ratio: float, ) -> bool: return ( dominant_line_count >= 8 and peak_weight_ratio >= 0.58 and concentration >= 0.86 and orthogonal_ratio <= 0.48 ) or ( dominant_line_count >= 12 and peak_weight_ratio >= 0.50 and concentration >= 0.82 and orthogonal_ratio <= 0.55 ) def circular_mean_degrees(angles: list[float], weights: list[float]) -> float: angle, _concentration = circular_mean_with_concentration( np.array(angles, dtype=np.float64), np.array(weights, dtype=np.float64), ) return angle def circular_mean_with_concentration( angles: np.ndarray, weights: np.ndarray, ) -> tuple[float, float]: total_weight = float(np.sum(weights)) if total_weight <= 0.0: return 0.0, 0.0 doubled_angles = np.deg2rad(angles * 2.0) vector_x = float(np.sum(np.cos(doubled_angles) * weights)) vector_y = float(np.sum(np.sin(doubled_angles) * weights)) angle = (math.degrees(math.atan2(vector_y, vector_x)) / 2.0) % 180.0 concentration = math.hypot(vector_x, vector_y) / total_weight return angle, concentration def normalized_line_angle(dx: float, dy: float) -> float: angle = math.degrees(math.atan2(dy, dx)) % 180.0 if angle >= 180.0: angle -= 180.0 return angle def angular_distance_degrees(first: float, second: float) -> float: diff = abs((first - second) % 180.0) return min(diff, 180.0 - diff) def score_confidence( *, segmentation_confidence: float, floor_area_ratio: float, line_count: int, dominant_line_count: int, peak_weight_ratio: float, concentration: float, ) -> float: segmentation_score = max(0.0, min(1.0, segmentation_confidence)) area_score = max(0.0, min(1.0, floor_area_ratio / 0.28)) line_score = max(0.0, min(1.0, line_count / 30.0)) dominant_count_score = max(0.0, min(1.0, dominant_line_count / 14.0)) peak_score = max(0.0, min(1.0, peak_weight_ratio)) concentration_score = max(0.0, min(1.0, concentration)) score = ( 0.18 * segmentation_score + 0.10 * area_score + 0.18 * line_score + 0.16 * dominant_count_score + 0.24 * peak_score + 0.14 * concentration_score ) return max(0.0, min(1.0, score)) def score_texture_confidence( *, segmentation_confidence: float, floor_area_ratio: float, texture_confidence: float, ) -> float: segmentation_score = max(0.0, min(1.0, segmentation_confidence)) area_score = max(0.0, min(1.0, floor_area_ratio / 0.28)) texture_score = max(0.0, min(1.0, texture_confidence)) score = 0.20 * segmentation_score + 0.14 * area_score + 0.66 * texture_score return max(0.0, min(1.0, score)) def final_orientation_confidence( *, source: str, hough_confidence: float, texture_confidence: float, ) -> float: if source == "texture": return texture_confidence if source == "combined": return max(hough_confidence, texture_confidence) * 0.96 return hough_confidence def label_direction(angle_degrees: float) -> str: if angle_degrees < 10.0 or angle_degrees >= 170.0: return "left_to_right" if angle_degrees < 35.0: return "slight_down_right" if angle_degrees < 70.0: return "diagonal_down_right" if angle_degrees < 110.0: return "top_to_bottom" if angle_degrees < 145.0: return "diagonal_up_right" return "slight_up_right" def create_overlay( image_bgr: np.ndarray, mask: np.ndarray, _lines: list[DetectedLine], angle_degrees: float | None, secondary_angle_degrees: float | None = None, ) -> np.ndarray: overlay = image_bgr.copy() if secondary_angle_degrees is not None: draw_direction_line(overlay, mask, secondary_angle_degrees, color_bgr=(255, 210, 20)) if angle_degrees is not None: draw_direction_line(overlay, mask, angle_degrees, color_bgr=(0, 255, 80)) return overlay def draw_direction_line( image_bgr: np.ndarray, mask: np.ndarray, angle_degrees: float, *, color_bgr: tuple[int, int, int], ) -> None: endpoints = direction_line_endpoints(mask, angle_degrees) if endpoints is None: return start, end = endpoints min_dimension = min(mask.shape[:2]) thickness = max(5, int(round(min_dimension / 155))) cv2.line(image_bgr, start, end, (0, 0, 0), thickness + 6, cv2.LINE_AA) cv2.line(image_bgr, start, end, color_bgr, thickness, cv2.LINE_AA) cv2.circle(image_bgr, start, thickness + 1, color_bgr, -1, cv2.LINE_AA) cv2.circle(image_bgr, end, thickness + 1, color_bgr, -1, cv2.LINE_AA) def direction_line_endpoints( mask: np.ndarray, angle_degrees: float, ) -> tuple[tuple[int, int], tuple[int, int]] | None: component_mask = largest_mask_component(mask) points = cv2.findNonZero(component_mask) if points is None: return None height, width = mask.shape[:2] floor_points = points.reshape(-1, 2).astype(np.float64) center = floor_points.mean(axis=0) center_x = int(round(center[0])) center_y = int(round(center[1])) if ( not (0 <= center_x < width and 0 <= center_y < height) or component_mask[center_y, center_x] == 0 ): distances = np.sum((floor_points - center) ** 2, axis=1) center = floor_points[int(np.argmin(distances))] unit = np.array( [ math.cos(math.radians(angle_degrees)), math.sin(math.radians(angle_degrees)), ], dtype=np.float64, ) projections = (floor_points - center) @ unit min_projection = float(np.percentile(projections, 1)) max_projection = float(np.percentile(projections, 99)) sample_count = max(2, int(round(max_projection - min_projection)) + 1) sample_count = min(sample_count, max(width, height) * 3) samples = np.linspace(min_projection, max_projection, sample_count) active_samples: list[float] = [] runs: list[list[float]] = [] current_run: list[float] = [] for sample in samples: x, y = center + sample * unit ix = int(round(x)) iy = int(round(y)) is_inside = 0 <= ix < width and 0 <= iy < height and component_mask[iy, ix] > 0 if is_inside: current_run.append(float(sample)) active_samples.append(float(sample)) elif current_run: runs.append(current_run) current_run = [] if current_run: runs.append(current_run) if runs: best_run = max(runs, key=len) start_projection = best_run[0] end_projection = best_run[-1] elif active_samples: start_projection = min(active_samples) end_projection = max(active_samples) else: start_projection = min_projection end_projection = max_projection minimum_length = max(30.0, min(width, height) * 0.08) if end_projection - start_projection < minimum_length: midpoint = (start_projection + end_projection) / 2.0 start_projection = midpoint - minimum_length / 2.0 end_projection = midpoint + minimum_length / 2.0 start = center + start_projection * unit end = center + end_projection * unit return clip_point_to_image(start, width, height), clip_point_to_image(end, width, height) def largest_mask_component(mask: np.ndarray) -> np.ndarray: binary_mask = np.where(mask > 0, 255, 0).astype(np.uint8) component_count, labels, stats, _centroids = cv2.connectedComponentsWithStats(binary_mask, 8) if component_count <= 1: return binary_mask largest_label = 1 + int(np.argmax(stats[1:, cv2.CC_STAT_AREA])) return np.where(labels == largest_label, 255, 0).astype(np.uint8) def clip_point_to_image( point: np.ndarray, width: int, height: int, ) -> tuple[int, int]: x = int(round(float(point[0]))) y = int(round(float(point[1]))) return max(0, min(width - 1, x)), max(0, min(height - 1, y))