""" Individual model inference functions. Critical production principle: if Model 3 (yield) throws an exception, the user should still get results from Models 1, 2, 4, 5, 6 — degraded, clearly labeled, but not a total failure. A farmer asking about climate risk shouldn't get a generic 500 error because the market price model had a transient issue. Every function here catches its own exceptions and returns a "degraded" output with status flags, rather than raising. The orchestrator (aggregator.py) decides what to do with partial failures — these functions just guarantee they never crash the caller. """ from typing import Any import numpy as np import pandas as pd, os from app.core.logging_config import get_logger from app.models_loader.registry import ModelRegistry logger = get_logger(__name__) def _load_crop_candidates(): import json, os path = os.path.normpath(os.path.join(os.path.dirname(__file__), '..', '..', 'data', 'lookup', 'crops.json')) # These are aggregate categories or non-standalone crops # that should not appear as recommendations EXCLUDE = { 'total foodgrain', 'pulses total', 'oilseeds total', 'other cereals', 'other cereals & millets', 'other kharif pulses', 'other rabi pulses', 'other summer pulses', 'other oilseeds', 'small millets', 'korra', 'samai', 'varagu', 'peas & beans (pulses)', 'mesta', 'sannhamp', 'dry ginger', 'guar seed', 'other cereals & millets', 'pome fruit', 'citrus fruit', } try: with open(path) as f: all_crops = json.load(f).keys() return sorted([c for c in all_crops if c not in EXCLUDE]) except: return [] CROP_CANDIDATES_FOR_RECOMMENDATION = _load_crop_candidates() def run_climate_model(registry: ModelRegistry, district: str, month: int = 6) -> dict[str, Any]: try: sarimax_models: dict = registry.get("sarimax_climate") model = sarimax_models.get(district) if model is None: logger.warning(f"No SARIMAX climate model trained for district='{district}'") return { "forecast_rainfall_mm": None, "forecast_temp_c": None, "anomaly_score": 0.0, "anomaly_label": "unknown", "degraded": True, } hist_path = "data/processed/climate_clean.csv" hist_df = pd.read_csv(hist_path) if os.path.exists(hist_path) else pd.DataFrame() dm = hist_df[(hist_df["district"]==district) & (hist_df["month"]==month)] if not hist_df.empty else pd.DataFrame() if not dm.empty: rainfall = round(float(dm["rainfall_mm"].mean()), 1) temp_c = round(float(dm["temp_c"].mean()), 1) if "temp_c" in dm.columns else None humidity = round(float(dm["humidity"].mean()), 1) if "humidity" in dm.columns else None da = hist_df[hist_df["district"]==district] ann_mean = float(da["rainfall_mm"].mean()) if not da.empty else rainfall ann_std = float(da["rainfall_mm"].std()) if not da.empty else 1.0 z = (rainfall - ann_mean) / (ann_std + 1e-9) alabel = "high" if abs(z) > 1.5 else "moderate" if abs(z) > 0.75 else "low" return { "forecast_rainfall_mm": rainfall, "forecast_temp_c": temp_c, "humidity": humidity, "anomaly_score": round(z, 3), "anomaly_label": alabel, "degraded": False, } forecast = float(model.forecast(steps=1).iloc[0]) recent_window = model.data.endog[-12:] hist_mean = float(np.mean(recent_window)) if len(recent_window) else forecast anomaly = (forecast - hist_mean) / (hist_mean + 1e-9) label = "high" if anomaly > 0.3 else "low" if anomaly < -0.3 else "normal" return { "forecast_rainfall_mm": round(forecast, 1), "forecast_temp_c": None, "anomaly_score": round(float(anomaly), 3), "anomaly_label": label, "degraded": False, } except Exception as exc: logger.error(f"Climate model inference failed for district='{district}': {exc}", exc_info=True) return { "forecast_rainfall_mm": None, "forecast_temp_c": None, "anomaly_score": 0.0, "anomaly_label": "unknown", "degraded": True, "error": "climate_model_failed", } def _load_trained_crops() -> set: import json, os path = os.path.normpath(os.path.join(os.path.dirname(__file__), '..', '..', 'data', 'lookup', 'trained_crops.json')) try: with open(path) as f: return set(json.load(f)) except: return set() _TRAINED_CROPS = _load_trained_crops() def run_feasibility_model( registry: ModelRegistry, feature_vector, crop: str = '', district: str = '' ) -> dict[str, Any]: try: # If model has never seen this crop, return not_suitable honestly # Check TNAU district-crop suitability lookup FIRST # This is ground truth from real agricultural data import json as _json, os as _os _suit_path = _os.path.normpath(_os.path.join(_os.path.dirname(__file__), '..', '..', 'data', 'lookup', 'district_crop_suitability.json')) try: with open(_suit_path) as _f: _suit_data = _json.load(_f) _district_data = _suit_data.get(district.lower() if district else '', {}) _crop_data = _district_data.get(crop.lower() if crop else '', None) if _crop_data is not None: _suit_label = _crop_data['suitability'] return { 'label': _suit_label, 'confidence': 0.95, 'top_factors': ['tnau_ground_truth'], 'degraded': False, 'source': 'tnau_district_data', } except Exception as _e: logger.warning(f'Could not load district_crop_suitability.json: {_e}') # Fall back to ML model if not in TNAU lookup if crop and _TRAINED_CROPS and crop.lower() not in _TRAINED_CROPS: return { 'label': 'not_suitable', 'confidence': 0.90, 'top_factors': ['no_training_data'], 'degraded': False, } bundle = registry.get("xgb_feasibility") model, label_encoder = bundle["model"], bundle["label_encoder"] proba = model.predict_proba(feature_vector.reshape(1, -1))[0] idx = int(np.argmax(proba)) label = label_encoder.inverse_transform([idx])[0] feature_cols = bundle.get("feature_cols", []) top_factors: list[str] = [] if hasattr(model, "feature_importances_") and feature_cols: importances = model.feature_importances_ ranked = sorted(zip(feature_cols, importances), key=lambda x: x[1], reverse=True) top_factors = [name for name, _ in ranked[:3]] return { "label": str(label), "confidence": min(float(proba.max()), 0.99), "top_factors": top_factors, "degraded": False, } except Exception as exc: logger.error(f"Feasibility model inference failed: {exc}", exc_info=True) return { "label": "unknown", "confidence": 0.0, "top_factors": [], "degraded": True, "error": "feasibility_model_failed", } def run_yield_model(registry: ModelRegistry, feature_vector: np.ndarray) -> dict[str, Any]: try: bundle = registry.get("xgb_yield") model = bundle["model"] pred = float(model.predict(feature_vector.reshape(1, -1))[0]) if pred < 0: # A negative yield is physically meaningless — clamp and flag # rather than return nonsense to the LLM/client. logger.warning(f"Yield model produced negative prediction {pred}, clamping to 0") pred = 0.0 return { "expected_yield_ton_ha": round(pred, 2), "uncertainty_ton_ha": round(pred * 0.15, 2), # placeholder heuristic until # ensemble-based uncertainty # is wired in (see training notes) "unit": "tonnes per hectare", "degraded": False, } except Exception as exc: logger.error(f"Yield model inference failed: {exc}", exc_info=True) return { "expected_yield_ton_ha": 0.0, "uncertainty_ton_ha": 0.0, "unit": "tonnes per hectare", "degraded": True, "error": "yield_model_failed", } def run_trend_model(registry: ModelRegistry, feature_vector: np.ndarray) -> dict[str, Any]: try: bundle = registry.get("dt_trend") model = bundle["model"] # dt_trend is trained on the same full feature vector produced by # FeatureBuilder (see ml_pipeline/training/train_dt_trend.py) — no # slicing needed. If a future model version trains on a reduced # feature subset, slice feature_vector here accordingly and update # this comment to document the new contract. pred = model.predict(feature_vector.reshape(1, -1))[0] proba = model.predict_proba(feature_vector.reshape(1, -1))[0] return { "trend": str(pred), "slope": None, # populated when slope is passed through as a raw feature "confidence": round(float(np.max(proba)), 3), "degraded": False, } except Exception as exc: logger.error(f"Trend model inference failed: {exc}", exc_info=True) return { "trend": "unknown", "slope": None, "confidence": 0.0, "degraded": True, "error": "trend_model_failed", } def run_recommendation_model( registry: ModelRegistry, district: str, month: int, build_feature_fn, exclude_crop: str, ) -> dict[str, Any]: try: bundle = registry.get("rf_recommendation") model = bundle["model"] classes = bundle.get("classes", list(model.classes_)) # Use "high" if available, fall back to "medium" — the score # still ranks crops by relative suitability, just on a # different scale. This handles the common case where market # data is insufficient to generate "high" suitability labels. if "high" in classes: score_idx = classes.index("high") elif "medium" in classes: score_idx = classes.index("medium") else: logger.error("rf_recommendation model has no usable suitability class") return {"alternatives": [], "degraded": True, "error": "recommendation_model_misconfigured"} results = [] # Filter candidates by district temperature compatibility import json as _json, os as _os, pandas as _pd _crops_path = _os.path.normpath(_os.path.join(_os.path.dirname(__file__), '..', '..', 'data', 'lookup', 'crops.json')) _climate_path = _os.path.normpath(_os.path.join(_os.path.dirname(__file__), '..', '..', 'data', 'processed', 'climate_clean.csv')) try: with open(_crops_path) as _f: _crops_data = _json.load(_f) _climate_df = _pd.read_csv(_climate_path) _district_temp = float(_climate_df[_climate_df['district']==district]['temp_c'].mean()) except: _crops_data = {} _district_temp = 28.0 for candidate_crop in CROP_CANDIDATES_FOR_RECOMMENDATION: if candidate_crop == exclude_crop: continue _props = _crops_data.get(candidate_crop, {}) _min_t = _props.get('min_temp_c', 0) _max_t = _props.get('max_temp_c', 50) if not (_min_t <= _district_temp <= _max_t): continue try: vec = build_feature_fn(district, candidate_crop, month) proba = model.predict_proba(vec.reshape(1, -1))[0] if "high" in classes: score = float(proba[score_idx]) else: # Use medium - low gap as a relative ranking score medium_idx = classes.index("medium") if "medium" in classes else score_idx low_idx = classes.index("low") if "low" in classes else -1 medium_score = float(proba[medium_idx]) low_score = float(proba[low_idx]) if low_idx >= 0 else 0 score = round(medium_score - low_score, 3) results.append({"crop": candidate_crop, "score": round(score, 3), "reason": None}) except Exception as inner_exc: logger.warning(f"Recommendation scoring failed for crop='{candidate_crop}': {inner_exc}") continue ranked = sorted(results, key=lambda r: r["score"], reverse=True)[:5] degraded = "high" not in classes # honest about data limitation return {"alternatives": ranked, "degraded": degraded} except Exception as exc: logger.error(f"Recommendation model inference failed: {exc}", exc_info=True) return {"alternatives": [], "degraded": True, "error": "recommendation_model_failed"} def run_market_model( registry: ModelRegistry, district: str, crop: str ) -> dict[str, Any]: try: from app.services.market_price_service import get_price price_data = get_price(crop, district) if price_data is None: return { 'price_forecast_per_quintal': None, 'price_per_kg': None, 'demand_level': 'unknown', 'price_volatility': 'unknown', 'message': 'Market price unavailable — no recent data for this crop.', 'degraded': True, } modal_kg = price_data['modal_price_per_kg'] min_kg = price_data['min_price_per_kg'] max_kg = price_data['max_price_per_kg'] modal_quintal = price_data['modal_price_per_quintal'] # Price spread as volatility indicator (formula not hardcoded) spread_pct = (max_kg - min_kg) / (modal_kg + 1e-9) if spread_pct > 0.3: volatility = 'high' elif spread_pct > 0.15: volatility = 'moderate' else: volatility = 'low' return { 'price_forecast_per_quintal': modal_quintal, 'price_per_kg': modal_kg, 'min_price_per_kg': min_kg, 'max_price_per_kg': max_kg, 'demand_level': 'moderate', 'price_volatility': volatility, 'source': price_data['source'], 'date': price_data['date'], 'degraded': False, } except Exception as exc: logger.error(f"Market model failed: {exc}", exc_info=True) return { 'price_forecast_per_quintal': None, 'price_per_kg': None, 'demand_level': 'unknown', 'price_volatility': 'unknown', 'message': 'Market price unavailable.', 'degraded': True, }