File size: 12,081 Bytes
711bc31 ac49be7 711bc31 ac49be7 711bc31 ac49be7 711bc31 ac49be7 711bc31 |
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 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 |
from typing import Callable
from plotly.graph_objects import Figure
import plotly.graph_objects as go
import pandas as pd
import geojson
from climateqa.engine.talk_to_data.ipcc.config import IPCC_INDICATOR_TO_COLORSCALE, IPCC_INDICATOR_TO_UNIT, IPCC_SCENARIO
from climateqa.engine.talk_to_data.ipcc.plot_informations import choropleth_map_informations, indicator_evolution_informations, indicator_specific_month_evolution_informations
from climateqa.engine.talk_to_data.ipcc.queries import indicator_for_given_year_query, indicator_per_year_and_specific_month_at_location_query, indicator_per_year_at_location_query
from climateqa.engine.talk_to_data.objects.plot import Plot
def generate_geojson_polygons(latitudes: list[float], longitudes: list[float], indicators: list[float]) -> geojson.FeatureCollection:
features = [
geojson.Feature(
geometry=geojson.Polygon([[
[lon - 0.5, lat - 0.5],
[lon + 0.5, lat - 0.5],
[lon + 0.5, lat + 0.5],
[lon - 0.5, lat + 0.5],
[lon - 0.5, lat - 0.5]
]]),
properties={"value": val},
id=str(idx)
)
for idx, (lat, lon, val) in enumerate(zip(latitudes, longitudes, indicators))
]
geojson_data = geojson.FeatureCollection(features)
return geojson_data
def plot_indicator_evolution_at_location_historical_and_projections(
params: dict,
) -> Callable[[pd.DataFrame], Figure]:
"""
Returns a function that generates a line plot showing the evolution of a climate indicator
(e.g., temperature, rainfall) over time at a specific location, including both historical data
and future projections for different climate scenarios.
Args:
params (dict): Dictionary with:
- indicator_column (str): Name of the climate indicator column to plot.
- location (str): Location (e.g., country, city) for which to plot the indicator.
Returns:
Callable[[pd.DataFrame], Figure]: Function that takes a DataFrame and returns a Plotly Figure
showing the indicator's evolution over time, with scenario lines and historical data.
"""
indicator = params["indicator_column"]
location = params["location"]
indicator_label = " ".join(word.capitalize() for word in indicator.split("_"))
unit = IPCC_INDICATOR_TO_UNIT.get(indicator, "")
def plot_data(df: pd.DataFrame) -> Figure:
df = df.sort_values(by='year')
years = df['year'].astype(int).tolist()
indicators = df[indicator].astype(float).tolist()
scenarios = df['scenario'].astype(str).tolist()
# Find last historical value for continuity
last_historical = [(y, v) for y, v, s in zip(years, indicators, scenarios) if s == 'historical']
last_historical_year, last_historical_indicator = last_historical[-1] if last_historical else (None, None)
fig = go.Figure()
for scenario in IPCC_SCENARIO:
x = [y for y, s in zip(years, scenarios) if s == scenario]
y = [v for v, s in zip(indicators, scenarios) if s == scenario]
# Connect historical to scenario
if scenario != 'historical' and last_historical_indicator is not None:
x = [last_historical_year] + x
y = [last_historical_indicator] + y
fig.add_trace(go.Scatter(
x=x,
y=y,
mode='lines',
name=scenario
))
fig.update_layout(
title=f'Yearly Evolution of {indicator_label} in {location} (Historical + SSP Scenarios)',
xaxis_title='Year',
yaxis_title=f'{indicator_label} ({unit})',
legend_title='Scenario',
height=800,
)
return fig
return plot_data
indicator_evolution_at_location_historical_and_projections: Plot = {
"name": "Indicator Evolution at Location (Historical + Projections)",
"description": (
"Shows how a climate indicator (e.g., rainfall, temperature) changes over time at a specific location, "
"including historical data and future projections. "
"Useful for questions about the value or trend of an indicator at a location for any year, "
"such as 'What will be the total rainfall in China in 2050?' or 'How does rainfall evolve in China over time?'. "
"Parameters: indicator_column (the climate variable), location (e.g., country, city)."
),
"params": ["indicator_column", "location"],
"plot_function": plot_indicator_evolution_at_location_historical_and_projections,
"sql_query": indicator_per_year_at_location_query,
"plot_information": indicator_evolution_informations,
"short_name": "Evolution"
}
def plot_indicator_monthly_evolution_at_location(
params: dict,
) -> Callable[[pd.DataFrame], Figure]:
"""
Returns a function that generates a line plot showing the evolution of a climate indicator
for a specific month over time at a specific location, including both historical data
and future projections for different climate scenarios.
Args:
params (dict): Dictionary with:
- indicator_column (str): Name of the climate indicator column to plot.
- location (str): Location (e.g., country, city) for which to plot the indicator.
- month (str): Month name to plot.
Returns:
Callable[[pd.DataFrame], Figure]: Function that takes a DataFrame and returns a Plotly Figure.
"""
indicator = params["indicator_column"]
location = params["location"]
month = params["month_name"]
indicator_label = " ".join(word.capitalize() for word in indicator.split("_"))
unit = IPCC_INDICATOR_TO_UNIT.get(indicator, "")
def plot_data(df: pd.DataFrame) -> Figure:
df = df.sort_values(by='year')
years = df['year'].astype(int).tolist()
indicators = df[indicator].astype(float).tolist()
scenarios = df['scenario'].astype(str).tolist()
# Find last historical value for continuity
last_historical = [(y, v) for y, v, s in zip(years, indicators, scenarios) if s == 'historical']
last_historical_year, last_historical_indicator = last_historical[-1] if last_historical else (None, None)
fig = go.Figure()
for scenario in IPCC_SCENARIO:
x = [y for y, s in zip(years, scenarios) if s == scenario]
y = [v for v, s in zip(indicators, scenarios) if s == scenario]
# Connect historical to scenario
if scenario != 'historical' and last_historical_indicator is not None:
x = [last_historical_year] + x
y = [last_historical_indicator] + y
fig.add_trace(go.Scatter(
x=x,
y=y,
mode='lines',
name=scenario
))
fig.update_layout(
title=f'Evolution of {indicator_label} in {month} in {location} (Historical + SSP Scenarios)',
xaxis_title='Year',
yaxis_title=f'{indicator_label} ({unit})',
legend_title='Scenario',
height=800,
)
return fig
return plot_data
indicator_specific_month_evolution_at_location: Plot = {
"name": "Indicator specific month Evolution at Location (Historical + Projections)",
"description": (
"Shows how a climate indicator (e.g., rainfall, temperature) for a specific month changes over time at a specific location, "
"including historical data and future projections. "
"Useful for questions about the value or trend of an indicator for a given month at a location, "
"such as 'How does July temperature evolve in Paris over time?'. "
"Parameters: indicator_column (the climate variable), location (e.g., country, city), month (1-12)."
),
"params": ["indicator_column", "location", "month"],
"plot_function": plot_indicator_monthly_evolution_at_location,
"sql_query": indicator_per_year_and_specific_month_at_location_query,
"plot_information": indicator_specific_month_evolution_informations,
"short_name": "Evolution for a specific month"
}
def plot_choropleth_map_of_country_indicator_for_specific_year(
params: dict,
) -> Callable[[pd.DataFrame], Figure]:
"""
Returns a function that generates a choropleth map (heatmap) showing the spatial distribution
of a climate indicator (e.g., temperature, rainfall) across all regions of a country for a specific year.
Args:
params (dict): Dictionary with:
- indicator_column (str): Name of the climate indicator column to plot.
- year (str or int, optional): Year for which to plot the indicator (default: 2050).
- country_name (str): Name of the country.
- location (str): Location (country or region) for the map.
Returns:
Callable[[pd.DataFrame], Figure]: Function that takes a DataFrame and returns a Plotly Figure
showing the indicator's spatial distribution as a choropleth map for the specified year.
"""
indicator = params["indicator_column"]
year = params.get('year')
if year is None:
year = 2050
country_name = params['country_name']
location = params['location']
indicator_label = " ".join(word.capitalize() for word in indicator.split("_"))
unit = IPCC_INDICATOR_TO_UNIT.get(indicator, "")
def plot_data(df: pd.DataFrame) -> Figure:
indicators = df[indicator].astype(float).tolist()
latitudes = df["latitude"].astype(float).tolist()
longitudes = df["longitude"].astype(float).tolist()
geojson_data = generate_geojson_polygons(latitudes, longitudes, indicators)
fig = go.Figure(go.Choroplethmapbox(
geojson=geojson_data,
locations=[str(i) for i in range(len(indicators))],
featureidkey="id",
z=indicators,
colorscale=IPCC_INDICATOR_TO_COLORSCALE[indicator],
zmin=min(indicators),
zmax=max(indicators),
marker_opacity=0.7,
marker_line_width=0,
colorbar_title=f"{indicator_label} ({unit})",
text=[f"{indicator_label}: {value:.2f} {unit}" for value in indicators], # Add hover text showing the indicator value
hoverinfo="text"
))
fig.update_layout(
mapbox_style="open-street-map",
mapbox_zoom=2,
height=800,
mapbox_center={
"lat": latitudes[len(latitudes)//2] if latitudes else 0,
"lon": longitudes[len(longitudes)//2] if longitudes else 0
},
coloraxis_colorbar=dict(title=f"{indicator_label} ({unit})"),
title=f"{indicator_label} in {year} in {location} ({country_name})"
)
return fig
return plot_data
choropleth_map_of_country_indicator_for_specific_year: Plot = {
"name": "Choropleth Map of a Country's Indicator Distribution for a Specific Year",
"description": (
"Displays a map showing the spatial distribution of a climate indicator (e.g., rainfall, temperature) "
"across all regions of a country for a specific year. "
"Can answer questions about the value of an indicator in a country or region for a given year, "
"such as 'What will be the total rainfall in China in 2050?' or 'How is rainfall distributed across China in 2050?'. "
"Parameters: indicator_column (the climate variable), year, location (country name)."
),
"params": ["indicator_column", "year", "location"],
"plot_function": plot_choropleth_map_of_country_indicator_for_specific_year,
"sql_query": indicator_for_given_year_query,
"plot_information": choropleth_map_informations,
"short_name": "Map",
}
IPCC_PLOTS = [
indicator_evolution_at_location_historical_and_projections,
choropleth_map_of_country_indicator_for_specific_year,
indicator_specific_month_evolution_at_location
] |