File size: 8,497 Bytes
b42373a fa17f1e b42373a fa17f1e b42373a fa17f1e b42373a fa17f1e b42373a fa17f1e b42373a fa17f1e b42373a fa17f1e b42373a fa17f1e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 | 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) |