-
Notifications
You must be signed in to change notification settings - Fork 0
/
ZNY.py
355 lines (311 loc) · 16.4 KB
/
ZNY.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
import discord
from discord import app_commands
from discord.ext import commands
import requests
from dotenv import load_dotenv
import os
import re
# Load environment variables from .env.example file
load_dotenv('.env.example')
# Retrieve the Discord bot token and channel ID from environment variables
TOKEN = os.getenv('DISCORD_BOT_TOKEN')
CHANNEL_ID = int(os.getenv('CHANNEL_ID')) # Ensure the channel ID is an integer
VATSIM_API_URL = 'https://data.vatsim.net/v3/vatsim-data.json'
AIRPLANES_LIVE_API_URL = 'http://api.airplanes.live/v2/'
# Initialize the bot
bot = commands.Bot(command_prefix="!", intents=discord.Intents.default())
@bot.event
async def on_ready():
print(f'Logged in as {bot.user}!')
try:
synced = await bot.tree.sync()
print(f'Synced {len(synced)} command(s)')
except Exception as e:
print(f'Error syncing commands: {e}')
# Define the slash command to get ATIS information
@bot.tree.command(name="atis", description="Fetch ATIS information from VATSIM")
@app_commands.describe(icao="The ICAO code of the airport")
async def fetch_atis(interaction: discord.Interaction, icao: str):
# Fetch data from VATSIM API
response = requests.get(VATSIM_API_URL)
if response.status_code == 200:
data = response.json()
atis_list = data.get('atis', [])
atis_info = None
# Search for the requested ICAO code in the ATIS data
for atis in atis_list:
if atis['callsign'].startswith(icao.upper()):
atis_info = atis
break
if atis_info:
atis_text = "\n".join(atis_info['text_atis'])
# Create an embed with the ATIS information
embed = discord.Embed(title=f"ATIS for {icao.upper()}", color=0x1e90ff)
embed.add_field(name="ATIS Information", value=atis_text, inline=False)
embed.add_field(name="Frequency", value=atis_info['frequency'], inline=True)
embed.add_field(name="ATIS Code", value=atis_info.get('atis_code', 'N/A'), inline=True)
await interaction.response.send_message(embed=embed)
else:
await interaction.response.send_message(f"No ATIS data available for {icao.upper()}.")
else:
print(f"Error fetching VATSIM data: {response.text}") # Log error details
await interaction.response.send_message(f"Error fetching VATSIM data: {response.status_code}")
# Define the slash command to get METAR and TAF information
@bot.tree.command(name="weather", description="Fetch METAR and TAF information from VATSIM")
@app_commands.describe(icao="The ICAO code of the airport")
async def fetch_metar(interaction: discord.Interaction, icao: str):
# Fetch METAR data from VATSIM METAR API
metar_url = f"https://metar.vatsim.net/{icao}"
print(f"Fetching METAR data from: {metar_url}") # Log the URL
response = requests.get(metar_url, headers={'Accept': 'text/plain'})
if response.status_code == 200:
metar_data = response.text.strip()
decoded_metar = decode_metar(metar_data)
# Create an embed with the METAR information
embed = discord.Embed(title=f"METAR for {icao.upper()} observation at {decoded_metar['observation_time']}", color=0x1e90ff)
embed.add_field(name="Raw METAR", value=f"`{metar_data}`", inline=False)
embed.add_field(name="Decoded METAR", value=decoded_metar['decoded'], inline=False)
# Fetch TAF data
taf_url = f"https://metar.vatsim.net/{icao}?taf"
taf_response = requests.get(taf_url, headers={'Accept': 'text/plain'})
if taf_response.status_code == 200:
taf_data = taf_response.text.strip()
embed.add_field(name="TAF", value=f"`{taf_data}`", inline=False)
else:
embed.add_field(name="TAF", value="No TAF data available.", inline=False)
await interaction.response.send_message(embed=embed)
else:
print(f"Error fetching METAR data: {response.text}") # Log error details
await interaction.response.send_message(f"Error fetching METAR data: {response.status_code}")
# Define the slash command to get online pilots information
@bot.tree.command(name="pilots", description="Fetch online pilots information from VATSIM")
async def fetch_pilots(interaction: discord.Interaction):
# Fetch data from VATSIM API
response = requests.get(VATSIM_API_URL)
if response.status_code == 200:
data = response.json()
pilots_list = data.get('pilots', [])
if pilots_list:
pilots_info = []
for pilot in pilots_list[:10]: # Limiting to 10 pilots for simplicity
pilots_info.append(
f"**{pilot['callsign']}** | {pilot['name']}\n"
f"Aircraft: {pilot['flight_plan']['aircraft_faa'] if pilot.get('flight_plan') else 'N/A'}\n"
f"Altitude: {pilot['altitude']} ft | Ground Speed: {pilot['groundspeed']} kts\n"
f"Location: {pilot['latitude']}, {pilot['longitude']}\n"
)
pilots_info_str = "\n".join(pilots_info)
embed = discord.Embed(title="Online Pilots", description=pilots_info_str, color=0x1e90ff)
await interaction.response.send_message(embed=embed)
else:
await interaction.response.send_message("No pilots currently online.")
else:
print(f"Error fetching pilots data: {response.text}") # Log error details
await interaction.response.send_message(f"Error fetching pilots data: {response.status_code}")
# Define the slash command to get online controllers information
@bot.tree.command(name="controllers", description="Fetch online controllers information from VATSIM")
async def fetch_controllers(interaction: discord.Interaction):
# Fetch data from VATSIM API
response = requests.get(VATSIM_API_URL)
if response.status_code == 200:
data = response.json()
controllers_list = data.get('controllers', [])
if controllers_list:
controllers_info = []
for controller in controllers_list[:10]: # Limiting to 10 controllers for simplicity
controllers_info.append(
f"**{controller['callsign']}** | {controller['name']}\n"
f"Frequency: {controller['frequency']}\n"
f"Position: {controller['facility']}\n"
)
controllers_info_str = "\n".join(controllers_info)
embed = discord.Embed(title="Online Controllers", description=controllers_info_str, color=0x1e90ff)
await interaction.response.send_message(embed=embed)
else:
await interaction.response.send_message("No controllers currently online.")
else:
print(f"Error fetching controllers data: {response.text}") # Log error details
await interaction.response.send_message(f"Error fetching controllers data: {response.status_code}")
# Define the slash command to get VATSIM server information
@bot.tree.command(name="servers", description="Fetch VATSIM server information")
async def fetch_servers(interaction: discord.Interaction):
# Fetch data from VATSIM API
response = requests.get(VATSIM_API_URL)
if response.status_code == 200:
data = response.json()
servers_list = data.get('servers', [])
if servers_list:
servers_info = []
for server in servers_list:
servers_info.append(
f"**{server['name']}** ({server['ident']})\n"
f"Location: {server['location']}\n"
f"Hostname/IP: {server['hostname_or_ip']}\n"
)
servers_info_str = "\n".join(servers_info)
embed = discord.Embed(title="VATSIM Servers", description=servers_info_str, color=0x1e90ff)
await interaction.response.send_message(embed=embed)
else:
await interaction.response.send_message("No servers currently available.")
else:
print(f"Error fetching servers data: {response.text}") # Log error details
await interaction.response.send_message(f"Error fetching servers data: {response.status_code}")
# Define the slash command to search for a callsign
@bot.tree.command(name="search", description="Search for a specific callsign on VATSIM")
@app_commands.describe(callsign="The callsign to search for")
async def search_callsign(interaction: discord.Interaction, callsign: str):
# Fetch data from VATSIM API
response = requests.get(VATSIM_API_URL)
if response.status_code == 200:
data = response.json()
callsign_upper = callsign.upper()
# Search in pilots
pilot_info = next((pilot for pilot in data.get('pilots', []) if pilot['callsign'].upper() == callsign_upper), None)
controller_info = next((controller for controller in data.get('controllers', []) if controller['callsign'].upper() == callsign_upper), None)
atis_info = next((atis for atis in data.get('atis', []) if atis['callsign'].upper() == callsign_upper), None)
if pilot_info or controller_info or atis_info:
embed = discord.Embed(title=f"Information for {callsign_upper}", color=0x1e90ff)
if pilot_info:
embed.add_field(
name="Pilot",
value=(
f"**Name**: {pilot_info['name']}\n"
f"**Callsign**: {pilot_info['callsign']}\n"
f"**Aircraft**: {pilot_info['flight_plan']['aircraft_faa'] if pilot_info.get('flight_plan') else 'N/A'}\n"
f"**Altitude**: {pilot_info['altitude']} ft\n"
f"**Ground Speed**: {pilot_info['groundspeed']} kts\n"
f"**Location**: {pilot_info['latitude']}, {pilot_info['longitude']}\n"
f"**Logon Time**: {pilot_info['logon_time']}\n"
),
inline=False
)
if controller_info:
embed.add_field(
name="Controller",
value=(
f"**Name**: {controller_info['name']}\n"
f"**Callsign**: {controller_info['callsign']}\n"
f"**Frequency**: {controller_info['frequency']}\n"
f"**Position**: {controller_info['facility']}\n"
f"**Logon Time**: {controller_info['logon_time']}\n"
),
inline=False
)
if atis_info:
atis_text = "\n".join(atis_info['text_atis'])
embed.add_field(
name="ATIS",
value=(
f"**Callsign**: {atis_info['callsign']}\n"
f"**Frequency**: {atis_info['frequency']}\n"
f"**ATIS Code**: {atis_info.get('atis_code', 'N/A')}\n"
f"**ATIS Information**: {atis_text}\n"
),
inline=False
)
await interaction.response.send_message(embed=embed)
else:
await interaction.response.send_message(f"No data found for the callsign '{callsign_upper}'.")
else:
print(f"Error fetching VATSIM data: {response.text}") # Log error details
await interaction.response.send_message(f"Error fetching VATSIM data: {response.status_code}")
# Define the slash command to search for real-life aircraft by callsign using Airplanes.live API
@bot.tree.command(name="irl_search", description="Search for a specific callsign using Airplanes.live API")
@app_commands.describe(callsign="The callsign to search for")
async def irl_search(interaction: discord.Interaction, callsign: str):
# Fetch data from Airplanes.live API
url = f"{AIRPLANES_LIVE_API_URL}callsign/{callsign.upper()}"
response = requests.get(url)
if response.status_code == 200:
data = response.json()
aircraft = data.get('ac', [])
if aircraft:
aircraft_info = aircraft[0] # Assuming we take the first match
last_position = aircraft_info.get('lastPosition', {})
embed = discord.Embed(
title=f"IRL Information for {callsign.upper()}",
description=(
f"**Hex Code**: {aircraft_info.get('hex', 'N/A')}\n"
f"**Type**: {aircraft_info.get('t', 'N/A')} ({aircraft_info.get('desc', 'N/A')})\n"
f"**Registration**: {aircraft_info.get('r', 'N/A')}\n"
f"**Altitude**: {aircraft_info.get('alt_baro', 'N/A')} ft\n"
f"**Ground Speed**: {aircraft_info.get('gs', 'N/A')} kts\n"
f"**Track**: {aircraft_info.get('track', 'N/A')}°\n"
f"**Squawk**: {aircraft_info.get('squawk', 'N/A')}\n"
f"**Latitude**: {last_position.get('lat', 'N/A')}\n"
f"**Longitude**: {last_position.get('lon', 'N/A')}\n"
f"**Last Seen**: {aircraft_info.get('seen', 'N/A')} seconds ago"
),
color=0x1e90ff
)
# Adding tracking link
tracking_link = f"https://globe.airplanes.live/?icao={aircraft_info.get('hex', '')}"
embed.add_field(name="Track on Airplanes.live", value=f"[Track here]({tracking_link})")
await interaction.response.send_message(embed=embed)
else:
await interaction.response.send_message(f"No aircraft found with callsign '{callsign.upper()}'.")
else:
print(f"Error fetching data from Airplanes.live API: {response.text}") # Log error details
await interaction.response.send_message(f"Error fetching data from Airplanes.live API: {response.status_code}")
# Function to decode a METAR string into human-readable information
def decode_metar(metar: str) -> dict:
decoded = []
# Extracting observation time (first group in METAR)
observation_time_match = re.search(r'\b(\d{6}Z)\b', metar)
observation_time = observation_time_match.group(1) if observation_time_match else "Unknown"
# Parse METAR elements using regex
wind_re = re.compile(r'(\d{3})(\d{2,3})(G\d{2,3})?KT')
visibility_re = re.compile(r'(\d{4})')
temp_dew_re = re.compile(r'(M?\d{2})/(M?\d{2})')
pressure_re = re.compile(r'(QNH|A)(\d{4})')
weather_re = re.compile(r'(\+|-|VC)?(MI|PR|BC|DR|BL|SH|TS|FZ)?(DZ|RA|SN|SG|IC|PL|GR|GS|UP|HZ|BR|FG|FU|VA|DU|SA|SS|DS)?')
cloud_re = re.compile(r'(FEW|SCT|BKN|OVC)(\d{3})')
# Decode wind
wind = wind_re.search(metar)
if wind:
direction, speed, gust = wind.groups()
wind_text = f"Wind: {direction}° at {speed} KT"
if gust:
wind_text += f", gusting to {gust[1:]} KT"
decoded.append(wind_text)
# Decode visibility
visibility = visibility_re.search(metar)
if visibility:
visibility_text = f"Visibility: {visibility.group(1)} meters"
decoded.append(visibility_text)
# Decode temperature and dew point
temp_dew = temp_dew_re.search(metar)
if temp_dew:
temp, dew = temp_dew.groups()
temp_text = f"Temperature: {temp.replace('M', '-') if 'M' in temp else temp}°C, Dew Point: {dew.replace('M', '-') if 'M' in dew else dew}°C"
decoded.append(temp_text)
# Decode pressure
pressure = pressure_re.search(metar)
if pressure:
unit, value = pressure.groups()
if unit == 'QNH':
pressure_text = f"Pressure: {value} hPa"
else: # assuming 'A' is used for inHg
pressure_text = f"Pressure: {value[:2]}.{value[2:]} inHg"
decoded.append(pressure_text)
# Decode weather phenomena
weather = weather_re.search(metar)
if weather and any(weather.groups()):
intensity, descriptor, phenomenon = weather.groups()
weather_text = "Weather: "
if intensity:
weather_text += f"{intensity} "
if descriptor:
weather_text += f"{descriptor} "
if phenomenon:
weather_text += f"{phenomenon}"
decoded.append(weather_text.strip())
# Decode cloud cover
cloud_matches = cloud_re.findall(metar)
if cloud_matches:
cloud_text = "Clouds: " + ", ".join([f"{amount} at {int(height) * 100} feet" for amount, height in cloud_matches])
decoded.append(cloud_text)
decoded_str = "\n".join(decoded) if decoded else "No significant weather observed."
return {"decoded": decoded_str, "observation_time": observation_time}
# Run the bot
bot.run(TOKEN)