""" Run the full AgriFound offline pipeline: 1. Simulate soil / weather / tree-physiology / pest-disease per orchard 2. Simulate cold-storage conditions + batches + quality inspections 3. Run analytics: tipping-point risk, causal pathway attribution, prescriptive recommendations, ripeness classification, storage risk 4. Write everything to CSV (data/) AND a ready-to-import MySQL dump (database/data_generated.sql) Usage: python3 run_pipeline.py """ import os import pandas as pd from master_data import ORCHARDS, STORAGE_FACILITIES, PEST_DISEASE from simulate_data import (simulate_weather, simulate_soil, simulate_tree, simulate_pest_disease, simulate_storage_conditions, simulate_batches_and_quality) from analytics import (score_tipping_points, attribute_causal_pathways, recommend_actions, classify_ripeness, score_storage_risk) DATA_DIR = os.path.join(os.path.dirname(__file__), "..", "data") DB_DIR = os.path.join(os.path.dirname(__file__), "..", "database") os.makedirs(DATA_DIR, exist_ok=True) def sql_escape(v): if v is None or (isinstance(v, float) and pd.isna(v)): return "NULL" if isinstance(v, str): return "'" + v.replace("\\", "\\\\").replace("'", "\\'") + "'" if isinstance(v, pd.Timestamp): return "'" + v.strftime("%Y-%m-%d %H:%M:%S") + "'" return str(v) def df_to_inserts(df: pd.DataFrame, table: str, columns: list, batch_size=500) -> str: lines = [] for start in range(0, len(df), batch_size): chunk = df.iloc[start:start + batch_size] values = [] for _, row in chunk.iterrows(): values.append("(" + ",".join(sql_escape(row[c]) for c in columns) + ")") if values: lines.append(f"INSERT INTO {table} ({','.join(columns)}) VALUES\n" + ",\n".join(values) + ";") return "\n".join(lines) def main(): sql_chunks = ["USE agrifound;\n"] all_soil, all_weather, all_tree, all_pest = [], [], [], [] all_alerts, all_causal, all_rec = [], [], [] all_batches, all_inspections, all_ripeness = [], [], [] all_storage_cond, all_storage_risk = [], [] batch_key_to_id = {} next_batch_id = 1 print("== Simulating orchard sensor data + running tipping-point analytics ==") for orchard_id in ORCHARDS: weather = simulate_weather(orchard_id) soil = simulate_soil(orchard_id, weather) tree = simulate_tree(orchard_id, weather, soil[soil.depth_cm == 10].reset_index(drop=True)) pest = simulate_pest_disease(orchard_id, soil[soil.depth_cm == 10].reset_index(drop=True), weather) all_weather.append(weather) all_soil.append(soil) all_tree.append(tree) if len(pest): all_pest.append(pest) soil_top = soil[soil.depth_cm == 10].reset_index(drop=True) alerts = score_tipping_points(orchard_id, soil_top) alerts = alerts.sort_values("risk_score", ascending=False).head(25) # cap per orchard for a readable demo for _, arow in alerts.iterrows(): alert_local_id = len(all_alerts) all_alerts.append({ "orchard_id": orchard_id, "ts": arow["ts"], "risk_score": arow["risk_score"], "risk_category": arow["risk_category"], "predicted_horizon_days": int(arow["predicted_horizon_days"]), "description": f"Cascading-failure risk detected ({arow['risk_category']})", "_local_id": alert_local_id, }) fired = attribute_causal_pathways(arow) pathway_keys = [f[0] for f in fired] for key, name, effect, catalog_id in fired: all_causal.append({"_alert_local_id": alert_local_id, "pathway_name": name, "contributing_factors": f"vwc={arow['vwc_pct']}, soil_temp={arow['soil_temp_c']}, ec={arow['ec_ds_m']}, no3={arow['no3_ppm']}", "effect_size": effect}) for action, water_sav, pest_sav in recommend_actions(pathway_keys): all_rec.append({"_alert_local_id": alert_local_id, "action": action, "expected_water_savings_pct": water_sav, "expected_pesticide_savings_pct": pest_sav}) print(f" orchard {orchard_id}: {len(weather)} weather rows, {len(soil)} soil rows, {len(alerts)} alerts kept") print("== Simulating cold storage + quality/ripeness ==") for facility_id, fcfg in STORAGE_FACILITIES.items(): cond = simulate_storage_conditions(facility_id, fcfg) all_storage_cond.append(cond) risk = score_storage_risk(cond, fcfg["target_temp_c"], fcfg["target_rh"]) risk["facility_id"] = facility_id for orchard_id in fcfg["orchard_ids"]: batches, insp = simulate_batches_and_quality(orchard_id, facility_id) for _, brow in batches.iterrows(): bid = next_batch_id next_batch_id += 1 batch_key_to_id[brow["batch_key"]] = bid all_batches.append({ "_id": bid, "facility_id": facility_id, "orchard_id": orchard_id, "variety": ORCHARDS[orchard_id]["variety"], "intake_date": brow["intake_date"], "quantity_kg": brow["quantity_kg"], "intake_quality_grade": brow["intake_quality_grade"], }) if len(insp): cls = classify_ripeness(insp) insp = insp.reset_index(drop=True) for i, irow in insp.iterrows(): ins_local_id = len(all_inspections) all_inspections.append({ "_local_id": ins_local_id, "_batch_key": irow["batch_key"], "orchard_id": irow["orchard_id"], "ts": irow["ts"], "firmness_kgf": irow["firmness_kgf"], "brix_pct": irow["brix_pct"], "starch_index": irow["starch_index"], "background_color": irow["background_color"], "defect_pct": irow["defect_pct"], "inspector": irow["inspector"], }) crow = cls.iloc[i] all_ripeness.append({ "_inspection_local_id": ins_local_id, "predicted_stage": crow["predicted_stage"], "confidence": crow["confidence"], "recommended_action": crow["recommended_action"], }) # tie storage risk to the most-recent batch at that facility (simplification for the demo) facility_batches = [b for b in all_batches if b["facility_id"] == facility_id] if facility_batches: latest_batch_id = facility_batches[-1]["_id"] for _, rrow in risk.iterrows(): all_storage_risk.append({ "_batch_id": latest_batch_id, "ts": rrow["ts"], "spoilage_risk_score": rrow["spoilage_risk_score"], "risk_category": rrow["risk_category"], "notes": rrow["notes"], }) print(f" facility {facility_id}: {len(cond)} condition rows, {len(facility_batches)} batches") # ---------------- assemble DataFrames with resolved keys ---------------- weather_df = pd.concat(all_weather, ignore_index=True) soil_df = pd.concat(all_soil, ignore_index=True) tree_df = pd.concat(all_tree, ignore_index=True) pest_df = pd.concat(all_pest, ignore_index=True) if all_pest else pd.DataFrame() alerts_df = pd.DataFrame(all_alerts) alerts_df["alert_id"] = range(1, len(alerts_df) + 1) local_to_alert_id = dict(zip(alerts_df["_local_id"], alerts_df["alert_id"])) causal_df = pd.DataFrame(all_causal) if len(causal_df): causal_df["alert_id"] = causal_df["_alert_local_id"].map(local_to_alert_id) rec_df = pd.DataFrame(all_rec) if len(rec_df): rec_df["alert_id"] = rec_df["_alert_local_id"].map(local_to_alert_id) batches_df = pd.DataFrame(all_batches) inspections_df = pd.DataFrame(all_inspections) inspections_df["inspection_id"] = range(1, len(inspections_df) + 1) inspections_df["batch_id"] = inspections_df["_batch_key"].map(batch_key_to_id) local_to_inspection_id = dict(zip(inspections_df["_local_id"], inspections_df["inspection_id"])) ripeness_df = pd.DataFrame(all_ripeness) if len(ripeness_df): ripeness_df["inspection_id"] = ripeness_df["_inspection_local_id"].map(local_to_inspection_id) storage_cond_df = pd.concat(all_storage_cond, ignore_index=True) storage_risk_df = pd.DataFrame(all_storage_risk) if len(storage_risk_df): storage_risk_df = storage_risk_df.rename(columns={"_batch_id": "batch_id"}) # ---------------- write CSVs for inspection ---------------- weather_df.to_csv(f"{DATA_DIR}/weather_readings.csv", index=False) soil_df.to_csv(f"{DATA_DIR}/soil_readings.csv", index=False) tree_df.to_csv(f"{DATA_DIR}/tree_physiology_readings.csv", index=False) if len(pest_df): pest_df.to_csv(f"{DATA_DIR}/pest_disease_observations.csv", index=False) alerts_df.to_csv(f"{DATA_DIR}/tipping_point_alerts.csv", index=False) if len(causal_df): causal_df.to_csv(f"{DATA_DIR}/causal_pathway_log.csv", index=False) if len(rec_df): rec_df.to_csv(f"{DATA_DIR}/prescriptive_recommendations.csv", index=False) batches_df.to_csv(f"{DATA_DIR}/storage_batches.csv", index=False) inspections_df.to_csv(f"{DATA_DIR}/quality_inspections.csv", index=False) if len(ripeness_df): ripeness_df.to_csv(f"{DATA_DIR}/ripeness_classifications.csv", index=False) storage_cond_df.to_csv(f"{DATA_DIR}/storage_conditions_log.csv", index=False) if len(storage_risk_df): storage_risk_df.to_csv(f"{DATA_DIR}/storage_risk_log.csv", index=False) # ---------------- build MySQL insert dump ---------------- sql_chunks.append(df_to_inserts(weather_df, "weather_readings", ["node_id", "ts", "air_temp_c", "humidity_pct", "leaf_wetness_hrs", "solar_radiation_wm2", "rainfall_mm", "wind_speed_kmh"])) sql_chunks.append(df_to_inserts(soil_df, "soil_readings", ["node_id", "ts", "depth_cm", "vwc_pct", "soil_temp_c", "ec_ds_m", "no3_ppm", "p_ppm", "k_ppm", "microbial_biomass_c_mgkg"])) sql_chunks.append(df_to_inserts(tree_df, "tree_physiology_readings", ["node_id", "ts", "sap_flow_g_hr", "fruit_diameter_mm", "trunk_tilt_deg", "phenological_stage"])) if len(pest_df): sql_chunks.append(df_to_inserts(pest_df, "pest_disease_observations", ["orchard_id", "ts", "catalog_id", "severity_pct", "source"])) alerts_cols = ["orchard_id", "ts", "risk_score", "risk_category", "predicted_horizon_days", "description"] sql_chunks.append(df_to_inserts(alerts_df, "tipping_point_alerts", alerts_cols)) if len(causal_df): sql_chunks.append(df_to_inserts(causal_df, "causal_pathway_log", ["alert_id", "pathway_name", "contributing_factors", "effect_size"])) if len(rec_df): sql_chunks.append(df_to_inserts(rec_df, "prescriptive_recommendations", ["alert_id", "action", "expected_water_savings_pct", "expected_pesticide_savings_pct"])) sql_chunks.append(df_to_inserts(batches_df.rename(columns={"_id": "batch_id"}), "storage_batches", ["batch_id", "facility_id", "orchard_id", "variety", "intake_date", "quantity_kg", "intake_quality_grade"])) sql_chunks.append(df_to_inserts(inspections_df, "quality_inspections", ["inspection_id", "batch_id", "orchard_id", "ts", "firmness_kgf", "brix_pct", "starch_index", "background_color", "defect_pct", "inspector"])) if len(ripeness_df): sql_chunks.append(df_to_inserts(ripeness_df, "ripeness_classifications", ["inspection_id", "predicted_stage", "confidence", "recommended_action"])) sql_chunks.append(df_to_inserts(storage_cond_df, "storage_conditions_log", ["facility_id", "ts", "temp_c", "humidity_pct", "co2_pct", "o2_pct", "ethylene_ppm"])) if len(storage_risk_df): sql_chunks.append(df_to_inserts(storage_risk_df, "storage_risk_log", ["batch_id", "ts", "spoilage_risk_score", "risk_category", "notes"])) with open(f"{DB_DIR}/data_generated.sql", "w") as f: f.write("\n\n".join(c for c in sql_chunks if c)) print("\n== DONE ==") print(f"CSV files written to: {DATA_DIR}") print(f"MySQL dump written to: {DB_DIR}/data_generated.sql") print(f"Rows: weather={len(weather_df)} soil={len(soil_df)} tree={len(tree_df)} " f"alerts={len(alerts_df)} causal={len(causal_df)} batches={len(batches_df)} " f"inspections={len(inspections_df)} storage_cond={len(storage_cond_df)} storage_risk={len(storage_risk_df)}") if __name__ == "__main__": main()