src/folium_map.py - master - Chaizee/ZenithCode_Incident-LLM-analytics

src/folium_map.py
204 строки6 KB
Chaizee
fix: fix speed
12 июн 2026, 16:28
12 июн 2026, 16:28b2745cc
from __future__ import annotations

from typing import Any

import pandas as pd

from src.gis_shapes import (
    aggregate_municipality_stats,
    join_stats_to_gdf,
    load_oblast_districts,
    shapefile_ready,
)

REGION_CENTER = (54.9893, 73.3682)

SEVERITY_HEX: dict[int, str] = {
    0: "#ADE1E5",
    1: "#99D19C",
    2: "#73AB84",
    3: "#79C7C5",
    4: "#5A9E8C",
    5: "#3F5F56",
}

OBLAST_MIN_ZOOM = 6
OBLAST_MAX_ZOOM = 12

MAP_DISPLAY_WIDTH = 520
MAP_DISPLAY_HEIGHT = 680

DISTRICT_FILL_OPACITY = 0.38
DISTRICT_FILL_OPACITY_HOVER = 0.55
SELECTED_BORDER = "#F0FF3D"

SATELLITE_TILES = (
    "https://server.arcgisonline.com/ArcGIS/rest/services/"
    "World_Imagery/MapServer/tile/{z}/{y}/{x}"
)

def mean_severity_color(value: float, has_data: bool) -> str:
    if not has_data or value <= 0:
        return SEVERITY_HEX[0]
    level = max(1, min(5, int(round(value))))
    return SEVERITY_HEX[level]

def _oblast_bounds(gdf) -> tuple[list[float], list[float], list[list[float]]]:
    minx, miny, maxx, maxy = gdf.total_bounds
    pad_lat = (maxy - miny) * 0.015
    pad_lon = (maxx - minx) * 0.015
    sw = [float(miny - pad_lat), float(minx - pad_lon)]
    ne = [float(maxy + pad_lat), float(maxx + pad_lon)]
    return sw, ne, [sw, ne]

def prepare_district_map_data(
    df: pd.DataFrame,
    *,
    category: str | None = None,
) -> pd.DataFrame:
    if not shapefile_ready():
        raise FileNotFoundError("Shapefile не найден. Запустите: python scripts/download_gis.py")

    work = df.copy()
    if "is_problem" in work.columns:
        work = work[work["is_problem"].astype(bool)]
    if category and category != "Все":
        work = work[work["category"].astype(str) == category]

    gdf = load_oblast_districts()
    stats = aggregate_municipality_stats(work)
    merged = join_stats_to_gdf(gdf, stats)

    merged["fill_color"] = merged.apply(
        lambda r: mean_severity_color(float(r["mean_severity"]), int(r["problem_count"]) > 0),
        axis=1,
    )
    return merged

def _merged_to_feature_collection(merged_gdf) -> dict[str, Any]:
    from shapely.geometry import mapping

    prop_cols = (
        "district_key",
        "district_name",
        "excel_municipalities",
        "problem_count",
        "mean_severity_str",
        "critical_count",
        "fill_color",
    )
    features: list[dict[str, Any]] = []
    for _, row in merged_gdf.iterrows():
        geom = row.geometry
        if geom is None or geom.is_empty:
            continue
        if geom.geom_type not in ("Polygon", "MultiPolygon"):
            continue
        props: dict[str, Any] = {}
        for col in prop_cols:
            if col not in merged_gdf.columns:
                continue
            val = row[col]
            if col in ("problem_count", "critical_count"):
                props[col] = int(val or 0)
            else:
                props[col] = str(val if val is not None else "")
        feat_id = str(props.get("district_key") or props.get("district_name") or len(features))
        features.append(
            {
                "type": "Feature",
                "id": feat_id,
                "properties": props,
                "geometry": mapping(geom),
            }
        )
    return {"type": "FeatureCollection", "features": features}

def build_district_folium_map(
    merged_gdf,
    *,
    selected_district_key: str = "",
) -> Any:
    import folium

    bounds_sw, bounds_ne, max_bounds = _oblast_bounds(merged_gdf)
    selected = str(selected_district_key or "").strip()

    def style_function(feature: dict) -> dict[str, Any]:
        props = feature.get("properties") or {}
        base = {
            "fillColor": props.get("fill_color", SEVERITY_HEX[0]),
            "color": "#FFFFFF",
            "weight": 1.4,
            "fillOpacity": DISTRICT_FILL_OPACITY,
        }
        if selected and str(props.get("district_key", "")) == selected:
            base["color"] = SELECTED_BORDER
            base["weight"] = 3.4
            base["fillOpacity"] = DISTRICT_FILL_OPACITY_HOVER
        return base

    def highlight_function(feature: dict) -> dict[str, Any]:
        props = feature.get("properties") or {}
        return {
            "fillColor": props.get("fill_color", SEVERITY_HEX[0]),
            "color": SELECTED_BORDER,
            "weight": 3.2,
            "fillOpacity": DISTRICT_FILL_OPACITY_HOVER,
        }

    m = folium.Map(
        location=list(REGION_CENTER),
        zoom_start=OBLAST_MIN_ZOOM,
        min_zoom=OBLAST_MIN_ZOOM,
        max_zoom=OBLAST_MAX_ZOOM,
        max_bounds=max_bounds,
        max_bounds_viscosity=1.0,
        tiles=SATELLITE_TILES,
        attr="Esri, Maxar, Earthstar",
        control_scale=False,
        attributionControl=False,
        zoom_control=True,
        scrollWheelZoom=True,
    )

    folium.GeoJson(
        _merged_to_feature_collection(merged_gdf),
        name="Районы",
        style_function=style_function,
        highlight_function=highlight_function,
    ).add_to(m)

    m.fit_bounds([bounds_sw, bounds_ne], padding=(8, 8), max_zoom=OBLAST_MIN_ZOOM)

    map_name = m.get_name()
    m.get_root().script.add_child(
        folium.Element(
            f"""
<script>
(function() {{
var map = {map_name};
map.whenReady(function() {{
map.fitBounds({max_bounds}, {{padding: [8, 8], maxZoom: {OBLAST_MIN_ZOOM}}});
var lockZoom = map.getZoom();
map.setMinZoom(lockZoom);
map.setMaxBounds({max_bounds});
}});
}})();
</script>
"""
        )
    )

    return m

def render_map_legend(region_name: str = "") -> None:
    import streamlit as st

    title = region_name or "Омская область"
    c0, c1, c2, c3, c4 = st.columns([1.2, 1, 1, 1, 1])
    c0.caption(f"**{title}**")
    c1.markdown(f'<span style="color:{SEVERITY_HEX[1]}">■</span> низкая', unsafe_allow_html=True)
    c2.markdown(f'<span style="color:{SEVERITY_HEX[3]}">■</span> средняя', unsafe_allow_html=True)
    c3.markdown(f'<span style="color:{SEVERITY_HEX[5]}">■</span> высокая / ЧП', unsafe_allow_html=True)
    c4.markdown(f'<span style="color:{SEVERITY_HEX[0]}">■</span> нет данных', unsafe_allow_html=True)