app.py

"""
Pantry API — Qwen2.5-VL-7B-Instruct su Hugging Face Spaces.

Stateless: nessun gr.State, nessuna cronologia. Il contesto della dispensa arriva
a ogni chiamata e non viene conservato dal codice applicativo.

Endpoint:
  /genera         proxy 1:1 con AiRequest dell'app Android
  /analizza_foto  foto + contesto -> {"prodotti":[...]}
  /piano          contesto -> {"giorni":[...]}
  /comando        contesto + messaggio -> {"azione":...,"prodotti":[...]}
  /chat           contesto + domanda -> testo
"""

from __future__ import annotations

import functools
import json
import os
import re
import threading
from typing import Optional

os.environ.setdefault("GRADIO_ANALYTICS_ENABLED", "False")

import gradio as gr
import torch
from PIL import Image
from qwen_vl_utils import process_vision_info
from transformers import AutoProcessor, Qwen2_5_VLForConditionalGeneration

try:
    import spaces

    def gpu(fn=None, **kwargs):
        return spaces.GPU(**kwargs)(fn) if fn is not None else spaces.GPU(**kwargs)
except ImportError:
    # Permette di aprire il file in locale senza il pacchetto spaces.
    def gpu(fn=None, **kwargs):
        def deco(f):
            @functools.wraps(f)
            def wrapper(*args, **kw):
                return f(*args, **kw)
            return wrapper
        return deco(fn) if fn is not None else deco


MODEL_ID = os.getenv("PANTRY_MODEL_ID", "Qwen/Qwen2.5-VL-7B-Instruct")
QUANTIZZAZIONE = os.getenv("PANTRY_QUANTIZZAZIONE", "bf16").lower()
MAX_TOKENS_CAP = 8192
DEFAULT_MAX_TOKENS = 2048

_modello = None
_processore = None
_lock_modello = threading.Lock()


def _carica_modello():
    """Caricamento pigro: la GPU di ZeroGPU esiste solo dentro @gpu."""
    global _modello, _processore
    if _modello is not None:
        return _modello, _processore
    with _lock_modello:
        if _modello is not None:
            return _modello, _processore
        if not torch.cuda.is_available():
            raise gr.Error(
                "Questo Space richiede una GPU. Imposta Hardware su GPU "
                "(ZeroGPU o dedicata) nelle impostazioni dello Space."
            )
        dtype = torch.bfloat16 if torch.cuda.is_bf16_supported() else torch.float16
        _processore = AutoProcessor.from_pretrained(MODEL_ID, trust_remote_code=True)
        parametri = {
            "torch_dtype": dtype,
            "device_map": "auto",
            "trust_remote_code": True,
        }
        if QUANTIZZAZIONE == "4bit":
            from transformers import BitsAndBytesConfig

            parametri.pop("torch_dtype", None)
            parametri["quantization_config"] = BitsAndBytesConfig(
                load_in_4bit=True,
                bnb_4bit_compute_dtype=dtype,
                bnb_4bit_quant_type="nf4",
            )
        _modello = Qwen2_5_VLForConditionalGeneration.from_pretrained(
            MODEL_ID, **parametri
        ).eval()
    return _modello, _processore


# ---------------------------------------------------------------------------
# Schemi: devono restare allineati a Schemi.kt dell'app.
# ---------------------------------------------------------------------------

SCHEMA_PRODOTTI = r"""
{
  "type": "object",
  "properties": {
    "azione": {
      "type": "string",
      "enum": ["aggiungi", "finito", "usa", "nessuna"]
    },
    "prodotti": {
      "type": "array",
      "items": {
        "type": "object",
        "properties": {
          "nome": {"type": "string"},
          "quantita": {"type": "number"},
          "unita": {"type": "string", "enum": ["g", "ml", "pz"]},
          "posto": {"type": "string"},
          "scadenza": {"type": "string"},
          "sicurezza": {"type": "number"}
        },
        "required": ["nome"]
      }
    }
  },
  "required": ["azione", "prodotti"]
}
"""

