Spaces:
Running
Running
import gradio as gr | |
import folium | |
from folium import plugins | |
import pandas as pd | |
from datetime import datetime | |
import time | |
import numpy as np | |
import plotly.graph_objects as go | |
from plotly.subplots import make_subplots | |
from opensky_api import OpenSkyApi # OpenSky API 클래스 import | |
# OpenSky API 인증 정보 | |
USERNAME = "seawolf2357" | |
PASSWORD = "Time2175!@" | |
api = OpenSkyApi(USERNAME, PASSWORD) | |
def get_states(bounds=None, max_retries=3): | |
"""Get current aircraft states from OpenSky Network with retry logic""" | |
for attempt in range(max_retries): | |
try: | |
if bounds: | |
# bbox = (min latitude, max latitude, min longitude, max longitude) | |
states = api.get_states(bbox=( | |
bounds.get('lamin', None), | |
bounds.get('lamax', None), | |
bounds.get('lomin', None), | |
bounds.get('lomax', None) | |
)) | |
else: | |
states = api.get_states() | |
if states: | |
return {'states': states.states} | |
else: | |
if attempt < max_retries - 1: | |
wait_time = min(2 ** attempt, 60) | |
print(f"Retrying in {wait_time} seconds...") | |
time.sleep(wait_time) | |
continue | |
return None | |
except Exception as e: | |
print(f"Error fetching data: {e}") | |
if attempt < max_retries - 1: | |
time.sleep(5) | |
continue | |
return None | |
return None | |
def create_monitoring_dashboard(states): | |
"""Create monitoring dashboard using Plotly""" | |
if not states: | |
return go.Figure() | |
# StateVector 객체에서 데이터 추출 | |
altitudes = [s.geo_altitude for s in states if s.geo_altitude is not None] | |
speeds = [s.velocity for s in states if s.velocity is not None] | |
countries = pd.Series([s.origin_country for s in states if s.origin_country]) | |
# Create subplots | |
fig = make_subplots( | |
rows=2, cols=2, | |
subplot_titles=('Altitude Distribution', 'Speed Distribution', | |
'Aircraft by Country', 'Aircraft Categories'), | |
specs=[ | |
[{"type": "xy"}, {"type": "xy"}], | |
[{"type": "xy"}, {"type": "domain"}] | |
] | |
) | |
# Add traces | |
fig.add_trace( | |
go.Histogram(x=altitudes, name="Altitude", marker_color='#4a90e2'), | |
row=1, col=1 | |
) | |
fig.add_trace( | |
go.Histogram(x=speeds, name="Speed", marker_color='#50C878'), | |
row=1, col=2 | |
) | |
# Top 10 countries | |
top_countries = countries.value_counts().head(10) | |
fig.add_trace( | |
go.Bar( | |
x=top_countries.index, | |
y=top_countries.values, | |
name="Countries", | |
marker_color='#FF6B6B' | |
), | |
row=2, col=1 | |
) | |
# Aircraft categories | |
categories = ['Commercial', 'Private', 'Military', 'Other'] | |
values = [40, 30, 20, 10] # Example distribution | |
fig.add_trace( | |
go.Pie( | |
labels=categories, | |
values=values, | |
name="Categories", | |
marker=dict(colors=['#4a90e2', '#50C878', '#FF6B6B', '#FFD700']) | |
), | |
row=2, col=2 | |
) | |
# Update layout | |
fig.update_layout( | |
height=800, | |
showlegend=True, | |
template="plotly_dark", | |
paper_bgcolor='rgba(0,0,0,0.3)', | |
plot_bgcolor='rgba(0,0,0,0.1)', | |
margin=dict(l=50, r=50, t=50, b=50), | |
font=dict(color='white'), | |
legend=dict( | |
bgcolor='rgba(0,0,0,0.3)', | |
bordercolor='rgba(255,255,255,0.2)', | |
borderwidth=1 | |
) | |
) | |
fig.update_xaxes(gridcolor='rgba(255,255,255,0.1)', zeroline=False) | |
fig.update_yaxes(gridcolor='rgba(255,255,255,0.1)', zeroline=False) | |
for i in fig['layout']['annotations']: | |
i['font'] = dict(size=12, color='white') | |
return fig | |
def create_map(region="world"): | |
"""Create aircraft tracking map""" | |
m = folium.Map( | |
location=[30, 0], | |
zoom_start=3, | |
tiles='CartoDB dark_matter' | |
) | |
bounds = { | |
"world": None, | |
"europe": {"lamin": 35.0, "lomin": -15.0, "lamax": 60.0, "lomax": 40.0}, | |
"north_america": {"lamin": 25.0, "lomin": -130.0, "lamax": 50.0, "lomax": -60.0}, | |
"asia": {"lamin": 10.0, "lomin": 60.0, "lamax": 50.0, "lomax": 150.0} | |
} | |
data = get_states(bounds.get(region)) | |
if not data or not data['states']: | |
return ( | |
m._repr_html_(), | |
create_monitoring_dashboard([]), | |
"No data available. Please try again later." | |
) | |
states = data['states'] | |
heat_data = [] | |
for state in states: | |
if state.latitude and state.longitude: | |
lat, lon = state.latitude, state.longitude | |
callsign = state.callsign if state.callsign else 'N/A' | |
altitude = state.geo_altitude if state.geo_altitude else 'N/A' | |
velocity = state.velocity if state.velocity else 'N/A' | |
heat_data.append([lat, lon, 1]) | |
popup_content = f""" | |
<div style="font-family: Arial; width: 200px;"> | |
<h4 style="color: #4a90e2;">Flight Information</h4> | |
<p><b>Callsign:</b> {callsign}</p> | |
<p><b>Altitude:</b> {altitude}m</p> | |
<p><b>Velocity:</b> {velocity}m/s</p> | |
<p><b>Origin:</b> {state.origin_country}</p> | |
</div> | |
""" | |
folium.Marker( | |
location=[lat, lon], | |
popup=folium.Popup(popup_content, max_width=300), | |
icon=folium.DivIcon( | |
html=f''' | |
<div style="transform: rotate({state.true_track if state.true_track else 0}deg)">✈️</div> | |
''', | |
icon_size=(20, 20) | |
) | |
).add_to(m) | |
plugins.HeatMap(heat_data, radius=15).add_to(m) | |
total_aircraft = len(states) | |
countries = len(set(s.origin_country for s in states if s.origin_country)) | |
avg_altitude = np.mean([s.geo_altitude for s in states if s.geo_altitude is not None]) if states else 0 | |
stats = f""" | |
📊 Real-time Statistics: | |
• Total Aircraft: {total_aircraft} | |
• Countries: {countries} | |
• Average Altitude: {avg_altitude:.0f}m | |
🔄 Last Updated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} | |
""" | |
return m._repr_html_(), create_monitoring_dashboard(states), stats | |
# Custom CSS | |
custom_css = """ | |
.gradio-container { | |
background: linear-gradient(135deg, #1a1a1a, #2d2d2d) !important; | |
} | |
.gr-button { | |
background: linear-gradient(135deg, #4a90e2, #357abd) !important; | |
border: none !important; | |
color: white !important; | |
} | |
.gr-button:hover { | |
background: linear-gradient(135deg, #357abd, #4a90e2) !important; | |
transform: translateY(-2px); | |
box-shadow: 0 5px 15px rgba(74, 144, 226, 0.4) !important; | |
} | |
""" | |
# Gradio interface | |
with gr.Blocks(css=custom_css) as demo: | |
gr.HTML( | |
""" | |
<h1 style="text-align: center; color: white;">🛩️ Global Aircraft Tracker</h1> | |
<p style="text-align: center; color: #ccc;">Real-time tracking of aircraft worldwide</p> | |
""" | |
) | |
gr.HTML("""<a href="https://visitorbadge.io/status?path=https%3A%2F%2Fimmunobiotech-opensky.hf.space"> | |
<img src="https://api.visitorbadge.io/api/visitors?path=https%3A%2F%2Fimmunobiotech-opensky.hf.space&countColor=%23263759" /> | |
</a>""") | |
with gr.Row(): | |
region_select = gr.Dropdown( | |
choices=["world", "europe", "north_america", "asia"], | |
value="world", | |
label="Select Region" | |
) | |
refresh_btn = gr.Button("🔄 Refresh") | |
map_html = gr.HTML() | |
stats_text = gr.Textbox(label="Statistics", lines=6) | |
dashboard_plot = gr.Plot(label="Monitoring Dashboard") | |
def update_map(region): | |
try: | |
return create_map(region) | |
except Exception as e: | |
print(f"Error updating map: {e}") | |
return ( | |
"<p>Map loading failed. Please try again.</p>", | |
go.Figure(), | |
f"Error: {str(e)}" | |
) | |
refresh_btn.click( | |
fn=update_map, | |
inputs=[region_select], | |
outputs=[map_html, dashboard_plot, stats_text] | |
) | |
region_select.change( | |
fn=update_map, | |
inputs=[region_select], | |
outputs=[map_html, dashboard_plot, stats_text] | |
) | |
# Launch with specific configurations | |
demo.launch( | |
show_error=True, | |
server_name="0.0.0.0", | |
server_port=7860, | |
share=False | |
) | |