""" main.py — FastAPI application Endpoints GET / → หน้าเว็บ Frontend (เสิร์ฟจากโฟลเดอร์ ../frontend) GET /api/health → สถานะระบบ AI + ฐานข้อมูล POST /api/scan → รับภาพ Base64 → AI อ่านป้าย → คืนข้อมูล Order POST /api/lookup → รับเลขทะเบียนที่พนักงานพิมพ์เอง → คืนข้อมูล Order หลักการออกแบบการตอบกลับ: ทุก endpoint ตอบ HTTP 200 พร้อม body {success, code, message, ...} เสมอ (ยกเว้น 422 จาก validation ของ Pydantic ซึ่งถูกดักแปลงเป็นรูปแบบเดียวกัน) เพื่อให้ Frontend บน iPad จัดการง่าย และไม่มีหน้าจอ error ดิบๆ โผล่หน้างาน """ from __future__ import annotations import asyncio import logging import time from contextlib import asynccontextmanager from pathlib import Path from typing import Any, Dict, Optional from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field from .config import settings from . import database as db from . import plate_utils as pu from .pipeline import pipeline # --------------------------------------------------------------------- # Logging # --------------------------------------------------------------------- logging.basicConfig( level=getattr(logging, settings.LOG_LEVEL.upper(), logging.INFO), format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) log = logging.getLogger("api") FRONTEND_DIR = settings.BASE_DIR.parent / "frontend" # --------------------------------------------------------------------- # Lifespan — โหลดโมเดล AI ครั้งเดียวตอนสตาร์ท # --------------------------------------------------------------------- @asynccontextmanager async def lifespan(app: FastAPI): log.info("=" * 60) log.info("เริ่มระบบสแกนป้ายทะเบียนรถบรรทุก — กำลังโหลดโมเดล AI…") log.info("(ครั้งแรกจะดาวน์โหลดโมเดล YOLO + EasyOCR ใช้เวลาสักครู่)") # โหลดใน thread แยก ไม่ให้บล็อก event loop ตอน startup await asyncio.to_thread(pipeline.load) log.info("สถานะระบบ AI: %s", "พร้อมใช้งาน" if pipeline.ready else f"ไม่พร้อม — {pipeline.load_error}") log.info("โหมดฐานข้อมูล: %s", "MySQL" if settings.USE_MYSQL else "Mockup") log.info("=" * 60) yield log.info("ปิดระบบเรียบร้อย") app = FastAPI( title="Thai Truck Plate Scanner", description="ระบบสแกนป้ายทะเบียนรถบรรทุกไทยเพื่อเรียกคิว Order (Self-Hosted)", version="1.0.0", lifespan=lifespan, ) app.add_middleware( CORSMiddleware, allow_origins=settings.CORS_ORIGINS, allow_credentials=False, allow_methods=["*"], allow_headers=["*"], ) # --------------------------------------------------------------------- # Schema # --------------------------------------------------------------------- class ScanRequest(BaseModel): image: str = Field(..., description="ภาพ JPEG เข้ารหัส Base64 (มีหรือไม่มี data URL prefix ก็ได้)") class LookupRequest(BaseModel): plate: str = Field(..., max_length=32, description="เลขทะเบียนที่พนักงานพิมพ์เอง") def envelope(success: bool, code: str, message: str, **extra: Any) -> Dict[str, Any]: """โครงสร้าง response มาตรฐานของทุก endpoint""" body = {"success": success, "code": code, "message": message} body.update(extra) return body def attach_order(plate: str, base: Dict[str, Any]) -> Dict[str, Any]: """ ค้นหา Order แล้วผนวกเข้ากับ response ถ้าไม่พบ → success=False code=ORDER_NOT_FOUND แต่ยังส่งเลขทะเบียนกลับไป เพื่อให้พนักงานเห็นว่า AI อ่านได้อะไร แล้วแก้ในช่องพิมพ์มือได้ทันที """ try: order = db.query_order_by_plate(plate) except Exception as exc: # noqa: BLE001 log.exception("query_order_by_plate ล้มเหลว") return envelope(False, "DB_ERROR", f"ค้นฐานข้อมูลไม่สำเร็จ: {type(exc).__name__}", plate=plate, **base) if not order: return envelope(False, "ORDER_NOT_FOUND", f"ไม่พบ Order ของทะเบียน {plate} ในระบบวันนี้", plate=plate, **base) return envelope(True, "OK", "พบข้อมูล Order", plate=plate, order=order, **base) # --------------------------------------------------------------------- # POST /api/scan — เส้นทางหลัก (ภาพ → AI → Order) # --------------------------------------------------------------------- # รันงาน AI ได้ทีละภาพเท่านั้น # # เหตุผล: เซิร์ฟเวอร์รัน worker เดียวและงาน AI เป็น CPU-bound # ถ้ามีคำขอเข้ามาซ้อน (พนักงานกดถ่ายรัว หรือ frontend ส่งซ้ำตอน timeout) # คำขอที่สองจะไปต่อคิวรอคำขอแรก ทำให้ยิ่งช้าและ timeout ตามกันเป็นลูกโซ่ # สู้ตอบกลับทันทีว่า "กำลังประมวลผลคันก่อนหน้า" ให้พนักงานรู้ตัวดีกว่า _scan_lock = asyncio.Semaphore(1) @app.post("/api/scan") async def api_scan(req: ScanRequest): if _scan_lock.locked(): return JSONResponse(envelope( False, "BUSY", "ระบบกำลังประมวลผลรถคันก่อนหน้า กรุณารอสักครู่แล้วกดถ่ายใหม่")) async with _scan_lock: return await _do_scan(req) async def _do_scan(req: ScanRequest): t0 = time.perf_counter() # ---- 1) ถอด Base64 เป็นภาพ OpenCV ---- try: bgr = pu.decode_base64_image(req.image) except ValueError as exc: return JSONResponse(envelope(False, "BAD_IMAGE", str(exc))) except Exception as exc: # noqa: BLE001 log.exception("decode ภาพล้มเหลว") return JSONResponse(envelope(False, "BAD_IMAGE", "อ่านไฟล์ภาพไม่สำเร็จ")) # ---- 2) รัน AI ใน thread pool (CPU-bound: ห้ามบล็อก event loop) ---- try: result = await asyncio.to_thread(pipeline.scan, bgr) except Exception as exc: # noqa: BLE001 log.exception("pipeline ล้มเหลวระดับ thread") return JSONResponse(envelope(False, "ENGINE_ERROR", "ระบบ AI ขัดข้องชั่วคราว กรุณาพิมพ์เลขทะเบียนเอง")) # เก็บภาพที่อ่านไม่ได้ไว้ปรับปรุงโมเดลภายหลัง debug_path = None if not result.ok: debug_path = await asyncio.to_thread(pipeline.save_debug, bgr, f"FAIL_{result.code}") meta = { "detector": result.detector, "raw_texts": result.raw_texts, "elapsed_ms": int((time.perf_counter() - t0) * 1000), } # ---- 3) AI อ่านไม่ได้ → แจ้งกลับอย่างนุ่มนวล (HTTP ยังเป็น 200) ---- if not result.ok: log.info("สแกนไม่สำเร็จ code=%s texts=%s", result.code, result.raw_texts[:5]) return JSONResponse(envelope(False, result.code, result.message, debug_image=debug_path, **meta)) # ---- 4) อ่านได้ → ไปค้น Order ---- log.info("อ่านป้ายได้: %s (conf=%.2f, votes=%d, %s)", result.plate, result.confidence, result.votes, result.detector) # อ่านได้เสียงเดียว = เสี่ยงอ่านผิดเป็นเลขที่ถูกรูปแบบ → เก็บภาพไว้ตรวจย้อนหลังด้วย if result.needs_confirm: await asyncio.to_thread(pipeline.save_debug, bgr, f"LOWCONF_{result.plate}") body = attach_order(result.plate, { **meta, "confidence": result.confidence, "votes": result.votes, "needs_confirm": result.needs_confirm, "province": result.province, "source": "ai", }) return JSONResponse(body) # --------------------------------------------------------------------- # POST /api/lookup — พิมพ์มือกรณีฉุกเฉิน # --------------------------------------------------------------------- @app.post("/api/lookup") async def api_lookup(req: LookupRequest): plate = pu.normalize_manual_plate(req.plate) if not plate: return JSONResponse(envelope( False, "INVALID_FORMAT", f"รูปแบบไม่ถูกต้อง — ป้ายรถบรรทุกต้องเป็นตัวเลข 6 หลัก หมวด " f"{settings.PLATE_PREFIX_MIN}–{settings.PLATE_PREFIX_MAX} เช่น 70-1234", )) log.info("ค้นหาด้วยการพิมพ์มือ: %s", plate) return JSONResponse(attach_order(plate, {"source": "manual", "confidence": 1.0})) # --------------------------------------------------------------------- # GET /api/health # --------------------------------------------------------------------- @app.get("/api/health") async def api_health(): # เตือนให้ชัดเมื่อระบบทำงานได้แต่ "ไม่เต็มประสิทธิภาพ" # ไม่งั้นจะดูเหมือนปกติทุกอย่าง ทั้งที่ช้าและแม่นน้อยกว่าที่ควร warnings = [] if pipeline.detector is None: warnings.append("ไม่มีโมเดลตรวจจับป้าย — ต้องอ่าน OCR ทั้งภาพ ช้ากว่าราว 10 เท่าและแม่นน้อยลงมาก") elif pipeline.detector_kind == "vehicle": warnings.append("ใช้โมเดลสำรอง (ตรวจจับตัวรถ) — แนะนำให้วางโมเดลป้ายทะเบียนที่ models/license_plate.pt") return { "status": "ok", "warnings": warnings, "ai": { "ready": pipeline.ready, "detector_kind": pipeline.detector_kind, "error": pipeline.load_error, "ocr_langs": settings.OCR_LANGS, "min_votes": pipeline.MIN_VOTES, "max_scan_seconds": settings.MAX_SCAN_SECONDS, }, "database": db.db_healthcheck(), "plate_range": f"{settings.PLATE_PREFIX_MIN}-{settings.PLATE_PREFIX_MAX}", "server_time": time.strftime("%Y-%m-%d %H:%M:%S"), } # --------------------------------------------------------------------- # ดักทุก exception ที่หลุดออกมา ไม่ให้ตอบ 500 ดิบๆ # --------------------------------------------------------------------- @app.exception_handler(Exception) async def unhandled_exception_handler(request: Request, exc: Exception): log.exception("Unhandled error ที่ %s", request.url.path) return JSONResponse( status_code=200, content=envelope(False, "SERVER_ERROR", "เซิร์ฟเวอร์ขัดข้องชั่วคราว กรุณาลองใหม่ หรือพิมพ์เลขทะเบียนเอง"), ) # --------------------------------------------------------------------- # เสิร์ฟ Frontend จาก FastAPI โดยตรง (จะได้ไม่ต้องพึ่ง XAMPP ก็ได้) # --------------------------------------------------------------------- if FRONTEND_DIR.is_dir(): @app.get("/") async def serve_index(): index = FRONTEND_DIR / "index.html" if index.is_file(): return FileResponse(index) return JSONResponse({"detail": "ไม่พบไฟล์ frontend/index.html"}, status_code=404) app.mount("/static", StaticFiles(directory=str(FRONTEND_DIR)), name="static") # --------------------------------------------------------------------- # รันตรงๆ ด้วย python -m app.main (หรือใช้ uvicorn ตาม README) # --------------------------------------------------------------------- if __name__ == "__main__": import uvicorn uvicorn.run("app.main:app", host=settings.HOST, port=settings.PORT, reload=False)