fr24.py

"""Thin client for the official FlightRadar24 API (fr24api.flightradar24.com).

Docs: https://fr24api.flightradar24.com/docs/getting-started
Auth is via a Bearer token supplied in the FR24_API_TOKEN env var.
"""
from __future__ import annotations

import datetime as dt
import os
import time
from functools import lru_cache

import requests

BASE = "https://fr24api.flightradar24.com"


class FR24Error(RuntimeError):
    pass


def _headers() -> dict:
    token = os.environ.get("FR24_API_TOKEN", "").strip()
    if not token:
        raise FR24Error(
            "FR24_API_TOKEN is not set. Put your FlightRadar24 API token in the "
            "environment (see .env.example)."
        )
    return {
        "Accept": "application/json",
        "Accept-Version": os.environ.get("FR24_API_VERSION", "v1"),
        "Authorization": f"Bearer {token}",
    }


def _clamp_bounds(north, south, west, east):
    """FR24 caps the queryable area; keep the box sane."""
    north = max(-90.0, min(90.0, float(north)))
    south = max(-90.0, min(90.0, float(south)))
    west = max(-180.0, min(180.0, float(west)))
    east = max(-180.0, min(180.0, float(east)))
    if north < south:
        north, south = south, north
    return north, south, west, east


FLIGHT_POSITIONS_URL = f"{BASE}/api/live/flight-positions/full"


def query_positions(params: dict, *, timeout=25):
    """Low-level call to live/flight-positions/full with arbitrary FR24 filters.

    Returns (data_list, request_url). At least one filter param is required by
    FR24 (bounds, routes, airports, callsigns, ...).
    """
    resp = requests.get(FLIGHT_POSITIONS_URL, headers=_headers(),
                        params=params, timeout=timeout)
    if resp.status_code == 400:
        raise FR24Error(f"FR24 bad request (400): {resp.text[:300]}")
    if resp.status_code == 401:
        raise FR24Error("FR24 rejected the token (401). Check FR24_API_TOKEN.")
    if resp.status_code == 402:
        raise FR24Error("FR24 says payment/credits required (402).")
    if resp.status_code == 429:
        raise FR24Error("FR24 rate limit hit (429). Slow down the refresh.")
    if resp.status_code >= 400:
        raise FR24Error(f"FR24 error {resp.status_code}: {resp.text[:300]}")

    payload = resp.json()
    data = payload.get("data", payload if isinstance(payload, list) else [])
    return data, resp.url


def live_positions(north, south, west, east, *, extra_params=None, timeout=25):
    """Return a list of live flight dicts inside the bounding box."""
    north, south, west, east = _clamp_bounds(north, south, west, east)
    params = {"bounds": f"{north},{south},{west},{east}"}
    if extra_params:
        params.update(extra_params)
    data, _ = query_positions(params, timeout=timeout)
    return data


def search_route(origin: str, destination: str, *, limit=200):
    """Live flights on a route, e.g. origin='LHR', destination='JFK'."""
    params = {"routes": f"{origin.strip().upper()}-{destination.strip().upper()}",
              "limit": limit}
    return query_positions(params)


def search_airport(airport: str, direction: str = "both", *, limit=200):
    """Live flights to/from an airport. direction in {inbound,outbound,both}."""
    direction = (direction or "both").strip().lower()
    if direction not in {"inbound", "outbound", "both"}:
        direction = "both"
    params = {"airports": f"{direction}:{airport.strip().upper()}", "limit": limit}
    return query_positions(params)


def airport_coords(code: str):
    """lat/lon for an IATA/ICAO code via the bundled OpenFlights table.

    (FR24's airport endpoints don't return coordinates on this plan, so this is
    resolved locally — fast and reliable.)
    """
    import airports
    return airports.coords(code)


def parse_eta(eta_value):
    """FR24 'eta' is an ISO timestamp. Return (iso_string, seconds_remaining)."""
    if not eta_value:
        return None, None
    try:
        if isinstance(eta_value, (int, float)):
            target = dt.datetime.fromtimestamp(eta_value, tz=dt.timezone.utc)
        else:
            s = str(eta_value).replace("Z", "+00:00")
            target = dt.datetime.fromisoformat(s)
            if target.tzinfo is None:
                target = target.replace(tzinfo=dt.timezone.utc)
        now = dt.datetime.now(tz=dt.timezone.utc)
        remaining = (target - now).total_seconds()
        return target.isoformat(), remaining
    except Exception:
        return str(eta_value), None


def human_duration(seconds):
    if seconds is None:
        return "—"
    seconds = int(max(0, seconds))
    h, rem = divmod(seconds, 3600)
    m, _ = divmod(rem, 60)
    if h:
        return f"{h}h {m:02d}m"
    return f"{m}m"