"""
Map creation utilities for the Sri Lanka Monitoring Dashboard.
Handles Folium choropleth maps, color schemes, and legends.
"""
import folium
import streamlit as st
# ============================================================
# METRIC CONFIGURATION
# ============================================================
# Consistent color palette for all sitrep metrics (red gradient)
SITREP_COLORS = ['#f7f7f7', '#fef0d9', '#fdcc8a', '#fc8d59', '#e34a33', '#b30000', '#7f0000']
METRIC_CONFIG = {
"people_affected": {
"label": "People Affected",
"thresholds": [1, 10000, 50000, 100000, 200000, 350000],
"format": lambda x: f"{x:,}"
},
"people_displaced": {
"label": "People Displaced",
"thresholds": [1, 500, 2000, 5000, 10000, 20000],
"format": lambda x: f"{x:,}"
},
"deaths": {
"label": "Deaths",
"thresholds": [1, 10, 25, 50, 75, 100],
"format": lambda x: str(x)
},
"missing": {
"label": "Missing Persons",
"thresholds": [1, 5, 10, 20, 30, 50],
"format": lambda x: str(x)
},
"houses_fully_damaged": {
"label": "Houses Fully Damaged",
"thresholds": [1, 50, 100, 250, 400, 500],
"format": lambda x: f"{x:,}"
},
"houses_partially_damaged": {
"label": "Houses Partially Damaged",
"thresholds": [1, 500, 2000, 5000, 7500, 10000],
"format": lambda x: f"{x:,}"
}
}
# ============================================================
# COLOR FUNCTIONS - SITREP
# ============================================================
def get_color_for_metric(value: int, metric: str) -> str:
"""Get color for a value based on the metric's thresholds using consistent palette."""
config = METRIC_CONFIG.get(metric, METRIC_CONFIG["people_affected"])
thresholds = config["thresholds"]
if value == 0:
return SITREP_COLORS[0] # Light gray for 0
for i, threshold in enumerate(thresholds):
if value < threshold:
return SITREP_COLORS[i]
return SITREP_COLORS[-1] # Highest color for values above all thresholds
def get_color(value: int, metric: str) -> str:
"""Alias for get_color_for_metric for backwards compatibility."""
return get_color_for_metric(value, metric)
def get_legend_html(metric: str) -> str:
"""Generate HTML legend for the selected metric with consistent colors."""
config = METRIC_CONFIG.get(metric, METRIC_CONFIG["people_affected"])
thresholds = config["thresholds"]
colors = SITREP_COLORS
def fmt(n):
"""Format number with K/M suffix for readability."""
if n >= 1000000:
return f"{n/1000000:.1f}M"
elif n >= 1000:
return f"{n/1000:.0f}K"
return str(n)
legend_items = [
f"0",
f"1-{fmt(thresholds[1]-1)}",
f"{fmt(thresholds[1])}-{fmt(thresholds[2]-1)}",
f"{fmt(thresholds[2])}-{fmt(thresholds[3]-1)}",
f"{fmt(thresholds[3])}-{fmt(thresholds[4]-1)}",
f"{fmt(thresholds[4])}-{fmt(thresholds[5]-1)}",
f"{fmt(thresholds[5])}+",
]
return " ".join(legend_items)
# ============================================================
# COLOR FUNCTIONS - FLOOD
# ============================================================
FLOOD_ALERT_COLORS = {
"Normal": "#28a745", # Green
"Alert": "#ffc107", # Yellow
"Minor Flood": "#fd7e14", # Orange
"Major Flood": "#dc3545", # Red
"default": "#6c757d" # Gray for unknown
}
TREND_ICONS = {
"Rising": "⬆️",
"Falling": "⬇️",
"Stable": "➡️",
"default": "—"
}
def get_flood_alert_color(remarks: str | None) -> str:
"""Get color for flood alert status based on remarks field."""
if not remarks or remarks == "-" or remarks == "Normal":
return FLOOD_ALERT_COLORS["Normal"]
return FLOOD_ALERT_COLORS.get(remarks, FLOOD_ALERT_COLORS["default"])
def get_trend_icon(trend: str | None) -> str:
"""Get icon for water level trend."""
if not trend:
return TREND_ICONS["default"]
return TREND_ICONS.get(trend, TREND_ICONS["default"])
def create_flood_tooltip(station_data: dict) -> str:
"""Create HTML tooltip for flood monitoring station."""
tributary = station_data.get("tributary", "Unknown")
station = station_data.get("station", "Unknown")
unit = station_data.get("unit", "m")
alert_level = station_data.get("alert_level", "N/A")
minor_flood = station_data.get("minor_flood_level", "N/A")
major_flood = station_data.get("major_flood_level", "N/A")
reading_1 = station_data.get("water_level_reading_1", "N/A")
reading_2 = station_data.get("water_level_reading_2", "N/A")
remarks = station_data.get("remarks") or "Normal"
trend = station_data.get("water_level_trend")
rainfall = station_data.get("rainfall_6hr_mm")
trend_icon = get_trend_icon(trend)
alert_color = get_flood_alert_color(remarks)
# Format water level display
if reading_2 not in [None, "N/A", "-"]:
water_level_display = f"{reading_2} {unit}"
if reading_1 not in [None, "N/A", "-"]:
water_level_display += f" (prev: {reading_1} {unit})"
elif reading_1 not in [None, "N/A", "-"]:
water_level_display = f"{reading_1} {unit}"
else:
water_level_display = "No data"
tooltip_html = f"""
{tributary}
{station}
{remarks}
Water Level: {water_level_display} {trend_icon}
Alert Level: {alert_level} {unit}
Minor Flood: {minor_flood} {unit}
Major Flood: {major_flood} {unit}
"""
if rainfall not in [None, "N/A", "-"]:
tooltip_html += f"
6h Rainfall: {rainfall} mm
"
tooltip_html += "
"
return tooltip_html
# COLOR FUNCTIONS - LANDSLIDE
# ============================================================
LANDSLIDE_COLORS = {
0: "#f0f0f0", # No warning - light gray
1: "#ffeb3b", # Level 1 - Yellow
2: "#ff9800", # Level 2 - Amber/Orange
3: "#f44336", # Level 3 - Red
}
def get_landslide_color(level: int) -> str:
"""Get color for landslide warning level."""
return LANDSLIDE_COLORS.get(level, LANDSLIDE_COLORS[0])
def get_landslide_legend_html() -> str:
"""Generate legend HTML for landslide warning levels."""
return (
'No Warning '
'🟡 Level 1 '
'🟠 Level 2 '
'🔴 Level 3'
)
# ============================================================
# MAP CREATION FUNCTIONS
# ============================================================
def create_empty_map(title: str = "Map") -> folium.Map:
"""Create an empty map centered on Sri Lanka."""
m = folium.Map(
location=[7.8731, 80.7718],
zoom_start=7,
tiles="cartodbpositron"
)
folium.map.Marker(
[7.8731, 80.7718],
icon=folium.DivIcon(
html=f'{title}
No data loaded
'
)
).add_to(m)
return m
def create_choropleth_map(
sitrep_data: dict | None,
districts_geojson: dict,
metric: str = "people_affected"
) -> folium.Map:
"""Create a choropleth map for situation report data."""
m = folium.Map(
location=[7.8731, 80.7718],
zoom_start=7,
tiles="cartodbpositron"
)
if not sitrep_data:
return create_empty_map("Situation Report")
# Build lookup from sitrep data
district_data = {d["district"]: d for d in sitrep_data.get("districts", [])}
config = METRIC_CONFIG.get(metric, METRIC_CONFIG["people_affected"])
format_fn = config["format"]
for feature in districts_geojson.get("features", []):
district_name = feature["properties"].get("district", "Unknown")
data = district_data.get(district_name, {})
value = data.get(metric, 0) if data else 0
color = get_color_for_metric(value, metric)
# Build tooltip
tooltip_html = f"""
{district_name}
{config['label']}: {format_fn(value)}
People Affected: {data.get('people_affected', 0):,}
People Displaced: {data.get('people_displaced', 0):,}
Deaths: {data.get('deaths', 0)}
Missing: {data.get('missing', 0)}
Houses Fully Damaged: {data.get('houses_fully_damaged', 0):,}
Houses Partially Damaged: {data.get('houses_partially_damaged', 0):,}
"""
folium.GeoJson(
feature,
style_function=lambda x, c=color: {
"fillColor": c,
"color": "#333",
"weight": 1,
"fillOpacity": 0.7
},
tooltip=folium.Tooltip(tooltip_html, sticky=True),
).add_to(m)
return m
def create_landslide_choropleth_map(
landslide_data: dict | None,
districts_geojson: dict,
divisions_geojson: dict | None = None,
landslide_observations: dict | None = None
) -> folium.Map:
"""
Create a choropleth map for landslide warning data.
If divisions_geojson is provided, the map will show division-level coloring
with district boundaries overlaid as a thicker stroke.
Otherwise, falls back to district-level coloring.
If landslide_observations is provided, adds point markers for observed landslides.
"""
m = folium.Map(
location=[7.8731, 80.7718],
zoom_start=7,
tiles="cartodbpositron"
)
if not landslide_data:
return create_empty_map("Landslide Warnings")
# Build division-level lookup from landslide data
from src.landslide_extractor import build_division_lookup, normalize_division_name
division_lookup = build_division_lookup(landslide_data)
# If we have divisions GeoJSON, render divisions with district overlay
if divisions_geojson:
# First, add division polygons (colored by warning level)
for feature in divisions_geojson.get("features", []):
division_name = feature["properties"].get("shapeName", "Unknown")
# Try to match division name with our lookup
norm_name = normalize_division_name(division_name)
# Try exact match first, then partial matching
matched_data = None
if norm_name in division_lookup:
matched_data = division_lookup[norm_name]
else:
# Try partial match - check if any lookup key is contained in norm_name or vice versa
for lookup_key, data in division_lookup.items():
lookup_norm = normalize_division_name(lookup_key)
if lookup_norm in norm_name or norm_name in lookup_norm:
matched_data = data
break
if matched_data:
level = matched_data["level"]
district = matched_data["district"]
else:
level = 0
district = "Unknown"
color = get_landslide_color(level)
level_name = {0: "No Warning", 1: "Yellow", 2: "Amber", 3: "Red"}.get(level, "Unknown")
level_emoji = {0: "⚪", 1: "🟡", 2: "🟠", 3: "🔴"}.get(level, "⚪")
tooltip_html = f"""
{division_name}
District: {district}
{level_emoji} Alert Level: {level_name}
"""
folium.GeoJson(
feature,
style_function=lambda x, c=color: {
"fillColor": c,
"color": "#666",
"weight": 0.5,
"fillOpacity": 0.7
},
tooltip=folium.Tooltip(tooltip_html, sticky=True),
).add_to(m)
# Then, add district boundaries as overlay (thicker stroke, no fill, non-interactive)
for feature in districts_geojson.get("features", []):
folium.GeoJson(
feature,
style_function=lambda x: {
"fillColor": "transparent",
"fillOpacity": 0,
"color": "#333",
"weight": 2.5,
"interactive": False,
},
interactive=False,
).add_to(m)
else:
# Fallback: district-level coloring (original behavior)
district_data = {d["district"]: d for d in landslide_data.get("districts", [])}
for feature in districts_geojson.get("features", []):
district_name = feature["properties"].get("district", "Unknown")
data = district_data.get(district_name, {})
max_level = data.get("max_level", 0) if data else 0
color = get_landslide_color(max_level)
# Build tooltip with division details
level_3_divs = data.get("level_3_divisions", [])
level_2_divs = data.get("level_2_divisions", [])
level_1_divs = data.get("level_1_divisions", [])
total_warnings = data.get("total_warnings", 0)
level_name = {0: "No Warning", 1: "Yellow", 2: "Amber", 3: "Red"}.get(max_level, "Unknown")
level_emoji = {0: "⚪", 1: "🟡", 2: "🟠", 3: "🔴"}.get(max_level, "⚪")
# Format division lists with truncation
def format_divisions(divs: list, max_show: int = 3) -> str:
if not divs:
return "None"
if len(divs) <= max_show:
return ", ".join(divs)
return ", ".join(divs[:max_show]) + f" (+{len(divs) - max_show} more)"
tooltip_html = f"""
{district_name}
{level_emoji} Alert Level: {level_name}
Total Divisions with Warnings: {total_warnings}
🔴 Level 3 (Red): {format_divisions(level_3_divs)}
🟠 Level 2 (Amber): {format_divisions(level_2_divs)}
🟡 Level 1 (Yellow): {format_divisions(level_1_divs)}
"""
folium.GeoJson(
feature,
style_function=lambda x, c=color: {
"fillColor": c,
"color": "#333",
"weight": 1,
"fillOpacity": 0.7
},
tooltip=folium.Tooltip(tooltip_html, sticky=True),
).add_to(m)
# Add landslide observations as point markers if available
if landslide_observations:
for feature in landslide_observations.get("features", []):
props = feature.get("properties", {})
geom = feature.get("geometry", {})
if geom.get("type") != "Point":
continue
coords = geom.get("coordinates", [])
if len(coords) < 2:
continue
lon, lat = coords[0], coords[1]
obs_type = props.get("Type", "Unknown")
obs_name = props.get("Name", "").strip()
# Determine marker color and icon based on observation type
if "Landslide observed" in obs_type:
marker_color = "red"
icon_name = "exclamation-triangle"
obs_status = "⚠️ Landslide Observed"
elif "No landslides observed" in obs_type:
marker_color = "green"
icon_name = "check"
obs_status = "✅ No Landslide Observed"
elif "cloud cover" in obs_type.lower():
marker_color = "gray"
icon_name = "cloud"
obs_status = "☁️ N/A (Cloud Cover)"
else:
marker_color = "blue"
icon_name = "info"
obs_status = obs_type
# Find which division this point is in (using division lookup for context)
# For now, just show the observation info
tooltip_html = f"""
📍 Landslide Observation
Status: {obs_status}
Source: {obs_name if obs_name else 'LHMP-LRRMD'}
Location: {lat:.4f}°N, {lon:.4f}°E
"""
folium.Marker(
location=[lat, lon],
tooltip=folium.Tooltip(tooltip_html, sticky=True),
icon=folium.Icon(color=marker_color, icon=icon_name, prefix='fa')
).add_to(m)
return m
def create_flood_map(
flood_data: dict | None,
rivers_geojson: dict,
water_bodies_geojson: dict | None = None
) -> folium.Map:
"""Create a map for flood warnings showing monitored rivers.
Uses the standard_name property from monitored_rivers.geojson to match with flood data.
"""
m = folium.Map(
location=[7.8731, 80.7718],
zoom_start=7,
tiles="cartodbpositron"
)
# Build station lookup from flood data using tributary names
# For rivers with multiple stations, prioritize the highest alert level
station_lookup = {}
if flood_data:
for basin_name, stations in flood_data.get("river_basins", {}).items():
for station in stations:
tributary = station.get("tributary", "").strip()
if tributary:
tributary_lower = tributary.lower()
# Check if we already have a station for this tributary
existing = station_lookup.get(tributary_lower)
if existing:
# Compare alert levels - keep the one with higher alert
existing_remarks = existing.get("remarks") or "Normal"
new_remarks = station.get("remarks") or "Normal"
# Alert priority: Major Flood > Minor Flood > Alert > Normal
priority = {"Major Flood": 4, "Minor Flood": 3, "Alert": 2, "Normal": 1}
existing_priority = priority.get(existing_remarks, 0)
new_priority = priority.get(new_remarks, 0)
# Only replace if new station has higher priority
if new_priority > existing_priority:
station_lookup[tributary_lower] = station
else:
station_lookup[tributary_lower] = station
# Add rivers layer - all rivers from monitored_rivers.geojson
for feature in rivers_geojson.get("features", []):
# Use standard_name for matching with scraped data
standard_name = feature["properties"].get("standard_name")
river_name = feature["properties"].get("name")
if not standard_name:
continue
# Look up station data using standard_name
station_data = station_lookup.get(standard_name.lower())
if station_data:
# River has monitoring data - color by alert status
remarks = station_data.get("remarks") or "Normal"
color = get_flood_alert_color(remarks)
weight = 3 if remarks not in ["Normal", None] else 2
opacity = 0.9 if remarks not in ["Normal", None] else 0.6
tooltip_html = create_flood_tooltip(station_data)
else:
# Show river without data in default color
color = FLOOD_ALERT_COLORS["Normal"]
weight = 2
opacity = 0.6
tooltip_html = f"{river_name or standard_name}
No current monitoring data"
folium.GeoJson(
feature,
style_function=lambda x, c=color, w=weight, o=opacity: {
"color": c,
"weight": w,
"opacity": o
},
tooltip=folium.Tooltip(tooltip_html, sticky=True),
).add_to(m)
# Add legend only when flood data is available
if flood_data:
legend_html = f"""
Flood Alert Status
Normal
Alert
Minor Flood
Major Flood
⬆️ Rising | ⬇️ Falling | ➡️ Stable
"""
m.get_root().html.add_child(folium.Element(legend_html))
return m