Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions scripts/save_fixtures.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,9 +13,13 @@
import sys
from dataclasses import dataclass

from dotenv import load_dotenv

from jetset.config import AppConfig
from jetset.fetcher import AirLabsAdapter

load_dotenv()

FIXTURES_DIR = "tests/fixtures"


Expand Down
22 changes: 11 additions & 11 deletions src/jetset/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ class App:
_FETCH_POLL_SECONDS = 1.0
# Display rotates a sliding window of this many flights; 4 metric pages each.
WINDOW_SIZE = 5
PAGES_PER_FLIGHT = 4
PAGES_PER_FLIGHT = 5

class Frame(NamedTuple):
flight: Flight
Expand Down Expand Up @@ -93,16 +93,16 @@ def _current_frame(self) -> Frame | None:
return None

window = min(self.WINDOW_SIZE, n)
# Slide the window across all captured flights once per refresh interval
# so fresh flights trickle into the rotation between (infrequent) fetches.
# The slide cadence scales with the catch size (refresh / N).
slide_interval = self.config.refresh / n
since = self.last_fetch if self.last_fetch is not None else time.time()
elapsed = time.time() - since
window_start = int(elapsed / slide_interval) % n if slide_interval else 0

# Within the window, rotate one flight at a time through its metric pages.
flight_in_window = (self.frame // self.PAGES_PER_FLIGHT) % window
# Show a sliding window of flights, one at a time. Each flight gets
# cycles_per_flight * PAGES_PER_FLIGHT frames (e.g. 1 cycle = all 4
# metric pages once) before advancing to the next. After every flight
# in the current window has been shown, slide the window by 1 so fresh
# flights trickle in.
frames_per_flight = self.PAGES_PER_FLIGHT * self.config.cycles_per_flight
frames_per_window_cycle = frames_per_flight * window
cycle = self.frame // frames_per_window_cycle
window_start = cycle % max(1, n - window + 1) if n else 0
flight_in_window = (self.frame % frames_per_window_cycle) // frames_per_flight
metric_page = self.frame % self.PAGES_PER_FLIGHT

return self.Frame(flights[(window_start + flight_in_window) % n], metric_page)
Expand Down
11 changes: 6 additions & 5 deletions src/jetset/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,11 +5,12 @@

@dataclass(frozen=True)
class AppConfig:
# IAH
home_lat: float = 29.9931
home_lon: float = -95.3416
range: int = 200
pause: int = 2
# HEB
home_lat: float = 30.0689232902742
home_lon: float = -95.25178382821562
range: int = 100 # in km
pause: int = 5 # how long in seconds to wait between cycling through the pages of a flight
cycles_per_flight: int = 2 # how many times to show all details per flight
refresh: int = 2700 # 45 min — one AirLabs bbox call per refresh ≈ 960/month
api_source: str = "airlabs"
# Where airline logos are cached. A global path (not a user home) so the
Expand Down
7 changes: 5 additions & 2 deletions src/jetset/display.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ def load_logo(airline_code: str, logo_dir: Path) -> Image.Image | None:
path = logo_dir / f"{airline_code}.png"
try:
return Image.open(path)
except (FileNotFoundError, PermissionError):
except FileNotFoundError, PermissionError:
return None


Expand Down Expand Up @@ -69,6 +69,7 @@ def metrics_label(flight: Flight, page: int = 0) -> str:
1 = speed (e.g. "450kn")
2 = vertical rate (e.g. "▲1500ft/min", "▼1200ft/min", or "▬0ft/min")
3 = track (e.g. "270°W")
4 = distance from home (e.g. "32km away")

Returns an empty string if the relevant field is missing.
"""
Expand All @@ -87,6 +88,8 @@ def metrics_label(flight: Flight, page: int = 0) -> str:
marker = _LEVEL
data += f"{marker}{abs(rate)}ft/min"
elif page == 3 and flight.track is not None:
data += f"{int(flight.track)}\u00b0{_cardinal(flight.track)}"
data += f"{int(flight.track)}°{_cardinal(flight.track)}"
elif page == 4 and flight.distance_km is not None:
data += f"{round(flight.distance_km)}km away"

return data
49 changes: 43 additions & 6 deletions src/jetset/fetcher.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,20 @@ def _ms_to_ft_per_min(ms: float | None) -> int | None:
return round(ms * 196.850394) if ms is not None else None


def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
"""Great-circle distance between two points in km."""
r = 6371
dlat = math.radians(lat2 - lat1)
dlon = math.radians(lon2 - lon1)
a = (
math.sin(dlat / 2) ** 2
+ math.cos(math.radians(lat1))
* math.cos(math.radians(lat2))
* math.sin(dlon / 2) ** 2
)
return r * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))


class AirLabsAdapter(FlightAPI):
"""Single data source for the display.

Expand All @@ -43,9 +57,9 @@ class AirLabsAdapter(FlightAPI):
interval — one bbox call per refresh.
"""

def __init__(self) -> None:
def __init__(self, api_key: str | None = None) -> None:
self._api = RequestsAPI("https://airlabs.co/api/v9")
self._api_key = os.environ.get("AIRLABS_API_KEY")
self._api_key = api_key or os.environ.get("AIRLABS_API_KEY")

@staticmethod
def _bbox(lat: float, lon: float, range_nm: float) -> str:
Expand All @@ -55,18 +69,30 @@ def _bbox(lat: float, lon: float, range_nm: float) -> str:
return f"{lat - half_lat},{lon - half_lon},{lat + half_lat},{lon + half_lon}"

@staticmethod
def to_flight(data: dict) -> Flight:
def to_flight(
data: dict,
home_lat: float | None = None,
home_lon: float | None = None,
) -> Flight:
"""Map an AirLabs flight (metric units) to a Flight (feet/knots/ft-min)."""
dep, arr = data.get("dep_iata"), data.get("arr_iata")
route = FlightRoute(Airport(dep), Airport(arr)) if dep and arr else None
f_lat, f_lng = data.get("lat"), data.get("lng")
distance_km = (
_haversine_km(home_lat, home_lon, f_lat, f_lng)
if home_lat is not None and home_lon is not None
and f_lat is not None and f_lng is not None
else None
)
return Flight(
callsign=(data.get("flight_icao") or "").strip(),
aircraft=data.get("aircraft_icao"),
route=route,
altitude=_meters_to_feet(data.get("alt")),
speed=_kmh_to_knots(data.get("speed")),
track=data.get("dir"),
track=float(d) if (d := data.get("dir")) is not None else None,
vertical_rate=_ms_to_ft_per_min(data.get("v_speed")),
distance_km=distance_km,
)

def nearby_flights(
Expand All @@ -88,10 +114,21 @@ def nearby_flights(
detail = body.get("error") if isinstance(body, dict) else body
logger.warning("AirLabs returned no usable data: %s", detail)
return []
raw_flights = [f for f in (body.get("response") or []) if f.get("flight_icao")]
raw_flights = [
f
for f in (body.get("response") or [])
if f.get("flight_icao") and f.get("status") == "en-route"
]
# Sort by proximity to home so the display shows the closest
# flights first. Flights without coordinates go to the end.
raw_flights.sort(
key=lambda f: _haversine_km(lat, lon, f.get("lat", 0), f.get("lng", 0))
if f.get("lat") is not None and f.get("lng") is not None
else float("inf")
)
if raw:
return raw_flights
return [self.to_flight(f) for f in raw_flights]
return [self.to_flight(f, lat, lon) for f in raw_flights]
except (requests.exceptions.RequestException, ValueError) as e:
logger.warning("Error fetching flights from AirLabs: %s", e)
return []
1 change: 1 addition & 0 deletions src/jetset/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ class Flight:
speed: int | None = None
track: float | None = None
vertical_rate: int | None = None
distance_km: float | None = None

def __post_init__(self) -> None:
logger.debug(
Expand Down
Loading
Loading