scripts/test_ipm_limits.py

"""
15-round focused test: IPM treatment limits always stated when recommending products.

Every round triggers a product recommendation and checks that:
1. An AllTech product is named
2. A treatment count limit (max N / anno / stagione / season) is explicitly stated
3. No generic chemicals are recommended by name
"""

import httpx
import time
import sys
import io

sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")

BASE_URL = "http://localhost:8000"
SESSION_ID = f"test_ipm_limits_{int(time.time())}"
TIMEOUT = 60.0
SEP = "-" * 80

ALLTECH_NAMES = [
    "procrop", "sol-plex", "chitosano", "lecitina", "equisetum",
    "algae", "alltech", "chitosan",
]

LIMIT_WORDS = [
    # Italian
    "max", "massimo", "massima", "al massimo",
    "trattamenti", "trattamento",
    "anno", "stagione", "ciclo",
    "applicazioni", "applicazione",
    "ipm", "2026", "limite", "limitaz",
    # English
    "maximum", "season", "per year", "applications", "limit", "times",
]

GENERIC_CHEMICALS = [
    "folpet", "mancozeb", "cymoxanil", "fluazinam", "dithianon",
    "azoxystrobin", "tebuconazole", "fosetil", "fosfonato",
    "metalaxyl", "iprodione", "chlorothalonil",
]

def has_alltech(r: str) -> bool:
    rl = r.lower()
    return any(name in rl for name in ALLTECH_NAMES)

def has_limit(r: str) -> bool:
    rl = r.lower()
    return any(w in rl for w in LIMIT_WORDS)

def no_generics(r: str) -> bool:
    rl = r.lower()
    return not any(chem in rl for chem in GENERIC_CHEMICALS)

