app.py

import json
import os
import re

import gradio as gr
from groq import Groq

# ============================================================
# GROQ CONFIG
# HuggingFace Spaces: Settings → Variables and Secrets
#   → New Secret → Name: GROQ_API_KEY → Value: gsk_...
# ============================================================
GROQ_API_KEY = os.environ.get("GROQ_API_KEY", "")
GROQ_MODEL   = "llama-3.1-8b-instant"

_client = None

def get_client():
    global _client
    if _client is None:
        if not GROQ_API_KEY:
            raise ValueError("GROQ_API_KEY not set. Add it in Space Settings → Secrets.")
        _client = Groq(api_key=GROQ_API_KEY)
    return _client

# ============================================================
FLAG_EXPLANATIONS = {
    "NEG_DEV":            "Current score is below the Gold baseline for this checkpoint.",
    "REL_COLLAPSE":       "Recent performance is dropping relative to earlier checkpoints.",
    "STRONG_COLLAPSE":    "Performance collapse is severe and sustained.",
    "LOW_AND_COLLAPSING": "Player is both below baseline and still getting worse.",
    "DEATH_SPIKE":        "Deaths increased quickly over the last 6-minute window.",
    "UNSUPPORTED_DEATHS": "Many deaths happened without nearby team support or trade value.",
    "LOW_IMPACT":         "Recent kills + assists are too low during a bad stretch.",
    "NO_OBJECTIVE":       "No objective participation detected in midgame.",
    "LOW_FARM":           "Farm gain is stagnating over the recent 6-minute window.",
    "NO_VISION":          "Vision contribution is too low for jungle/support while already collapsing.",
}

SAMPLE_ROWS = [
    {"minute":6,"role":"JUNGLE","kda":"2/0/3","gold":4800,"cs":42,"score_10":7.2,
     "gold_avg":5.0,"delta":2.2,"delta_recent":2.1,"delta_base":2.1,"rel_collapse":0.0,
     "risk_score":0,"risk_label":"LOW","flags":[],"deaths_gain_2":0,"kills_gain_2":2,
     "assists_gain_2":3,"cs_gain_2":20,"ward_gain_2":1,"unsupported_death_ratio":0.0,
     "objective_involvement":0.5,"game_num":1},
    {"minute":12,"role":"JUNGLE","kda":"2/2/4","gold":7200,"cs":71,"score_10":4.1,
     "gold_avg":5.1,"delta":-1.0,"delta_recent":-1.4,"delta_base":2.1,"rel_collapse":-3.5,
     "risk_score":3,"risk_label":"MED","flags":["REL_COLLAPSE","DEATH_SPIKE(+2)"],
     "deaths_gain_2":2,"kills_gain_2":0,"assists_gain_2":1,"cs_gain_2":14,"ward_gain_2":0,
     "unsupported_death_ratio":0.5,"objective_involvement":0.2,"game_num":1},
    {"minute":18,"role":"JUNGLE","kda":"2/5/5","gold":9100,"cs":91,"score_10":2.3,
     "gold_avg":5.2,"delta":-2.9,"delta_recent":-2.6,"delta_base":0.4,"rel_collapse":-3.0,
     "risk_score":5,"risk_label":"HIGH",
     "flags":["NEG_DEV","REL_COLLAPSE","LOW_AND_COLLAPSING","DEATH_SPIKE(+3)","NO_OBJECTIVE"],
     "deaths_gain_2":3,"kills_gain_2":0,"assists_gain_2":1,"cs_gain_2":10,"ward_gain_2":0,
     "unsupported_death_ratio":0.8,"objective_involvement":0.0,"game_num":1},
]

# ============================================================
def clean_flag(flag):
    return re.sub(r"\(.*?\)", "", flag).strip()

def parse_flags(flag_input):
    if isinstance(flag_input, list):
        return [str(x).strip() for x in flag_input if str(x).strip()]
    if not flag_input:
        return []
    if isinstance(flag_input, str):
        return [x.strip() for x in flag_input.split(",") if x.strip()]
    return []

def coerce_number(value, default=0.0):
    try:
        return float(value) if value not in (None, "") else default
    except:
        return default

