Исторические треки

Готовые к запуску примеры поиска архивных рейсов, получения полных сведений о рейсе и запроса исторического трафика аэропорта. Сроки хранения данных и требования к тарифу см. в полном справочнике Исторические данные ADS-B.

#	Вопрос	Используемые данные
1	Какими были последние 5 рейсов борта с регистрационным номером N636JB?	`/ultra/history/flights`
2	Полные метаданные последнего вылета из EGLL	`/ultra/history/flights` → `/ultra/history/flight/{flight_id}`
3	Сколько прилётов принял KJFK за последние 24 часа?	`/ultra/history/airport/{icao}/traffic`

Требования

Авторизация

x-api-key в каждом запросе (прямая подписка). Подробнее →

Тариф

Для /ultra/history/... требуется тариф Pro, Ultra или Mega — 401/403 означает, что ваш тариф ниже необходимого уровня.

Подробнее →

Установка

Приведённые ниже скрипты используют requests для вызовов API и rich для вывода в терминал. Установите их один раз:

pip install requests rich

1. Последние 5 рейсов по регистрационному номеру

Вопрос: по каким маршрутам N636JB летал в последнее время и сколько длился каждый полёт?

Укажите фильтр registration и задайте limit=5, чтобы получить только самые последние архивные рейсы.

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"

REGISTRATION = "N636JB"


def fetch_recent_flights(registration: str, limit: int = 5) -> list[dict] | None:
    r = requests.get(
        f"{BASE}/ultra/history/flights",
        headers=HEADERS,
        params={"registration": registration, "limit": limit},
        timeout=(10, 25),
    )
    if r.status_code in (401, 403):
        msg = r.json().get("message", "Plan not entitled.")
        console.print(
            f"[red]Access denied:[/red] {msg}\n"
            "[yellow]Upgrade to Pro or above for /ultra/history/... endpoints.[/yellow]"
        )
        return None
    r.raise_for_status()
    return r.json().get("flights", [])


