Coverage for app/utils.py: 12%
782 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-13 03:30 +0000
1"""Shared utilities available to all blueprints."""
3import csv
4import functools
5import io
6import logging
7import math
8import os
9from collections import defaultdict
10from collections.abc import Callable
11from functools import wraps
12from typing import Any
14from flask import ( # pyright: ignore[reportMissingImports]
15 abort,
16 redirect,
17 session,
18 url_for,
19)
20from werkzeug.routing import ( # pyright: ignore[reportMissingImports]
21 BaseConverter,
22 ValidationError,
23)
25_log = logging.getLogger(__name__)
28def to_libpq_url(database_url: str) -> str:
29 """Strip a SQLAlchemy dialect+driver suffix (e.g. postgresql+psycopg://)
30 down to a plain postgresql:// URL, which is what libpq CLI tools
31 (pg_dump, psql) expect — they don't understand the +driver suffix."""
32 scheme, sep, rest = database_url.partition("://")
33 return f"{scheme.split('+', 1)[0]}{sep}{rest}" if sep else database_url
36# ── Tracks GIF export ─────────────────────────────────────────────────────────
38_GIF_W, _GIF_H = 800, 480
39_GIF_PAD = 30 # pixel padding around the track area
40_GIF_STEP_MS = 600 # ms per frame in the animated GIF
41_GIF_HOLD_MS = 3000 # ms for the final "all tracks" frame
42_TRACK_COLOUR = (180, 30, 200) # current/newest track: vivid purple
43_HISTORY_COLOUR = (
44 130,
45 20,
46 150,
47) # previously drawn tracks: slightly darker purple (GIF uses flat colour; web uses opacity fade)
48_BG_COLOUR = (248, 248, 252) # near-white background
51def _mercator_y(lat_deg: float) -> float:
52 """Web-Mercator y value for a latitude (not clamped)."""
53 lat = math.radians(max(-85.0, min(85.0, lat_deg)))
54 return math.log(math.tan(math.pi / 4 + lat / 2))
57def _build_gif_projection(
58 all_coords: list[tuple[float, float]],
59 canvas_w: int = _GIF_W,
60 canvas_h: int = _GIF_H,
61 pad: int = _GIF_PAD,
62) -> "tuple[Any, Any] | None":
63 """Return (project_fn, bbox) or None if there are fewer than 2 unique points."""
64 if len(all_coords) < 2:
65 return None
66 lons = [c[0] for c in all_coords]
67 lats = [c[1] for c in all_coords]
68 min_lon, max_lon = min(lons), max(lons)
69 min_lat, max_lat = min(lats), max(lats)
70 # Add small margin so tracks don't touch the edge
71 dlon = max(max_lon - min_lon, 0.1) * 0.1
72 dlat = max(max_lat - min_lat, 0.1) * 0.1
73 min_lon -= dlon
74 max_lon += dlon
75 min_lat -= dlat
76 max_lat += dlat
78 usable_w = canvas_w - 2 * pad
79 usable_h = canvas_h - 2 * pad
81 y_min = _mercator_y(min_lat)
82 y_max = _mercator_y(max_lat)
83 y_range = y_max - y_min or 1e-9 # Mercator log-tan units (NOT degrees)
84 x_range = max_lon - min_lon or 1e-9 # degrees
86 # Web Mercator geographic aspect: scale_y_mercator = scale_x * 180/π
87 # (1 Mercator-y unit = 111 km at any latitude; 1° lon at centre = cos(φ)*111 km,
88 # so equal-km scales satisfy scale_y = scale_x * 180/π).
89 # Pick scale_x so BOTH axes fit: scale_x ≤ w/x_range AND scale_x * 180/π * y_range ≤ h.
90 scale_x = min(
91 usable_w / x_range,
92 usable_h * math.pi / (180.0 * y_range),
93 )
94 scale_y = scale_x * 180.0 / math.pi # pixels per Mercator-y unit
96 off_x = pad + (usable_w - x_range * scale_x) / 2
97 off_y = pad + (usable_h - y_range * scale_y) / 2
99 def project(lon: float, lat: float) -> tuple[int, int]:
100 px = off_x + (lon - min_lon) * scale_x
101 py = off_y + (y_max - _mercator_y(lat)) * scale_y
102 return int(px), int(py)
104 return project, (min_lon, min_lat, max_lon, max_lat)
107def _coords_from_geojson(geojson: dict[str, Any] | None) -> list[tuple[float, float]]:
108 """Extract (lon, lat) pairs from a GeoJSON Feature or FeatureCollection.
110 Silently skips a malformed coordinate entry (wrong shape, non-numeric,
111 or non-finite) rather than raising — GpsTrack.geojson is a DB-stored
112 JSON field with no enforced schema; a corrupted entry should degrade to
113 "fewer points plotted", not crash track-image/GIF rendering with an
114 unhandled 500. Non-finite is included because Python's json module
115 accepts the non-standard "Infinity"/"NaN" tokens on load, and a NaN/inf
116 "coordinate" isn't a valid lon/lat regardless — downstream Mercator
117 projection math (_build_gif_projection) isn't finite-safe either.
118 """
119 if not geojson:
120 return []
121 if geojson.get("type") == "Feature":
122 geom = geojson.get("geometry") or {}
123 result = []
124 for c in geom.get("coordinates", []):
125 try:
126 if len(c) < 2:
127 continue
128 lon, lat = float(c[0]), float(c[1])
129 if math.isfinite(lon) and math.isfinite(lat):
130 result.append((lon, lat))
131 except (TypeError, ValueError):
132 continue
133 return result
134 if geojson.get("type") == "FeatureCollection":
135 result = []
136 for feat in geojson.get("features", []):
137 result.extend(_coords_from_geojson(feat))
138 return result
139 return []
142_TILE_UA = "OpenHangar flight track images (https://github.com/e2jk/OpenHangar)"
143# How long a downloaded tile is reused from the on-disk cache: OpenStreetMap
144# asks for tiles to be cached (its tile usage policy), OpenAIP data changes
145# more often.
146_TILE_MAX_AGE_S = {"osm": 30 * 86400, "opi": 7 * 86400}
147_TILE_FETCH_WORKERS = 4
150def _tile_cache_dir() -> str | None:
151 """Where downloaded map tiles are kept between requests
152 (upload_layout.TILE_CACHE_DIRNAME), or None outside the app."""
153 from flask import current_app, has_app_context
154 from services.upload_layout import ( # pyright: ignore[reportMissingImports]
155 TILE_CACHE_DIRNAME,
156 )
158 if not has_app_context():
159 return None
160 return os.path.join(
161 current_app.config.get("UPLOAD_FOLDER", "/data/uploads"), TILE_CACHE_DIRNAME
162 )
165def _fetch_tile_bytes(
166 source: str, z: int, x: int, y: int, url: str, cache_dir: str | None
167) -> bytes:
168 """One tile's PNG: from the on-disk cache when fresh enough, else
169 downloaded (and kept there)."""
170 import time
171 import urllib.request
173 path = (
174 os.path.join(cache_dir, source, str(z), str(x), f"{y}.png")
175 if cache_dir
176 else None
177 )
178 if (
179 path
180 and os.path.isfile(path)
181 and time.time() - os.path.getmtime(path) < _TILE_MAX_AGE_S[source]
182 ):
183 with open(path, "rb") as fh:
184 return fh.read()
185 req = urllib.request.Request(url, headers={"User-Agent": _TILE_UA})
186 with urllib.request.urlopen(req, timeout=5) as resp: # nosec B310 # url has a hardcoded https:// prefix
187 raw: bytes = resp.read()
188 if path:
189 try:
190 os.makedirs(os.path.dirname(path), exist_ok=True)
191 tmp = f"{path}.{os.getpid()}.part"
192 with open(tmp, "wb") as fh:
193 fh.write(raw)
194 os.replace(tmp, path)
195 except OSError as exc: # a full or read-only disk: just don't cache
196 _log.debug("Could not cache map tile %s: %s", path, exc)
197 return raw
200def _ensure_tiles(
201 z: int,
202 tiles: list[tuple[int, int]],
203 openaip_key: str | None,
204 cache: dict[Any, Any],
205) -> None:
206 """Put each of *tiles* (x, y at zoom *z*) in *cache* ready to paste, as
207 ("img", z, x, y): the OpenStreetMap tile toned down (_tone_down), with
208 the OpenAIP tile on top when there's a key (zoom ≤ 14). Missing ones are
209 fetched in parallel; a tile that can't be had is left out."""
210 from concurrent.futures import ThreadPoolExecutor
212 from PIL import Image as _Img # pyright: ignore[reportMissingImports]
214 todo = [(x, y) for x, y in dict.fromkeys(tiles) if ("img", z, x, y) not in cache]
215 if not todo:
216 return
217 cache_dir = _tile_cache_dir()
218 with_opi = bool(openaip_key) and z <= 14
220 def fetch(xy: tuple[int, int]) -> tuple[tuple[int, int], Any]:
221 x, y = xy
222 try:
223 base = _fetch_tile_bytes(
224 "osm",
225 z,
226 x,
227 y,
228 f"https://tile.openstreetmap.org/{z}/{x}/{y}.png",
229 cache_dir,
230 )
231 img = _tone_down(_Img.open(io.BytesIO(base)))
232 if with_opi:
233 opi = _fetch_tile_bytes(
234 "opi",
235 z,
236 x,
237 y,
238 f"https://api.tiles.openaip.net/api/data/openaip/{z}/{x}/{y}.png?apiKey={openaip_key}",
239 cache_dir,
240 )
241 overlay = _Img.open(io.BytesIO(opi)).convert("RGBA")
242 img.paste(overlay.convert("RGB"), (0, 0), overlay)
243 return xy, img
244 except Exception as exc: # noqa: BLE001 -- one bad tile must not fail the whole image
245 _log.debug("Map tile %s/%s/%s unavailable: %s", z, x, y, exc)
246 return xy, None
248 with ThreadPoolExecutor(max_workers=_TILE_FETCH_WORKERS) as pool:
249 for (x, y), img in pool.map(fetch, todo):
250 if img is not None:
251 cache[("img", z, x, y)] = img
254def _make_tile_background(
255 project: Callable[[float, float], tuple[int, int]],
256 min_lon: float,
257 max_lon: float,
258 min_lat: float,
259 max_lat: float,
260 canvas_w: int,
261 canvas_h: int,
262 openaip_key: str | None = None,
263 tile_cache: dict[Any, Any] | None = None,
264 max_tiles: int = 36,
265) -> Any:
266 """Fetch OSM raster tiles and composite them into a background PIL Image.
268 Returns a PIL Image, or None on failure (caller falls back to plain fill).
269 Tiles are fetched from OpenStreetMap; zoom level is chosen automatically.
270 Max 36 tiles are fetched to keep export time reasonable.
272 tile_cache, if provided, is shared across the frames of one image: it holds
273 each tile ready to paste (see _ensure_tiles), so a tile is fetched,
274 decoded and toned once per request.
275 """
276 from PIL import Image as _Img # pyright: ignore[reportMissingImports]
278 # Compute pixels-per-degree-lon from the projection function
279 mid_lat = (min_lat + max_lat) / 2.0
280 px0x = project(min_lon, mid_lat)[0]
281 px1x = project(min_lon + 1.0, mid_lat)[0]
282 scale_x = float(px1x - px0x) # canvas pixels per degree-lon
283 if scale_x <= 0:
284 return None
286 # Choose zoom so each tile is ~256 canvas pixels wide (clamped 2–14)
287 z = round(math.log2(max(scale_x * 360.0 / 256.0, 1.0)))
288 z = max(2, min(z, 14))
289 n = 2**z
291 def _lon_to_tx(lon: float) -> int:
292 return int(int((lon + 180.0) / 360.0 * n) % n)
294 def _lat_to_ty(lat: float) -> int:
295 lat_r = math.radians(max(-85.0, min(85.0, lat)))
296 return int(
297 (1.0 - math.log(math.tan(lat_r) + 1.0 / math.cos(lat_r)) / math.pi)
298 / 2.0
299 * n
300 )
302 def _tile_nw_lonlat(tx: int, ty: int) -> tuple[float, float]:
303 lon = tx * 360.0 / n - 180.0
304 lat = math.degrees(math.atan(math.sinh(math.pi * (1.0 - 2.0 * ty / n))))
305 return lon, lat
307 tx_min = _lon_to_tx(min_lon)
308 tx_max = _lon_to_tx(max_lon)
309 ty_min = _lat_to_ty(max_lat)
310 ty_max = _lat_to_ty(min_lat)
312 # Guard against tile count explosion
313 if (tx_max - tx_min + 1) * (ty_max - ty_min + 1) > max_tiles:
314 return None
316 bg = _Img.new("RGB", (canvas_w, canvas_h), _BG_COLOUR)
317 cache: dict[Any, Any] = tile_cache if tile_cache is not None else {}
318 _ensure_tiles(
319 z,
320 [
321 (tx % n, ty)
322 for tx in range(tx_min, tx_max + 1)
323 for ty in range(ty_min, ty_max + 1)
324 ],
325 openaip_key,
326 cache,
327 )
329 for tx in range(tx_min, tx_max + 1):
330 for ty in range(ty_min, ty_max + 1):
331 tile = cache.get(("img", z, tx % n, ty))
332 if tile is None:
333 continue # unavailable: that area stays plain background
334 # Compute pixel bounds for this tile by projecting its NW corner
335 # and the NW corner of the tile to its SE — then size the resize
336 # to exactly span that gap plus 1 px overlap, eliminating seams
337 # caused by integer rounding of adjacent corner coordinates.
338 lon_nw, lat_nw = _tile_nw_lonlat(tx, ty)
339 lon_se, lat_se = _tile_nw_lonlat(tx + 1, ty + 1)
340 px, py = project(lon_nw, lat_nw)
341 px_se, py_se = project(lon_se, lat_se)
342 tile_w = max(1, px_se - px + 1)
343 tile_h = max(1, py_se - py + 1)
344 bg.paste(tile.resize((tile_w, tile_h), _Img.Resampling.LANCZOS), (px, py))
346 return bg
349# Leaflet's map background, which the web maps' faded base layer lets through.
350_LEAFLET_BG = (221, 221, 221)
353def _tone_down(tile: Any) -> Any:
354 """A base-map tile toned down the way the web maps do it (base.css
355 .oh-basemap: grayscale(.6) brightness(1.05) contrast(.85) opacity(.8)),
356 so the airspace layer and the track stand out on the images too."""
357 from PIL import Image as _Img # pyright: ignore[reportMissingImports]
358 from PIL import ImageEnhance # pyright: ignore[reportMissingImports]
360 rgb = tile.convert("RGB")
361 rgb = _Img.blend(rgb, rgb.convert("L").convert("RGB"), 0.6)
362 rgb = ImageEnhance.Brightness(rgb).enhance(1.05)
363 rgb = ImageEnhance.Contrast(rgb).enhance(0.85)
364 return _Img.blend(_Img.new("RGB", rgb.size, _LEAFLET_BG), rgb, 0.8)
367def _canvas_geo_bounds(
368 project_fn: Callable[[float, float], tuple[int, int]],
369 canvas_w: int,
370 canvas_h: int,
371 min_lon: float,
372 max_lon: float,
373 min_lat: float,
374 max_lat: float,
375) -> tuple[float, float, float, float]:
376 """Return (lon_min, lat_min, lon_max, lat_max) spanning the full canvas.
378 Expands the geographic bbox beyond the track bounding box so the full
379 canvas area is covered by map tiles, eliminating plain-background padding.
380 Returns the original bbox if the inverse projection cannot be computed.
381 """
382 mid_lat = (min_lat + max_lat) / 2.0
383 # Longitude is linear with pixel x in Mercator
384 px0, _ = project_fn(min_lon, mid_lat)
385 px1, _ = project_fn(min_lon + 1.0, mid_lat)
386 scale_x = float(px1 - px0)
387 if scale_x <= 0:
388 return min_lon, min_lat, max_lon, max_lat
389 c_lon_min = min_lon - float(px0) / scale_x
390 c_lon_max = min_lon + float(canvas_w - px0) / scale_x
392 # Latitude: derive Mercator scale from two known projection points
393 _, py_bot = project_fn(min_lon, min_lat) # larger py (bottom of canvas area)
394 _, py_top = project_fn(min_lon, max_lat) # smaller py (top of canvas area)
395 merc_bot = _mercator_y(min_lat)
396 merc_top = _mercator_y(max_lat)
397 merc_range = merc_top - merc_bot
398 py_range = float(py_bot - py_top)
399 if merc_range <= 0 or py_range <= 0:
400 return c_lon_min, min_lat, c_lon_max, max_lat
401 scale_y = py_range / merc_range # pixels per Mercator-y unit
402 # Derived: merc_y at canvas row py = merc_top + (py_top - py) / scale_y
403 merc_canvas_top = merc_top + float(py_top) / scale_y
404 merc_canvas_bot = merc_top + float(py_top - canvas_h) / scale_y
405 # Clamp to avoid math.atan overflow at extreme Mercator values
406 merc_canvas_top = max(-10.0, min(10.0, merc_canvas_top))
407 merc_canvas_bot = max(-10.0, min(10.0, merc_canvas_bot))
408 c_lat_max = math.degrees(2.0 * math.atan(math.exp(merc_canvas_top)) - math.pi / 2.0)
409 c_lat_min = math.degrees(2.0 * math.atan(math.exp(merc_canvas_bot)) - math.pi / 2.0)
410 return c_lon_min, max(-85.0, c_lat_min), c_lon_max, min(85.0, c_lat_max)
413def _canvas_tile_background(
414 project_fn: Callable[[float, float], tuple[int, int]],
415 min_lon: float,
416 max_lon: float,
417 min_lat: float,
418 max_lat: float,
419 canvas_w: int,
420 canvas_h: int,
421 *,
422 openaip_key: str | None,
423 tile_cache: dict[Any, Any],
424 max_tiles_canvas: int,
425 max_tiles: int,
426) -> Any:
427 """The map background for a whole canvas showing the track bbox
428 (*min_lon*…*max_lat*): tiles over the full canvas — not only the track's
429 own extent, which left bands of plain background around a track whose
430 shape doesn't fill the canvas — or, when that needs more than
431 *max_tiles_canvas* tiles, over the track's extent with *max_tiles*.
432 None when neither fits (caller falls back to a plain fill)."""
433 c_min_lon, c_min_lat, c_max_lon, c_max_lat = _canvas_geo_bounds(
434 project_fn, canvas_w, canvas_h, min_lon, max_lon, min_lat, max_lat
435 )
436 bg = _make_tile_background(
437 project_fn,
438 c_min_lon,
439 c_max_lon,
440 c_min_lat,
441 c_max_lat,
442 canvas_w,
443 canvas_h,
444 openaip_key=openaip_key,
445 tile_cache=tile_cache,
446 max_tiles=max_tiles_canvas,
447 )
448 if bg is None:
449 bg = _make_tile_background(
450 project_fn,
451 min_lon,
452 max_lon,
453 min_lat,
454 max_lat,
455 canvas_w,
456 canvas_h,
457 openaip_key=openaip_key,
458 tile_cache=tile_cache,
459 max_tiles=max_tiles,
460 )
461 return bg
464def sort_tracks_oldest_first(
465 rows: list[dict[str, Any]],
466) -> list[dict[str, Any]]:
467 """Return rows sorted ascending by their 'date' string key (oldest first).
469 Both the web animation and the GIF export use this ordering so that
470 chronological playback and progressive opacity fading are consistent
471 across rendering environments. Call this once in the route handler and
472 pass the result to both the template context and generate_tracks_gif().
473 """
474 return sorted(rows, key=lambda r: r.get("date", ""))
477def generate_tracks_gif(
478 track_rows: list[dict[str, Any]],
479 _font_path: str = "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
480 _openaip_key: str | None = None,
481 canvas_w: int = _GIF_W,
482 canvas_h: int = _GIF_H,
483 high_res: bool = False,
484) -> bytes:
485 """Render an animated GIF of the flight tracks, oldest-first.
487 track_rows must already be sorted oldest-first; use sort_tracks_oldest_first().
488 Each frame adds one more track (cumulative) and re-fits the map to the
489 bounding box of tracks seen so far, creating a progressive zoom-out effect
490 that mirrors the web animation. Previous tracks are drawn in a lighter
491 colour; the newest one in vivid purple. Returns raw GIF bytes ready to
492 stream to the browser.
494 high_res=True doubles line widths, font sizes and padding, uses a 256-colour
495 palette, and raises the tile-fetch cap to 64 — producing a sharper map with
496 readable labels at the cost of a larger file.
497 """
498 from PIL import Image, ImageDraw, ImageFont # pyright: ignore[reportMissingImports]
500 _q_scale = 2 if high_res else 1
501 _pad = _GIF_PAD * _q_scale
502 _line_curr = 3 * _q_scale
503 _line_hist = 2 * _q_scale
504 _font_sz = 14 * _q_scale
505 _font_sz_sm = 11 * _q_scale
506 _txt_margin = 8 * _q_scale
507 _q_colors = 256 if high_res else 128
508 _max_tiles = 64 if high_res else 36
509 # Canvas-extent calls cover the full 1600×960 canvas which can require
510 # ~70–90 tiles at certain zoom levels (wide flat tracks at z=7), so use a
511 # higher cap than the track-bbox call. 128 = 2× the high-res track cap.
512 _max_tiles_canvas = _max_tiles * 2 if high_res else _max_tiles
514 # Pre-compute per-track coordinates (avoids re-parsing GeoJSON in the inner loop)
515 per_track_coords = [_coords_from_geojson(r.get("geojson")) for r in track_rows]
516 all_coords: list[tuple[float, float]] = [c for tc in per_track_coords for c in tc]
518 if len(all_coords) < 2:
519 # Fallback: single blank frame
520 img = Image.new("RGB", (canvas_w, canvas_h), _BG_COLOUR)
521 buf = io.BytesIO()
522 img.save(buf, format="GIF")
523 return buf.getvalue()
525 font: Any # PIL font type varies across Pillow versions
526 font_sm: Any
527 try:
528 font = ImageFont.truetype(_font_path, _font_sz)
529 font_sm = ImageFont.truetype(_font_path, _font_sz_sm)
530 except OSError:
531 font = font_sm = ImageFont.load_default()
533 def draw_shadow_text(draw: Any, text: str, font: Any) -> None:
534 draw.text(
535 (_txt_margin + 1, _txt_margin + 1), text, fill=(255, 255, 255), font=font
536 )
537 draw.text((_txt_margin, _txt_margin), text, fill=(40, 40, 60), font=font)
539 def draw_track(
540 draw: Any,
541 coords: list[tuple[float, float]],
542 colour: tuple[int, int, int],
543 width: int,
544 project_fn: Callable[[float, float], tuple[int, int]],
545 ) -> None:
546 pts = [project_fn(lon, lat) for lon, lat in coords]
547 if len(pts) >= 2:
548 draw.line(pts, fill=colour, width=width)
549 r = width + 2 * _q_scale
550 sx, sy = pts[0]
551 draw.ellipse([sx - r, sy - r, sx + r, sy + r], fill=(40, 160, 60))
552 ex, ey = pts[-1]
553 draw.ellipse([ex - r, ey - r, ex + r, ey + r], fill=(200, 40, 40))
555 # Shared tile cache: (z, tx, ty) → raw PNG bytes; ("opi", z, tx, ty) for OpenAIP.
556 # Avoids re-fetching tiles that appear in multiple frames at the same zoom level.
557 tile_cache: dict[Any, Any] = {}
559 def _frame_bg(
560 project_fn: Callable[[float, float], tuple[int, int]],
561 min_lon: float,
562 max_lon: float,
563 min_lat: float,
564 max_lat: float,
565 ) -> Any:
566 bg = _canvas_tile_background(
567 project_fn,
568 min_lon,
569 max_lon,
570 min_lat,
571 max_lat,
572 canvas_w,
573 canvas_h,
574 openaip_key=_openaip_key,
575 tile_cache=tile_cache,
576 max_tiles_canvas=_max_tiles_canvas,
577 max_tiles=_max_tiles,
578 )
579 return (
580 bg.copy()
581 if bg is not None
582 else Image.new("RGB", (canvas_w, canvas_h), _BG_COLOUR)
583 )
585 frames: list[Any] = []
586 durations: list[int] = []
587 accumulated_coords: list[tuple[float, float]] = []
589 # The map with every earlier track drawn, for the view it was drawn in:
590 # a frame whose view didn't change (the new track lies within the area
591 # already shown — most frames once the first few flights are in) starts
592 # from it instead of fetching the map and redrawing all earlier tracks.
593 history_img: Any = None
594 history_view: tuple[float, float, float, float] | None = None
596 for frame_idx in range(len(track_rows)):
597 accumulated_coords.extend(per_track_coords[frame_idx])
598 proj_result = _build_gif_projection(
599 accumulated_coords, canvas_w=canvas_w, canvas_h=canvas_h, pad=_pad
600 )
601 if proj_result is None:
602 continue # not enough coords yet (e.g. leading rows with no geojson)
604 project_fn, view = proj_result
605 f_min_lon, f_min_lat, f_max_lon, f_max_lat = view
606 if view != history_view:
607 history_img = _frame_bg(
608 project_fn, f_min_lon, f_max_lon, f_min_lat, f_max_lat
609 )
610 history_draw = ImageDraw.Draw(history_img)
611 for i in range(frame_idx):
612 draw_track(
613 history_draw,
614 per_track_coords[i],
615 _HISTORY_COLOUR,
616 _line_hist,
617 project_fn,
618 )
619 history_view = view
620 img = history_img.copy()
621 draw = ImageDraw.Draw(img)
622 draw_track(
623 draw, per_track_coords[frame_idx], _TRACK_COLOUR, _line_curr, project_fn
624 )
625 draw_track(
626 ImageDraw.Draw(history_img),
627 per_track_coords[frame_idx],
628 _HISTORY_COLOUR,
629 _line_hist,
630 project_fn,
631 )
633 row = track_rows[frame_idx]
634 label = f"{row.get('date', '')} {row.get('dep', '')} → {row.get('arr', '')}"
635 draw_shadow_text(draw, label, font)
636 draw.text(
637 (_txt_margin, canvas_h - _font_sz_sm - _txt_margin),
638 f"{frame_idx + 1} / {len(track_rows)}",
639 fill=(80, 80, 100),
640 font=font_sm,
641 )
643 frames.append(img)
644 durations.append(_GIF_STEP_MS)
646 # Final frame: all tracks at equal weight using the full bounding box, longer hold
647 if frames:
648 proj_result_final = _build_gif_projection(
649 all_coords, canvas_w=canvas_w, canvas_h=canvas_h, pad=_pad
650 )
651 # proj_result_final is guaranteed non-None: all_coords has ≥ 2 points (checked above)
652 project_final, (f_min_lon, f_min_lat, f_max_lon, f_max_lat) = proj_result_final # type: ignore[misc]
653 img = _frame_bg(project_final, f_min_lon, f_max_lon, f_min_lat, f_max_lat)
654 draw = ImageDraw.Draw(img)
655 for tc in per_track_coords:
656 draw_track(draw, tc, _TRACK_COLOUR, _line_hist, project_final)
657 draw_shadow_text(draw, f"All {len(track_rows)} tracks", font)
658 frames.append(img)
659 durations.append(_GIF_HOLD_MS)
661 # Quantise all frames to a shared 128-colour palette built from the last frame
662 # (widest view, most visually representative). A single global palette means
663 # identical background areas compress very well with GIF's LZW codec.
664 palette_src = frames[-1].quantize(colors=_q_colors, dither=Image.Dither.NONE)
665 quantized: list[Any] = [
666 f.quantize(palette=palette_src, dither=Image.Dither.NONE) for f in frames
667 ]
669 buf = io.BytesIO()
670 quantized[0].save(
671 buf,
672 format="GIF",
673 save_all=True,
674 append_images=quantized[1:],
675 loop=0,
676 duration=durations,
677 optimize=True,
678 )
679 return buf.getvalue()
682def generate_single_track_image(
683 geojson: dict[str, Any] | None,
684 date: str = "",
685 dep: str = "",
686 arr: str = "",
687 _font_path: str = "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
688 _openaip_key: str | None = None,
689 canvas_w: int = _GIF_W,
690 canvas_h: int = _GIF_H,
691 high_res: bool = False,
692) -> bytes:
693 """Render a single GPS track as a static PNG.
695 Returns raw PNG bytes. If the track has fewer than 2 points a blank
696 canvas is returned so the caller can always stream a valid image.
697 """
698 from PIL import Image, ImageDraw, ImageFont # pyright: ignore[reportMissingImports]
700 all_coords = _coords_from_geojson(geojson)
702 _q_scale = 2 if high_res else 1
703 _pad = _GIF_PAD * _q_scale
704 _line_w = 3 * _q_scale
705 _font_sz = 14 * _q_scale
706 _txt_margin = 8 * _q_scale
707 _max_tiles = 64 if high_res else 36
708 _max_tiles_canvas = _max_tiles * 2 if high_res else _max_tiles
710 def _blank_png() -> bytes:
711 buf = io.BytesIO()
712 Image.new("RGB", (canvas_w, canvas_h), _BG_COLOUR).save(buf, format="PNG")
713 return buf.getvalue()
715 if len(all_coords) < 2:
716 return _blank_png()
718 proj_result = _build_gif_projection(
719 all_coords, canvas_w=canvas_w, canvas_h=canvas_h, pad=_pad
720 )
721 if proj_result is None:
722 return _blank_png()
724 project_fn, (f_min_lon, f_min_lat, f_max_lon, f_max_lat) = proj_result
726 tile_cache: dict[Any, Any] = {}
727 bg = _canvas_tile_background(
728 project_fn,
729 f_min_lon,
730 f_max_lon,
731 f_min_lat,
732 f_max_lat,
733 canvas_w,
734 canvas_h,
735 openaip_key=_openaip_key,
736 tile_cache=tile_cache,
737 max_tiles_canvas=_max_tiles_canvas,
738 max_tiles=_max_tiles,
739 )
740 img = (
741 bg.copy()
742 if bg is not None
743 else Image.new("RGB", (canvas_w, canvas_h), _BG_COLOUR)
744 )
746 draw = ImageDraw.Draw(img)
747 pts = [project_fn(lon, lat) for lon, lat in all_coords]
748 if len(pts) >= 2:
749 draw.line(pts, fill=_TRACK_COLOUR, width=_line_w)
750 r = _line_w + 2 * _q_scale
751 sx, sy = pts[0]
752 draw.ellipse([sx - r, sy - r, sx + r, sy + r], fill=(40, 160, 60))
753 ex, ey = pts[-1]
754 draw.ellipse([ex - r, ey - r, ex + r, ey + r], fill=(200, 40, 40))
756 label = f"{date} {dep} → {arr}".strip(" →").strip()
757 if label:
758 try:
759 font: Any = ImageFont.truetype(_font_path, _font_sz)
760 except OSError:
761 font = ImageFont.load_default()
762 draw.text(
763 (_txt_margin + 1, _txt_margin + 1), label, fill=(255, 255, 255), font=font
764 )
765 draw.text((_txt_margin, _txt_margin), label, fill=(40, 40, 60), font=font)
767 buf = io.BytesIO()
768 img.save(buf, format="PNG")
769 return buf.getvalue()
772def generate_single_track_gif(
773 geojson: dict[str, Any] | None,
774 date: str = "",
775 dep: str = "",
776 arr: str = "",
777 _font_path: str = "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
778 _openaip_key: str | None = None,
779 canvas_w: int = _GIF_W,
780 canvas_h: int = _GIF_H,
781 high_res: bool = False,
782) -> bytes:
783 """Render an animated GIF of a single flight track drawn progressively.
785 The track's coordinate array is split into N equal chunks (3–10 depending
786 on track length). Each frame draws all chunks revealed so far, fitting the
787 map bounding box to the accumulated coords so the view zooms out as the
788 route grows — mirroring the per-flight behaviour of the web Animate button.
790 Falls back to a single-frame GIF wrapping the still PNG when the track has
791 fewer than 20 GPS points (too sparse for meaningful animation).
792 """
793 from PIL import Image, ImageDraw, ImageFont # pyright: ignore[reportMissingImports]
795 all_coords = _coords_from_geojson(geojson)
797 _q_scale = 2 if high_res else 1
798 _pad = _GIF_PAD * _q_scale
799 _line_curr = 3 * _q_scale
800 _line_hist = 2 * _q_scale
801 _font_sz = 14 * _q_scale
802 _font_sz_sm = 11 * _q_scale
803 _txt_margin = 8 * _q_scale
804 _q_colors = 256 if high_res else 128
805 _max_tiles = 64 if high_res else 36
806 _max_tiles_canvas = _max_tiles * 2 if high_res else _max_tiles
808 def _blank_gif() -> bytes:
809 buf = io.BytesIO()
810 Image.new("RGB", (canvas_w, canvas_h), _BG_COLOUR).save(buf, format="GIF")
811 return buf.getvalue()
813 if len(all_coords) < 2:
814 return _blank_gif()
816 # Sparse-track fallback: wrap the still image in a single-frame GIF
817 if len(all_coords) < 20:
818 png = generate_single_track_image(
819 geojson,
820 date=date,
821 dep=dep,
822 arr=arr,
823 _font_path=_font_path,
824 _openaip_key=_openaip_key,
825 canvas_w=canvas_w,
826 canvas_h=canvas_h,
827 high_res=high_res,
828 )
829 img = Image.open(io.BytesIO(png)).convert("RGB")
830 buf = io.BytesIO()
831 img.quantize(colors=_q_colors, dither=Image.Dither.NONE).save(buf, format="GIF")
832 return buf.getvalue()
834 # Split coords into N equal chunks (3–10)
835 n_chunks = max(3, min(10, len(all_coords) // 10))
836 chunk_size = len(all_coords) // n_chunks
837 chunks: list[list[tuple[float, float]]] = [
838 list(all_coords[i * chunk_size : (i + 1) * chunk_size])
839 for i in range(n_chunks - 1)
840 ]
841 chunks.append(
842 list(all_coords[(n_chunks - 1) * chunk_size :])
843 ) # last catches remainder
845 try:
846 font: Any = ImageFont.truetype(_font_path, _font_sz)
847 font_sm: Any = ImageFont.truetype(_font_path, _font_sz_sm)
848 except OSError:
849 font = font_sm = ImageFont.load_default()
851 def draw_shadow_text(draw: Any, text: str, fnt: Any) -> None:
852 draw.text(
853 (_txt_margin + 1, _txt_margin + 1), text, fill=(255, 255, 255), font=fnt
854 )
855 draw.text((_txt_margin, _txt_margin), text, fill=(40, 40, 60), font=fnt)
857 def draw_track(
858 draw: Any,
859 coords: list[tuple[float, float]],
860 colour: tuple[int, int, int],
861 width: int,
862 project_fn: Callable[[float, float], tuple[int, int]],
863 ) -> None:
864 pts = [project_fn(lon, lat) for lon, lat in coords]
865 if len(pts) >= 2:
866 draw.line(pts, fill=colour, width=width)
867 r = width + 2 * _q_scale
868 sx, sy = pts[0]
869 draw.ellipse([sx - r, sy - r, sx + r, sy + r], fill=(40, 160, 60))
870 ex, ey = pts[-1]
871 draw.ellipse([ex - r, ey - r, ex + r, ey + r], fill=(200, 40, 40))
873 tile_cache: dict[Any, Any] = {}
875 def _frame_bg(
876 project_fn: Callable[[float, float], tuple[int, int]],
877 min_lon: float,
878 max_lon: float,
879 min_lat: float,
880 max_lat: float,
881 ) -> Any:
882 bg = _canvas_tile_background(
883 project_fn,
884 min_lon,
885 max_lon,
886 min_lat,
887 max_lat,
888 canvas_w,
889 canvas_h,
890 openaip_key=_openaip_key,
891 tile_cache=tile_cache,
892 max_tiles_canvas=_max_tiles_canvas,
893 max_tiles=_max_tiles,
894 )
895 return (
896 bg.copy()
897 if bg is not None
898 else Image.new("RGB", (canvas_w, canvas_h), _BG_COLOUR)
899 )
901 label = f"{date} {dep} → {arr}".strip(" →").strip()
903 frames: list[Any] = []
904 durations: list[int] = []
905 accumulated: list[tuple[float, float]] = []
907 for chunk_idx, chunk in enumerate(chunks):
908 accumulated.extend(chunk)
909 proj_result = _build_gif_projection(
910 accumulated, canvas_w=canvas_w, canvas_h=canvas_h, pad=_pad
911 )
912 if proj_result is None:
913 continue
915 project_fn, (f_min_lon, f_min_lat, f_max_lon, f_max_lat) = proj_result
916 img = _frame_bg(project_fn, f_min_lon, f_max_lon, f_min_lat, f_max_lat)
917 draw = ImageDraw.Draw(img)
919 for i in range(chunk_idx):
920 draw_track(draw, chunks[i], _HISTORY_COLOUR, _line_hist, project_fn)
921 draw_track(draw, chunk, _TRACK_COLOUR, _line_curr, project_fn)
923 if label:
924 draw_shadow_text(draw, label, font)
925 draw.text(
926 (_txt_margin, canvas_h - _font_sz_sm - _txt_margin),
927 f"{chunk_idx + 1} / {n_chunks}",
928 fill=(80, 80, 100),
929 font=font_sm,
930 )
931 frames.append(img)
932 durations.append(_GIF_STEP_MS)
934 if not frames:
935 return _blank_gif()
937 # Final hold frame: full track, all chunks at equal weight
938 proj_result_final = _build_gif_projection(
939 list(all_coords), canvas_w=canvas_w, canvas_h=canvas_h, pad=_pad
940 )
941 # guaranteed non-None: all_coords has ≥ 20 points
942 project_final, (f_min_lon, f_min_lat, f_max_lon, f_max_lat) = proj_result_final # type: ignore[misc]
943 img = _frame_bg(project_final, f_min_lon, f_max_lon, f_min_lat, f_max_lat)
944 draw = ImageDraw.Draw(img)
945 for chunk in chunks:
946 draw_track(draw, chunk, _TRACK_COLOUR, _line_hist, project_final)
947 if label:
948 draw_shadow_text(draw, label, font)
949 frames.append(img)
950 durations.append(_GIF_HOLD_MS)
952 palette_src = frames[-1].quantize(colors=_q_colors, dither=Image.Dither.NONE)
953 quantized: list[Any] = [
954 f.quantize(palette=palette_src, dither=Image.Dither.NONE) for f in frames
955 ]
956 buf = io.BytesIO()
957 quantized[0].save(
958 buf,
959 format="GIF",
960 save_all=True,
961 append_images=quantized[1:],
962 loop=0,
963 duration=durations,
964 optimize=True,
965 )
966 return buf.getvalue()
969@functools.lru_cache(maxsize=1)
970def _load_aircraft_types() -> dict[str, tuple[str, str]]:
971 """Return {type_designator: (manufacturer, model)} from aircraft_types.csv."""
972 path = os.path.join(os.path.dirname(__file__), "data", "aircraft_types.csv")
973 result: dict[str, tuple[str, str]] = {}
974 try:
975 with open(path, newline="", encoding="utf-8") as f:
976 for row in csv.DictReader(f):
977 des = row.get("type_designator", "").strip().upper()
978 mfr = row.get("manufacturer", "").strip()
979 model = row.get("model", "").strip()
980 if des:
981 result[des] = (mfr, model)
982 except OSError as exc:
983 _log.warning("aircraft_types.csv not found: %s", exc)
984 return result
987def _load_aircraft_type_variants() -> list[tuple[str, str, str, str]]:
988 """Return all (type_designator, full_name, manufacturer, model) tuples — one per CSV row.
990 Unlike _load_aircraft_types(), duplicate designators are preserved so
991 the search endpoint can surface every variant (e.g. all PA-28-181 models
992 that share the P28A ICAO code). manufacturer and model are kept separate
993 so callers can pre-fill form fields for the exact selected variant rather
994 than an arbitrary one sharing the same ICAO code.
995 """
996 path = os.path.join(os.path.dirname(__file__), "data", "aircraft_types.csv")
997 result: list[tuple[str, str, str, str]] = []
998 seen: set[tuple[str, str]] = set()
999 try:
1000 with open(path, newline="", encoding="utf-8") as f:
1001 for row in csv.DictReader(f):
1002 des = row.get("type_designator", "").strip().upper()
1003 mfr = row.get("manufacturer", "").strip()
1004 model = row.get("model", "").strip()
1005 name = f"{mfr} {model}".strip()
1006 if des and (des, name) not in seen:
1007 result.append((des, name, mfr, model))
1008 seen.add((des, name))
1009 except OSError as exc:
1010 _log.warning("aircraft_types.csv not found: %s", exc)
1011 return result
1014@functools.lru_cache(maxsize=1)
1015def _load_aircraft_type_engine_data() -> dict[str, tuple[int, str]]:
1016 """Return {type_designator: (engine_count, engine_type)} from aircraft_types.csv.
1018 For designators with multiple rows (variants), uses data from the first row.
1019 """
1020 path = os.path.join(os.path.dirname(__file__), "data", "aircraft_types.csv")
1021 result: dict[str, tuple[int, str]] = {}
1022 try:
1023 with open(path, newline="", encoding="utf-8") as f:
1024 for row in csv.DictReader(f):
1025 des = row.get("type_designator", "").strip().upper()
1026 if not des or des in result:
1027 continue
1028 try:
1029 ec = int(row.get("engine_count", "1"))
1030 except ValueError:
1031 ec = 1
1032 et = row.get("engine_type", "").strip()
1033 result[des] = (ec, et)
1034 except OSError as exc:
1035 _log.warning("aircraft_types.csv not found: %s", exc)
1036 return result
1039def get_aircraft_type_engine_info(icao_code: str) -> tuple[int, str] | None:
1040 """Return (engine_count, engine_type) for the given ICAO code, or None."""
1041 return _load_aircraft_type_engine_data().get(icao_code.strip().upper())
1044@functools.lru_cache(maxsize=1)
1045def _build_model_name_prefix_lookup() -> dict[str, str]:
1046 """Return {compact_first_word_of_model: icao_code} for resolve_aircraft_type_icao fallback.
1048 Enables matching pilot-logbook type strings like "DR401" or "PA28-161 TDI"
1049 against ICAO codes like "DR40" or "P28A" via the CSV model-name column.
1050 """
1051 path = os.path.join(os.path.dirname(__file__), "data", "aircraft_types.csv")
1052 result: dict[str, str] = {}
1053 try:
1054 with open(path, newline="", encoding="utf-8") as f:
1055 for row in csv.DictReader(f):
1056 des = row.get("type_designator", "").strip().upper()
1057 model = row.get("model", "").strip()
1058 if not des or not model:
1059 continue
1060 first_word = model.split()[0]
1061 key = first_word.replace("-", "").replace(" ", "").upper()
1062 if key and key not in result:
1063 result[key] = des
1064 except OSError as exc:
1065 _log.warning("aircraft_types.csv not found: %s", exc)
1066 return result
1069@functools.lru_cache(maxsize=1)
1070def _sorted_model_prefix_keys() -> list[str]:
1071 """Keys of _build_model_name_prefix_lookup(), longest-first.
1073 Cached separately from the lookup dict itself: found by fuzzing that
1074 resolve_aircraft_type_icao() re-sorted these ~1-2k keys on *every* call
1075 that reaches the fallback branch (i.e. almost every call — an exact ICAO
1076 code is rarely what a user types or a logbook CSV column holds), even
1077 though the lookup dict underneath never changes between calls.
1078 """
1079 return sorted(_build_model_name_prefix_lookup(), key=len, reverse=True)
1082@functools.lru_cache(maxsize=4096)
1083def resolve_aircraft_type_icao(aircraft_type: str | None) -> str | None:
1084 """Return the matching ICAO type designator for *aircraft_type*, or None.
1086 Cached: a pure function of the text over static type data, and a
1087 "make model variant" text that only the word-by-word step below
1088 resolves walks every model-name key first (~2 ms) — the landing
1089 recency resolves each recent flight's type on every dashboard view."""
1090 if not aircraft_type:
1091 return None
1092 types = _load_aircraft_types()
1093 norm = aircraft_type.strip().upper()
1094 if norm in types:
1095 return norm
1096 # Try stripping hyphens and spaces (e.g. "PA-28" → "PA28")
1097 compact = norm.replace("-", "").replace(" ", "")
1098 if compact in types:
1099 return compact
1100 # Try matching against the first word of each CSV model name (longest key wins).
1101 # e.g. "DR401" startswith "DR401" (from "DR-401 135CDI") → DR40
1102 # "PA28161TDI" startswith "PA28161" (from "PA-28-161 Cherokee…") → P28A
1103 # Guard: reject if the unmatched tail starts with a digit — that signals a
1104 # different model number (e.g. "C172RG" wrongly matching key "C17").
1105 prefix_lookup = _build_model_name_prefix_lookup()
1106 for key in _sorted_model_prefix_keys():
1107 if len(key) < 4:
1108 break # keys are sorted longest-first; stop once they're too short
1109 if compact.startswith(key):
1110 tail = compact[len(key) :]
1111 if tail and tail[0].isdigit():
1112 continue # "C172RG" matching "C17" — tail "2RG" starts with digit
1113 return prefix_lookup[key]
1114 # Word by word, for a "make model variant" text such as a managed
1115 # aircraft's "ROBIN DR-401 155CDI" (Flight.display_aircraft_type): the
1116 # manufacturer in front breaks the match above, and the engine variant
1117 # after the model ("155CDI") trips its digit guard. A word matching a
1118 # model name's first word exactly ("DR-401" → DR40) settles it. Only
1119 # words with a digit, so a manufacturer or a common word can't match.
1120 for word in norm.split():
1121 key = word.replace("-", "")
1122 if len(key) >= 4 and any(c.isdigit() for c in key) and key in prefix_lookup:
1123 return prefix_lookup[key]
1124 return None
1127# EASA aeroplane class ratings that landing recency (FCL.060(b)) counts by
1128# class rather than by type. Anything else — turboprops, jets, helicopters,
1129# gyrocopters, electric — is counted per ICAO type, the stricter reading.
1130AIRCRAFT_CLASS_SEP_LAND = "sep_land"
1131AIRCRAFT_CLASS_SEP_SEA = "sep_sea"
1132AIRCRAFT_CLASS_MEP_LAND = "mep_land"
1133AIRCRAFT_CLASS_MEP_SEA = "mep_sea"
1136@functools.lru_cache(maxsize=1)
1137def _load_aircraft_classes() -> dict[str, str]:
1138 """{type_designator: class} for the piston aeroplanes rated by class.
1140 From the ICAO data's description and engines: one piston engine is SEP,
1141 two are MEP; a seaplane is the "sea" class, a landplane or an amphibian
1142 (flown mostly from land) the "land" one. The data can't tell a touring
1143 motor glider from an aeroplane, so a TMG is counted as SEP (land)."""
1144 path = os.path.join(os.path.dirname(__file__), "data", "aircraft_types.csv")
1145 result: dict[str, str] = {}
1146 try:
1147 with open(path, newline="", encoding="utf-8") as f:
1148 for row in csv.DictReader(f):
1149 des = row.get("type_designator", "").strip().upper()
1150 if not des or des in result or row.get("engine_type") != "Piston":
1151 continue
1152 sea = row.get("description") == "SeaPlane"
1153 if row.get("description") not in ("LandPlane", "Amphibian", "SeaPlane"):
1154 continue
1155 engines = row.get("engine_count", "").strip()
1156 if engines == "1":
1157 result[des] = (
1158 AIRCRAFT_CLASS_SEP_SEA if sea else AIRCRAFT_CLASS_SEP_LAND
1159 )
1160 elif engines == "2":
1161 result[des] = (
1162 AIRCRAFT_CLASS_MEP_SEA if sea else AIRCRAFT_CLASS_MEP_LAND
1163 )
1164 except OSError as exc:
1165 _log.warning("aircraft_types.csv not found: %s", exc)
1166 return result
1169def aircraft_class_for_icao(icao: str | None) -> str | None:
1170 """The EASA class (AIRCRAFT_CLASS_*) recency counts *icao* under, or
1171 None when it is counted per type."""
1172 return _load_aircraft_classes().get((icao or "").strip().upper())
1175@functools.lru_cache(maxsize=1)
1176def _load_airport_names() -> dict[str, str]:
1177 """Return {ICAO ident: airport name} for all airports in airports.csv."""
1178 path = os.path.join(os.path.dirname(__file__), "data", "airports.csv")
1179 result: dict[str, str] = {}
1180 try:
1181 with open(path, newline="", encoding="utf-8") as f:
1182 for row in csv.DictReader(f):
1183 ident = row.get("ident", "").strip()
1184 name = row.get("name", "").strip()
1185 if ident and name:
1186 result[ident] = name
1187 except OSError as exc:
1188 _log.warning("airports.csv not found: %s", exc)
1189 return result
1192@functools.lru_cache(maxsize=1)
1193def _load_airport_geo() -> dict[str, dict[str, Any]]:
1194 """Return {ICAO ident: {lat, lon, elevation_ft, country, region, name}}
1195 for all airports in airports.csv that have coordinates. Used by the
1196 pilot flight-statistics report (distance, farthest points, elevation
1197 extremes) — kept separate from _load_airport_names() since most callers
1198 only need the name."""
1199 path = os.path.join(os.path.dirname(__file__), "data", "airports.csv")
1200 result: dict[str, dict[str, Any]] = {}
1201 try:
1202 with open(path, newline="", encoding="utf-8") as f:
1203 for row in csv.DictReader(f):
1204 ident = row.get("ident", "").strip()
1205 lat_raw = row.get("latitude_deg", "").strip()
1206 lon_raw = row.get("longitude_deg", "").strip()
1207 if not ident or not lat_raw or not lon_raw:
1208 continue
1209 try:
1210 lat = float(lat_raw)
1211 lon = float(lon_raw)
1212 except ValueError:
1213 continue
1214 elev_raw = row.get("elevation_ft", "").strip()
1215 result[ident] = {
1216 "lat": lat,
1217 "lon": lon,
1218 "elevation_ft": float(elev_raw) if elev_raw else None,
1219 "country": row.get("iso_country", "").strip(),
1220 "region": row.get("iso_region", "").strip(),
1221 "name": row.get("name", "").strip(),
1222 }
1223 except OSError as exc:
1224 _log.warning("airports.csv not found: %s", exc)
1225 return result
1228_EARTH_RADIUS_NM = 3440.065
1231def haversine_nm(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
1232 """Great-circle distance between two points, in nautical miles."""
1233 p1, p2 = math.radians(lat1), math.radians(lat2)
1234 dphi = math.radians(lat2 - lat1)
1235 dlambda = math.radians(lon2 - lon1)
1236 a = (
1237 math.sin(dphi / 2) ** 2
1238 + math.cos(p1) * math.cos(p2) * math.sin(dlambda / 2) ** 2
1239 )
1240 return _EARTH_RADIUS_NM * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
1243_alog = logging.getLogger("openhangar.activity")
1246def _sl(value: object) -> str:
1247 """Sanitize a value for log output — strips CR/LF to prevent log injection (CWE-117)."""
1248 return str(value).replace("\r\n", "").replace("\n", "").replace("\r", "")
1251def activity(event: str, **fields: object) -> None:
1252 """Emit a structured [ACTIVITY] log entry with user_id and ip automatically included."""
1253 from flask import request, session
1255 uid = session.get("user_id", "")
1256 ip = request.remote_addr or ""
1257 parts = [f"[ACTIVITY] {event}", f"user_id={_sl(uid)}", f"ip={_sl(ip)}"]
1258 parts.extend(f"{k}={_sl(v)}" for k, v in fields.items())
1259 _alog.info(" ".join(parts))
1262class AircraftRefConverter(BaseConverter):
1263 """URL converter for the ``aircraft_id`` slot: accepts either the numeric
1264 primary key (unchanged, always works) or the aircraft's registration
1265 (e.g. ``OO-GRN``), so routes like ``/aircraft/<aircraft_id>/flights``
1266 also resolve ``/aircraft/OO-GRN/flights``.
1268 Authorization is unaffected: this only resolves the URL segment to a
1269 primary key (or the reverse, for link generation) — every view still
1270 does its own tenant-scoped lookup exactly as before, so a registration
1271 belonging to another tenant 404s the same way a wrong numeric id does.
1273 Registrations containing '/' or spaces (rare, but not forbidden by the
1274 model) are sanitized the same way upload filenames already are
1275 elsewhere in this blueprint; such an aircraft simply isn't reachable via
1276 its pretty URL (only via the numeric id, which always works).
1277 """
1279 regex = r"[^/]+"
1281 def to_python(self, value: str) -> int:
1282 if value.isdigit():
1283 return int(value)
1285 from models import Aircraft, db
1287 needle = value.upper()
1288 ac = (
1289 Aircraft.query.filter(
1290 db.func.upper(
1291 db.func.replace(
1292 db.func.replace(Aircraft.registration, "/", "-"), " ", "-"
1293 )
1294 )
1295 == needle
1296 )
1297 .order_by(Aircraft.id)
1298 .first()
1299 )
1300 if ac is None:
1301 raise ValidationError()
1302 return int(ac.id)
1304 def to_url(self, value: Any) -> str:
1305 if isinstance(value, str) and not value.isdigit():
1306 return super().to_url(value)
1308 from models import Aircraft, db
1310 ac = db.session.get(Aircraft, int(value))
1311 reg = ac.registration if ac and ac.registration else str(value)
1312 safe_reg = reg.replace("/", "-").replace(" ", "-")
1313 return super().to_url(safe_reg)
1316def login_required(f: Callable[..., Any]) -> Callable[..., Any]:
1317 """Redirect unauthenticated users to the login page."""
1319 @wraps(f)
1320 def decorated(*args: Any, **kwargs: Any) -> Any:
1321 if not session.get("user_id"):
1322 return redirect(url_for("auth.login"))
1323 return f(*args, **kwargs)
1325 return decorated
1328def require_instance_admin(f: Callable[..., Any]) -> Callable[..., Any]:
1329 """Abort 403 unless the current user is the instance admin."""
1331 @wraps(f)
1332 def decorated(*args: Any, **kwargs: Any) -> Any:
1333 from models import User, db
1335 user_id = session.get("user_id")
1336 if not user_id:
1337 return redirect(url_for("auth.login"))
1338 user = db.session.get(User, user_id)
1339 if not user or not user.is_instance_admin:
1340 abort(403)
1341 return f(*args, **kwargs)
1343 return decorated
1346def current_user_role() -> str | None:
1347 """Return the Role of the current user in their tenant, or None."""
1348 from models import TenantUser
1350 user_id = session.get("user_id")
1351 if not user_id:
1352 return None
1353 tu = TenantUser.query.filter_by(user_id=user_id).first()
1354 return tu.role if tu else None
1357def check_update_available() -> bool:
1358 """Return True when a newer release is available than the running instance.
1360 Reads the ``update_available`` AppSetting written by the background
1361 version-check service (computed once per check cycle, not per request).
1362 Falls back to a live comparison against ``latest_version`` if the flag has
1363 not been written yet (fresh install before the first background check).
1364 Returns False on any error. Must be called inside a request (or
1365 application) context.
1366 """
1367 try:
1368 from models import AppSetting, db # pyright: ignore[reportMissingImports]
1370 flag = db.session.get(AppSetting, "update_available")
1371 if flag is not None:
1372 return bool(flag.value == "true")
1374 # Fallback for fresh installs: compute from latest_version directly.
1375 from packaging.version import Version # pyright: ignore[reportMissingImports]
1377 current = os.environ.get("OPENHANGAR_VERSION", "development")
1378 if current == "development":
1379 return False
1380 latest_s = db.session.get(AppSetting, "latest_version")
1381 latest = latest_s.value if latest_s else None
1382 return bool(latest and Version(latest) > Version(current))
1383 except Exception: # noqa: BLE001 -- cosmetic nav-badge hint, degrade to "no update" on any error
1384 return False
1387def check_legacy_logbook_data() -> bool:
1388 """Return True when pre-unification logbook/crew data was preserved
1389 but not migrated during the ``flight_crew``/``pilot_logbook_entries``
1390 table removal.
1392 Reads the ``legacy_logbook_data_present`` AppSetting written by the
1393 Alembic migration that dropped those two tables — set only when either
1394 table still had rows at upgrade time (both tables are dropped outright,
1395 with no flag set, when they were already empty). Returns False on any
1396 error. Must be called inside a request (or application) context.
1397 """
1398 try:
1399 from models import AppSetting, db # pyright: ignore[reportMissingImports]
1401 flag = db.session.get(AppSetting, "legacy_logbook_data_present")
1402 return bool(flag is not None and flag.value == "true")
1403 except Exception: # noqa: BLE001 -- cosmetic nav-badge hint, degrade to "no legacy data" on any error
1404 return False
1407def require_role(*roles: str) -> Callable[..., Any]:
1408 """Decorator: abort 403 if the current user's role is not in *roles*."""
1410 def decorator(f: Callable[..., Any]) -> Callable[..., Any]:
1411 @wraps(f)
1412 def decorated(*args: Any, **kwargs: Any) -> Any:
1413 if current_user_role() not in roles:
1414 abort(403)
1415 return f(*args, **kwargs)
1417 return decorated
1419 return decorator
1422def require_pilot_access(f: Callable[..., Any]) -> Callable[..., Any]:
1423 """Decorator: abort 403 unless the user has pilot access.
1425 Pilot access is granted by ADMIN/OWNER/PILOT/STUDENT/INSTRUCTOR role,
1426 or by the per-user is_pilot capability flag.
1427 """
1429 @wraps(f)
1430 def decorated(*args: Any, **kwargs: Any) -> Any:
1431 from models import Role, User, db
1433 role = current_user_role()
1434 if role in (Role.ADMIN, Role.OWNER, Role.PILOT, Role.STUDENT, Role.INSTRUCTOR):
1435 return f(*args, **kwargs)
1436 uid = session.get("user_id")
1437 if uid:
1438 user = db.session.get(User, uid)
1439 if user and user.is_pilot:
1440 return f(*args, **kwargs)
1441 return abort(403)
1443 return decorated
1446def require_maint_access(f: Callable[..., Any]) -> Callable[..., Any]:
1447 """Decorator: abort 403 unless the user has maintenance access.
1449 Maintenance access is granted by ADMIN/OWNER/MAINTENANCE/INSTRUCTOR role,
1450 or by the per-user is_maintenance capability flag.
1451 """
1453 @wraps(f)
1454 def decorated(*args: Any, **kwargs: Any) -> Any:
1455 from models import Role, User, db
1457 role = current_user_role()
1458 if role in (Role.ADMIN, Role.OWNER, Role.MAINTENANCE, Role.INSTRUCTOR):
1459 return f(*args, **kwargs)
1460 uid = session.get("user_id")
1461 if uid:
1462 user = db.session.get(User, uid)
1463 if user and user.is_maintenance:
1464 return f(*args, **kwargs)
1465 return abort(403)
1467 return decorated
1470def tenant_pilots(tenant_id: int) -> list[tuple[int, str]]:
1471 """``(user_id, label)`` for the active users in *tenant_id* with pilot
1472 access (same rule as ``require_pilot_access``: a pilot-capable role or
1473 the ``is_pilot`` flag), sorted case-insensitively by label.
1475 The label is the user's display name; when two pilots share one, both
1476 get their e-mail local part appended so every label is unique and a
1477 picked suggestion maps back to exactly one account.
1478 """
1479 from models import Role, TenantUser, User, db
1480 from sqlalchemy import or_ # pyright: ignore[reportMissingImports]
1482 pilot_roles = (Role.ADMIN, Role.OWNER, Role.PILOT, Role.STUDENT, Role.INSTRUCTOR)
1483 users = (
1484 db.session.query(User)
1485 .join(TenantUser, TenantUser.user_id == User.id)
1486 .filter(TenantUser.tenant_id == tenant_id)
1487 .filter(User.is_active.is_(True))
1488 .filter(or_(TenantUser.role.in_(pilot_roles), User.is_pilot.is_(True)))
1489 .all()
1490 )
1491 name_counts: dict[str, int] = defaultdict(int)
1492 for u in users:
1493 name_counts[u.display_name.casefold()] += 1
1494 labelled = [
1495 (
1496 u.id,
1497 f"{u.display_name} ({u.email.split('@')[0]})"
1498 if name_counts[u.display_name.casefold()] > 1
1499 else u.display_name,
1500 )
1501 for u in users
1502 ]
1503 return sorted(labelled, key=lambda p: p[1].casefold())
1506def tenant_pilot_names(tenant_id: int) -> list[str]:
1507 """Plain display names of ``tenant_pilots`` — type-ahead suggestions for
1508 free-text crew name fields that never link to an account (e.g. the
1509 standalone pilot-logbook entry form's PIC name)."""
1510 return sorted({label for _uid, label in tenant_pilots(tenant_id)}, key=str.casefold)
1513def user_can_access_aircraft(aircraft_id: int) -> bool:
1514 """Return True when the current user may access this aircraft.
1516 ADMIN and OWNER bypass the check entirely. Other roles need either a
1517 UserAllAircraftAccess row (all-planes grant) or a per-aircraft
1518 UserAircraftAccess row.
1519 """
1520 from models import Role, TenantUser, UserAircraftAccess, UserAllAircraftAccess
1522 role = current_user_role()
1523 if role in (Role.ADMIN, Role.OWNER):
1524 return True
1525 uid = session.get("user_id")
1526 if not uid:
1527 return False
1528 tu = TenantUser.query.filter_by(user_id=uid).first()
1529 if (
1530 tu
1531 and UserAllAircraftAccess.query.filter_by(
1532 user_id=uid, tenant_id=tu.tenant_id
1533 ).first()
1534 ):
1535 return True
1536 return (
1537 UserAircraftAccess.query.filter_by(user_id=uid, aircraft_id=aircraft_id).first()
1538 is not None
1539 )
1542def accessible_aircraft(tenant_id: int, include_archived: bool = False) -> Any:
1543 """Return a query of Aircraft the current user is allowed to see.
1545 ADMIN and OWNER see every aircraft in the tenant. A user with a
1546 UserAllAircraftAccess row for the tenant also sees all aircraft.
1547 Other roles see only aircraft granted via UserAircraftAccess.
1549 Archived aircraft are excluded unless include_archived is True — pass it
1550 for views that must keep showing an archived aircraft's history.
1551 """
1552 from models import Aircraft, Role, UserAircraftAccess, UserAllAircraftAccess
1554 base = Aircraft.query.filter_by(tenant_id=tenant_id).order_by(Aircraft.registration)
1555 if not include_archived:
1556 base = base.filter(Aircraft.archived_at.is_(None))
1557 role = current_user_role()
1558 if role in (Role.ADMIN, Role.OWNER):
1559 return base
1560 uid = session.get("user_id")
1561 if not uid:
1562 from sqlalchemy import false
1564 return base.filter(false())
1565 if UserAllAircraftAccess.query.filter_by(user_id=uid, tenant_id=tenant_id).first():
1566 return base
1567 ids = [
1568 row.aircraft_id
1569 for row in (
1570 UserAircraftAccess.query.filter_by(user_id=uid)
1571 .with_entities(UserAircraftAccess.aircraft_id)
1572 .all()
1573 )
1574 ]
1575 if not ids:
1576 from sqlalchemy import false
1578 return base.filter(false())
1579 return base.filter(Aircraft.id.in_(ids))
1582def compute_aircraft_statuses(
1583 aircraft_list: Any,
1584 triggers: Any,
1585 hobbs_by_id: Any,
1586 landings_by_id: Any = None,
1587 flight_hours_by_id: Any = None,
1588) -> dict[int, str]:
1589 """Return {aircraft_id: 'grounded'|'overdue'|'due_soon'|'ok'} for every aircraft.
1591 Grounded (expired insurance or unresolved grounding snag) takes priority.
1592 Among maintenance: overdue > due_soon > ok — maintenance triggers,
1593 component TBO / calendar life limits, and expiring insurance all count.
1595 ``landings_by_id``/``flight_hours_by_id`` default to an empty mapping —
1596 callers that don't (yet) pass them just get 'ok' from any landings- or
1597 flight-hours-basis trigger, same as passing no ``hobbs_by_id`` entry
1598 does for engine-hours-basis ones.
1599 """
1600 from services.component_limits import fleet_limit_statuses
1602 if landings_by_id is None:
1603 landings_by_id = {}
1604 if flight_hours_by_id is None:
1605 flight_hours_by_id = {}
1607 by_aircraft = defaultdict(list)
1608 for t in triggers:
1609 by_aircraft[t.aircraft_id].append(t)
1610 limit_status_by_ac = fleet_limit_statuses(aircraft_list)
1612 result = {}
1613 for ac in aircraft_list:
1614 if ac.is_grounded:
1615 result[ac.id] = "grounded"
1616 continue
1617 hobbs = hobbs_by_id.get(ac.id)
1618 landings = landings_by_id.get(ac.id)
1619 flight_hours = flight_hours_by_id.get(ac.id)
1620 statuses = [
1621 t.status(
1622 current_engine_hours=hobbs,
1623 current_landings=landings,
1624 current_flight_hours=flight_hours,
1625 )
1626 for t in by_aircraft.get(ac.id, [])
1627 ]
1628 statuses.append(limit_status_by_ac.get(ac.id, "ok"))
1629 ins = ac.insurance_status
1630 if ins == "expiring_soon":
1631 statuses.append("due_soon")
1632 if "overdue" in statuses:
1633 result[ac.id] = "overdue"
1634 elif "due_soon" in statuses:
1635 result[ac.id] = "due_soon"
1636 else:
1637 result[ac.id] = "ok"
1638 return result