392 lines
16 KiB
Python
392 lines
16 KiB
Python
"""pantry-vision — camera-driven grocery cataloguing for the kitchen display, from
|
|
docs/project-plan.md Phase 17.
|
|
|
|
The workflow this exists for: come home, put down the shopping bag, hold one item up
|
|
to the kitchen display's camera, the system proposes what it is and how long it keeps,
|
|
you confirm (editing anything it got wrong) before it's written anywhere, then put the
|
|
item away. The display also shows the resulting inventory sorted by what expires
|
|
soonest, and Grocy's recipes.
|
|
|
|
SECURITY BOUNDARY, same shape as admin-canvas/server.py's, with one deliberate
|
|
difference: this service DOES get a published port (see the README and
|
|
setup-container-host.sh's PANTRY_VISION_PORT), because unlike admin-canvas — which is
|
|
only ever called by Home Assistant on the compose network — this one is called
|
|
directly by the kitchen-display kiosk, a separate physical device on the LAN. Every
|
|
request still requires the bearer token below; an unset token fails closed (rejects
|
|
everything), never "auth optional".
|
|
|
|
Four endpoints, all token-gated:
|
|
- POST /identify raw image bytes -> Ollama vision model -> a *proposal*
|
|
- POST /confirm JSON, human-reviewed/edited -> writes to Grocy stock
|
|
- GET /inventory proxies Grocy stock, sorted by soonest-expiring first
|
|
- GET /recipes proxies Grocy recipes (+ fulfillment, best-effort)
|
|
|
|
/identify NEVER writes anything by itself — same "propose, never auto-commit" rule
|
|
this project already applies to identity-merge confirmation (see the Identity store
|
|
row in docs/project-plan.md §2): the vision model's guess is shown to the person
|
|
standing at the display, who can edit the name/category/date/location before /confirm
|
|
is ever called. A wrong guess costs a tap to fix, not a wrong fact silently written
|
|
into the household's inventory.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from datetime import date, datetime
|
|
from http import HTTPStatus
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import urlsplit
|
|
|
|
LOG = logging.getLogger("pantry-vision")
|
|
|
|
TOKEN = os.environ.get("PANTRY_VISION_TOKEN", "")
|
|
|
|
OLLAMA_HOST = os.environ.get("OLLAMA_HOST", "http://llm-host:11434").rstrip("/")
|
|
# TBD/unverified — see README.md's "Pick a vision model" section. Not every Ollama
|
|
# model can see images; this must be one that can (llava, qwen2.5vl, etc.) and must
|
|
# actually be pulled on the LLM host (`ollama pull <name>`) before /identify works.
|
|
OLLAMA_VISION_MODEL = os.environ.get("OLLAMA_VISION_MODEL", "llava")
|
|
OLLAMA_TIMEOUT = float(os.environ.get("OLLAMA_TIMEOUT", "120"))
|
|
|
|
GROCY_URL = os.environ.get("GROCY_URL", "http://grocy:80").rstrip("/")
|
|
GROCY_API_KEY = os.environ.get("GROCY_API_KEY", "")
|
|
# Fresh-install defaults ("Default" location, "Piece" quantity unit) — VERIFY against
|
|
# your own Grocy instance (Settings -> Locations / Quantity units) and override if
|
|
# they differ. Same "don't build against a guess" flag as every other cross-service ID
|
|
# in this project (ADMIN_WEB_URL placeholders, weather_entity_id, etc.).
|
|
GROCY_DEFAULT_LOCATION_ID = os.environ.get("GROCY_DEFAULT_LOCATION_ID", "1")
|
|
GROCY_DEFAULT_QU_ID = os.environ.get("GROCY_DEFAULT_QU_ID", "2")
|
|
|
|
MAX_IMAGE_BYTES = int(os.environ.get("PANTRY_VISION_MAX_IMAGE_MB", "15")) * 1024 * 1024
|
|
MAX_JSON_BYTES = 64 * 1024
|
|
|
|
_IDENTIFY_PROMPT = """You are looking at a single grocery item held up to a kitchen \
|
|
camera, on a plain background. Identify it and answer ONLY with a JSON object, no \
|
|
other text, in exactly this shape:
|
|
{"name": "<short product name>", "category": "<produce|dairy|meat|frozen|pantry|bakery|beverage|other>", \
|
|
"estimated_shelf_life_days": <integer, typical days from today until it goes bad, \
|
|
assuming normal fridge/pantry storage for its category>, "confidence": "<high|medium|low>"}
|
|
If you cannot identify the item, still return the JSON shape with your best guess and \
|
|
"confidence": "low"."""
|
|
|
|
_JSON_OBJECT_RE = re.compile(r"\{.*\}", re.DOTALL)
|
|
|
|
|
|
def _http_json(method: str, url: str, payload: dict | None = None, headers: dict | None = None,
|
|
timeout: float = 30.0) -> dict:
|
|
data = json.dumps(payload).encode("utf-8") if payload is not None else None
|
|
req = urllib.request.Request(url, data=data, method=method)
|
|
req.add_header("Content-Type", "application/json")
|
|
for key, value in (headers or {}).items():
|
|
req.add_header(key, value)
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
body = resp.read()
|
|
return json.loads(body) if body else {}
|
|
|
|
|
|
def _grocy_headers() -> dict:
|
|
return {"GROCY-API-KEY": GROCY_API_KEY}
|
|
|
|
|
|
def _grocy_get(path: str):
|
|
return _http_json("GET", f"{GROCY_URL}{path}", headers=_grocy_headers(), timeout=15)
|
|
|
|
|
|
def _grocy_post(path: str, payload: dict):
|
|
return _http_json("POST", f"{GROCY_URL}{path}", payload=payload, headers=_grocy_headers(), timeout=15)
|
|
|
|
|
|
def _identify_via_ollama(image_bytes: bytes) -> dict:
|
|
"""Calls the vision model and returns a best-effort proposal — degrades to a
|
|
low-confidence placeholder rather than raising, mirroring digest-engine's
|
|
llm_client._fallback_document: a broken/blank result is worse than one flagged
|
|
as unreliable that a human has to review anyway (which they always do — see the
|
|
module docstring, /identify never writes anything by itself).
|
|
"""
|
|
payload = {
|
|
"model": OLLAMA_VISION_MODEL,
|
|
"prompt": _IDENTIFY_PROMPT,
|
|
"images": [base64.b64encode(image_bytes).decode("ascii")],
|
|
"stream": False,
|
|
"options": {"temperature": 0.2},
|
|
}
|
|
try:
|
|
result = _http_json("POST", f"{OLLAMA_HOST}/api/generate", payload=payload, timeout=OLLAMA_TIMEOUT)
|
|
text = result.get("response", "")
|
|
except Exception:
|
|
LOG.warning("pantry-vision: Ollama call failed", exc_info=True)
|
|
return {
|
|
"name": "Unknown item",
|
|
"category": "other",
|
|
"estimated_shelf_life_days": 7,
|
|
"confidence": "low",
|
|
"note": f"Could not reach the vision model at {OLLAMA_HOST} — filled in manually.",
|
|
}
|
|
|
|
match = _JSON_OBJECT_RE.search(text)
|
|
if match:
|
|
try:
|
|
proposal = json.loads(match.group(0))
|
|
proposal.setdefault("name", "Unknown item")
|
|
proposal.setdefault("category", "other")
|
|
proposal.setdefault("estimated_shelf_life_days", 7)
|
|
proposal.setdefault("confidence", "low")
|
|
return proposal
|
|
except ValueError:
|
|
pass
|
|
|
|
LOG.warning("pantry-vision: could not parse a JSON proposal out of the model's response")
|
|
return {
|
|
"name": "Unknown item",
|
|
"category": "other",
|
|
"estimated_shelf_life_days": 7,
|
|
"confidence": "low",
|
|
"note": "The vision model's response wasn't valid JSON — filled in manually.",
|
|
"raw_response": text[:500],
|
|
}
|
|
|
|
|
|
def _find_or_create_product(name: str) -> int:
|
|
"""Best-effort product lookup/create against Grocy's object API.
|
|
|
|
VERIFY against a real Grocy instance — this assumes GET /api/objects/products
|
|
returns a flat list of {id, name, ...} and that POST to the same path with just
|
|
name/location_id/qu_id_purchase/qu_id_stock is enough to create a minimal product.
|
|
Grocy's actual required-field set depends on its own settings and has not been
|
|
checked against a live instance; see README.md's verification list.
|
|
"""
|
|
products = _grocy_get("/api/objects/products")
|
|
needle = name.strip().lower()
|
|
for product in products:
|
|
if str(product.get("name", "")).strip().lower() == needle:
|
|
return int(product["id"])
|
|
|
|
created = _grocy_post(
|
|
"/api/objects/products",
|
|
{
|
|
"name": name,
|
|
"location_id": int(GROCY_DEFAULT_LOCATION_ID),
|
|
"qu_id_purchase": int(GROCY_DEFAULT_QU_ID),
|
|
"qu_id_stock": int(GROCY_DEFAULT_QU_ID),
|
|
},
|
|
)
|
|
return int(created["created_object_id"])
|
|
|
|
|
|
def _add_to_stock(product_id: int, amount: float, best_before_date: str) -> dict:
|
|
return _grocy_post(
|
|
f"/api/stock/products/{product_id}/add",
|
|
{"amount": amount, "best_before_date": best_before_date, "transaction_type": "purchase"},
|
|
)
|
|
|
|
|
|
def _days_left(best_before: str | None) -> int | None:
|
|
if not best_before:
|
|
return None
|
|
try:
|
|
due = datetime.strptime(best_before, "%Y-%m-%d").date()
|
|
except ValueError:
|
|
return None
|
|
return (due - date.today()).days
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
server_version = "pantry-vision/1"
|
|
|
|
def log_message(self, format, *args): # noqa: A002
|
|
LOG.info("%s - %s", self.address_string(), format % args)
|
|
|
|
def _authorized(self) -> bool:
|
|
if not TOKEN:
|
|
return False
|
|
return self.headers.get("Authorization", "") == f"Bearer {TOKEN}"
|
|
|
|
def _respond(self, status: HTTPStatus, payload) -> None:
|
|
body = json.dumps(payload).encode("utf-8")
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Access-Control-Allow-Origin", "*")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def _read_body(self, max_bytes: int) -> bytes:
|
|
try:
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
except ValueError:
|
|
raise ValueError("missing or invalid Content-Length") from None
|
|
if length <= 0:
|
|
return b""
|
|
if length > max_bytes:
|
|
raise ValueError(f"body too large ({length} > {max_bytes} bytes)")
|
|
return self.rfile.read(length)
|
|
|
|
def do_OPTIONS(self): # noqa: N802
|
|
# The frontend (pantry-web, a separate origin/port) fetches this API directly
|
|
# from the browser, so a plain CORS preflight has to succeed.
|
|
self.send_response(HTTPStatus.NO_CONTENT)
|
|
self.send_header("Access-Control-Allow-Origin", "*")
|
|
self.send_header("Access-Control-Allow-Headers", "Authorization, Content-Type")
|
|
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
|
|
self.end_headers()
|
|
|
|
def do_GET(self): # noqa: N802
|
|
if not self._authorized():
|
|
self._respond(HTTPStatus.UNAUTHORIZED, {"error": "missing or invalid bearer token"})
|
|
return
|
|
path = urlsplit(self.path).path
|
|
if path == "/inventory":
|
|
self._handle_inventory()
|
|
elif path == "/recipes":
|
|
self._handle_recipes()
|
|
else:
|
|
self._respond(HTTPStatus.NOT_FOUND, {"error": "no such endpoint"})
|
|
|
|
def do_POST(self): # noqa: N802
|
|
if not self._authorized():
|
|
self._respond(HTTPStatus.UNAUTHORIZED, {"error": "missing or invalid bearer token"})
|
|
return
|
|
path = urlsplit(self.path).path
|
|
if path == "/identify":
|
|
self._handle_identify()
|
|
elif path == "/confirm":
|
|
self._handle_confirm()
|
|
else:
|
|
self._respond(HTTPStatus.NOT_FOUND, {"error": "no such endpoint"})
|
|
|
|
def _handle_identify(self) -> None:
|
|
try:
|
|
image = self._read_body(MAX_IMAGE_BYTES)
|
|
except ValueError as exc:
|
|
self._respond(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, {"error": str(exc)})
|
|
return
|
|
if not image:
|
|
self._respond(HTTPStatus.BAD_REQUEST, {"error": "empty image body"})
|
|
return
|
|
|
|
proposal = _identify_via_ollama(image)
|
|
self._respond(HTTPStatus.OK, proposal)
|
|
|
|
def _handle_confirm(self) -> None:
|
|
try:
|
|
raw = self._read_body(MAX_JSON_BYTES)
|
|
payload = json.loads(raw or b"{}")
|
|
except (ValueError, json.JSONDecodeError) as exc:
|
|
self._respond(HTTPStatus.BAD_REQUEST, {"error": f"bad request body: {exc}"})
|
|
return
|
|
|
|
name = str(payload.get("name", "")).strip()
|
|
best_before = str(payload.get("best_before_date", "")).strip()
|
|
amount = float(payload.get("quantity") or 1)
|
|
|
|
if not name:
|
|
self._respond(HTTPStatus.BAD_REQUEST, {"error": "'name' is required"})
|
|
return
|
|
if not re.match(r"^\d{4}-\d{2}-\d{2}$", best_before):
|
|
self._respond(HTTPStatus.BAD_REQUEST, {"error": "'best_before_date' must be YYYY-MM-DD"})
|
|
return
|
|
if not GROCY_API_KEY:
|
|
self._respond(HTTPStatus.SERVICE_UNAVAILABLE, {"error": "GROCY_API_KEY is not configured"})
|
|
return
|
|
|
|
try:
|
|
product_id = _find_or_create_product(name)
|
|
_add_to_stock(product_id, amount, best_before)
|
|
except (urllib.error.URLError, urllib.error.HTTPError, KeyError, ValueError) as exc:
|
|
LOG.warning("pantry-vision: Grocy write failed", exc_info=True)
|
|
self._respond(HTTPStatus.BAD_GATEWAY, {"error": f"Grocy write failed: {exc}"})
|
|
return
|
|
|
|
LOG.info("pantry-vision: added %s x%s (best before %s) to Grocy", name, amount, best_before)
|
|
self._respond(HTTPStatus.OK, {"ok": True, "product_id": product_id})
|
|
|
|
def _handle_inventory(self) -> None:
|
|
try:
|
|
stock = _grocy_get("/api/stock")
|
|
except (urllib.error.URLError, urllib.error.HTTPError) as exc:
|
|
LOG.warning("pantry-vision: Grocy /api/stock unreachable", exc_info=True)
|
|
self._respond(HTTPStatus.BAD_GATEWAY, {"error": f"Grocy unreachable: {exc}"})
|
|
return
|
|
|
|
# VERIFY: assumes each stock row carries a nested "product" object with at
|
|
# least a "name" (Grocy's documented default `?embed` behaviour varies by
|
|
# version) — falls back to "product_id" as the display name if not, rather
|
|
# than dropping the row, same "degrade, don't blank" rule as everywhere else
|
|
# in this project's renderers.
|
|
items = []
|
|
for row in stock if isinstance(stock, list) else []:
|
|
product = row.get("product") or {}
|
|
name = product.get("name") or f"Product #{row.get('product_id')}"
|
|
best_before = row.get("best_before_date")
|
|
items.append(
|
|
{
|
|
"name": name,
|
|
"amount": row.get("amount"),
|
|
"best_before_date": best_before,
|
|
"days_left": _days_left(best_before),
|
|
}
|
|
)
|
|
|
|
items.sort(key=lambda i: (i["days_left"] is None, i["days_left"]))
|
|
self._respond(HTTPStatus.OK, {"items": items})
|
|
|
|
def _handle_recipes(self) -> None:
|
|
try:
|
|
recipes = _grocy_get("/api/objects/recipes")
|
|
except (urllib.error.URLError, urllib.error.HTTPError) as exc:
|
|
LOG.warning("pantry-vision: Grocy recipes unreachable", exc_info=True)
|
|
self._respond(HTTPStatus.BAD_GATEWAY, {"error": f"Grocy unreachable: {exc}"})
|
|
return
|
|
|
|
results = []
|
|
for recipe in recipes if isinstance(recipes, list) else []:
|
|
entry = {"id": recipe.get("id"), "name": recipe.get("name"), "fulfilled": None}
|
|
# Best-effort only: Grocy's fulfillment endpoint needs the recipes feature
|
|
# actually set up (ingredients entered per recipe) to mean anything, and
|
|
# its exact response shape is unverified — see README.md. A failure here
|
|
# degrades to "fulfilled: null" (rendered as "unknown" client-side), never
|
|
# a broken page.
|
|
try:
|
|
fulfillment = _grocy_get(f"/api/recipes/{entry['id']}/fulfillment")
|
|
entry["fulfilled"] = bool(fulfillment.get("need_fulfilled"))
|
|
except Exception:
|
|
pass
|
|
results.append(entry)
|
|
|
|
self._respond(HTTPStatus.OK, {"recipes": results})
|
|
|
|
|
|
def main() -> int:
|
|
logging.basicConfig(
|
|
level=os.environ.get("LOG_LEVEL", "INFO").upper(),
|
|
format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
|
|
)
|
|
|
|
if not TOKEN:
|
|
LOG.error(
|
|
"PANTRY_VISION_TOKEN is not set — every request will be rejected until it is. "
|
|
"See pantry-vision.env.example."
|
|
)
|
|
if not GROCY_API_KEY:
|
|
LOG.warning("GROCY_API_KEY is not set — /confirm will fail until it is configured.")
|
|
|
|
port = int(os.environ.get("PANTRY_VISION_PORT", "8095"))
|
|
server = ThreadingHTTPServer(("0.0.0.0", port), Handler)
|
|
LOG.info("pantry-vision listening on :%d (Ollama: %s, Grocy: %s)", port, OLLAMA_HOST, GROCY_URL)
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
pass
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|