Spaces:
Running
Running
File size: 6,015 Bytes
0d4a0ba | 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 | 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
|