|
|
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() |
|
|
|
|
|
|
|
|
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] |
|
|
|
|
|
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() |
|
|
|
|
|
|
|
|
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] |
|
|
|
|
|
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], |
|
|
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 |
|
|
] |