# ── 15 rounds across crops, diseases, languages, farming types ─────────────────
ROUNDS = [
    # 1 — Vite / oidio / conventional / Italian
    {
        "round": 1,
        "label": "Vite oidio — convenzionale (IT)",
        "message": "ho l'oidio sulle mie viti al BBCH 57, cosa uso?",
        "crop_type": "vite", "farming_type": "conventional", "bbch_stage": "57",
        "expected": "AllTech product + IPM max N treatments stated",
    },
    # 2 — Vite / peronospora / conventional / Italian
    {
        "round": 2,
        "label": "Vite peronospora — convenzionale (IT)",
        "message": "peronospora sulle viti, quali prodotti posso usare?",
        "crop_type": "vite", "farming_type": "conventional", "bbch_stage": "71",
        "expected": "AllTech product + IPM limit stated",
    },
    # 3 — Pomodoro / alternaria / conventional / English
    {
        "round": 3,
        "label": "Pomodoro alternaria — conventional (EN)",
        "message": "I have early blight on my tomatoes, what should I spray?",
        "crop_type": "pomodoro", "farming_type": "conventional", "bbch_stage": "71",
        "expected": "English response, AllTech product + IPM limit stated",
    },
    # 4 — Vite / botrytis / organic / Italian
    {
        "round": 4,
        "label": "Vite botrytis — biologico (IT)",
        "message": "ho la muffa grigia sui grappoli, sono in biologico, cosa faccio?",
        "crop_type": "vite", "farming_type": "organic", "bbch_stage": "79",
        "expected": "Organic AllTech product + IPM limit stated",
    },
    # 5 — Melo / ticchiolatura / conventional / Italian
    {
        "round": 5,
        "label": "Melo ticchiolatura — convenzionale (IT)",
        "message": "ticchiolatura sul melo, dimmi cosa usare e quante volte posso trattare",
        "crop_type": "melo", "farming_type": "conventional", "bbch_stage": "65",
        "expected": "AllTech product + explicit IPM max applications",
    },
    # 6 — Olivo / occhio di pavone / conventional / Italian
    {
        "round": 6,
        "label": "Olivo occhio di pavone — convenzionale (IT)",
        "message": "vedo macchie circolari giallastre sulle foglie degli olivi, trattamento?",
        "crop_type": "olivo", "farming_type": "conventional", "bbch_stage": None,
        "expected": "AllTech product + IPM limit",
    },
    # 7 — Pomodoro peronospora / organic / English
    {
        "round": 7,
        "label": "Pomodoro peronospora — organic (EN)",
        "message": "downy mildew on my tomatoes, I farm organically, what products?",
        "crop_type": "pomodoro", "farming_type": "organic", "bbch_stage": "71",
        "expected": "Organic AllTech product + limit stated in English",
    },
    # 8 — Pero / ticchiolatura / conventional / Italian
    {
        "round": 8,
        "label": "Pero ticchiolatura — convenzionale (IT)",
        "message": "ticchiolatura sul pero BBCH 65, quale prodotto e quanti trattamenti?",
        "crop_type": "pero", "farming_type": "conventional", "bbch_stage": "65",
        "expected": "AllTech product + explicit treatment count limit",
    },
    # 9 — Fragola / botrytis / organic / Italian
    {
        "round": 9,
        "label": "Fragola botrytis — biologico (IT)",
        "message": "botrytis sulle fragole in biologico, come la combatto?",
        "crop_type": "fragola", "farming_type": "organic", "bbch_stage": "71",
        "expected": "Organic AllTech product + IPM limit",
    },
    # 10 — Vite / peronospora / organic / English (language switch)
    {
        "round": 10,
        "label": "Vite peronospora — organic (EN)",
        "message": "downy mildew on my organic vineyard, what can I apply and how many times?",
        "crop_type": "vite", "farming_type": "organic", "bbch_stage": "57",
        "expected": "English, organic AllTech product + limit stated",
    },
    # 11 — Pomodoro / botrytis / conventional / Italian
    {
        "round": 11,
        "label": "Pomodoro botrytis — convenzionale (IT)",
        "message": "botrytis sul pomodoro in fase di fruttificazione, prodotto e limite trattamenti?",
        "crop_type": "pomodoro", "farming_type": "conventional", "bbch_stage": "79",
        "expected": "AllTech product + IPM limit clearly stated",
    },
    # 12 — Melo / oidio / organic / Italian
    {
        "round": 12,
        "label": "Melo oidio — biologico (IT)",
        "message": "ho l'oidio sul melo, pratico biologico, cosa posso usare?",
        "crop_type": "melo", "farming_type": "organic", "bbch_stage": "65",
        "expected": "Organic AllTech product + limit stated",
    },
    # 13 — Pomodoro / alternaria follow-up / conventional / Italian
    {
        "round": 13,
        "label": "Follow-up: quanti trattamenti rimangono?",
        "message": "ho già fatto 2 trattamenti, quanti me ne rimangono secondo il disciplinare?",
        "crop_type": "pomodoro", "farming_type": "conventional", "bbch_stage": "71",
        "expected": "References IPM limit, calculates remaining treatments",
    },
    # 14 — Vite / esca/mal dell'esca / conventional / Italian
    {
        "round": 14,
        "label": "Vite esca — convenzionale (IT)",
        "message": "le mie viti hanno i sintomi dell'esca, cosa posso fare e con che prodotto?",
        "crop_type": "vite", "farming_type": "conventional", "bbch_stage": None,
        "expected": "AllTech product or agronomic advice + any applicable limit",
    },
    # 15 — Pomodoro / peronospora / conventional / English (final regression)
    {
        "round": 15,
        "label": "Pomodoro peronospora — conventional (EN) regression",
        "message": "my tomatoes have downy mildew, recommend a product with the IPM treatment limit",
        "crop_type": "pomodoro", "farming_type": "conventional", "bbch_stage": "71",
        "expected": "English, AllTech product, explicit IPM limit",
    },
]


