# NOTAMs

NOTAMs integrieren — aktive Meldungen, Gültigkeitszeiten und strukturierte Zuordnung der Felder.



Ausführbare Beispiele für die Arbeit mit [NOTAMs](https://www.icao.int/safety/pages/notam-service-profiles.aspx): Zusammenfassungen aktiver Meldungen, Filterung nach Flugplatz und Vorflugprüfungen für mehrere Flughäfen. Feldbeschreibungen finden Sie in der Referenz zu [NOTAMs](/docs/v31/notams).

| # | Frage                                                                  | Verwendete Daten     |
| - | ---------------------------------------------------------------------- | -------------------- |
| 1 | Alle aktiven NOTAMs für KJFK abrufen und eine Zusammenfassung anzeigen | `GET /notams/{icao}` |
| 2 | Pistenbezogene NOTAMs (Geltungsbereich Flugplatz) in EGLL filtern      | `GET /notams/{icao}` |
| 3 | NOTAM-Prüfung vor dem Flug: Abflug- und Zielflughafen                  | `GET /notams/{icao}` |

<DocRequirements>
  <DocRequirementAuth />
</DocRequirements>

### Installation [#install]

Die folgenden Skripte verwenden `requests` für API-Aufrufe und `rich` für die Terminalausgabe. Einmalig installieren:

```bash
pip install requests rich
```

***

### 1. Zusammenfassung aktiver NOTAMs für KJFK [#1-active-notam-summary-for-kjfk]

**Frage:** Welche [NOTAMs](https://www.icao.int/safety/pages/notam-service-profiles.aspx) sind derzeit in KJFK aktiv?

[NOTAMs](https://www.icao.int/safety/pages/notam-service-profiles.aspx) werden häufig aktualisiert — cachen Sie sie höchstens 5 Minuten lang. Dieses Skript ruft alle aktiven Meldungen ab und zeigt eine sortierbare Übersichtstabelle an.

```python
import os
from datetime import datetime

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"


def fetch_notams(icao: str) -> dict | None:
    r = requests.get(
        f"{BASE}/notams/{icao}",
        headers=HEADERS,
        timeout=(10, 15),
    )
    if r.status_code == 404:
        return None
    r.raise_for_status()
    return r.json()


def format_dt(val: str | None) -> str:
    if not val:
        return "—"
    try:
        return datetime.fromisoformat(val).strftime("%m/%d %H:%Mz")
    except ValueError:
        return val[:16]


def main() -> None:
    data = fetch_notams("KJFK")
    if not data:
        console.print("[red]No NOTAM data returned for KJFK.[/red]")
        return

    notams = data.get("notams", [])
    total = data.get("total", len(notams))

    sorted_notams = sorted(
        notams,
        key=lambda n: n.get("effective") or "",
    )

    table = Table(
        title=f"KJFK NOTAMs — {total} active",
        show_lines=True,
    )
    table.add_column("NOTAM ID", style="cyan", no_wrap=True)
    table.add_column("Type", justify="center")
    table.add_column("Scope", justify="center")
    table.add_column("Effective", no_wrap=True)
    table.add_column("Expiration", no_wrap=True)
    table.add_column("Body")

    for n in sorted_notams:
        body = n.get("body") or n.get("raw", "")
        snippet = (body[:60] + "…") if len(body) > 60 else body
        table.add_row(
            n.get("notam_id") or n.get("notam_id_domestic") or "—",
            n.get("type") or "—",
            n.get("scope") or "—",
            format_dt(n.get("effective")),
            format_dt(n.get("expiration")),
            snippet,
        )

    console.print(table)


if __name__ == "__main__":
    main()
```

**Ausgabe:** Eine Tabelle aller aktiven [NOTAMs](https://www.icao.int/safety/pages/notam-service-profiles.aspx), sortiert nach Beginn der Gültigkeit, mit ID, Typ, Geltungsbereich, Gültigkeitszeitraum und einer Vorschau des Textes mit 60 Zeichen.

***

### 2. Flugplatz-NOTAMs in EGLL [#2-aerodrome-notams-at-egll]

**Frage:** Welche [NOTAMs](https://www.icao.int/safety/pages/notam-service-profiles.aspx) in EGLL haben den Geltungsbereich Flugplatz (Sperrungen von Pisten/Rollwegen, Ausfälle der Befeuerung)?

Flugplatz-[NOTAMs](https://www.icao.int/safety/pages/notam-service-profiles.aspx) (`scope == "AERODROME"`) wirken sich direkt auf den Bodenbetrieb aus. Filtern Sie vor dem Rollbriefing nach diesen Meldungen. Sie können Einträge mit `scope: "FIR"` auch serverseitig mit `?exclude_scope=FIR` ausschließen, statt clientseitig zu filtern.

```python
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_notams(icao: str) -> dict | None:
    r = requests.get(
        f"{BASE}/notams/{icao}",
        headers=HEADERS,
        timeout=(10, 15),
    )
    if r.status_code == 404:
        return None
    r.raise_for_status()
    return r.json()


def main() -> None:
    data = fetch_notams("EGLL")
    if not data:
        console.print("[red]No NOTAM data returned for EGLL.[/red]")
        return

    all_notams = data.get("notams", [])
    aerodrome = [n for n in all_notams if n.get("scope") == "AERODROME"]

    if not aerodrome:
        console.print("[yellow]No aerodrome NOTAMs active at EGLL.[/yellow]")
        return

    console.print(
        Panel(
            f"[bold]{len(aerodrome)}[/bold] aerodrome-scoped NOTAM(s) at EGLL "
            f"(scope=AERODROME: runway/taxiway closures, lighting outages)",
            style="yellow",
        )
    )

    table = Table(show_lines=True)
    table.add_column("NOTAM ID", style="cyan", no_wrap=True)
    table.add_column("Type", justify="center")
    table.add_column("Q-Code")
    table.add_column("Body")

    for n in aerodrome:
        body = n.get("body") or n.get("raw", "")
        snippet = (body[:80] + "…") if len(body) > 80 else body
        table.add_row(
            n.get("notam_id") or n.get("notam_id_domestic") or "—",
            n.get("type") or "—",
            n.get("q_code") or "—",
            snippet,
        )

    console.print(table)


if __name__ == "__main__":
    main()
```

**Ausgabe:** Ein Panel mit der Anzahl der Flugplatz-[NOTAMs](https://www.icao.int/safety/pages/notam-service-profiles.aspx), gefolgt von einer Tabelle mit ID, Typ, Q-Code und gekürztem Text. Gibt „No aerodrome [NOTAMs](https://www.icao.int/safety/pages/notam-service-profiles.aspx) active“ aus, wenn keine Meldung zutrifft.

***

### 3. NOTAM-Prüfung vor dem Flug: Abflug und Ankunft [#3-pre-flight-notam-check-departure-and-arrival]

**Frage:** Welche [NOTAMs](https://www.icao.int/safety/pages/notam-service-profiles.aspx) sind sowohl an meinem Abflughafen (KJFK) als auch an meinem Zielflughafen (KLAX) aktiv?

Rufen Sie beide Flughäfen parallel ab, um die Latenz zu minimieren. Prüfen Sie alle [NOTAMs](https://www.icao.int/safety/pages/notam-service-profiles.aspx) — verlassen Sie sich nicht allein auf die automatische Auswertung.

```python
import os
import concurrent.futures

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 = "KJFK"
ARRIVAL = "KLAX"


def fetch_notams(icao: str) -> tuple[str, dict | None]:
    try:
        r = requests.get(
            f"{BASE}/notams/{icao}",
            headers=HEADERS,
            timeout=(10, 15),
        )
        if r.status_code == 404:
            return icao, None
        r.raise_for_status()
        return icao, r.json()
    except Exception as exc:  # noqa: BLE001
        console.print(f"[red]Error fetching {icao}: {exc}[/red]")
        return icao, None


def print_airport_notams(icao: str, data: dict | None) -> None:
    if not data:
        console.print(Panel(f"[red]No data for {icao}[/red]"))
        return

    notams = data.get("notams", [])
    total = data.get("total", len(notams))

    console.print(
        Panel(
            f"[bold cyan]{icao}[/bold cyan] — {total} active NOTAM(s)",
            style="blue",
        )
    )

    for n in notams[:5]:
        body = n.get("body") or n.get("raw", "")
        notam_id = n.get("notam_id") or n.get("notam_id_domestic") or "—"
        scope = n.get("scope") or "—"
        snippet = (body[:80] + "…") if len(body) > 80 else body
        console.print(f"  [cyan]{notam_id}[/cyan] [{scope}] {snippet}")

    if total > 5:
        console.print(f"  [dim]… and {total - 5} more NOTAM(s)[/dim]")

    console.print()


def main() -> None:
    airports = [DEPARTURE, ARRIVAL]

    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
        results = dict(pool.map(lambda a: fetch_notams(a), airports))

    for icao in airports:
        print_airport_notams(icao, results.get(icao))

    console.print(
        "[bold yellow]⚠ Review all NOTAMs — do not rely on automated "
        "parsing alone.[/bold yellow]"
    )


if __name__ == "__main__":
    main()
```

**Ausgabe:** Zwei Abschnitte — einer pro Flughafen —, die jeweils die Gesamtzahl der [NOTAMs](https://www.icao.int/safety/pages/notam-service-profiles.aspx) und die ersten 5 Meldungen als kompakte Liste zeigen. Am Ende wird eine Erinnerung ausgegeben, alle [NOTAMs](https://www.icao.int/safety/pages/notam-service-profiles.aspx) manuell zu prüfen.