def normalize_row(row):
    out = dict(row)
    out["minute"]      = int(coerce_number(out.get("minute"), 0))
    out["role"]        = str(out.get("role", "UNKNOWN")).upper()
    out["kda"]         = str(out.get("kda", "0/0/0"))
    out["gold"]        = int(coerce_number(out.get("gold"), 0))
    out["cs"]          = int(coerce_number(out.get("cs"), 0))
    out["score_10"]    = round(coerce_number(out.get("score_10"), 0.0), 2)
    out["gold_avg"]    = round(coerce_number(out.get("gold_avg"), 5.0), 2)
    out["delta"]       = round(coerce_number(out.get("delta"), 0.0), 2)
    out["delta_recent"] = round(coerce_number(out.get("delta_recent"), out["delta"]), 2)
    out["delta_base"]   = round(coerce_number(out.get("delta_base"), out["delta_recent"]), 2)
    out["rel_collapse"] = round(coerce_number(out.get("rel_collapse"), 0.0), 2)
    out["risk_score"]   = int(coerce_number(out.get("risk_score"), 0))
    out["risk_label"]   = str(out.get("risk_label", "LOW")).upper()
    out["flags"]        = parse_flags(out.get("flags", []))
    out["deaths_gain_2"]  = int(coerce_number(out.get("deaths_gain_2"), 0))
    out["kills_gain_2"]   = int(coerce_number(out.get("kills_gain_2"), 0))
    out["assists_gain_2"] = int(coerce_number(out.get("assists_gain_2"), 0))
    out["cs_gain_2"]      = int(coerce_number(out.get("cs_gain_2"), 0))
    out["ward_gain_2"]    = int(coerce_number(out.get("ward_gain_2"), 0))
    out["unsupported_death_ratio"] = round(coerce_number(out.get("unsupported_death_ratio"), 0.0), 2)
    out["objective_involvement"]   = round(coerce_number(out.get("objective_involvement"), 0.0), 2)
    return out

def classify_state(row):
    flags = {clean_flag(f) for f in row["flags"]}
    risk  = row["risk_score"]
    if risk >= 4 or "STRONG_COLLAPSE" in flags or "LOW_AND_COLLAPSING" in flags:
        return "critical_breakdown"
    if "NO_OBJECTIVE" in flags:
        return "objective_misalignment"
    if "UNSUPPORTED_DEATHS" in flags or "DEATH_SPIKE" in flags:
        return "unsafe_isolation_pattern"
    if "REL_COLLAPSE" in flags or "NEG_DEV" in flags:
        return "performance_drop"
    if "LOW_FARM" in flags or "NO_VISION" in flags:
        return "resource_stagnation"
    return "stable"

def build_evidence(row):
    evidence = []
    for f in row["flags"]:
        base = clean_flag(f)
        if base in FLAG_EXPLANATIONS:
            evidence.append(f"{f}: {FLAG_EXPLANATIONS[base]}")
    if row["delta"] <= -2:
        evidence.append(f"delta={row['delta']:.2f}: clearly below Gold checkpoint baseline")
    if row["rel_collapse"] <= -1.5:
        evidence.append(f"rel_collapse={row['rel_collapse']:.2f}: recent trend worse than earlier game")
    if row["unsupported_death_ratio"] >= 0.5:
        evidence.append(f"unsupported_death_ratio={row['unsupported_death_ratio']:.2f}: deaths without team support")
    if row["objective_involvement"] == 0 and row["minute"] >= 12:
        evidence.append("objective_involvement=0 after midgame threshold")
    if row["cs_gain_2"] <= 12 and row["role"] != "SUPPORT" and row["minute"] >= 9:
        evidence.append(f"cs_gain_2={row['cs_gain_2']}: low 6-minute farm growth")
    if not evidence:
        evidence.append("No major breakdown signals in current checkpoint.")
    return evidence

def llm_suggestion(row, state, evidence):
    try:
        flags_str    = ", ".join(row["flags"]) if row["flags"] else "none"
        evidence_str = "\n".join(f"- {e}" for e in evidence)
        system = (
            "You are a real-time League of Legends gameplay coach. "
            "You monitor a player at 3-minute checkpoints and send short, direct, "
            "actionable interventions like a human coach sitting next to the player. "
            "Rules: one sentence only, no bullet points, no generic advice, "
            "be specific to the role and minute, do not mention trolling or cheating."
        )
        user = (
            f"Player role: {row['role']}\n"
            f"Game minute: {row['minute']}\n"
            f"KDA: {row['kda']}   Gold: {row['gold']:,}   CS: {row['cs']}\n"
            f"Score_10: {row['score_10']:.2f} vs Gold avg: {row['gold_avg']:.2f} "
            f"(delta={row['delta']:+.2f})\n"
            f"Trend: delta_recent={row['delta_recent']:+.2f}  "
            f"rel_collapse={row['rel_collapse']:+.2f}\n"
            f"Risk: {row['risk_label']} (score={row['risk_score']})\n"
            f"Active flags: {flags_str}\n"
            f"Evidence:\n{evidence_str}\n"
            f"Current state: {state}\n\n"
            "Give one short, specific coaching intervention for this player right now."
        )
        response = get_client().chat.completions.create(
            model=GROQ_MODEL,
            messages=[{"role": "system", "content": system},
                      {"role": "user",   "content": user}],
            max_tokens=100,
            temperature=0.7,
        )
        return state, response.choices[0].message.content.strip()
    except Exception as e:
        return state, f"[Groq error: {e}] Avoid isolated plays and regroup before the next objective."

