import folium import gradio as gr from folium.plugins import MarkerCluster import re from datetime import datetime import os def parse_rep_file(file_content): """ Parse EEW REP file content and extract earthquake and station data """ lines = file_content.split('\n') # Initialize data structures earthquake_data = {} stations = [] for i, line in enumerate(lines): line = line.strip() if not line: continue # Parse header line if line.startswith('Reporting time'): parts = line.split() earthquake_data['reporting_time'] = f"{parts[1]} {parts[2]}" # Extract quality metrics metrics_match = re.search(r'averr=([\d.]+).*?Q=([\d-]+).*?Gap=(\d+).*?Avg_wei=([\d.]+).*?n=(\d+).*?n_c=(\d+).*?n_m=(\d+).*?Padj=([\d.]+).*?no_eq=(\d+)', line) if metrics_match: earthquake_data.update({ 'averr': float(metrics_match.group(1)), 'Q': int(metrics_match.group(2)), 'gap': int(metrics_match.group(3)), 'avg_wei': float(metrics_match.group(4)), 'n_total': int(metrics_match.group(5)), 'n_location': int(metrics_match.group(6)), 'n_magnitude': int(metrics_match.group(7)), 'padj': float(metrics_match.group(8)), 'event_num': int(metrics_match.group(9)) }) # Parse event summary line (contains origin time and location) elif re.match(r'^\d{4}\s+\d{1,2}\s+\d{1,2}', line): parts = line.split() if len(parts) >= 12: try: earthquake_data.update({ 'year': int(parts[0]), 'month': int(parts[1]), 'day': int(parts[2]), 'hour': int(parts[3]), 'minute': int(parts[4]), 'second': float(parts[5]), 'latitude': float(parts[6]), 'longitude': float(parts[7]), 'depth': float(parts[8]), 'magnitude_ml': float(parts[9]), 'magnitude_pd_s': float(parts[10]), 'magnitude_pv': float(parts[11]), 'magnitude_pd': float(parts[12]) if len(parts) > 12 else 0.0, 'magnitude_tc': float(parts[13]) if len(parts) > 13 else 0.0, 'process_time': float(parts[14]) if len(parts) > 14 else 0.0 }) except (ValueError, IndexError): pass # Parse station data (starts after the header lines) elif re.match(r'^[A-Z]\d{3}', line) or re.match(r'^\s+[A-Z]', line): parts = line.split() if len(parts) >= 20: try: station = { 'code': parts[0].strip(), 'component': parts[1], 'network': parts[2], 'location': parts[3], 'latitude': float(parts[4]), 'longitude': float(parts[5]), 'pga': float(parts[6]), # cm/s² 'pgv': float(parts[7]), # cm/s 'pgd': float(parts[8]), # cm 'tc': float(parts[9]), # S-P time residual 'mtc': float(parts[10]), # Tau-c magnitude 'mpv': float(parts[11]), # Velocity magnitude 'mpd': float(parts[12]), # Displacement magnitude 'perror': float(parts[13]), # Travel time residual 'distance': float(parts[14]), # km 'weight': float(parts[15]), # Station weight 'p_arrival': f"{parts[16]} {parts[17]}", 'pick_weight': int(parts[18]), 'update_sec': int(parts[19]), 'ps_ratio': float(parts[20]) if len(parts) > 20 else 0.0, 'used_sec': int(parts[21]) if len(parts) > 21 else 0 } stations.append(station) except (ValueError, IndexError): pass return earthquake_data, stations def create_earthquake_map(rep_content, filename=""): """ Create an interactive map showing earthquake epicenter and stations """ earthquake_data, stations = parse_rep_file(rep_content) if not earthquake_data or 'latitude' not in earthquake_data: return None, "Error: Could not parse earthquake data from REP file" # Create map centered on earthquake epicenter m = folium.Map( location=[earthquake_data['latitude'], earthquake_data['longitude']], zoom_start=8, tiles='OpenStreetMap' ) # Add earthquake epicenter marker magnitude = earthquake_data.get('magnitude_ml', 0) mag_color = 'red' if magnitude >= 7.0 else 'orange' if magnitude >= 6.0 else 'yellow' popup_content = f""" Earthquake Event #{earthquake_data.get('event_num', 'N/A')}
Magnitude: {magnitude:.1f} ML
Location: {earthquake_data['latitude']:.4f}°, {earthquake_data['longitude']:.4f}°
Depth: {earthquake_data.get('depth', 'N/A')} km
Origin Time: {earthquake_data.get('year', '')}-{earthquake_data.get('month', ''):02d}-{earthquake_data.get('day', ''):02d} {earthquake_data.get('hour', ''):02d}:{earthquake_data.get('minute', ''):02d}:{earthquake_data.get('second', ''):.1f}
Processing Time: {earthquake_data.get('process_time', 0):.1f} seconds
Stations: {len(stations)} total
Quality: Gap={earthquake_data.get('gap', 'N/A')}°, RMS={earthquake_data.get('averr', 'N/A'):.1f}s """ folium.CircleMarker( location=[earthquake_data['latitude'], earthquake_data['longitude']], radius=max(5, magnitude * 2), color=mag_color, fill=True, fill_color=mag_color, fill_opacity=0.7, popup=folium.Popup(popup_content, max_width=300) ).add_to(m) # Add station markers station_cluster = MarkerCluster(name="Stations").add_to(m) for station in stations: # Color code stations by PGA pga = station.get('pga', 0) if pga > 50: color = 'darkred' elif pga > 20: color = 'red' elif pga > 10: color = 'orange' elif pga > 5: color = 'yellow' else: color = 'green' station_popup = f""" Station: {station['code']}
Network: {station['network']}
Component: {station['component']}
Location: {station['latitude']:.4f}°, {station['longitude']:.4f}°
Distance: {station.get('distance', 'N/A')} km
PGA: {station.get('pga', 0):.2f} cm/s²
PGV: {station.get('pgv', 0):.3f} cm/s
PGD: {station.get('pgd', 0):.3f} cm
P-arrival: {station.get('p_arrival', 'N/A')}
Pick Weight: {station.get('pick_weight', 'N/A')}
Travel Time Residual: {station.get('perror', 0):.2f}s """ folium.CircleMarker( location=[station['latitude'], station['longitude']], radius=3, color=color, fill=True, fill_color=color, fill_opacity=0.8, popup=folium.Popup(station_popup, max_width=250) ).add_to(station_cluster) # Add layer control folium.LayerControl().add_to(m) # Add title title_html = f'''

Earthquake EEW Report

Event #{earthquake_data.get('event_num', 'N/A')} - Magnitude {magnitude:.1f} ML

File: {filename}

''' m.get_root().html.add_child(folium.Element(title_html)) return m, f"Successfully processed {len(stations)} stations" def process_rep_file(file): """ Process uploaded REP file and return map HTML """ if file is None: return None, "Please upload a REP file" try: content = file.decode('utf-8') filename = getattr(file, 'name', 'uploaded_file.rep') m, message = create_earthquake_map(content, filename) if m is None: return None, message # Save map to HTML string import io import base64 map_html = m.get_root().render() return map_html, message except Exception as e: return None, f"Error processing file: {str(e)}" # Gradio Interface def create_gradio_interface(): with gr.Blocks(title="EEW REP File Map Viewer") as interface: gr.Markdown(""" # 🌍 Earthquake EEW Report Map Viewer Upload an EEW REP file to visualize the earthquake epicenter and seismic stations on an interactive map. ## Features: - **Epicenter**: Red/orange/yellow circle based on magnitude - **Stations**: Colored by PGA intensity (green=low, red=high) - **Interactive**: Click markers for detailed information - **Clustering**: Stations grouped for better visibility at low zoom ## REP File Format: Files should be in the standard EEW report format with `.rep` extension. """) with gr.Row(): with gr.Column(scale=1): file_input = gr.File( label="Upload REP File", file_types=[".rep"], type="binary" ) process_btn = gr.Button("Generate Map", variant="primary") with gr.Column(scale=2): map_output = gr.HTML(label="Interactive Map") status_output = gr.Textbox(label="Status", interactive=False) # Sample files section gr.Markdown("### Sample REP Files") gr.Markdown("Try these sample files to see the visualization:") # Event handlers process_btn.click( fn=process_rep_file, inputs=[file_input], outputs=[map_output, status_output] ) # Auto-process on file upload file_input.change( fn=process_rep_file, inputs=[file_input], outputs=[map_output, status_output] ) return interface # Main execution if __name__ == "__main__": interface = create_gradio_interface() interface.launch( server_name="0.0.0.0", server_port=7860, show_error=True, theme=gr.themes.Soft() )