Spaces:
Running
Running
| import geopandas as gpd | |
| import folium | |
| import shapely | |
| from src.utils import MAP_CENTER, MAP_ZOOM, UNIT_NAME_COL | |
| TILE = "CartoDB positron" | |
| def _clean_geometries(gdf: gpd.GeoDataFrame) -> gpd.GeoDataFrame: | |
| """Drop null/empty geometries and repair invalid ones before any spatial operation.""" | |
| gdf = gdf[~gdf["geometry"].isna()].copy() | |
| gdf = gdf[~gdf["geometry"].is_empty].copy() | |
| gdf["geometry"] = shapely.make_valid(gdf["geometry"]) | |
| # Drop any geometry that still has non-finite bounds | |
| def _has_finite_bounds(geom): | |
| try: | |
| b = geom.bounds | |
| return all(v == v and abs(v) != float("inf") for v in b) # NaN check + inf check | |
| except Exception: | |
| return False | |
| gdf = gdf[gdf["geometry"].apply(_has_finite_bounds)].copy() | |
| return gdf | |
| def _format_popup(row: dict) -> str: | |
| fields = [ | |
| ("Art", row.get("scientificName") or row.get("species")), | |
| ("Datum", row.get("eventDate") or f"{row.get('year','')}-{str(row.get('month','')).zfill(2)}-{str(row.get('day','')).zfill(2)}"), | |
| ("Datenquelle", row.get("datasetName")), | |
| ("Institution", row.get("institutionCode")), | |
| ("Erfasser", row.get("recordedBy")), | |
| ("Ort", row.get("locality")), | |
| ("Bundesland/Region", row.get("stateProvince")), | |
| ("Land", row.get("country")), | |
| ("GBIF-ID", row.get("gbifID")), | |
| ("Koordinatengenauigkeit", row.get("coordinateUncertaintyInMeters")), | |
| ] | |
| rows_html = "".join( | |
| f"<tr><td style='font-weight:bold;padding-right:8px'>{k}</td><td>{v}</td></tr>" | |
| for k, v in fields | |
| if v and str(v).strip() and str(v) not in ("nan", "None", "--") | |
| ) | |
| return f"<table style='font-size:12px;min-width:220px'>{rows_html}</table>" | |
| def build_current_map(occurrences_gdf: gpd.GeoDataFrame, scope: str, | |
| boundary_gdf: gpd.GeoDataFrame | None = None, | |
| outline_gdf: gpd.GeoDataFrame | None = None) -> folium.Map: | |
| center = MAP_CENTER.get(scope, (51.2, 10.4)) | |
| zoom = MAP_ZOOM.get(scope, 6) | |
| m = folium.Map(location=center, zoom_start=zoom, tiles=TILE) | |
| if outline_gdf is not None and not outline_gdf.empty: | |
| simplified_outline = _clean_geometries(outline_gdf.copy()) | |
| simplified_outline["geometry"] = simplified_outline["geometry"].simplify(tolerance=0.005, preserve_topology=True) | |
| folium.GeoJson( | |
| simplified_outline.__geo_interface__, | |
| style_function=lambda _: {"color": "#1a1a1a", "weight": 2.5, "fillOpacity": 0}, | |
| name="Übergeordnete Grenze", | |
| ).add_to(m) | |
| if boundary_gdf is not None and not boundary_gdf.empty: | |
| simplified = _clean_geometries(boundary_gdf.copy()) | |
| simplified["geometry"] = simplified["geometry"].simplify(tolerance=0.005, preserve_topology=True) | |
| folium.GeoJson( | |
| simplified.__geo_interface__, | |
| style_function=lambda _: {"color": "#444", "weight": 0.8, "fillOpacity": 0}, | |
| name="Grenzen", | |
| ).add_to(m) | |
| points_layer = folium.FeatureGroup(name="Fundpunkte") | |
| for _, row in occurrences_gdf.iterrows(): | |
| lat = row.get("decimalLatitude") | |
| lon = row.get("decimalLongitude") | |
| if lat is None or lon is None: | |
| continue | |
| popup_html = _format_popup(row.to_dict()) | |
| folium.CircleMarker( | |
| location=[lat, lon], | |
| radius=5, | |
| color="#c0392b", | |
| fill=True, | |
| fill_color="#e74c3c", | |
| fill_opacity=0.7, | |
| popup=folium.Popup(popup_html, max_width=320), | |
| tooltip=row.get("scientificName", ""), | |
| ).add_to(points_layer) | |
| points_layer.add_to(m) | |
| folium.LayerControl().add_to(m) | |
| return m | |
| def build_choropleth_map(occurrences_gdf: gpd.GeoDataFrame, boundary_gdf: gpd.GeoDataFrame, | |
| scope: str, period_label: str) -> folium.Map: | |
| center = MAP_CENTER.get(scope, (51.2, 10.4)) | |
| zoom = MAP_ZOOM.get(scope, 6) | |
| m = folium.Map(location=center, zoom_start=zoom, tiles=TILE) | |
| unit_col = UNIT_NAME_COL.get(scope, "GEN") | |
| if unit_col not in boundary_gdf.columns: | |
| return m | |
| # Count occurrences per unit | |
| counts = occurrences_gdf.groupby(unit_col).size().reset_index(name="count") \ | |
| if unit_col in occurrences_gdf.columns else None | |
| # Simplify for performance (clean invalid geometries first) | |
| boundary_plot = _clean_geometries(boundary_gdf.copy()) | |
| boundary_plot["geometry"] = boundary_plot["geometry"].simplify(tolerance=0.005, preserve_topology=True) | |
| if counts is not None and not counts.empty: | |
| boundary_plot = boundary_plot.merge(counts, on=unit_col, how="left") | |
| boundary_plot["count"] = boundary_plot["count"].fillna(0).astype(int) | |
| folium.Choropleth( | |
| geo_data=boundary_plot.__geo_interface__, | |
| data=boundary_plot[[unit_col, "count"]], | |
| columns=[unit_col, "count"], | |
| key_on=f"feature.properties.{unit_col}", | |
| fill_color="YlOrRd", | |
| fill_opacity=0.75, | |
| line_opacity=0.4, | |
| legend_name=f"Funde {period_label}", | |
| nan_fill_color="#eeeeee", | |
| name=f"Funde {period_label}", | |
| ).add_to(m) | |
| # Tooltip overlay | |
| tooltip_gdf = boundary_plot[[unit_col, "count", "geometry"]].copy() | |
| folium.GeoJson( | |
| tooltip_gdf.__geo_interface__, | |
| style_function=lambda _: {"color": "#555", "weight": 0.5, "fillOpacity": 0}, | |
| tooltip=folium.GeoJsonTooltip( | |
| fields=[unit_col, "count"], | |
| aliases=["Einheit:", "Funde:"], | |
| localize=True, | |
| ), | |
| name="Info", | |
| ).add_to(m) | |
| else: | |
| folium.GeoJson( | |
| boundary_plot.__geo_interface__, | |
| style_function=lambda _: {"color": "#888", "weight": 0.8, "fillColor": "#eeeeee", "fillOpacity": 0.4}, | |
| ).add_to(m) | |
| folium.LayerControl().add_to(m) | |
| return m | |