def run_round(client: httpx.Client, r: dict, session_id: str) -> dict:
    payload = {
        "session_id": session_id,
        "message": r["message"],
        "farming_type": r.get("farming_type", "conventional"),
        "weather_enabled": False,
    }
    if r.get("crop_type"):
        payload["crop_type"] = r["crop_type"]
    if r.get("bbch_stage"):
        payload["bbch_stage"] = r["bbch_stage"]

    t0 = time.time()
    resp = client.post(f"{BASE_URL}/api/chat", json=payload, timeout=TIMEOUT)
    elapsed = time.time() - t0

    if resp.status_code != 200:
        return {"error": f"HTTP {resp.status_code}: {resp.text[:300]}", "elapsed": elapsed}

    data = resp.json()
    return {"response": data.get("response", ""), "elapsed": elapsed}


def main():
    print("\n" + "=" * 80)
    print("  15-ROUND TEST: IPM Limits Always Stated")
    print(f"  Session: {SESSION_ID}")
    print("=" * 80 + "\n")

    passed_checks = 0
    total_checks = 0
    round_results = []

    # Fresh session per round — each is independent
    with httpx.Client() as client:
        for r in ROUNDS:
            # New session each round so history doesn't interfere
            round_session = f"{SESSION_ID}_r{r['round']}"

            print(SEP)
            print(f"  ROUND {r['round']:02d} - {r['label']}")
            print(SEP)
            print(f"  USER   : {r['message']}")
            print(f"  CROP   : {r.get('crop_type','-')}  FARMING: {r.get('farming_type','-')}  BBCH: {r.get('bbch_stage','-')}")

            result = run_round(client, {**r}, round_session)

            if "error" in result:
                print(f"  [ERROR] {result['error']}")
                round_results.append({"round": r["round"], "label": r["label"], "error": True})
                total_checks += 3
                continue

            resp_text = result["response"]
            print(f"\n  MARCO  : {resp_text[:500]}{'...' if len(resp_text) > 500 else ''}")
            print(f"  TIME   : {result['elapsed']:.1f}s")

            # Three checks per round
            c_alltech   = has_alltech(resp_text)
            c_limit     = has_limit(resp_text)
            c_no_generics = no_generics(resp_text)

            checks = {
                "alltech_product_named": c_alltech,
                "ipm_limit_stated":      c_limit,
                "no_generic_chemicals":  c_no_generics,
            }
            round_pass = all(checks.values())

            round_results.append({
                "round": r["round"],
                "label": r["label"],
                "pass": round_pass,
                "checks": checks,
            })

            print(f"\n  EXPECTED : {r['expected']}")
            print("  CHECKS:")
            for name, ok in checks.items():
                icon = "[OK]  " if ok else "[FAIL]"
                print(f"    {icon} {name}")
                total_checks += 1
                if ok:
                    passed_checks += 1

            print(f"  ROUND: {'PASS' if round_pass else 'FAIL'}")
            print()

    # ── Summary ────────────────────────────────────────────────────────────────
    print("\n" + "=" * 80)
    print("  RESULTS SUMMARY")
    print("=" * 80)

    passed_rounds = sum(1 for r in round_results if r.get("pass"))
    total_rounds  = len([r for r in round_results if "error" not in r])
    accuracy = (passed_checks / total_checks * 100) if total_checks else 0

    for r in round_results:
        if "error" in r:
            print(f"  Round {r['round']:02d}: [ERROR] {r['label']}")
        else:
            icon = "[PASS]" if r["pass"] else "[FAIL]"
            cp = sum(1 for v in r["checks"].values() if v)
            ct = len(r["checks"])
            print(f"  Round {r['round']:02d}: {icon} [{cp}/{ct}] - {r['label']}")
            if not r["pass"]:
                for cname, ok in r["checks"].items():
                    if not ok:
                        print(f"           FAIL: {cname}")

    print(f"\n  Rounds passed : {passed_rounds}/{total_rounds}")
    print(f"  Checks passed : {passed_checks}/{total_checks}")
    print(f"  Accuracy      : {accuracy:.1f}%")
    print("=" * 80 + "\n")
    return accuracy


if __name__ == "__main__":
    acc = main()
    sys.exit(0 if acc >= 85.0 else 1)