В аэропорту
Готовые к запуску примеры работы с погодой на уровне аэропорта: METAR и TAF для аэропортов вылета, назначения и запасных. Запросы с bbox для маршрута (ветер на высотах, PIREP, AIRMET) см. в разделе На маршруте.
| # | Вопрос | Используемые данные |
|---|---|---|
| 1 | Какова текущая категория полётных условий в крупных аэропортах США прямо сейчас? | METAR |
| 2 | Превышает ли высота по плотности высоту ВПП больше чем на заданный вами порог? | METAR |
| 3 | Какой будет погода в момент моего вылета (в ближайшие несколько часов)? | TAF |
| 4 | Нью-Йорк → Лос-Анджелес — погода при взлёте и посадке: сейчас и по прогнозу? | METAR + TAF |
| 5 | Сводка текущих METAR в выбранных международных аэропортах | METAR |
| 6 | Проходит ли текущий METAR пользовательский чек-лист по ветру, видимости и нижней границе облаков? | METAR |
Требования
x-api-key в каждом запросе (прямая подписка). Подробнее →Установка
Скрипты ниже используют requests для вызовов API, rich для таблиц в терминале и plotly + pandas для карты категорий полётных условий. Установите один раз:
pip install requests rich plotly pandas1. Карта текущих категорий полётных условий (METAR в реальном времени)
Вопрос: какова текущая категория полётных условий по данным METAR в четырёх крупных аэропортах США?
METAR возвращает категорию полётных условий (VFR / MVFR / IFR / LIFR). Этот скрипт отмечает каждый аэропорт на карте США цветными подписями согласно таблице категорий полётных условий.
import os
import webbrowser
from pathlib import Path
import pandas as pd
import plotly.express as px
import requests
HEADERS = {
"X-RapidAPI-Key": os.getenv("RAPIDAPI_KEY", "YOUR_RAPIDAPI_KEY"),
"X-RapidAPI-Host": "skylink-api.p.rapidapi.com",
}
BASE = "https://skylink-api.p.rapidapi.com"
# Demo airports: code → (latitude, longitude, city name)
AIRPORTS = {
"KJFK": (40.6413, -73.7781, "New York (JFK)"),
"KLAX": (33.9425, -118.4081, "Los Angeles"),
"KORD": (41.9742, -87.9073, "Chicago"),
"KDEN": (39.8561, -104.6737, "Denver"),
}
# Flight category → color + label (see METAR reference)
CONDITIONS = {
"VFR": ("#22c55e", "VFR - visual flight rules"),
"MVFR": ("#3b82f6", "MVFR - marginal VFR"),
"IFR": ("#ef4444", "IFR - instrument flight rules"),
"LIFR": ("#a855f7", "LIFR - low IFR"),
}
def fetch_metar(icao: str) -> dict | None:
r = requests.get(
f"{BASE}/weather/metar/{icao}",
headers=HEADERS,
params={"parsed": "true"},
timeout=(10, 15),
)
if r.status_code == 404:
return None
r.raise_for_status()
return r.json()
def build_map() -> Path:
rows = []
for icao, (lat, lon, city) in AIRPORTS.items():
metar = fetch_metar(icao)
if not metar:
continue
p = metar["parsed"]
code = p["flight_rules"]
_color, label = CONDITIONS.get(code, ("#6b7280", code))
wind = p["wind"]["speed"]
rows.append({
"city": city,
"code": icao,
"lat": lat,
"lon": lon,
"rating_code": code,
"conditions": label,
"wind": f"{wind} kt (~{round(wind * 1.15)} mph, ~{round(wind * 1.852)} km/h)",
"map_label": f"{city}: {label}",
})
df = pd.DataFrame(rows)
color_map = {k: v[0] for k, v in CONDITIONS.items()}
fig = px.scatter_geo(
df,
lat="lat",
lon="lon",
color="rating_code",
color_discrete_map=color_map,
hover_name="city",
hover_data={
"conditions": True,
"wind": True,
"code": True,
"rating_code": False,
"lat": False,
"lon": False,
"map_label": False,
},
scope="usa",
title="Current flight category by airport (METAR)",
labels={"rating_code": "API code"},
)
fig.update_traces(
marker={"size": 18, "line": {"width": 2, "color": "white"}},
text=df["map_label"],
textposition="top center",
)
fig.add_annotation(
text="Green = VFR | Blue = MVFR | Red/magenta = IFR/LIFR",
xref="paper", yref="paper", x=0.5, y=-0.08, showarrow=False,
font={"size": 12},
)
out = Path("weather_map.html")
fig.write_html(out, include_plotlyjs="cdn")
return out
if __name__ == "__main__":
path = build_map()
print(f"Saved {path} - opening interactive map in your browser.")
print("Hover a marker for wind and category. Colors follow METAR flight_rules.")
webbrowser.open(path.resolve().as_uri())Результат: HTML-карта с четырьмя аэропортами США, окрашенными по flight_rules, со всплывающими данными о ветре и подписью категории.
2. Высота по плотности и высота ВПП
Вопрос: превышает ли parsed.density_altitude в этом аэропорту высоту ВПП больше чем на заданный вами порог?
METAR содержит высоту по плотности (насколько разрежённым ощущается воздух). Сравните её с высотой ВПП: большая разница означает более длинный разбег при взлёте.
import os
import requests
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
console = Console()
HEADERS = {
"X-RapidAPI-Key": os.getenv("RAPIDAPI_KEY", "YOUR_RAPIDAPI_KEY"),
"X-RapidAPI-Host": "skylink-api.p.rapidapi.com",
}
BASE = "https://skylink-api.p.rapidapi.com"
def fetch_metar(icao: str) -> dict:
r = requests.get(
f"{BASE}/weather/metar/{icao}",
headers=HEADERS,
params={"parsed": "true"},
timeout=(10, 15),
)
r.raise_for_status()
return r.json()
def takeoff_performance_report(
icao: str,
runway_elevation_ft: int,
concern_if_da_above_runway_ft: int = 1000,
) -> None:
metar = fetch_metar(icao)
name = metar.get("airport_name", icao)
p = metar["parsed"]
da = p["density_altitude"]
gap = da - runway_elevation_ft
hot_day = gap >= concern_if_da_above_runway_ft
table = Table(title=f"Takeoff air density - {name} ({icao})")
table.add_column("What", style="cyan")
table.add_column("Value", justify="right")
table.add_row("Runway elevation", f"{runway_elevation_ft} ft above sea level")
table.add_row("Air feels like elevation", f"{da} ft (density altitude)")
table.add_row("Difference", f"+{gap} ft - air is thinner than runway height suggests")
table.add_row("Outside temperature", f"{p['temperature']} C / {round(p['temperature'] * 9/5 + 32)} F")
table.add_row("Barometer", f"{p['altimeter']} inHg")
console.print(table)
if hot_day:
msg = (
f"The air is significantly thinner than normal for this runway ({gap} ft gap). "
"Planes need more runway to lift off and climb slower. "
"Check your aircraft handbook before assuming normal takeoff distance."
)
console.print(Panel(msg, title="WARN: possible reduced takeoff performance", style="red"))
else:
msg = (
f"Density altitude is within {concern_if_da_above_runway_ft} ft of runway elevation. "
"Takeoff performance is likely close to normal (still verify for your aircraft)."
)
console.print(Panel(msg, title="OK: no major thin-air warning for this threshold", style="green"))
if __name__ == "__main__":
# JFK runway is almost at sea level (~13 ft)
takeoff_performance_report("KJFK", runway_elevation_ft=13)Результат: таблица со сравнением высоты ВПП, высоты по плотности и температуры; панель предупреждает, если разница превышает порог.
3. Какой будет погода в момент моего вылета?
Вопрос: я планирую взлететь из Чикаго в текущее время UTC — каких условий ожидать?
TAF — это временная шкала прогнозных блоков. Наиболее вероятные условия = основной блок FROM; возможные кратковременные периоды = перекрывающиеся строки TEMPO / BECMG.
import os
from datetime import datetime, timezone
import requests
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
console = Console()
HEADERS = {
"X-RapidAPI-Key": os.getenv("RAPIDAPI_KEY", "YOUR_RAPIDAPI_KEY"),
"X-RapidAPI-Host": "skylink-api.p.rapidapi.com",
}
BASE = "https://skylink-api.p.rapidapi.com"
CONDITIONS = {
"VFR": "VFR - visual flight rules",
"MVFR": "MVFR - marginal VFR",
"IFR": "IFR - instrument flight rules",
"LIFR": "LIFR - low IFR",
}
LAYER_MEANING = {
"FROM": "Most likely for this time window",
"TEMPO": "Possible brief worse spell (may not happen)",
"BECMG": "Conditions gradually changing toward",
}
def parse_time(iso: str) -> datetime:
return datetime.fromisoformat(iso.replace("Z", "+00:00"))
def fetch_taf(icao: str) -> dict | None:
r = requests.get(
f"{BASE}/weather/taf/{icao}",
headers=HEADERS,
params={"parsed": "true"},
timeout=(10, 15),
)
if r.status_code == 404:
return None
r.raise_for_status()
return r.json()
def print_departure_forecast(icao: str, departure: datetime) -> None:
taf = fetch_taf(icao)
if not taf:
console.print(f"[red]No forecast (TAF) published for {icao} - common at small airfields.[/]")
return
name = taf.get("airport_name", icao)
console.print(Panel(
f"Forecast airport: {name} ({icao})\n"
f"Departure time (UTC): {departure.strftime('%Y-%m-%d %H:%M')} Z\n"
f"Below = parsed TAF layers active at that instant.",
title="Question: forecast at departure time",
))
forecast = taf["parsed"]["forecast"]
baseline = next(
(
p for p in forecast
if p["type"] == "FROM"
and parse_time(p["start_time"]) <= departure <= parse_time(p["end_time"])
),
None,
)
modifiers = [
p for p in forecast
if p is not baseline
and parse_time(p["start_time"]) <= departure <= parse_time(p["end_time"])
]
table = Table(title="Forecast layers at your departure time")
table.add_column("Role", style="cyan")
table.add_column("Conditions")
table.add_column("Weather details")
if baseline:
code = baseline["flight_rules"]
wx = ", ".join(w["repr"] for w in baseline["wx_codes"]) or "No significant weather"
table.add_row(
LAYER_MEANING.get("FROM", "Most likely"),
CONDITIONS.get(code, code),
wx,
)
else:
table.add_row("Most likely", "-", "No matching forecast block for this time")
for p in modifiers:
code = p["flight_rules"]
wx = ", ".join(w["repr"] for w in p["wx_codes"]) or "-"
prob = (p.get("probability") or {}).get("value")
role = LAYER_MEANING.get(p["type"], p["type"])
if prob:
role += f" ({prob}% chance)"
table.add_row(role, CONDITIONS.get(code, code), wx)
console.print(table)
console.print(
"\n[dim]Show the baseline FROM row and any overlapping TEMPO/BECMG rows separately.[/dim]"
)
if __name__ == "__main__":
print_departure_forecast("KORD", datetime.now(timezone.utc))Результат: панель со временем вылета и таблица с действующей строкой FROM и перекрывающимися периодами-модификаторами.
4. Нью-Йорк → Лос-Анджелес — погода в обоих аэропортах?
Вопрос: я лечу по маршруту KJFK → KLAX — какая погода в аэропортах вылета и назначения прямо сейчас и каков наихудший прогноз на ближайшие ~24 часа?
import os
import concurrent.futures
import requests
from rich.console import Console
from rich.table import Table
console = Console()
HEADERS = {
"X-RapidAPI-Key": os.getenv("RAPIDAPI_KEY", "YOUR_RAPIDAPI_KEY"),
"X-RapidAPI-Host": "skylink-api.p.rapidapi.com",
}
BASE = "https://skylink-api.p.rapidapi.com"
CONDITIONS = {
"VFR": "VFR - visual flight rules",
"MVFR": "MVFR - marginal VFR",
"IFR": "IFR - instrument flight rules",
"LIFR": "LIFR - low IFR",
}
SEVERITY = {"VFR": 0, "MVFR": 1, "IFR": 2, "LIFR": 3}
STYLE = {"VFR": "green", "MVFR": "blue", "IFR": "red", "LIFR": "magenta"}
def fetch_metar(icao: str) -> dict | None:
r = requests.get(
f"{BASE}/weather/metar/{icao}",
headers=HEADERS,
params={"parsed": "true"},
timeout=(10, 15),
)
if r.status_code == 404:
return None
r.raise_for_status()
return r.json()
def fetch_taf(icao: str) -> dict | None:
r = requests.get(
f"{BASE}/weather/taf/{icao}",
headers=HEADERS,
params={"parsed": "true"},
timeout=(10, 15),
)
if r.status_code == 404:
return None
r.raise_for_status()
return r.json()
def worst_forecast_label(taf: dict | None) -> str:
if not taf:
return "No forecast available"
from_blocks = [p for p in taf["parsed"]["forecast"] if p["type"] == "FROM"]
if not from_blocks:
return "No forecast available"
worst = max(from_blocks, key=lambda p: SEVERITY.get(p["flight_rules"], -1))
code = worst["flight_rules"]
return CONDITIONS.get(code, code)
def print_route_weather(dep_icao: str, arr_icao: str) -> None:
with concurrent.futures.ThreadPoolExecutor() as ex:
metar_dep = ex.submit(fetch_metar, dep_icao).result()
taf_dep = ex.submit(fetch_taf, dep_icao).result()
metar_arr = ex.submit(fetch_metar, arr_icao).result()
taf_arr = ex.submit(fetch_taf, arr_icao).result()
table = Table(title=f"Route weather: {dep_icao} (departure) -> {arr_icao} (arrival)")
table.add_column("Airport", style="cyan")
table.add_column("Weather RIGHT NOW", max_width=36)
table.add_column("Worst forecast in next ~24 h", max_width=36)
for icao, metar, taf in [
(dep_icao, metar_dep, taf_dep),
(arr_icao, metar_arr, taf_arr),
]:
if metar:
code = metar["parsed"]["flight_rules"]
now = CONDITIONS.get(code, code)
style = STYLE.get(code, "white")
airport = metar.get("airport_name", icao)
else:
now, style, airport = "No observation", "white", icao
table.add_row(
f"{airport}\n({icao})",
f"[{style}]{now}[/]",
worst_forecast_label(taf),
)
console.print(table)
console.print(
"\n[dim]RIGHT NOW = live METAR observation. "
"Worst forecast = poorest 'most likely' TAF block in the next day.[/dim]"
)
if __name__ == "__main__":
print_route_weather("KJFK", "KLAX")Результат: две строки (вылет и прилёт): текущие flight_rules из METAR и наихудшая категория FROM из TAF.
5. Дашборд METAR по нескольким аэропортам
Вопрос: получить декодированный METAR для фиксированного списка аэропортов и вывести ветер, температуру, видимость и flight_rules.
import os
import requests
from rich.console import Console
from rich.table import Table
console = Console()
HEADERS = {
"X-RapidAPI-Key": os.getenv("RAPIDAPI_KEY", "YOUR_RAPIDAPI_KEY"),
"X-RapidAPI-Host": "skylink-api.p.rapidapi.com",
}
BASE = "https://skylink-api.p.rapidapi.com"
AIRPORTS = ["KJFK", "KLAX", "EGLL", "YSSY"] # New York, LA, London, Sydney
CONDITIONS = {
"VFR": "VFR - visual flight rules",
"MVFR": "MVFR - marginal VFR",
"IFR": "IFR - instrument flight rules",
"LIFR": "LIFR - low IFR",
}
STYLE = {"VFR": "green", "MVFR": "blue", "IFR": "red", "LIFR": "magenta"}
def fetch_metar(icao: str) -> dict | None:
r = requests.get(
f"{BASE}/weather/metar/{icao}",
headers=HEADERS,
params={"parsed": "true"},
timeout=(10, 15),
)
if r.status_code == 404:
return None
r.raise_for_status()
return r.json()
def visibility_text(parsed: dict) -> str:
v = parsed["visibility"]["value"]
repr_ = parsed["visibility"]["repr"]
if v is None:
return "More than 6 miles visibility"
# Some international stations return coded values (e.g. 9999 = 10 km+) - not statute miles.
if v > 50 or repr_ in ("9999", "P6"):
return f"{repr_} (coded - check raw METAR for units)"
return f"{v} miles visibility"
def print_live_dashboard(airports: list[str]) -> None:
table = Table(title="Live airport weather (METAR - updates every 30–60 min)")
table.add_column("Airport", style="cyan", max_width=28)
table.add_column("Flight category", max_width=30)
table.add_column("Wind")
table.add_column("Temp")
table.add_column("Visibility")
for icao in airports:
data = fetch_metar(icao)
if not data:
table.add_row(icao, "No data", "-", "-", "-")
continue
p = data["parsed"]
code = p["flight_rules"]
label = CONDITIONS.get(code, code)
style = STYLE.get(code, "white")
gust = p["wind"]["gust"]
wind = f"{p['wind']['speed']} kt"
if gust:
wind += f" (gusts {gust} kt)"
table.add_row(
data["airport_name"],
f"[{style}]{label}[/]",
wind,
f"{p['temperature']} C",
visibility_text(p),
)
console.print(table)
console.print(
"\n[dim]In a real dashboard, refresh every 5 minutes - "
"that is often enough and saves API quota.[/dim]"
)
if __name__ == "__main__":
print_live_dashboard(AIRPORTS)Результат: таблица в терминале с названием аэропорта, подписью flight_rules, ветром, температурой и видимостью.
6. Чек-лист личных минимумов (демо)
Вопрос: проходит ли текущий METAR проверку по настраиваемым ограничениям ветра, порывов, видимости и нижней границы облаков?
Не заменяет официальный метеобрифинг или нормативные минимумы.
import os
import requests
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
console = Console()
HEADERS = {
"X-RapidAPI-Key": os.getenv("RAPIDAPI_KEY", "YOUR_RAPIDAPI_KEY"),
"X-RapidAPI-Host": "skylink-api.p.rapidapi.com",
}
BASE = "https://skylink-api.p.rapidapi.com"
# Demo personal limits - change these to match your skills / aircraft
MY_LIMITS = {
"max_wind_kt": 25,
"max_gust_kt": 35,
"min_visibility_sm": 1.0,
"min_ceiling_ft": 500,
}
def fetch_metar(icao: str) -> dict:
r = requests.get(
f"{BASE}/weather/metar/{icao}",
headers=HEADERS,
params={"parsed": "true"},
timeout=(10, 15),
)
r.raise_for_status()
return r.json()
def print_safety_check(icao: str) -> None:
metar = fetch_metar(icao)
name = metar.get("airport_name", icao)
p = metar["parsed"]
code = p["flight_rules"]
limits = Table(title=f"Your demo limits (edit MY_LIMITS in the script)")
limits.add_column("Rule")
limits.add_column("Your limit", justify="right")
limits.add_row("Max steady wind", f"{MY_LIMITS['max_wind_kt']} kt")
limits.add_row("Max gusts", f"{MY_LIMITS['max_gust_kt']} kt")
limits.add_row("Min visibility", f"{MY_LIMITS['min_visibility_sm']} miles")
limits.add_row("Min cloud ceiling", f"{MY_LIMITS['min_ceiling_ft']} ft")
console.print(limits)
problems: list[str] = []
wind = p["wind"]["speed"]
if wind > MY_LIMITS["max_wind_kt"]:
problems.append(f"Wind too strong: {wind} kt (your max {MY_LIMITS['max_wind_kt']} kt)")
gust = p["wind"]["gust"]
if gust and gust > MY_LIMITS["max_gust_kt"]:
problems.append(f"Gusts too strong: {gust} kt (your max {MY_LIMITS['max_gust_kt']} kt)")
vis = p["visibility"]["value"]
if vis is not None and vis < MY_LIMITS["min_visibility_sm"]:
problems.append(
f"Visibility too low: {vis} miles (you need at least {MY_LIMITS['min_visibility_sm']} mi)"
)
ceiling_layers = [c for c in p["clouds"] if c["type"] in ("BKN", "OVC")]
if ceiling_layers:
ceiling_ft = ceiling_layers[0]["base"] * 100
if ceiling_ft < MY_LIMITS["min_ceiling_ft"]:
problems.append(
f"Clouds too low: base at {ceiling_ft} ft (you need at least {MY_LIMITS['min_ceiling_ft']} ft)"
)
safe = len(problems) == 0
if safe:
body = (
f"All demo checks passed at {name}.\n"
f"Current overall rating from METAR: {code}.\n"
"This does NOT replace a full weather briefing."
)
title = f"Within demo limits - {icao}"
style = "green"
else:
body = "Problems found:\n" + "\n".join(f"- {x}" for x in problems)
title = f"Exceeds demo limits - {icao}"
style = "red"
console.print(Panel(body, title=title, style=style))
if __name__ == "__main__":
print_safety_check("KJFK")Результат: таблица ограничений, затем панель со списком пройденных проверок или с каждым нарушенным правилом относительно MY_LIMITS.
Погода на маршруте (эндпоинты с bbox)
Вопросы 7–9 запрашивают данные в пределах ограничивающего прямоугольника (bounding box) вдоль вашего маршрута. Постройте прямоугольник по координатам аэропортов вылета и назначения (с запасом) или по путевым точкам. Пустой reports / total: 0 — нормальная ситуация, если PIREP не подавались или нет действующих предупреждений, — это не сбой API.
Вспомогательная функция, используемая ниже:
def route_bbox(
dep_lat: float, dep_lon: float,
arr_lat: float, arr_lon: float,
pad_deg: float = 2.0,
) -> str:
"""SW corner first, then NE - matches API bbox format."""
lat1 = min(dep_lat, arr_lat) - pad_deg
lat2 = max(dep_lat, arr_lat) + pad_deg
lon1 = min(dep_lon, arr_lon) - pad_deg
lon2 = max(dep_lon, arr_lon) + pad_deg
return f"{lat1},{lon1},{lat2},{lon2}"Координаты KJFK (40.6413, -73.7781) и KLAX (33.9425, -118.4081).
AIRMET/SIGMET
Активные AIRMET и SIGMET в пределах ограничивающего прямоугольника: IFR, турбулентность, обледенение и конвекция для планирования на маршруте.
На маршруте
Примеры интеграции погоды на маршруте — ветер на высотах, PIREP и AIRMET/SIGMET в пределах ограничивающего прямоугольника (bounding box).