rainViewerMap / app.py
nakas's picture
ok7
8356f47
import gradio as gr
import folium
import requests
import json
from datetime import datetime
import time
import urllib.parse
class RainViewerMap:
def __init__(self):
self.api_url = "https://api.rainviewer.com/public/weather-maps.json"
self.tile_host = "https://tilecache.rainviewer.com"
self.nhc_api_base = "https://mapservices.weather.noaa.gov/tropical/rest/services/tropical/NHC_tropical_weather/MapServer"
self.storm_cache = {}
def get_radar_data(self):
"""Fetch available radar timestamps from RainViewer API"""
try:
response = requests.get(self.api_url)
data = response.json()
return data
except Exception as e:
print(f"Error fetching radar data: {e}")
return None
def get_hurricane_data(self):
"""Fetch active hurricane/tropical storm data from NOAA NHC"""
active_storms = []
error_messages = []
try:
# Get specific Eastern Pacific storms (EP1-EP5) and Atlantic storms (AT1-AT5)
storm_layers = [
{'name': 'EP1', 'forecast_points_id': 136, 'forecast_track_id': 137, 'forecast_cone_id': 138},
{'name': 'EP2', 'forecast_points_id': 154, 'forecast_track_id': 155, 'forecast_cone_id': 156},
{'name': 'AT1', 'forecast_points_id': 6, 'forecast_track_id': 7, 'forecast_cone_id': 8},
{'name': 'AT2', 'forecast_points_id': 24, 'forecast_track_id': 25, 'forecast_cone_id': 26},
{'name': 'AT3', 'forecast_points_id': 42, 'forecast_track_id': 43, 'forecast_cone_id': 44}
]
for storm_info in storm_layers:
storm_name = storm_info['name']
points_id = storm_info['forecast_points_id']
track_id = storm_info['forecast_track_id']
cone_id = storm_info['forecast_cone_id']
try:
# Get forecast points
points_url = f"{self.nhc_api_base}/{points_id}/query"
points_params = {
'where': '1=1',
'outFields': '*',
'f': 'geojson',
'returnGeometry': 'true'
}
points_response = requests.get(points_url, params=points_params, timeout=15)
if points_response.status_code == 200:
points_data = points_response.json()
if points_data.get('features'):
# Get track data
track_url = f"{self.nhc_api_base}/{track_id}/query"
track_response = requests.get(track_url, params=points_params, timeout=15)
track_data = None
if track_response.status_code == 200:
track_data = track_response.json()
# Get cone data
cone_url = f"{self.nhc_api_base}/{cone_id}/query"
cone_response = requests.get(cone_url, params=points_params, timeout=15)
cone_data = None
if cone_response.status_code == 200:
cone_data = cone_response.json()
active_storms.append({
'source': f'NOAA NHC {storm_name}',
'storm_id': storm_name,
'forecast_points': points_data,
'forecast_track': track_data,
'forecast_cone': cone_data,
'type': 'nhc_storm'
})
# Get storm name from first feature for logging
first_feature = points_data['features'][0]
storm_actual_name = first_feature['properties'].get('stormname', storm_name)
error_messages.append(f"Successfully loaded {storm_actual_name} ({storm_name})")
except Exception as e:
error_messages.append(f"Error fetching {storm_name}: {e}")
# If no real storms found, add informational message
if not active_storms:
error_messages.append("No active storms found - this could mean no storms are currently active")
# Store debug info
self.storm_cache['errors'] = error_messages
self.storm_cache['last_update'] = datetime.now().isoformat()
return active_storms
except Exception as e:
error_messages.append(f"General error fetching hurricane data: {e}")
self.storm_cache['errors'] = error_messages
print(f"Hurricane data fetch error: {e}")
return []
def get_storm_track_data(self, layer_id):
"""Fetch track data for a specific storm layer"""
try:
# Look for corresponding track layer (usually layer_id - 1 or nearby)
track_layer_id = layer_id - 1 if layer_id > 0 else layer_id + 1
query_url = f"{self.nhc_api_base}/{track_layer_id}/query"
params = {
'where': '1=1',
'outFields': '*',
'f': 'geojson'
}
response = requests.get(query_url, params=params)
if response.status_code == 200:
return response.json()
return None
except Exception as e:
print(f"Error fetching track data: {e}")
return None
def create_map(self, lat=40.7128, lon=-74.0060, zoom=8, show_radar=True, time_index=0, show_hurricanes=True):
"""Create a Folium map with optional radar overlay and hurricane tracking"""
# Create base map
m = folium.Map(
location=[lat, lon],
zoom_start=zoom,
tiles='OpenStreetMap'
)
if show_radar:
radar_data = self.get_radar_data()
if radar_data and 'radar' in radar_data:
# Get available timestamps
past_data = radar_data['radar']['past']
nowcast_data = radar_data['radar']['nowcast']
all_times = past_data + nowcast_data
if all_times and time_index < len(all_times):
# Get the radar tile path for selected time
selected_time = all_times[time_index]
tile_path = selected_time['path']
timestamp = selected_time['time']
# Create tile URL template
tile_url = f"{self.tile_host}{tile_path}/512/{{z}}/{{x}}/{{y}}/2/1_1.png"
# Add radar overlay
folium.raster_layers.TileLayer(
tiles=tile_url,
attr='RainViewer',
name='Radar',
overlay=True,
control=True,
opacity=0.6
).add_to(m)
# Add timestamp info
dt = datetime.fromtimestamp(timestamp)
folium.Marker(
[lat + 0.1, lon + 0.1],
popup=f"Radar Time: {dt.strftime('%Y-%m-%d %H:%M:%S')}",
icon=folium.Icon(color='blue', icon='info-sign')
).add_to(m)
# Add hurricane tracking
if show_hurricanes:
self.add_hurricane_layers(m)
# Add layer control
folium.LayerControl().add_to(m)
return m._repr_html_()
def add_hurricane_layers(self, folium_map):
"""Add hurricane tracking layers to the map"""
hurricane_data = self.get_hurricane_data()
for storm_info in hurricane_data:
storm_type = storm_info.get('type', 'unknown')
if storm_type == 'nhc_storm':
# Handle real NHC storm data
storm_id = storm_info.get('storm_id', 'Unknown')
forecast_points = storm_info.get('forecast_points', {})
forecast_track = storm_info.get('forecast_track', {})
forecast_cone = storm_info.get('forecast_cone', {})
# Add forecast track line
if forecast_track and forecast_track.get('features'):
for track_feature in forecast_track['features']:
if track_feature['geometry']['type'] == 'LineString':
coords = track_feature['geometry']['coordinates']
# Convert from [lon, lat] to [lat, lon] for Folium
track_coords = [[coord[1], coord[0]] for coord in coords]
track_props = track_feature.get('properties', {})
storm_name = track_props.get('stormname', storm_id)
folium.PolyLine(
locations=track_coords,
color='red',
weight=4,
opacity=0.8,
popup=f"Forecast Track: {storm_name}"
).add_to(folium_map)
# Add forecast cone if available
if forecast_cone and forecast_cone.get('features'):
for cone_feature in forecast_cone['features']:
if cone_feature['geometry']['type'] == 'Polygon':
coords = cone_feature['geometry']['coordinates'][0] # Outer ring
# Convert from [lon, lat] to [lat, lon] for Folium
cone_coords = [[coord[1], coord[0]] for coord in coords]
cone_props = cone_feature.get('properties', {})
storm_name = cone_props.get('stormname', storm_id)
folium.Polygon(
locations=cone_coords,
color='orange',
weight=2,
opacity=0.6,
fillColor='yellow',
fillOpacity=0.2,
popup=f"Forecast Cone: {storm_name}"
).add_to(folium_map)
# Add forecast points with spinning hurricane markers
if forecast_points and forecast_points.get('features'):
for feature in forecast_points['features']:
if feature['geometry']['type'] == 'Point':
coords = feature['geometry']['coordinates']
props = feature.get('properties', {})
# Extract detailed storm information
storm_name = props.get('stormname', 'Unknown Storm')
storm_type_detail = props.get('stormtype', 'Unknown')
max_wind = props.get('maxwind', 'N/A')
pressure = props.get('mslp', 'N/A')
gust = props.get('gust', 'N/A')
tau = props.get('tau', 0) # Forecast hour
date_label = props.get('datelbl', 'Unknown Time')
ss_num = props.get('ssnum', 0) # Saffir-Simpson number
advisory_num = props.get('advisnum', 'N/A')
advisory_date = props.get('advdate', 'N/A')
# Determine marker properties based on storm intensity
color = self.get_storm_color(max_wind)
radius = 8 + (ss_num * 2) # Larger markers for stronger storms
# Create detailed popup
popup_text = f"""
<div style='font-family: Arial; font-size: 12px; line-height: 1.4; min-width: 250px;'>
<div style='text-align: center; background: {color}; color: white; padding: 8px; margin: -8px -8px 8px -8px; border-radius: 5px 5px 0 0;'>
<span style='font-size: 20px;'>🌀</span>
<strong style='font-size: 16px;'>{storm_name}</strong>
</div>
<table style='width: 100%; border-collapse: collapse;'>
<tr><td style='padding: 3px; font-weight: bold;'>Type:</td><td style='padding: 3px;'>{storm_type_detail}</td></tr>
<tr><td style='padding: 3px; font-weight: bold;'>Max Winds:</td><td style='padding: 3px; color: {color}; font-weight: bold;'>{max_wind} kt</td></tr>
<tr><td style='padding: 3px; font-weight: bold;'>Gusts:</td><td style='padding: 3px;'>{gust} kt</td></tr>
<tr><td style='padding: 3px; font-weight: bold;'>Pressure:</td><td style='padding: 3px;'>{pressure} mb</td></tr>
<tr><td style='padding: 3px; font-weight: bold;'>Category:</td><td style='padding: 3px;'>{ss_num if ss_num > 0 else 'N/A'}</td></tr>
<tr><td style='padding: 3px; font-weight: bold;'>Time:</td><td style='padding: 3px;'>{date_label}</td></tr>
<tr><td style='padding: 3px; font-weight: bold;'>Forecast:</td><td style='padding: 3px;'>+{tau}h</td></tr>
<tr><td style='padding: 3px; font-weight: bold;'>Advisory:</td><td style='padding: 3px;'>#{advisory_num}</td></tr>
<tr><td style='padding: 3px; font-weight: bold;'>Position:</td><td style='padding: 3px;'>{coords[1]:.1f}°N, {abs(coords[0]):.1f}°W</td></tr>
</table>
<div style='text-align: center; margin-top: 8px; font-size: 10px; color: #666;'>
Updated: {advisory_date}
</div>
</div>
"""
# Different marker styles for current vs forecast positions
if tau == 0: # Current position - SPINNING HURRICANE!
# Create spinning hurricane icon for current position
spinning_hurricane_html = f"""
<div style="
font-size: {radius * 2.5}px;
color: {color};
text-shadow: 3px 3px 6px rgba(0,0,0,0.7);
animation: spin 2s linear infinite;
display: flex;
align-items: center;
justify-content: center;
width: {radius * 4}px;
height: {radius * 4}px;
background: radial-gradient(circle, rgba(255,255,255,0.9) 0%, rgba(255,255,255,0.4) 70%, transparent 100%);
border-radius: 50%;
border: 3px solid {color};
box-shadow: 0 0 20px rgba(255,255,255,0.8), 0 0 40px {color};
">
🌀
</div>
<style>
@keyframes spin {{
from {{ transform: rotate(0deg); }}
to {{ transform: rotate(360deg); }}
}}
</style>
"""
# Add spinning hurricane marker for current position
folium.Marker(
location=[coords[1], coords[0]],
popup=folium.Popup(popup_text, max_width=300),
icon=folium.DivIcon(
html=spinning_hurricane_html,
class_name="spinning-hurricane",
icon_size=(radius * 4, radius * 4),
icon_anchor=(radius * 2, radius * 2)
)
).add_to(folium_map)
# Add multiple pulsing circles around current position
for i, pulse_radius in enumerate([radius + 10, radius + 20, radius + 30]):
folium.CircleMarker(
location=[coords[1], coords[0]],
radius=pulse_radius,
color=color,
fill=False,
weight=3 - i,
opacity=0.7 - (i * 0.2),
popup=f"Current Position: {storm_name}"
).add_to(folium_map)
else: # Forecast position
# For forecast positions, use smaller hurricane icon
forecast_icon_html = f"""
<div style="
font-size: {radius + 4}px;
color: {color};
text-shadow: 2px 2px 4px rgba(0,0,0,0.5);
display: flex;
align-items: center;
justify-content: center;
width: {radius * 2.5}px;
height: {radius * 2.5}px;
background: rgba(255,255,255,0.9);
border-radius: 50%;
border: 2px solid {color};
box-shadow: 0 0 10px rgba(0,0,0,0.3);
">
🌀
</div>
"""
folium.Marker(
location=[coords[1], coords[0]],
popup=folium.Popup(popup_text, max_width=300),
icon=folium.DivIcon(
html=forecast_icon_html,
class_name="forecast-hurricane",
icon_size=(radius * 2.5, radius * 2.5),
icon_anchor=(radius * 1.25, radius * 1.25)
)
).add_to(folium_map)
# Add forecast time label
folium.Marker(
location=[coords[1] - 0.15, coords[0]],
icon=folium.DivIcon(
html=f'<div style="font-size: 9px; background: white; padding: 2px 4px; border-radius: 3px; border: 1px solid {color}; box-shadow: 0 1px 3px rgba(0,0,0,0.3);">{date_label}</div>',
class_name="forecast-label",
icon_size=(70, 20),
icon_anchor=(35, 10)
)
).add_to(folium_map)
def add_current_storms_markers(self, folium_map, storm_data):
"""Add markers for storms from CurrentStorms.json format"""
# This method handles the specific format from NHC CurrentStorms.json
# Implementation depends on the actual JSON structure
pass
def add_storm_tracks(self, folium_map, track_data):
"""Add storm track lines to the map"""
for feature in track_data.get('features', []):
if feature['geometry']['type'] == 'LineString':
coords = feature['geometry']['coordinates']
# Convert coordinates from [lon, lat] to [lat, lon] for Folium
track_coords = [[coord[1], coord[0]] for coord in coords]
props = feature.get('properties', {})
storm_name = props.get('STORMNAME', 'Storm Track')
folium.PolyLine(
locations=track_coords,
color='red',
weight=3,
opacity=0.8,
popup=f"Track: {storm_name}"
).add_to(folium_map)
def get_storm_color(self, max_wind):
"""Get color based on storm intensity (Saffir-Simpson scale)"""
try:
wind_speed = float(max_wind) if max_wind != 'N/A' else 0
if wind_speed >= 137: # Category 5
return 'purple'
elif wind_speed >= 113: # Category 4
return 'red'
elif wind_speed >= 96: # Category 3
return 'orange'
elif wind_speed >= 83: # Category 2
return 'yellow'
elif wind_speed >= 64: # Category 1
return 'green'
elif wind_speed >= 34: # Tropical Storm
return 'blue'
else: # Tropical Depression
return 'lightblue'
except:
return 'gray'
def get_available_times(self):
"""Get list of available radar times for dropdown"""
radar_data = self.get_radar_data()
if radar_data and 'radar' in radar_data:
past_data = radar_data['radar']['past']
nowcast_data = radar_data['radar']['nowcast']
all_times = past_data + nowcast_data
time_options = []
for i, time_data in enumerate(all_times):
timestamp = time_data['time']
dt = datetime.fromtimestamp(timestamp)
time_options.append(f"{i}: {dt.strftime('%Y-%m-%d %H:%M:%S')}")
return time_options
return ["No data available"]
def create_gradio_app():
rain_viewer = RainViewerMap()
# Get available times for dropdown with error handling
try:
time_options = rain_viewer.get_available_times()
if not time_options:
time_options = ["No data available"]
except Exception as e:
print(f"Error getting initial time options: {e}")
time_options = ["Loading..."]
with gr.Blocks(title="RainViewer Radar Map") as demo:
gr.Markdown("# Weather Radar & Hurricane Tracker")
gr.Markdown("Interactive weather map with radar data from RainViewer and hurricane tracking from NOAA NHC")
with gr.Row():
with gr.Column(scale=1):
lat_input = gr.Number(
label="Latitude",
value=40.7128,
info="Map center latitude"
)
lon_input = gr.Number(
label="Longitude",
value=-74.0060,
info="Map center longitude"
)
zoom_input = gr.Slider(
minimum=1,
maximum=15,
value=8,
step=1,
label="Zoom Level"
)
show_radar = gr.Checkbox(
label="Show Radar",
value=True
)
show_hurricanes = gr.Checkbox(
label="Show Hurricanes",
value=True
)
time_dropdown = gr.Dropdown(
choices=time_options,
label="Radar Time",
value=time_options[-1] if time_options else None,
info="Select radar timestamp"
)
gr.Markdown("### 🌀 Active Hurricane Information")
storm_list = gr.HTML(
value="Loading storm data...",
label="Active Storms"
)
update_btn = gr.Button("Update Map", variant="primary")
refresh_times_btn = gr.Button("Refresh Times")
refresh_storms_btn = gr.Button("Refresh Storms")
with gr.Column(scale=2):
map_html = gr.HTML(
value=rain_viewer.create_map(),
label="Radar Map"
)
def update_map(lat, lon, zoom, show_radar_flag, show_hurricanes_flag, selected_time):
time_index = 0
if selected_time and ":" in selected_time and selected_time != "No data available" and selected_time != "Loading...":
try:
time_index = int(selected_time.split(":")[0])
# Validate time_index against available data
available_times = rain_viewer.get_available_times()
if time_index >= len(available_times):
time_index = len(available_times) - 1 if available_times else 0
except (ValueError, IndexError):
time_index = 0 # Default to first time if parsing fails
return rain_viewer.create_map(lat, lon, zoom, show_radar_flag, time_index, show_hurricanes_flag)
def refresh_times():
new_times = rain_viewer.get_available_times()
# Return both updated choices and a safe default value
safe_value = new_times[0] if new_times else None # Use first item instead of last
return gr.Dropdown(choices=new_times, value=safe_value)
def get_storm_list_html():
"""Generate HTML for storm list display"""
hurricane_data = rain_viewer.get_hurricane_data()
if not hurricane_data:
return "<p>No active storms currently detected.</p>"
html_content = "<div style='font-family: Arial, sans-serif;'>"
html_content += "<h4>Active Storms:</h4>"
for i, storm_info in enumerate(hurricane_data):
storm_type = storm_info.get('type', 'unknown')
source = storm_info.get('source', 'Unknown')
html_content += f"<div style='margin: 10px 0; padding: 10px; border: 1px solid #ddd; border-radius: 5px;'>"
html_content += f"<strong>Storm #{i+1}</strong> - Source: {source}<br>"
if storm_type == 'nhc_storm':
storm_id = storm_info.get('storm_id', 'Unknown')
forecast_points = storm_info.get('forecast_points', {})
features = forecast_points.get('features', [])
if features:
# Get current position (tau=0) and latest forecast
current_pos = None
latest_forecast = None
for feature in features:
props = feature.get('properties', {})
tau = props.get('tau', 999)
if tau == 0:
current_pos = props
if not latest_forecast or tau > latest_forecast.get('tau', -1):
latest_forecast = props
display_props = current_pos if current_pos else latest_forecast
if display_props:
storm_name = display_props.get('stormname', storm_id)
storm_type_detail = display_props.get('stormtype', 'Unknown')
max_wind = display_props.get('maxwind', 'N/A')
pressure = display_props.get('mslp', 'N/A')
advisory_num = display_props.get('advisnum', 'N/A')
advisory_date = display_props.get('advdate', 'N/A')
ss_num = display_props.get('ssnum', 0)
lat = display_props.get('lat', 'N/A')
lon = display_props.get('lon', 'N/A')
# Create detailed storm information table with spinning icon
html_content += f"""
<div style='background: linear-gradient(135deg, #f5f7fa 0%, #c3cfe2 100%); padding: 15px; border-radius: 10px; margin: 5px 0;'>
<div style='text-align: center; margin-bottom: 10px;'>
<span style='font-size: 24px; animation: spin 3s linear infinite;'>🌀</span>
<h3 style='margin: 5px 0; color: {self.get_storm_color(max_wind)};'>{storm_name}</h3>
</div>
<table style='width: 100%; border-collapse: collapse; font-size: 12px;'>
<tr style='background: rgba(255,255,255,0.7);'>
<td style='padding: 5px; border: 1px solid #ddd; font-weight: bold;'>Type:</td>
<td style='padding: 5px; border: 1px solid #ddd;'>{storm_type_detail}</td>
</tr>
<tr>
<td style='padding: 5px; border: 1px solid #ddd; font-weight: bold;'>Max Winds:</td>
<td style='padding: 5px; border: 1px solid #ddd; color: {self.get_storm_color(max_wind)};'><strong>{max_wind} kt</strong></td>
</tr>
<tr style='background: rgba(255,255,255,0.7);'>
<td style='padding: 5px; border: 1px solid #ddd; font-weight: bold;'>Category:</td>
<td style='padding: 5px; border: 1px solid #ddd;'>{f'Category {ss_num}' if ss_num > 0 else 'Tropical Storm/Depression'}</td>
</tr>
<tr>
<td style='padding: 5px; border: 1px solid #ddd; font-weight: bold;'>Pressure:</td>
<td style='padding: 5px; border: 1px solid #ddd;'>{pressure} mb</td>
</tr>
<tr style='background: rgba(255,255,255,0.7);'>
<td style='padding: 5px; border: 1px solid #ddd; font-weight: bold;'>Position:</td>
<td style='padding: 5px; border: 1px solid #ddd;'>{lat}°N, {abs(float(lon)) if lon != 'N/A' else 'N/A'}°W</td>
</tr>
<tr>
<td style='padding: 5px; border: 1px solid #ddd; font-weight: bold;'>Advisory:</td>
<td style='padding: 5px; border: 1px solid #ddd;'>#{advisory_num}</td>
</tr>
<tr style='background: rgba(255,255,255,0.7);'>
<td style='padding: 5px; border: 1px solid #ddd; font-weight: bold;'>Updated:</td>
<td style='padding: 5px; border: 1px solid #ddd;'>{advisory_date}</td>
</tr>
<tr>
<td style='padding: 5px; border: 1px solid #ddd; font-weight: bold;'>Forecast Points:</td>
<td style='padding: 5px; border: 1px solid #ddd;'>{len(features)} positions</td>
</tr>
</table>
<div style='margin-top: 10px; font-size: 11px; color: #666;'>
<strong>🌀 Forecast Track:</strong> Click hurricane markers on map for detailed timing
</div>
</div>
<style>
@keyframes spin {{
from {{ transform: rotate(0deg); }}
to {{ transform: rotate(360deg); }}
}}
</style>
"""
elif storm_type == 'sample':
html_content += "<em>Sample data for demonstration</em><br>"
html_content += "Name: Sample Hurricane<br>"
html_content += "Max Wind: 75 kt<br>"
html_content += "Pressure: 980 mb<br>"
html_content += "</div>"
# Add error information if available
errors = rain_viewer.storm_cache.get('errors', [])
if errors:
html_content += "<div style='margin-top: 15px; padding: 10px; background-color: #fff3cd; border: 1px solid #ffeaa7; border-radius: 5px;'>"
html_content += "<strong>Debug Info:</strong><br>"
for error in errors[-3:]: # Show last 3 errors
html_content += f"• {error}<br>"
html_content += "</div>"
html_content += "</div>"
return html_content
update_btn.click(
fn=update_map,
inputs=[lat_input, lon_input, zoom_input, show_radar, show_hurricanes, time_dropdown],
outputs=map_html
)
refresh_times_btn.click(
fn=refresh_times,
outputs=time_dropdown
)
refresh_storms_btn.click(
fn=get_storm_list_html,
outputs=storm_list
)
# Initialize components on load
def initialize_app():
storm_html = get_storm_list_html()
# Also refresh time options on load to ensure they're current
current_times = rain_viewer.get_available_times()
safe_time_value = current_times[0] if current_times else None
return storm_html, gr.Dropdown(choices=current_times, value=safe_time_value)
demo.load(
fn=initialize_app,
outputs=[storm_list, time_dropdown]
)
# Auto-update only on manual controls (not time dropdown to prevent conflicts)
for input_component in [lat_input, lon_input, zoom_input, show_radar, show_hurricanes]:
input_component.change(
fn=update_map,
inputs=[lat_input, lon_input, zoom_input, show_radar, show_hurricanes, time_dropdown],
outputs=map_html
)
# Only update map when time dropdown is explicitly changed by user
time_dropdown.select(
fn=update_map,
inputs=[lat_input, lon_input, zoom_input, show_radar, show_hurricanes, time_dropdown],
outputs=map_html
)
return demo
if __name__ == "__main__":
app = create_gradio_app()
app.launch(share=True)