File size: 4,207 Bytes
5ef46eb 1f9ead2 5ef46eb | 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 | import torch
import gradio as gr
import numpy as np
import pandas as pd
from scipy.stats import norm
from src.model import EnhancedLSTM
def simulate_risk_score(rainfall, evaporation, inflow, outflow, population, water_usage_model, district):
"""
Simulate and calculate the risk score for flood and drought, along with storage change details.
Parameters:
rainfall (float): Monthly rainfall in mm.
evaporation (float): Monthly potential evapotranspiration (PET) in mm.
inflow (float): Monthly reservoir inflow in m³.
outflow (float): Monthly reservoir outflow in m³.
population (int): Population of the region.
water_usage_model (callable): Predictive model for water usage based on population.
Returns:
dict: Risk scores for flood and drought, including changes in inflow, outflow, and storage.
"""
water_balance = rainfall - evaporation
#water_usage = 2e6
df = pd.read_csv('data/usage.csv')['District']
unique_districts = df.unique()
df = pd.DataFrame(unique_districts, columns=['District'])
one_hot = pd.get_dummies(df['District'], prefix='District')
df = pd.concat([df, one_hot], axis=1)
input = np.array(df[df["District"] == district].iloc[:, 1:].values.astype(int))
input = np.append(input, [[population]], axis=1)
input = torch.tensor(input, dtype=torch.float32).unsqueeze(0)
print(input.size())
with torch.no_grad():
water_usage = water_usage_model(input).sum().item() * 1e6
adjusted_inflow = inflow * (1 + water_balance / 100) # Modify inflow based on water balance (rainfall/evaporation)
adjusted_outflow = outflow * (1 - water_balance / 100) # Modify outflow based on water balance
# Calculate net water balance
net_water_balance = water_balance + (adjusted_inflow - adjusted_outflow) / 1e6
# Calculate storage change in the district
storage_change = adjusted_inflow - adjusted_outflow # Change in storage is the difference between inflow and outflow
# Aggregate water balance for SPEI calculation
aggregated_balance = [water_balance, net_water_balance]
mu, sigma = np.mean(aggregated_balance), np.std(aggregated_balance)
print(mu,sigma)
spei = (net_water_balance - mu) / sigma
# Define flood and drought risks based on SPEI value
if spei <= -2:
drought_risk = "High Risk"
flood_risk = "Low Risk"
elif spei >= 2:
drought_risk = "Low Risk"
flood_risk = "High Risk"
else:
drought_risk = "Moderate Risk"
flood_risk = "Moderate Risk"
drought_score = max(0, min(100, (1 - norm.cdf(spei)) * 100)) * (water_usage / 1e6)
flood_score = max(0, min(100, norm.cdf(spei) * 100)) * (water_usage / 1e6)
return {
"SPEI": spei,
"Drought Risk": drought_risk,
"Flood Risk": flood_risk,
"Drought Score": drought_score,
"Flood Score": flood_score,
"Adjusted Inflow": adjusted_inflow,
"Adjusted Outflow": adjusted_outflow,
"Storage Change": storage_change,
}
input_size = 14
output_size = 3
model = EnhancedLSTM(input_size=input_size, lstm_layer_sizes=[128]*3, linear_layer_size=[64]*6, output_size=output_size)
model.load_state_dict(torch.load('./model/res_1.pt', map_location='cpu'))
model.eval()
def predict(rainfall, evaporation, inflow, outflow, population, district):
result = simulate_risk_score(rainfall, evaporation, inflow, outflow, population, district)
return result
# Gradio input-output setup
inputs = [
gr.Number(label="Rainfall (mm)"),
gr.Number(label="Evaporation (mm)"),
gr.Number(label="Inflow (m³)"),
gr.Number(label="Outflow (m³)"),
gr.Number(label="Population"),
gr.Textbox(label="District")
]
outputs = gr.JSON(label="Risk Assessment Results")
# Create the Gradio interface
gr_interface = gr.Interface(
fn=predict,
inputs=inputs,
outputs=outputs,
title="Flood and Drought Risk Assessment",
description="Simulate and assess flood and drought risks based on environmental and demographic factors."
)
# Run the interface locally for testing
if __name__ == "__main__":
gr_interface.launch() |