| import os |
| import json |
| import shutil |
| import base64 |
| import uuid |
| import cv2 |
| import numpy as np |
| from fastapi import FastAPI, UploadFile, File, Form, HTTPException, BackgroundTasks |
| from fastapi.responses import FileResponse, JSONResponse |
| from fastapi.staticfiles import StaticFiles |
| from fastapi.middleware.cors import CORSMiddleware |
| from contextlib import asynccontextmanager |
| from barcode_scanner import scan_barcode, scan_all_barcodes |
| from ocr import read_chassis, postprocess_with_hint, ocr_image, get_ocr |
| from preprocess import preprocess_chassis |
| from evaluate import evaluate, get_pairs |
|
|
| @asynccontextmanager |
| async def lifespan(app: FastAPI): |
| get_ocr() |
| yield |
|
|
| app = FastAPI(title="Chassis OCR API", description="API backend for Chassis OCR PWA", lifespan=lifespan) |
| app.add_middleware( |
| CORSMiddleware, |
| allow_origins=["*"], |
| allow_credentials=True, |
| allow_methods=["*"], |
| allow_headers=["*"], |
| ) |
|
|
| TEMP_DIR = "temp_uploads" |
| RESULTS_DIR = "results" |
| CONFIG_PATH = "config.json" |
| os.makedirs(TEMP_DIR, exist_ok=True) |
| os.makedirs(RESULTS_DIR, exist_ok=True) |
|
|
| def cv2_to_base64(img): |
| _, buffer = cv2.imencode('.jpg', img) |
| return base64.b64encode(buffer).decode('utf-8') |
|
|
| @app.get("/api/status") |
| def get_status(): |
| return { |
| "status": "online", |
| "message": "OCR Backend is active" |
| } |
|
|
| @app.get("/api/test-pairs") |
| def get_test_pairs(): |
| barcode_dir = "images/barcode" |
| chassis_dir = "images/chassis" |
| if not os.path.exists(barcode_dir) or not os.path.exists(chassis_dir): |
| return [] |
| barcodes = {os.path.splitext(f)[0] for f in os.listdir(barcode_dir) if f.lower().endswith(('.jpg', '.jpeg', '.png'))} |
| chassis = {os.path.splitext(f)[0] for f in os.listdir(chassis_dir) if f.lower().endswith(('.jpg', '.jpeg', '.png'))} |
| common = sorted(list(barcodes & chassis)) |
| return common |
|
|
| @app.get("/api/scan-barcode/{key}") |
| def scan_barcode_by_key(key: str): |
| for ext in ['.jpg', '.jpeg', '.png', '.JPG', '.PNG']: |
| path = os.path.join("images/barcode", f"{key}{ext}") |
| if os.path.exists(path): |
| result = scan_barcode(path) |
| return {"success": result is not None, "barcode": result} |
| raise HTTPException(status_code=404, detail=f"Barcode image for key '{key}' not found") |
|
|
| @app.post("/api/scan-barcode") |
| async def api_scan_barcode(file: UploadFile = File(...)): |
| temp_filename = f"{uuid.uuid4()}_{file.filename}" |
| temp_path = os.path.join(TEMP_DIR, temp_filename) |
| try: |
| with open(temp_path, "wb") as buffer: |
| shutil.copyfileobj(file.file, buffer) |
| result = scan_barcode(temp_path) |
| return {"success": result is not None, "barcode": result} |
| except Exception as e: |
| raise HTTPException(status_code=500, detail=str(e)) |
| finally: |
| if os.path.exists(temp_path): |
| os.remove(temp_path) |
|
|
| @app.get("/api/config") |
| def get_config(): |
| if os.path.exists(CONFIG_PATH): |
| with open(CONFIG_PATH) as f: |
| return json.load(f) |
| return {} |
|
|
| @app.post("/api/config") |
| async def save_config(config_data: dict): |
| try: |
| with open(CONFIG_PATH, "w") as f: |
| json.dump(config_data, f, indent=2) |
| return {"status": "success", "message": "Configuration updated successfully"} |
| except Exception as e: |
| raise HTTPException(status_code=500, detail=str(e)) |
|
|
| eval_status = {"running": False, "progress": 0, "total": 0, "results": []} |
|
|
| def run_evaluation_task(): |
| global eval_status |
| try: |
| eval_status["running"] = True |
| eval_status["progress"] = 0 |
| evaluate() |
| report_path = os.path.join(RESULTS_DIR, "report.json") |
| if os.path.exists(report_path): |
| with open(report_path) as f: |
| eval_status["results"] = json.load(f) |
| eval_status["progress"] = len(eval_status["results"]) |
| eval_status["total"] = len(eval_status["results"]) |
| except Exception as e: |
| print(f"Error in evaluation background task: {e}") |
| finally: |
| eval_status["running"] = False |
|
|
| @app.post("/api/evaluate") |
| def trigger_evaluation(background_tasks: BackgroundTasks): |
| global eval_status |
| if eval_status["running"]: |
| return {"status": "already_running", "message": "Evaluation task is currently running"} |
| eval_status = {"running": True, "progress": 0, "total": 50, "results": []} |
| background_tasks.add_task(run_evaluation_task) |
| return {"status": "started", "message": "Batch evaluation started in the background"} |
|
|
| @app.get("/api/evaluate/status") |
| def get_evaluation_status(): |
| report_path = os.path.join(RESULTS_DIR, "report.json") |
| results = [] |
| if os.path.exists(report_path): |
| try: |
| with open(report_path) as f: |
| results = json.load(f) |
| except Exception: |
| pass |
| return { |
| "running": eval_status["running"], |
| "progress": eval_status["progress"], |
| "total": eval_status["total"], |
| "has_existing_report": len(results) > 0, |
| "results": results if not eval_status["running"] else eval_status["results"] |
| } |
|
|
| @app.post("/api/match") |
| async def match_chassis( |
| barcode_val: str = Form(...), |
| chassis_file: UploadFile = File(None), |
| chassis_key: str = Form(None) |
| ): |
| if not chassis_file and not chassis_key: |
| raise HTTPException(status_code=400, detail="Either chassis_file or chassis_key must be provided") |
|
|
| chassis_path = None |
| temp_path = None |
|
|
| if chassis_key: |
| for ext in ['.jpg', '.jpeg', '.png', '.JPG', '.PNG']: |
| p = os.path.join("images/chassis", f"{chassis_key}{ext}") |
| if os.path.exists(p): |
| chassis_path = p |
| break |
| if not chassis_path: |
| raise HTTPException(status_code=404, detail=f"Chassis image for key '{chassis_key}' not found in images/chassis") |
| else: |
| temp_filename = f"{uuid.uuid4()}_{chassis_file.filename}" |
| temp_path = os.path.join(TEMP_DIR, temp_filename) |
| with open(temp_path, "wb") as buffer: |
| shutil.copyfileobj(chassis_file.file, buffer) |
| chassis_path = temp_path |
|
|
| try: |
| save_comp = True |
| comp_filename = chassis_key if chassis_key else os.path.splitext(chassis_file.filename)[0] |
| variations = preprocess_chassis(chassis_path, save_comparison=save_comp) |
| original_img = cv2.imread(chassis_path) |
| base64_original = cv2_to_base64(original_img) |
|
|
| base64_variations = [] |
| labels = ["CLAHE", "Bilateral", "Otsu", "Adaptive"] |
| for idx, var in enumerate(variations): |
| base64_variations.append({ |
| "label": labels[idx], |
| "base64": cv2_to_base64(var) |
| }) |
|
|
| best_text, best_conf, best_score = "", 0.0, -1 |
| winning_label = "" |
| variation_details = [] |
|
|
| for idx, var in enumerate(variations): |
| text, conf = ocr_image(var) |
| score = conf * max(len(text), 1) |
| variation_details.append({ |
| "label": labels[idx], |
| "text": text, |
| "confidence": conf, |
| "score": score |
| }) |
| if score > best_score: |
| best_text, best_conf, best_score = text, conf, score |
| winning_label = labels[idx] |
|
|
| corrected_text, is_match = postprocess_with_hint(best_text, barcode_val) |
| status = "FAILED" |
| if best_text == barcode_val: |
| status = "EXACT" |
| elif is_match: |
| status = "CORRECTED" |
|
|
| return { |
| "success": is_match, |
| "status": status, |
| "barcode_val": barcode_val, |
| "raw_ocr": best_text, |
| "corrected_ocr": corrected_text, |
| "confidence": best_conf, |
| "winning_label": winning_label, |
| "variations": base64_variations, |
| "original": base64_original, |
| "variation_details": variation_details, |
| "comparison_url": f"/results/{comp_filename}_comparison.jpg" if save_comp else None |
| } |
|
|
| except Exception as e: |
| raise HTTPException(status_code=500, detail=str(e)) |
| finally: |
| if temp_path and os.path.exists(temp_path): |
| os.remove(temp_path) |
|
|
| app.mount("/results", StaticFiles(directory="results"), name="results") |
| app.mount("/", StaticFiles(directory="web", html=True), name="static") |
|
|
| if __name__ == "__main__": |
| import uvicorn |
| uvicorn.run("server:app", host="0.0.0.0", port=8000, reload=True) |
