""" AgriFound simulation engine. Generates physically-informed, agronomically-grounded synthetic sensor/quality/storage data standing in for the real 50+ node IoT mesh described in the proposal (which doesn't exist yet). Where possible, parameters are anchored to real open sources: - Growing-season temperature & rainfall normals per region (NHB India apple crop profile; see docs/DATA_SOURCES.md) - Soil tipping-point thresholds (VWC, EC, NO3, temperature) as stated in the proposal's Module II (Soil Agent) section - Postharvest physiology benchmarks for firmness/Brix/starch index (standard horticultural literature) This is NOT real sensor data. It is a clearly-labeled simulator that lets the rest of the stack (analytics + PHP dashboard) be built and demonstrated end-to-end. Swap in `fetch_open_weather.py` + real IoT feeds for production use. """ import numpy as np import pandas as pd from datetime import datetime, timedelta from config import SEASON_START, SEASON_END, STORAGE_END, RANDOM_SEED, THRESHOLDS from master_data import REGIONS, ORCHARDS, node_ids_for_orchard, STORAGE_FACILITIES rng = np.random.default_rng(RANDOM_SEED) def date_range(start, end, freq="D"): return pd.date_range(start=start, end=end, freq=freq) # --------------------------------------------------------------------- # WEATHER # --------------------------------------------------------------------- def simulate_weather(orchard_id: int) -> pd.DataFrame: region = REGIONS[ORCHARDS[orchard_id]["region_id"]] dates = date_range(SEASON_START, SEASON_END) n = len(dates) doy = dates.dayofyear.values.astype(float) # seasonal sinusoid calibrated so the Apr-Oct mean matches the real # regional growing-season average temperature season_curve = np.sin((doy - 100) / 214 * np.pi) # peaks mid-season air_temp = region["temp_c"] + 6 * season_curve + rng.normal(0, 1.8, n) humidity = np.clip(55 + 20 * np.sin((doy - 60) / 200 * np.pi) + rng.normal(0, 6, n), 25, 98) # Rainfall: distribute the real annual total stochastically across the season, # with extra convective-storm intensity in Apr-May (matches recent Kashmir # hailstorm/cloudburst reporting -- see docs/DATA_SOURCES.md) daily_rain_base = region["rain_mm"] / 365.0 rain_prob = 0.28 is_spring = (dates.month.isin([4, 5])).astype(float) storm_boost = 1 + 1.6 * is_spring # spring storms hit harder rain_event = rng.random(n) < rain_prob rainfall = np.where( rain_event, rng.gamma(shape=1.4, scale=daily_rain_base / rain_prob * 1.1) * storm_boost, 0.0, ) # occasional extreme cloudburst/hailstorm event in spring extreme_idx = np.where((dates.month.isin([4, 5])) & (rng.random(n) < 0.015))[0] rainfall[extreme_idx] += rng.uniform(40, 85, len(extreme_idx)) leaf_wetness = np.clip(rainfall.copy(), 0, None) leaf_wetness = np.where(leaf_wetness > 0, rng.uniform(2, 10, n), rng.uniform(0, 1.5, n)) solar_rad = np.clip(220 + 140 * season_curve - 3.2 * rainfall + rng.normal(0, 25, n), 60, 420) wind = np.clip(rng.normal(8, 4, n) + 6 * (rainfall > 20), 0, 45) node_id = node_ids_for_orchard(orchard_id)["weather"] return pd.DataFrame({ "node_id": node_id, "ts": dates, "air_temp_c": air_temp.round(1), "humidity_pct": humidity.round(1), "leaf_wetness_hrs": leaf_wetness.round(1), "solar_radiation_wm2": solar_rad.round(1), "rainfall_mm": rainfall.round(1), "wind_speed_kmh": wind.round(1), }) # --------------------------------------------------------------------- # SOIL (3 depths, with drying/rewetting hysteresis = "soil memory") # --------------------------------------------------------------------- def simulate_soil(orchard_id: int, weather: pd.DataFrame) -> pd.DataFrame: soil_type = REGIONS[ORCHARDS[orchard_id]["region_id"]]["soil"] field_capacity = {"loamy": 32, "silty_loam": 34, "alluvial_loam": 33, "sandy_loam": 26}.get(soil_type, 30) wilting_point = field_capacity * 0.45 compaction_bias = ORCHARDS[orchard_id]["compaction_bias"] depths = [10, 30, 60] node_ids = node_ids_for_orchard(orchard_id) frames = [] for depth in depths: n = len(weather) vwc = np.zeros(n) vwc[0] = field_capacity * 0.75 # depth damps rainfall response and evapotranspiration draw-down infil_factor = {10: 0.9, 30: 0.55, 60: 0.3}[depth] et_factor = {10: 1.0, 30: 0.6, 60: 0.3}[depth] # compaction reduces infiltration -> shallower effective wetting, more drought sensitivity infil_factor *= (1 - 0.5 * compaction_bias) for i in range(1, n): infiltration = weather["rainfall_mm"].iloc[i] * infil_factor * 0.35 et_loss = (0.9 + weather["air_temp_c"].iloc[i] / 40.0) * et_factor drainage = max(0, (vwc[i - 1] - field_capacity)) * 0.4 # hysteresis: drying is slower to release than wetting is to absorb (soil "memory") delta = infiltration - et_loss - drainage if delta < 0: delta *= 0.85 # drying resists / lags vwc[i] = np.clip(vwc[i - 1] + delta + rng.normal(0, 0.4), wilting_point * 0.6, field_capacity * 1.05) soil_temp = (weather["air_temp_c"].values * (0.55 if depth == 10 else 0.35 if depth == 30 else 0.2) + {10: 8, 30: 12, 60: 14}[depth] + rng.normal(0, 0.6, n)) ec = np.clip(0.9 + 0.6 * compaction_bias + rng.normal(0, 0.25, n) + 0.02 * np.cumsum(np.where(weather["rainfall_mm"].values < 1, 0.02, -0.05)).clip(-5, 15), 0.2, 4.5) no3 = np.clip(70 - 25 * compaction_bias - 0.5 * np.maximum(soil_temp - 28, 0) * 3 + rng.normal(0, 5, n), 5, 140) p_ppm = np.clip(18 + rng.normal(0, 3, n), 4, 40) k_ppm = np.clip(140 + rng.normal(0, 15, n), 40, 260) microbial_c = np.clip(220 - 6 * np.maximum(soil_temp - 28, 0) + rng.normal(0, 12, n), 20, 320) frames.append(pd.DataFrame({ "node_id": node_ids[f"soil_{depth}"], "ts": weather["ts"].values, "depth_cm": depth, "vwc_pct": vwc.round(2), "soil_temp_c": soil_temp.round(1), "ec_ds_m": ec.round(2), "no3_ppm": no3.round(1), "p_ppm": p_ppm.round(1), "k_ppm": k_ppm.round(1), "microbial_biomass_c_mgkg": microbial_c.round(1), })) return pd.concat(frames, ignore_index=True) # --------------------------------------------------------------------- # TREE PHYSIOLOGY # --------------------------------------------------------------------- def simulate_tree(orchard_id: int, weather: pd.DataFrame, soil_top: pd.DataFrame) -> pd.DataFrame: n = len(weather) dates = weather["ts"] doy = dates.dt.dayofyear.values.astype(float) # phenology: bloom (Apr) -> fruit set (May) -> growth (Jun-Aug) -> maturity/harvest (Sep-Oct) stage = np.select( [dates.dt.month == 4, dates.dt.month == 5, dates.dt.month.isin([6, 7, 8]), dates.dt.month.isin([9, 10])], ["bloom", "fruit_set", "growth", "maturity"], default="dormant", ) vwc = soil_top["vwc_pct"].values stress = np.clip((THRESHOLDS["vwc_critical_pct"] + 4 - vwc) / 10, 0, 1) # 0=no stress,1=high stress sap_flow = np.clip(45 + 25 * np.sin((doy - 110) / 190 * np.pi) * (1 - 0.6 * stress) + rng.normal(0, 4, n), 2, 110) # fruit diameter: logistic growth from fruit-set to harvest, slowed by water stress fruit_diam = np.zeros(n) growth_rate_base = 0.9 for i in range(1, n): if stage[i] in ("fruit_set", "growth", "maturity"): growth = growth_rate_base * (1 - fruit_diam[i - 1] / 78.0) * (1 - 0.5 * stress[i]) fruit_diam[i] = max(0, fruit_diam[i - 1] + growth + rng.normal(0, 0.15)) else: fruit_diam[i] = fruit_diam[i - 1] trunk_tilt = np.abs(rng.normal(0.4, 0.15, n)) storm_idx = np.where(weather["rainfall_mm"].values > 35)[0] trunk_tilt[storm_idx] += rng.uniform(0.8, 2.5, len(storm_idx)) node_id = node_ids_for_orchard(orchard_id)["tree"] return pd.DataFrame({ "node_id": node_id, "ts": dates.values, "sap_flow_g_hr": sap_flow.round(1), "fruit_diameter_mm": fruit_diam.round(2), "trunk_tilt_deg": trunk_tilt.round(2), "phenological_stage": stage, }) # --------------------------------------------------------------------- # PEST / DISEASE OBSERVATIONS (linked to the named causal pathways) # --------------------------------------------------------------------- def simulate_pest_disease(orchard_id: int, soil_top: pd.DataFrame, weather: pd.DataFrame) -> pd.DataFrame: dates = weather["ts"] rows = [] catalog_by_name = { "Apple Scab": 1, "San Jose Scale": 2, "Codling Moth": 3, "Apple Blotch Leaf Miner": 4, } wet_spring = (dates.dt.month.isin([4, 5])) & (weather["leaf_wetness_hrs"] > 5) for i, is_wet in enumerate(wet_spring): if is_wet and rng.random() < 0.4: rows.append((dates.iloc[i], catalog_by_name["Apple Scab"], float(np.clip(rng.normal(18, 8), 2, 60)))) pupation_ok = (soil_top["soil_temp_c"] > 18) & (soil_top["soil_temp_c"] < 28) & (soil_top["vwc_pct"] > 15) for i, ok in enumerate(pupation_ok.values): if ok and dates.iloc[i].month in (6, 7, 8) and rng.random() < 0.06: rows.append((dates.iloc[i], catalog_by_name["Codling Moth"], float(np.clip(rng.normal(12, 5), 1, 45)))) warm_dry = (weather["air_temp_c"] > 26) & (weather["humidity_pct"] < 45) for i, ok in enumerate(warm_dry.values): if ok and rng.random() < 0.03: rows.append((dates.iloc[i], catalog_by_name["Apple Blotch Leaf Miner"], float(np.clip(rng.normal(15, 6), 1, 50)))) if not rows: return pd.DataFrame(columns=["orchard_id", "ts", "catalog_id", "severity_pct", "source"]) df = pd.DataFrame(rows, columns=["ts", "catalog_id", "severity_pct"]) df["orchard_id"] = orchard_id df["source"] = "model_inferred" return df[["orchard_id", "ts", "catalog_id", "severity_pct", "source"]] # --------------------------------------------------------------------- # COLD STORAGE CONDITIONS # --------------------------------------------------------------------- def simulate_storage_conditions(facility_id: int, facility_cfg: dict) -> pd.DataFrame: dates = date_range(SEASON_START, STORAGE_END) n = len(dates) target_t = facility_cfg["target_temp_c"] target_rh = facility_cfg["target_rh"] temp = target_t + rng.normal(0, 0.35, n) humidity = target_rh + rng.normal(0, 1.2, n) # occasional compressor-fault / door-open excursions fault_idx = np.where(rng.random(n) < 0.02)[0] for idx in fault_idx: span = min(n - idx, rng.integers(2, 9)) temp[idx:idx + span] += rng.uniform(2.5, 7.0) humidity[idx:idx + span] -= rng.uniform(5, 15) co2 = np.clip(1.5 + rng.normal(0, 0.3, n) + 0.3 * (temp - target_t).clip(0, None), 0.5, 6) o2 = np.clip(2.5 - 0.15 * (temp - target_t).clip(0, None) + rng.normal(0, 0.2, n), 0.8, 5) ethylene = np.clip(0.3 + 0.15 * np.cumsum((temp - target_t).clip(0, None)) / 50 + rng.normal(0, 0.05, n), 0.05, 5) return pd.DataFrame({ "facility_id": facility_id, "ts": dates, "temp_c": temp.round(2), "humidity_pct": humidity.round(1), "co2_pct": co2.round(2), "o2_pct": o2.round(2), "ethylene_ppm": ethylene.round(3), }) # --------------------------------------------------------------------- # STORAGE BATCHES + QUALITY / RIPENESS INSPECTIONS # --------------------------------------------------------------------- def simulate_batches_and_quality(orchard_id: int, facility_id: int): harvest_dates = pd.date_range("2024-09-10", "2024-10-20", freq="7D") batches = [] inspections = [] batch_id_counter_start = orchard_id * 100 # placeholder key used only within this run for bi, hdate in enumerate(harvest_dates): qty = float(rng.uniform(8000, 22000)) grade = rng.choice(["A", "A", "B", "C"], p=[0.45, 0.3, 0.18, 0.07]) batch_key = batch_id_counter_start + bi batches.append(dict(batch_key=batch_key, orchard_id=orchard_id, facility_id=facility_id, intake_date=hdate.date(), quantity_kg=round(qty, 1), intake_quality_grade=grade)) # quality trajectory over ~140 days in storage: firmness/starch/brix/color evolve storage_days = pd.date_range(hdate, hdate + timedelta(days=140), freq="14D") firmness0 = rng.uniform(7.6, 9.2) starch0 = rng.uniform(2.0, 4.0) brix0 = rng.uniform(11.5, 14.0) colors = ["green", "green_yellow", "yellow", "yellow", "red_blush"] for di, d in enumerate(storage_days): t = di / max(1, len(storage_days) - 1) firmness = firmness0 - t * rng.uniform(2.0, 3.4) starch = starch0 + t * rng.uniform(4.5, 6.5) brix = brix0 + t * rng.uniform(-0.5, 1.2) defect = np.clip(t * rng.uniform(2, 9) + rng.normal(0, 1), 0, 35) color = colors[min(len(colors) - 1, int(t * (len(colors) - 1) + rng.uniform(-0.3, 0.3)))] inspections.append(dict( batch_key=batch_key, orchard_id=orchard_id, ts=d, firmness_kgf=round(max(2.5, firmness), 2), brix_pct=round(max(9.0, brix), 2), starch_index=round(min(10, max(1, starch)), 1), background_color=color, defect_pct=round(defect, 1), inspector="sensor_nir", )) return pd.DataFrame(batches), pd.DataFrame(inspections) if __name__ == "__main__": w = simulate_weather(1) s = simulate_soil(1, w) print(w.head(), "\n", s.head())