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"""
🌀
{storm_name}
Type: | {storm_type_detail} |
Max Winds: | {max_wind} kt |
Gusts: | {gust} kt |
Pressure: | {pressure} mb |
Category: | {ss_num if ss_num > 0 else 'N/A'} |
Time: | {date_label} |
Forecast: | +{tau}h |
Advisory: | #{advisory_num} |
Position: | {coords[1]:.1f}°N, {abs(coords[0]):.1f}°W |
Updated: {advisory_date}
"""
# 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"""
🌀
"""
# 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"""
🌀
"""
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'{date_label}
',
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 "No active storms currently detected.
"
html_content = ""
html_content += "
Active Storms:
"
for i, storm_info in enumerate(hurricane_data):
storm_type = storm_info.get('type', 'unknown')
source = storm_info.get('source', 'Unknown')
html_content += f"
"
html_content += f"
Storm #{i+1} - Source: {source}
"
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"""
🌀
{storm_name}
Type: |
{storm_type_detail} |
Max Winds: |
{max_wind} kt |
Category: |
{f'Category {ss_num}' if ss_num > 0 else 'Tropical Storm/Depression'} |
Pressure: |
{pressure} mb |
Position: |
{lat}°N, {abs(float(lon)) if lon != 'N/A' else 'N/A'}°W |
Advisory: |
#{advisory_num} |
Updated: |
{advisory_date} |
Forecast Points: |
{len(features)} positions |
🌀 Forecast Track: Click hurricane markers on map for detailed timing
"""
elif storm_type == 'sample':
html_content += "
Sample data for demonstration"
html_content += "Name: Sample Hurricane
"
html_content += "Max Wind: 75 kt
"
html_content += "Pressure: 980 mb
"
html_content += "
"
# Add error information if available
errors = rain_viewer.storm_cache.get('errors', [])
if errors:
html_content += "
"
html_content += "Debug Info:
"
for error in errors[-3:]: # Show last 3 errors
html_content += f"• {error}
"
html_content += "
"
html_content += "
"
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)