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

refactor: delete the legacy session model

Browse source 
Two session models shared one process. `/v1/*` and `/v1/stream` are the live
one; `/api/sessions*` and `/api/compare_live_vs_still` were the earlier model,
still constructed on every `create_app()` through `build_legacy_scans` even
though nothing called them. The web client and `stream_client.py` speak only
`/v1`.

Deleted: `observe.py` (ScanService/PerceptionService/PerceptionAdapter),
`yolo_adapter.py` (imported only by the legacy builder), `pipeline.py` and
`test_suite.py` (the old single-image CLI, reachable only from each other),
`test_observe.py` (its Session* tests), the eight legacy route handlers,
`build_legacy_scans` with its two facades, `create_fastapi_app`, and the
`scans` parameter on `create_app`.

`IdentityRankingTests` moved to `test_identity.py` before `test_observe.py` was
deleted: those five ranking rules are the live path's contract, not the legacy
model's. `test_api.py` keeps every `/v1` and WebSocket test.

Verified: 246 tests OK (2 skipped), and a route dump of a freshly built app
shows no `/api/*` path with all nine `/v1` routes and the WebSocket intact.
This commit is contained in:
kiyo9w 2026-09-28 17:47:50 +09:00
parent 004212a8cf
commit eca03e0c91
10 changed files with 54 additions and 1348 deletions
Download patch file Download diff file Expand all files Collapse all files

51
recognization/code/test_api.py
View file

@ -30,33 +30,6 @@ from test_live import FakeDetector, FakeInventory, FakeMatcher # noqa: E402
from fastapi.testclient import TestClient # noqa: E402
def make_scans(engine):
"""Legacy session model wired to the fake engine (the real factory is exercised
against the real model in test_real_model)."""
from observe import PerceptionAdapter, PerceptionService, ScanService
def detect(frame_bytes):
image = engine.decode_image(frame_bytes)
boxes = engine.detect(image)
return {"width": image.shape[1], "height": image.shape[0], "boxes": boxes,
"empty_slots": [], "quality": {"decoded": True}}
def classify(box):
scores = engine.score_crop(engine.crop(engine._last_image, box["box"]))
return scores
# keep the frame for the classify callback
def detect_with_frame(frame_bytes):
image = engine.decode_image(frame_bytes)
engine._last_image = image
boxes = engine.detect(image)
return {"width": image.shape[1], "height": image.shape[0], "boxes": boxes,
"empty_slots": [], "quality": {"decoded": True}}
adapter = PerceptionAdapter(detect_fn=detect_with_frame, classify_fn=classify,
model_version="fake", matcher_method="hybrid")
return ScanService(PerceptionService(adapter))
def make_app(*, token=None, max_sessions=4, detector=None):
engine = RecognitionEngine(
@ -64,7 +37,7 @@ def make_app(*, token=None, max_sessions=4, detector=None):
settings=EngineSettings(classify_workers=2),
)
settings = ServerSettings(token=token, max_sessions=max_sessions)
app = create_app(engine, settings, web_dir="/nonexistent", scans=make_scans(engine))
app = create_app(engine, settings, web_dir="/nonexistent")
return TestClient(app), engine
@ -549,27 +522,5 @@ class WebSocketProtocolTests(unittest.TestCase):
engine.shutdown()
class LegacyRouteTests(unittest.TestCase):
def test_legacy_sessions_still_respond(self):
"""The earlier session model stays mounted; it is backed by the engine now."""
client, engine = make_app()
try:
started = client.post("/api/sessions", json={"zone_id": "z1", "catalog_version": "v1"})
self.assertEqual(started.status_code, 200)
session_id = started.json()["session"]["session_id"]
frames = client.post(
f"/api/sessions/{session_id}/frames",
json={
"frame_id": "f1",
"sequence": 1,
"frame_bytes_b64": base64.b64encode(jpeg_bytes()).decode(),
},
)
self.assertEqual(frames.status_code, 200)
self.assertIn("observation", frames.json())
finally:
engine.shutdown()
if __name__ == "__main__":
unittest.main()