import math import os import pickle import pandas as pd from huggingface_hub import hf_hub_download from concurrent.futures import ThreadPoolExecutor repo_id = "Navanihk/rainfallPrediction" # Use /app as primary cache directory for Hugging Face Spaces cache_dir = '/app/hf_cache' fallback_cache_dir = '/tmp/hf_cache' os.makedirs(cache_dir, exist_ok=True) os.makedirs(fallback_cache_dir, exist_ok=True) def predict_rainfall_for_date(df_long, model, district_name, target_date, lags=7): """ Predict rainfall for a specific district and date. df_long: original dataset (with columns ['district','state','date','rainfall']) model: trained LightGBM model district_name: name of the district to predict target_date: pd.Timestamp or string in 'YYYY-MM-DD' lags: number of lag days to use """ # Ensure target_date is Timestamp target_date = pd.Timestamp(target_date) if not pd.api.types.is_datetime64_any_dtype(df_long['date']): df_long = df_long.copy() df_long['date'] = pd.to_datetime(df_long['date']) # Get district data up to the day before target_date district_data = df_long[(df_long['district']==district_name) & (df_long['date'] < target_date)].sort_values('date') # Check if enough data for lags if len(district_data) < lags: raise ValueError(f"Not enough data to create {lags} lag features for {district_name}") # Get last 'lags' days of rainfall (most recent first) lag_values = list(district_data['rainfall'].tail(lags).values[::-1]) # Create feature vector: lag + date features features = lag_values + [target_date.month, target_date.day, target_date.dayofyear] # Predict pred_rain = model.predict([features])[0] return { 'district': district_name, 'state': district_data['state'].iloc[0], 'date': target_date, 'predicted_rainfall': pred_rain } def rainFallPrediction(location,RainFallData): """Predict average annual rainfall for a given location using RainFallData""" try: # Try to download the model with fallback cache directories try: model_path = hf_hub_download(repo_id=repo_id, filename="rainfall.pkl", cache_dir=cache_dir) except PermissionError: # Try fallback cache directory model_path = hf_hub_download(repo_id=repo_id, filename="rainfall.pkl", cache_dir=fallback_cache_dir) # Load the actual model from the downloaded file with open(model_path, 'rb') as f: model = pickle.load(f) # model = pickle.load(open('rainfall.pkl','rb')) date = "2025-01-10" print("predict_rainfall_for_date called") pred_result = predict_rainfall_for_date(RainFallData, model, location, date) print(pred_result) return pred_result['predicted_rainfall'] except Exception as e: print(f"Error loading ML model: {e}") # Fallback to simple lookup from the provided data location_lower = location.lower() # Try to find matching district in the provided data if 'district' in RainFallData.columns and 'rainfall' in RainFallData.columns: matching_rows = RainFallData[RainFallData['district'].str.lower().str.contains(location_lower, na=False)] if not matching_rows.empty: return matching_rows['rainfall'].iloc[0] # Default rainfall values for major cities (mm/year) default_rainfall = { "karur": 800, "new delhi": 700, "delhi": 700, "mumbai": 2200, "bangalore": 970, "chennai": 1400, "hyderabad": 800, "pune": 600, "kolkata": 1600, "ahmedabad": 800, "jaipur": 650 } return default_rainfall.get(location_lower, 750) # Default to 750mm if location not found def daily_to_annual(daily_mm, method="simple", month=None, monthly_avg=None, mu=None, sigma=None, sims=10000): if method == "simple": return daily_mm * 365 if method == "monthly_factor": if month is None or monthly_avg is None: raise ValueError("month and monthly_avg (array of 12) required") days = calendar.monthrange(2024, month)[1] # only for days count # naive: scale predicted day to that month then sum using monthly averages monthly_est = [monthly_avg[i] if i+1 != month else daily_mm * days for i in range(12)] return sum(monthly_est) if method == "monte_carlo": if mu is None or sigma is None: raise ValueError("mu and sigma required for monte_carlo") draws = np.random.normal(mu, sigma, size=(sims, 365)) draws[draws < 0] = 0.0 totals = draws.sum(axis=1) return {"mean": totals.mean(), "median": np.median(totals), "p10": np.percentile(totals,10), "p90": np.percentile(totals,90)} raise ValueError("unknown method") def get_groundwater_level(lat, lon, date): """ Fetch groundwater level from India WRIS API for a given lat, lon, and date. This is a simplified implementation using Erode, Tamil Nadu as an example. """ # In production, you would reverse geocode lat/lon to district/state # Here, we use Erode, Tamil Nadu for demonstration import requests url = ( "https://indiawris.gov.in/Dataset/Ground Water Level?stateName=Tamil%20Nadu&districtName=Karur&agencyName=CGWB&startdate=2025-09-14&enddate=2025-09-14&download=false&page=0&size=1" ) return float(15) try: resp = requests.post(url, timeout=10) data = resp.json() print(data) if resp: print(data) if data: gw_level = data['data'][0].get('dataValue', 15) return float(gw_level) except Exception as e: print(f"Error fetching groundwater level: {e}") return 15 # fallback default def get_soil_type(lat, lon): """ Fetch soil type from OpenEPI API for a given lat, lon. """ import requests url = f"https://api.openepi.io/soil/type?lat={lat}&lon={lon}&top_k=3" try: resp = requests.get(url, timeout=10) if resp.status_code == 200: data = resp.json() # Parse soil type from response (update as per actual API response) if data: return data['properties'].get('most_probable_soil_type', 'Loamy') except Exception as e: print(f"Error fetching soil type: {e}") return "Loamy" # fallback default def RooftTopCalculation(input_data,RainFallData): """ Calculate rainwater harvesting parameters based on input data """ # Extract input parameters location = input_data.get('location', 'Unknown') roof_area = input_data.get('roof_area', 0) open_space = input_data.get('open_space', 0) roof_type = input_data.get('roof_type', 'null') dwellers = input_data.get('dwellers', 1) coordinates = get_coordinates(location) lat, lon = coordinates[0], coordinates[1] # Fetch groundwater level and soil type in parallel import requests from datetime import datetime today = datetime.now().strftime('%Y-%m-%d') with ThreadPoolExecutor(max_workers=5) as executor: future_gw = executor.submit(get_groundwater_level, lat, lon, today) future_soil = executor.submit(get_soil_type, lat, lon) groundwater_level = future_gw.result() print("get_groundwater_level called") soil_type = future_soil.result() print("get_soil_type called") # Default values for calculations (these would typically come from ML models or databases) # Using Delhi-like values as defaults default_rainfall = rainFallPrediction(location,RainFallData) # mm/year default_runoff_coefficient = 0.8 # typical for concrete/tiled roofs # groundwater_level = 15 # meters below ground level # soil_type = "Loamy" aquifer_type = "Alluvial Aquifer" print("reult",groundwater_level,soil_type) # Calculate runoff volume (in liters/year) runoff_volume = roof_area * default_rainfall * default_runoff_coefficient # Calculate harvestable water (considering losses - typically 15-20% loss) loss_factor = 0.15 harvestable_water = int(runoff_volume * (1 - loss_factor)) # Calculate recharge potential (typically 60-70% of harvestable water) recharge_potential = int(harvestable_water * 0.65) # Determine structure suggestion based on available space and groundwater level structure_suggestion = determine_structure(open_space, groundwater_level, roof_area) # Calculate recommended dimensions dimensions = calculate_dimensions(structure_suggestion, harvestable_water) # Calculate capacity capacity = dimensions['length'] * dimensions['width'] * dimensions['depth'] * 1000 # in liters # Cost analysis estimated_cost = calculate_cost(structure_suggestion, dimensions, roof_area) water_saved_value = calculate_water_savings(harvestable_water) roi_years = round(estimated_cost / water_saved_value, 1) if water_saved_value > 0 else 0 # Feasibility score feasibility_score = calculate_feasibility(roof_area, open_space, groundwater_level) # Prepare response response = { "user": { "name": "User", # Default name as not provided in input "location": { "lat": coordinates[0], "lon": coordinates[1], "address": location }, "roof_area": roof_area, "open_space": open_space, "dwellers": dwellers }, "reference_data": { "rainfall": f"{default_rainfall} mm/year", "aquifer": aquifer_type, "groundwater_level": f"{groundwater_level} m bgl", "soil_type": soil_type, "runoff_coefficient": default_runoff_coefficient }, "calculations": { "runoff_volume": f"{int(runoff_volume)} liters/year", "harvestable_water": f"{harvestable_water} liters/year", "recharge_potential": f"{recharge_potential} liters/year", "structure_suggestion": structure_suggestion, "recommended_dimensions": dimensions, "capacity": f"{int(capacity)} liters" }, "cost_analysis": { "estimated_cost": f"₹{estimated_cost:,}", "water_saved_value": f"₹{int(water_saved_value):,}/year", "roi": f"{roi_years} years" }, "gis_data": { "map_coordinates": coordinates, "feasibility_score": feasibility_score }, "support": { "regional_language": get_regional_language(location), "faq_link": "https://cgwb.gov.in/rtrwh-faqs" } } return response def get_coordinates(location): """Get coordinates for a location (simplified - would use geocoding service in production)""" location_coordinates = { "karur": [10.9601, 78.0766], "new delhi": [28.6139, 77.2090], "delhi": [28.6139, 77.2090], "mumbai": [19.0760, 72.8777], "bangalore": [12.9716, 77.5946], "chennai": [13.0827, 80.2707], "hyderabad": [17.3850, 78.4867], "salem": [11.6643, 78.1460], "kolkata": [22.5726, 88.3639], "madurai": [9.9252, 78.1198], "erode": [11.3410, 77.7172] } location_lower = location.lower() return location_coordinates.get(location_lower, [28.6139, 77.2090]) # Default to Delhi def determine_structure(open_space, groundwater_level, roof_area): """Determine the best rainwater harvesting structure""" if open_space >= 100 and groundwater_level > 10: return "Recharge Pit" elif open_space >= 50 and groundwater_level > 5: return "Percolation Tank" elif roof_area >= 100: return "Storage Tank" else: return "Rain Barrel" def calculate_dimensions(structure_type, harvestable_water): """Calculate recommended dimensions based on structure type""" if structure_type == "Recharge Pit": # Calculate based on recharge requirements volume_needed = harvestable_water / 1000 # Convert to cubic meters depth = min(3, max(2, volume_needed / 4)) # Depth between 2-3m area_needed = volume_needed / depth side = math.sqrt(area_needed) return { "length": round(side, 1), "width": round(side, 1), "depth": round(depth, 1) } elif structure_type == "Percolation Tank": return {"length": 3.0, "width": 3.0, "depth": 2.0} elif structure_type == "Storage Tank": return {"length": 2.0, "width": 2.0, "depth": 2.5} else: # Rain Barrel return {"length": 1.0, "width": 1.0, "depth": 1.5} def calculate_cost(structure_type, dimensions, roof_area): """Calculate estimated cost in INR""" base_costs = { "Recharge Pit": 15000, "Percolation Tank": 25000, "Storage Tank": 20000, "Rain Barrel": 5000 } base_cost = base_costs.get(structure_type, 15000) volume = dimensions['length'] * dimensions['width'] * dimensions['depth'] # Additional cost based on volume and roof area additional_cost = (volume * 2000) + (roof_area * 50) return int(base_cost + additional_cost) def calculate_water_savings(harvestable_water): """Calculate annual water savings value in INR""" # Assuming water cost of ₹12 per 1000 liters water_cost_per_1000_liters = 12 annual_savings = (harvestable_water / 1000) * water_cost_per_1000_liters return annual_savings def calculate_feasibility(roof_area, open_space, groundwater_level): """Calculate feasibility score""" score = 0 if roof_area >= 100: score += 30 elif roof_area >= 50: score += 20 else: score += 10 if open_space >= 100: score += 30 elif open_space >= 50: score += 20 else: score += 10 if groundwater_level > 10: score += 40 elif groundwater_level > 5: score += 30 else: score += 20 if score >= 80: return "High" elif score >= 60: return "Medium" else: return "Low" def get_regional_language(location): """Get regional language based on location""" language_map = { "karur": "Tamil", "new delhi": "Hindi", "delhi": "Hindi", "mumbai": "Marathi", "bangalore": "Kannada", "chennai": "Tamil", "hyderabad": "Telugu", "pune": "Marathi", "kolkata": "Bengali", "ahmedabad": "Gujarati", "jaipur": "Hindi" } return language_map.get(location.lower(), "Tamil")