# src/server/simulation/crop_growth.py """ Crop growth model for AquaGuard-RL. Implements yield response to water, soil fertility, and temperature based on FAO Penman-Monteith-inspired water stress functions. References: - FAO Crop Yield Response to Water (Doorenbos & Kassam): https://www.fao.org/3/i2800e/i2800e.pdf - FAO AQUASTAT crop water requirements: https://www.fao.org/aquastat/en/ """ from __future__ import annotations import logging from typing import Dict logger = logging.getLogger(__name__) # Temperature above which yield penalty applies (°C anomaly threshold) TEMP_PENALTY_THRESHOLD_C = 2.0 # Yield loss per °C above threshold (Lobell et al., 6% per °C) YIELD_LOSS_PER_DEGREE = 0.06 class CropGrowthModel: """ Simulates crop yield as a function of water availability, soil health, and climate. The yield model: actual_yield = base_yield × water_stress_factor × soil_fertility_factor × temperature_factor × irrigation_efficiency_factor All factors are in [0, 1] (multiplicative penalties). """ def compute_yield( self, base_yield: float, water_available_mm: float, water_requirement_mm: float, optimal_water_mm: float, wilting_point_mm: float, soil_fertility: float, temperature_anomaly: float, irrigation_method: str, ) -> float: """ Compute crop yield for a given season and conditions. Args: base_yield: Potential yield under ideal conditions (t/ha). water_available_mm: Total water available to the crop this season (mm). water_requirement_mm: Full-season water requirement (mm). optimal_water_mm: Water level at which stress begins below this point (mm). wilting_point_mm: Minimum water for survival (below = near-zero yield). soil_fertility: Soil fertility index [0, 1]. temperature_anomaly: Deviation from historical average temperature (°C). irrigation_method: Active irrigation method ('flood', 'sprinkler', 'drip'). Returns: Estimated yield in tons per hectare. """ wsf = self.compute_water_stress(water_available_mm, water_requirement_mm, wilting_point_mm) sff = self.compute_soil_fertility_factor(soil_fertility) tf = self.compute_temperature_factor(temperature_anomaly) ief = self.compute_irrigation_efficiency_factor(irrigation_method) yield_t_per_ha = base_yield * wsf * sff * tf * ief yield_t_per_ha = max(0.0, yield_t_per_ha) logger.debug( f"Yield: base={base_yield:.2f} × wsf={wsf:.3f} × sff={sff:.3f} × " f"tf={tf:.3f} × ief={ief:.3f} = {yield_t_per_ha:.3f} t/ha" ) return yield_t_per_ha def compute_water_stress( self, water_available_mm: float, water_requirement_mm: float, wilting_point_mm: float, ) -> float: """ Compute Penman-Monteith-inspired water stress factor. The stress function has three regimes: - Below wilting point (< 30% requirement): severe stress → factor ~0.1 - Deficit stress (30-100% of requirement): linear interpolation - Adequate water (≥ requirement): no stress → factor = 1.0 Args: water_available_mm: Available water this season (mm). water_requirement_mm: Full crop water requirement (mm). wilting_point_mm: Minimum water for crop survival (mm). Returns: Water stress factor in [0.05, 1.0]. """ w = water_available_mm req = water_requirement_mm wp = wilting_point_mm if w >= req: return 1.0 # Optimal or excess water elif w < wp: return 0.05 # Catastrophic: below wilting point elif w < 0.4 * req: # Severe stress: wilting point to 40% of requirement # Linearly interpolate from 0.05 to 0.25 t = (w - wp) / (0.4 * req - wp + 1e-6) return 0.05 + t * 0.20 elif w < 0.7 * req: # Moderate stress: 40-70% of requirement # Linearly interpolate from 0.25 to 0.60 t = (w - 0.4 * req) / (0.3 * req + 1e-6) return 0.25 + t * 0.35 else: # Mild stress: 70-100% of requirement # Linearly interpolate from 0.60 to 1.0 t = (w - 0.7 * req) / (0.3 * req + 1e-6) return 0.60 + t * 0.40 def compute_soil_fertility_factor(self, soil_fertility: float) -> float: """ Compute yield multiplier based on soil fertility. Uses a sublinear response: degraded soils (fertility < 0.5) have disproportionately lower yields due to nutrient limitations. Args: soil_fertility: Soil fertility index [0, 1]. Returns: Fertility factor in [0.1, 1.0]. """ fertility = max(0.0, min(1.0, soil_fertility)) if fertility >= 0.7: # Good soil: near-linear response return 0.80 + (fertility - 0.7) / 0.3 * 0.20 elif fertility >= 0.4: # Moderate soil: reduced yield return 0.50 + (fertility - 0.4) / 0.3 * 0.30 else: # Poor soil: severe yield limitation return 0.10 + fertility / 0.4 * 0.40 def compute_temperature_factor(self, temperature_anomaly: float) -> float: """ Compute yield multiplier based on temperature deviation from historical average. No penalty for anomalies < +2°C. Above +2°C, applies 6% yield loss per degree (calibrated from Lobell et al. 2011, Nature Climate Change). Args: temperature_anomaly: Deviation from historical average in °C. Returns: Temperature factor in [0.1, 1.0]. """ if temperature_anomaly <= TEMP_PENALTY_THRESHOLD_C: return 1.0 excess_degrees = temperature_anomaly - TEMP_PENALTY_THRESHOLD_C penalty = excess_degrees * YIELD_LOSS_PER_DEGREE return max(0.1, 1.0 - penalty) def compute_irrigation_efficiency_factor(self, irrigation_method: str) -> float: """ Compute yield multiplier from irrigation method efficiency. Drip and sprinkler irrigation provide more uniform water distribution, reducing waterlogging and improving nutrient uptake. Args: irrigation_method: Method name ('flood', 'sprinkler', 'drip'). Returns: Irrigation efficiency factor. """ factors = { "flood": 1.00, "sprinkler": 1.08, # 8% yield improvement (uniform coverage) "drip": 1.15, # 15% yield improvement (root zone delivery) } return factors.get(irrigation_method, 1.00) def compute_water_productivity( self, yield_t_per_ha: float, water_used_mm: float, area_ha: float, ) -> float: """ Compute water productivity (kg/m³). A key metric for irrigation efficiency assessment. Args: yield_t_per_ha: Crop yield in tons per hectare. water_used_mm: Total water used in mm. area_ha: Cropped area in hectares. Returns: Water productivity in kg grain per m³ of water. """ if water_used_mm <= 0 or area_ha <= 0: return 0.0 total_yield_kg = yield_t_per_ha * area_ha * 1000 # t → kg total_water_m3 = water_used_mm / 1000 * area_ha * 10000 # mm × ha → m³ return total_yield_kg / total_water_m3