tngo/product-recognition-demo
1
0
Fork 0
forked from vnguyen/product-recognition-demo
Code Pull requests Activity

fix: keep catalog IO off the live event loop

Browse source
This commit is contained in:
kiyo9w 2026-09-28 12:57:21 +09:00
parent 62166f77b8
commit 2894979a35
1 changed files with 93 additions and 57 deletions
Download patch file Download diff file Expand all files Collapse all files

150
recognization/code/app.py
View file

@ -309,42 +309,53 @@ def create_app(
# ---------------------------------------------------------------- health # ---------------------------------------------------------------- health
@app.get("/v1/health") @app.get("/v1/health")
async def health() -> Dict[str, Any]: async def health() -> Dict[str, Any]:
return { loop = asyncio.get_running_loop()
"status": "ok",
"model": { def _health() -> Dict[str, Any]:
"version": engine.model_version(), gallery = (
"device": getattr(engine.settings, "device", None) or "auto", engine.identity_gallery()
"imgsz": getattr(engine.settings, "imgsz", 640), if getattr(engine.settings, "identity_backend", "") == "gallery"
"warm": True, else None
}, )
"catalog": {"version": engine.catalog_version(), "sku_count": engine.sku_count()}, return {
"identity": { "status": "ok",
"backend": getattr(engine.settings, "identity_backend", "matcher"), "model": {
"gallery_views": ( "version": engine.model_version(),
engine.identity_gallery().view_count() "device": getattr(engine.settings, "device", None) or "auto",
if getattr(engine.settings, "identity_backend", "") == "gallery" and engine.identity_gallery() "imgsz": getattr(engine.settings, "imgsz", 640),
else None "warm": True,
), },
}, "catalog": {"version": engine.catalog_version(), "sku_count": engine.sku_count()},
"ocr": { "identity": {
# Without this an operator could not tell a broken OCR from a shelf "backend": getattr(engine.settings, "identity_backend", "matcher"),
# with no readable tags. "gallery_views": gallery.view_count() if gallery is not None else None,
"engine": getattr(engine.ocr_engine, "name", "none"), },
"available": bool(getattr(engine.ocr_engine, "available", lambda: False)()), "ocr": {
}, # Without this an operator could not tell a broken OCR from a shelf
"recording": recorder.stats() if recorder is not None else None, # with no readable tags.
"sessions_active": len(sessions), "engine": getattr(engine.ocr_engine, "name", "none"),
"uptime_s": round(engine.uptime_s(), 1), "available": bool(getattr(engine.ocr_engine, "available", lambda: False)()),
} },
"sessions_active": sum(session is not None for session in sessions.values()),
"uptime_s": round(engine.uptime_s(), 1),
}
return await loop.run_in_executor(None, _health)
# ---------------------------------------------------------------- catalog # ---------------------------------------------------------------- catalog
@app.get("/v1/catalog", dependencies=[Depends(authorize)]) @app.get("/v1/catalog", dependencies=[Depends(authorize)])
async def catalog() -> Dict[str, Any]: async def catalog() -> Dict[str, Any]:
return {"catalog_version": engine.catalog_version(), "products": engine.catalog()} loop = asyncio.get_running_loop()
def _catalog() -> Dict[str, Any]:
return {"catalog_version": engine.catalog_version(), "products": engine.catalog()}
return await loop.run_in_executor(None, _catalog)
@app.get("/v1/catalog/{sku_id}/image", dependencies=[Depends(authorize)]) @app.get("/v1/catalog/{sku_id}/image", dependencies=[Depends(authorize)])
async def catalog_image(sku_id: str) -> Response: async def catalog_image(sku_id: str) -> Response:
path = engine.catalog_image_path(sku_id) loop = asyncio.get_running_loop()
path = await loop.run_in_executor(None, engine.catalog_image_path, sku_id)
if not path: if not path:
raise _error("NOT_FOUND", f"no catalog image for '{sku_id}'", 404) raise _error("NOT_FOUND", f"no catalog image for '{sku_id}'", 404)
return FileResponse(path) return FileResponse(path)
@ -353,7 +364,8 @@ def create_app(
async def update_stock(sku_id: str, body: StockUpdateBody) -> Dict[str, Any]: async def update_stock(sku_id: str, body: StockUpdateBody) -> Dict[str, Any]:
if body.delta == 0: if body.delta == 0:
raise _error("VALIDATION_ERROR", "delta must not be zero", 422) raise _error("VALIDATION_ERROR", "delta must not be zero", 422)
product = engine.update_stock(sku_id, body.delta) loop = asyncio.get_running_loop()
product = await loop.run_in_executor(None, engine.update_stock, sku_id, body.delta)
if product is None: if product is None:
raise _error("NOT_FOUND", f"unknown sku '{sku_id}'", 404) raise _error("NOT_FOUND", f"unknown sku '{sku_id}'", 404)
return {"status": "success", "product": product} return {"status": "success", "product": product}
@ -378,7 +390,7 @@ def create_app(
@app.get("/v1/samples/{sample_id}", dependencies=[Depends(authorize)]) @app.get("/v1/samples/{sample_id}", dependencies=[Depends(authorize)])
async def get_sample(sample_id: str) -> Response: async def get_sample(sample_id: str) -> Response:
item = samples.get(sample_id) item = samples.get(sample_id)
if not item or not os.path.exists(item["path"]): if not item or not await asyncio.get_running_loop().run_in_executor(None, os.path.exists, item["path"]):
raise _error("NOT_FOUND", f"no sample '{sample_id}'", 404) raise _error("NOT_FOUND", f"no sample '{sample_id}'", 404)
return FileResponse(item["path"]) return FileResponse(item["path"])
@ -640,11 +652,16 @@ def create_app(
if settings.token and header and not bearer_matches(header, settings.token): if settings.token and header and not bearer_matches(header, settings.token):
await websocket.close(code=1008) await websocket.close(code=1008)
return return
loop = asyncio.get_running_loop()
async with sessions_lock: async with sessions_lock:
if len(sessions) >= settings.max_sessions: if len(sessions) >= settings.max_sessions:
await websocket.close(code=1013) await websocket.close(code=1013)
return return
session_id = str(uuid.uuid4()) session_id = str(uuid.uuid4())
sessions[session_id] = None # reserve before any disk work
session = None
try:
model_version = await loop.run_in_executor(None, engine.model_version)
session = LiveSession( session = LiveSession(
session_id, session_id,
engine, engine,
@ -656,33 +673,46 @@ def create_app(
tags=settings.live_tags, tags=settings.live_tags,
), ),
) )
session.refresh_catalog() await loop.run_in_executor(None, session.refresh_catalog)
catalog_version = session._catalog_version
if settings.live_tags: if settings.live_tags:
session.enable_tags() await loop.run_in_executor(None, session.enable_tags)
sessions[session_id] = session async with sessions_lock:
await websocket.accept() sessions[session_id] = session
send_lock = asyncio.Lock() except BaseException:
loop = asyncio.get_running_loop() if session is not None:
session.close()
async with sessions_lock:
sessions.pop(session_id, None)
raise
try:
await websocket.accept()
send_lock = asyncio.Lock()
async def send(payload: Dict[str, Any]) -> None: async def send(payload: Dict[str, Any]) -> None:
async with send_lock: async with send_lock:
await websocket.send_text(json.dumps(payload)) await websocket.send_text(json.dumps(payload))
await send( await send(
{ {
"type": "hello", "type": "hello",
"protocol": 1, "protocol": 1,
"session_id": session_id, "session_id": session_id,
"catalog_version": engine.catalog_version(), "catalog_version": catalog_version,
"model": { "model": {
"version": engine.model_version(), "version": model_version,
"device": getattr(engine.settings, "device", None) or "auto", "device": getattr(engine.settings, "device", None) or "auto",
"imgsz": getattr(engine.settings, "imgsz", 640), "imgsz": getattr(engine.settings, "imgsz", 640),
}, },
"config": session.config(), "config": session.config(),
"limits": {"max_frame_bytes": MAX_FRAME_BYTES}, "limits": {"max_frame_bytes": MAX_FRAME_BYTES},
} }
) )
except BaseException:
session.close()
async with sessions_lock:
sessions.pop(session_id, None)
raise
async def processing_loop() -> None: async def processing_loop() -> None:
while not session.closed: while not session.closed:
@ -851,7 +881,14 @@ def create_app(
# Base64 expands ~4/3. Reject the text before decoding it. # Base64 expands ~4/3. Reject the text before decoding it.
if len(raw) > (MAX_FRAME_BYTES * 4) // 3 + 8: if len(raw) > (MAX_FRAME_BYTES * 4) // 3 + 8:
raise _error("PAYLOAD_TOO_LARGE", f"frame larger than {MAX_FRAME_BYTES} bytes", 413) raise _error("PAYLOAD_TOO_LARGE", f"frame larger than {MAX_FRAME_BYTES} bytes", 413)
frame_bytes = base64.b64decode(raw) if raw else b"" loop = asyncio.get_running_loop()
try:
frame_bytes = (
await loop.run_in_executor(None, lambda: base64.b64decode(raw, validate=True))
if raw else b""
)
except (ValueError, base64.binascii.Error):
raise _error("BAD_REQUEST", "frame_bytes_b64 is not valid base64", 400)
if len(frame_bytes) > MAX_FRAME_BYTES: if len(frame_bytes) > MAX_FRAME_BYTES:
raise _error("PAYLOAD_TOO_LARGE", f"frame larger than {MAX_FRAME_BYTES} bytes", 413) raise _error("PAYLOAD_TOO_LARGE", f"frame larger than {MAX_FRAME_BYTES} bytes", 413)
request = FrameRequest( request = FrameRequest(
@ -868,7 +905,6 @@ def create_app(
) )
if not request.frame_id: if not request.frame_id:
raise _error("BAD_REQUEST", "frame_id required", 400) raise _error("BAD_REQUEST", "frame_id required", 400)
loop = asyncio.get_running_loop()
observation = await loop.run_in_executor(None, scans.observe_frame, request) observation = await loop.run_in_executor(None, scans.observe_frame, request)
return {"status": "success", "observation": observation.as_dict(), "session": session.snapshot()} return {"status": "success", "observation": observation.as_dict(), "session": session.snapshot()}
except ObserveError as exc: except ObserveError as exc: