modules/advisor.py

import base64
import datetime
import os
from io import BytesIO

from huggingface_hub import InferenceClient
from PIL import Image

ADVISOR_MODEL_ID = os.getenv("ADVISOR_MODEL_ID", "Qwen/Qwen2.5-Coder-3B-Instruct")
ADVISOR_PROVIDER = os.getenv("ADVISOR_PROVIDER", "nscale")

_client = None


def _get_client() -> InferenceClient:
    global _client
    if _client is None:
        _client = InferenceClient(model=ADVISOR_MODEL_ID, provider=ADVISOR_PROVIDER, token=os.getenv("HF_TOKEN"))
    return _client


def _watering_status(last_watered: str | None) -> str:
    if not last_watered:
        return "Never watered yet (no record)."
    days_since = (datetime.date.today() - datetime.date.fromisoformat(last_watered)).days
    if days_since <= 0:
        return f"Watered today ({last_watered})."
    if days_since == 1:
        return f"Last watered yesterday ({last_watered})."
    return f"Last watered {days_since} days ago (on {last_watered})."


def _build_system_prompt(
    plant_info: dict,
    plant_name: str | None = None,
    genus: str | None = None,
    last_watered: str | None = None,
    neighbors: list[dict] | None = None,
) -> str:
    name = plant_name or genus or "this plant"
    neighbor_line = ""
    if neighbors:
        names = ", ".join(f"{n['name']} ({n['genus']})" for n in neighbors)
        neighbor_line = f"- Plants growing right next to it in the garden: {names}\n"
    return (
        "You are an expert gardening assistant with deep knowledge of houseplant "
        "and garden plant care. Be practical, encouraging, and specific.\n\n"
        f"The user is asking about their plant: {name} (genus: {genus}). "
        "Known care profile for this plant:\n"
        f"- Sunlight: {plant_info.get('sunlight')}\n"
        f"- Soil: {plant_info.get('soil')}\n"
        f"- Watering frequency: every {plant_info.get('watering_frequency_days')} days\n"
        f"- Fertilization: {plant_info.get('fertilization_type')}\n"
        f"- Watering status: {_watering_status(last_watered)}\n"
        f"{neighbor_line}\n"
        "Always factor the watering status into your answer: if it is overdue "
        "compared to the recommended frequency, say so and recommend watering; "
        "if it was watered recently, take that into account (don't tell the "
        "user to water again right away) and consider overwatering as a "
        "possible cause if the question describes a problem.\n\n"
        "Use this profile as context, but also draw on your general gardening "
        "knowledge for issues it doesn't cover (pests, diseases, yellowing "
        "leaves, repotting, etc.). If nearby plants are listed, factor in "
        "companion-planting effects (competition for light, water or nutrients, "
        "shared pests/diseases, or beneficial pairings) when relevant to the "
        "question. Give a precise, actionable, gardening-advice "
        "focused answer and never recommend dangerous or toxic substances. "
        "Answer in 2-4 sentences, in the same language as the question."
    )


def ask_about_plant(
    question: str,
    plant_info: dict,
    plant_name: str | None = None,
    genus: str | None = None,
    last_watered: str | None = None,
    neighbors: list[dict] | None = None,
) -> str:
    """Ask the advisor a question about a specific plant, grounded in its care data."""
    try:
        completion = _get_client().chat_completion(
            messages=[
                {"role": "system", "content": _build_system_prompt(plant_info, plant_name, genus, last_watered, neighbors)},
                {"role": "user", "content": question},
            ],
            max_tokens=300,
        )
        return completion.choices[0].message.content
    except Exception as e:
        # TEMP debug: surface the real HF Inference error in the Space logs
        print(f"[advisor] HF Inference error: {e!r}")
        return "Sorry, the assistant is unavailable right now — please try again later."


def diagnose_plant_health(
    image: Image.Image,
    plant_name: str | None = None,
    genus: str | None = None,
) -> str:
    """Ask a vision-language model to assess a plant's health from a photo."""
    name = plant_name or genus or "this plant"
    buf = BytesIO()
    image.convert("RGB").save(buf, format="JPEG")
    b64 = base64.b64encode(buf.getvalue()).decode()

    prompt = (
        f"This is a photo of a houseplant named '{name}' (genus: {genus}). "
        "Look closely at its leaves, stems and soil for signs of trouble: "
        "yellowing or browning leaves, wilting, spots, pests, mold, or "
        "dry/overwatered soil. Start your reply with exactly one status word "
        "on its own — 'Healthy', 'Needs attention', or 'Sick' — then a short "
        "explanation (1-3 sentences) of what you see and what to do about it."
    )

    try:
        completion = _get_client().chat_completion(
            messages=[{
                "role": "user",
                "content": [
                    {"type": "text", "text": prompt},
                    {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}},
                ],
            }],
            max_tokens=250,
        )
        return completion.choices[0].message.content.strip()
    except Exception as e:
        print(f"[advisor] HF Inference health-check error: {e!r}")
        return "Sorry, the health check is unavailable right now — please try again later."