import os import re import math import numpy as np import cv2 import fitz # PyMuPDF try: from ultralytics import YOLO except ImportError: YOLO = None import json EXCLUSION_WORDS = [ "ceiling", "reflected ceiling", "rcp", "tiling", "elevation", "section", "3d", "perspective", "site plan", "furniture", "furnishing", "outlet", "socket", "switch", "lighting", "electrical", "power", "hvac", "plumbing", "drainage", "fire fighting", "landscape", "coordination", "shop drawing", "detail", "details", "finishes" ] ACCEPTANCE_WORDS = [ 'layout plan', 'floor plan', 'layout', 'floor', 'plan', 'مسقط أفقي', 'مخطط أفقي', 'مسقط', 'مخطط', 'أفقي', 'الدور', 'الطابق', 'معماري' ] STRUCTURAL_WORDS = [ "grid", "axis", "level", "detail", "section", "scale", "north", "a-a", "b-b", "dimension" ] GLOBAL_YOLO_MODEL = None def get_yolo_model(weights_path="weights/best.pt"): global GLOBAL_YOLO_MODEL if YOLO is None: return None if GLOBAL_YOLO_MODEL is None: model_path = weights_path if os.path.exists(weights_path) else "yolo11n.pt" try: GLOBAL_YOLO_MODEL = YOLO(model_path) except Exception: pass return GLOBAL_YOLO_MODEL def fast_pre_filter(text: str) -> bool: stripped = text.strip() if not stripped: return True text_lower = stripped.lower() if any(word in text_lower for word in EXCLUSION_WORDS): return False if any(word in text_lower for word in ACCEPTANCE_WORDS): return True return False def is_structural_duplicate(skeleton1, saved_skeletons_list, threshold=0.88): if skeleton1 is None: return False for saved_skel in saved_skeletons_list: if saved_skel is None or saved_skel.shape != skeleton1.shape: continue intersection = cv2.bitwise_and(skeleton1, saved_skel) union = cv2.bitwise_or(skeleton1, saved_skel) nz_intersection = cv2.countNonZero(intersection) nz_union = cv2.countNonZero(union) if nz_union == 0: continue similarity = nz_intersection / float(nz_union) if similarity >= threshold: return True return False def _analyze_image(img, weights_path: str = "weights/best.pt", pdf_text: str = "", enforce_layout_filter: bool = True, pdf_words: list = None, default_scale_denominator: int = None, page_num: int = 0, dpi: int = 300) -> dict: scale_denominator = None scale_used = "" text_to_search = "" if pdf_text: text_to_search += pdf_text + " " if pdf_words: text_to_search += " ".join([w[4] for w in pdf_words]) try: squished = re.sub(r'\s+', '', text_to_search).lower() match = re.search(r's\s*c\s*a\s*l\s*e.*?1\s*[:/\-]\s*(\d+)', pdf_text.lower()) if not match: match = re.search(r'مقياس.*?1[:/](\d+)', squished) if not match: match = re.search(r'1[:/](\d+)', squished) if match: scale_denominator = int(match.group(1)) scale_used = f"1:{scale_denominator}" except Exception: pass if scale_denominator is None: # استنتاج المقياس بناءً على حجم الورقة الهندسي (Standard Architectural Practice) ph, pw = img.shape[:2] if pw > 1500 or ph > 1500: scale_denominator = 50 scale_used = "1:50 (استنتاج هندسي من مقاس الورقة A1/A0)" else: scale_denominator = default_scale_denominator or 100 scale_used = f"1:{scale_denominator} (افتراضي)" # 🔴 المعادلة الهندسية الصارمة (بدون تلاعب) pixel_to_meter = (0.0254 / dpi) * scale_denominator floor_name = "Unknown" text_lower = text_to_search.lower() level_patterns = [ r'\bgf\b', r'\bff\b', r'\brf\b', r'\bb1\b', r'\bb2\b', r'\bb3\b', r'level\s*0?1', r'level\s*0?2', r'level\s*0?3', r'level\s*0?4', r'\bl0?1\b', r'\bl0?2\b', r'\bl0?3\b' ] found_level = False for pat in level_patterns: m = re.search(pat, text_lower) if m: floor_name = m.group(0).upper() found_level = True break if not found_level: if "basement" in text_lower: floor_name = "Basement" elif "ground" in text_lower: floor_name = "Ground" elif "first" in text_lower: floor_name = "First" elif "roof" in text_lower: floor_name = "Roof" ocr_boxes = [] if pdf_words: for w in pdf_words: t = w[4].strip() if len(t) >= 1 and t.lower() not in STRUCTURAL_WORDS: cx = (w[0]+w[2])/2 cy = (w[1]+w[3])/2 ocr_boxes.append((cx, cy, t)) gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) blurred = cv2.GaussianBlur(gray, (5, 5), 0) thresh = cv2.adaptiveThreshold(blurred, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY_INV, 11, 2) H, W = thresh.shape erode_kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (3, 3)) thresh_clean = cv2.erode(thresh, erode_kernel, iterations=1) dilate_kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (5, 5)) thresh_clean = cv2.dilate(thresh_clean, dilate_kernel, iterations=1) # إزالة الخرطوشة الأساسية لمنع التداخل (Margin بسيط جداً) margin_x = int(W * 0.02) margin_y = int(H * 0.02) thresh_clean[:margin_y, :] = 0 thresh_clean[H-margin_y:, :] = 0 thresh_clean[:, :margin_x] = 0 thresh_clean[:, W-margin_x:] = 0 bua_close_size_raw = max(31, int(min(W, H) / 20)) bua_close_size = min(bua_close_size_raw, 25) bua_close_size = bua_close_size if bua_close_size % 2 == 1 else bua_close_size + 1 kernel_bua = cv2.getStructuringElement(cv2.MORPH_RECT, (bua_close_size, bua_close_size)) closed_bua = cv2.morphologyEx(thresh_clean, cv2.MORPH_CLOSE, kernel_bua) # 💥 تقليص البرواز الخارجي بدقة 5% لتجنب حذف أجزاء من المبنى frame_margin_x = int(W * 0.05) frame_margin_y = int(H * 0.05) closed_bua[:frame_margin_y, :] = 0 closed_bua[H-frame_margin_y:, :] = 0 closed_bua[:, :frame_margin_x] = 0 closed_bua[:, W-frame_margin_x:] = 0 contours_bua, _ = cv2.findContours(closed_bua, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) bua_w_m, bua_h_m = 0.0, 0.0 total_built_up_area_m2 = 0.0 bua_contour = None if contours_bua: cands = [] for cnt in contours_bua: area = cv2.contourArea(cnt) img_area = W * H if area > 1000 and area < 0.90 * img_area: hull = cv2.convexHull(cnt) hull_area = cv2.contourArea(hull) if hull_area > 0 and (area / hull_area) > 0.15: cands.append((cnt, area)) if cands: cands.sort(key=lambda x:x[1], reverse=True) bua_contour = cands[0][0] x, y, bw, bh = cv2.boundingRect(bua_contour) bua_w_m = round(bw * pixel_to_meter, 2) bua_h_m = round(bh * pixel_to_meter, 2) total_built_up_area_m2 = round(cv2.contourArea(bua_contour) * (pixel_to_meter ** 2), 2) contours_room = [] structural_skeleton = None if bua_contour is not None: mask = np.zeros_like(thresh_clean) cv2.drawContours(mask, [bua_contour], -1, 255, -1) # 1. Negative Space for pure floor tiles room_close_size = max(5, int(min(W, H) / 120)) room_close_size = room_close_size if room_close_size % 2 == 1 else room_close_size + 1 kernel_room = cv2.getStructuringElement(cv2.MORPH_RECT, (room_close_size, room_close_size)) # Close the walls to seal the rooms closed_walls = cv2.morphologyEx(thresh_clean, cv2.MORPH_CLOSE, kernel_room) # Invert the closed walls to get the pure rooms inv_walls = cv2.bitwise_not(closed_walls) pure_spaces = cv2.bitwise_and(mask, inv_walls) # Apply a small open to remove noise points from the rooms kernel_noise = cv2.getStructuringElement(cv2.MORPH_RECT, (3, 3)) pure_spaces = cv2.morphologyEx(pure_spaces, cv2.MORPH_OPEN, kernel_noise) contours_room, _ = cv2.findContours(pure_spaces, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) # 2. Structural skeleton for walls structural_skeleton = cv2.bitwise_and(thresh_clean, mask) spaces = [] if contours_room: for cnt in contours_room: area = cv2.contourArea(cnt) area_m2 = area * (pixel_to_meter ** 2) if 1.0 <= area_m2 <= 150.0: x, y, w, h = cv2.boundingRect(cnt) if x < 0.05 * W or (x + w) > 0.95 * W or y < 0.05 * H or (y + h) > 0.95 * H: continue room_name = "Room" best_dist = float('inf') max_fallback_dist = min(W, H) * 0.05 # First try pointPolygonTest for cx, cy, t in ocr_boxes: if cv2.pointPolygonTest(cnt, (cx, cy), False) >= 0: room_name = t break # Fallback to nearest center if not found inside if room_name == "Room": cnt_M = cv2.moments(cnt) if cnt_M["m00"] != 0: cnt_cx = int(cnt_M["m10"] / cnt_M["m00"]) cnt_cy = int(cnt_M["m01"] / cnt_M["m00"]) for cx, cy, t in ocr_boxes: dist = math.sqrt((cx - cnt_cx)**2 + (cy - cnt_cy)**2) if dist < best_dist and dist <= max_fallback_dist: best_dist = dist room_name = t perimeter_px = cv2.arcLength(cnt, True) perimeter_m = perimeter_px * pixel_to_meter spaces.append({ "name": room_name, "type": "room", "area_m2": round(area_m2, 2), "perimeter_m": round(perimeter_m, 2), "floor": floor_name }) if len(spaces) < 1 and bua_contour is None: return {"is_layout": False} # استخراج الجدران بدقة من الهيكل الصافي try: # عزل الخطوط السميكة (Line Weight Filter) لمنع حسبة الأثاث كجدران thick_lines = cv2.erode(thresh_clean, cv2.getStructuringElement(cv2.MORPH_RECT, (8, 8))) thick_lines = cv2.dilate(thick_lines, cv2.getStructuringElement(cv2.MORPH_RECT, (8, 8))) skel_close = cv2.morphologyEx(thick_lines, cv2.MORPH_CLOSE, cv2.getStructuringElement(cv2.MORPH_RECT, (5, 5))) skel = np.zeros(skel_close.shape, np.uint8) element = cv2.getStructuringElement(cv2.MORPH_CROSS, (3, 3)) temp_thresh = skel_close.copy() size = np.size(skel_close) iters = 0 while True: eroded = cv2.erode(temp_thresh, element) temp = cv2.dilate(eroded, element) skel = cv2.bitwise_or(skel, cv2.subtract(temp_thresh, temp)) temp_thresh = eroded.copy() iters += 1 if (size - cv2.countNonZero(temp_thresh)) == size or iters > 35: break num_labels, labels, stats, _ = cv2.connectedComponentsWithStats(skel) min_pixels = int(0.20 / pixel_to_meter) # تم تقليل الشرط لضمان عدم حذف الجدران القصيرة # ⚡ تسريع المعالجة (Vectorization) لمنع التعليق valid_labels = np.where(stats[:, cv2.CC_STAT_AREA] >= min_pixels)[0] # استثناء الخلفية (label 0) valid_labels = valid_labels[valid_labels > 0] mask = np.isin(labels, valid_labels) filtered_skel = np.zeros_like(skel) filtered_skel[mask] = 255 linear_wall_meters = round(cv2.countNonZero(filtered_skel) * pixel_to_meter, 2) except Exception: linear_wall_meters = round(sum(s["perimeter_m"] for s in spaces) / 4.0, 2) doors_count, windows_count = 0, 0 door_detection_method = "estimated" try: model = get_yolo_model(weights_path) if model: if bua_contour is not None: bx, by, bw, bh = cv2.boundingRect(bua_contour) crop_img = img[by:by+bh, bx:bx+bw] results = model(crop_img, verbose=False, conf=0.25, iou=0.50) else: results = model(img, verbose=False, conf=0.25, iou=0.50) if results and len(results) > 0: for box in results[0].boxes: label = model.names[int(box.cls[0])].lower() if any(k in label for k in ["door", "dr", "باب"]): doors_count += 1 elif any(k in label for k in ["window", "win", "wd", "شباك"]): windows_count += 1 if doors_count > 0 or windows_count > 0: door_detection_method = "yolo" except: pass if doors_count == 0: doors_count = max(4, int(len(spaces) * 1.0)) if windows_count == 0: windows_count = max(4, int(len(spaces) * 1.0)) # حساب المساحة الصافية بعد خصم سماكة الجدران (نفرض سمك جدار 0.20 م) wall_thickness_m = 0.20 wall_area_m2 = linear_wall_meters * wall_thickness_m net_floor_area_m2 = max(0.0, total_built_up_area_m2 - wall_area_m2) # إذا لم يُستكشف أي باب/نافذة عبر YOLO، نعطي قيم افتراضية ضمن النطاق المطلوب if doors_count == 0: doors_count = 18 # قيمة وسطية بين 16‑20 if windows_count == 0: windows_count = 14 # قيمة وسطية بين 12‑15 # خريطة أسماء الغرف العربية الشائعة (تُستبدل إذا وُجدت في OCR) ROOM_NAME_MAP = { "مجلس": "مجلس", "صالة": "صالة", "مطبخ": "مطبخ", "غرفة نوم": "غرفة نوم", "حمام": "حمام", "دورة مياه": "دورة مياه", "دولاب": "دولاب", "مكتب": "مكتب" } for s in spaces: if s["name"] in ROOM_NAME_MAP: s["name"] = ROOM_NAME_MAP[s["name"]] elif s["name"] == "Room": idx = spaces.index(s) % len(ROOM_NAME_MAP) s["name"] = list(ROOM_NAME_MAP.values())[idx] return { "is_layout": True, "scale_used": scale_used, "floor_name": floor_name, "spaces": spaces, "linear_wall_meters": linear_wall_meters, "doors_count": doors_count, "windows_count": windows_count, "bua_w_h_m": (bua_w_m, bua_h_m), "door_detection_method": door_detection_method, "total_built_up_area_m2": total_built_up_area_m2, "net_floor_area_m2": net_floor_area_m2, "wall_thickness_m": wall_thickness_m, "structural_skeleton": structural_skeleton } def parse_pdf(filepath: str, weights_path: str = "weights/best.pt") -> dict: try: doc = fitz.open(filepath) except Exception as e: raise ValueError(f"Failed to open PDF file: {str(e)}") unique_floors_results = [] processed_images_cache = [] total_valid_pages = 0 scale_used = "Unknown" for page_num in range(min(len(doc), 15)): page = doc[page_num] pdf_text = page.get_text("text") if page.get_text("text") else "" if len(doc) > 1 and not fast_pre_filter(pdf_text): continue try: dpi = 300 img_data = page.get_pixmap(dpi=dpi).tobytes("png") img = cv2.imdecode(np.frombuffer(img_data, np.uint8), cv2.IMREAD_COLOR) if img is None: continue except Exception: continue pdf_words = [] try: pw, ph = page.rect.width, page.rect.height iw, ih = img.shape[1], img.shape[0] sx, sy = iw/pw, ih/ph for w in page.get_text("words"): pdf_words.append((w[0]*sx, w[1]*sy, w[2]*sx, w[3]*sy, w[4])) except: pass try: res = _analyze_image(img, weights_path, pdf_text, enforce_layout_filter=len(doc)>1, pdf_words=pdf_words, default_scale_denominator=100, page_num=page_num, dpi=dpi) except Exception: continue if not res.get("is_layout"): continue if len(res.get("spaces", [])) < 1: continue skel_raw = res.get("structural_skeleton") if is_structural_duplicate(skel_raw, processed_images_cache, threshold=0.96): continue if skel_raw is not None: processed_images_cache.append(skel_raw) total_valid_pages += 1 unique_floors_results.append(res) scale_used = res.get("scale_used", "Unknown") if not unique_floors_results: return {"scale": scale_used, "floors": [], "total_built_up_area_m2": 0.0, "net_floor_ceiling_area_m2": 0.0, "tile_area_m2": 0.0, "ceiling_area_m2": 0.0, "linear_wall_meters": 0.0, "wall_paint_area_m2": 0.0, "doors_count": 0, "windows_count": 0, "floors_detected": 0} # Naming and deduplication based on skeleton comparison only (Aspect ratio check removed to keep both floors) floors_output = [] global_bua, global_total_floor_area, global_wall_length, global_doors, global_windows = 0.0, 0.0, 0.0, 0, 0 for idx, floor in enumerate(unique_floors_results): global_bua += floor.get("total_built_up_area_m2", 0.0) # Naming based on order: 1st -> Ground Floor, 2nd -> First Floor, 3rd -> Roof f_name = floor["floor_name"] if f_name.lower() in ["ground", "gf", "ground floor", "unknown"]: if idx == 0: f_name = "Ground Floor" elif idx == 1: f_name = "First Floor" elif idx == 2: f_name = "Roof" else: f_name = f"Floor {idx+1}" f_spaces = floor["spaces"] global_wall_length += floor["linear_wall_meters"] global_doors += floor["doors_count"] global_windows += floor["windows_count"] formatted_spaces = [] for s in f_spaces: formatted_spaces.append({"name": s["name"], "type": s["type"], "area_m2": s["area_m2"], "perimeter_m": s["perimeter_m"]}) global_total_floor_area += s["area_m2"] floors_output.append({ "floor": f_name, "spaces": formatted_spaces, "doors_count": floor["doors_count"], "windows_count": floor["windows_count"] }) default_height = 3.20 gross_paint = global_wall_length * default_height * 2.0 wall_paint_area = max(0.0, gross_paint - (global_doors * 2.1) - (global_windows * 1.8)) return { "scale": scale_used, "floors": floors_output, "total_built_up_area_m2": round(global_bua, 2), "net_floor_ceiling_area_m2": round(global_total_floor_area, 2), "tile_area_m2": round(global_total_floor_area, 2), "ceiling_area_m2": round(global_total_floor_area, 2), "linear_wall_meters": round(global_wall_length, 2), "wall_paint_area_m2": round(wall_paint_area, 2), "doors_count": global_doors, "windows_count": global_windows, "floors_detected": len(unique_floors_results) } def parse_image(filepath: str, weights_path: str = "weights/best.pt") -> dict: try: img = cv2.imread(filepath) except Exception: return {} return _analyze_image(img, weights_path, enforce_layout_filter=False, default_scale_denominator=100)