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)