from celery_app import app from config import settings from utils.geometry import angle_difference, calculate_reverse_yaw from sqlalchemy import create_engine, text from sqlalchemy.orm import Session @app.task(bind=True) def run_validation(self, house_id: str): """Post-analysis validation checks.""" engine = create_engine(settings.database_url) issues = [] try: with Session(engine) as session: # Load all panoramas for this house pano_result = session.execute( text( "SELECT id FROM panoramas WHERE house_id = :house_id" ), {"house_id": house_id}, ) pano_ids = [row[0] for row in pano_result.fetchall()] # Load all approved connections/hotspots hotspot_result = session.execute( text( "SELECT id, data FROM ai_suggestions " "WHERE house_id = :house_id " "AND suggestion_type IN ('connection', 'hotspot') " "AND status = 'approved'" ), {"house_id": house_id}, ) suggestions = hotspot_result.fetchall() # Parse suggestions into structured connections connections = [] for sug_id, data_str in suggestions: try: conn = eval(data_str) connections.append(conn) except Exception: continue # --- Check 1: Orphan panoramas (no hotspot connections) --- connected_ids = set() for conn in connections: connected_ids.add(conn.get("from_pano_id")) connected_ids.add(conn.get("to_pano_id")) for pano_id in pano_ids: if pano_id not in connected_ids: issues.append( { "type": "orphan_pano", "severity": "error", "description": ( f"Panorama {pano_id} has no hotspot connections" ), "pano_id": pano_id, "hotspot_id": None, } ) # --- Check 2: One-way connections --- connection_pairs = set() for conn in connections: pair = ( conn.get("from_pano_id"), conn.get("to_pano_id"), ) connection_pairs.add(pair) for from_id, to_id in connection_pairs: reverse_pair = (to_id, from_id) if reverse_pair not in connection_pairs: issues.append( { "type": "one_way_connection", "severity": "warning", "description": ( f"One-way connection: {from_id} -> {to_id}, " f"missing reverse" ), "pano_id": from_id, "hotspot_id": None, } ) # --- Check 3: Reverse yaw mismatch --- for conn in connections: from_id = conn.get("from_pano_id") to_id = conn.get("to_pano_id") yaw = conn.get("yaw") if yaw is None: continue # Find reverse connection for rev_conn in connections: rev_from = rev_conn.get("from_pano_id") rev_to = rev_conn.get("to_pano_id") rev_yaw = rev_conn.get("yaw") if rev_from == to_id and rev_to == from_id and rev_yaw is not None: expected_reverse = calculate_reverse_yaw(yaw) diff = angle_difference(rev_yaw, expected_reverse) if diff > settings.yaw_tolerance: issues.append( { "type": "reverse_yaw_mismatch", "severity": "warning", "description": ( f"Reverse yaw mismatch for {from_id} -> {to_id}: " f"expected {expected_reverse:.1f}, " f"got {rev_yaw:.1f} (diff {diff:.1f})" ), "pano_id": from_id, "hotspot_id": None, } ) break # --- Check 4: Hotspot overlap --- # Group hotspots by source pano from collections import defaultdict hotspots_by_pano = defaultdict(list) for conn in connections: from_id = conn.get("from_pano_id") yaw = conn.get("yaw") if yaw is not None: hotspots_by_pano[from_id].append( {"yaw": yaw, "conn": conn} ) for pano_id, hotspots_list in hotspots_by_pano.items(): for i in range(len(hotspots_list)): for j in range(i + 1, len(hotspots_list)): yaw1 = hotspots_list[i]["yaw"] yaw2 = hotspots_list[j]["yaw"] diff = angle_difference(yaw1, yaw2) if diff < 5.0: # Within 5 degrees issues.append( { "type": "hotspot_overlap", "severity": "warning", "description": ( f"Hotspot overlap on pano {pano_id}: " f"yaws {yaw1:.1f} and {yaw2:.1f} " f"within {diff:.1f} degrees" ), "pano_id": pano_id, "hotspot_id": None, } ) # --- Check 5: Low confidence --- for conn in connections: confidence = conn.get("confidence", 0.0) if confidence < settings.confidence_threshold: issues.append( { "type": "low_confidence", "severity": "info", "description": ( f"Low confidence connection " f"{conn.get('from_pano_id')} -> " f"{conn.get('to_pano_id')}: " f"{confidence:.2f}" ), "pano_id": conn.get("from_pano_id"), "hotspot_id": None, } ) # --- Check 6: Unnamed rooms --- room_result = session.execute( text( "SELECT id, name FROM rooms WHERE house_id = :house_id" ), {"house_id": house_id}, ) for room_id, name in room_result.fetchall(): if not name or name.strip() == "": issues.append( { "type": "unnamed_room", "severity": "warning", "description": f"Room {room_id} has no name", "pano_id": None, "hotspot_id": None, } ) # --- Check 7: Rooms with no entry hotspot --- room_pano_result = session.execute( text( "SELECT room_id, pano_id FROM panorama_rooms pr " "JOIN panoramas p ON p.id = pr.panorama_id " "WHERE p.house_id = :house_id" ), {"house_id": house_id}, ) room_panos = defaultdict(list) for room_id, pano_id in room_pano_result.fetchall(): room_panos[room_id].append(pano_id) for room_id, pano_list in room_panos.items(): has_entry = False for conn in connections: if conn.get("to_pano_id") in pano_list: has_entry = True break if not has_entry: issues.append( { "type": "no_entry_hotspot", "severity": "warning", "description": ( f"Room {room_id} has no entry hotspot" ), "pano_id": None, "hotspot_id": None, } ) # Save validation results session.execute( text( "DELETE FROM ai_suggestions WHERE house_id = :house_id " "AND suggestion_type = 'validation'" ), {"house_id": house_id}, ) session.execute( text( "INSERT INTO ai_suggestions (house_id, suggestion_type, " "data, confidence, status) " "VALUES (:house_id, 'validation', :data, 1.0, 'completed')" ), { "house_id": house_id, "data": str(issues), }, ) session.commit() except Exception as e: return {"success": False, "issues": [], "error": str(e)} return { "success": True, "issues": issues, }