SCHEMA_PIANO = r"""
{
  "type": "object",
  "properties": {
    "giorni": {
      "type": "array",
      "items": {
        "type": "object",
        "properties": {
          "numero": {"type": "integer"},
          "pasti": {
            "type": "array",
            "items": {
              "type": "object",
              "properties": {
                "tipo": {
                  "type": "string",
                  "enum": ["colazione", "pranzo", "cena", "spuntino"]
                },
                "nome": {"type": "string"},
                "ingredienti": {
                  "type": "array",
                  "items": {
                    "type": "object",
                    "properties": {
                      "voceId": {"type": "integer"},
                      "nome": {"type": "string"},
                      "quantita": {"type": "number"},
                      "unita": {"type": "string", "enum": ["g", "ml", "pz"]}
                    },
                    "required": ["voceId", "nome", "quantita", "unita"]
                  }
                }
              },
              "required": ["tipo", "nome", "ingredienti"]
            }
          }
        },
        "required": ["numero", "pasti"]
      }
    }
  },
  "required": ["giorni"]
}
"""


# ---------------------------------------------------------------------------
# Prompt e generazione
# ---------------------------------------------------------------------------

def _componi_sistema(system_instruction: str, json_output: bool, schema_json: str) -> str:
    sistema = (system_instruction or "").strip()
    if json_output:
        sistema += (
            "\n\nRispondi SOLO con un oggetto JSON valido, senza markdown, "
            "senza commenti e senza testo prima o dopo. Usa esattamente le "
            "chiavi e i tipi dello schema."
        )
    if schema_json and schema_json.strip():
        sistema += "\n\nSchema JSON da rispettare:\n" + schema_json.strip()
    return sistema


def _prepara_ingressi(
    system_instruction: str,
    user_text: str,
    immagine: Optional[Image.Image],
    json_output: bool,
    schema_json: str,
):
    modello, processore = _carica_modello()
    sistema = _componi_sistema(system_instruction, json_output, schema_json)

    contenuto = []
    if immagine is not None:
        contenuto.append({"type": "image", "image": immagine})
    contenuto.append({"type": "text", "text": user_text})

    messaggi = [
        {"role": "system", "content": [{"type": "text", "text": sistema}]},
        {"role": "user", "content": contenuto},
    ]

    testo_prompt = processore.apply_chat_template(
        messaggi, tokenize=False, add_generation_prompt=True
    )
    immagini, video = process_vision_info(messaggi)
    ingressi = processore(
        text=[testo_prompt],
        images=immagini,
        videos=video,
        padding=True,
        return_tensors="pt",
    ).to(modello.device)
    return modello, processore, ingressi


def _genera_grezzo(
    system_instruction: str,
    user_text: str,
    immagine: Optional[Image.Image],
    json_output: bool,
    schema_json: str,
    max_tokens: int,
    temperatura: float,
) -> str:
    modello, processore, ingressi = _prepara_ingressi(
        system_instruction, user_text, immagine, json_output, schema_json
    )
    token_max = max(16, min(int(max_tokens or DEFAULT_MAX_TOKENS), MAX_TOKENS_CAP))
    parametri = dict(max_new_tokens=token_max)
    if temperatura > 0:
        parametri.update(
            do_sample=True,
            temperature=max(0.0, min(float(temperatura), 1.5)),
            top_p=0.95,
        )
    else:
        parametri.update(do_sample=False)

    with torch.inference_mode():
        generati = modello.generate(**ingressi, **parametri)

    ritagliati = [
        uscita[len(ingresso):]
        for ingresso, uscita in zip(ingressi.input_ids, generati)
    ]
    testo = processore.batch_decode(
        ritagliati,
        skip_special_tokens=True,
        clean_up_tokenization_spaces=False,
    )[0]
    return testo.strip()


def _estrai_json(testo: str) -> str:
    testo = (testo or "").strip()
    if testo.startswith("```"):
        testo = re.sub(r"^```(?:json)?\s*", "", testo, flags=re.IGNORECASE)
        testo = re.sub(r"\s*```$", "", testo)
    inizio = testo.find("{")
    fine = testo.rfind("}")
    if inizio >= 0 and fine > inizio:
        return testo[inizio:fine + 1]
    return testo


def _valida_json(testo: str, schema_json: str) -> str:
    if not (schema_json or "").strip():
        return testo
    import jsonschema

    schema = json.loads(schema_json)
    dati = json.loads(testo)
    jsonschema.validate(dati, schema)
    return json.dumps(dati, ensure_ascii=False)


