# src/server/simulation/groundwater.py """ Groundwater dynamics model for AquaGuard-RL. Implements a simplified multi-zone confined aquifer model calibrated with Central Ground Water Board (CGWB) data for North India. Model features: - Seasonal water balance: extraction vs recharge vs rainfall - Inter-zone lateral flow (Darcy's Law approximation) - Aquifer collapse detection and permanent land degradation - Sustainable extraction enforcement References: - CGWB Annual Report 2023: https://cgwb.gov.in/en/reports - Darcy's Law: Q = K × A × (dh/dl) """ from __future__ import annotations import logging from typing import Dict, List, Tuple logger = logging.getLogger(__name__) # Fraction of depth differential transferred between adjacent zones per season INTER_ZONE_CONDUCTIVITY = 0.15 class GroundwaterModel: """ Simulates multi-zone aquifer dynamics for one growing season at a time. The model tracks: 1. Natural annual recharge (CGWB district estimates) 2. Monsoon rainfall recharge (12% of rainfall reaches aquifer — CGWB all-India) 3. Groundwater extraction for irrigation (bounded by extraction limits) 4. Inter-zone lateral flow driven by hydraulic head gradients 5. Aquifer collapse detection (irreversible) All depth values are in meters (depth to water table, positive = deeper/more depleted). """ def advance( self, current_depth_m: float, extraction_limit_m: float, rainfall_mm: float, recharge_rate_mm_yr: float, storage_coefficient: float, irrigation_demand_mm: float, gwi_fraction: float, rainfall_recharge_fraction: float, ) -> Tuple[float, float]: """ Advance groundwater depth for one growing season (~4 months). The water balance equation: net_change = total_recharge - actual_extraction depth_change = -net_change / storage_coefficient / 1000 Args: current_depth_m: Current depth to water table in meters. extraction_limit_m: Agent-imposed extraction limit in meters/season. rainfall_mm: Actual realized rainfall for this season in mm. recharge_rate_mm_yr: Natural annual recharge rate in mm/year. storage_coefficient: Aquifer storage coefficient (dimensionless). irrigation_demand_mm: Irrigation water demand in mm. gwi_fraction: Fraction of irrigation met by groundwater (0–1). rainfall_recharge_fraction: Fraction of rainfall recharging aquifer (0–1). Returns: Tuple of (new_depth_m, water_extracted_m) where water_extracted_m is the volume extracted expressed as equivalent aquifer depth change. """ # Seasonal natural recharge (1/3 of annual for a ~4-month season) seasonal_natural_recharge_mm = recharge_rate_mm_yr * (1.0 / 3.0) # Rainfall recharge (12% of rainfall reaches the aquifer, CGWB estimate) rainfall_recharge_mm = rainfall_mm * rainfall_recharge_fraction # Total recharge this season total_recharge_mm = seasonal_natural_recharge_mm + rainfall_recharge_mm # Groundwater extraction from irrigation gw_extraction_demand_mm = irrigation_demand_mm * gwi_fraction # Enforce extraction limit (convert m/season → mm equivalent for water balance) # extraction_limit_m is max depth drawdown allowed per season max_extraction_mm = extraction_limit_m * 1000.0 * storage_coefficient actual_extraction_mm = min(gw_extraction_demand_mm, max_extraction_mm) # Net change in water storage (positive = more water stored = table rising) net_change_mm = total_recharge_mm - actual_extraction_mm # Convert storage change to water table depth change # ΔH = ΔStorage_mm / (StorageCoefficient × 1000) # Positive net_change → water level rises → depth decreases (negative depth change) depth_change_m = -(net_change_mm / 1000.0) / storage_coefficient new_depth_m = max(0.1, current_depth_m + depth_change_m) # floor at 0.1m (artesian) water_extracted_m = actual_extraction_mm / 1000.0 logger.debug( f"GW balance: depth {current_depth_m:.2f}→{new_depth_m:.2f}m | " f"recharge={total_recharge_mm:.1f}mm (natural={seasonal_natural_recharge_mm:.1f}, " f"rain={rainfall_recharge_mm:.1f}) | extraction={actual_extraction_mm:.1f}mm | " f"net={net_change_mm:.1f}mm" ) return new_depth_m, water_extracted_m def apply_lateral_flow( self, zone_states: Dict[str, Dict], zone_data: Dict[str, Dict], ) -> None: """ Apply inter-zone lateral groundwater flow based on Darcy's Law approximation. Water flows from zones with shallower tables (less depleted) to zones with deeper tables (more depleted), driven by hydraulic head gradients. This models the hydrological connectivity of shared aquifer systems — aggressive extraction in one zone affects neighboring zones. Modifies zone_states in-place. Args: zone_states: Current zone simulation states (modified in-place). zone_data: Static zone parameters (used for storage coefficients). """ zone_ids = [zid for zid, zs in zone_states.items() if not zs.get("is_collapsed", False)] if len(zone_ids) < 2: return # Snapshot current depths before applying flows depths_before = {zid: zone_states[zid]["gw_depth_m"] for zid in zone_ids} for i, zone_a_id in enumerate(zone_ids): for zone_b_id in zone_ids[i + 1:]: depth_a = depths_before[zone_a_id] depth_b = depths_before[zone_b_id] depth_diff = depth_a - depth_b if abs(depth_diff) < 0.5: continue # No meaningful hydraulic gradient # Darcy-approximated lateral flow # Water flows FROM deeper (more depleted) zone TO shallower (less depleted) # depth_diff = depth_a - depth_b; positive means A is deeper # Flow magnitude proportional to depth differential flow_magnitude = abs(depth_diff) * INTER_ZONE_CONDUCTIVITY sc_a = zone_data[zone_a_id]["storage_coefficient"] sc_b = zone_data[zone_b_id]["storage_coefficient"] # Depth change: zones converge toward each other change_a = flow_magnitude * (sc_b / (sc_a + sc_b)) change_b = flow_magnitude * (sc_a / (sc_a + sc_b)) if depth_diff > 0: # Zone A is deeper → gains water (depth decreases), Zone B loses water (depth increases) zone_states[zone_a_id]["gw_depth_m"] = max(0.1, zone_states[zone_a_id]["gw_depth_m"] - change_a) zone_states[zone_b_id]["gw_depth_m"] = max(0.1, zone_states[zone_b_id]["gw_depth_m"] + change_b) else: # Zone B is deeper → gains water (depth decreases), Zone A loses water (depth increases) zone_states[zone_b_id]["gw_depth_m"] = max(0.1, zone_states[zone_b_id]["gw_depth_m"] - change_b) zone_states[zone_a_id]["gw_depth_m"] = max(0.1, zone_states[zone_a_id]["gw_depth_m"] + change_a) logger.debug( f"Lateral flow applied. Depths: " + " | ".join(f"{zid}={zone_states[zid]['gw_depth_m']:.2f}m" for zid in zone_ids) ) def compute_sustainable_extraction( self, recharge_rate_mm_yr: float, storage_coefficient: float, safety_factor: float = 1.5, ) -> float: """ Compute the maximum sustainable seasonal extraction in meters. Allows up to safety_factor × natural recharge, enabling some managed overdraft in drought years while preventing long-term depletion. Args: recharge_rate_mm_yr: Natural annual recharge rate in mm/year. storage_coefficient: Aquifer storage coefficient. safety_factor: Multiplier on natural recharge (default 1.5 = 50% overdraft). Returns: Maximum sustainable extraction in meters/season. """ seasonal_recharge_mm = recharge_rate_mm_yr / 3.0 max_extraction_mm = seasonal_recharge_mm * safety_factor return (max_extraction_mm / 1000.0) / storage_coefficient