422 lines
14 KiB
Python
422 lines
14 KiB
Python
"""Zones tab — zone → place → linked-faces management + "go here" destination.
|
|
|
|
Hierarchy (replaces the old flat places):
|
|
Zone (name + description)
|
|
└─ Place (name + description + optional reference photos + linked face ids)
|
|
|
|
Routes live under /api/zones. Toggle + CRUD changes write
|
|
data/.recognition_state.json (the SAME file faces use); the Gemini child polls
|
|
it at 1 Hz and re-primes / announces mid-session. The "go here" endpoints set a
|
|
navigation target the robot will head to once N2 locomotion is wired — for now
|
|
they just record the target and feed Gemini the place's reference.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, File, HTTPException, Query, UploadFile
|
|
from fastapi.responses import FileResponse, StreamingResponse
|
|
from pydantic import BaseModel
|
|
|
|
from Project.Sanad.config import BASE_DIR
|
|
from Project.Sanad.core.logger import get_logger
|
|
from Project.Sanad.dashboard.routes._safe_io import check_upload_size
|
|
from Project.Sanad.vision import recognition_state
|
|
|
|
log = get_logger("zones_routes")
|
|
|
|
router = APIRouter()
|
|
|
|
STATE_PATH = BASE_DIR / "data" / ".recognition_state.json"
|
|
|
|
|
|
# ── lazy subsystem accessors ────────────────────────────────
|
|
|
|
def _get_camera():
|
|
try:
|
|
from Project.Sanad.main import camera # type: ignore
|
|
return camera
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _get_zone_gallery():
|
|
try:
|
|
from Project.Sanad.main import zone_gallery # type: ignore
|
|
return zone_gallery
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _get_face_gallery():
|
|
try:
|
|
from Project.Sanad.main import gallery # type: ignore
|
|
return gallery
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _require_zones():
|
|
g = _get_zone_gallery()
|
|
if g is None:
|
|
raise HTTPException(503, "Zone gallery subsystem unavailable.")
|
|
return g
|
|
|
|
|
|
def _bump_zones_version() -> int:
|
|
cur = recognition_state.read(STATE_PATH)
|
|
v = cur.zones_version + 1
|
|
recognition_state.mutate(STATE_PATH, zones_version=v)
|
|
return v
|
|
|
|
|
|
def _validate_image(content: bytes, filename: str | None = None) -> None:
|
|
check_upload_size(content)
|
|
if len(content) < 16:
|
|
raise HTTPException(400, "Image too small / empty.")
|
|
if not (content[:3] == b"\xff\xd8\xff" or content[:8] == b"\x89PNG\r\n\x1a\n"):
|
|
raise HTTPException(400, f"Only JPEG/PNG accepted (got {filename or 'unknown'}).")
|
|
|
|
|
|
def _safe_photo_name(name: str) -> None:
|
|
if "/" in name or ".." in name or "\x00" in name:
|
|
raise HTTPException(400, "Invalid photo name.")
|
|
|
|
|
|
def _resolve_faces(face_ids: list[int]) -> list[dict]:
|
|
"""Turn linked face ids into [{id, name}] using the face gallery."""
|
|
g = _get_face_gallery()
|
|
out = []
|
|
for fid in face_ids:
|
|
name = None
|
|
if g is not None:
|
|
try:
|
|
e = g.get(fid)
|
|
name = e.name if e else None
|
|
except Exception:
|
|
name = None
|
|
out.append({"id": fid, "name": name})
|
|
return out
|
|
|
|
|
|
def _place_to_dict(p) -> dict:
|
|
d = p.to_dict()
|
|
d["faces"] = _resolve_faces(p.face_ids)
|
|
return d
|
|
|
|
|
|
def _zone_to_dict(z) -> dict:
|
|
return {
|
|
"id": z.id, "name": z.name, "description": z.description,
|
|
"added_at": z.added_at,
|
|
"places": [_place_to_dict(p) for p in z.places],
|
|
}
|
|
|
|
|
|
def _nav_target_dict(st, gallery) -> Optional[dict]:
|
|
zid, pid = st.nav_target_zone_id, st.nav_target_place_id
|
|
if not zid or not pid:
|
|
return None
|
|
zone_name = place_name = None
|
|
if gallery is not None:
|
|
try:
|
|
z = gallery.get_zone(zid)
|
|
zone_name = z.name if z else None
|
|
p = gallery.get_place(zid, pid)
|
|
place_name = p.name if p else None
|
|
except Exception:
|
|
pass
|
|
return {"zone_id": zid, "place_id": pid,
|
|
"zone_name": zone_name, "place_name": place_name}
|
|
|
|
|
|
# ── state + toggle ──────────────────────────────────────────
|
|
|
|
@router.get("/state")
|
|
async def get_state():
|
|
st = recognition_state.read(STATE_PATH)
|
|
g = _get_zone_gallery()
|
|
zones_count = places_count = 0
|
|
if g is not None:
|
|
try:
|
|
zones = g.list_zones()
|
|
zones_count = len(zones)
|
|
places_count = sum(len(z.places) for z in zones)
|
|
except Exception:
|
|
pass
|
|
return {
|
|
"zone_rec_enabled": st.zone_rec_enabled,
|
|
"zones_version": st.zones_version,
|
|
"zones_count": zones_count,
|
|
"places_count": places_count,
|
|
"nav_target": _nav_target_dict(st, g),
|
|
}
|
|
|
|
|
|
@router.post("/zone-rec")
|
|
async def set_zone_rec(on: bool = Query(...)):
|
|
"""Enable / disable the robot's knowledge of zones & places (hot)."""
|
|
st = recognition_state.mutate(STATE_PATH, zone_rec_enabled=bool(on))
|
|
log.info("zone recognition %s", "ON" if on else "OFF")
|
|
return {"ok": True, "zone_rec_enabled": st.zone_rec_enabled}
|
|
|
|
|
|
@router.post("/sync")
|
|
async def sync_zones():
|
|
v = _bump_zones_version()
|
|
log.info("zones sync requested → v.%d", v)
|
|
return {"ok": True, "zones_version": v}
|
|
|
|
|
|
# ── zones CRUD ──────────────────────────────────────────────
|
|
|
|
class NamePayload(BaseModel):
|
|
name: Optional[str] = None
|
|
|
|
|
|
class DescribePayload(BaseModel):
|
|
description: Optional[str] = None
|
|
|
|
|
|
class FacesPayload(BaseModel):
|
|
face_ids: list[int] = []
|
|
|
|
|
|
@router.get("")
|
|
async def list_zones():
|
|
g = _require_zones()
|
|
zones = g.list_zones()
|
|
return {"zones": [_zone_to_dict(z) for z in zones], "total": len(zones)}
|
|
|
|
|
|
@router.post("/create")
|
|
async def create_zone(name: Optional[str] = Query(default=None),
|
|
description: Optional[str] = Query(default=None)):
|
|
g = _require_zones()
|
|
if not (name or "").strip() and not (description or "").strip():
|
|
raise HTTPException(400, "A zone needs at least a name or a description.")
|
|
z = g.create_zone(name=name, description=description)
|
|
_bump_zones_version()
|
|
return {"ok": True, "zone": _zone_to_dict(z)}
|
|
|
|
|
|
@router.post("/{zone_id}/rename")
|
|
async def rename_zone(zone_id: int, payload: NamePayload):
|
|
g = _require_zones()
|
|
try:
|
|
g.rename_zone(zone_id, payload.name)
|
|
except FileNotFoundError as exc:
|
|
raise HTTPException(404, str(exc))
|
|
_bump_zones_version()
|
|
return {"ok": True, "zone": _zone_to_dict(g.get_zone(zone_id))}
|
|
|
|
|
|
@router.post("/{zone_id}/describe")
|
|
async def describe_zone(zone_id: int, payload: DescribePayload):
|
|
g = _require_zones()
|
|
try:
|
|
g.describe_zone(zone_id, payload.description)
|
|
except FileNotFoundError as exc:
|
|
raise HTTPException(404, str(exc))
|
|
_bump_zones_version()
|
|
return {"ok": True, "zone": _zone_to_dict(g.get_zone(zone_id))}
|
|
|
|
|
|
@router.delete("/{zone_id}")
|
|
async def delete_zone(zone_id: int):
|
|
g = _require_zones()
|
|
try:
|
|
g.delete_zone(zone_id)
|
|
except FileNotFoundError as exc:
|
|
raise HTTPException(404, str(exc))
|
|
# If the active destination was inside this zone, clear it.
|
|
st = recognition_state.read(STATE_PATH)
|
|
if st.nav_target_zone_id == zone_id:
|
|
recognition_state.mutate(STATE_PATH, nav_target_zone_id=0, nav_target_place_id=0)
|
|
_bump_zones_version()
|
|
return {"ok": True, "deleted": zone_id}
|
|
|
|
|
|
# ── places CRUD (within a zone) ─────────────────────────────
|
|
|
|
@router.post("/{zone_id}/places/create")
|
|
async def create_place(
|
|
zone_id: int,
|
|
name: Optional[str] = Query(default=None),
|
|
description: Optional[str] = Query(default=None),
|
|
face_ids: list[int] = Query(default=[]),
|
|
files: Optional[list[UploadFile]] = File(default=None),
|
|
):
|
|
g = _require_zones()
|
|
if g.get_zone(zone_id) is None:
|
|
raise HTTPException(404, f"zone_{zone_id} not found")
|
|
if not (name or "").strip() and not (description or "").strip():
|
|
raise HTTPException(400, "A place needs at least a name or a description.")
|
|
image_bytes: list[bytes] = []
|
|
for f in (files or []):
|
|
content = await f.read()
|
|
if not content:
|
|
continue
|
|
_validate_image(content, f.filename)
|
|
image_bytes.append(content)
|
|
p = g.create_place(zone_id, name=name, description=description,
|
|
face_ids=face_ids, image_bytes_list=image_bytes or None)
|
|
_bump_zones_version()
|
|
return {"ok": True, "place": _place_to_dict(p)}
|
|
|
|
|
|
@router.post("/{zone_id}/places/{place_id}/rename")
|
|
async def rename_place(zone_id: int, place_id: int, payload: NamePayload):
|
|
g = _require_zones()
|
|
try:
|
|
g.rename_place(zone_id, place_id, payload.name)
|
|
except FileNotFoundError as exc:
|
|
raise HTTPException(404, str(exc))
|
|
_bump_zones_version()
|
|
return {"ok": True, "place": _place_to_dict(g.get_place(zone_id, place_id))}
|
|
|
|
|
|
@router.post("/{zone_id}/places/{place_id}/describe")
|
|
async def describe_place(zone_id: int, place_id: int, payload: DescribePayload):
|
|
g = _require_zones()
|
|
try:
|
|
g.describe_place(zone_id, place_id, payload.description)
|
|
except FileNotFoundError as exc:
|
|
raise HTTPException(404, str(exc))
|
|
_bump_zones_version()
|
|
return {"ok": True, "place": _place_to_dict(g.get_place(zone_id, place_id))}
|
|
|
|
|
|
@router.post("/{zone_id}/places/{place_id}/faces")
|
|
async def set_place_faces(zone_id: int, place_id: int, payload: FacesPayload):
|
|
"""Replace the set of saved faces linked to this place."""
|
|
g = _require_zones()
|
|
try:
|
|
g.set_place_faces(zone_id, place_id, payload.face_ids)
|
|
except FileNotFoundError as exc:
|
|
raise HTTPException(404, str(exc))
|
|
_bump_zones_version()
|
|
return {"ok": True, "place": _place_to_dict(g.get_place(zone_id, place_id))}
|
|
|
|
|
|
@router.post("/{zone_id}/places/{place_id}/capture")
|
|
async def capture_to_place(zone_id: int, place_id: int):
|
|
g = _require_zones()
|
|
cam = _get_camera()
|
|
if cam is None or not cam.is_running():
|
|
raise HTTPException(409, "Camera is not running. Toggle Vision ON first.")
|
|
jpeg = cam.get_fresh_frame(max_age_s=0.5, timeout_s=1.5)
|
|
if not jpeg:
|
|
raise HTTPException(409, "Camera has no frame yet.")
|
|
try:
|
|
fname = g.add_photo(zone_id, place_id, jpeg)
|
|
except FileNotFoundError as exc:
|
|
raise HTTPException(404, str(exc))
|
|
_bump_zones_version()
|
|
return {"ok": True, "added": fname, "place": _place_to_dict(g.get_place(zone_id, place_id))}
|
|
|
|
|
|
@router.post("/{zone_id}/places/{place_id}/upload")
|
|
async def upload_to_place(zone_id: int, place_id: int,
|
|
files: list[UploadFile] = File(...)):
|
|
g = _require_zones()
|
|
if g.get_place(zone_id, place_id) is None:
|
|
raise HTTPException(404, f"zone_{zone_id}/place_{place_id} not found")
|
|
added: list[str] = []
|
|
for f in files:
|
|
content = await f.read()
|
|
_validate_image(content, f.filename)
|
|
try:
|
|
added.append(g.add_photo(zone_id, place_id, content))
|
|
except FileNotFoundError as exc:
|
|
raise HTTPException(404, str(exc))
|
|
_bump_zones_version()
|
|
return {"ok": True, "added": added, "place": _place_to_dict(g.get_place(zone_id, place_id))}
|
|
|
|
|
|
@router.delete("/{zone_id}/places/{place_id}")
|
|
async def delete_place(zone_id: int, place_id: int):
|
|
g = _require_zones()
|
|
try:
|
|
g.delete_place(zone_id, place_id)
|
|
except FileNotFoundError as exc:
|
|
raise HTTPException(404, str(exc))
|
|
st = recognition_state.read(STATE_PATH)
|
|
if st.nav_target_zone_id == zone_id and st.nav_target_place_id == place_id:
|
|
recognition_state.mutate(STATE_PATH, nav_target_zone_id=0, nav_target_place_id=0)
|
|
_bump_zones_version()
|
|
return {"ok": True, "deleted": place_id}
|
|
|
|
|
|
@router.delete("/{zone_id}/places/{place_id}/photo/{photo_name}")
|
|
async def delete_place_photo(zone_id: int, place_id: int, photo_name: str):
|
|
g = _require_zones()
|
|
_safe_photo_name(photo_name)
|
|
try:
|
|
g.delete_photo(zone_id, place_id, photo_name)
|
|
except FileNotFoundError as exc:
|
|
raise HTTPException(404, str(exc))
|
|
_bump_zones_version()
|
|
return {"ok": True, "deleted": photo_name}
|
|
|
|
|
|
@router.get("/{zone_id}/places/{place_id}/photo/{photo_name}")
|
|
async def get_place_photo(zone_id: int, place_id: int, photo_name: str,
|
|
download: int = Query(default=0)):
|
|
g = _require_zones()
|
|
_safe_photo_name(photo_name)
|
|
path = g.get_photo(zone_id, place_id, photo_name)
|
|
if path is None:
|
|
raise HTTPException(404, "Photo not found.")
|
|
media = "image/png" if path.suffix.lower() == ".png" else "image/jpeg"
|
|
headers = {}
|
|
if download:
|
|
headers["Content-Disposition"] = (
|
|
f'attachment; filename="zone_{zone_id}_place_{place_id}_{photo_name}"')
|
|
return FileResponse(path, media_type=media, headers=headers)
|
|
|
|
|
|
@router.get("/{zone_id}/places/{place_id}/download.zip")
|
|
async def download_place_zip(zone_id: int, place_id: int):
|
|
g = _require_zones()
|
|
try:
|
|
data = g.zip_place(zone_id, place_id)
|
|
except FileNotFoundError as exc:
|
|
raise HTTPException(404, str(exc))
|
|
return StreamingResponse(
|
|
io.BytesIO(data), media_type="application/zip",
|
|
headers={
|
|
"Content-Disposition": f'attachment; filename="zone_{zone_id}_place_{place_id}.zip"',
|
|
"Content-Length": str(len(data)),
|
|
},
|
|
)
|
|
|
|
|
|
# ── "go here" navigation target ─────────────────────────────
|
|
|
|
@router.post("/{zone_id}/places/{place_id}/go")
|
|
async def go_to_place(zone_id: int, place_id: int):
|
|
"""Set this place as the active destination. Records the target and lets
|
|
the Gemini child pick it up (reference photo + goal). Actual robot motion
|
|
is wired by N2 locomotion — until then this just establishes the goal."""
|
|
g = _require_zones()
|
|
p = g.get_place(zone_id, place_id)
|
|
if p is None:
|
|
raise HTTPException(404, f"zone_{zone_id}/place_{place_id} not found")
|
|
recognition_state.mutate(STATE_PATH,
|
|
nav_target_zone_id=zone_id,
|
|
nav_target_place_id=place_id)
|
|
log.info("nav target set → zone_%d/place_%d (%s)", zone_id, place_id,
|
|
p.name or "(unnamed)")
|
|
return {"ok": True, "nav_target": {"zone_id": zone_id, "place_id": place_id,
|
|
"place_name": p.name}}
|
|
|
|
|
|
@router.post("/nav/clear")
|
|
async def clear_nav_target():
|
|
recognition_state.mutate(STATE_PATH, nav_target_zone_id=0, nav_target_place_id=0)
|
|
log.info("nav target cleared")
|
|
return {"ok": True, "nav_target": None}
|