def fmt_duration(seconds: int | None) -> str:
    if seconds is None:
        return "—"
    h, m = divmod(seconds // 60, 60)
    return f"{h}:{m:02d}"


def main() -> None:
    flights = fetch_recent_flights(REGISTRATION)
    if flights is None:
        return

    if not flights:
        console.print(f"[yellow]No archived flights found for {REGISTRATION}.[/yellow]")
        return

    table = Table(title=f"Recent Flights — {REGISTRATION}", border_style="cyan")
    table.add_column("Flight ID", style="dim")
    table.add_column("From")
    table.add_column("To")
    table.add_column("Departure (UTC)")
    table.add_column("Duration", justify="right")

    for f in flights:
        table.add_row(
            (f.get("flight_id") or "")[:8],
            f.get("departure_icao") or "—",
            f.get("arrival_icao") or "—",
            (f.get("departure_time") or "—")[:16],
            fmt_duration(f.get("duration_sec")),
        )

    console.print(table)


if __name__ == "__main__":
    main()

Результат: таблица из пяти последних архивных рейсов: ID рейса (первые 8 символов) | откуда | куда | время вылета | продолжительность (ч:мм).

2. Цепочка «поиск → сведения»: полные метаданные последнего вылета из EGLL

Вопрос: какие данные о воздушном судне, маршруте и времени описывают последний рейс, вылетевший из лондонского Хитроу?

Цепочка из двух шагов: найдите последний вылет из EGLL, чтобы получить flight_id, затем запросите полную запись со сведениями, включая тип воздушного судна и признак подмены GPS-сигнала (GPS spoofing).

import os

import requests
from rich.console import Console
from rich.panel import Panel

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"

DEPARTURE_ICAO = "EGLL"


def handle_plan_gate(r: requests.Response) -> bool:
    """Return True if request was blocked by plan gating."""
    if r.status_code in (401, 403):
        msg = r.json().get("message", "Plan not entitled.")
        console.print(
            f"[red]Access denied:[/red] {msg}\n"
            "[yellow]Upgrade to Pro or above for /ultra/history/... endpoints.[/yellow]"
        )
        return True
    return False


def fetch_latest_departure(departure_icao: str) -> str | None:
    r = requests.get(
        f"{BASE}/ultra/history/flights",
        headers=HEADERS,
        params={"departure_icao": departure_icao, "limit": 1},
        timeout=(10, 25),
    )
    if handle_plan_gate(r):
        return None
    r.raise_for_status()
    flights = r.json().get("flights", [])
    if not flights:
        return None
    return flights[0]["flight_id"]


def fetch_flight_detail(flight_id: str) -> dict | None:
    r = requests.get(
        f"{BASE}/ultra/history/flight/{flight_id}",
        headers=HEADERS,
        timeout=(10, 25),
    )
    if r.status_code == 404:
        return None
    if handle_plan_gate(r):
        return None
    r.raise_for_status()
    return r.json()


def fmt_duration(seconds: int | None) -> str:
    if seconds is None:
        return "—"
    h, m = divmod(seconds // 60, 60)
    return f"{h}h {m:02d}m"


def main() -> None:
    flight_id = fetch_latest_departure(DEPARTURE_ICAO)
    if not flight_id:
        console.print(f"[yellow]No recent departures found for {DEPARTURE_ICAO}.[/yellow]")
        return

    detail = fetch_flight_detail(flight_id)
    if detail is None:
        console.print(f"[yellow]Flight detail not available for ID {flight_id[:8]}.[/yellow]")
        return

    spoofing = detail.get("gps_spoofing_suspected")
    spoofing_str = "[red]YES — review track[/red]" if spoofing else "No"

    lines = [
        f"[bold]Registration:[/bold]    {detail.get('registration') or '—'}",
        f"[bold]Callsign:[/bold]        {detail.get('callsign') or '—'}",
        f"[bold]Aircraft Type:[/bold]   {detail.get('aircraft_type') or '—'}",
        f"[bold]Route:[/bold]           {detail.get('departure_icao') or '—'}  →  {detail.get('arrival_icao') or '—'}",
        f"[bold]Departure:[/bold]       {detail.get('departure_airport_name') or detail.get('departure_icao') or '—'}",
        f"[bold]Arrival:[/bold]         {detail.get('arrival_airport_name') or detail.get('arrival_icao') or '—'}",
        f"[bold]Dep time (UTC):[/bold]  {(detail.get('departure_time') or '—')[:16]}",
        f"[bold]Arr time (UTC):[/bold]  {(detail.get('arrival_time') or '—')[:16]}",
        f"[bold]Duration:[/bold]        {fmt_duration(detail.get('duration_sec'))}",
        f"[bold]Distance:[/bold]        {detail.get('distance_nm') or '—'} nm",
        f"[bold]GPS Spoofing:[/bold]    {spoofing_str}",
    ]
    console.print(
        Panel("\n".join(lines), title=f"Flight Detail — {flight_id[:8]}", border_style="cyan")
    )


if __name__ == "__main__":
    main()

Результат: панель со сведениями: регистрационный номер, позывной, тип ВС, полный маршрут (названия аэропортов), время вылета и прилёта, продолжительность, расстояние и признак подмены GPS-сигнала.

3. Прилёты в KJFK за последние 24 часа

Вопрос: сколько рейсов прибыло в JFK за последние 24 часа и какими были первые пять?

Вычислите start и end в UTC и запросите эндпоинт трафика аэропорта с direction=arr. Учтите, что доступное окно ограничено сроком хранения данных на вашем тарифе (90 дней для пути Ultra).

import datetime
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"

AIRPORT = "KJFK"


def fetch_airport_traffic(
    icao: str,
    start: str,
    end: str,
    direction: str = "arr",
) -> dict | None:
    r = requests.get(
        f"{BASE}/ultra/history/airport/{icao}/traffic",
        headers=HEADERS,
        params={"start": start, "end": end, "direction": direction, "limit": 50},
        timeout=(10, 25),
    )
    if r.status_code in (401, 403):
        msg = r.json().get("message", "Plan not entitled.")
        console.print(
            f"[red]Access denied:[/red] {msg}\n"
            "[yellow]Upgrade to Pro or above for /ultra/history/... endpoints.[/yellow]"
        )
        return None
    r.raise_for_status()
    return r.json()


def main() -> None:
    now = datetime.datetime.now(datetime.timezone.utc)
    start = (now - datetime.timedelta(hours=24)).strftime("%Y-%m-%dT%H:%M:%SZ")
    end = now.strftime("%Y-%m-%dT%H:%M:%SZ")

    result = fetch_airport_traffic(AIRPORT, start=start, end=end, direction="arr")
    if result is None:
        return

    flights = result.get("flights", [])
    total = result.get("count", len(flights))

    console.print(
        f"[bold cyan]{AIRPORT}[/bold cyan] received "
        f"[bold]{total:,}[/bold] arrivals in the last 24 hours."
    )

    if not flights:
        console.print("[yellow]No individual flight records returned.[/yellow]")
        return

    table = Table(title=f"Sample Arrivals — {AIRPORT} (first 5)", border_style="cyan")
    table.add_column("Callsign")
    table.add_column("From")
    table.add_column("Arrival time (UTC)")

    for f in flights[:5]:
        table.add_row(
            f.get("callsign") or "—",
            f.get("departure_airport_icao") or "—",
            (f.get("landing_time") or "—")[:16],
        )

    console.print(table)
    console.print(
        "[dim]Window is subject to your plan's retention limit "
        "(Ultra path: up to 90 days).[/dim]"
    )


if __name__ == "__main__":
    main()

Результат: общее число прилётов за 24-часовое окно, а затем таблица-выборка из первых пяти прилётов: позывной | откуда | время прилёта.

Установка

1. Последние 5 рейсов по регистрационному номеру

2. Цепочка «поиск → сведения»: полные метаданные последнего вылета из EGLL

3. Прилёты в KJFK за последние 24 часа

На этой странице