File size: 8,412 Bytes
f38691a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 | """
TerraMind - Cache Builder
Precomputes and serialises all district-level lookup tables needed
for edge inference and the district intelligence novelty layer.
Outputs JSON files under artifacts/edge/:
- district_crop_frequency.json
- district_crop_stats.json
- district_irrigation_infra.json
- district_crop_irrigated_area.json
- state_district_map.json
- agro_climatic_zones.json (stub - configurable later)
"""
from __future__ import annotations
import json
from pathlib import Path
import numpy as np
import pandas as pd
from backend.core.config import EDGE_ARTIFACTS, CENTRAL_ARTIFACTS
from backend.core.logging_config import log
from backend.utils.data_loader import (
load_combined_yield_data,
load_icrisat_main,
load_icrisat_source,
load_icrisat_irrigation,
)
from backend.utils.feature_engineering import (
build_crop_frequency_prior,
compute_district_crop_stats,
)
from backend.utils.naming_maps import ICRISAT_CROP_PREFIX_MAP, normalize_crop
class NumpyEncoder(json.JSONEncoder):
"""Handle numpy types in JSON serialisation."""
def default(self, obj):
if isinstance(obj, (np.integer,)):
return int(obj)
if isinstance(obj, (np.floating,)):
return float(obj)
if isinstance(obj, np.ndarray):
return obj.tolist()
if pd.isna(obj):
return None
return super().default(obj)
def _save_json(data, path: Path):
with open(path, "w", encoding="utf-8") as f:
json.dump(data, f, cls=NumpyEncoder, indent=2)
log.info("Saved cache: %s (%d bytes)", path.name, path.stat().st_size)
def build_all_caches():
"""Build every cache file needed for edge / district intelligence."""
log.info("=== Building edge caches ===")
# 1. Yield data -> crop frequency + crop stats
yield_df = load_combined_yield_data()
freq = build_crop_frequency_prior(yield_df)
freq_dict: dict = {}
for _, row in freq.iterrows():
key = f"{row['State']}|{row['District']}"
freq_dict.setdefault(key, []).append({
"crop": row["Crop"],
"area_share": round(float(row["area_share"]), 4),
"rank": int(row["frequency_rank"]),
})
_save_json(freq_dict, EDGE_ARTIFACTS / "district_crop_frequency.json")
stats = compute_district_crop_stats(yield_df)
stats_dict: dict = {}
for _, row in stats.iterrows():
key = f"{row['State']}|{row['District']}|{row['Crop']}"
stats_dict[key] = {
"mean_yield": round(float(row["mean_yield"]), 3),
"std_yield": round(float(row["std_yield"]), 3),
"median_yield": round(float(row["median_yield"]), 3),
"n_years": int(row["n_years"]),
"first_year": int(row["first_year"]) if pd.notna(row["first_year"]) else None,
"last_year": int(row["last_year"]) if pd.notna(row["last_year"]) else None,
}
_save_json(stats_dict, EDGE_ARTIFACTS / "district_crop_stats.json")
# State-district map
sd_map: dict = {}
for state in yield_df["State"].dropna().unique():
districts = sorted(yield_df.loc[yield_df["State"] == state, "District"].dropna().unique().tolist())
if districts:
sd_map[state] = districts
_save_json(sd_map, EDGE_ARTIFACTS / "state_district_map.json")
_save_json(sd_map, CENTRAL_ARTIFACTS / "state_district_map.json")
# 2. ICRISAT source -> irrigation infrastructure
src = load_icrisat_source()
if src is not None:
infra_dict: dict = {}
for (state, dist), grp in src.groupby(["State", "District"], observed=True):
latest = grp.sort_values("Year").iloc[-1]
def safe_float(val):
if pd.isna(val):
return 0.0
try:
return float(val)
except (ValueError, TypeError):
return 0.0
infra_dict[f"{state}|{dist}"] = {
"canals": safe_float(latest.get("CANALS AREA (1000 ha)")),
"tanks": safe_float(latest.get("TANKS AREA (1000 ha)")),
"tube_wells": safe_float(latest.get("TUBE WELLS AREA (1000 ha)")),
"other_wells": safe_float(latest.get("OTHER WELLS AREA (1000 ha)")),
"other": safe_float(latest.get("OTHER SOURCES AREA (1000 ha)")),
"net_irrigated": safe_float(latest.get("NET AREA (1000 ha)")),
}
_save_json(infra_dict, EDGE_ARTIFACTS / "district_irrigation_infra.json")
# 3. ICRISAT irrigation -> crop irrigated area percentages
irr = load_icrisat_irrigation()
icr_main = load_icrisat_main()
if irr is not None and icr_main is not None:
irr_pct_dict: dict = {}
for (state, dist), irr_grp in irr.groupby(["State", "District"], observed=True):
main_grp = icr_main[(icr_main["State"] == state) & (icr_main["District"] == dist)]
if main_grp.empty:
continue
latest_irr = irr_grp.sort_values("Year").iloc[-1]
latest_main = main_grp.sort_values("Year").iloc[-1]
for prefix, crop_name in ICRISAT_CROP_PREFIX_MAP.items():
irr_col = f"{prefix} IRRIGATED AREA (1000 ha)"
area_col = f"{prefix} AREA (1000 ha)"
if irr_col in latest_irr.index and area_col in latest_main.index:
irr_val = latest_irr.get(irr_col)
area_val = latest_main.get(area_col)
if pd.notna(irr_val) and pd.notna(area_val) and area_val > 0:
pct = round(float(irr_val / area_val * 100), 1)
key = f"{state}|{dist}|{crop_name}"
irr_pct_dict[key] = min(pct, 100.0)
_save_json(irr_pct_dict, EDGE_ARTIFACTS / "district_crop_irrigated_area.json")
# 4. Best season per (district, crop)
season_stats = yield_df.groupby(
["State", "District", "Crop", "Season"], observed=True
).agg(
mean_yield=("Yield", "mean"),
count=("Yield", "count"),
).reset_index()
best_season: dict = {}
for (state, dist, crop), grp in season_stats.groupby(["State", "District", "Crop"], observed=True):
if grp.empty:
continue
best = grp.loc[grp["mean_yield"].idxmax()]
best_season[f"{state}|{dist}|{crop}"] = best["Season"]
_save_json(best_season, EDGE_ARTIFACTS / "district_crop_best_season.json")
# 5. 10-year trajectory per (district, crop)
trajectory: dict = {}
recent = yield_df[yield_df["Year_Num"] >= yield_df["Year_Num"].max() - 10].copy()
for (state, dist, crop), grp in recent.groupby(["State", "District", "Crop"], observed=True):
yearly = grp.groupby("Year_Num", observed=True)["Yield"].mean().sort_index()
trajectory[f"{state}|{dist}|{crop}"] = {
"years": yearly.index.astype(int).tolist(),
"yields": [round(float(v), 3) for v in yearly.values],
}
_save_json(trajectory, EDGE_ARTIFACTS / "district_crop_trajectory.json")
# 6. Stub agro-climatic zones - configurable mapping file
agro_zones = {
"western himalayan": ["jammu and kashmir", "himachal pradesh", "uttarakhand"],
"eastern himalayan": ["assam", "sikkim", "meghalaya", "arunachal pradesh", "nagaland", "manipur", "mizoram", "tripura"],
"upper gangetic plains": ["uttar pradesh"],
"lower gangetic plains": ["west bengal", "bihar"],
"middle gangetic plains": ["bihar", "jharkhand"],
"trans gangetic plains": ["punjab", "haryana", "delhi", "chandigarh"],
"eastern plateau and hills": ["chhattisgarh", "jharkhand", "odisha"],
"central plateau and hills": ["madhya pradesh", "rajasthan", "uttar pradesh"],
"western plateau and hills": ["maharashtra"],
"southern plateau and hills": ["karnataka", "andhra pradesh", "telangana"],
"east coast": ["tamil nadu", "andhra pradesh", "odisha"],
"west coast": ["goa", "kerala", "karnataka"],
"gujarat plains": ["gujarat"],
"western dry region": ["rajasthan"],
"island region": ["andaman and nicobar", "lakshadweep"],
}
_save_json(agro_zones, EDGE_ARTIFACTS / "agro_climatic_zones.json")
log.info("=== All edge caches built successfully ===")
if __name__ == "__main__":
build_all_caches()
|