""" Interactive Web Application: Dorayaki (どら焼き) Supermarket Shelf Product Recognition & Inventory System Features: - Light, fresh pastel green visual design - Brand Logo: Dorayaki - Live camera photo capture + File upload + Select from D:\\recognization\\add - Clean bounding box detection (NO class labels, NO confidence text) - Dual language support (English / Japanese) - Fine-grained classification: SURF, ORB, BRISK, Hybrid (Paper method) """ import os import sys import io import cv2 import json import base64 import argparse import numpy as np from typing import Optional, List, Dict current_dir = os.path.dirname(os.path.abspath(__file__)) project_root = os.path.dirname(current_dir) if os.path.exists(os.path.join(project_root, "stock")): os.chdir(project_root) if current_dir not in sys.path: sys.path.insert(0, current_dir) from pipeline import ShelfRecognitionPipeline def encode_image_base64(img_bgr: np.ndarray, quality: int = 85) -> str: """Encodes an OpenCV image to a base64 JPEG data URL.""" if img_bgr is None or img_bgr.size == 0: return "" success, buffer = cv2.imencode(".jpg", img_bgr, [cv2.IMWRITE_JPEG_QUALITY, quality]) if not success: return "" return f"data:image/jpeg;base64,{base64.b64encode(buffer).decode('utf-8')}" def decode_image_base64(data_url: str) -> Optional[np.ndarray]: """Decodes a base64 data URL into an OpenCV BGR image.""" try: if "," in data_url: data_url = data_url.split(",")[1] img_bytes = base64.b64decode(data_url) np_arr = np.frombuffer(img_bytes, np.uint8) return cv2.imdecode(np_arr, cv2.IMREAD_COLOR) except Exception as e: print(f"[Error] Failed to decode base64 image: {e}") return None def get_available_shelf_images(add_dir: str = "add") -> List[Dict[str, str]]: """Lists images from D:\\recognization\\add and root image.jpg.""" items = [] if os.path.exists("image.jpg"): items.append({"path": "image.jpg", "label": "image.jpg (Main Shelf)"}) if os.path.exists(add_dir): valid_exts = (".webp", ".jpg", ".jpeg", ".png") files = sorted( [f for f in os.listdir(add_dir) if f.lower().endswith(valid_exts)], key=lambda x: (len(x), x) ) for f in files: p = os.path.join(add_dir, f).replace("\\", "/") items.append({"path": p, "label": f"{add_dir}/{f}"}) # Also include runs samples if present mosaic = "runs/detect/train/train_batch0.jpg" if os.path.exists(mosaic): items.append({"path": mosaic, "label": "train_batch0.jpg (Mosaic)"}) return items def create_fastapi_app(pipeline: ShelfRecognitionPipeline): from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse, JSONResponse from fastapi.middleware.cors import CORSMiddleware app = FastAPI(title="Dorayaki - Supermarket Shelf Intelligence") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) HTML_PAGE = """ Dorayaki - Smart Supermarket Shelf AI
Dorayaki どら焼き
Smart Supermarket Shelf AI & Inventory Recognition
PM
AI
Dev
Reviewer
Algorithm:
👆 Click on any product packaging to identify & view real-time inventory
Product Details Ready
🏷️
No product selected.
Click any merchandise item on the shelf image to inspect details.
Feature Matching Inliers --
Visual keypoint matching will appear here
""" @app.get("/", response_class=HTMLResponse) async def index(): return HTMLResponse(content=HTML_PAGE) @app.get("/api/shelf_samples") async def get_shelf_samples(): samples = get_available_shelf_images("add") return {"status": "success", "samples": samples} @app.post("/api/load_shelf") async def load_shelf(payload: dict): img_path = payload.get("image_path", "image.jpg") init_res = pipeline.set_active_shelf_image(img_path) if init_res.get("status") == "success": return { "status": "success", "boxes": init_res["boxes"], "empty_slots": init_res.get("empty_slots", []), "image_base64": encode_image_base64(pipeline.current_image) } return JSONResponse(status_code=400, content={"status": "error", "message": "Failed to load image"}) @app.post("/api/upload_shelf") async def upload_shelf(payload: dict): data_url = payload.get("image_base64", "") img_bgr = decode_image_base64(data_url) if img_bgr is None: return JSONResponse(status_code=400, content={"status": "error", "message": "Failed to decode image"}) init_res = pipeline.set_active_shelf_image(img_bgr) if init_res.get("status") == "success": return { "status": "success", "boxes": init_res["boxes"], "empty_slots": init_res.get("empty_slots", []), "image_base64": encode_image_base64(pipeline.current_image) } return JSONResponse(status_code=400, content={"status": "error", "message": "Detection failed"}) @app.post("/api/click") async def handle_click(payload: dict): x = int(payload.get("x", 0)) y = int(payload.get("y", 0)) method = payload.get("method", "hybrid") res = pipeline.process_click(x, y, method=method) if res.get("status") == "success": return { "status": "success", "target_box": res["target_box"], "classification": res["classification"], "inventory": res["inventory"], "crop_base64": encode_image_base64(res["crop_image"]), "vis_base64": encode_image_base64(res["match_visualization"]) } elif res.get("status") == "empty_slot": return { "status": "empty_slot", "empty_slot": res["empty_slot"], "message": res["message"] } return JSONResponse(status_code=404, content=res) @app.post("/api/update_stock") async def update_stock(payload: dict): sku_id = payload.get("sku_id") delta = int(payload.get("delta", -1)) ok = pipeline.inventory.update_stock(sku_id, delta) if ok: p = pipeline.inventory.get_product(sku_id) return {"status": "success", "number_on": p["number_on"], "number_out": p["number_out"]} return JSONResponse(status_code=400, content={"status": "error"}) return app def main(): parser = argparse.ArgumentParser(description="Dorayaki - Smart Supermarket Shelf AI") parser.add_argument("--port", type=int, default=8080) parser.add_argument("--host", type=str, default="0.0.0.0") args = parser.parse_args() pipeline = ShelfRecognitionPipeline() app = create_fastapi_app(pipeline) import uvicorn print(f"\n[Dorayaki Server] Starting on http://localhost:{args.port} (or http://127.0.0.1:{args.port})") uvicorn.run(app, host=args.host, port=args.port, log_level="info") if __name__ == "__main__": main()