def _genera_con_validazione(
    system_instruction: str,
    user_text: str,
    immagine: Optional[Image.Image],
    json_output: bool,
    schema_json: str,
    max_tokens: int,
    temperatura: float,
) -> str:
    primo = _genera_grezzo(
        system_instruction, user_text, immagine, json_output,
        schema_json, max_tokens, temperatura,
    )
    if not json_output:
        return primo

    candidato = _estrai_json(primo)
    try:
        return _valida_json(candidato, schema_json)
    except Exception as errore:
        # Un solo tentativo di riparazione: il modello vede l'errore e riscrive.
        riparazione = _genera_grezzo(
            system_instruction=system_instruction,
            user_text=(
                f"{user_text}\n\nLa risposta precedente era:\n{primo}\n\n"
                f"Non è valida: {errore}. Riscrivi l'oggetto completo, solo JSON."
            ),
            immagine=immagine,
            json_output=True,
            schema_json=schema_json,
            max_tokens=max_tokens,
            temperatura=0.0,
        )
        try:
            return _valida_json(_estrai_json(riparazione), schema_json)
        except Exception as secondo:
            raise gr.Error(f"Il modello non ha prodotto un JSON valido: {secondo}") from secondo


# ---------------------------------------------------------------------------
# Endpoint
# ---------------------------------------------------------------------------

@gpu(duration=60)
def genera(
    system_instruction: str,
    user_text: str,
    immagine: Optional[Image.Image],
    json_output: bool,
    schema_json: str,
    max_tokens: int = DEFAULT_MAX_TOKENS,
    temperatura: float = 0.2,
) -> str:
    return _genera_con_validazione(
        system_instruction, user_text, immagine, json_output,
        schema_json, max_tokens, temperatura,
    )


@gpu(duration=60)
def analizza_foto(immagine: Optional[Image.Image], contesto: str) -> str:
    if immagine is None:
        raise gr.Error("Carica una foto del frigo, della dispensa o dello scontrino.")
    sistema = (
        "Sei l'assistente della dispensa di casa.\n"
        "Leggi la foto della spesa e riconosci i prodotti acquistati.\n"
        "Per ogni prodotto scrivi nome, quantita' e unita' (solo g, ml o pz).\n"
        "Scegli il posto fra quelli elencati nel contesto.\n"
        "Aggiungi sicurezza da 0 a 1: quanto sei sicuro della lettura.\n"
        "Per una foto, azione vale sempre 'nessuna'.\n"
        "Non inventare prodotti che nella foto non si vedono.\n"
        "Rispondi solo con il JSON dello schema, senza spiegazioni.\n\n"
        + (contesto or "")
    )
    return _genera_con_validazione(
        sistema,
        "Leggi questa foto della spesa e riconosci i prodotti.",
        immagine,
        True,
        SCHEMA_PRODOTTI,
        1024,
        0.0,
    )


@gpu(duration=60)
def piano(contesto: str) -> str:
    sistema = (
        "Sei il coach della dispensa di casa.\n"
        "Prepara un piano di 3 giorni: colazione, pranzo e cena per ogni giorno.\n"
        "Aggiungi gli spuntini solo se il profilo prevede piu' di 3 pasti al giorno.\n"
        "Rispetta calorie, macro, tempo ai fornelli e livello di cucina del profilo.\n"
        "Sale, olio e acqua si considerano sempre presenti.\n"
        "Per ogni altro ingrediente usa SOLO le voci dell'elenco, indicando il loro "
        "identificativo in voceId.\n"
        "Rispondi solo con il JSON dello schema, senza spiegazioni.\n\n"
        + (contesto or "")
    )
    return _genera_con_validazione(
        sistema,
        "Prepara il piano dei 3 giorni.",
        None,
        True,
        SCHEMA_PIANO,
        MAX_TOKENS_CAP,
        0.2,
    )


