"""
Bio-Bite — Recovery Nutrition Engine
====================================
Reads the recovery data your smartwatch already collects (strain, sleep, HRV)
and turns it into a personalized recovery meal and a next-day plan.
Pipeline: USER INPUT -> hybrid text/wearable routing -> FAISS top-3
-> validated RAG generation -> grounded AI OUTPUT
- Dataset : read directly from the Hugging Face Dataset repo
- Embedder : benjac8/biobite-retriever (E5 snapshot; best neural model in Part 3 v2)
- Generator: Qwen/Qwen2.5-3B-Instruct (winner of the 18-case Part 4 v2 benchmark)
"""
import html
import hashlib
import json
import os
import re
import traceback
from datetime import datetime
import faiss
import gradio as gr
import numpy as np
import pandas as pd
import torch
from datasets import load_dataset
from PIL import Image, ImageDraw, ImageEnhance, ImageFilter, ImageOps
from sentence_transformers import SentenceTransformer
from transformers import AutoModelForCausalLM, AutoTokenizer
try:
import pytesseract
except ImportError: # OCR remains optional in local development.
pytesseract = None
from biobite_logic import (
build_ingredient_preview,
build_profile_query,
build_state_description,
find_metric_conflicts,
grounded_explanation,
grounded_plan,
ingredient_present,
metric_category_scores,
optional_number,
parse_json_object,
parse_watch_ocr_text,
selected_metric_values,
validate_food_preferences,
validate_generation,
)
# ZeroGPU support (falls back gracefully when running locally / on CPU)
try:
import spaces
ZERO_GPU = True
except ImportError: # local run
ZERO_GPU = False
class _Dummy:
@staticmethod
def GPU(*a, **k):
def deco(fn):
return fn
return deco
spaces = _Dummy()
# --------------------------------------------------------------------------
# Configuration
# --------------------------------------------------------------------------
SEED = 42
HF_DATASET = "benjac8/bio-bite-recovery-nutrition"
RECSYS_CONFIG_FILE = os.environ.get("BIOBITE_RECSYS_CONFIG", "recsys_config.json")
if os.path.exists(RECSYS_CONFIG_FILE):
with open(RECSYS_CONFIG_FILE) as config_file:
RECSYS_CONFIG = json.load(config_file)
else:
RECSYS_CONFIG = {}
EMBED_MODEL = os.environ.get(
"BIOBITE_EMBED_MODEL",
RECSYS_CONFIG.get("embedding_model_id", "intfloat/e5-small-v2"),
)
QUERY_PREFIX = os.environ.get(
"BIOBITE_QUERY_PREFIX",
RECSYS_CONFIG.get(
"query_prefix", "query: "
),
)
GEN_MODEL = os.environ.get("BIOBITE_GEN_MODEL", "Qwen/Qwen2.5-3B-Instruct")
EMB_FILE = os.environ.get("BIOBITE_EMBEDDINGS_FILE", "biobite_embeddings.parquet")
GUIDANCE_FILE = "recovery_guidance.json"
SPOONACULAR_KEY = os.environ.get("SPOONACULAR_API_KEY", "")
DATASET_REVISION = os.environ.get(
"BIOBITE_DATASET_REVISION", RECSYS_CONFIG.get("dataset_revision")
)
EXPECTED_ROW_ALIGNMENT = RECSYS_CONFIG.get("row_alignment_sha256", "")
LOCAL_DATASET_CSV = os.environ.get("BIOBITE_DATASET_CSV", "")
ROW_ID_FIELDS = RECSYS_CONFIG.get(
"row_id_fields", ["Physiological_State", "Recipe_Name", "Ingredients"]
)
MAIN_INGREDIENT_CHOICES = [
"No preference", "Steak / beef", "Chicken", "Salmon", "Tuna", "Eggs",
"Tofu", "Tempeh", "Lentils", "Chickpeas", "Greek yogurt",
]
EXTRA_INGREDIENT_CHOICES = [
"Rice", "Quinoa", "Pasta", "Potatoes", "Sweet potato", "Oats", "Spinach",
"Broccoli", "Tomato", "Avocado", "Mushrooms", "Bell pepper", "Beans",
]
EXCLUDED_INGREDIENT_CHOICES = [
"Peanuts", "Tree nuts", "Dairy", "Gluten", "Fish / shellfish", "Eggs",
"Soy", "Sesame", "Caffeine",
]
np.random.seed(SEED)
torch.manual_seed(SEED)
# --------------------------------------------------------------------------
# Load data, index and models (once, at startup)
# --------------------------------------------------------------------------
print("Loading dataset from Hugging Face…")
df = (
pd.read_csv(LOCAL_DATASET_CSV)
if LOCAL_DATASET_CSV
else load_dataset(
HF_DATASET, split="train", revision=DATASET_REVISION or None
).to_pandas()
)
def _attach_and_verify_row_ids(frame):
"""Create stable content IDs and fail fast if embeddings no longer align."""
row_ids = []
sequence_hash = hashlib.sha256()
for position, (_, row) in enumerate(frame.iterrows()):
key = "\x1f".join(str(row.get(field, "")) for field in ROW_ID_FIELDS)
row_id = hashlib.sha256(key.encode("utf-8")).hexdigest()[:16]
row_ids.append(row_id)
sequence_hash.update(f"{position}:{row_id}\n".encode("utf-8"))
actual = sequence_hash.hexdigest()
if EXPECTED_ROW_ALIGNMENT and actual != EXPECTED_ROW_ALIGNMENT:
raise RuntimeError(
"Dataset/embedding alignment check failed. The pinned dataset rows "
"do not match the saved embedding order."
)
result = frame.copy()
result["row_id"] = row_ids
return result
df = _attach_and_verify_row_ids(df)
print("Loading embeddings…")
doc_emb = pd.read_parquet(EMB_FILE).to_numpy().astype("float32")
if len(doc_emb) != len(df):
raise RuntimeError(
f"Dataset/embedding row count mismatch: {len(df)} rows vs {len(doc_emb)} vectors"
)
index = faiss.IndexFlatIP(doc_emb.shape[1])
index.add(doc_emb)
with open(GUIDANCE_FILE) as fh:
NEXT_DAY_GUIDANCE = json.load(fh)
print("Loading models…")
# The embedder runs OUTSIDE @spaces.GPU (during retrieval), where no real GPU
# exists on ZeroGPU — so it must stay on CPU. Encoding one query takes ~50 ms.
embedder = SentenceTransformer(EMBED_MODEL, device="cpu")
# The generator runs INSIDE @spaces.GPU. ZeroGPU requires models to be placed on
# cuda at module level (a CUDA emulation layer makes this work at startup, and
# the real GPU is attached inside the decorated function).
if ZERO_GPU:
gen_device, gen_dtype = "cuda", torch.float16
elif torch.cuda.is_available():
gen_device, gen_dtype = "cuda", torch.float16
else:
gen_device, gen_dtype = "cpu", torch.float32
gen_tokenizer = AutoTokenizer.from_pretrained(GEN_MODEL)
gen_tokenizer.pad_token_id = gen_tokenizer.eos_token_id
gen_model = AutoModelForCausalLM.from_pretrained(
GEN_MODEL,
torch_dtype=gen_dtype,
low_cpu_mem_usage=True,
)
gen_model.to(gen_device)
gen_model.eval()
print(f"Ready — {len(df)} recipes indexed. "
f"ZeroGPU={ZERO_GPU}, generator on {gen_device}, embedder on cpu.")
# --------------------------------------------------------------------------
# Retrieval
# --------------------------------------------------------------------------
def retrieve(user_text, k=3, diet=None, max_prep=None, category=None, pool=500,
required_ingredients=None, excluded_ingredients=None):
"""Top-k recovery recipes for a free-text description of the user's day."""
q = embedder.encode(
[QUERY_PREFIX + user_text], convert_to_numpy=True, normalize_embeddings=True
).astype("float32")
# Structured filters are cheap over 10k rows and should never reduce the UI
# below the assignment's required three recommendations. Search the full
# index when filters are active, then rank rather than hard-filter category.
search_k = len(df) if (diet != "Any" or max_prep or excluded_ingredients) else pool
scores, idx = index.search(q, search_k)
cand = df.iloc[idx[0]].copy()
cand["similarity"] = scores[0]
# Diet and explicit exclusions are safety choices and remain hard rules.
if diet and diet != "Any":
cand = cand[cand["diet_tag"] == diet]
if excluded_ingredients and len(cand):
searchable = cand["Recipe_Name"].astype(str) + " " + cand["Ingredients"].astype(str)
safe = searchable.map(lambda value: not any(
ingredient_present(value, blocked) for blocked in excluded_ingredients
))
cand = cand[safe]
if len(cand) == 0:
raise ValueError("No recipes satisfy the selected diet and exclusions")
searchable = cand["Recipe_Name"].astype(str) + " " + cand["Ingredients"].astype(str)
cand["_main_match"] = False
if required_ingredients:
cand["_main_match"] = searchable.map(
lambda value: ingredient_present(value, required_ingredients[0])
)
cand["_category_match"] = (
cand["recovery_category"] == category if category else True
)
cand["_within_time"] = (
cand["Prep_Time"] <= max_prep if max_prep else True
)
cand = cand.sort_values(
["_within_time", "_main_match", "_category_match", "similarity"],
ascending=[False, False, False, False],
kind="stable",
)
return cand.head(k).drop(
columns=["_main_match", "_category_match", "_within_time"]
)
def infer_recovery_category(user_text, sleep=None, strain=None, hrv=None, k=30):
"""Hybrid router: semantic retrieval plus explainable wearable-range fit."""
hits = retrieve(user_text, k=k, pool=500)
# Shift cosine scores to positive weights, then normalise category totals.
weights = hits["similarity"] - hits["similarity"].min() + 0.01
semantic = weights.groupby(hits["recovery_category"]).sum().to_dict()
semantic_total = sum(semantic.values()) or 1.0
metric = metric_category_scores(sleep=sleep, strain=strain, hrv=hrv)
has_metrics = any(value is not None for value in (sleep, strain, hrv))
metric_total = sum(metric.values()) or 1.0
combined = {}
for category in NEXT_DAY_GUIDANCE:
semantic_score = semantic.get(category, 0.0) / semantic_total
combined[category] = semantic_score if not has_metrics else (
0.65 * semantic_score + 0.35 * (metric[category] / metric_total)
)
return max(combined, key=combined.get), combined
# --------------------------------------------------------------------------
# Input validation — deterministic and explainable (no model needed)
# --------------------------------------------------------------------------
MAX_INPUT_CHARS = 800
# Any one of these signals the user is describing a training / recovery day.
TOPIC_WORDS = {
"train", "training", "trained", "workout", "work-out", "gym", "lift", "lifted",
"lifting", "squat", "squats", "deadlift", "bench", "press", "crossfit", "run",
"ran", "running", "jog", "jogging", "marathon", "cycle", "cycling", "bike",
"ride", "swim", "swam", "swimming", "row", "rowing", "yoga", "pilates",
"football", "soccer", "basketball", "tennis", "climb", "climbing", "hike",
"hiking", "cardio", "session", "exercise", "sport", "sports", "match", "game",
"practice", "sleep", "slept", "sleeping", "rest", "rested", "resting", "nap",
"tired", "exhausted", "drained", "wrecked", "sore", "fatigue", "fatigued",
"recovery", "recover", "stress", "stressed", "stressful", "anxious", "anxiety",
"burnt", "burnout", "hrv", "strain", "heart", "rate", "dehydrated",
"dehydration", "sweat", "sweated", "sweating", "hydration", "thirsty",
"muscle", "muscles", "body", "energy", "day", "today", "hours", "hour",
}
def validate_input(text):
"""Return (status, cleaned_text, message).
status: 'ok' | 'warn' | 'error'
error -> we cannot proceed
warn -> we proceed, but tell the user the result may be poor
"""
if text is None or not str(text).strip():
return ("error", "",
"Please describe your day first — for example "
"“Heavy leg day at the gym, slept 5 hours, feeling wrecked.”")
t = str(text).strip()
if not re.search(r"[A-Za-z-]", t):
return ("error", t,
"That doesn't look like a description of your day. "
"Try something like “Ran 10km this morning and I'm drained.”")
# Non-Latin script (e.g. Hebrew) — the embedding model is English-only
letters = [c for c in t if c.isalpha()]
if letters and sum(1 for c in letters if ord(c) > 591) / len(letters) > 0.3:
return ("error", t,
"Bio-Bite currently understands English only. "
"Please describe your day in English.")
# Check the topic BEFORE truncating, so a long entry isn't misjudged
on_topic = bool(set(re.findall(r"[a-z]+", t.lower())) & TOPIC_WORDS)
# On-topic text can be very short and still useful ("ran 20km")
if len(t) < (6 if on_topic else 10):
return ("error", t,
"That's a little short — tell me about your training, sleep or "
"stress today so I can match the right recovery meal.")
if len(t) > MAX_INPUT_CHARS:
t = t[:MAX_INPUT_CHARS]
if not on_topic:
return ("warn", t,
"I couldn't spot anything about training, sleep, stress or hydration "
"in that, so this match may be off. Mentioning your workout, sleep or "
"how you feel will give a much better result.")
return ("ok", t, "")
def notice_html(message, kind="warn"):
color = "#3ddc84" if kind == "warn" else "#ff8a7a"
return (f''
f'{message}
')
# --------------------------------------------------------------------------
# Generation (single combined call keeps latency ~15s within ZeroGPU quota)
# --------------------------------------------------------------------------
PROMPT = """You are a recovery-nutrition recipe assistant for an educational prototype.
The athlete describes their day as: "{state}" ({numbers})
Their recovery goal is: {category}
Nutritional need: {need}
A recommended recipe from our database, to use as inspiration:
- Name: {name}
- Ingredients: {ingredients}
- Prep time: {prep} minutes
The athlete's additional request is: "{constraint}"
Required ingredients — EVERY item must appear in the final Ingredients field: {required_ingredients}
Excluded ingredients — NONE may appear in the final recipe: {excluded_ingredients}
Selected diet: {diet}
Hard maximum preparation time: {max_prep}
Adapt the recipe into a NEW dish that respects every structured choice while still
meeting the nutritional need. Required ingredients are hard constraints, not ideas.
Never use an excluded ingredient or violate the selected diet or maximum preparation
time. The science explanation and tomorrow's plan are added later from verified,
deterministic guidance; do not include them in your answer.
Write in ENGLISH only.
Reply with ONE valid JSON object and NOTHING else, with exactly these keys:
- "Recipe_Name": string (an original name for the new dish)
- "Ingredients": string (comma-separated)
- "Instructions": string (numbered steps)
- "prep_time_min": integer
"""
# Records why the last generation failed so the UI can explain it accurately.
LAST_ERROR = {"reason": None, "detail": ""}
@spaces.GPU(duration=30)
def _generate_raw(prompt):
"""The ONLY function that touches the GPU.
Keeping the boundary to `str -> str` means just a string is serialised to the
ZeroGPU worker process, and prompt building / JSON parsing happen on CPU
(which also avoids burning GPU quota on non-GPU work).
"""
messages = [{"role": "user", "content": prompt}]
rendered = gen_tokenizer.apply_chat_template(
messages, tokenize=False, add_generation_prompt=True
)
encoded = gen_tokenizer(rendered, return_tensors="pt").to(gen_model.device)
with torch.inference_mode():
output_ids = gen_model.generate(
**encoded,
max_new_tokens=320,
do_sample=False,
pad_token_id=gen_tokenizer.eos_token_id,
eos_token_id=gen_tokenizer.eos_token_id,
)
new_tokens = output_ids[0, encoded["input_ids"].shape[1]:]
return gen_tokenizer.decode(new_tokens, skip_special_tokens=True).strip()
def generate_biobite(source_row, state, constraint, category, numbers,
diet="Any", max_prep=None, required_ingredients=None,
excluded_ingredients=None):
"""RAG generation: retrieved recipe + coded science -> new recipe + plan."""
g = NEXT_DAY_GUIDANCE[category]
prompt = PROMPT.format(
state=state, numbers=numbers, category=category,
need=source_row["Nutritional_Need"], name=source_row["Recipe_Name"],
ingredients=source_row["Ingredients"], prep=source_row["Prep_Time"],
constraint=constraint, diet=diet,
required_ingredients=", ".join(required_ingredients or []) or "none",
excluded_ingredients=", ".join(excluded_ingredients or []) or "none",
max_prep=f"{max_prep} minutes" if max_prep else "not specified",
)
errors = []
for attempt in range(2):
attempt_prompt = prompt
if attempt and errors:
attempt_prompt += (
"\nYour previous answer failed these checks: " + "; ".join(errors) +
"\nCorrect every issue. Return one complete JSON object only."
)
try:
raw = _generate_raw(attempt_prompt)
except Exception as exc: # GPU worker error, quota, timeout…
print(f"GPU GENERATION FAILED [{type(exc).__name__}]: {exc}")
traceback.print_exc()
LAST_ERROR["reason"] = "gpu"
LAST_ERROR["detail"] = f"{type(exc).__name__}: {exc}"
return None
LAST_ERROR["reason"] = "validation"
LAST_ERROR["detail"] = raw[:300]
obj, errors = validate_generation(
parse_json_object(raw), selected_diet=diet,
constraint=constraint, selected_max=max_prep,
required_ingredients=required_ingredients,
excluded_ingredients=excluded_ingredients,
)
if obj is not None:
break
LAST_ERROR["detail"] = "; ".join(errors) + " | " + raw[:220]
print(f"GENERATION VALIDATION FAILED [attempt {attempt + 1}]:", LAST_ERROR["detail"])
if obj is None:
return None
# Scientific and training claims stay deterministic. The model is used for
# the new recipe, while coded guidance supplies the explanation and plan.
obj["why_it_works"] = grounded_explanation(category)
obj["next_day_plan"] = grounded_plan(category, g)
return obj
# --------------------------------------------------------------------------
# Bonus: fetch a real dish photo from a live recipe API
# --------------------------------------------------------------------------
def fetch_dish_image(recipe_name):
"""Live-data bonus. Returns an image URL or None (never breaks the app)."""
if not SPOONACULAR_KEY:
return None
try:
import requests
r = requests.get(
"https://api.spoonacular.com/recipes/complexSearch",
params={"query": recipe_name, "number": 1, "apiKey": SPOONACULAR_KEY},
timeout=6,
)
hits = r.json().get("results", [])
return hits[0].get("image") if hits else None
except Exception:
return None
# --------------------------------------------------------------------------
# HTML rendering
# --------------------------------------------------------------------------
PANEL = "background:#111a13;border:1px solid #1f3324;border-radius:14px;"
GREEN = "#3ddc84"
GREEN_DIM = "#7fbf9a"
TEXT = "#dfeee4"
MUTED = "#8aa695"
def cards_html(rows, required_ingredients=None):
"""Render the three retrieved dataset matches.
The preview is built by ``build_ingredient_preview`` so any ingredient the
card claims via an "Includes" badge is guaranteed to be visible in the list.
A badge is only ever emitted for labels confirmed by ``ingredient_present``
against the COMPLETE Ingredients field.
"""
cards = []
for _, r in rows.iterrows():
category = html.escape(str(r["recovery_category"]))
recipe_name = html.escape(str(r["Recipe_Name"]))
cuisine = html.escape(str(r["cuisine"]))
diet_tag = html.escape(str(r["diet_tag"]))
ingredient_text = str(r["Ingredients"])
items, remaining, matched = build_ingredient_preview(
ingredient_text, required_ingredients, limit=5
)
preview = html.escape(", ".join(items))
if remaining > 0:
preview += f" + {remaining} more"
match_badge = (
f'✓ Includes '
f'{html.escape(", ".join(matched))}
' if matched else ""
)
# Labels the user asked for that this dataset recipe does NOT contain are
# never advertised as present; we say plainly that the generated recipe
# will supply them instead.
missing = [str(value) for value in (required_ingredients or [])
if str(value).strip() and str(value) not in matched]
missing_note = (
f''
f'The personalized recipe will include {html.escape(", ".join(missing))}.
'
if missing else ""
)
cards.append(f"""
{category}
{recipe_name}
⏱ {r['Prep_Time']} min · 🔥 ~{r['calories']} kcal
🥩 ~{r['protein_g']}g protein · 🌾 ~{r['carbs_g']}g carbs
✨ ~{r['magnesium_mg']}mg magnesium
🥣 {preview}
{cuisine} · {diet_tag}
{match_badge}
{missing_note}
""")
return (''
+ "".join(cards) + "
")
def recipe_html(obj, image_url=None, servings=2):
safe_url = html.escape(str(image_url), quote=True) if image_url else None
img = (f'
'
if safe_url else "")
name = html.escape(str(obj["Recipe_Name"]))
ingredients = html.escape(str(obj["Ingredients"]))
steps = html.escape(str(obj["Instructions"])).replace("\n", "
")
why = html.escape(str(obj["why_it_works"]))
return f"""
{img}
🍳 {name}
Ready in {obj['prep_time_min']} minutes · {int(servings)} serving{'s' if int(servings) != 1 else ''}
Ingredients
{ingredients}
Instructions
{steps}
🔬 Why this works for you
{why}
📅 Tomorrow's Recovery Plan
Based on your recovery state: {html.escape(category)}
"""
# --------------------------------------------------------------------------
# Main callback
# --------------------------------------------------------------------------
def read_watch_screenshot(image):
"""Read labelled recovery values locally; never retain or log the image."""
if image is None:
return (
notice_html("Upload a WHOOP or watch screenshot first.", "error"),
gr.update(), gr.update(), gr.update(), gr.update(), "{}",
)
if pytesseract is None:
return (
notice_html(
"Screenshot reading is temporarily unavailable. You can still enter the values manually.",
"error",
),
gr.update(), gr.update(), gr.update(), gr.update(), "{}",
)
try:
source_image = image if isinstance(image, Image.Image) else Image.fromarray(image)
grayscale = ImageOps.autocontrast(source_image.convert("L"))
def prepare(candidate):
scale = max(2, min(4, 1800 // max(candidate.width, 1)))
return candidate.resize(
(candidate.width * scale, candidate.height * scale),
Image.Resampling.LANCZOS,
).filter(ImageFilter.SHARPEN)
enlarged = prepare(grayscale)
top_panel = prepare(grayscale.crop((
0,
int(grayscale.height * 0.08),
grayscale.width,
max(int(grayscale.height * 0.62), 1),
)))
top_inverted = ImageOps.invert(top_panel)
top_thresholded = top_panel.point(lambda pixel: 255 if pixel > 145 else 0)
# WHOOP's home screen places the three dashboard values inside large
# coloured rings. Full-page OCR can miss those isolated white digits
# even when it reads the labels below them. Read each ring from its
# stable relative position as a second, tightly-scoped OCR pass. We
# only trust the layout when all three values form a valid WHOOP row,
# which keeps this fallback conservative for non-WHOOP screenshots.
def read_ring_value(x_start, x_end, low, high):
ring = grayscale.crop((
int(grayscale.width * x_start),
int(grayscale.height * 0.215),
max(int(grayscale.width * x_end), 1),
max(int(grayscale.height * 0.275), 1),
))
ring = prepare(ring)
inverted = ImageOps.autocontrast(ImageOps.invert(ring))
binary = inverted.point(lambda pixel: 255 if pixel > 128 else 0)
# The coloured recovery ring becomes a dark arc after inversion
# and can make OCR treat "93" as "9". Remove only black connected
# components that touch the crop border; the central digits never
# touch that border, so their shapes remain unchanged.
border_clean = binary.copy()
for x in range(border_clean.width):
for y in (0, border_clean.height - 1):
if border_clean.getpixel((x, y)) == 0:
ImageDraw.floodfill(border_clean, (x, y), 255)
for y in range(border_clean.height):
for x in (0, border_clean.width - 1):
if border_clean.getpixel((x, y)) == 0:
ImageDraw.floodfill(border_clean, (x, y), 255)
variants = (
border_clean,
inverted,
)
readings = []
for variant in variants:
for psm in (7, 8):
readings.append(pytesseract.image_to_string(
variant,
config=f"--psm {psm} -c tessedit_char_whitelist=0123456789.%",
))
candidates = []
for token in re.findall(r"\d{1,3}(?:[.,]\d+)?", " ".join(readings)):
value = float(token.replace(",", "."))
if low <= value <= high:
candidates.append(value)
if not candidates:
return None
# Prefer a decimal for Strain when available; WHOOP Strain is
# commonly shown to one decimal place (for example 14.7).
decimal_values = [value for value in candidates if value % 1]
if high == 21 and decimal_values:
return decimal_values[0]
# For percentage rings, a second OCR pass often restores a digit
# clipped by the first pass ("9" versus "93"). Prefer that fuller
# candidate when both are present, while still allowing a genuine
# single-digit score when it is the only reading.
fuller_scores = [value for value in candidates if value >= 10]
if high == 100 and fuller_scores:
return fuller_scores[0]
return candidates[0]
ring_sleep = read_ring_value(0.075, 0.285, 0, 100)
ring_recovery = read_ring_value(0.395, 0.625, 0, 100)
ring_strain = read_ring_value(0.705, 0.95, 0, 21)
ring_hint = ""
if all(value is not None for value in (ring_sleep, ring_recovery, ring_strain)):
ring_hint = (
f"WHOOP SLEEP {ring_sleep:g}% RECOVERY {ring_recovery:g}% "
f"STRAIN {ring_strain:g}"
)
extracted = "\n".join([
ring_hint,
pytesseract.image_to_string(enlarged, config="--psm 11"),
pytesseract.image_to_string(top_panel, config="--psm 6"),
pytesseract.image_to_string(top_inverted, config="--psm 11"),
pytesseract.image_to_string(top_thresholded, config="--psm 11"),
])
parsed = parse_watch_ocr_text(extracted)
found = [label for label, key in (("Sleep", "sleep"), ("Strain", "strain"), ("HRV", "hrv"))
if parsed[key] is not None]
timestamp = datetime.now().astimezone().strftime("%d %b %Y, %H:%M")
source_info = {
"source": parsed["source"],
"timestamp": timestamp,
"workout": parsed["workout"],
"sleep_score": parsed["sleep_score"],
"recovery_score": parsed["recovery_score"],
"hrv_relative_percent": parsed["hrv_relative_percent"],
}
if not found:
status = notice_html(
"I could not confidently find labelled Sleep, Strain or HRV values. "
"Try a tighter, clearer screenshot or enter the values manually.",
"error",
)
else:
values = []
if parsed["sleep"] is not None:
values.append(f"Sleep {parsed['sleep']:g} h")
if parsed["strain"] is not None:
values.append(f"Strain {parsed['strain']:g}/21")
if parsed["hrv"] is not None:
values.append(f"HRV {parsed['hrv']:g} ms")
informational = []
if parsed["sleep_score"] is not None and parsed["sleep"] is None:
informational.append(
f"Sleep score {parsed['sleep_score']:g}% is not sleep duration"
)
if parsed["recovery_score"] is not None:
informational.append(f"Recovery score {parsed['recovery_score']:g}%")
if parsed["hrv_relative_percent"] is not None and parsed["hrv"] is None:
informational.append(
f"relative HRV change {parsed['hrv_relative_percent']:g}% is not HRV in ms"
)
detail = (
"
Also detected: "
+ html.escape(" · ".join(informational))
+ ". These values were not inserted into incompatible fields."
if informational else ""
)
status = notice_html(
f"✓ Read from {html.escape(parsed['source'])}: "
f"{html.escape(' · '.join(values))}
"
"Please review the values below before generating. The image is not added to the project dataset or logs."
f"{detail}"
)
return (
status,
gr.update(value=found),
gr.update(value=parsed["sleep"] if parsed["sleep"] is not None else 7),
gr.update(value=parsed["strain"] if parsed["strain"] is not None else 10),
gr.update(value=parsed["hrv"] if parsed["hrv"] is not None else 45),
json.dumps(source_info),
)
except Exception as exc: # OCR failure must never block manual entry.
print(f"OCR FAILED [{type(exc).__name__}]: {exc}")
return (
notice_html(
"I could not read that screenshot. Try a tighter crop or use manual entry.",
"error",
),
gr.update(), gr.update(), gr.update(), gr.update(), "{}",
)
def food_preference_feedback(main_ingredient, include_ingredients,
excluded_ingredients, diet):
required, _, errors = validate_food_preferences(
main_ingredient, include_ingredients, excluded_ingredients, diet
)
if errors:
return notice_html("
".join(f"• {html.escape(error)}" for error in errors), "error")
if required:
return notice_html(
"✓ The personalized recipe will include: " +
html.escape(", ".join(required)) + "."
)
return ""
def reset_form():
"""Restore valid visual defaults while keeping every wearable metric disabled."""
return (
None, "", [], 7, 10, 45, [], "", "Any", 30, 2,
"No preference", [], [], "No preference", "Any equipment", "", "",
"", "", "", "", "", "{}",
)
def strength_starter(_source):
return QUICK_STARTERS["strength"]
def endurance_starter(_source):
return QUICK_STARTERS["endurance"]
def stress_starter(_source):
return QUICK_STARTERS["stress"]
def run(state, feelings, metric_selection, main_ingredient, include_ingredients,
excluded_ingredients, constraint, diet, max_prep, servings, cuisine,
equipment, sleep_hours, strain, hrv, wearable_source):
# ---- Layer 1: validate the input -------------------------------------
try:
sleep_value, strain_value, hrv_value = selected_metric_values(
metric_selection, sleep_hours, strain, hrv
)
except (TypeError, ValueError) as exc:
return (notice_html(f"Please check the wearable values: {html.escape(str(exc))}.", "error"),
"", "", "", "")
state = build_state_description(
state, feelings, any(value is not None for value in (sleep_value, strain_value, hrv_value))
)
status, state, message = validate_input(state)
if status == "error":
return (notice_html(message, "error"), "", "", "", "")
warning = notice_html(message) if status == "warn" else ""
try:
required, excluded, food_errors = validate_food_preferences(
main_ingredient, include_ingredients, excluded_ingredients, diet
)
if food_errors:
friendly_errors = "
".join(
f"• {html.escape(error)}" for error in food_errors
)
return (
notice_html(
"Please fix these food preferences before generating:
" +
friendly_errors,
"error",
),
"", "", "", "",
)
profile_query = build_profile_query(
state, sleep=sleep_value, strain=strain_value, hrv=hrv_value
)
nums = []
if sleep_value is not None:
nums.append(f"slept {sleep_value:g}h")
if strain_value is not None:
nums.append(f"strain {strain_value:g}/21")
if hrv_value is not None:
nums.append(f"HRV {hrv_value:g} ms")
numbers = ", ".join(nums) if nums else "no wearable numbers given"
category, route_scores = infer_recovery_category(
profile_query, sleep=sleep_value, strain=strain_value, hrv=hrv_value
)
conflicts = find_metric_conflicts(state, sleep=sleep_value, strain=strain_value)
if conflicts:
conflict_text = "; ".join(conflicts).capitalize() + "."
warning += notice_html(conflict_text)
# ---- Layer 2: retrieve, relaxing filters if they are too strict ---
want_prep = int(max_prep) if max_prep else None
top3 = retrieve(
profile_query, k=3, diet=diet, max_prep=want_prep, category=category,
pool=2000 if required else 500,
required_ingredients=required,
excluded_ingredients=excluded,
)
relaxed = ""
if want_prep and (top3["Prep_Time"] > want_prep).any():
relaxed = (f"No {'' if diet == 'Any' else diet + ' '}meals under "
f"{want_prep} min matched your state, so the time limit "
f"was relaxed.")
elif diet and diet != "Any" and (top3["diet_tag"] != diet).any():
relaxed = f"Not enough {diet} matches, so the diet filter was relaxed."
if (top3["recovery_category"] != category).any():
category_note = (
"To keep three recommendations while respecting your food choices, "
"some cards come from a nearby recovery category."
)
relaxed = f"{relaxed} {category_note}".strip()
if required and not ingredient_present(
f"{top3.iloc[0]['Recipe_Name']} {top3.iloc[0]['Ingredients']}", required[0]
):
ingredient_note = (
f"The dataset has no close {html.escape(required[0])} match for this "
"recovery state. The personalized recipe will still be required to include it."
)
relaxed = f"{relaxed} {ingredient_note}".strip()
readable_category = category.replace("-", " ").title()
reason_bits = []
if feelings:
reason_bits.append(", ".join(feelings).lower())
if nums:
reason_bits.append(numbers)
reason = " and ".join(reason_bits) or "your recovery description"
header = (f'Your match: {html.escape(readable_category)}
'
f'Based on {html.escape(reason)}.
')
# Summary (warnings + match header) is rendered above the primary result;
# the three dataset cards go into a closed Accordion below it.
summary = warning + (notice_html(relaxed) if relaxed else "") + header
cards = cards_html(top3, required)
if not constraint or not constraint.strip():
constraint = "Keep it simple with easy-to-find ingredients."
preference_details = [f"Make {int(servings)} serving(s)"]
if cuisine and cuisine != "No preference":
preference_details.append(f"Use a {cuisine} style")
if equipment and equipment != "Any equipment":
preference_details.append(f"Cooking setup: {equipment}")
constraint = constraint.strip() + ". " + ". ".join(preference_details)
# ---- Layer 3: generate, retry only if a retry can help ------------
obj = generate_biobite(
top3.iloc[0], state, constraint, category, numbers,
diet=diet, max_prep=want_prep,
required_ingredients=required,
excluded_ingredients=excluded,
)
note = ""
if obj is None:
obj = template_fallback(
top3.iloc[0], category,
required_ingredients=required,
excluded_ingredients=excluded,
max_prep=want_prep,
diet=diet,
)
if LAST_ERROR["reason"] == "gpu":
detail = LAST_ERROR["detail"].lower()
if "quota" in detail or "exceeded" in detail:
note = notice_html(
"The free daily GPU allowance for this Space has run out, so "
"this is a deterministic recipe that still respects your selected "
"ingredients, exclusions and time limit. The allowance resets every 24 hours.")
else:
note = notice_html(
"The AI generator is temporarily unavailable, so this is the "
"deterministic recipe that still respects your structured choices.")
elif LAST_ERROR["reason"] == "validation":
note = notice_html(
"The generated recipe did not pass the format or constraint checks, "
"so a deterministic recipe that respects your structured choices is shown.")
else:
note = notice_html(
"The generator returned an unexpected response, so the best "
"safe deterministic recipe is shown instead.")
img = fetch_dish_image(obj["Recipe_Name"])
try:
source_info = json.loads(wearable_source or "{}")
except (TypeError, json.JSONDecodeError):
source_info = {}
source_label = html.escape(str(source_info.get("source", "Manual entry")))
timestamp = html.escape(str(source_info.get("timestamp", "This session")))
row_ids = ", ".join(html.escape(str(value)) for value in top3["row_id"].tolist())
generation_mode = "AI-generated recipe" if not note else "validated deterministic fallback"
technical = f"""
Decision trace
Source: {source_label} · {timestamp}
Metrics used: {html.escape(numbers)}
Route: {html.escape(category)} · score {route_scores[category]:.3f}
Retrieved row IDs: {row_ids}
Output mode: {generation_mode}
"""
return (
summary,
note + recipe_html(obj, img, servings=servings),
plan_html(obj["next_day_plan"], category),
cards,
technical,
)
# ---- Layer 4: nothing should ever reach the user as a crash ----------
except Exception as exc: # noqa: BLE001
msg = str(exc).lower()
if "quota" in msg or "gpu" in msg:
friendly = ("The free daily GPU allowance for this Space has run out. "
"It resets each day — please try again later.")
else:
friendly = ("Something went wrong while building your Bio-Bite. "
"Please try again in a moment.")
print("ERROR in run():", exc)
return (notice_html(friendly, "error"), "", "", "", "")
# --------------------------------------------------------------------------
# UI
# --------------------------------------------------------------------------
QUICK_STARTERS = {
"strength": (
"Heavy CrossFit workout today", ["Sore muscles", "Low energy"],
["Sleep", "Strain", "HRV"], "Steak / beef", ["Potatoes"], [],
"Keep it simple", "omnivore", 25, 4, 18, 32, "{}",
),
"endurance": (
"I ran a half marathon this morning", ["Low energy", "Dehydrated"],
["Sleep", "Strain", "HRV"], "Eggs", ["Rice", "Sweet potato"], [],
"Carb-heavy", "vegetarian", 40, 7, 19, 41, "{}",
),
"stress": (
"A very stressful week at work", ["Stressed", "Low energy"],
["Sleep", "Strain", "HRV"], "Tofu", ["Spinach"], ["Caffeine"],
"One-pan meal", "Any", 30, 5, 8, 29, "{}",
),
}
THEME = gr.themes.Base(
primary_hue=gr.themes.colors.green,
secondary_hue=gr.themes.colors.emerald,
neutral_hue=gr.themes.colors.gray,
).set(
body_background_fill="#070b08",
body_text_color="#dfeee4",
background_fill_primary="#0d1410",
background_fill_secondary="#111a13",
block_background_fill="#0d1410",
block_border_color="#1f3324",
block_label_text_color="#3ddc84",
block_title_text_color="#3ddc84",
border_color_primary="#1f3324",
input_background_fill="#111a13",
input_border_color="#25402c",
input_placeholder_color="#6d8a79",
body_text_color_subdued="#8aa695",
block_info_text_color="#8aa695",
button_primary_background_fill="#1f9e5a",
button_primary_background_fill_hover="#28c46f",
button_primary_text_color="#04120a",
button_primary_border_color="#1f9e5a",
# Secondary buttons ("Read screenshot", "Start over") previously fell back to
# Gradio's light default, rendering as white blocks on the dark UI.
button_secondary_background_fill="#16241b",
button_secondary_background_fill_hover="#1e3527",
button_secondary_text_color="#dfeee4",
button_secondary_border_color="#2c4a35",
button_cancel_background_fill="#16241b",
button_cancel_background_fill_hover="#1e3527",
button_cancel_text_color="#dfeee4",
button_cancel_border_color="#2c4a35",
# Checkboxes / radio pills were unreadable (white chip, invisible label).
checkbox_background_color="#111a13",
checkbox_background_color_selected="#1f9e5a",
checkbox_background_color_hover="#1a2a1f",
checkbox_border_color="#2c4a35",
checkbox_border_color_selected="#3ddc84",
checkbox_border_color_hover="#3ddc84",
checkbox_label_background_fill="#111a13",
checkbox_label_background_fill_selected="#16301f",
checkbox_label_background_fill_hover="#1a2a1f",
checkbox_label_text_color="#dfeee4",
checkbox_label_text_color_selected="#eafff2",
checkbox_label_border_color="#2c4a35",
)
CSS = """
.gradio-container, body { background: #070b08 !important; }
#bb-hero {
background: linear-gradient(135deg, #0d1a12 0%, #070b08 70%);
border: 1px solid #1f3324; border-left: 4px solid #3ddc84;
border-radius: 16px; padding: 22px 24px; margin-bottom: 16px;
}
#bb-hero h1 { color: #3ddc84 !important; margin: 0 0 8px 0; font-size: 30px; }
#bb-hero p { color: #a9c6b5 !important; margin: 0; font-size: 15px; line-height: 1.6; }
#bb-hero b { color: #eafff2 !important; }
.bb-sec { color:#3ddc84 !important; font-weight:600; margin: 14px 0 6px 0 !important; }
.bb-step { background:#0d1410;border:1px solid #1f3324;border-radius:16px;padding:16px;margin:10px 0; }
.bb-match { background:#0d1a12;border:1px solid #245c38;border-radius:12px;
padding:13px 15px;margin:10px 0;color:#dfeee4; }
.bb-match b { color:#3ddc84;font-size:15px; }
.bb-match span { color:#a9c6b5;font-size:13px; }
footer { display: none !important; }
/* ---- Contrast fixes: nothing may render light-on-light in the dark theme ---- */
/* Secondary buttons (Read screenshot, Start over) */
button.secondary, .gr-button-secondary, button[class*="secondary"] {
background: #16241b !important;
color: #dfeee4 !important;
border: 1px solid #2c4a35 !important;
}
button.secondary:hover, .gr-button-secondary:hover, button[class*="secondary"]:hover {
background: #1e3527 !important;
border-color: #3ddc84 !important;
}
/* Checkbox / radio pills and their labels */
.gradio-container input[type="checkbox"], .gradio-container input[type="radio"] {
accent-color: #3ddc84 !important;
background-color: #111a13 !important;
border: 1px solid #2c4a35 !important;
}
.gradio-container label, .gradio-container label span,
.gradio-container .wrap label span, fieldset label span {
color: #dfeee4 !important;
}
.gradio-container fieldset label {
background: #111a13 !important;
border: 1px solid #2c4a35 !important;
border-radius: 8px !important;
}
.gradio-container fieldset label:has(input:checked) {
background: #16301f !important;
border-color: #3ddc84 !important;
}
/* File-upload dropzone */
.gradio-container .file-preview, .gradio-container [data-testid="block-label"] { color: #3ddc84 !important; }
/* Dropdown menus were light on light in some Gradio builds */
.gradio-container ul[role="listbox"], .gradio-container .options,
.gradio-container li[role="option"] {
background: #111a13 !important;
color: #dfeee4 !important;
}
.gradio-container li[role="option"]:hover,
.gradio-container li[role="option"][aria-selected="true"] {
background: #1e3527 !important;
color: #eafff2 !important;
}
/* Accordion headers */
.gradio-container .label-wrap, .gradio-container .label-wrap span { color: #3ddc84 !important; }
/* Primary result: recipe (wider) beside the plan on desktop, stacked on mobile. */
.bb-primary { align-items: flex-start; }
@media (max-width: 768px) {
.bb-primary { flex-direction: column !important; }
.bb-primary > div { width: 100% !important; min-width: 0 !important; }
}
"""
# Force the dark palette regardless of the visitor's browser setting
FORCE_DARK = """
function() {
const u = new URL(window.location);
if (u.searchParams.get('__theme') !== 'dark') {
u.searchParams.set('__theme', 'dark');
window.location.replace(u.href);
}
}
"""
with gr.Blocks(title="Bio-Bite", theme=THEME, css=CSS, js=FORCE_DARK) as demo:
# Keep State JSON-serializable as a string. Gradio 5.9's API-schema builder
# crashes on an unconstrained dict (additionalProperties: true), which makes
# every UI event appear as "No API found" even though the app has started.
wearable_source = gr.State("{}")
gr.HTML(
f"""
🥗 Bio-Bite — your recovery, on a plate
Your watch says you slept 5 hours and hit a strain of 18.
So what should you eat?
Bio-Bite turns your recovery data into a personalized meal and a plan for tomorrow.
"""
)
gr.Markdown("## 1 · Add your recovery data", elem_classes="bb-sec")
gr.Markdown(
"Upload a **WHOOP or watch screenshot**, or enter only the metrics you have. "
"You always review the detected values before they are used."
)
with gr.Row(elem_classes="bb-step"):
with gr.Column(scale=1):
screenshot = gr.Image(
type="pil",
sources=["upload", "clipboard"],
label="Recovery screenshot",
height=250,
)
read_screenshot = gr.Button("📷 Read screenshot", variant="secondary")
ocr_status = gr.HTML()
gr.Markdown(
"🔒 The image is processed for this request and is not added to the dataset or logs. "
"Direct WHOOP sign-in will be enabled only after official OAuth credentials are configured."
)
with gr.Column(scale=1):
metric_selection = gr.CheckboxGroup(
["Sleep", "Strain", "HRV"],
value=[],
label="Use these metrics",
info="Unchecked values are ignored—even if a number is visible.",
)
sleep_hours = gr.Number(
value=7, minimum=0, maximum=10, step=0.25,
label="Sleep last night (hours)",
)
strain = gr.Number(
value=10, minimum=0, maximum=21, step=0.1,
label="WHOOP Strain (0–21)",
)
hrv = gr.Number(
value=45, minimum=10, maximum=150, step=1,
label="HRV (ms)",
)
gr.Markdown("## 2 · Tell us how you feel", elem_classes="bb-sec")
with gr.Row(elem_classes="bb-step"):
with gr.Column(scale=1):
feelings = gr.CheckboxGroup(
["Sore muscles", "Low energy", "Stressed", "Well recovered", "Dehydrated"],
label="Today I feel…",
)
with gr.Column(scale=1):
state = gr.Textbox(
label="Anything else? — optional",
placeholder="e.g. Heavy leg day, 10 km run, rest day…",
lines=2,
)
gr.Markdown("## 3 · Choose the meal", elem_classes="bb-sec")
with gr.Column(elem_classes="bb-step"):
with gr.Row():
diet = gr.Dropdown(
["Any", "omnivore", "vegetarian", "vegan", "pescatarian", "gluten-free"],
value="Any", label="Diet",
)
max_prep = gr.Slider(10, 60, value=30, step=5, label="Max prep time (min)")
servings = gr.Slider(1, 4, value=2, step=1, label="Servings")
with gr.Row():
main_ingredient = gr.Dropdown(
MAIN_INGREDIENT_CHOICES,
value="No preference",
label="Main ingredient — guaranteed in the personalized recipe",
)
include_ingredients = gr.Dropdown(
EXTRA_INGREDIENT_CHOICES,
multiselect=True,
max_choices=3,
value=[],
label="Also include — up to 3",
)
excluded_ingredients = gr.Dropdown(
EXCLUDED_INGREDIENT_CHOICES,
multiselect=True,
value=[],
label="Avoid / allergies",
)
with gr.Row():
cuisine = gr.Dropdown(
["No preference", "Mediterranean", "Asian-inspired", "Mexican-inspired", "Middle Eastern"],
value="No preference",
label="Cuisine style",
)
equipment = gr.Dropdown(
["Any equipment", "One pan", "Microwave only", "Oven", "No-cook"],
value="Any equipment",
label="Available setup",
)
constraint = gr.Textbox(
label="Other requirement — optional",
placeholder="e.g. spicy, mild flavors, high-protein",
lines=1,
)
preference_status = gr.HTML()
with gr.Row():
btn = gr.Button("🍽️ Generate My Bio-Bite", variant="primary", size="lg", scale=4)
clear_btn = gr.Button("Start over", variant="secondary", scale=1)
with gr.Accordion("⚡ Try a ready-made demo scenario", open=False):
with gr.Row():
starter_strength = gr.Button("🏋️ Strength + poor sleep")
starter_endurance = gr.Button("🏃 Endurance + dehydration")
starter_stress = gr.Button("🧠 Stress + low HRV")
# Results hierarchy: compact summary → primary result (recipe + plan side by
# side) → the three dataset matches in a closed Accordion → technical trace.
out_recs = gr.HTML()
with gr.Row(elem_classes="bb-primary"):
with gr.Column(scale=3, min_width=320):
gr.Markdown("### ✨ Your personalized Bio-Bite", elem_classes="bb-sec")
out_recipe = gr.HTML()
with gr.Column(scale=2, min_width=260):
gr.Markdown("### 📅 Your plan for tomorrow", elem_classes="bb-sec")
out_plan = gr.HTML()
with gr.Accordion("How we created this recommendation — 3 dataset matches",
open=False):
out_cards = gr.HTML()
with gr.Accordion("How Bio-Bite decided — technical trace", open=False):
out_technical = gr.HTML()
gr.HTML(
f"""
⚠️
Educational prototype — not medical, nutritional or
training advice. Recipes come from a synthetic dataset generated by a language
model and have not been reviewed by a registered dietitian. Consult a qualified
professional for personal guidance.
Dataset:
benjac8/bio-bite-recovery-nutrition
· Embeddings: {html.escape(EMBED_MODEL)} · Generation: {html.escape(GEN_MODEL)}