| import pandas as pd |
| import numpy as np |
| import plotly.graph_objects as go |
| import plotly.express as px |
| import xarray as xr |
| import solara |
| import requests |
| from uuid import uuid4 |
| from sklearn.neighbors import NearestNeighbors |
| from scipy.stats import linregress |
| import geopandas as gpd |
| from constants import REGION_COLUMN |
| from Datasets import DATASETS |
| def get_plotly_theme(): |
| return "plotly_dark" if solara.lab.theme.dark_effective else "plotly_white" |
|
|
| def get_location_bounds(query, format="json", limit=1): |
| """Fetch lat/lon bounds for a location using Nominatim API.""" |
| url = "https://nominatim.openstreetmap.org/search" |
| params = {"q": query, "format": format, "addressdetails": 1, "limit": limit} |
| headers = {"User-Agent": "SimpleGeocodingApp/1.0"} |
| try: |
| response = requests.get(url, params=params, headers=headers, timeout=10).json() |
| if not response: |
| return None |
| bounds = response[0]["boundingbox"] |
| return { |
| "location": response[0].get("display_name", "Unknown"), |
| "bounds": { |
| "min_lat": float(bounds[0]), |
| "max_lat": float(bounds[1]), |
| "min_lon": float(bounds[2]), |
| "max_lon": float(bounds[3]) |
| } |
| } |
| except Exception: |
| return None |
|
|
| def load_dataset(dataset_config): |
| dataset = xr.open_dataset(dataset_config["average_path"], decode_timedelta=False) |
| variables = list(dataset.data_vars) |
| float_vars = [var for var in variables if dataset[var].dtype in ["float32", "float64"]] |
| |
| data = dataset.to_dataframe().reset_index().dropna() |
| data = data.rename(columns={"latitude": "lat", "longitude": "lon"}) |
| |
| if len(data) >= 4: |
| sample_indices = np.random.choice(len(data), size=4, replace=False) |
| sample_coords = data[["lat", "lon"]].iloc[sample_indices].values |
| coords = data[["lat", "lon"]].values |
| |
| knn = NearestNeighbors(n_neighbors=2).fit(coords) |
| distances, _ = knn.kneighbors(sample_coords, n_neighbors=2) |
| grid_spacing = distances[:, 1].mean() |
| else: |
| grid_spacing = 0.25 |
| |
| dataset.close() |
| return data, float_vars, grid_spacing |
|
|
| def calculate_map_zoom(lat_range, lon_range): |
| range_max = max(lat_range, lon_range) |
| return (1 if range_max > 100 else |
| 2 if range_max > 30 else |
| 4 if range_max > 15 else |
| 5 if range_max > 5 else |
| 7 if range_max > 1 else |
| 9 if range_max > 0.1 else 10) |
|
|
| def create_heatmap(data, dataset_name, value_column, grid_size, color_scale, map_style, opacity, center=None, zoom=None): |
| if center is None: |
| center = {"lat": data["lat"].mean(), "lon": data["lon"].mean()} |
| if zoom is None: |
| zoom = calculate_map_zoom(data["lat"].max() - data["lat"].min(), |
| data["lon"].max() - data["lon"].min()) |
| |
| features = [ |
| { |
| "type": "Feature", "id": str(uuid4()), |
| "geometry": { |
| "type": "Polygon", |
| "coordinates": [[ |
| [row["lon"] - grid_size/2, row["lat"] - grid_size/2], |
| [row["lon"] + grid_size/2, row["lat"] - grid_size/2], |
| [row["lon"] + grid_size/2, row["lat"] + grid_size/2], |
| [row["lon"] - grid_size/2, row["lat"] + grid_size/2], |
| [row["lon"] - grid_size/2, row["lat"] - grid_size/2] |
| ]] |
| }, |
| "properties": {"value": row[value_column], "lat": row["lat"], "lon": row["lon"]} |
| } |
| for _, row in data.iterrows() |
| ] |
| |
| geojson = {"type": "FeatureCollection", "features": features} |
| fig = go.Figure( |
| go.Choroplethmap( |
| geojson=geojson, |
| locations=[f["id"] for f in features], |
| z=[f["properties"]["value"] for f in features], |
| colorscale=color_scale, |
| marker_opacity=opacity, |
| marker_line_width=0, |
| colorbar=dict(title=dataset_name), |
| customdata=np.stack(([f["properties"]["lat"] for f in features], |
| [f["properties"]["lon"] for f in features]), axis=-1), |
| hovertemplate="<b>Lat: %{customdata[0]:.3f}</b><br>Lon: %{customdata[1]:.3f}<br>Value: %{z}<extra></extra>" |
| ) |
| ) |
| fig.update_layout( |
| map=dict(style=map_style, center=center, zoom=zoom), |
| height=900, |
| width=1250, |
| title=f"Heatmap - {dataset_name}", |
| template=get_plotly_theme() |
| ) |
| return fig |
|
|
| def create_timeseries_plot(file_path, dataset_name, lat, lon, plotted_line_color, spatial_mean_data=None, show_spatial_mean=False, from_year=None, to_year=None, use_timeframe=False, region="India", show_trend_line=False, trend_line_color="red", spatial_mean_color="yellow"): |
| try: |
| dataset = xr.open_dataset(file_path, decode_timedelta=False) |
| variable = list(dataset.data_vars)[0] |
| |
| lat_idx = np.abs((dataset.get("lat", dataset.get("latitude")).values - lat)).argmin() |
| lon_idx = np.abs((dataset.get("lon", dataset.get("longitude")).values - lon)).argmin() |
| |
| timeseries = dataset[variable].isel(lat=lat_idx, lon=lon_idx).to_dataframe().reset_index() |
| |
| years = timeseries["year"].values |
| if len(years) == 0: |
| raise ValueError("No years found in the dataset") |
| start_year = int(min(years)) |
| end_year = int(max(years)) |
| |
| if use_timeframe and from_year is not None and to_year is not None: |
| timeseries = timeseries[(timeseries["year"] >= from_year) & (timeseries["year"] <= to_year)] |
| start_year = from_year |
| end_year = to_year |
| |
| x = timeseries["year"].values |
| y = timeseries[variable].values |
| |
| |
| if show_trend_line and len(x) > 2: |
| x_norm = x - x.mean() |
| coeffs = np.polyfit(x_norm, y, deg=2) |
| poly = np.poly1d(coeffs) |
| trend_line = poly(x_norm) |
| midpoint_idx = len(x) // 2 |
| slope = 2 * coeffs[0] * x_norm[midpoint_idx] + coeffs[1] |
| _, _, r_value, p_value, _ = linregress(x, y) |
| trend = "Increasing" if p_value < 0.05 and slope > 0 else "Decreasing" if p_value < 0.05 and slope < 0 else "No Significant Trend" |
| else: |
| trend_line = np.full_like(x, np.nan) |
| slope = 0 |
| p_value = 1.0 |
| trend = "No Significant Trend" |
| |
| print(f"Creating timeseries plot: dataset_name={dataset_name}, start_year={start_year}, end_year={end_year}, lat={lat:.2f}, lon={lon:.2f}, region={region}") |
| |
| fig = px.line( |
| timeseries, |
| x="year", |
| y=variable, |
| title=f"{dataset_name}, {start_year}-{end_year} for Lat: {lat:.2f}, Lon: {lon:.2f}", |
| labels={"year": "Year", variable: dataset_name}, |
| color_discrete_sequence=[plotted_line_color], |
| render_mode="svg" |
| ) |
| fig.data[0].name = dataset_name |
| |
| if show_trend_line and len(x) > 2: |
| fig.add_scatter( |
| x=x, |
| y=trend_line, |
| mode="lines", |
| name=f"{dataset_name} Trend", |
| line=dict(color=trend_line_color, dash="dash") |
| ) |
| |
| if show_spatial_mean and spatial_mean_data is not None: |
| spatial_mean_df = spatial_mean_data.get("timeseries", pd.DataFrame()) |
| if not spatial_mean_df.empty: |
| fig.add_scatter( |
| x=spatial_mean_df["year"], |
| y=spatial_mean_df[variable], |
| mode="lines", |
| name=f"{region} Spatial Mean", |
| line=dict(color=spatial_mean_color, width=2) |
| ) |
| |
| fig.add_annotation( |
| x=timeseries["year"].max(), |
| y=timeseries[variable].min(), |
| text=f"Slope: {slope:.4f}<br>p-value: {p_value:.4f}<br>Trend: {trend}", |
| showarrow=False, |
| xanchor="right", |
| yanchor="top", |
| bgcolor="white" if not solara.lab.theme.dark_effective else "#333333", |
| bordercolor="black", |
| borderpad=4, |
| opacity=0.8, |
| font=dict(size=12, color="black" if not solara.lab.theme.dark_effective else "white") |
| ) |
| |
| fig.update_layout( |
| xaxis_title="Year", |
| yaxis_title=dataset_name, |
| height=550, |
| width=650, |
| showlegend=True, |
| legend=dict( |
| x=0.99, |
| y=0.99, |
| xanchor="right", |
| yanchor="top", |
| bgcolor="rgba(0, 0, 0, 0)" |
| ), |
| template=get_plotly_theme() |
| ) |
| |
| dataset.close() |
| return fig |
| |
| except Exception as e: |
| print(f"Error creating timeseries plot: {e}") |
| dataset.close() if 'dataset' in locals() else None |
| return go.Figure() |
|
|
| def get_region_bounds(geodataframe, region): |
| geometry = geodataframe[geodataframe[REGION_COLUMN] == region].geometry.iloc[0] |
| minx, miny, maxx, maxy = geometry.bounds |
| return {"min_lon": minx, "max_lon": maxx, "min_lat": miny, "max_lat": maxy} |
|
|
| def filter_region_data(data, geodataframe, region): |
| if not region or region == "India": |
| return data |
| |
| try: |
| points = gpd.GeoDataFrame( |
| data, geometry=gpd.points_from_xy(data["lon"], data["lat"]), crs=geodataframe.crs |
| ) |
| points.sindex |
| |
| region_geometry = geodataframe[geodataframe[REGION_COLUMN] == region].geometry.iloc[0] |
| |
| bbox = region_geometry.bounds |
| candidates = points[points.geometry.bounds.minx >= bbox[0]] |
| candidates = candidates[candidates.geometry.bounds.maxx <= bbox[2]] |
| candidates = candidates[candidates.geometry.bounds.miny >= bbox[1]] |
| candidates = candidates[candidates.geometry.bounds.maxy <= bbox[3]] |
| |
| filtered_points = candidates[candidates.geometry.within(region_geometry)] |
| |
| return filtered_points.drop(columns=["geometry"]) |
| |
| except Exception as e: |
| print(f"Error filtering region data: {e}") |
| return pd.DataFrame() |
|
|
| def calculate_spatial_mean(dataset_config, geodataframe, region, from_year=None, to_year=None, use_timeframe=False): |
| try: |
| dataset = xr.open_dataset(dataset_config["timeseries_path"], decode_timedelta=False) |
| variable = list(dataset.data_vars)[0] |
| no_data_value = dataset_config["no_data"] |
|
|
| data = dataset[variable].to_dataframe().reset_index() |
| data = data.rename(columns={"latitude": "lat", "longitude": "lon"}) |
|
|
| data = data[data[variable] != no_data_value] |
|
|
| if use_timeframe and from_year is not None and to_year is not None: |
| data = data[(data["year"] >= from_year) & (data["year"] <= to_year)] |
|
|
| if region != "India" and geodataframe is not None: |
| data = filter_region_data(data, geodataframe, region) |
| |
| spatial_means = data.groupby("year")[variable].mean().reset_index() |
| overall_mean = spatial_means[variable].mean() |
| |
| dataset.close() |
| return {"overall_mean": overall_mean, "timeseries": spatial_means} |
| |
| except Exception as e: |
| print(f"Error calculating spatial mean: {e}") |
| return {"overall_mean": None, "timeseries": pd.DataFrame()} |
|
|
| def calculate_selected_spatial_mean(dataset_config, selected_coords, from_year=None, to_year=None, use_timeframe=False): |
| try: |
| if not selected_coords: |
| return {"overall_mean": None, "timeseries": pd.DataFrame()} |
| |
| dataset = xr.open_dataset(dataset_config["timeseries_path"], decode_timedelta=False) |
| variable = list(dataset.data_vars)[0] |
| no_data_value = dataset_config["no_data"] |
|
|
| data_list = [] |
| for lat, lon in selected_coords: |
| lat_idx = np.abs((dataset.get("lat", dataset.get("latitude")).values - lat)).argmin() |
| lon_idx = np.abs((dataset.get("lon", dataset.get("longitude")).values - lon)).argmin() |
| point_data = dataset[variable].isel(lat=lat_idx, lon=lon_idx).to_dataframe().reset_index() |
| data_list.append(point_data) |
| |
| data = pd.concat(data_list, ignore_index=True) |
| data = data[data[variable] != no_data_value] |
|
|
| if use_timeframe and from_year is not None and to_year is not None: |
| data = data[(data["year"] >= from_year) & (data["year"] <= to_year)] |
|
|
| spatial_means = data.groupby("year")[variable].mean().reset_index() |
| overall_mean = spatial_means[variable].mean() |
| |
| dataset.close() |
| return {"overall_mean": overall_mean, "timeseries": spatial_means} |
| |
| except Exception as e: |
| print(f"Error calculating selected spatial mean: {e}") |
| return {"overall_mean": None, "timeseries": pd.DataFrame()} |
|
|
| def create_selected_timeseries_plot(file_path, dataset_name, selected_coords, plotted_line_color, selected_spatial_mean_data=None, from_year=None, to_year=None, use_timeframe=False, show_trend_line=False, trend_line_color="red", spatial_mean_color="yellow"): |
| try: |
| if not selected_coords: |
| return go.Figure() |
| |
| dataset = xr.open_dataset(file_path, decode_timedelta=False) |
| variable = list(dataset.data_vars)[0] |
| no_data_value = next(d for d in DATASETS if d["name"] == dataset_name)["no_data"] |
|
|
| data_list = [] |
| for lat, lon in selected_coords: |
| lat_idx = np.abs((dataset.get("lat", dataset.get("latitude")).values - lat)).argmin() |
| lon_idx = np.abs((dataset.get("lon", dataset.get("longitude")).values - lon)).argmin() |
| point_data = dataset[variable].isel(lat=lat_idx, lon=lon_idx).to_dataframe().reset_index() |
| data_list.append(point_data) |
| |
| data = pd.concat(data_list, ignore_index=True) |
| data = data[data[variable] != no_data_value] |
|
|
| timeseries = data.groupby("year")[variable].mean().reset_index() |
|
|
| years = timeseries["year"].values |
| if len(years) == 0: |
| raise ValueError("No years found in the dataset") |
| start_year = int(min(years)) |
| end_year = int(max(years)) |
|
|
| if use_timeframe and from_year is not None and to_year is not None: |
| timeseries = timeseries[(timeseries["year"] >= from_year) & (timeseries["year"] <= to_year)] |
| start_year = from_year |
| end_year = to_year |
|
|
| x = timeseries["year"].values |
| y = timeseries[variable].values |
| |
| |
| if show_trend_line and len(x) > 2: |
| x_norm = x - x.mean() |
| coeffs = np.polyfit(x_norm, y, deg=2) |
| poly = np.poly1d(coeffs) |
| trend_line = poly(x_norm) |
| midpoint_idx = len(x) // 2 |
| slope = 2 * coeffs[0] * x_norm[midpoint_idx] + coeffs[1] |
| _, _, r_value, p_value, _ = linregress(x, y) |
| trend = "Increasing" if p_value < 0.05 and slope > 0 else "Decreasing" if p_value < 0.05 and slope < 0 else "No Significant Trend" |
| else: |
| trend_line = np.full_like(x, np.nan) |
| slope = 0 |
| p_value = 1.0 |
| trend = "No Significant Trend" |
| |
| fig = px.line( |
| timeseries, |
| x="year", |
| y=variable, |
| title=f"{dataset_name}, Spatial Mean of Selected Points, {start_year}-{end_year}", |
| labels={"year": "Year", variable: dataset_name}, |
| color_discrete_sequence=[plotted_line_color] |
| ) |
| fig.data[0].name = dataset_name |
| |
| if show_trend_line and len(x) > 2: |
| fig.add_scatter( |
| x=x, |
| y=trend_line, |
| mode="lines", |
| name=f"{dataset_name} Trend", |
| line=dict(color=trend_line_color, dash="dash") |
| ) |
| |
| if selected_spatial_mean_data is not None: |
| spatial_mean_df = selected_spatial_mean_data.get("timeseries", pd.DataFrame()) |
| if not spatial_mean_df.empty: |
| fig.add_scatter( |
| x=spatial_mean_df["year"], |
| y=spatial_mean_df[variable], |
| mode="lines", |
| name="Selected Points Spatial Mean", |
| line=dict(color=spatial_mean_color, width=2) |
| ) |
| |
| fig.add_annotation( |
| x=timeseries["year"].max(), |
| y=timeseries[variable].min(), |
| text=f"Slope: {slope:.4f}<br>p-value: {p_value:.4f}<br>Trend: {trend}", |
| showarrow=False, |
| xanchor="right", |
| yanchor="top", |
| bgcolor="white" if not solara.lab.theme.dark_effective else "#333333", |
| bordercolor="black", |
| borderpad=4, |
| opacity=0.8, |
| font=dict(size=12, color="black" if not solara.lab.theme.dark_effective else "white") |
| ) |
| |
| fig.update_layout( |
| xaxis_title="Year", |
| yaxis_title=f"Mean {dataset_name}", |
| height=550, |
| width=650, |
| showlegend=True, |
| legend=dict( |
| x=0.99, |
| y=0.99, |
| xanchor="right", |
| yanchor="top", |
| bgcolor="rgba(0, 0, 0, 0)" |
| ), |
| template=get_plotly_theme() |
| ) |
| |
| dataset.close() |
| return fig |
| |
| except Exception as e: |
| print(f"Error creating selected timeseries plot: {e}") |
| dataset.close() if 'dataset' in locals() else None |
| return go.Figure() |
|
|
| def get_available_years(dataset_config): |
| try: |
| dataset = xr.open_dataset(dataset_config["timeseries_path"], decode_timedelta=False) |
| years = dataset["year"].values.tolist() |
| dataset.close() |
| return sorted(years) |
| except Exception as e: |
| print(f"Error getting available years: {e}") |
| return [] |