@gpu(duration=60)
def comando(contesto: str, messaggio: str) -> str:
    sistema = (
        "Sei l'assistente della dispensa di casa.\n"
        "Capisci se il messaggio dell'utente modifica la dispensa e lo traduci "
        "nei dati richiesti.\n"
        "- azione 'aggiungi': ha comprato qualcosa (serve quantita' e unita').\n"
        "- azione 'finito': una cosa e' finita (basta il nome).\n"
        "- azione 'usa': ne ha consumata una quantita' (serve quantita' e unita').\n"
        "- azione 'nessuna': non e' un comando sulla dispensa.\n"
        "Usa solo i posti e le unita' elencati; non convertire mai fra g, ml e pz.\n"
        "Rispondi solo con il JSON dello schema, senza spiegazioni.\n\n"
        + (contesto or "")
    )
    return _genera_con_validazione(
        sistema, messaggio, None, True, SCHEMA_PRODOTTI, 1024, 0.0
    )


@gpu(duration=60)
def chat(contesto: str, domanda: str) -> str:
    sistema = (
        "Sei il coach della dispensa di casa.\n"
        "Rispondi in italiano, in modo concreto e breve.\n"
        "Usa SOLO le voci elencate nel contesto, con le quantita' indicate.\n"
        "Sale, olio e acqua si considerano sempre presenti.\n"
        "Se un piatto richiede qualcosa che non c'e', non proporlo.\n\n"
        + (contesto or "")
    )
    return _genera_con_validazione(
        sistema, domanda, None, False, "", 2048, 0.4
    )


# ---------------------------------------------------------------------------
# Interfaccia
# ---------------------------------------------------------------------------

with gr.Blocks(title="Pantry API — Qwen 7B") as demo:
    gr.Markdown(
        "# Pantry API\n"
        "Qwen2.5-VL-7B-Instruct. Lo Space è senza stato: a ogni chiamata "
        "arriva il contesto completo della dispensa."
    )

    with gr.Tab("Generico"):
        g_sistema = gr.Textbox(label="System instruction", lines=6)
        g_utente = gr.Textbox(label="User text", lines=4)
        g_immagine = gr.Image(type="pil", label="Immagine (opzionale)")
        g_json = gr.Checkbox(label="JSON output", value=False)
        g_schema = gr.Textbox(label="Schema JSON", lines=4)
        g_max = gr.Slider(64, MAX_TOKENS_CAP, value=DEFAULT_MAX_TOKENS, step=64,
                          label="Max new tokens")
        g_temp = gr.Slider(0.0, 1.0, value=0.2, step=0.1, label="Temperatura")
        g_out = gr.Textbox(label="Risposta", lines=14)
        gr.Button("Genera").click(
            genera,
            [g_sistema, g_utente, g_immagine, g_json, g_schema, g_max, g_temp],
            g_out,
            api_name="/genera",
        )

    with gr.Tab("Foto"):
        f_immagine = gr.Image(type="pil", label="Foto frigo / dispensa / scontrino")
        f_contesto = gr.Textbox(label="Contesto dispensa", lines=8)
        f_out = gr.Code(label="Prodotti (JSON)", language="json")
        gr.Button("Analizza").click(
            analizza_foto, [f_immagine, f_contesto], f_out,
            api_name="/analizza_foto",
        )

    with gr.Tab("Piano"):
        p_contesto = gr.Textbox(label="Contesto dispensa", lines=10)
        p_out = gr.Code(label="Piano (JSON)", language="json")
        gr.Button("Prepara piano").click(
            piano, [p_contesto], p_out, api_name="/piano"
        )

    with gr.Tab("Comando"):
        c_contesto = gr.Textbox(label="Contesto dispensa", lines=8)
        c_messaggio = gr.Textbox(label="Messaggio")
        c_out = gr.Code(label="Comando (JSON)", language="json")
        gr.Button("Interpreta").click(
            comando, [c_contesto, c_messaggio], c_out, api_name="/comando"
        )

    with gr.Tab("Chat"):
        h_contesto = gr.Textbox(label="Contesto dispensa", lines=8)
        h_domanda = gr.Textbox(label="Domanda")
        h_out = gr.Textbox(label="Risposta", lines=12)
        gr.Button("Chiedi").click(
            chat, [h_contesto, h_domanda], h_out, api_name="/chat"
        )


demo.queue(default_concurrency_limit=1, max_size=8)

if __name__ == "__main__":
    demo.launch()