agricultural_mcp/resources.py

import typing as t
import pandas as pd
import gradio as gr
from data_loader import AgriculturalDataLoader

# -------------------------------------------------------------------
# Hypothèse: AgriculturalDataLoader.load_all_files() concatène 10 CSV
# et renvoie un DataFrame avec au moins ces colonnes (si présentes):
# ["millesime","raisonsoci","siret","pacage","refca","numilot","numparcell",
#  "nomparc","surfparc","rang","kqte","teneurn","teneurp","teneurk",
#  "keq","volumebo","codeamm","codegnis","materiel","mainoeuvre", ...]
# -------------------------------------------------------------------

class AgriculturalResources:
    def __init__(self):
        self.data_loader = AgriculturalDataLoader()
        self.data_cache: t.Optional[pd.DataFrame] = None

    def load_data(self) -> pd.DataFrame:
        if self.data_cache is None:
            df = self.data_loader.load_all_files()

            # Normalisation minimale & robustesse
            df = df.copy()
            # Harmonise noms connus (au cas où)
            rename_map = {
                "raisonsoci": "raisonsociale",
                "numparcelle": "numparcell",
                "NomParc": "nomparc",
                "SurfParc": "surfparc",
            }
            for k, v in rename_map.items():
                if k in df.columns and v not in df.columns:
                    df[v] = df[k]

            # Types & trim
            for col in ["millesime", "siret", "pacage", "refca", "numilot", "numparcell", "rang",
                        "codeamm", "codegnis"]:
                if col in df.columns:
                    df[col] = df[col].astype(str).str.strip()

            for col in ["nomparc", "raisonsociale", "materiel", "mainoeuvre"]:
                if col in df.columns:
                    df[col] = df[col].astype(str).str.strip()

            for col in ["surfparc", "kqte", "teneurn", "teneurp", "teneurk", "keq", "volumebo"]:
                if col in df.columns:
                    # coerce = NaN si non convertible
                    df[col] = pd.to_numeric(df[col], errors="coerce")

            # IDs composites utiles
            if {"millesime", "numparcell"}.issubset(df.columns):
                df["parcelle_id"] = df["millesime"] + ":" + df["numparcell"]
            else:
                df["parcelle_id"] = None

            if {"millesime", "numparcell", "rang"}.issubset(df.columns):
                df["intervention_id"] = df["millesime"] + ":" + df["numparcell"] + ":" + df["rang"]
            else:
                df["intervention_id"] = None

            self.data_cache = df

        return self.data_cache

    # -------------------------
    # Utilitaires internes
    # -------------------------
    def _safe_first(self, df: pd.DataFrame) -> t.Optional[pd.Series]:
        if df is None or df.empty:
            return None
        return df.iloc[0]

    def _notnull(self, d: dict) -> dict:
        # Retire les champs None/NaN pour des payloads plus propres
        return {k: v for k, v in d.items() if pd.notna(v)}

    # -------------------------
    # LISTINGS / DISCOVERY
    # -------------------------

    @gr.mcp.resource("dataset://years")
    def list_years(self) -> t.List[str]:
        """Liste des millésimes disponibles dans l'ensemble des fichiers."""
        df = self.load_data()
        if "millesime" not in df.columns:
            return []
        years = sorted(df["millesime"].dropna().astype(str).unique())
        return years

    @gr.mcp.resource("exploitation://{siret}/parcelles")
    def list_parcelles_by_exploitation(self, siret: str, millesime: t.Optional[str] = None) -> t.List[dict]:
        """Liste les parcelles d'une exploitation (optionnellement filtrées par millésime)."""
        df = self.load_data()
        q = df[df["siret"] == siret] if "siret" in df.columns else df.iloc[0:0]
        if millesime:
            q = q[q["millesime"] == str(millesime)]
        cols = [c for c in ["parcelle_id","millesime","numparcell","nomparc","surfparc","refca","numilot"] if c in q.columns]
        out = q[cols].drop_duplicates().to_dict(orient="records")
        return out

    @gr.mcp.resource("parcelles://search")
    def search_parcelles(self, query: str = "", millesime: t.Optional[str] = None, limit: int = 50) -> t.List[dict]:
        """Recherche de parcelles par nom/numéro, filtrable par millésime."""
        df = self.load_data()
        q = df
        if millesime:
            q = q[q["millesime"] == str(millesime)]
        if query:
            mask = False
            if "numparcell" in q.columns:
                mask = q["numparcell"].str.contains(query, case=False, na=False)
            if "nomparc" in q.columns:
                mask = mask | q["nomparc"].str.contains(query, case=False, na=False)
            q = q[mask]
        cols = [c for c in ["parcelle_id","millesime","numparcell","nomparc","surfparc","refca","numilot","siret"] if c in q.columns]
        return q[cols].drop_duplicates().head(limit).to_dict(orient="records")

    # -------------------------
    # RESSOURCES CANONIQUES
    # -------------------------

    @gr.mcp.resource("exploitation://{siret}/{millesime}")
    def get_exploitation(self, siret: str, millesime: str) -> dict:
        """Infos d'une exploitation (pour un millésime donné)."""
        df = self.load_data()
        q = df[(df["siret"] == siret) & (df["millesime"] == str(millesime))] if {"siret","millesime"}.issubset(df.columns) else df.iloc[0:0]
        row = self._safe_first(q.sort_values(by=[c for c in ["millesime"] if c in q.columns], ascending=False))
        if row is None:
            return {}
        return self._notnull({
            "millesime": row.get("millesime"),
            "siret": row.get("siret"),
            "raison_sociale": row.get("raisonsociale"),
            "pacage": row.get("pacage"),
        })

    @gr.mcp.resource("parcelle://{millesime}/{numparcell}")
    def get_parcelle(self, millesime: str, numparcell: str) -> dict:
        """Infos d'une parcelle (identifiée par millésime + numparcell)."""
        df = self.load_data()
        q = df[(df["millesime"] == str(millesime)) & (df["numparcell"] == str(numparcell))]
        row = self._safe_first(q)
        if row is None:
            return {}
        return self._notnull({
            "parcelle_id": row.get("parcelle_id"),
            "millesime": row.get("millesime"),
            "numparcell": row.get("numparcell"),
            "nomparc": row.get("nomparc"),
            "surfparc": row.get("surfparc"),
            "siret": row.get("siret"),
            "refca": row.get("refca"),
            "numilot": row.get("numilot"),
        })

    @gr.mcp.resource("intervention://{millesime}/{numparcell}/{rang}")
    def get_intervention(self, millesime: str, numparcell: str, rang: str) -> dict:
        """Infos d'une intervention (clé composite millésime + numparcell + rang)."""
        df = self.load_data()
        q = df[(df["millesime"] == str(millesime)) & (df["numparcell"] == str(numparcell)) & (df["rang"] == str(rang))]
        row = self._safe_first(q)
        if row is None:
            return {}
        return self._notnull({
            "intervention_id": row.get("intervention_id"),
            "millesime": row.get("millesime"),
            "numparcell": row.get("numparcell"),
            "rang": row.get("rang"),
            "mainoeuvre": row.get("mainoeuvre"),
            "materiel": row.get("materiel"),
            "codeamm": row.get("codeamm"),
            "codegnis": row.get("codegnis"),
            "kqte": row.get("kqte"),
            "teneurn": row.get("teneurn"),
            "teneurp": row.get("teneurp"),
            "teneurk": row.get("teneurk"),
            "keq": row.get("keq"),
            "volumebo": row.get("volumebo"),
        })

    @gr.mcp.resource("intrant://{codeamm}")
    def get_intrant(self, codeamm: str, millesime: t.Optional[str] = None) -> dict:
        """Infos d’un intrant (filtrable par millésime)."""
        df = self.load_data()
        q = df[df["codeamm"] == str(codeamm)] if "codeamm" in df.columns else df.iloc[0:0]
        if millesime:
            q = q[q["millesime"] == str(millesime)]
        row = self._safe_first(q)
        if row is None:
            return {}
        return self._notnull({
            "codeamm": row.get("codeamm"),
            "codegnis": row.get("codegnis"),
            "millesime": row.get("millesime"),
            "kqte": row.get("kqte"),
            "teneurn": row.get("teneurn"),
            "teneurp": row.get("teneurp"),
            "teneurk": row.get("teneurk"),
            "keq": row.get("keq"),
            "volumebo": row.get("volumebo"),
        })

    @gr.mcp.resource("materiel://{millesime}/{numparcell}/{rang}")
    def get_materiel(self, millesime: str, numparcell: str, rang: str) -> dict:
        """Matériel utilisé pour une intervention donnée."""
        df = self.load_data()
        q = df[(df["millesime"] == str(millesime)) & (df["numparcell"] == str(numparcell)) & (df["rang"] == str(rang))]
        row = self._safe_first(q)
        if row is None:
            return {}
        return self._notnull({
            "millesime": row.get("millesime"),
            "numparcell": row.get("numparcell"),
            "rang": row.get("rang"),
            "materiel": row.get("materiel"),
        })

    @gr.mcp.resource("maindoeuvre://{millesime}/{numparcell}/{rang}")
    def get_main_oeuvre(self, millesime: str, numparcell: str, rang: str) -> dict:
        """Main d’œuvre associée à une intervention donnée."""
        df = self.load_data()
        q = df[(df["millesime"] == str(millesime)) & (df["numparcell"] == str(numparcell)) & (df["rang"] == str(rang))]
        row = self._safe_first(q)
        if row is None:
            return {}
        return self._notnull({
            "millesime": row.get("millesime"),
            "numparcell": row.get("numparcell"),
            "rang": row.get("rang"),
            "mainoeuvre": row.get("mainoeuvre"),
        })

# -------------------------------------------------------------------
# Instance pour utilisation
# -------------------------------------------------------------------
resources = AgriculturalResources()