Spaces:
Running
Running
| import requests | |
| import json | |
| from datetime import datetime, timedelta | |
| import gradio as gr | |
| import pandas as pd | |
| import traceback | |
| import plotly.express as px | |
| import plotly.graph_objects as go | |
| class NasaSsdCneosApi: | |
| def __init__(self): | |
| self.fireball_url = "https://ssd-api.jpl.nasa.gov/fireball.api" | |
| self.ca_url = "https://ssd-api.jpl.nasa.gov/cad.api" | |
| self.nea_url = "https://ssd-api.jpl.nasa.gov/sbdb_query.api" | |
| self.scout_url = "https://ssd-api.jpl.nasa.gov/scout.api" | |
| def get_fireballs(self, limit=10, date_min=None, energy_min=None): | |
| try: | |
| params = {'limit': limit} | |
| if date_min: | |
| params['date-min'] = date_min | |
| if energy_min: | |
| params['energy-min'] = energy_min | |
| response = requests.get(self.fireball_url, params=params) | |
| response.raise_for_status() | |
| return response.json() | |
| except Exception as e: | |
| print("Fireball API Error:", e) | |
| traceback.print_exc() | |
| return None | |
| def get_close_approaches(self, dist_max=None, date_min=None, date_max=None, | |
| h_min=None, h_max=None, v_inf_min=None, v_inf_max=None, | |
| limit=10): | |
| try: | |
| params = {'limit': limit, 'dist-max': dist_max, 'date-min': date_min, | |
| 'date-max': date_max, 'h-min': h_min, 'h-max': h_max, | |
| 'v-inf-min': v_inf_min, 'v-inf-max': v_inf_max, 'sort': 'date'} | |
| params = {k: v for k, v in params.items() if v is not None} | |
| response = requests.get(self.ca_url, params=params) | |
| response.raise_for_status() | |
| return response.json() | |
| except Exception as e: | |
| print("Close Approaches API Error:", e) | |
| traceback.print_exc() | |
| return None | |
| def get_nea_data(self, des=None, spk_id=None, h_max=None): | |
| try: | |
| # Build query parameter for NEAs - select asteroid with NEA flag | |
| query_params = { | |
| 'sb-nea': 'true' # Filter for Near-Earth Asteroids | |
| } | |
| if des: | |
| query_params['sb-spk'] = des | |
| if spk_id: | |
| query_params['sb-spkid'] = spk_id | |
| if h_max: | |
| query_params['sb-h-max'] = h_max | |
| # Add fields to return | |
| query_params['fields'] = 'spkid,full_name,pdes,neo,H,G,diameter,extent,albedo,rot_per,GM,BV,UB,IR,spec_B,spec_T,H_sigma,diameter_sigma,orbit_id,epoch,epoch_mjd,epoch_cal,a,e,i,om,w,ma,ad,n,tp,tp_cal,per,per_y,q,moid,moid_ld,moid_jup' | |
| query_params['limit'] = 100 # Set a reasonable limit | |
| response = requests.get(self.nea_url, params=query_params) | |
| response.raise_for_status() | |
| return response.json() | |
| except Exception as e: | |
| print("NEA API Error:", e) | |
| traceback.print_exc() | |
| return None | |
| def get_scout_data(self, limit=10, nea_comet="NEA"): | |
| try: | |
| params = {'limit': limit} | |
| if nea_comet: | |
| params['nea-comet'] = nea_comet.lower() | |
| response = requests.get(self.scout_url, params=params) | |
| response.raise_for_status() | |
| return response.json() | |
| except Exception as e: | |
| print("Scout API Error:", e) | |
| traceback.print_exc() | |
| return None | |
| def format_response(self, data, format_type): | |
| try: | |
| if not data: | |
| return None | |
| fields = data.get('fields') | |
| rows = data.get('data') | |
| if not fields or not rows: | |
| return None | |
| df = pd.DataFrame([dict(zip(fields, row)) for row in rows]) | |
| if format_type == 'fireballs': | |
| return df.rename(columns={ | |
| 'date': 'Date/Time', 'energy': 'Energy (kt)', | |
| 'impact-e': 'Impact Energy (10^10 J)', 'lat': 'Latitude', | |
| 'lon': 'Longitude', 'alt': 'Altitude (km)', | |
| 'vel': 'Velocity (km/s)' | |
| }) | |
| elif format_type == 'close_approaches': | |
| return df.rename(columns={ | |
| 'des': 'Object', 'orbit_id': 'Orbit ID', 'cd': 'Time (TDB)', | |
| 'dist': 'Nominal Distance (au)', 'dist_min': 'Minimum Distance (au)', | |
| 'dist_max': 'Maximum Distance (au)', 'v_rel': 'Velocity (km/s)', | |
| 'h': 'H (mag)' | |
| }) | |
| elif format_type == 'nea': | |
| name_columns = { | |
| 'full_name': 'Full Name', 'pdes': 'Designation', | |
| 'H': 'Absolute Magnitude (mag)', 'diameter': 'Diameter (km)', | |
| 'q': 'Perihelion (au)', 'ad': 'Aphelion (au)', | |
| 'i': 'Inclination (deg)', 'e': 'Eccentricity', | |
| 'moid': 'MOID (au)', 'moid_ld': 'MOID (LD)' | |
| } | |
| # Use only columns that exist in the dataframe | |
| valid_columns = {k: v for k, v in name_columns.items() if k in df.columns} | |
| return df.rename(columns=valid_columns) | |
| elif format_type == 'scout': | |
| # Handle Scout API response - column names may vary | |
| # Adjust these column mappings based on actual response structure | |
| if 'score' in df.columns: | |
| df = df.rename(columns={ | |
| 'object': 'Object', 'score': 'Rating', | |
| 'diameter': 'Diameter (m)', 'ca_dist': 'Close Approach', | |
| 'nobs': 'Observations' | |
| }) | |
| return df | |
| return df | |
| except Exception as e: | |
| print("Data formatting error:", e) | |
| traceback.print_exc() | |
| return None | |
| # Gradio Interface Functions | |
| def fetch_fireballs(limit, date_min, energy_min): | |
| api = NasaSsdCneosApi() | |
| # Convert empty strings to None | |
| date_min = date_min if date_min else None | |
| energy_min = float(energy_min) if energy_min else None | |
| data = api.get_fireballs( | |
| limit=int(limit), | |
| date_min=date_min, | |
| energy_min=energy_min | |
| ) | |
| df = api.format_response(data, 'fireballs') | |
| if df is None or df.empty: | |
| return "No data available", None | |
| # Create world map of fireballs | |
| if 'Latitude' in df.columns and 'Longitude' in df.columns: | |
| fig = px.scatter_geo(df, | |
| lat='Latitude', | |
| lon='Longitude', | |
| size='Energy (kt)', | |
| hover_name='Date/Time', | |
| projection='natural earth', | |
| title='Fireball Events') | |
| return df, fig | |
| return df, None | |
| def fetch_close_approaches(limit, dist_max, date_min, date_max, h_min, h_max, v_inf_min, v_inf_max): | |
| api = NasaSsdCneosApi() | |
| # Convert empty strings to None | |
| dist_max = float(dist_max) if dist_max else None | |
| date_min = date_min if date_min else None | |
| date_max = date_max if date_max else None | |
| h_min = float(h_min) if h_min else None | |
| h_max = float(h_max) if h_max else None | |
| v_inf_min = float(v_inf_min) if v_inf_min else None | |
| v_inf_max = float(v_inf_max) if v_inf_max else None | |
| data = api.get_close_approaches( | |
| limit=int(limit), | |
| dist_max=dist_max, | |
| date_min=date_min, | |
| date_max=date_max, | |
| h_min=h_min, | |
| h_max=h_max, | |
| v_inf_min=v_inf_min, | |
| v_inf_max=v_inf_max | |
| ) | |
| df = api.format_response(data, 'close_approaches') | |
| if df is None or df.empty: | |
| return "No data available", None | |
| # Create scatter plot of distance vs velocity | |
| fig = px.scatter(df, | |
| x='Nominal Distance (au)', | |
| y='Velocity (km/s)', | |
| hover_name='Object', | |
| size='H (mag)', | |
| color='H (mag)', | |
| title='Close Approaches - Distance vs Velocity') | |
| return df, fig | |
| def fetch_nea_data(des, spk_id, h_max): | |
| api = NasaSsdCneosApi() | |
| # Convert empty strings to None | |
| des = des if des else None | |
| spk_id = spk_id if spk_id else None | |
| h_max = float(h_max) if h_max else None | |
| data = api.get_nea_data( | |
| des=des, | |
| spk_id=spk_id, | |
| h_max=h_max | |
| ) | |
| df = api.format_response(data, 'nea') | |
| if df is None or df.empty: | |
| return "No data available", None | |
| # Create a scatter plot of perihelion vs aphelion colored by inclination | |
| if not df.empty and 'Perihelion (au)' in df.columns and 'Aphelion (au)' in df.columns: | |
| fig = px.scatter(df, | |
| x='Perihelion (au)', | |
| y='Aphelion (au)', | |
| hover_name='Designation' if 'Designation' in df.columns else None, | |
| color='Inclination (deg)' if 'Inclination (deg)' in df.columns else None, | |
| size='Diameter (km)' if 'Diameter (km)' in df.columns else None, | |
| title='NEA Orbital Parameters') | |
| return df, fig | |
| return df, None | |
| def fetch_scout_data(limit, nea_comet): | |
| api = NasaSsdCneosApi() | |
| data = api.get_scout_data( | |
| limit=int(limit), | |
| nea_comet=nea_comet | |
| ) | |
| df = api.format_response(data, 'scout') | |
| if df is None or df.empty: | |
| return "No data available", None | |
| # Create a scatter plot based on available columns | |
| if not df.empty: | |
| # Use columns that are available in the dataframe | |
| x_col = 'Diameter (m)' if 'Diameter (m)' in df.columns else df.columns[0] | |
| y_col = 'Close Approach' if 'Close Approach' in df.columns else df.columns[1] | |
| hover_col = 'Object' if 'Object' in df.columns else None | |
| color_col = 'Rating' if 'Rating' in df.columns else None | |
| size_col = 'Observations' if 'Observations' in df.columns else None | |
| fig = px.scatter(df, | |
| x=x_col, | |
| y=y_col, | |
| hover_name=hover_col, | |
| color=color_col, | |
| size=size_col, | |
| title='Scout Objects') | |
| return df, fig | |
| return df, None | |
| # Create Gradio interface | |
| with gr.Blocks(title="NASA SSD/CNEOS API Explorer") as demo: | |
| gr.Markdown("# NASA SSD/CNEOS API Explorer") | |
| gr.Markdown("Access data from NASA's Center for Near Earth Object Studies") | |
| with gr.Tab("Fireballs"): | |
| gr.Markdown("### Fireball Events") | |
| gr.Markdown("Get information about recent fireball events detected by sensors.") | |
| with gr.Row(): | |
| with gr.Column(): | |
| fireball_limit = gr.Slider(minimum=1, maximum=100, value=10, step=1, label="Limit") | |
| fireball_date = gr.Textbox(label="Minimum Date (YYYY-MM-DD)", placeholder="e.g. 2023-01-01") | |
| fireball_energy = gr.Textbox(label="Minimum Energy (kt)", placeholder="e.g. 0.5") | |
| fireball_submit = gr.Button("Fetch Fireballs") | |
| with gr.Column(): | |
| fireball_results = gr.DataFrame(label="Fireball Results") | |
| fireball_map = gr.Plot(label="Fireball Map") | |
| fireball_submit.click(fetch_fireballs, inputs=[fireball_limit, fireball_date, fireball_energy], outputs=[fireball_results, fireball_map]) | |
| with gr.Tab("Close Approaches"): | |
| gr.Markdown("### Close Approaches") | |
| gr.Markdown("Get information about close approaches of near-Earth objects.") | |
| with gr.Row(): | |
| with gr.Column(): | |
| ca_limit = gr.Slider(minimum=1, maximum=100, value=10, step=1, label="Limit") | |
| ca_dist_max = gr.Textbox(label="Maximum Distance (AU)", placeholder="e.g. 0.05") | |
| ca_date_min = gr.Textbox(label="Minimum Date (YYYY-MM-DD)", placeholder="e.g. 2023-01-01") | |
| ca_date_max = gr.Textbox(label="Maximum Date (YYYY-MM-DD)", placeholder="e.g. 2023-12-31") | |
| ca_h_min = gr.Textbox(label="Minimum H (mag)", placeholder="e.g. 20") | |
| ca_h_max = gr.Textbox(label="Maximum H (mag)", placeholder="e.g. 30") | |
| ca_v_min = gr.Textbox(label="Minimum Velocity (km/s)", placeholder="e.g. 10") | |
| ca_v_max = gr.Textbox(label="Maximum Velocity (km/s)", placeholder="e.g. 30") | |
| ca_submit = gr.Button("Fetch Close Approaches") | |
| with gr.Column(): | |
| ca_results = gr.DataFrame(label="Close Approach Results") | |
| ca_plot = gr.Plot(label="Close Approach Plot") | |
| ca_submit.click(fetch_close_approaches, | |
| inputs=[ca_limit, ca_dist_max, ca_date_min, ca_date_max, ca_h_min, ca_h_max, ca_v_min, ca_v_max], | |
| outputs=[ca_results, ca_plot]) | |
| with gr.Tab("NEA Data"): | |
| gr.Markdown("### Near-Earth Asteroid Data") | |
| gr.Markdown("Get information about specific near-Earth asteroids.") | |
| with gr.Row(): | |
| with gr.Column(): | |
| nea_des = gr.Textbox(label="Designation", placeholder="e.g. 2020 SW") | |
| nea_spk = gr.Textbox(label="SPK-ID", placeholder="e.g. 54101815") | |
| nea_h_max = gr.Textbox(label="Maximum H (mag)", placeholder="e.g. 25") | |
| nea_submit = gr.Button("Fetch NEA Data") | |
| with gr.Column(): | |
| nea_results = gr.DataFrame(label="NEA Results") | |
| nea_plot = gr.Plot(label="NEA Orbital Parameters") | |
| nea_submit.click(fetch_nea_data, inputs=[nea_des, nea_spk, nea_h_max], outputs=[nea_results, nea_plot]) | |
| with gr.Tab("Scout Data"): | |
| gr.Markdown("### Scout System Data") | |
| gr.Markdown("Get information about newly discovered objects from NASA's Scout system.") | |
| with gr.Row(): | |
| with gr.Column(): | |
| scout_limit = gr.Slider(minimum=1, maximum=100, value=10, step=1, label="Limit") | |
| scout_type = gr.Radio(["NEA", "comet"], label="Object Type", value="NEA") | |
| scout_submit = gr.Button("Fetch Scout Data") | |
| with gr.Column(): | |
| scout_results = gr.DataFrame(label="Scout Results") | |
| scout_plot = gr.Plot(label="Scout Objects Plot") | |
| scout_submit.click(fetch_scout_data, inputs=[scout_limit, scout_type], outputs=[scout_results, scout_plot]) | |
| gr.Markdown("### About") | |
| gr.Markdown(""" | |
| This application provides access to NASA's Solar System Dynamics (SSD) and Center for Near Earth Object Studies (CNEOS) API. | |
| Data is retrieved in real-time from NASA's servers. All data is courtesy of NASA/JPL-Caltech. | |
| Created by [Your Name] using Gradio and Hugging Face Spaces. | |
| """) | |
| # Create requirements.txt file | |
| requirements = """ | |
| gradio>=3.50.0 | |
| pandas>=1.5.0 | |
| plotly>=5.14.0 | |
| requests>=2.28.0 | |
| """ | |
| with open("requirements.txt", "w") as f: | |
| f.write(requirements) | |
| if __name__ == "__main__": | |
| demo.launch() |