def draft_fallback(row, state):
    role, minute = row["role"], row["minute"]
    if state == "critical_breakdown":
        return "mitigation",      f"Minute {minute}: {role} is in a high-risk state. Avoid isolated plays, regroup before the next objective."
    if state == "objective_misalignment":
        return "objective_focus", f"Minute {minute}: reset vision and regroup 20-40s before the next objective."
    if state == "unsafe_isolation_pattern":
        return "safer_play",      f"Minute {minute}: avoid side-lane exposure, wait for a teammate before committing."
    if state == "performance_drop":
        return "re_engagement",   f"Minute {minute}: farm one safe wave, then return to team setup."
    if state == "resource_stagnation":
        return "resource_recovery", f"Minute {minute}: prioritize safe CS and vision before the next fight."
    return "stable",              f"Minute {minute}: state is stable. Keep syncing around the next objective."

def analyze_row(row, use_llm=True):
    row        = normalize_row(row)
    state      = classify_state(row)
    evidence   = build_evidence(row)
    confidence = min(0.98, max(0.35, 0.35 + 0.08 * row["risk_score"] + 0.06 * len(row["flags"])))
    stype, refined = llm_suggestion(row, state, evidence) if use_llm else draft_fallback(row, state)
    return {
        "input": row, "state": state, "confidence": round(confidence, 2),
        "evidence": evidence, "suggestion_type": stype,
        "refined_suggestion": refined,
        "source": "groq-llm" if use_llm else "rule-based",
    }

def _analyze_list(rows, use_llm):
    outputs, log_text = [], []
    for idx, row in enumerate(rows, start=1):
        try:
            result = analyze_row(row, use_llm=use_llm)
            outputs.append({
                "game":       row.get("game_num", "?"),
                "minute":     result["input"]["minute"],
                "role":       result["input"]["role"],
                "kda":        result["input"]["kda"],
                "score_10":   result["input"]["score_10"],
                "risk_label": result["input"]["risk_label"],
                "state":      result["state"],
                "source":     result["source"],
                "suggestion": result["refined_suggestion"],
            })
            log_text.append(
                f"[Game {row.get('game_num','?')}  min {result['input']['minute']}]  "
                f"{result['input']['risk_label']}  →  {result['refined_suggestion']}"
            )
        except Exception as e:
            log_text.append(f"Row {idx} failed: {e}")
    return outputs, "\n\n".join(log_text)

def analyze_uploaded_file(file, use_llm):
    if file is None:
        return [], "No file uploaded."
    try:
        path = file if isinstance(file, str) else file.name
        with open(path, "r") as f:
            rows = json.load(f)
        if isinstance(rows, dict):
            rows = [rows]
    except Exception as e:
        return [], f"Failed to read file: {e}"
    return _analyze_list(rows, use_llm)

