Coverage for app/utils.py: 11%
713 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-23 09:59 +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 []
142def _make_tile_background(
143 project: Callable[[float, float], tuple[int, int]],
144 min_lon: float,
145 max_lon: float,
146 min_lat: float,
147 max_lat: float,
148 canvas_w: int,
149 canvas_h: int,
150 openaip_key: str | None = None,
151 tile_cache: dict[Any, bytes] | None = None,
152 max_tiles: int = 36,
153) -> Any:
154 """Fetch OSM raster tiles and composite them into a background PIL Image.
156 Returns a PIL Image, or None on failure (caller falls back to plain fill).
157 Tiles are fetched from OpenStreetMap; zoom level is chosen automatically.
158 Max 36 tiles are fetched to keep export time reasonable.
160 tile_cache, if provided, is a dict keyed by (z, tx, ty) or ("opi", z, tx, ty)
161 holding raw PNG bytes so repeated calls across frames skip network fetches.
162 """
163 import urllib.request
165 from PIL import Image as _Img # pyright: ignore[reportMissingImports]
167 # Compute pixels-per-degree-lon from the projection function
168 mid_lat = (min_lat + max_lat) / 2.0
169 px0x = project(min_lon, mid_lat)[0]
170 px1x = project(min_lon + 1.0, mid_lat)[0]
171 scale_x = float(px1x - px0x) # canvas pixels per degree-lon
172 if scale_x <= 0:
173 return None
175 # Choose zoom so each tile is ~256 canvas pixels wide (clamped 2–14)
176 z = round(math.log2(max(scale_x * 360.0 / 256.0, 1.0)))
177 z = max(2, min(z, 14))
178 n = 2**z
180 def _lon_to_tx(lon: float) -> int:
181 return int(int((lon + 180.0) / 360.0 * n) % n)
183 def _lat_to_ty(lat: float) -> int:
184 lat_r = math.radians(max(-85.0, min(85.0, lat)))
185 return int(
186 (1.0 - math.log(math.tan(lat_r) + 1.0 / math.cos(lat_r)) / math.pi)
187 / 2.0
188 * n
189 )
191 def _tile_nw_lonlat(tx: int, ty: int) -> tuple[float, float]:
192 lon = tx * 360.0 / n - 180.0
193 lat = math.degrees(math.atan(math.sinh(math.pi * (1.0 - 2.0 * ty / n))))
194 return lon, lat
196 tx_min = _lon_to_tx(min_lon)
197 tx_max = _lon_to_tx(max_lon)
198 ty_min = _lat_to_ty(max_lat)
199 ty_max = _lat_to_ty(min_lat)
201 # Guard against tile count explosion
202 if (tx_max - tx_min + 1) * (ty_max - ty_min + 1) > max_tiles:
203 return None
205 bg = _Img.new("RGB", (canvas_w, canvas_h), _BG_COLOUR)
206 ua = "OpenHangar flight-logbook GIF export (https://github.com/e2jk/OpenHangar)"
208 for tx in range(tx_min, tx_max + 1):
209 for ty in range(ty_min, ty_max + 1):
210 try:
211 tx_w = tx % n
212 # Compute pixel bounds for this tile by projecting its NW corner
213 # and the NW corner of the tile to its SE — then size the resize
214 # to exactly span that gap plus 1 px overlap, eliminating seams
215 # caused by integer rounding of adjacent corner coordinates.
216 lon_nw, lat_nw = _tile_nw_lonlat(tx, ty)
217 lon_se, lat_se = _tile_nw_lonlat(tx + 1, ty + 1)
218 px, py = project(lon_nw, lat_nw)
219 px_se, py_se = project(lon_se, lat_se)
220 tile_w = max(1, px_se - px + 1)
221 tile_h = max(1, py_se - py + 1)
223 # Base map: CARTO light (same as web animation)
224 base_key = (z, tx_w, ty)
225 if tile_cache is not None and base_key in tile_cache:
226 raw = tile_cache[base_key]
227 else:
228 base_url = (
229 f"https://a.basemaps.cartocdn.com/light_all/{z}/{tx_w}/{ty}.png"
230 )
231 req = urllib.request.Request(base_url, headers={"User-Agent": ua})
232 with urllib.request.urlopen(req, timeout=5) as resp: # nosec B310 # base_url has a hardcoded https:// prefix
233 raw = resp.read()
234 if tile_cache is not None:
235 tile_cache[base_key] = raw
236 tile = _Img.open(io.BytesIO(raw)).convert("RGBA")
237 tile = tile.resize((tile_w, tile_h), _Img.Resampling.LANCZOS)
238 bg.paste(tile.convert("RGB"), (px, py))
239 # Aviation overlay: OpenAIP (only at zoom ≤ 14, requires key)
240 if openaip_key and z <= 14:
241 opi_key = ("opi", z, tx_w, ty)
242 if tile_cache is not None and opi_key in tile_cache:
243 opi_raw = tile_cache[opi_key]
244 else:
245 opi_url = f"https://api.tiles.openaip.net/api/data/openaip/{z}/{tx_w}/{ty}.png?apiKey={openaip_key}"
246 opi_req = urllib.request.Request(
247 opi_url, headers={"User-Agent": ua}
248 )
249 with urllib.request.urlopen(opi_req, timeout=5) as opi_resp: # nosec B310 # opi_url has a hardcoded https:// prefix
250 opi_raw = opi_resp.read()
251 if tile_cache is not None:
252 tile_cache[opi_key] = opi_raw
253 opi_tile = _Img.open(io.BytesIO(opi_raw)).convert("RGBA")
254 opi_tile = opi_tile.resize(
255 (tile_w, tile_h), _Img.Resampling.LANCZOS
256 )
257 bg.paste(opi_tile.convert("RGB"), (px, py), opi_tile)
258 except Exception as exc: # noqa: BLE001 -- one bad tile must not fail the whole image
259 _log.debug("OPI tile unavailable, leaving area as background: %s", exc)
261 return bg
264def _canvas_geo_bounds(
265 project_fn: Callable[[float, float], tuple[int, int]],
266 canvas_w: int,
267 canvas_h: int,
268 min_lon: float,
269 max_lon: float,
270 min_lat: float,
271 max_lat: float,
272) -> tuple[float, float, float, float]:
273 """Return (lon_min, lat_min, lon_max, lat_max) spanning the full canvas.
275 Expands the geographic bbox beyond the track bounding box so the full
276 canvas area is covered by map tiles, eliminating plain-background padding.
277 Returns the original bbox if the inverse projection cannot be computed.
278 """
279 mid_lat = (min_lat + max_lat) / 2.0
280 # Longitude is linear with pixel x in Mercator
281 px0, _ = project_fn(min_lon, mid_lat)
282 px1, _ = project_fn(min_lon + 1.0, mid_lat)
283 scale_x = float(px1 - px0)
284 if scale_x <= 0:
285 return min_lon, min_lat, max_lon, max_lat
286 c_lon_min = min_lon - float(px0) / scale_x
287 c_lon_max = min_lon + float(canvas_w - px0) / scale_x
289 # Latitude: derive Mercator scale from two known projection points
290 _, py_bot = project_fn(min_lon, min_lat) # larger py (bottom of canvas area)
291 _, py_top = project_fn(min_lon, max_lat) # smaller py (top of canvas area)
292 merc_bot = _mercator_y(min_lat)
293 merc_top = _mercator_y(max_lat)
294 merc_range = merc_top - merc_bot
295 py_range = float(py_bot - py_top)
296 if merc_range <= 0 or py_range <= 0:
297 return c_lon_min, min_lat, c_lon_max, max_lat
298 scale_y = py_range / merc_range # pixels per Mercator-y unit
299 # Derived: merc_y at canvas row py = merc_top + (py_top - py) / scale_y
300 merc_canvas_top = merc_top + float(py_top) / scale_y
301 merc_canvas_bot = merc_top + float(py_top - canvas_h) / scale_y
302 # Clamp to avoid math.atan overflow at extreme Mercator values
303 merc_canvas_top = max(-10.0, min(10.0, merc_canvas_top))
304 merc_canvas_bot = max(-10.0, min(10.0, merc_canvas_bot))
305 c_lat_max = math.degrees(2.0 * math.atan(math.exp(merc_canvas_top)) - math.pi / 2.0)
306 c_lat_min = math.degrees(2.0 * math.atan(math.exp(merc_canvas_bot)) - math.pi / 2.0)
307 return c_lon_min, max(-85.0, c_lat_min), c_lon_max, min(85.0, c_lat_max)
310def sort_tracks_oldest_first(
311 rows: list[dict[str, Any]],
312) -> list[dict[str, Any]]:
313 """Return rows sorted ascending by their 'date' string key (oldest first).
315 Both the web animation and the GIF export use this ordering so that
316 chronological playback and progressive opacity fading are consistent
317 across rendering environments. Call this once in the route handler and
318 pass the result to both the template context and generate_tracks_gif().
319 """
320 return sorted(rows, key=lambda r: r.get("date", ""))
323def generate_tracks_gif(
324 track_rows: list[dict[str, Any]],
325 _font_path: str = "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
326 _openaip_key: str | None = None,
327 canvas_w: int = _GIF_W,
328 canvas_h: int = _GIF_H,
329 high_res: bool = False,
330) -> bytes:
331 """Render an animated GIF of the flight tracks, oldest-first.
333 track_rows must already be sorted oldest-first; use sort_tracks_oldest_first().
334 Each frame adds one more track (cumulative) and re-fits the map to the
335 bounding box of tracks seen so far, creating a progressive zoom-out effect
336 that mirrors the web animation. Previous tracks are drawn in a lighter
337 colour; the newest one in vivid purple. Returns raw GIF bytes ready to
338 stream to the browser.
340 high_res=True doubles line widths, font sizes and padding, uses a 256-colour
341 palette, and raises the tile-fetch cap to 64 — producing a sharper map with
342 readable labels at the cost of a larger file.
343 """
344 from PIL import Image, ImageDraw, ImageFont # pyright: ignore[reportMissingImports]
346 _q_scale = 2 if high_res else 1
347 _pad = _GIF_PAD * _q_scale
348 _line_curr = 3 * _q_scale
349 _line_hist = 2 * _q_scale
350 _font_sz = 14 * _q_scale
351 _font_sz_sm = 11 * _q_scale
352 _txt_margin = 8 * _q_scale
353 _q_colors = 256 if high_res else 128
354 _max_tiles = 64 if high_res else 36
355 # Canvas-extent calls cover the full 1600×960 canvas which can require
356 # ~70–90 tiles at certain zoom levels (wide flat tracks at z=7), so use a
357 # higher cap than the track-bbox call. 128 = 2× the high-res track cap.
358 _max_tiles_canvas = _max_tiles * 2 if high_res else _max_tiles
360 # Pre-compute per-track coordinates (avoids re-parsing GeoJSON in the inner loop)
361 per_track_coords = [_coords_from_geojson(r.get("geojson")) for r in track_rows]
362 all_coords: list[tuple[float, float]] = [c for tc in per_track_coords for c in tc]
364 if len(all_coords) < 2:
365 # Fallback: single blank frame
366 img = Image.new("RGB", (canvas_w, canvas_h), _BG_COLOUR)
367 buf = io.BytesIO()
368 img.save(buf, format="GIF")
369 return buf.getvalue()
371 font: Any # PIL font type varies across Pillow versions
372 font_sm: Any
373 try:
374 font = ImageFont.truetype(_font_path, _font_sz)
375 font_sm = ImageFont.truetype(_font_path, _font_sz_sm)
376 except OSError:
377 font = font_sm = ImageFont.load_default()
379 def draw_shadow_text(draw: Any, text: str, font: Any) -> None:
380 draw.text(
381 (_txt_margin + 1, _txt_margin + 1), text, fill=(255, 255, 255), font=font
382 )
383 draw.text((_txt_margin, _txt_margin), text, fill=(40, 40, 60), font=font)
385 def draw_track(
386 draw: Any,
387 coords: list[tuple[float, float]],
388 colour: tuple[int, int, int],
389 width: int,
390 project_fn: Callable[[float, float], tuple[int, int]],
391 ) -> None:
392 pts = [project_fn(lon, lat) for lon, lat in coords]
393 if len(pts) >= 2:
394 draw.line(pts, fill=colour, width=width)
395 r = width + 2 * _q_scale
396 sx, sy = pts[0]
397 draw.ellipse([sx - r, sy - r, sx + r, sy + r], fill=(40, 160, 60))
398 ex, ey = pts[-1]
399 draw.ellipse([ex - r, ey - r, ex + r, ey + r], fill=(200, 40, 40))
401 # Shared tile cache: (z, tx, ty) → raw PNG bytes; ("opi", z, tx, ty) for OpenAIP.
402 # Avoids re-fetching tiles that appear in multiple frames at the same zoom level.
403 tile_cache: dict[Any, bytes] = {}
405 def _frame_bg(
406 project_fn: Callable[[float, float], tuple[int, int]],
407 min_lon: float,
408 max_lon: float,
409 min_lat: float,
410 max_lat: float,
411 ) -> Any:
412 fetch_lon_min, fetch_lat_min, fetch_lon_max, fetch_lat_max = (
413 min_lon,
414 min_lat,
415 max_lon,
416 max_lat,
417 )
418 if high_res:
419 fetch_lon_min, fetch_lat_min, fetch_lon_max, fetch_lat_max = (
420 _canvas_geo_bounds(
421 project_fn,
422 canvas_w,
423 canvas_h,
424 min_lon,
425 max_lon,
426 min_lat,
427 max_lat,
428 )
429 )
430 bg = _make_tile_background(
431 project_fn,
432 fetch_lon_min,
433 fetch_lon_max,
434 fetch_lat_min,
435 fetch_lat_max,
436 canvas_w,
437 canvas_h,
438 openaip_key=_openaip_key,
439 tile_cache=tile_cache,
440 max_tiles=_max_tiles_canvas,
441 )
442 if bg is None and high_res:
443 # Canvas extent still exceeded the raised cap; fall back to track bbox
444 bg = _make_tile_background(
445 project_fn,
446 min_lon,
447 max_lon,
448 min_lat,
449 max_lat,
450 canvas_w,
451 canvas_h,
452 openaip_key=_openaip_key,
453 tile_cache=tile_cache,
454 max_tiles=_max_tiles,
455 )
456 return (
457 bg.copy()
458 if bg is not None
459 else Image.new("RGB", (canvas_w, canvas_h), _BG_COLOUR)
460 )
462 frames: list[Any] = []
463 durations: list[int] = []
464 accumulated_coords: list[tuple[float, float]] = []
466 for frame_idx in range(len(track_rows)):
467 accumulated_coords.extend(per_track_coords[frame_idx])
468 proj_result = _build_gif_projection(
469 accumulated_coords, canvas_w=canvas_w, canvas_h=canvas_h, pad=_pad
470 )
471 if proj_result is None:
472 continue # not enough coords yet (e.g. leading rows with no geojson)
474 project_fn, (f_min_lon, f_min_lat, f_max_lon, f_max_lat) = proj_result
475 img = _frame_bg(project_fn, f_min_lon, f_max_lon, f_min_lat, f_max_lat)
476 draw = ImageDraw.Draw(img)
478 for i in range(frame_idx):
479 draw_track(
480 draw, per_track_coords[i], _HISTORY_COLOUR, _line_hist, project_fn
481 )
482 draw_track(
483 draw, per_track_coords[frame_idx], _TRACK_COLOUR, _line_curr, project_fn
484 )
486 row = track_rows[frame_idx]
487 label = f"{row.get('date', '')} {row.get('dep', '')} → {row.get('arr', '')}"
488 draw_shadow_text(draw, label, font)
489 draw.text(
490 (_txt_margin, canvas_h - _font_sz_sm - _txt_margin),
491 f"{frame_idx + 1} / {len(track_rows)}",
492 fill=(80, 80, 100),
493 font=font_sm,
494 )
496 frames.append(img)
497 durations.append(_GIF_STEP_MS)
499 # Final frame: all tracks at equal weight using the full bounding box, longer hold
500 if frames:
501 proj_result_final = _build_gif_projection(
502 all_coords, canvas_w=canvas_w, canvas_h=canvas_h, pad=_pad
503 )
504 # proj_result_final is guaranteed non-None: all_coords has ≥ 2 points (checked above)
505 project_final, (f_min_lon, f_min_lat, f_max_lon, f_max_lat) = proj_result_final # type: ignore[misc]
506 img = _frame_bg(project_final, f_min_lon, f_max_lon, f_min_lat, f_max_lat)
507 draw = ImageDraw.Draw(img)
508 for tc in per_track_coords:
509 draw_track(draw, tc, _TRACK_COLOUR, _line_hist, project_final)
510 draw_shadow_text(draw, f"All {len(track_rows)} tracks", font)
511 frames.append(img)
512 durations.append(_GIF_HOLD_MS)
514 # Quantise all frames to a shared 128-colour palette built from the last frame
515 # (widest view, most visually representative). A single global palette means
516 # identical background areas compress very well with GIF's LZW codec.
517 palette_src = frames[-1].quantize(colors=_q_colors, dither=Image.Dither.NONE)
518 quantized: list[Any] = [
519 f.quantize(palette=palette_src, dither=Image.Dither.NONE) for f in frames
520 ]
522 buf = io.BytesIO()
523 quantized[0].save(
524 buf,
525 format="GIF",
526 save_all=True,
527 append_images=quantized[1:],
528 loop=0,
529 duration=durations,
530 optimize=True,
531 )
532 return buf.getvalue()
535def generate_single_track_image(
536 geojson: dict[str, Any] | None,
537 date: str = "",
538 dep: str = "",
539 arr: str = "",
540 _font_path: str = "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
541 _openaip_key: str | None = None,
542 canvas_w: int = _GIF_W,
543 canvas_h: int = _GIF_H,
544 high_res: bool = False,
545) -> bytes:
546 """Render a single GPS track as a static PNG.
548 Returns raw PNG bytes. If the track has fewer than 2 points a blank
549 canvas is returned so the caller can always stream a valid image.
550 """
551 from PIL import Image, ImageDraw, ImageFont # pyright: ignore[reportMissingImports]
553 all_coords = _coords_from_geojson(geojson)
555 _q_scale = 2 if high_res else 1
556 _pad = _GIF_PAD * _q_scale
557 _line_w = 3 * _q_scale
558 _font_sz = 14 * _q_scale
559 _txt_margin = 8 * _q_scale
560 _max_tiles = 64 if high_res else 36
561 _max_tiles_canvas = _max_tiles * 2 if high_res else _max_tiles
563 def _blank_png() -> bytes:
564 buf = io.BytesIO()
565 Image.new("RGB", (canvas_w, canvas_h), _BG_COLOUR).save(buf, format="PNG")
566 return buf.getvalue()
568 if len(all_coords) < 2:
569 return _blank_png()
571 proj_result = _build_gif_projection(
572 all_coords, canvas_w=canvas_w, canvas_h=canvas_h, pad=_pad
573 )
574 if proj_result is None:
575 return _blank_png()
577 project_fn, (f_min_lon, f_min_lat, f_max_lon, f_max_lat) = proj_result
579 tile_cache: dict[Any, bytes] = {}
580 fetch_lon_min, fetch_lat_min, fetch_lon_max, fetch_lat_max = (
581 f_min_lon,
582 f_min_lat,
583 f_max_lon,
584 f_max_lat,
585 )
586 if high_res:
587 fetch_lon_min, fetch_lat_min, fetch_lon_max, fetch_lat_max = _canvas_geo_bounds(
588 project_fn, canvas_w, canvas_h, f_min_lon, f_max_lon, f_min_lat, f_max_lat
589 )
590 bg = _make_tile_background(
591 project_fn,
592 fetch_lon_min,
593 fetch_lon_max,
594 fetch_lat_min,
595 fetch_lat_max,
596 canvas_w,
597 canvas_h,
598 openaip_key=_openaip_key,
599 tile_cache=tile_cache,
600 max_tiles=_max_tiles_canvas,
601 )
602 if bg is None and high_res:
603 bg = _make_tile_background(
604 project_fn,
605 f_min_lon,
606 f_max_lon,
607 f_min_lat,
608 f_max_lat,
609 canvas_w,
610 canvas_h,
611 openaip_key=_openaip_key,
612 tile_cache=tile_cache,
613 max_tiles=_max_tiles,
614 )
615 img = (
616 bg.copy()
617 if bg is not None
618 else Image.new("RGB", (canvas_w, canvas_h), _BG_COLOUR)
619 )
621 draw = ImageDraw.Draw(img)
622 pts = [project_fn(lon, lat) for lon, lat in all_coords]
623 if len(pts) >= 2:
624 draw.line(pts, fill=_TRACK_COLOUR, width=_line_w)
625 r = _line_w + 2 * _q_scale
626 sx, sy = pts[0]
627 draw.ellipse([sx - r, sy - r, sx + r, sy + r], fill=(40, 160, 60))
628 ex, ey = pts[-1]
629 draw.ellipse([ex - r, ey - r, ex + r, ey + r], fill=(200, 40, 40))
631 label = f"{date} {dep} → {arr}".strip(" →").strip()
632 if label:
633 try:
634 font: Any = ImageFont.truetype(_font_path, _font_sz)
635 except OSError:
636 font = ImageFont.load_default()
637 draw.text(
638 (_txt_margin + 1, _txt_margin + 1), label, fill=(255, 255, 255), font=font
639 )
640 draw.text((_txt_margin, _txt_margin), label, fill=(40, 40, 60), font=font)
642 buf = io.BytesIO()
643 img.save(buf, format="PNG")
644 return buf.getvalue()
647def generate_single_track_gif(
648 geojson: dict[str, Any] | None,
649 date: str = "",
650 dep: str = "",
651 arr: str = "",
652 _font_path: str = "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
653 _openaip_key: str | None = None,
654 canvas_w: int = _GIF_W,
655 canvas_h: int = _GIF_H,
656 high_res: bool = False,
657) -> bytes:
658 """Render an animated GIF of a single flight track drawn progressively.
660 The track's coordinate array is split into N equal chunks (3–10 depending
661 on track length). Each frame draws all chunks revealed so far, fitting the
662 map bounding box to the accumulated coords so the view zooms out as the
663 route grows — mirroring the per-flight behaviour of the web Animate button.
665 Falls back to a single-frame GIF wrapping the still PNG when the track has
666 fewer than 20 GPS points (too sparse for meaningful animation).
667 """
668 from PIL import Image, ImageDraw, ImageFont # pyright: ignore[reportMissingImports]
670 all_coords = _coords_from_geojson(geojson)
672 _q_scale = 2 if high_res else 1
673 _pad = _GIF_PAD * _q_scale
674 _line_curr = 3 * _q_scale
675 _line_hist = 2 * _q_scale
676 _font_sz = 14 * _q_scale
677 _font_sz_sm = 11 * _q_scale
678 _txt_margin = 8 * _q_scale
679 _q_colors = 256 if high_res else 128
680 _max_tiles = 64 if high_res else 36
681 _max_tiles_canvas = _max_tiles * 2 if high_res else _max_tiles
683 def _blank_gif() -> bytes:
684 buf = io.BytesIO()
685 Image.new("RGB", (canvas_w, canvas_h), _BG_COLOUR).save(buf, format="GIF")
686 return buf.getvalue()
688 if len(all_coords) < 2:
689 return _blank_gif()
691 # Sparse-track fallback: wrap the still image in a single-frame GIF
692 if len(all_coords) < 20:
693 png = generate_single_track_image(
694 geojson,
695 date=date,
696 dep=dep,
697 arr=arr,
698 _font_path=_font_path,
699 _openaip_key=_openaip_key,
700 canvas_w=canvas_w,
701 canvas_h=canvas_h,
702 high_res=high_res,
703 )
704 img = Image.open(io.BytesIO(png)).convert("RGB")
705 buf = io.BytesIO()
706 img.quantize(colors=_q_colors, dither=Image.Dither.NONE).save(buf, format="GIF")
707 return buf.getvalue()
709 # Split coords into N equal chunks (3–10)
710 n_chunks = max(3, min(10, len(all_coords) // 10))
711 chunk_size = len(all_coords) // n_chunks
712 chunks: list[list[tuple[float, float]]] = [
713 list(all_coords[i * chunk_size : (i + 1) * chunk_size])
714 for i in range(n_chunks - 1)
715 ]
716 chunks.append(
717 list(all_coords[(n_chunks - 1) * chunk_size :])
718 ) # last catches remainder
720 try:
721 font: Any = ImageFont.truetype(_font_path, _font_sz)
722 font_sm: Any = ImageFont.truetype(_font_path, _font_sz_sm)
723 except OSError:
724 font = font_sm = ImageFont.load_default()
726 def draw_shadow_text(draw: Any, text: str, fnt: Any) -> None:
727 draw.text(
728 (_txt_margin + 1, _txt_margin + 1), text, fill=(255, 255, 255), font=fnt
729 )
730 draw.text((_txt_margin, _txt_margin), text, fill=(40, 40, 60), font=fnt)
732 def draw_track(
733 draw: Any,
734 coords: list[tuple[float, float]],
735 colour: tuple[int, int, int],
736 width: int,
737 project_fn: Callable[[float, float], tuple[int, int]],
738 ) -> None:
739 pts = [project_fn(lon, lat) for lon, lat in coords]
740 if len(pts) >= 2:
741 draw.line(pts, fill=colour, width=width)
742 r = width + 2 * _q_scale
743 sx, sy = pts[0]
744 draw.ellipse([sx - r, sy - r, sx + r, sy + r], fill=(40, 160, 60))
745 ex, ey = pts[-1]
746 draw.ellipse([ex - r, ey - r, ex + r, ey + r], fill=(200, 40, 40))
748 tile_cache: dict[Any, bytes] = {}
750 def _frame_bg(
751 project_fn: Callable[[float, float], tuple[int, int]],
752 min_lon: float,
753 max_lon: float,
754 min_lat: float,
755 max_lat: float,
756 ) -> Any:
757 fetch_lon_min, fetch_lat_min, fetch_lon_max, fetch_lat_max = (
758 min_lon,
759 min_lat,
760 max_lon,
761 max_lat,
762 )
763 if high_res:
764 fetch_lon_min, fetch_lat_min, fetch_lon_max, fetch_lat_max = (
765 _canvas_geo_bounds(
766 project_fn, canvas_w, canvas_h, min_lon, max_lon, min_lat, max_lat
767 )
768 )
769 bg = _make_tile_background(
770 project_fn,
771 fetch_lon_min,
772 fetch_lon_max,
773 fetch_lat_min,
774 fetch_lat_max,
775 canvas_w,
776 canvas_h,
777 openaip_key=_openaip_key,
778 tile_cache=tile_cache,
779 max_tiles=_max_tiles_canvas,
780 )
781 if bg is None and high_res:
782 bg = _make_tile_background(
783 project_fn,
784 min_lon,
785 max_lon,
786 min_lat,
787 max_lat,
788 canvas_w,
789 canvas_h,
790 openaip_key=_openaip_key,
791 tile_cache=tile_cache,
792 max_tiles=_max_tiles,
793 )
794 return (
795 bg.copy()
796 if bg is not None
797 else Image.new("RGB", (canvas_w, canvas_h), _BG_COLOUR)
798 )
800 label = f"{date} {dep} → {arr}".strip(" →").strip()
802 frames: list[Any] = []
803 durations: list[int] = []
804 accumulated: list[tuple[float, float]] = []
806 for chunk_idx, chunk in enumerate(chunks):
807 accumulated.extend(chunk)
808 proj_result = _build_gif_projection(
809 accumulated, canvas_w=canvas_w, canvas_h=canvas_h, pad=_pad
810 )
811 if proj_result is None:
812 continue
814 project_fn, (f_min_lon, f_min_lat, f_max_lon, f_max_lat) = proj_result
815 img = _frame_bg(project_fn, f_min_lon, f_max_lon, f_min_lat, f_max_lat)
816 draw = ImageDraw.Draw(img)
818 for i in range(chunk_idx):
819 draw_track(draw, chunks[i], _HISTORY_COLOUR, _line_hist, project_fn)
820 draw_track(draw, chunk, _TRACK_COLOUR, _line_curr, project_fn)
822 if label:
823 draw_shadow_text(draw, label, font)
824 draw.text(
825 (_txt_margin, canvas_h - _font_sz_sm - _txt_margin),
826 f"{chunk_idx + 1} / {n_chunks}",
827 fill=(80, 80, 100),
828 font=font_sm,
829 )
830 frames.append(img)
831 durations.append(_GIF_STEP_MS)
833 if not frames:
834 return _blank_gif()
836 # Final hold frame: full track, all chunks at equal weight
837 proj_result_final = _build_gif_projection(
838 list(all_coords), canvas_w=canvas_w, canvas_h=canvas_h, pad=_pad
839 )
840 # guaranteed non-None: all_coords has ≥ 20 points
841 project_final, (f_min_lon, f_min_lat, f_max_lon, f_max_lat) = proj_result_final # type: ignore[misc]
842 img = _frame_bg(project_final, f_min_lon, f_max_lon, f_min_lat, f_max_lat)
843 draw = ImageDraw.Draw(img)
844 for chunk in chunks:
845 draw_track(draw, chunk, _TRACK_COLOUR, _line_hist, project_final)
846 if label:
847 draw_shadow_text(draw, label, font)
848 frames.append(img)
849 durations.append(_GIF_HOLD_MS)
851 palette_src = frames[-1].quantize(colors=_q_colors, dither=Image.Dither.NONE)
852 quantized: list[Any] = [
853 f.quantize(palette=palette_src, dither=Image.Dither.NONE) for f in frames
854 ]
855 buf = io.BytesIO()
856 quantized[0].save(
857 buf,
858 format="GIF",
859 save_all=True,
860 append_images=quantized[1:],
861 loop=0,
862 duration=durations,
863 optimize=True,
864 )
865 return buf.getvalue()
868@functools.lru_cache(maxsize=1)
869def _load_aircraft_types() -> dict[str, tuple[str, str]]:
870 """Return {type_designator: (manufacturer, model)} from aircraft_types.csv."""
871 path = os.path.join(os.path.dirname(__file__), "data", "aircraft_types.csv")
872 result: dict[str, tuple[str, str]] = {}
873 try:
874 with open(path, newline="", encoding="utf-8") as f:
875 for row in csv.DictReader(f):
876 des = row.get("type_designator", "").strip().upper()
877 mfr = row.get("manufacturer", "").strip()
878 model = row.get("model", "").strip()
879 if des:
880 result[des] = (mfr, model)
881 except OSError as exc:
882 _log.warning("aircraft_types.csv not found: %s", exc)
883 return result
886def _load_aircraft_type_variants() -> list[tuple[str, str, str, str]]:
887 """Return all (type_designator, full_name, manufacturer, model) tuples — one per CSV row.
889 Unlike _load_aircraft_types(), duplicate designators are preserved so
890 the search endpoint can surface every variant (e.g. all PA-28-181 models
891 that share the P28A ICAO code). manufacturer and model are kept separate
892 so callers can pre-fill form fields for the exact selected variant rather
893 than an arbitrary one sharing the same ICAO code.
894 """
895 path = os.path.join(os.path.dirname(__file__), "data", "aircraft_types.csv")
896 result: list[tuple[str, str, str, str]] = []
897 seen: set[tuple[str, str]] = set()
898 try:
899 with open(path, newline="", encoding="utf-8") as f:
900 for row in csv.DictReader(f):
901 des = row.get("type_designator", "").strip().upper()
902 mfr = row.get("manufacturer", "").strip()
903 model = row.get("model", "").strip()
904 name = f"{mfr} {model}".strip()
905 if des and (des, name) not in seen:
906 result.append((des, name, mfr, model))
907 seen.add((des, name))
908 except OSError as exc:
909 _log.warning("aircraft_types.csv not found: %s", exc)
910 return result
913@functools.lru_cache(maxsize=1)
914def _load_aircraft_type_engine_data() -> dict[str, tuple[int, str]]:
915 """Return {type_designator: (engine_count, engine_type)} from aircraft_types.csv.
917 For designators with multiple rows (variants), uses data from the first row.
918 """
919 path = os.path.join(os.path.dirname(__file__), "data", "aircraft_types.csv")
920 result: dict[str, tuple[int, str]] = {}
921 try:
922 with open(path, newline="", encoding="utf-8") as f:
923 for row in csv.DictReader(f):
924 des = row.get("type_designator", "").strip().upper()
925 if not des or des in result:
926 continue
927 try:
928 ec = int(row.get("engine_count", "1"))
929 except ValueError:
930 ec = 1
931 et = row.get("engine_type", "").strip()
932 result[des] = (ec, et)
933 except OSError as exc:
934 _log.warning("aircraft_types.csv not found: %s", exc)
935 return result
938def get_aircraft_type_engine_info(icao_code: str) -> tuple[int, str] | None:
939 """Return (engine_count, engine_type) for the given ICAO code, or None."""
940 return _load_aircraft_type_engine_data().get(icao_code.strip().upper())
943@functools.lru_cache(maxsize=1)
944def _build_model_name_prefix_lookup() -> dict[str, str]:
945 """Return {compact_first_word_of_model: icao_code} for resolve_aircraft_type_icao fallback.
947 Enables matching pilot-logbook type strings like "DR401" or "PA28-161 TDI"
948 against ICAO codes like "DR40" or "P28A" via the CSV model-name column.
949 """
950 path = os.path.join(os.path.dirname(__file__), "data", "aircraft_types.csv")
951 result: dict[str, str] = {}
952 try:
953 with open(path, newline="", encoding="utf-8") as f:
954 for row in csv.DictReader(f):
955 des = row.get("type_designator", "").strip().upper()
956 model = row.get("model", "").strip()
957 if not des or not model:
958 continue
959 first_word = model.split()[0]
960 key = first_word.replace("-", "").replace(" ", "").upper()
961 if key and key not in result:
962 result[key] = des
963 except OSError as exc:
964 _log.warning("aircraft_types.csv not found: %s", exc)
965 return result
968@functools.lru_cache(maxsize=1)
969def _sorted_model_prefix_keys() -> list[str]:
970 """Keys of _build_model_name_prefix_lookup(), longest-first.
972 Cached separately from the lookup dict itself: found by fuzzing that
973 resolve_aircraft_type_icao() re-sorted these ~1-2k keys on *every* call
974 that reaches the fallback branch (i.e. almost every call — an exact ICAO
975 code is rarely what a user types or a logbook CSV column holds), even
976 though the lookup dict underneath never changes between calls.
977 """
978 return sorted(_build_model_name_prefix_lookup(), key=len, reverse=True)
981def resolve_aircraft_type_icao(aircraft_type: str | None) -> str | None:
982 """Return the matching ICAO type designator for *aircraft_type*, or None."""
983 if not aircraft_type:
984 return None
985 types = _load_aircraft_types()
986 norm = aircraft_type.strip().upper()
987 if norm in types:
988 return norm
989 # Try stripping hyphens and spaces (e.g. "PA-28" → "PA28")
990 compact = norm.replace("-", "").replace(" ", "")
991 if compact in types:
992 return compact
993 # Try matching against the first word of each CSV model name (longest key wins).
994 # e.g. "DR401" startswith "DR401" (from "DR-401 135CDI") → DR40
995 # "PA28161TDI" startswith "PA28161" (from "PA-28-161 Cherokee…") → P28A
996 # Guard: reject if the unmatched tail starts with a digit — that signals a
997 # different model number (e.g. "C172RG" wrongly matching key "C17").
998 prefix_lookup = _build_model_name_prefix_lookup()
999 for key in _sorted_model_prefix_keys():
1000 if len(key) < 4:
1001 break # keys are sorted longest-first; stop once they're too short
1002 if compact.startswith(key):
1003 tail = compact[len(key) :]
1004 if tail and tail[0].isdigit():
1005 continue # "C172RG" matching "C17" — tail "2RG" starts with digit
1006 return prefix_lookup[key]
1007 return None
1010@functools.lru_cache(maxsize=1)
1011def _load_airport_names() -> dict[str, str]:
1012 """Return {ICAO ident: airport name} for all airports in airports.csv."""
1013 path = os.path.join(os.path.dirname(__file__), "data", "airports.csv")
1014 result: dict[str, str] = {}
1015 try:
1016 with open(path, newline="", encoding="utf-8") as f:
1017 for row in csv.DictReader(f):
1018 ident = row.get("ident", "").strip()
1019 name = row.get("name", "").strip()
1020 if ident and name:
1021 result[ident] = name
1022 except OSError as exc:
1023 _log.warning("airports.csv not found: %s", exc)
1024 return result
1027@functools.lru_cache(maxsize=1)
1028def _load_airport_geo() -> dict[str, dict[str, Any]]:
1029 """Return {ICAO ident: {lat, lon, elevation_ft, country, region, name}}
1030 for all airports in airports.csv that have coordinates. Used by the
1031 pilot flight-statistics report (distance, farthest points, elevation
1032 extremes) — kept separate from _load_airport_names() since most callers
1033 only need the name."""
1034 path = os.path.join(os.path.dirname(__file__), "data", "airports.csv")
1035 result: dict[str, dict[str, Any]] = {}
1036 try:
1037 with open(path, newline="", encoding="utf-8") as f:
1038 for row in csv.DictReader(f):
1039 ident = row.get("ident", "").strip()
1040 lat_raw = row.get("latitude_deg", "").strip()
1041 lon_raw = row.get("longitude_deg", "").strip()
1042 if not ident or not lat_raw or not lon_raw:
1043 continue
1044 try:
1045 lat = float(lat_raw)
1046 lon = float(lon_raw)
1047 except ValueError:
1048 continue
1049 elev_raw = row.get("elevation_ft", "").strip()
1050 result[ident] = {
1051 "lat": lat,
1052 "lon": lon,
1053 "elevation_ft": float(elev_raw) if elev_raw else None,
1054 "country": row.get("iso_country", "").strip(),
1055 "region": row.get("iso_region", "").strip(),
1056 "name": row.get("name", "").strip(),
1057 }
1058 except OSError as exc:
1059 _log.warning("airports.csv not found: %s", exc)
1060 return result
1063_EARTH_RADIUS_NM = 3440.065
1066def haversine_nm(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
1067 """Great-circle distance between two points, in nautical miles."""
1068 p1, p2 = math.radians(lat1), math.radians(lat2)
1069 dphi = math.radians(lat2 - lat1)
1070 dlambda = math.radians(lon2 - lon1)
1071 a = (
1072 math.sin(dphi / 2) ** 2
1073 + math.cos(p1) * math.cos(p2) * math.sin(dlambda / 2) ** 2
1074 )
1075 return _EARTH_RADIUS_NM * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
1078_alog = logging.getLogger("openhangar.activity")
1081def _sl(value: object) -> str:
1082 """Sanitize a value for log output — strips CR/LF to prevent log injection (CWE-117)."""
1083 return str(value).replace("\r\n", "").replace("\n", "").replace("\r", "")
1086def activity(event: str, **fields: object) -> None:
1087 """Emit a structured [ACTIVITY] log entry with user_id and ip automatically included."""
1088 from flask import request, session
1090 uid = session.get("user_id", "")
1091 ip = request.remote_addr or ""
1092 parts = [f"[ACTIVITY] {event}", f"user_id={_sl(uid)}", f"ip={_sl(ip)}"]
1093 parts.extend(f"{k}={_sl(v)}" for k, v in fields.items())
1094 _alog.info(" ".join(parts))
1097class AircraftRefConverter(BaseConverter):
1098 """URL converter for the ``aircraft_id`` slot: accepts either the numeric
1099 primary key (unchanged, always works) or the aircraft's registration
1100 (e.g. ``OO-GRN``), so routes like ``/aircraft/<aircraft_id>/flights``
1101 also resolve ``/aircraft/OO-GRN/flights``.
1103 Authorization is unaffected: this only resolves the URL segment to a
1104 primary key (or the reverse, for link generation) — every view still
1105 does its own tenant-scoped lookup exactly as before, so a registration
1106 belonging to another tenant 404s the same way a wrong numeric id does.
1108 Registrations containing '/' or spaces (rare, but not forbidden by the
1109 model) are sanitized the same way upload filenames already are
1110 elsewhere in this blueprint; such an aircraft simply isn't reachable via
1111 its pretty URL (only via the numeric id, which always works).
1112 """
1114 regex = r"[^/]+"
1116 def to_python(self, value: str) -> int:
1117 if value.isdigit():
1118 return int(value)
1120 from models import Aircraft, db
1122 needle = value.upper()
1123 ac = (
1124 Aircraft.query.filter(
1125 db.func.upper(
1126 db.func.replace(
1127 db.func.replace(Aircraft.registration, "/", "-"), " ", "-"
1128 )
1129 )
1130 == needle
1131 )
1132 .order_by(Aircraft.id)
1133 .first()
1134 )
1135 if ac is None:
1136 raise ValidationError()
1137 return int(ac.id)
1139 def to_url(self, value: Any) -> str:
1140 if isinstance(value, str) and not value.isdigit():
1141 return super().to_url(value)
1143 from models import Aircraft, db
1145 ac = db.session.get(Aircraft, int(value))
1146 reg = ac.registration if ac and ac.registration else str(value)
1147 safe_reg = reg.replace("/", "-").replace(" ", "-")
1148 return super().to_url(safe_reg)
1151def login_required(f: Callable[..., Any]) -> Callable[..., Any]:
1152 """Redirect unauthenticated users to the login page."""
1154 @wraps(f)
1155 def decorated(*args: Any, **kwargs: Any) -> Any:
1156 if not session.get("user_id"):
1157 return redirect(url_for("auth.login"))
1158 return f(*args, **kwargs)
1160 return decorated
1163def require_instance_admin(f: Callable[..., Any]) -> Callable[..., Any]:
1164 """Abort 403 unless the current user is the instance admin."""
1166 @wraps(f)
1167 def decorated(*args: Any, **kwargs: Any) -> Any:
1168 from models import User, db
1170 user_id = session.get("user_id")
1171 if not user_id:
1172 return redirect(url_for("auth.login"))
1173 user = db.session.get(User, user_id)
1174 if not user or not user.is_instance_admin:
1175 abort(403)
1176 return f(*args, **kwargs)
1178 return decorated
1181def current_user_role() -> str | None:
1182 """Return the Role of the current user in their tenant, or None."""
1183 from models import TenantUser
1185 user_id = session.get("user_id")
1186 if not user_id:
1187 return None
1188 tu = TenantUser.query.filter_by(user_id=user_id).first()
1189 return tu.role if tu else None
1192def check_update_available() -> bool:
1193 """Return True when a newer release is available than the running instance.
1195 Reads the ``update_available`` AppSetting written by the background
1196 version-check service (computed once per check cycle, not per request).
1197 Falls back to a live comparison against ``latest_version`` if the flag has
1198 not been written yet (fresh install before the first background check).
1199 Returns False on any error. Must be called inside a request (or
1200 application) context.
1201 """
1202 try:
1203 from models import AppSetting, db # pyright: ignore[reportMissingImports]
1205 flag = db.session.get(AppSetting, "update_available")
1206 if flag is not None:
1207 return bool(flag.value == "true")
1209 # Fallback for fresh installs: compute from latest_version directly.
1210 from packaging.version import Version # pyright: ignore[reportMissingImports]
1212 current = os.environ.get("OPENHANGAR_VERSION", "development")
1213 if current == "development":
1214 return False
1215 latest_s = db.session.get(AppSetting, "latest_version")
1216 latest = latest_s.value if latest_s else None
1217 return bool(latest and Version(latest) > Version(current))
1218 except Exception: # noqa: BLE001 -- cosmetic nav-badge hint, degrade to "no update" on any error
1219 return False
1222def check_legacy_logbook_data() -> bool:
1223 """Return True when pre-unification logbook/crew data was preserved
1224 but not migrated during the ``flight_crew``/``pilot_logbook_entries``
1225 table removal.
1227 Reads the ``legacy_logbook_data_present`` AppSetting written by the
1228 Alembic migration that dropped those two tables — set only when either
1229 table still had rows at upgrade time (both tables are dropped outright,
1230 with no flag set, when they were already empty). Returns False on any
1231 error. Must be called inside a request (or application) context.
1232 """
1233 try:
1234 from models import AppSetting, db # pyright: ignore[reportMissingImports]
1236 flag = db.session.get(AppSetting, "legacy_logbook_data_present")
1237 return bool(flag is not None and flag.value == "true")
1238 except Exception: # noqa: BLE001 -- cosmetic nav-badge hint, degrade to "no legacy data" on any error
1239 return False
1242def require_role(*roles: str) -> Callable[..., Any]:
1243 """Decorator: abort 403 if the current user's role is not in *roles*."""
1245 def decorator(f: Callable[..., Any]) -> Callable[..., Any]:
1246 @wraps(f)
1247 def decorated(*args: Any, **kwargs: Any) -> Any:
1248 if current_user_role() not in roles:
1249 abort(403)
1250 return f(*args, **kwargs)
1252 return decorated
1254 return decorator
1257def require_pilot_access(f: Callable[..., Any]) -> Callable[..., Any]:
1258 """Decorator: abort 403 unless the user has pilot access.
1260 Pilot access is granted by ADMIN/OWNER/PILOT/STUDENT/INSTRUCTOR role,
1261 or by the per-user is_pilot capability flag.
1262 """
1264 @wraps(f)
1265 def decorated(*args: Any, **kwargs: Any) -> Any:
1266 from models import Role, User, db
1268 role = current_user_role()
1269 if role in (Role.ADMIN, Role.OWNER, Role.PILOT, Role.STUDENT, Role.INSTRUCTOR):
1270 return f(*args, **kwargs)
1271 uid = session.get("user_id")
1272 if uid:
1273 user = db.session.get(User, uid)
1274 if user and user.is_pilot:
1275 return f(*args, **kwargs)
1276 return abort(403)
1278 return decorated
1281def require_maint_access(f: Callable[..., Any]) -> Callable[..., Any]:
1282 """Decorator: abort 403 unless the user has maintenance access.
1284 Maintenance access is granted by ADMIN/OWNER/MAINTENANCE/INSTRUCTOR role,
1285 or by the per-user is_maintenance capability flag.
1286 """
1288 @wraps(f)
1289 def decorated(*args: Any, **kwargs: Any) -> Any:
1290 from models import Role, User, db
1292 role = current_user_role()
1293 if role in (Role.ADMIN, Role.OWNER, Role.MAINTENANCE, Role.INSTRUCTOR):
1294 return f(*args, **kwargs)
1295 uid = session.get("user_id")
1296 if uid:
1297 user = db.session.get(User, uid)
1298 if user and user.is_maintenance:
1299 return f(*args, **kwargs)
1300 return abort(403)
1302 return decorated
1305def tenant_pilots(tenant_id: int) -> list[tuple[int, str]]:
1306 """``(user_id, label)`` for the active users in *tenant_id* with pilot
1307 access (same rule as ``require_pilot_access``: a pilot-capable role or
1308 the ``is_pilot`` flag), sorted case-insensitively by label.
1310 The label is the user's display name; when two pilots share one, both
1311 get their e-mail local part appended so every label is unique and a
1312 picked suggestion maps back to exactly one account.
1313 """
1314 from models import Role, TenantUser, User, db
1315 from sqlalchemy import or_ # pyright: ignore[reportMissingImports]
1317 pilot_roles = (Role.ADMIN, Role.OWNER, Role.PILOT, Role.STUDENT, Role.INSTRUCTOR)
1318 users = (
1319 db.session.query(User)
1320 .join(TenantUser, TenantUser.user_id == User.id)
1321 .filter(TenantUser.tenant_id == tenant_id)
1322 .filter(User.is_active.is_(True))
1323 .filter(or_(TenantUser.role.in_(pilot_roles), User.is_pilot.is_(True)))
1324 .all()
1325 )
1326 name_counts: dict[str, int] = defaultdict(int)
1327 for u in users:
1328 name_counts[u.display_name.casefold()] += 1
1329 labelled = [
1330 (
1331 u.id,
1332 f"{u.display_name} ({u.email.split('@')[0]})"
1333 if name_counts[u.display_name.casefold()] > 1
1334 else u.display_name,
1335 )
1336 for u in users
1337 ]
1338 return sorted(labelled, key=lambda p: p[1].casefold())
1341def tenant_pilot_names(tenant_id: int) -> list[str]:
1342 """Plain display names of ``tenant_pilots`` — type-ahead suggestions for
1343 free-text crew name fields that never link to an account (e.g. the
1344 standalone pilot-logbook entry form's PIC name)."""
1345 return sorted({label for _uid, label in tenant_pilots(tenant_id)}, key=str.casefold)
1348def user_can_access_aircraft(aircraft_id: int) -> bool:
1349 """Return True when the current user may access this aircraft.
1351 ADMIN and OWNER bypass the check entirely. Other roles need either a
1352 UserAllAircraftAccess row (all-planes grant) or a per-aircraft
1353 UserAircraftAccess row.
1354 """
1355 from models import Role, TenantUser, UserAircraftAccess, UserAllAircraftAccess
1357 role = current_user_role()
1358 if role in (Role.ADMIN, Role.OWNER):
1359 return True
1360 uid = session.get("user_id")
1361 if not uid:
1362 return False
1363 tu = TenantUser.query.filter_by(user_id=uid).first()
1364 if (
1365 tu
1366 and UserAllAircraftAccess.query.filter_by(
1367 user_id=uid, tenant_id=tu.tenant_id
1368 ).first()
1369 ):
1370 return True
1371 return (
1372 UserAircraftAccess.query.filter_by(user_id=uid, aircraft_id=aircraft_id).first()
1373 is not None
1374 )
1377def accessible_aircraft(tenant_id: int, include_archived: bool = False) -> Any:
1378 """Return a query of Aircraft the current user is allowed to see.
1380 ADMIN and OWNER see every aircraft in the tenant. A user with a
1381 UserAllAircraftAccess row for the tenant also sees all aircraft.
1382 Other roles see only aircraft granted via UserAircraftAccess.
1384 Archived aircraft are excluded unless include_archived is True — pass it
1385 for views that must keep showing an archived aircraft's history.
1386 """
1387 from models import Aircraft, Role, UserAircraftAccess, UserAllAircraftAccess
1389 base = Aircraft.query.filter_by(tenant_id=tenant_id).order_by(Aircraft.registration)
1390 if not include_archived:
1391 base = base.filter(Aircraft.archived_at.is_(None))
1392 role = current_user_role()
1393 if role in (Role.ADMIN, Role.OWNER):
1394 return base
1395 uid = session.get("user_id")
1396 if not uid:
1397 from sqlalchemy import false
1399 return base.filter(false())
1400 if UserAllAircraftAccess.query.filter_by(user_id=uid, tenant_id=tenant_id).first():
1401 return base
1402 ids = [
1403 row.aircraft_id
1404 for row in (
1405 UserAircraftAccess.query.filter_by(user_id=uid)
1406 .with_entities(UserAircraftAccess.aircraft_id)
1407 .all()
1408 )
1409 ]
1410 if not ids:
1411 from sqlalchemy import false
1413 return base.filter(false())
1414 return base.filter(Aircraft.id.in_(ids))
1417def compute_aircraft_statuses(
1418 aircraft_list: Any,
1419 triggers: Any,
1420 hobbs_by_id: Any,
1421 landings_by_id: Any = None,
1422 flight_hours_by_id: Any = None,
1423) -> dict[int, str]:
1424 """Return {aircraft_id: 'grounded'|'overdue'|'due_soon'|'ok'} for every aircraft.
1426 Grounded (expired insurance or unresolved grounding snag) takes priority.
1427 Among maintenance: overdue > due_soon > ok — maintenance triggers,
1428 component TBO / calendar life limits, and expiring insurance all count.
1430 ``landings_by_id``/``flight_hours_by_id`` default to an empty mapping —
1431 callers that don't (yet) pass them just get 'ok' from any landings- or
1432 flight-hours-basis trigger, same as passing no ``hobbs_by_id`` entry
1433 does for engine-hours-basis ones.
1434 """
1435 from services.component_limits import fleet_limit_statuses
1437 if landings_by_id is None:
1438 landings_by_id = {}
1439 if flight_hours_by_id is None:
1440 flight_hours_by_id = {}
1442 by_aircraft = defaultdict(list)
1443 for t in triggers:
1444 by_aircraft[t.aircraft_id].append(t)
1445 limit_status_by_ac = fleet_limit_statuses(aircraft_list)
1447 result = {}
1448 for ac in aircraft_list:
1449 if ac.is_grounded:
1450 result[ac.id] = "grounded"
1451 continue
1452 hobbs = hobbs_by_id.get(ac.id)
1453 landings = landings_by_id.get(ac.id)
1454 flight_hours = flight_hours_by_id.get(ac.id)
1455 statuses = [
1456 t.status(
1457 current_engine_hours=hobbs,
1458 current_landings=landings,
1459 current_flight_hours=flight_hours,
1460 )
1461 for t in by_aircraft.get(ac.id, [])
1462 ]
1463 statuses.append(limit_status_by_ac.get(ac.id, "ok"))
1464 ins = ac.insurance_status
1465 if ins == "expiring_soon":
1466 statuses.append("due_soon")
1467 if "overdue" in statuses:
1468 result[ac.id] = "overdue"
1469 elif "due_soon" in statuses:
1470 result[ac.id] = "due_soon"
1471 else:
1472 result[ac.id] = "ok"
1473 return result