| from __future__ import annotations | |
| from pathlib import Path | |
| import httpx | |
| from app.config import Settings | |
| class CameraError(RuntimeError): | |
| pass | |
| def snapshot(settings: Settings, *, client: httpx.Client | None = None) -> Path: | |
| """HAL snapshot used by Autonomous Lamp. Returns the saved JPEG path.""" | |
| url = ( | |
| f"{settings.camera_url.rstrip('/')}/camera/snapshot" | |
| f"?save=true&width={settings.snapshot_width}&quality={settings.snapshot_quality}" | |
| ) | |
| own = client is None | |
| http = client or httpx.Client(timeout=30.0) | |
| try: | |
| response = http.get(url) | |
| response.raise_for_status() | |
| payload = response.json() | |
| except httpx.HTTPError as exc: | |
| raise CameraError(f"Lamp camera snapshot failed: {exc}") from exc | |
| finally: | |
| if own: | |
| http.close() | |
| path = payload.get("path") if isinstance(payload, dict) else None | |
| if not path: | |
| raise CameraError(f"snapshot JSON missing path: {payload!r}") | |
| return Path(path) | |
| def aim(settings: Settings, direction: str = "down", *, client: httpx.Client | None = None) -> None: | |
| """HAL servo aim. Call before snapshot when the paper is on the desk.""" | |
| url = f"{settings.camera_url.rstrip('/')}/servo/aim" | |
| own = client is None | |
| http = client or httpx.Client(timeout=15.0) | |
| try: | |
| response = http.post(url, json={"direction": direction}) | |
| response.raise_for_status() | |
| except httpx.HTTPError as exc: | |
| raise CameraError(f"Lamp servo aim failed: {exc}") from exc | |
| finally: | |
| if own: | |
| http.close() | |