pyr0ball
2b4875d77c
- IntakeSession.observe_frame now catches ValueError from crop_to_book (empty-platform capture), cleans up any orphan wide-image file/dir, resets the strategy, and does not advance session state. The error is surfaced via a new last_error field. - /api/observe adds an "error" field to the response when a capture fires but fails, instead of an unhandled 500. - observe_frame only reports fired=True (and only runs capture logic) when state is WAITING_FRONT or WAITING_BACK, so a completed book left on the platform can no longer produce a no-op "fired" capture from an auto-strategy re-triggering. - observe_frame now holds a per-session threading.Lock for its whole body, since the frontend polls /api/observe every ~500ms without chaining requests and Catalogue.update_row() does a non-atomic read-all-then-rewrite-whole-file rewrite of catalogue.csv. Adds tests for empty-platform front/back captures and the BOOK_COMPLETE no-op-fire case, plus an API-level empty-platform test.
128 lines
4.1 KiB
Python
128 lines
4.1 KiB
Python
import numpy as np
|
|
from fastapi.testclient import TestClient
|
|
|
|
from bookmark.capture_strategies.manual_button import ManualButtonStrategy
|
|
from bookmark.catalogue import Catalogue
|
|
from bookmark.main import create_app
|
|
from bookmark.session import IntakeSession
|
|
from bookmark.sizing import MarkerRegion
|
|
|
|
|
|
def make_reference(size=100, color=200):
|
|
return np.full((size, size, 3), color, dtype=np.uint8)
|
|
|
|
|
|
def make_book_frame(size=100, ref_color=200, book_color=50, box=(20, 20, 40, 40)):
|
|
frame = np.full((size, size, 3), ref_color, dtype=np.uint8)
|
|
x, y, w, h = box
|
|
frame[y:y + h, x:x + w] = book_color
|
|
return frame
|
|
|
|
|
|
class FakeCamera:
|
|
def __init__(self, frame: np.ndarray) -> None:
|
|
self.frame = frame
|
|
|
|
def read_frame(self) -> np.ndarray:
|
|
return self.frame
|
|
|
|
|
|
class FailingCamera:
|
|
def __init__(self, error_message: str = "Camera not initialized") -> None:
|
|
self.error_message = error_message
|
|
|
|
def read_frame(self) -> np.ndarray:
|
|
raise RuntimeError(self.error_message)
|
|
|
|
|
|
def make_client(tmp_path):
|
|
catalogue = Catalogue(tmp_path / "catalogue.csv")
|
|
strategy = ManualButtonStrategy()
|
|
reference = make_reference()
|
|
markers = [MarkerRegion(name="small_boundary", bucket="small", x=0, y=0, width=10, height=10)]
|
|
session = IntakeSession(
|
|
catalogue=catalogue,
|
|
strategy=strategy,
|
|
reference_frame=reference,
|
|
markers=markers,
|
|
captures_dir=tmp_path / "captures",
|
|
)
|
|
camera = FakeCamera(reference)
|
|
app = create_app(session, camera)
|
|
return TestClient(app), session, camera
|
|
|
|
|
|
def test_session_state_starts_waiting_front(tmp_path):
|
|
client, _, _ = make_client(tmp_path)
|
|
response = client.get("/api/session-state")
|
|
assert response.status_code == 200
|
|
assert response.json()["state"] == "waiting_front"
|
|
|
|
|
|
def test_observe_without_trigger_does_not_fire(tmp_path):
|
|
client, _, _ = make_client(tmp_path)
|
|
response = client.post("/api/observe")
|
|
assert response.status_code == 200
|
|
assert response.json()["fired"] is False
|
|
|
|
|
|
def test_manual_trigger_then_observe_captures_front(tmp_path):
|
|
client, _, camera = make_client(tmp_path)
|
|
camera.frame = make_book_frame()
|
|
client.post("/api/manual-trigger")
|
|
response = client.post("/api/observe")
|
|
body = response.json()
|
|
assert body["fired"] is True
|
|
assert body["state"] == "waiting_back"
|
|
assert body["book_id"] is not None
|
|
|
|
|
|
def test_preview_returns_404_before_any_observe(tmp_path):
|
|
client, _, _ = make_client(tmp_path)
|
|
response = client.get("/api/preview.jpg")
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_preview_returns_jpeg_after_observe(tmp_path):
|
|
client, _, _ = make_client(tmp_path)
|
|
client.post("/api/observe")
|
|
response = client.get("/api/preview.jpg")
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"] == "image/jpeg"
|
|
|
|
|
|
def test_observe_on_empty_platform_does_not_500_and_reports_error(tmp_path):
|
|
client, session, camera = make_client(tmp_path)
|
|
# camera.frame is already the reference frame (empty platform) by default.
|
|
client.post("/api/manual-trigger")
|
|
response = client.post("/api/observe")
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["fired"] is False
|
|
assert body["state"] == "waiting_front"
|
|
assert "error" in body
|
|
assert session.catalogue.read_all() == []
|
|
|
|
|
|
def test_observe_returns_503_on_camera_failure(tmp_path):
|
|
catalogue = Catalogue(tmp_path / "catalogue.csv")
|
|
strategy = ManualButtonStrategy()
|
|
reference = make_reference()
|
|
markers = [MarkerRegion(name="small_boundary", bucket="small", x=0, y=0, width=10, height=10)]
|
|
session = IntakeSession(
|
|
catalogue=catalogue,
|
|
strategy=strategy,
|
|
reference_frame=reference,
|
|
markers=markers,
|
|
captures_dir=tmp_path / "captures",
|
|
)
|
|
camera = FailingCamera("Camera connection lost")
|
|
app = create_app(session, camera)
|
|
client = TestClient(app)
|
|
|
|
response = client.post("/api/observe")
|
|
assert response.status_code == 503
|
|
body = response.json()
|
|
assert body["error"] == "Camera unavailable"
|
|
assert "Camera connection lost" in body["detail"]
|