# ============================================================
# GRADIO UI
# ============================================================
with gr.Blocks(title="PandaSkill Gameplay Agent") as demo:

    gr.Markdown("""
# 🐼 PandaSkill — Gameplay Process Manager Agent
Adapted from **McGee et al. (2026)** — AI facilitator using collective intelligence principles,
applied to League of Legends player performance monitoring.

Upload a `_rows.json` file from `test_player.py` to get per-checkpoint AI coaching interventions.
""")

    use_llm_toggle = gr.Checkbox(
        value=True,
        label="Use Groq LLM  (uncheck = rule-based fallback, no API needed)"
    )

    with gr.Tab("📂 Upload JSON  ← start here"):
        gr.Markdown("Upload the `*_rows.json` file exported by `test_player.py` after running locally.")
        upload_file = gr.File(label="Upload *_rows.json", file_types=[".json"])
        btn_upload  = gr.Button("▶  Run Agent on File", variant="primary")
        out_table   = gr.JSON(label="All Checkpoint Results")
        out_log     = gr.Textbox(label="Coaching Suggestions (one per checkpoint)",
                                 lines=25)
        btn_upload.click(analyze_uploaded_file,
                         inputs=[upload_file, use_llm_toggle],
                         outputs=[out_table, out_log])

    with gr.Tab("🎮 Live Demo"):
        gr.Markdown(
            "No file needed — runs on a built-in JUNGLE sample showing a "
            "performance collapse at minute 12–18."
        )
        btn_demo   = gr.Button("▶  Run Demo", variant="secondary")
        demo_table = gr.JSON(label="Demo Results")
        demo_log   = gr.Textbox(label="Demo Suggestions", lines=12)
        btn_demo.click(_analyze_list,
                       inputs=[gr.State(SAMPLE_ROWS), use_llm_toggle],
                       outputs=[demo_table, demo_log])

    with gr.Tab("🔢 Single Snapshot"):
        with gr.Row():
            m_min   = gr.Number(value=18,    label="minute")
            m_role  = gr.Dropdown(["TOP","JUNGLE","MID","BOT","SUPPORT"], value="JUNGLE", label="role")
            m_kda   = gr.Textbox(value="2/5/5", label="kda")
            m_gold  = gr.Number(value=9100,  label="gold")
            m_cs    = gr.Number(value=91,    label="cs")
        with gr.Row():
            m_s10   = gr.Number(value=2.3,   label="score_10")
            m_avg   = gr.Number(value=5.2,   label="gold_avg")
            m_dlt   = gr.Number(value=-2.9,  label="delta")
            m_dr    = gr.Number(value=-2.6,  label="delta_recent")
            m_db    = gr.Number(value=0.4,   label="delta_base")
            m_rc    = gr.Number(value=-3.0,  label="rel_collapse")
        with gr.Row():
            m_rs    = gr.Number(value=5,     label="risk_score")
            m_rl    = gr.Dropdown(["LOW","MED","HIGH"], value="HIGH", label="risk_label")
            m_fl    = gr.Textbox(value="NEG_DEV, REL_COLLAPSE, LOW_AND_COLLAPSING, DEATH_SPIKE(+3), NO_OBJECTIVE", label="flags")
        with gr.Row():
            m_d2    = gr.Number(value=3,   label="deaths_gain_2")
            m_k2    = gr.Number(value=0,   label="kills_gain_2")
            m_a2    = gr.Number(value=1,   label="assists_gain_2")
            m_cs2   = gr.Number(value=10,  label="cs_gain_2")
            m_w2    = gr.Number(value=0,   label="ward_gain_2")
            m_udr   = gr.Number(value=0.8, label="unsupported_death_ratio")
            m_obj   = gr.Number(value=0.0, label="objective_involvement")
        btn_m  = gr.Button("▶  Analyze", variant="primary")
        out_mr = gr.JSON(label="Agent Output")
        out_mm = gr.Textbox(label="Coaching Intervention")

        def run_manual(minute,role,kda,gold,cs,score_10,gold_avg,delta,
                       delta_recent,delta_base,rel_collapse,risk_score,risk_label,
                       flags,deaths_gain_2,kills_gain_2,assists_gain_2,cs_gain_2,
                       ward_gain_2,unsupported_death_ratio,objective_involvement,use_llm):
            row = dict(minute=minute,role=role,kda=kda,gold=gold,cs=cs,
                       score_10=score_10,gold_avg=gold_avg,delta=delta,
                       delta_recent=delta_recent,delta_base=delta_base,
                       rel_collapse=rel_collapse,risk_score=risk_score,
                       risk_label=risk_label,flags=flags,
                       deaths_gain_2=deaths_gain_2,kills_gain_2=kills_gain_2,
                       assists_gain_2=assists_gain_2,cs_gain_2=cs_gain_2,
                       ward_gain_2=ward_gain_2,
                       unsupported_death_ratio=unsupported_death_ratio,
                       objective_involvement=objective_involvement)
            result = analyze_row(row, use_llm=use_llm)
            return result, result["refined_suggestion"]

        btn_m.click(run_manual,
            inputs=[m_min,m_role,m_kda,m_gold,m_cs,m_s10,m_avg,m_dlt,
                    m_dr,m_db,m_rc,m_rs,m_rl,m_fl,
                    m_d2,m_k2,m_a2,m_cs2,m_w2,m_udr,m_obj,use_llm_toggle],
            outputs=[out_mr, out_mm])

    gr.Markdown("""
---
**Reference:** McGee, E. S., Cagan, J., & McComb, C. (2026).
Guiding Generalized Team Problem-Solving Through a Collective Intelligence-Based AI Facilitator.
*ASME Journal of Mechanical Design.*
""")

if __name__ == "__main__":
    demo.launch(
        server_name="0.0.0.0",
        server_port=7860,
        share=True,
        ssr_mode=False
    )