"""Fridge to Fork Gradio app.
The Space runs on CPU and calls the authenticated Modal backend for both model
jobs. Recipe generation is deliberately gated behind editable ingredient fields:
`/generate` receives only the current confirmed field values, never the raw
vision extraction state.
The look is a warm, kitchen-table redesign (cream paper, Spectral serif
headlines over Hanken Grotesque, a leafy garden green hero with a clementine
citrus pop). Because the product must stay a Gradio app, that design is
recreated here with Gradio theming + CSS + custom-HTML cards rather than a
separate frontend framework.
"""
import contextlib
import html
import json
import mimetypes
import os
import re
from pathlib import Path
from typing import Any
import gradio as gr
import requests
try:
from dotenv import load_dotenv
# Local dev: pull F2F_BACKEND_URL / F2F_AUTH_TOKEN from a .env. On the HF Space
# these are injected as real env vars and no .env exists, so this no-ops.
load_dotenv()
except ModuleNotFoundError:
pass
BACKEND_URL_ENV = "F2F_BACKEND_URL"
AUTH_TOKEN_ENV = "F2F_AUTH_TOKEN"
MAX_IMAGES = 3
MAX_INGREDIENT_SLOTS = 40
BACKEND_TIMEOUT_SECONDS = 90
IMAGE_EXTENSIONS = [".jpg", ".jpeg", ".png", ".webp"]
LOADING_COLD_START_NOTE = (
"First call this session can take up to about 1 minute while the kitchen warms up."
)
CUISINE_CHOICES = [
"No preference",
"American",
"Italian",
"Mexican",
"Chinese",
"Japanese",
"Korean",
"Indian",
"Mediterranean",
"Thai",
]
# Tap-to-add exclusions that sit beside the free-text kid-notes field. Selected
# presets are merged with whatever the cook types before going to /generate.
KID_NOTE_PRESETS = [
"no spicy",
"no visible veg",
"no seafood",
]
FREE_STAPLE_NAMES = {
"salt",
"kosher salt",
"sea salt",
"table salt",
"pepper",
"black pepper",
"ground pepper",
"freshly ground pepper",
"freshly ground black pepper",
"salt and pepper",
"water",
"tap water",
"cooking water",
"oil",
"cooking oil",
"neutral oil",
"vegetable oil",
"canola oil",
"olive oil",
"avocado oil",
"cooking spray",
}
STAPLE_ROOT_TOKENS = {"salt", "pepper", "oil", "water", "spray"}
QUANTITY_OR_USAGE_TOKENS = {
"a",
"an",
"and",
"as",
"boiling",
"can",
"cans",
"carton",
"cartons",
"clove",
"cloves",
"cold",
"cup",
"cups",
"dash",
"drizzle",
"enough",
"for",
"frying",
"g",
"gram",
"grams",
"handful",
"handfuls",
"hot",
"kg",
"large",
"lb",
"lbs",
"liter",
"liters",
"litre",
"litres",
"medium",
"ml",
"more",
"needed",
"of",
"optional",
"or",
"ounce",
"ounces",
"oz",
"package",
"packages",
"pan",
"pinch",
"pint",
"pints",
"plus",
"pound",
"pounds",
"quart",
"quarts",
"skillet",
"slice",
"slices",
"small",
"splash",
"tablespoon",
"tablespoons",
"taste",
"tbsp",
"tbsps",
"teaspoon",
"teaspoons",
"the",
"to",
"tsp",
"tsps",
"warm",
}
# Inline icons (stroke = currentColor), mirrored from the design's icon set so
# the recipe cards carry the same leaf / cart / clock / people detailing. Sizes
# are set in CSS so one constant can be reused at any scale.
_SVG_ATTRS = (
'viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="1.8" '
'stroke-linecap="round" stroke-linejoin="round" aria-hidden="true"'
)
ICON_CLOCK = f''
ICON_USERS = (
f''
)
ICON_LEAF = (
f''
)
ICON_CART = (
f''
)
ICON_SPARKLE = (
f''
)
class FriendlyError(Exception):
"""An error safe to show in the Gradio UI."""
def _loading_html(label: str) -> str:
safe_label = html.escape(" ".join(str(label).split()) or "Working...")
safe_note = html.escape(LOADING_COLD_START_NOTE)
return f"""
{safe_label}
{safe_note}
""".strip()
# Guided flow: four steps the user walks through one screen at a time. The
# labels drive the progress rail; the numbers are the source of truth for which
# gr.Column is visible. _show_step() returns updates for
# [stepper, step_snap, step_confirm, step_tune, step_cook] in that order.
WIZARD_STEPS = (
("Snap", "Your fridge"),
("Confirm", "What we spotted"),
("Tune", "A couple of dials"),
("Cook", "Tonight's dinners"),
)
def _stepper_html(active: int) -> str:
"""Progress rail markup for the guided flow; `active` is 1-based."""
nodes = []
for number, (label, _) in enumerate(WIZARD_STEPS, start=1):
if number < active:
state = "done"
elif number == active:
state = "active"
else:
state = "todo"
nodes.append(
f'
'
f'{number}'
f'{html.escape(label)}'
"
"
)
return (
''
+ "".join(nodes)
+ ""
)
def _show_step(active: int) -> list[Any]:
"""Reveal exactly one step.
Returns updates for [stepper, step_snap, step_confirm, step_tune, step_cook];
callers append this to a handler's outputs to navigate the wizard.
"""
updates: list[Any] = [gr.update(value=_stepper_html(active))]
updates += [
gr.update(visible=(number == active))
for number in range(1, len(WIZARD_STEPS) + 1)
]
return updates
def _clear_status_update() -> Any:
return gr.update(value="", visible=False)
def _show_step_and_clear_status(active: int) -> list[Any]:
return _show_step(active) + [_clear_status_update()]
def _backend_base_url() -> str:
raw_url = os.environ.get(BACKEND_URL_ENV, "").strip()
if not raw_url:
raise FriendlyError(f"Set {BACKEND_URL_ENV} before using the app.")
return raw_url.rstrip("/")
def _auth_headers() -> dict[str, str]:
token = os.environ.get(AUTH_TOKEN_ENV, "").strip()
if not token:
raise FriendlyError(f"Set {AUTH_TOKEN_ENV} before using the app.")
return {"Authorization": f"Bearer {token}"}
def _backend_endpoint(path: str) -> str:
return f"{_backend_base_url()}/{path.lstrip('/')}"
def _friendly_backend_error(response: requests.Response) -> FriendlyError:
if response.status_code == 401:
return FriendlyError("Backend authentication failed. Check F2F_AUTH_TOKEN.")
try:
payload = response.json()
except ValueError:
payload = {}
detail = payload.get("detail") if isinstance(payload, dict) else None
if response.status_code in {413, 415, 422} and isinstance(detail, str):
return FriendlyError(detail)
if response.status_code >= 500:
return FriendlyError(
"The backend had trouble processing that request. Try again in a moment."
)
if isinstance(detail, str) and detail:
return FriendlyError(detail)
return FriendlyError("The backend rejected the request.")
def _post_multipart(path: str, image_paths: list[Path]) -> dict[str, Any]:
files: list[tuple[str, tuple[str, Any, str]]] = []
with contextlib.ExitStack() as stack:
for image_path in image_paths:
mime_type = (
mimetypes.guess_type(image_path.name)[0] or "application/octet-stream"
)
handle = stack.enter_context(image_path.open("rb"))
files.append(("images", (image_path.name, handle, mime_type)))
response = requests.post(
_backend_endpoint(path),
headers=_auth_headers(),
files=files,
timeout=BACKEND_TIMEOUT_SECONDS,
)
if response.status_code >= 400:
raise _friendly_backend_error(response)
try:
return response.json()
except ValueError as exc:
raise FriendlyError("Backend returned something other than JSON.") from exc
def _post_json(path: str, payload: dict[str, Any]) -> dict[str, Any]:
response = requests.post(
_backend_endpoint(path),
headers=_auth_headers(),
json=payload,
timeout=BACKEND_TIMEOUT_SECONDS,
)
if response.status_code >= 400:
raise _friendly_backend_error(response)
try:
return response.json()
except ValueError as exc:
raise FriendlyError("Backend returned something other than JSON.") from exc
def _post_form_with_optional_image(
path: str,
data: dict[str, str],
image_path: Path | None,
) -> dict[str, Any]:
files: list[tuple[str, tuple[str, Any, str]]] | None = None
with contextlib.ExitStack() as stack:
if image_path is not None:
mime_type = (
mimetypes.guess_type(image_path.name)[0] or "application/octet-stream"
)
handle = stack.enter_context(image_path.open("rb"))
files = [("plate_photo", (image_path.name, handle, mime_type))]
response = requests.post(
_backend_endpoint(path),
headers=_auth_headers(),
data=data,
files=files,
timeout=BACKEND_TIMEOUT_SECONDS,
)
if response.status_code >= 400:
raise _friendly_backend_error(response)
try:
return response.json()
except ValueError as exc:
raise FriendlyError("Backend returned something other than JSON.") from exc
def _upload_paths(files: Any) -> list[Path]:
if files is None:
raise FriendlyError("Choose 1-3 fridge, pantry, or counter photos first.")
if not isinstance(files, list):
files = [files]
paths: list[Path] = []
for file_obj in files:
if isinstance(file_obj, (str, Path)):
paths.append(Path(file_obj))
elif hasattr(file_obj, "name"):
paths.append(Path(file_obj.name))
elif isinstance(file_obj, dict) and file_obj.get("path"):
paths.append(Path(file_obj["path"]))
paths = [path for path in paths if path.exists()]
if not paths:
raise FriendlyError("Choose 1-3 fridge, pantry, or counter photos first.")
if len(paths) > MAX_IMAGES:
raise FriendlyError(f"Use at most {MAX_IMAGES} photos for one extraction.")
return paths
def _optional_upload_path(file_obj: Any) -> Path | None:
if file_obj is None:
return None
if isinstance(file_obj, list):
file_obj = file_obj[0] if file_obj else None
if file_obj is None:
return None
path: Path | None = None
if isinstance(file_obj, (str, Path)):
path = Path(file_obj)
elif hasattr(file_obj, "name"):
path = Path(file_obj.name)
elif isinstance(file_obj, dict) and file_obj.get("path"):
path = Path(file_obj["path"])
if path is None or not path.exists():
raise FriendlyError("Choose a valid plate photo or clear the upload.")
return path
def _clean_ingredient_name(value: Any) -> str:
if not isinstance(value, str):
return ""
return " ".join(value.split())
def _normalize_food_name(value: str) -> str:
normalized = re.sub(r"[^a-z0-9]+", " ", value.casefold())
return " ".join(normalized.split())
FREE_STAPLE_TOKENS = {
token
for staple_name in FREE_STAPLE_NAMES
for token in _normalize_food_name(staple_name).split()
}
def _is_quantity_token(value: str) -> bool:
return bool(re.fullmatch(r"\d+(?:[./]\d+)?|half|quarter", value))
def _food_tokens(value: str) -> set[str]:
return {
token
for token in _normalize_food_name(value).split()
if token not in QUANTITY_OR_USAGE_TOKENS and not _is_quantity_token(token)
}
def _is_free_staple(value: str) -> bool:
tokens = _food_tokens(value)
if not tokens:
return False
normalized = " ".join(tokens)
if normalized in FREE_STAPLE_NAMES:
return True
return bool(tokens & STAPLE_ROOT_TOKENS) and tokens.issubset(FREE_STAPLE_TOKENS)
def _renderable_ingredient_items(items: Any) -> list[str]:
if not isinstance(items, list):
return []
renderable: list[str] = []
seen: set[str] = set()
for item in items:
# Dinner items are {"name", "qty"} objects; bare strings are tolerated
# so a stale backend during a deploy window still renders.
if isinstance(item, dict):
ingredient = _clean_ingredient_name(item.get("name"))
qty = _clean_ingredient_name(item.get("qty"))
else:
ingredient = _clean_ingredient_name(item)
qty = ""
key = _normalize_food_name(ingredient)
if not ingredient or not key or key in seen or _is_free_staple(ingredient):
continue
renderable.append(f"{ingredient} — {qty}" if qty else ingredient)
seen.add(key)
return renderable
def _ingredient_label(item: dict[str, Any]) -> str:
name = _clean_ingredient_name(item.get("name"))
qty = _clean_ingredient_name(item.get("qty"))
if qty:
return f"{qty} {name}".strip()
return name
def _ingredients_from_extract(payload: dict[str, Any]) -> list[str]:
items = payload.get("items", [])
if not isinstance(items, list):
raise FriendlyError(
"Backend returned an ingredient payload in an unexpected shape."
)
ingredients: list[str] = []
seen: set[str] = set()
for item in items:
if not isinstance(item, dict):
continue
ingredient = _ingredient_label(item)
key = ingredient.casefold()
if ingredient and key not in seen:
ingredients.append(ingredient)
seen.add(key)
return ingredients[:MAX_INGREDIENT_SLOTS]
def _merge_kid_notes(preset_values: Any, free_text: Any) -> str:
"""Combine the selected preset chips with the typed notes into one string.
The backend treats `kid_notes` as a free-text list of exclusions, so we just
concatenate the active presets and the cook's own text (deduped, comma-joined).
"""
parts: list[str] = []
seen: set[str] = set()
candidates: list[Any] = []
if isinstance(preset_values, list):
candidates.extend(preset_values)
elif preset_values:
candidates.append(preset_values)
candidates.append(free_text)
for candidate in candidates:
cleaned = _clean_ingredient_name(candidate)
key = cleaned.casefold()
if cleaned and key not in seen:
parts.append(cleaned)
seen.add(key)
return ", ".join(parts)
def _confirmed_pantry_from_slots(slot_values: tuple[Any, ...]) -> list[str]:
confirmed: list[str] = []
seen: set[str] = set()
for value in slot_values:
ingredient = _clean_ingredient_name(value)
key = ingredient.casefold()
if ingredient and key not in seen:
confirmed.append(ingredient)
seen.add(key)
return confirmed
def _slot_component_updates(ingredients: list[str]) -> list[Any]:
ingredients = ingredients[:MAX_INGREDIENT_SLOTS]
group_updates = [
gr.update(visible=index < len(ingredients))
for index in range(MAX_INGREDIENT_SLOTS)
]
textbox_updates = [
gr.update(value=ingredients[index] if index < len(ingredients) else "")
for index in range(MAX_INGREDIENT_SLOTS)
]
return group_updates + textbox_updates
def _make_button_update(ingredients: list[str]) -> Any:
return gr.update(interactive=bool(ingredients))
def _ingredient_textboxes_changed(*slot_values):
ingredients = _confirmed_pantry_from_slots(slot_values)
return _make_button_update(ingredients)
def _confirm_next_clicked(*slot_values):
if _confirmed_pantry_from_slots(slot_values):
return _show_step_and_clear_status(3)
return _show_step(2) + [
gr.update(value="Add at least one ingredient before continuing.", visible=True)
]
def _reset_cook_outputs():
# 13 outputs (9 cook results + 4 chat components). NEVER let an inputs=0
# dep over these hit exactly 10 outputs: it would duplicate the harness's
# make-loading fingerprint (outputs==10 ∧ make_button ∧ dinners_html).
return (
"",
gr.update(interactive=True, value="Make dinner"),
gr.update(visible=False),
gr.update(choices=[], value=None, visible=False, interactive=False),
gr.update(value=None, visible=False),
gr.update(visible=False, interactive=False, value="I made this"),
gr.update(value="", visible=False),
{},
[],
gr.update(visible=False),
[],
gr.update(value=""),
gr.update(interactive=True, value="Ask"),
)
def _preserve_cook_outputs():
return tuple(gr.update() for _ in range(len(_reset_cook_outputs())))
def _pantry_changed_since_generation(generated_pantry, slot_values):
if not isinstance(generated_pantry, list):
return False
previous_pantry = _confirmed_pantry_from_slots(tuple(generated_pantry))
if not previous_pantry:
return False
return _confirmed_pantry_from_slots(slot_values) != previous_pantry
def _reset_cook_outputs_if_pantry_changed(generated_pantry, *slot_values):
if _pantry_changed_since_generation(generated_pantry, slot_values):
return _reset_cook_outputs()
return _preserve_cook_outputs()
def _reset_generation_outputs_if_pantry_changed(generated_pantry, *slot_values):
if _pantry_changed_since_generation(generated_pantry, slot_values):
return (gr.update(value="", visible=False),) + _reset_cook_outputs()
return (gr.update(),) + _preserve_cook_outputs()
def _has_generated_dinner_options(dinners_payload, generated_pantry):
if not isinstance(generated_pantry, list) or not generated_pantry:
return False
if not isinstance(dinners_payload, dict):
return False
dinners = dinners_payload.get("dinners")
return isinstance(dinners, list) and bool(dinners)
def _reset_generation_outputs_if_dinner_generated(dinners_payload, generated_pantry):
if _has_generated_dinner_options(dinners_payload, generated_pantry):
return (
(gr.update(value="", visible=False),)
+ _reset_cook_outputs()
+ (gr.update(),)
)
return (gr.update(),) + _preserve_cook_outputs() + (gr.update(),)
def _extract_loading():
return (
gr.update(value=_loading_html("Reading your fridge…"), visible=True),
gr.update(interactive=False, value="Looking inside…"),
)
def _extract_clicked(files):
try:
paths = _upload_paths(files)
payload = _post_multipart("/extract", paths)
ingredients = _ingredients_from_extract(payload)
status = (
"Review the ingredients, make any edits, then press Continue →."
if ingredients
else "No ingredients were detected. Add ingredients manually, then press Continue →."
)
# Success: hand the user to Step 2 (Confirm) — even with nothing detected,
# so they can add ingredients manually before Tune/Cook.
return (
_slot_component_updates(ingredients)
+ [
gr.update(value=status, visible=True),
gr.update(visible=True),
_make_button_update(ingredients),
payload,
gr.update(value="", visible=False),
gr.update(interactive=True, value="Scan again"),
]
+ _show_step(2)
)
except (FriendlyError, requests.RequestException) as exc:
message = (
str(exc)
if isinstance(exc, FriendlyError)
else "Could not reach the backend."
)
# Failure: stay on Step 1 so the user can retry the scan.
return (
_slot_component_updates([])
+ [
gr.update(value=message, visible=True),
gr.update(visible=False),
gr.update(interactive=False),
{},
gr.update(value="", visible=False),
gr.update(interactive=True, value="Find ingredients"),
]
+ _show_step(1)
)
def _add_ingredient_clicked(new_ingredient, *slot_values):
ingredients = _confirmed_pantry_from_slots(slot_values)
ingredient = _clean_ingredient_name(new_ingredient)
if not ingredient:
status = "Type an ingredient to add."
elif len(ingredients) >= MAX_INGREDIENT_SLOTS:
status = f"Ingredient list is capped at {MAX_INGREDIENT_SLOTS} items."
elif ingredient.casefold() in {item.casefold() for item in ingredients}:
status = f"{ingredient} is already on the list."
else:
ingredients.append(ingredient)
status = "Ingredient added."
return _slot_component_updates(ingredients) + [
gr.update(value=""),
gr.update(value=status, visible=True),
_make_button_update(ingredients),
]
def _delete_ingredient_clicked(index, *slot_values):
updated_slots = list(slot_values)
if index < len(updated_slots):
updated_slots[index] = ""
ingredients = _confirmed_pantry_from_slots(tuple(updated_slots))
status = (
"Ingredient removed."
if ingredients
else "Ingredient removed. Add at least one ingredient before making dinner."
)
return _slot_component_updates(ingredients) + [
gr.update(value=status, visible=True),
_make_button_update(ingredients),
]
def _render_list(items: Any, list_class: str, empty_text: str = "None") -> str:
renderable_items = _renderable_ingredient_items(items)
if not renderable_items:
return f'
{html.escape(empty_text)}
'
return (
f'
'
+ "".join(f"
{html.escape(item)}
" for item in renderable_items)
+ "
"
)
def _render_dinners(payload: dict[str, Any]) -> str:
dinners = payload.get("dinners", [])
if not isinstance(dinners, list) or not dinners:
return ""
time_cap_note = payload.get("time_cap_note")
note_markup = ""
if isinstance(time_cap_note, str) and time_cap_note.strip():
note_markup = (
'
'
f"{html.escape(time_cap_note.strip())}"
"
"
)
cards: list[str] = []
for dinner in dinners:
if not isinstance(dinner, dict):
continue
title = html.escape(str(dinner.get("title", "Dinner option")))
time_min = html.escape(str(dinner.get("time_min", "?")))
serves = html.escape(str(dinner.get("serves", "?")))
steps = dinner.get("steps", [])
if isinstance(steps, list):
step_markup = (
""
+ "".join(f"
{html.escape(str(step))}
" for step in steps)
+ ""
)
else:
step_markup = ""
card_number = len(cards) + 1
cards.append(
f"""
Dinner {card_number}
{ICON_CLOCK} about {time_min} min{ICON_USERS} serves ~{serves}