"""Multi-cue floor direction estimation in renderer coordinates.""" from dataclasses import asdict, dataclass import math import cv2 import numpy as np @dataclass(frozen=True) class DirectionCue: source: str angle_degrees: float confidence: float support: int region_support: int ambiguity_degrees: float | None details: dict def metadata(self) -> dict: payload = asdict(self) payload["angleDegrees"] = payload.pop("angle_degrees") payload["regionSupport"] = payload.pop("region_support") payload["ambiguityDegrees"] = payload.pop("ambiguity_degrees") return payload def normalize_angle(angle_degrees: float, period: float = 180.0) -> float: return float(angle_degrees % period) def angular_distance(first: float, second: float, period: float = 180.0) -> float: difference = abs((first - second) % period) return float(min(difference, period - difference)) def periodic_mean( angles_degrees: np.ndarray, weights: np.ndarray, *, period: float, ) -> tuple[float, float]: if len(angles_degrees) == 0: return 0.0, 0.0 weights = np.maximum(np.asarray(weights, dtype=np.float64), 0.0) total_weight = float(weights.sum()) if total_weight <= 1e-9: return 0.0, 0.0 phase = np.asarray(angles_degrees, dtype=np.float64) * (2.0 * np.pi / period) vector_x = float(np.sum(np.cos(phase) * weights)) vector_y = float(np.sum(np.sin(phase) * weights)) angle = math.atan2(vector_y, vector_x) * (period / (2.0 * np.pi)) concentration = math.hypot(vector_x, vector_y) / total_weight return normalize_angle(angle, period), float(np.clip(concentration, 0.0, 1.0)) def build_surface_uv_grids( *, image_shape: tuple[int, int], surface_uv: np.ndarray | None, surface_indices: np.ndarray | None, surface_plane_ids: np.ndarray | None, ) -> tuple[np.ndarray | None, np.ndarray | None]: height, width = image_shape if surface_uv is None or surface_indices is None: return None, None if surface_uv.ndim != 2 or surface_uv.shape[1] != 2: return None, None if len(surface_uv) != len(surface_indices): return None, None flat_size = height * width indices = np.asarray(surface_indices, dtype=np.int64) valid_indices = (indices >= 0) & (indices < flat_size) uv_grid = np.full((flat_size, 2), np.nan, dtype=np.float32) uv_grid[indices[valid_indices]] = surface_uv[valid_indices].astype(np.float32) plane_grid = None if surface_plane_ids is not None and len(surface_plane_ids) == len(indices): plane_grid = np.full(flat_size, 255, dtype=np.uint8) plane_grid[indices[valid_indices]] = surface_plane_ids[valid_indices].astype(np.uint8) return ( uv_grid.reshape(height, width, 2), plane_grid.reshape(height, width) if plane_grid is not None else None, ) def rectify_floor_image( image_rgb: np.ndarray, surface_mask: np.ndarray, *, render_transform: list[float] | np.ndarray | None, surface_uv: np.ndarray | None, surface_indices: np.ndarray | None, surface_plane_ids: np.ndarray | None, ) -> tuple[np.ndarray, np.ndarray, dict] | None: height, width = surface_mask.shape[:2] uv_grid, plane_grid = build_surface_uv_grids( image_shape=(height, width), surface_uv=surface_uv, surface_indices=surface_indices, surface_plane_ids=surface_plane_ids, ) if uv_grid is not None: result = rectify_from_uv(image_rgb, uv_grid, plane_grid) if result is not None: gray, mask = result return gray, mask, {"source": "surface-uv", "width": gray.shape[1], "height": gray.shape[0]} transform = normalize_transform(render_transform) if transform is None: return None result = rectify_from_transform(image_rgb, surface_mask, transform) if result is None: return None gray, mask = result return gray, mask, {"source": "floor-transform", "width": gray.shape[1], "height": gray.shape[0]} def rectify_from_uv( image_rgb: np.ndarray, uv_grid: np.ndarray, plane_grid: np.ndarray | None, ) -> tuple[np.ndarray, np.ndarray] | 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()) < 800: 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 scale = choose_rectification_scale(span, int(valid.sum())) output_width = int(np.clip(round(float(span[0]) * scale) + 1, 96, 896)) output_height = int(np.clip(round(float(span[1]) * scale) + 1, 96, 896)) if output_width < 96 or output_height < 96: return None gray = cv2.cvtColor(image_rgb, cv2.COLOR_RGB2GRAY) values = gray[valid].astype(np.float64) xy = np.rint((uv - lower) * scale).astype(np.int32) xs = np.clip(xy[:, 0], 0, output_width - 1) ys = np.clip(xy[:, 1], 0, output_height - 1) flat_indices = ys * output_width + xs pixel_count = output_width * output_height counts = np.bincount(flat_indices, minlength=pixel_count) sums = np.bincount(flat_indices, weights=values, minlength=pixel_count) covered = counts > 0 if int(covered.sum()) < max(600, int(pixel_count * 0.1)): return None fill_value = int(np.median(values)) output = np.full(pixel_count, fill_value, dtype=np.uint8) output[covered] = np.clip(sums[covered] / counts[covered], 0, 255).astype(np.uint8) mask = (covered.reshape(output_height, output_width).astype(np.uint8) * 255) return output.reshape(output_height, output_width), mask def rectify_from_transform( image_rgb: np.ndarray, surface_mask: np.ndarray, transform: np.ndarray, ) -> tuple[np.ndarray, np.ndarray] | None: ys, xs = np.where(surface_mask > 0) if len(xs) < 800: return None stride = max(1, len(xs) // 30000) source_points = np.column_stack((xs[::stride], ys[::stride])).astype(np.float32).reshape(1, -1, 2) mapped = cv2.perspectiveTransform(source_points, transform)[0] mapped = mapped[np.isfinite(mapped).all(axis=1)] if len(mapped) < 500: return None lower = np.percentile(mapped, 0.5, axis=0) upper = np.percentile(mapped, 99.5, axis=0) span = upper - lower if not np.isfinite(span).all() or float(np.min(span)) <= 1e-5: return None scale = choose_rectification_scale(span, len(mapped)) output_width = int(np.clip(round(float(span[0]) * scale) + 1, 96, 896)) output_height = int(np.clip(round(float(span[1]) * scale) + 1, 96, 896)) if output_width < 96 or output_height < 96: return None normalization = np.array([ [scale, 0.0, -float(lower[0]) * scale], [0.0, scale, -float(lower[1]) * scale], [0.0, 0.0, 1.0], ], dtype=np.float64) warp = normalization @ transform gray = cv2.cvtColor(image_rgb, cv2.COLOR_RGB2GRAY) rectified = cv2.warpPerspective( gray, warp, (output_width, output_height), flags=cv2.INTER_LINEAR, borderMode=cv2.BORDER_REPLICATE, ) rectified_mask = cv2.warpPerspective( (surface_mask > 0).astype(np.uint8) * 255, warp, (output_width, output_height), flags=cv2.INTER_NEAREST, ) return rectified, rectified_mask def choose_rectification_scale(span: np.ndarray, sample_count: int) -> float: aspect = float(span[0] / span[1]) target_long_side = int(np.clip( round(math.sqrt(sample_count * max(aspect, 1.0 / aspect))), 320, 896, )) return target_long_side / max(float(np.max(span)), 1e-6) def estimate_material_cue(gray: np.ndarray, mask: np.ndarray) -> DirectionCue | None: valid = mask > 0 if int(valid.sum()) < 1000: return None kernel_size = max(3, int(round(min(gray.shape) * 0.012))) | 1 inner_mask = cv2.erode(mask, np.ones((kernel_size, kernel_size), dtype=np.uint8)) if int(np.count_nonzero(inner_mask)) < 800: inner_mask = mask valid = inner_mask > 0 fill_value = int(np.median(gray[valid])) prepared = gray.copy() prepared[~valid] = fill_value prepared = cv2.createCLAHE(clipLimit=1.8, tileGridSize=(8, 8)).apply(prepared) sigma = max(3.0, min(gray.shape) / 36.0) illumination = cv2.GaussianBlur(prepared, (0, 0), sigmaX=sigma) detail = prepared.astype(np.float32) - illumination.astype(np.float32) gx = cv2.Sobel(detail, cv2.CV_32F, 1, 0, ksize=3) gy = cv2.Sobel(detail, cv2.CV_32F, 0, 1, ksize=3) magnitude = np.hypot(gx, gy) gradient_values = magnitude[valid] if gradient_values.size < 800 or float(np.percentile(gradient_values, 90)) < 2.0: return None threshold = float(np.percentile(gradient_values, 68)) block_angles = [] block_weights = [] block_regions = [] rows, columns = 4, 4 height, width = gray.shape[:2] for row in range(rows): y0, y1 = row * height // rows, (row + 1) * height // rows for column in range(columns): x0, x1 = column * width // columns, (column + 1) * width // columns block_valid = valid[y0:y1, x0:x1] selected = block_valid & (magnitude[y0:y1, x0:x1] >= threshold) if int(selected.sum()) < 45: continue angles = ( np.degrees(np.arctan2( gy[y0:y1, x0:x1][selected], gx[y0:y1, x0:x1][selected], )) + 90.0 ) % 180.0 weights = magnitude[y0:y1, x0:x1][selected].astype(np.float64) angle, concentration = periodic_mean(angles, weights, period=180.0) if concentration < 0.22: continue block_angles.append(angle) block_weights.append(float(weights.sum()) * concentration) block_regions.append(row * columns + column) tensor_cue = cue_from_region_angles( source="rectified-structure-tensor", angles=block_angles, weights=block_weights, regions=block_regions, period=180.0, minimum_regions=3, ) line_cue = estimate_rectified_line_cue(gray, mask) if line_cue is None: line_cue = estimate_rectified_line_cue(prepared, mask) return fuse_material_cues(tensor_cue, line_cue) def estimate_rectified_line_cue(gray: np.ndarray, mask: np.ndarray) -> DirectionCue | None: detector = cv2.createLineSegmentDetector(cv2.LSD_REFINE_STD) detected = detector.detect(gray)[0] if detected is None: return None height, width = gray.shape[:2] min_length = max(18.0, min(height, width) * 0.055) angles = [] weights = [] regions = [] for raw_line in detected.reshape(-1, 4): x1, y1, x2, y2 = [float(value) for value in raw_line] dx, dy = x2 - x1, y2 - y1 length = math.hypot(dx, dy) if length < min_length: continue sample_x = np.clip(np.rint(np.linspace(x1, x2, 9)).astype(np.int32), 0, width - 1) sample_y = np.clip(np.rint(np.linspace(y1, y2, 9)).astype(np.int32), 0, height - 1) mask_support = float(np.mean(mask[sample_y, sample_x] > 0)) if mask_support < 0.78: continue midpoint_x = (x1 + x2) * 0.5 midpoint_y = (y1 + y2) * 0.5 angles.append(normalize_angle(math.degrees(math.atan2(dy, dx)))) weights.append(length * mask_support) regions.append( min(3, int(midpoint_y * 4 / max(height, 1))) * 4 + min(3, int(midpoint_x * 4 / max(width, 1))) ) return cue_from_region_angles( source="rectified-repeated-lines", angles=angles, weights=weights, regions=regions, period=180.0, minimum_regions=3, ) def cue_from_region_angles( *, source: str, angles: list[float], weights: list[float], regions: list[int], period: float, minimum_regions: int, ) -> DirectionCue | None: if len(angles) < 4: return None angle_array = np.asarray(angles, dtype=np.float64) weight_array = np.asarray(weights, dtype=np.float64) angle, concentration = periodic_mean(angle_array, weight_array, period=period) tolerance = 12.0 if period == 180.0 else 9.0 agreement = np.array( [angular_distance(float(value), angle, period) <= tolerance for value in angle_array], dtype=bool, ) if int(agreement.sum()) < 4: return None agreeing_regions = {region for region, keep in zip(regions, agreement, strict=False) if keep} if len(agreeing_regions) < minimum_regions: return None agreement_ratio = float(weight_array[agreement].sum() / max(float(weight_array.sum()), 1e-6)) region_score = min(1.0, len(agreeing_regions) / 7.0) support_score = min(1.0, int(agreement.sum()) / 12.0) confidence = float(np.clip( 0.38 * concentration + 0.28 * agreement_ratio + 0.20 * region_score + 0.14 * support_score, 0.0, 1.0, )) if confidence < 0.48: return None return DirectionCue( source=source, angle_degrees=round(angle, 3), confidence=round(confidence, 3), support=int(agreement.sum()), region_support=len(agreeing_regions), ambiguity_degrees=None, details={ "concentration": round(concentration, 4), "agreementRatio": round(agreement_ratio, 4), }, ) def fuse_material_cues( tensor_cue: DirectionCue | None, line_cue: DirectionCue | None, ) -> DirectionCue | None: if tensor_cue is None: return line_cue if line_cue is None: return tensor_cue difference = angular_distance(tensor_cue.angle_degrees, line_cue.angle_degrees) if difference > 15.0: stronger = max((tensor_cue, line_cue), key=lambda cue: cue.confidence) return DirectionCue( source=stronger.source, angle_degrees=stronger.angle_degrees, confidence=round(stronger.confidence * 0.72, 3), support=stronger.support, region_support=stronger.region_support, ambiguity_degrees=None, details={**stronger.details, "cueDisagreementDegrees": round(difference, 3)}, ) angles = np.array([tensor_cue.angle_degrees, line_cue.angle_degrees]) weights = np.array([tensor_cue.confidence, line_cue.confidence]) angle, concentration = periodic_mean(angles, weights, period=180.0) confidence = float(np.clip( max(tensor_cue.confidence, line_cue.confidence) * 0.72 + min(tensor_cue.confidence, line_cue.confidence) * 0.28 + concentration * 0.08, 0.0, 1.0, )) return DirectionCue( source="rectified-material-consensus", angle_degrees=round(angle, 3), confidence=round(confidence, 3), support=tensor_cue.support + line_cue.support, region_support=max(tensor_cue.region_support, line_cue.region_support), ambiguity_degrees=None, details={ "tensorAngleDegrees": tensor_cue.angle_degrees, "lineAngleDegrees": line_cue.angle_degrees, "cueAgreementDegrees": round(difference, 3), }, ) def normalize_vector(vector: np.ndarray) -> np.ndarray | None: vector = np.asarray(vector, dtype=np.float64) length = float(np.linalg.norm(vector)) if not np.isfinite(vector).all() or length <= 1e-7: return None return vector / length def normalize_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-12: return None return transform def world_direction_to_render_angle( direction: np.ndarray, *, render_u_axis: np.ndarray | None, render_v_axis: np.ndarray | None, intrinsics: np.ndarray | None, render_transform: np.ndarray | None, ) -> float | None: direction = normalize_vector(direction) if direction is None: return None if render_u_axis is not None and render_v_axis is not None: u_axis = normalize_vector(render_u_axis) v_axis = normalize_vector(render_v_axis) if u_axis is not None and v_axis is not None: return normalize_angle(math.degrees(math.atan2( float(direction @ v_axis), float(direction @ u_axis), ))) if intrinsics is None or render_transform is None: return None intrinsics = np.asarray(intrinsics, dtype=np.float64).reshape(3, 3) vanishing_point = intrinsics @ direction mapped = render_transform @ vanishing_point if not np.isfinite(mapped).all() or float(np.linalg.norm(mapped[:2])) <= 1e-7: return None return normalize_angle(math.degrees(math.atan2(float(mapped[1]), float(mapped[0])))) def choose_depth_axis( base_angle_degrees: float, *, floor_normal: np.ndarray, render_u_axis: np.ndarray | None, render_v_axis: np.ndarray | None, intrinsics: np.ndarray | None, render_transform: np.ndarray | None, ) -> float: camera_forward = np.array([0.0, 0.0, 1.0], dtype=np.float64) floor_normal = normalize_vector(floor_normal) if floor_normal is None: return normalize_angle(base_angle_degrees) projected_forward = camera_forward - floor_normal * float(camera_forward @ floor_normal) forward_angle = world_direction_to_render_angle( projected_forward, render_u_axis=render_u_axis, render_v_axis=render_v_axis, intrinsics=intrinsics, render_transform=render_transform, ) first = normalize_angle(base_angle_degrees) second = normalize_angle(base_angle_degrees + 90.0) if forward_angle is None: return first return first if angular_distance(first, forward_angle) <= angular_distance(second, forward_angle) else second def estimate_wall_normal_cue( normals_map: np.ndarray | None, valid_mask: np.ndarray | None, wall_mask: np.ndarray, *, floor_normal: np.ndarray, render_u_axis: np.ndarray | None, render_v_axis: np.ndarray | None, intrinsics: np.ndarray | None, render_transform: np.ndarray | None, ) -> DirectionCue | None: if normals_map is None or normals_map.shape[:2] != wall_mask.shape: return None floor_normal = normalize_vector(floor_normal) if floor_normal is None: return None lengths = np.linalg.norm(normals_map, axis=2) valid = (wall_mask > 0) & np.isfinite(normals_map).all(axis=2) & (lengths > 1e-4) if valid_mask is not None and valid_mask.shape == wall_mask.shape: valid &= valid_mask.astype(bool) normalized = np.zeros_like(normals_map, dtype=np.float64) normalized[valid] = normals_map[valid] / lengths[valid, None] vertical_alignment = np.abs(normalized @ floor_normal) valid &= vertical_alignment <= 0.42 ys, xs = np.where(valid) if len(xs) < 250: return None stride = max(1, len(xs) // 6000) ys, xs = ys[::stride], xs[::stride] normals = normalized[ys, xs] horizontal = normals - (normals @ floor_normal)[:, None] * floor_normal horizontal_lengths = np.linalg.norm(horizontal, axis=1) keep = horizontal_lengths > 0.6 horizontal = horizontal[keep] / horizontal_lengths[keep, None] ys, xs = ys[keep], xs[keep] if len(horizontal) < 200: return None angles = [] weights = [] regions = [] height, width = wall_mask.shape for direction, y, x in zip(horizontal, ys, xs, strict=False): angle = world_direction_to_render_angle( direction, render_u_axis=render_u_axis, render_v_axis=render_v_axis, intrinsics=intrinsics, render_transform=render_transform, ) if angle is None: continue angles.append(angle % 90.0) weights.append(1.0) regions.append(min(3, int(y * 4 / max(height, 1))) * 4 + min(3, int(x * 4 / max(width, 1)))) if len(angles) < 200: return None base_angle, concentration = periodic_mean( np.asarray(angles), np.asarray(weights), period=90.0, ) agreement = np.array( [angular_distance(angle, base_angle, 90.0) <= 10.0 for angle in angles], dtype=bool, ) agreeing_regions = {region for region, accepted in zip(regions, agreement, strict=False) if accepted} if int(agreement.sum()) < 160 or len(agreeing_regions) < 2: return None agreement_ratio = float(agreement.mean()) confidence = float(np.clip( 0.46 * concentration + 0.30 * agreement_ratio + 0.14 * min(1.0, len(agreeing_regions) / 6.0) + 0.10 * min(1.0, int(agreement.sum()) / 1200.0), 0.0, 1.0, )) angle = choose_depth_axis( base_angle, floor_normal=floor_normal, render_u_axis=render_u_axis, render_v_axis=render_v_axis, intrinsics=intrinsics, render_transform=render_transform, ) return DirectionCue( source="wall-normal-manhattan", angle_degrees=round(angle, 3), confidence=round(confidence, 3), support=int(agreement.sum()), region_support=len(agreeing_regions), ambiguity_degrees=90.0, details={ "baseAxisDegrees": round(base_angle, 3), "concentration": round(concentration, 4), "agreementRatio": round(agreement_ratio, 4), }, ) def estimate_architectural_line_cue( image_rgb: np.ndarray, structure_mask: np.ndarray, *, floor_normal: np.ndarray, render_u_axis: np.ndarray | None, render_v_axis: np.ndarray | None, intrinsics: np.ndarray | None, render_transform: np.ndarray | None, ) -> DirectionCue | None: if intrinsics is None: return None floor_normal = normalize_vector(floor_normal) if floor_normal is None: return None intrinsics = np.asarray(intrinsics, dtype=np.float64).reshape(3, 3) gray = cv2.cvtColor(image_rgb, cv2.COLOR_RGB2GRAY) detector = cv2.createLineSegmentDetector(cv2.LSD_REFINE_STD) detected = detector.detect(gray)[0] if detected is None: return None height, width = gray.shape minimum_length = max(30.0, min(height, width) * 0.07) vertical_vanishing = intrinsics @ floor_normal finite_vertical = abs(float(vertical_vanishing[2])) > 1e-7 vertical_xy = ( vertical_vanishing[:2] / vertical_vanishing[2] if finite_vertical else vertical_vanishing[:2] ) angles = [] weights = [] regions = [] for raw_line in detected.reshape(-1, 4): x1, y1, x2, y2 = [float(value) for value in raw_line] line_vector = np.array([x2 - x1, y2 - y1], dtype=np.float64) length = float(np.linalg.norm(line_vector)) if length < minimum_length: continue line_unit = line_vector / length sample_x = np.clip(np.rint(np.linspace(x1, x2, 11)).astype(np.int32), 0, width - 1) sample_y = np.clip(np.rint(np.linspace(y1, y2, 11)).astype(np.int32), 0, height - 1) structure_support = float(np.mean(structure_mask[sample_y, sample_x] > 0)) if structure_support < 0.58: continue midpoint = np.array([(x1 + x2) * 0.5, (y1 + y2) * 0.5], dtype=np.float64) vertical_direction = vertical_xy - midpoint if finite_vertical else vertical_xy vertical_length = float(np.linalg.norm(vertical_direction)) if vertical_length > 1e-6: vertical_direction /= vertical_length if abs(float(line_unit @ vertical_direction)) >= math.cos(math.radians(14.0)): continue p1 = np.array([x1, y1, 1.0], dtype=np.float64) p2 = np.array([x2, y2, 1.0], dtype=np.float64) image_line = np.cross(p1, p2) interpretation_plane = intrinsics.T @ image_line world_direction = np.cross(interpretation_plane, floor_normal) angle = world_direction_to_render_angle( world_direction, render_u_axis=render_u_axis, render_v_axis=render_v_axis, intrinsics=intrinsics, render_transform=render_transform, ) if angle is None: continue angles.append(angle % 90.0) weights.append(length * structure_support) regions.append( min(3, int(midpoint[1] * 4 / max(height, 1))) * 4 + min(3, int(midpoint[0] * 4 / max(width, 1))) ) base_cue = cue_from_region_angles( source="architectural-manhattan", angles=angles, weights=weights, regions=regions, period=90.0, minimum_regions=3, ) if base_cue is None: return None angle = choose_depth_axis( base_cue.angle_degrees, floor_normal=floor_normal, render_u_axis=render_u_axis, render_v_axis=render_v_axis, intrinsics=intrinsics, render_transform=render_transform, ) return DirectionCue( source=base_cue.source, angle_degrees=round(angle, 3), confidence=base_cue.confidence, support=base_cue.support, region_support=base_cue.region_support, ambiguity_degrees=90.0, details={**base_cue.details, "baseAxisDegrees": base_cue.angle_degrees}, ) def fuse_structural_cues(cues: list[DirectionCue]) -> DirectionCue | None: cues = [cue for cue in cues if cue is not None and cue.confidence >= 0.42] if not cues: return None if len(cues) == 1: return cues[0] base_angles = np.array([cue.angle_degrees % 90.0 for cue in cues]) weights = np.array([cue.confidence for cue in cues]) angle, concentration = periodic_mean(base_angles, weights, period=90.0) disagreement = max( angular_distance(float(value), angle, 90.0) for value in base_angles ) if disagreement > 14.0: strongest = max(cues, key=lambda cue: cue.confidence) return DirectionCue( source=strongest.source, angle_degrees=strongest.angle_degrees, confidence=round(strongest.confidence * 0.72, 3), support=strongest.support, region_support=strongest.region_support, ambiguity_degrees=90.0, details={**strongest.details, "structuralCueDisagreementDegrees": round(disagreement, 3)}, ) strongest = max(cues, key=lambda cue: cue.confidence) representative = strongest.angle_degrees if angular_distance(representative, angle) > angular_distance(representative + 90.0, angle): representative += 90.0 confidence = float(np.clip( np.average([cue.confidence for cue in cues], weights=weights) + concentration * 0.12, 0.0, 1.0, )) return DirectionCue( source="room-manhattan-consensus", angle_degrees=round(normalize_angle(representative), 3), confidence=round(confidence, 3), support=sum(cue.support for cue in cues), region_support=max(cue.region_support for cue in cues), ambiguity_degrees=90.0, details={ "cueSources": [cue.source for cue in cues], "baseAxisDegrees": round(angle, 3), "concentration": round(concentration, 4), "cueDisagreementDegrees": round(disagreement, 3), }, ) def select_direction( material_cue: DirectionCue | None, structural_cue: DirectionCue | None, ) -> tuple[DirectionCue | None, str]: if material_cue is not None and material_cue.confidence >= 0.82: if structural_cue is None: return material_cue, "strong-material-direction" axis_difference = min( angular_distance(material_cue.angle_degrees, structural_cue.angle_degrees), angular_distance(material_cue.angle_degrees, structural_cue.angle_degrees + 90.0), ) if axis_difference <= 14.0: boosted = DirectionCue( source="material-room-consensus", angle_degrees=material_cue.angle_degrees, confidence=round(min(1.0, material_cue.confidence * 0.78 + structural_cue.confidence * 0.30), 3), support=material_cue.support + structural_cue.support, region_support=max(material_cue.region_support, structural_cue.region_support), ambiguity_degrees=None, details={ "materialSource": material_cue.source, "structuralSource": structural_cue.source, "axisDifferenceDegrees": round(axis_difference, 3), }, ) return boosted, "material-and-room-agree" return material_cue, "strong-diagonal-material-direction" if structural_cue is not None and structural_cue.confidence >= 0.52: return structural_cue, "room-structure-direction" if material_cue is not None and material_cue.confidence >= 0.62: return material_cue, "moderate-material-direction" return None, "insufficient-observable-direction-evidence" def direction_label(angle_degrees: float) -> str: angle = normalize_angle(angle_degrees) if angle < 12.0 or angle >= 168.0: return "across_render_u" if 78.0 <= angle < 102.0: return "along_render_v" return "diagonal" def direction_axis_for_overlay( *, angle_degrees: float, image_shape: tuple[int, int], surface_mask: np.ndarray, surface_uv: np.ndarray | None, surface_indices: np.ndarray | None, surface_plane_ids: np.ndarray | None, render_transform: np.ndarray | None, ) -> dict | None: height, width = image_shape direction = np.array([ math.cos(math.radians(angle_degrees)), math.sin(math.radians(angle_degrees)), ]) if surface_uv is not None and surface_indices is not None and len(surface_uv) == len(surface_indices): valid = np.isfinite(surface_uv).all(axis=1) if surface_plane_ids is not None and len(surface_plane_ids) == len(surface_uv): plane_values = surface_plane_ids[valid & (surface_plane_ids != 255)] if plane_values.size: dominant_plane = int(np.bincount(plane_values.astype(np.int64)).argmax()) valid &= surface_plane_ids == dominant_plane uv = surface_uv[valid] indices = np.asarray(surface_indices)[valid] if len(uv) >= 100: center = np.median(uv, axis=0) span = np.percentile(uv, 95, axis=0) - np.percentile(uv, 5, axis=0) radius = max(float(np.min(span)) * 0.28, 1e-3) endpoints = [center - direction * radius, center + direction * radius] image_points = [] for endpoint in endpoints: nearest = int(np.argmin(np.sum((uv - endpoint) ** 2, axis=1))) pixel_index = int(indices[nearest]) image_points.append((float(pixel_index % width), float(pixel_index // width))) return { "startX": image_points[0][0], "startY": image_points[0][1], "endX": image_points[1][0], "endY": image_points[1][1], "source": "surface-uv", } if render_transform is None: return None ys, xs = np.where(surface_mask > 0) if len(xs) < 50: return None sample_points = np.column_stack((xs, ys)).astype(np.float32).reshape(1, -1, 2) mapped = cv2.perspectiveTransform(sample_points, render_transform.astype(np.float32))[0] finite = np.isfinite(mapped).all(axis=1) mapped = mapped[finite] if len(mapped) < 50: return None center = np.median(mapped, axis=0) span = np.percentile(mapped, 95, axis=0) - np.percentile(mapped, 5, axis=0) radius = max(float(np.min(span)) * 0.28, 1e-3) floor_points = np.array( [[center - direction * radius, center + direction * radius]], dtype=np.float32, ) inverse = np.linalg.inv(render_transform).astype(np.float32) image_points = cv2.perspectiveTransform(floor_points, inverse)[0] if not np.isfinite(image_points).all(): return None return { "startX": float(image_points[0, 0]), "startY": float(image_points[0, 1]), "endX": float(image_points[1, 0]), "endY": float(image_points[1, 1]), "source": "floor-transform", } def find_floor_direction( image_rgb: np.ndarray, surface_mask: np.ndarray, *, wall_mask: np.ndarray, structure_mask: np.ndarray, normals_map: np.ndarray | None, geometry_valid_mask: np.ndarray | None, intrinsics: np.ndarray | list[float] | None, floor_normal: np.ndarray | list[float] | None, render_u_axis: np.ndarray | list[float] | None, render_v_axis: np.ndarray | list[float] | None, render_transform: np.ndarray | list[float] | None, surface_uv: np.ndarray | None, surface_indices: np.ndarray | None, surface_plane_ids: np.ndarray | None, ) -> dict: height, width = surface_mask.shape[:2] transform = normalize_transform(render_transform) camera_matrix = None if intrinsics is not None: raw_intrinsics = np.asarray(intrinsics, dtype=np.float64) if raw_intrinsics.size == 9 and np.isfinite(raw_intrinsics).all(): camera_matrix = raw_intrinsics.reshape(3, 3) normal = normalize_vector(np.asarray(floor_normal)) if floor_normal is not None else None u_axis = normalize_vector(np.asarray(render_u_axis)) if render_u_axis is not None else None v_axis = normalize_vector(np.asarray(render_v_axis)) if render_v_axis is not None else None rectified = rectify_floor_image( image_rgb, surface_mask, render_transform=transform, surface_uv=surface_uv, surface_indices=surface_indices, surface_plane_ids=surface_plane_ids, ) material_cue = estimate_material_cue(rectified[0], rectified[1]) if rectified is not None else None wall_cue = None architecture_cue = None if normal is not None: wall_cue = estimate_wall_normal_cue( normals_map, geometry_valid_mask, wall_mask, floor_normal=normal, render_u_axis=u_axis, render_v_axis=v_axis, intrinsics=camera_matrix, render_transform=transform, ) architecture_cue = estimate_architectural_line_cue( image_rgb, structure_mask, floor_normal=normal, render_u_axis=u_axis, render_v_axis=v_axis, intrinsics=camera_matrix, render_transform=transform, ) structural_cue = fuse_structural_cues([ cue for cue in (wall_cue, architecture_cue) if cue is not None ]) selected, selection_reason = select_direction(material_cue, structural_cue) warnings = [] needs_user_direction = selected is None if selected is None: selected = DirectionCue( source="canonical-render-axis", angle_degrees=0.0, confidence=0.0, support=0, region_support=0, ambiguity_degrees=90.0, details={"reason": "direction-is-not-observable-in-this-image"}, ) warnings.append("direction_not_observable_user_rotation_recommended") if selected.ambiguity_degrees is not None: warnings.append("direction_has_90_degree_axis_ambiguity") axis = direction_axis_for_overlay( angle_degrees=selected.angle_degrees, image_shape=(height, width), surface_mask=surface_mask, surface_uv=surface_uv, surface_indices=surface_indices, surface_plane_ids=surface_plane_ids, render_transform=transform, ) floor_area_ratio = float(np.count_nonzero(surface_mask)) / max(int(surface_mask.size), 1) return { "source": "backend-direction-finder-v2", "directionMethod": selected.source, "selectionReason": selection_reason, "angleDegrees": selected.angle_degrees, "renderAngleDegrees": selected.angle_degrees, "secondaryAngleDegrees": normalize_angle(selected.angle_degrees + 90.0), "directionLabel": direction_label(selected.angle_degrees), "confidence": selected.confidence, "floorAreaRatio": round(floor_area_ratio, 4), "lineCount": 0, "dominantLineCount": 0, "gridPatternDetected": bool(material_cue and material_cue.ambiguity_degrees == 90.0), "axisAmbiguous90": selected.ambiguity_degrees == 90.0, "needsUserDirection": needs_user_direction, "warnings": warnings, "dominantLines": [], "directionAxis": axis, "rectification": rectified[2] if rectified is not None else None, "cues": { "material": material_cue.metadata() if material_cue is not None else None, "wallNormals": wall_cue.metadata() if wall_cue is not None else None, "architecture": architecture_cue.metadata() if architecture_cue is not None else None, "structural": structural_cue.metadata() if structural_cue is not None else None, }, }