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()