pcb-defect-detector-project/new-version
0
1# app.py - نسخة متوافقة مع DatasetStorage2from huggingface_hub import HfApi, upload_file3import os4import uuid5import json6import base647from datetime import datetime8from fastapi import FastAPI, File, UploadFile, HTTPException, Depends, Form9from fastapi.responses import HTMLResponse, Response10from fastapi.staticfiles import StaticFiles11from fastapi.middleware.cors import CORSMiddleware12from sqlalchemy.orm import Session13from ultralytics import YOLO14from typing import Optional, List15 16# استخدام نظام التخزين الجديد17from database import User, Board, get_db18from auth import get_current_user, authenticate_user, create_access_token, get_password_hash19from utils.image_utils import process_image20 21# إضافات PDF22from reportlab.lib.pagesizes import A423from reportlab.lib import colors24from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle25from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle26from reportlab.lib.enums import TA_CENTER, TA_RIGHT27from reportlab.pdfbase import pdfmetrics28from reportlab.pdfbase.ttfonts import TTFont29import io30 31# إعدادات32os.environ["HF_TOKEN"] = os.environ.get("HF_TOKEN", "")33 34# تحميل النموذج35MODEL_PATH = "last.pt"36model = None37if os.path.exists(MODEL_PATH):38 model = YOLO(MODEL_PATH)39 print("✅ تم تحميل النموذج بنجاح")40else:41 print("⚠️ النموذج غير موجود")42 43app = FastAPI(title="PCB Detector API")44 45# CORS46app.add_middleware(47 CORSMiddleware,48 allow_origins=["*"],49 allow_credentials=True,50 allow_methods=["*"],51 allow_headers=["*"],52)53 54# الملفات الثابتة55os.makedirs("static/uploads", exist_ok=True)56os.makedirs("static/annotated", exist_ok=True)57app.mount("/static", StaticFiles(directory="static"), name="static")58 59# ============================================60# إنشاء المستخدم الافتراضي عند بدء التشغيل61# ============================================62 63@app.on_event("startup")64async def create_default_user():65 """إنشاء مستخدم افتراضي إذا لم يكن موجوداً"""66 try:67 existing_user = User.get_by_username("admin")68 if not existing_user:69 hashed_password = get_password_hash("admin123")70 new_user = User.create("admin", "admin@pcb.com", hashed_password)71 if new_user:72 print("✅ تم إنشاء المستخدم الافتراضي: admin / admin123")73 else:74 print("⚠️ فشل في إنشاء المستخدم الافتراضي")75 else:76 print("✅ المستخدم الافتراضي موجود بالفعل")77 except Exception as e:78 print(f"⚠️ خطأ في إنشاء المستخدم الافتراضي: {e}")79 80# ============================================81# API Endpoints82# ============================================83 84@app.get("/health")85async def health_check():86 return {"status": "healthy", "message": "PCB Detector API is running"}87 88# -------------------- المستخدمين --------------------89 90@app.post("/api/register")91async def register(user_data: dict):92 username = user_data.get("username")93 email = user_data.get("email")94 password = user_data.get("password")95 96 if not username or not email or not password:97 raise HTTPException(400, "جميع الحقول مطلوبة")98 99 if len(password) > 72:100 password = password[:72]101 102 # التحقق من وجود المستخدم103 existing = User.get_by_username(username)104 if existing:105 raise HTTPException(400, "اسم المستخدم موجود بالفعل")106 107 hashed_password = get_password_hash(password)108 new_user = User.create(username, email, hashed_password)109 110 if not new_user:111 raise HTTPException(500, "فشل في إنشاء المستخدم")112 113 token = create_access_token({"sub": new_user.id})114 return {115 "access_token": token,116 "token_type": "bearer",117 "user": {118 "id": new_user.id,119 "username": new_user.username,120 "email": new_user.email,121 "created_at": new_user.created_at122 }123 }124 125@app.post("/api/login")126async def login(user_data: dict):127 username = user_data.get("username")128 password = user_data.get("password")129 130 if not username or not password:131 raise HTTPException(400, "اسم المستخدم وكلمة المرور مطلوبة")132 133 if len(password) > 72:134 password = password[:72]135 136 # تحقق من بيانات المستخدم137 user = authenticate_user(username, password) # تحتاج لتعديل دالة authenticate_user138 if not user:139 raise HTTPException(401, "اسم المستخدم أو كلمة المرور غير صحيحة")140 141 token = create_access_token({"sub": user.id})142 return {143 "access_token": token,144 "token_type": "bearer",145 "user": {146 "id": user.id,147 "username": user.username,148 "email": user.email,149 "created_at": user.created_at150 }151 }152 153@app.get("/api/me")154async def get_me(current_user: User = Depends(get_current_user)):155 return {156 "id": current_user.id,157 "username": current_user.username,158 "email": current_user.email,159 "created_at": current_user.created_at160 }161 162# -------------------- اللوحات --------------------163 164@app.post("/api/boards")165async def create_board(166 title: str = Form(...),167 description: str = Form(None),168 current_user: User = Depends(get_current_user)169):170 new_board = Board.create(current_user.id, title, description or "")171 if not new_board:172 raise HTTPException(500, "فشل في حفظ اللوحة")173 174 return {175 "id": new_board.id,176 "title": new_board.title,177 "description": new_board.description,178 "image_path": new_board.image_path,179 "annotated_image_path": new_board.annotated_image_path,180 "defects_data": new_board.defects_data,181 "report_data": new_board.report_data,182 "created_at": new_board.created_at,183 "updated_at": new_board.updated_at184 }185 186@app.get("/api/boards")187async def get_boards(current_user: User = Depends(get_current_user)):188 boards = Board.get_all_by_user(current_user.id)189 result = []190 for b in boards:191 result.append({192 "id": b.id,193 "title": b.title,194 "description": b.description,195 "image_path": b.image_path,196 "annotated_image_path": b.annotated_image_path,197 "defects_data": b.defects_data,198 "report_data": b.report_data,199 "created_at": b.created_at,200 "updated_at": b.updated_at201 })202 return result203 204 205@app.put("/api/boards/{board_id}")206async def update_board(207 board_id: int,208 defects_data: str = Form(None),209 report_data: str = Form(None),210 current_user: User = Depends(get_current_user)211):212 """تحديث بيانات اللوحة (النتائج)"""213 board = Board.get_by_id(current_user.id, board_id)214 if not board:215 raise HTTPException(404, "اللوحة غير موجودة")216 217 updates = {}218 if defects_data:219 updates["defects_data"] = json.loads(defects_data) if isinstance(defects_data, str) else defects_data220 if report_data:221 updates["report_data"] = json.loads(report_data) if isinstance(report_data, str) else report_data222 223 if board.update(**updates):224 return {"success": True}225 else:226 raise HTTPException(500, "فشل في تحديث اللوحة")227 228@app.delete("/api/boards/{board_id}")229async def delete_board(board_id: int, current_user: User = Depends(get_current_user)):230 board = Board.get_by_id(current_user.id, board_id)231 if not board:232 raise HTTPException(404, "اللوحة غير موجودة")233 234 if board.delete():235 return {"success": True}236 else:237 raise HTTPException(500, "فشل في حذف اللوحة")238 239@app.post("/api/boards/upload-image")240async def upload_board_image(241 file: UploadFile = File(...),242 board_id: int = Form(...),243 current_user: User = Depends(get_current_user)244):245 board = Board.get_by_id(current_user.id, board_id)246 if not board:247 raise HTTPException(404, "اللوحة غير موجودة")248 249 filename = f"board_{board_id}_{uuid.uuid4()}.jpg"250 filepath = f"static/annotated/{filename}"251 os.makedirs("static/annotated", exist_ok=True)252 253 contents = await file.read()254 with open(filepath, "wb") as f:255 f.write(contents)256 257 # تحديث مسار الصورة في اللوحة258 board.annotated_image_path = f"/static/annotated/{filename}"259 board.update(annotated_image_path=board.annotated_image_path)260 261 return {"success": True}262 263# -------------------- الكشف --------------------264 265@app.post("/detect")266async def detect_without_auth(267 file: UploadFile = File(...),268 conf_threshold: float = Form(0.5),269 iou_threshold: float = Form(0.45),270 max_det: int = Form(300),271 imgsz: int = Form(640),272 lang: str = Form("ar")273):274 """كشف العيوب - لا يحتاج تسجيل دخول"""275 if not model:276 raise HTTPException(500, "النموذج غير محمل")277 278 temp_path = f"static/temp_{uuid.uuid4()}.jpg"279 contents = await file.read()280 with open(temp_path, "wb") as f:281 f.write(contents)282 283 result = await process_image(temp_path, model, conf_threshold, iou_threshold, max_det, imgsz, lang)284 os.remove(temp_path)285 286 return result287 288# ============================================289# PDF Report Endpoint (يحتاج تعديل لاحق)290# ============================================291 292@app.post("/api/detect/pdf")293async def detect_and_generate_pdf(294 file: UploadFile = File(...),295 conf_threshold: float = Form(0.5),296 iou_threshold: float = Form(0.45),297 max_det: int = Form(300),298 imgsz: int = Form(640),299 lang: str = Form("ar"),300 current_user: User = Depends(get_current_user)301):302 """كشف العيوب وتوليد PDF مباشرة"""303 if not model:304 raise HTTPException(500, "النموذج غير محمل")305 306 # إضافة مكتبات إعادة تشكيل النص العربي307 import arabic_reshaper308 from bidi.algorithm import get_display309 310 temp_path = f"static/temp_{uuid.uuid4()}.jpg"311 contents = await file.read()312 with open(temp_path, "wb") as f:313 f.write(contents)314 315 result = await process_image(temp_path, model, conf_threshold, iou_threshold, max_det, imgsz, lang)316 os.remove(temp_path)317 318 # إنشاء PDF من النتيجة319 buffer = io.BytesIO()320 doc = SimpleDocTemplate(buffer, pagesize=A4, rightMargin=72, leftMargin=72, topMargin=72, bottomMargin=72)321 styles = getSampleStyleSheet()322 323 # دالة مساعدة لإعادة تشكيل النص العربي324 def reshape_arabic(text):325 """إعادة تشكيل النص العربي للعرض الصحيح في PDF"""326 reshaped = arabic_reshaper.reshape(text)327 return get_display(reshaped)328 329 # محاولة تحميل خط عربي - مسارات مختلفة تناسب Hugging Face Spaces330 font_loaded = False331 possible_font_paths = [332 # مسارات شائعة في Hugging Face Spaces333 '/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf',334 '/usr/share/fonts/truetype/liberation/LiberationSans-Regular.ttf',335 '/usr/share/fonts/truetype/noto/NotoSans-Regular.ttf',336 '/usr/share/fonts/truetype/ubuntu/Ubuntu-Regular.ttf',337 # مسارات إضافية338 '/System/Library/Fonts/Arial.ttf',339 'C:\\Windows\\Fonts\\Arial.ttf'340 ]341 342 try:343 for font_path in possible_font_paths:344 if os.path.exists(font_path):345 pdfmetrics.registerFont(TTFont('ArabicFont', font_path))346 font_loaded = True347 print(f"✅ تم تحميل الخط من: {font_path}")348 break349 350 # إنشاء أنماط النص مع إعادة التشكيل351 arabic_style = ParagraphStyle(352 'ArabicStyle', 353 parent=styles['Normal'], 354 fontName='ArabicFont' if font_loaded else 'Helvetica', 355 fontSize=11, 356 alignment=TA_RIGHT357 )358 title_style = ParagraphStyle(359 'TitleStyle', 360 parent=styles['Title'], 361 fontName='ArabicFont' if font_loaded else 'Helvetica', 362 fontSize=18, 363 alignment=TA_CENTER, 364 textColor=colors.HexColor('#2563eb')365 )366 367 # إذا لم يتم العثور على خط، نستخدم النص المعاد تشكيله مع الخط الافتراضي368 if not font_loaded:369 print("⚠️ تحذير: لم يتم العثور على خط عربي، سيتم استخدام إعادة تشكيل النص فقط")370 371 except Exception as e:372 print(f"خطأ في تحميل الخط: {e}")373 arabic_style = styles['Normal']374 title_style = styles['Title']375 376 story = []377 # إعادة تشكيل جميع النصوص العربية378 story.append(Paragraph(reshape_arabic("📋 تقرير كشف عيوب PCB"), title_style))379 story.append(Spacer(1, 12))380 story.append(Paragraph(reshape_arabic(f"<b>تاريخ الفحص:</b> {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"), arabic_style))381 story.append(Paragraph(reshape_arabic(f"<b>عدد العيوب:</b> {result.get('num_defects', 0)}"), arabic_style))382 story.append(Spacer(1, 12))383 384 decision = result.get('board_decision', '✅ مقبول')385 decision_color = '#10b981'386 if 'مرفوض' in decision:387 decision_color = '#ef4444'388 elif 'مقبول مشروط' in decision:389 decision_color = '#f59e0b'390 391 story.append(Paragraph(reshape_arabic("<b>القرار النهائي:</b>"), arabic_style))392 story.append(Paragraph(f'<font color="{decision_color}">{reshape_arabic(decision)}</font>', arabic_style))393 story.append(Spacer(1, 12))394 395 detections = result.get('detections', [])396 if detections:397 story.append(Paragraph(reshape_arabic("<b>🔍 العيوب المكتشفة</b>"), arabic_style))398 story.append(Spacer(1, 6))399 400 # إعادة تشكيل محتوى الجدول401 table_data = [[reshape_arabic("نوع العيب"), reshape_arabic("نسبة الثقة")]]402 for d in detections[:20]:403 defect_name = d.get('defect_type_ar', d.get('defect_type', 'غير معروف'))404 confidence = f"{d.get('confidence', 0) * 100:.1f}%"405 table_data.append([reshape_arabic(defect_name), confidence])406 407 table = Table(table_data, colWidths=[300, 100])408 table.setStyle(TableStyle([409 ('BACKGROUND', (0, 0), (-1, 0), colors.HexColor('#2563eb')),410 ('TEXTCOLOR', (0, 0), (-1, 0), colors.white),411 ('ALIGN', (0, 0), (-1, -1), 'CENTER'),412 ('VALIGN', (0, 0), (-1, -1), 'MIDDLE'),413 ('FONTNAME', (0, 0), (-1, -1), 'ArabicFont' if font_loaded else 'Helvetica'),414 ('FONTSIZE', (0, 0), (-1, -1), 10),415 ('GRID', (0, 0), (-1, -1), 0.5, colors.grey),416 ]))417 story.append(table)418 else:419 story.append(Paragraph(reshape_arabic("✅ لم يتم اكتشاف أي عيوب"), arabic_style))420 421 doc.build(story)422 buffer.seek(0)423 424 return Response(content=buffer.getvalue(), media_type="application/pdf", headers={425 "Content-Disposition": f"attachment; filename=pcb_detection_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf"426 })427# ============================================428# Statistics Dashboard Endpoint429# ============================================430 431@app.get("/api/stats")432async def get_statistics(current_user: User = Depends(get_current_user)):433 """جلب إحصائيات العيوب للمستخدم الحالي"""434 435 boards = Board.get_all_by_user(current_user.id)436 437 defect_counts = {438 "missing_hole": 0,439 "mouse_bite": 0,440 "open_circuit": 0,441 "short_circuit": 0,442 "spur": 0,443 "spurious_copper": 0444 }445 446 severity_counts = {447 "critical": 0,448 "high": 0,449 "medium": 0,450 "low": 0451 }452 453 total_defects = 0454 total_boards = len(boards)455 456 print(f"📊 عدد اللوحات للمستخدم {current_user.id}: {total_boards}")457 458 for board in boards:459 print(f"📋 لوحة: {board.id} - {board.title}")460 print(f" defects_data type: {type(board.defects_data)}")461 print(f" defects_data: {board.defects_data}")462 463 if board.defects_data:464 # محاولة قراءة البيانات465 defects = None466 if isinstance(board.defects_data, list):467 defects = board.defects_data468 elif isinstance(board.defects_data, str):469 try:470 defects = json.loads(board.defects_data)471 except:472 defects = []473 else:474 defects = []475 476 for defect in defects:477 defect_type = defect.get('defect_type', '')478 if defect_type in defect_counts:479 defect_counts[defect_type] += 1480 total_defects += 1481 482 if defect_type in ["short_circuit", "open_circuit"]:483 severity = "critical"484 elif defect_type in ["missing_hole"]:485 severity = "high"486 elif defect_type in ["spur", "spurious_copper"]:487 severity = "medium"488 elif defect_type in ["mouse_bite"]:489 severity = "low"490 else:491 severity = "medium"492 493 severity_counts[severity] += 1494 495 defect_names_ar = {496 "missing_hole": "ثقب مفقود",497 "mouse_bite": "عضّة فأر",498 "open_circuit": "دارة مفتوحة",499 "short_circuit": "دارة قصيرة",500 "spur": "نتوء نحاسي",501 "spurious_copper": "نحاس زائد"502 }503 504 # ترتيب العيوب حسب الكثرة505 sorted_defects = [(k, v) for k, v in defect_counts.items() if v > 0]506 sorted_defects.sort(key=lambda x: x[1], reverse=True)507 most_common = [{"type": defect_names_ar.get(d[0], d[0]), "count": d[1]} for d in sorted_defects[:3]]508 509 print(f"📊 النتائج النهائية: total_defects={total_defects}, defect_counts={defect_counts}")510 511 return {512 "total_boards": total_boards,513 "total_defects": total_defects,514 "defect_counts": {defect_names_ar.get(k, k): v for k, v in defect_counts.items()},515 "severity_counts": severity_counts,516 "most_common_defects": most_common,517 "model_accuracy": 99.01518 }519# ============================================520# Batch Processing Endpoint521# ============================================522 523@app.post("/api/detect/batch")524async def detect_batch(525 files: List[UploadFile] = File(...),526 conf_threshold: float = Form(0.5),527 iou_threshold: float = Form(0.45),528 max_det: int = Form(300),529 imgsz: int = Form(640),530 lang: str = Form("ar"),531 current_user: User = Depends(get_current_user)532):533 """معالجة دفعة من الصور واكتشاف العيوب في كل منها"""534 if not model:535 raise HTTPException(500, "النموذج غير محمل")536 537 results = []538 539 for file in files:540 if not file.content_type.startswith("image/"):541 results.append({542 "filename": file.filename,543 "success": False,544 "error": "الملف ليس صورة"545 })546 continue547 548 try:549 temp_path = f"static/temp_{uuid.uuid4()}.jpg"550 contents = await file.read()551 with open(temp_path, "wb") as f:552 f.write(contents)553 554 result = await process_image(temp_path, model, conf_threshold, iou_threshold, max_det, imgsz, lang)555 os.remove(temp_path)556 557 results.append({558 "filename": file.filename,559 "success": True,560 "num_defects": result.get("num_defects", 0),561 "detections": result.get("detections", []),562 "board_decision": result.get("board_decision", "✅ مقبول"),563 "annotated_image_base64": result.get("annotated_image_base64", None)564 })565 566 except Exception as e:567 results.append({568 "filename": file.filename,569 "success": False,570 "error": str(e)571 })572 573 total_images = len(files)574 successful = sum(1 for r in results if r.get("success", False))575 total_defects = sum(r.get("num_defects", 0) for r in results if r.get("success", False))576 577 return {578 "success": True,579 "total_images": total_images,580 "successful_count": successful,581 "failed_count": total_images - successful,582 "total_defects": total_defects,583 "results": results584 }585 586# ============================================587# الصفحة الرئيسية588# ============================================589 590@app.get("/", response_class=HTMLResponse)591async def main_page():592 try:593 with open("templates/index.html", "r", encoding="utf-8") as f:594 return f.read()595 except FileNotFoundError:596 return HTMLResponse(content="<h1>PCB Detector</h1><p>API is running. Visit /docs for API documentation.</p>")