Coverage for app/utils.py: 10%
671 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-01 12:21 +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 functools import wraps
11from typing import Any, Callable
13from flask import abort, redirect, session, url_for # pyright: ignore[reportMissingImports]
14from werkzeug.routing import ( # pyright: ignore[reportMissingImports]
15 BaseConverter,
16 ValidationError,
17)
19_log = logging.getLogger(__name__)
22def to_libpq_url(database_url: str) -> str:
23 """Strip a SQLAlchemy dialect+driver suffix (e.g. postgresql+psycopg://)
24 down to a plain postgresql:// URL, which is what libpq CLI tools
25 (pg_dump, psql) expect — they don't understand the +driver suffix."""
26 scheme, sep, rest = database_url.partition("://")
27 return f"{scheme.split('+', 1)[0]}{sep}{rest}" if sep else database_url
30# ── Tracks GIF export ─────────────────────────────────────────────────────────
32_GIF_W, _GIF_H = 800, 480
33_GIF_PAD = 30 # pixel padding around the track area
34_GIF_STEP_MS = 600 # ms per frame in the animated GIF
35_GIF_HOLD_MS = 3000 # ms for the final "all tracks" frame
36_TRACK_COLOUR = (180, 30, 200) # current/newest track: vivid purple
37_HISTORY_COLOUR = (
38 130,
39 20,
40 150,
41) # previously drawn tracks: slightly darker purple (GIF uses flat colour; web uses opacity fade)
42_BG_COLOUR = (248, 248, 252) # near-white background
45def _mercator_y(lat_deg: float) -> float:
46 """Web-Mercator y value for a latitude (not clamped)."""
47 lat = math.radians(max(-85.0, min(85.0, lat_deg)))
48 return math.log(math.tan(math.pi / 4 + lat / 2))
51def _build_gif_projection(
52 all_coords: list[tuple[float, float]],
53 canvas_w: int = _GIF_W,
54 canvas_h: int = _GIF_H,
55 pad: int = _GIF_PAD,
56) -> "tuple[Any, Any] | None":
57 """Return (project_fn, bbox) or None if there are fewer than 2 unique points."""
58 if len(all_coords) < 2:
59 return None
60 lons = [c[0] for c in all_coords]
61 lats = [c[1] for c in all_coords]
62 min_lon, max_lon = min(lons), max(lons)
63 min_lat, max_lat = min(lats), max(lats)
64 # Add small margin so tracks don't touch the edge
65 dlon = max(max_lon - min_lon, 0.1) * 0.1
66 dlat = max(max_lat - min_lat, 0.1) * 0.1
67 min_lon -= dlon
68 max_lon += dlon
69 min_lat -= dlat
70 max_lat += dlat
72 usable_w = canvas_w - 2 * pad
73 usable_h = canvas_h - 2 * pad
75 y_min = _mercator_y(min_lat)
76 y_max = _mercator_y(max_lat)
77 y_range = y_max - y_min or 1e-9 # Mercator log-tan units (NOT degrees)
78 x_range = max_lon - min_lon or 1e-9 # degrees
80 # Web Mercator geographic aspect: scale_y_mercator = scale_x * 180/π
81 # (1 Mercator-y unit = 111 km at any latitude; 1° lon at centre = cos(φ)*111 km,
82 # so equal-km scales satisfy scale_y = scale_x * 180/π).
83 # Pick scale_x so BOTH axes fit: scale_x ≤ w/x_range AND scale_x * 180/π * y_range ≤ h.
84 scale_x = min(
85 usable_w / x_range,
86 usable_h * math.pi / (180.0 * y_range),
87 )
88 scale_y = scale_x * 180.0 / math.pi # pixels per Mercator-y unit
90 off_x = pad + (usable_w - x_range * scale_x) / 2
91 off_y = pad + (usable_h - y_range * scale_y) / 2
93 def project(lon: float, lat: float) -> tuple[int, int]:
94 px = off_x + (lon - min_lon) * scale_x
95 py = off_y + (y_max - _mercator_y(lat)) * scale_y
96 return int(px), int(py)
98 return project, (min_lon, min_lat, max_lon, max_lat)
101def _coords_from_geojson(geojson: dict[str, Any] | None) -> list[tuple[float, float]]:
102 """Extract (lon, lat) pairs from a GeoJSON Feature or FeatureCollection.
104 Silently skips a malformed coordinate entry (wrong shape, non-numeric,
105 or non-finite) rather than raising — GpsTrack.geojson is a DB-stored
106 JSON field with no enforced schema; a corrupted entry should degrade to
107 "fewer points plotted", not crash track-image/GIF rendering with an
108 unhandled 500. Non-finite is included because Python's json module
109 accepts the non-standard "Infinity"/"NaN" tokens on load, and a NaN/inf
110 "coordinate" isn't a valid lon/lat regardless — downstream Mercator
111 projection math (_build_gif_projection) isn't finite-safe either.
112 """
113 if not geojson:
114 return []
115 if geojson.get("type") == "Feature":
116 geom = geojson.get("geometry") or {}
117 result = []
118 for c in geom.get("coordinates", []):
119 try:
120 if len(c) < 2:
121 continue
122 lon, lat = float(c[0]), float(c[1])
123 if math.isfinite(lon) and math.isfinite(lat):
124 result.append((lon, lat))
125 except (TypeError, ValueError):
126 continue
127 return result
128 if geojson.get("type") == "FeatureCollection":
129 result = []
130 for feat in geojson.get("features", []):
131 result.extend(_coords_from_geojson(feat))
132 return result
133 return []
136def _make_tile_background(
137 project: Callable[[float, float], tuple[int, int]],
138 min_lon: float,
139 max_lon: float,
140 min_lat: float,
141 max_lat: float,
142 canvas_w: int,
143 canvas_h: int,
144 openaip_key: str | None = None,
145 tile_cache: dict[Any, bytes] | None = None,
146 max_tiles: int = 36,
147) -> Any:
148 """Fetch OSM raster tiles and composite them into a background PIL Image.
150 Returns a PIL Image, or None on failure (caller falls back to plain fill).
151 Tiles are fetched from OpenStreetMap; zoom level is chosen automatically.
152 Max 36 tiles are fetched to keep export time reasonable.
154 tile_cache, if provided, is a dict keyed by (z, tx, ty) or ("opi", z, tx, ty)
155 holding raw PNG bytes so repeated calls across frames skip network fetches.
156 """
157 import urllib.request
158 from PIL import Image as _Img # pyright: ignore[reportMissingImports]
160 # Compute pixels-per-degree-lon from the projection function
161 mid_lat = (min_lat + max_lat) / 2.0
162 px0x = project(min_lon, mid_lat)[0]
163 px1x = project(min_lon + 1.0, mid_lat)[0]
164 scale_x = float(px1x - px0x) # canvas pixels per degree-lon
165 if scale_x <= 0:
166 return None
168 # Choose zoom so each tile is ~256 canvas pixels wide (clamped 2–14)
169 z = int(round(math.log2(max(scale_x * 360.0 / 256.0, 1.0))))
170 z = max(2, min(z, 14))
171 n = 2**z
173 def _lon_to_tx(lon: float) -> int:
174 return int(int((lon + 180.0) / 360.0 * n) % n)
176 def _lat_to_ty(lat: float) -> int:
177 lat_r = math.radians(max(-85.0, min(85.0, lat)))
178 return int(
179 (1.0 - math.log(math.tan(lat_r) + 1.0 / math.cos(lat_r)) / math.pi)
180 / 2.0
181 * n
182 )
184 def _tile_nw_lonlat(tx: int, ty: int) -> tuple[float, float]:
185 lon = tx * 360.0 / n - 180.0
186 lat = math.degrees(math.atan(math.sinh(math.pi * (1.0 - 2.0 * ty / n))))
187 return lon, lat
189 tx_min = _lon_to_tx(min_lon)
190 tx_max = _lon_to_tx(max_lon)
191 ty_min = _lat_to_ty(max_lat)
192 ty_max = _lat_to_ty(min_lat)
194 # Guard against tile count explosion
195 if (tx_max - tx_min + 1) * (ty_max - ty_min + 1) > max_tiles:
196 return None
198 bg = _Img.new("RGB", (canvas_w, canvas_h), _BG_COLOUR)
199 ua = "OpenHangar flight-logbook GIF export (https://github.com/e2jk/OpenHangar)"
201 for tx in range(tx_min, tx_max + 1):
202 for ty in range(ty_min, ty_max + 1):
203 try:
204 tx_w = tx % n
205 # Compute pixel bounds for this tile by projecting its NW corner
206 # and the NW corner of the tile to its SE — then size the resize
207 # to exactly span that gap plus 1 px overlap, eliminating seams
208 # caused by integer rounding of adjacent corner coordinates.
209 lon_nw, lat_nw = _tile_nw_lonlat(tx, ty)
210 lon_se, lat_se = _tile_nw_lonlat(tx + 1, ty + 1)
211 px, py = project(lon_nw, lat_nw)
212 px_se, py_se = project(lon_se, lat_se)
213 tile_w = max(1, px_se - px + 1)
214 tile_h = max(1, py_se - py + 1)
216 # Base map: CARTO light (same as web animation)
217 base_key = (z, tx_w, ty)
218 if tile_cache is not None and base_key in tile_cache:
219 raw = tile_cache[base_key]
220 else:
221 base_url = (
222 f"https://a.basemaps.cartocdn.com/light_all/{z}/{tx_w}/{ty}.png"
223 )
224 req = urllib.request.Request(base_url, headers={"User-Agent": ua})
225 with urllib.request.urlopen(req, timeout=5) as resp: # nosec B310 # base_url has a hardcoded https:// prefix
226 raw = resp.read()
227 if tile_cache is not None:
228 tile_cache[base_key] = raw
229 tile = _Img.open(io.BytesIO(raw)).convert("RGBA")
230 tile = tile.resize((tile_w, tile_h), _Img.Resampling.LANCZOS)
231 bg.paste(tile.convert("RGB"), (px, py))
232 # Aviation overlay: OpenAIP (only at zoom ≤ 14, requires key)
233 if openaip_key and z <= 14:
234 opi_key = ("opi", z, tx_w, ty)
235 if tile_cache is not None and opi_key in tile_cache:
236 opi_raw = tile_cache[opi_key]
237 else:
238 opi_url = f"https://api.tiles.openaip.net/api/data/openaip/{z}/{tx_w}/{ty}.png?apiKey={openaip_key}"
239 opi_req = urllib.request.Request(
240 opi_url, headers={"User-Agent": ua}
241 )
242 with urllib.request.urlopen(opi_req, timeout=5) as opi_resp: # nosec B310 # opi_url has a hardcoded https:// prefix
243 opi_raw = opi_resp.read()
244 if tile_cache is not None:
245 tile_cache[opi_key] = opi_raw
246 opi_tile = _Img.open(io.BytesIO(opi_raw)).convert("RGBA")
247 opi_tile = opi_tile.resize(
248 (tile_w, tile_h), _Img.Resampling.LANCZOS
249 )
250 bg.paste(opi_tile.convert("RGB"), (px, py), opi_tile)
251 except Exception as exc:
252 _log.debug("OPI tile unavailable, leaving area as background: %s", exc)
254 return bg
257def _canvas_geo_bounds(
258 project_fn: Callable[[float, float], tuple[int, int]],
259 canvas_w: int,
260 canvas_h: int,
261 min_lon: float,
262 max_lon: float,
263 min_lat: float,
264 max_lat: float,
265) -> tuple[float, float, float, float]:
266 """Return (lon_min, lat_min, lon_max, lat_max) spanning the full canvas.
268 Expands the geographic bbox beyond the track bounding box so the full
269 canvas area is covered by map tiles, eliminating plain-background padding.
270 Returns the original bbox if the inverse projection cannot be computed.
271 """
272 mid_lat = (min_lat + max_lat) / 2.0
273 # Longitude is linear with pixel x in Mercator
274 px0, _ = project_fn(min_lon, mid_lat)
275 px1, _ = project_fn(min_lon + 1.0, mid_lat)
276 scale_x = float(px1 - px0)
277 if scale_x <= 0:
278 return min_lon, min_lat, max_lon, max_lat
279 c_lon_min = min_lon - float(px0) / scale_x
280 c_lon_max = min_lon + float(canvas_w - px0) / scale_x
282 # Latitude: derive Mercator scale from two known projection points
283 _, py_bot = project_fn(min_lon, min_lat) # larger py (bottom of canvas area)
284 _, py_top = project_fn(min_lon, max_lat) # smaller py (top of canvas area)
285 merc_bot = _mercator_y(min_lat)
286 merc_top = _mercator_y(max_lat)
287 merc_range = merc_top - merc_bot
288 py_range = float(py_bot - py_top)
289 if merc_range <= 0 or py_range <= 0:
290 return c_lon_min, min_lat, c_lon_max, max_lat
291 scale_y = py_range / merc_range # pixels per Mercator-y unit
292 # Derived: merc_y at canvas row py = merc_top + (py_top - py) / scale_y
293 merc_canvas_top = merc_top + float(py_top) / scale_y
294 merc_canvas_bot = merc_top + float(py_top - canvas_h) / scale_y
295 # Clamp to avoid math.atan overflow at extreme Mercator values
296 merc_canvas_top = max(-10.0, min(10.0, merc_canvas_top))
297 merc_canvas_bot = max(-10.0, min(10.0, merc_canvas_bot))
298 c_lat_max = math.degrees(2.0 * math.atan(math.exp(merc_canvas_top)) - math.pi / 2.0)
299 c_lat_min = math.degrees(2.0 * math.atan(math.exp(merc_canvas_bot)) - math.pi / 2.0)
300 return c_lon_min, max(-85.0, c_lat_min), c_lon_max, min(85.0, c_lat_max)
303def sort_tracks_oldest_first(
304 rows: list[dict[str, Any]],
305) -> list[dict[str, Any]]:
306 """Return rows sorted ascending by their 'date' string key (oldest first).
308 Both the web animation and the GIF export use this ordering so that
309 chronological playback and progressive opacity fading are consistent
310 across rendering environments. Call this once in the route handler and
311 pass the result to both the template context and generate_tracks_gif().
312 """
313 return sorted(rows, key=lambda r: r.get("date", ""))
316def generate_tracks_gif(
317 track_rows: list[dict[str, Any]],
318 _font_path: str = "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
319 _openaip_key: str | None = None,
320 canvas_w: int = _GIF_W,
321 canvas_h: int = _GIF_H,
322 high_res: bool = False,
323) -> bytes:
324 """Render an animated GIF of the flight tracks, oldest-first.
326 track_rows must already be sorted oldest-first; use sort_tracks_oldest_first().
327 Each frame adds one more track (cumulative) and re-fits the map to the
328 bounding box of tracks seen so far, creating a progressive zoom-out effect
329 that mirrors the web animation. Previous tracks are drawn in a lighter
330 colour; the newest one in vivid purple. Returns raw GIF bytes ready to
331 stream to the browser.
333 high_res=True doubles line widths, font sizes and padding, uses a 256-colour
334 palette, and raises the tile-fetch cap to 64 — producing a sharper map with
335 readable labels at the cost of a larger file.
336 """
337 from PIL import Image, ImageDraw, ImageFont # pyright: ignore[reportMissingImports]
339 _q_scale = 2 if high_res else 1
340 _pad = _GIF_PAD * _q_scale
341 _line_curr = 3 * _q_scale
342 _line_hist = 2 * _q_scale
343 _font_sz = 14 * _q_scale
344 _font_sz_sm = 11 * _q_scale
345 _txt_margin = 8 * _q_scale
346 _q_colors = 256 if high_res else 128
347 _max_tiles = 64 if high_res else 36
348 # Canvas-extent calls cover the full 1600×960 canvas which can require
349 # ~70–90 tiles at certain zoom levels (wide flat tracks at z=7), so use a
350 # higher cap than the track-bbox call. 128 = 2× the high-res track cap.
351 _max_tiles_canvas = _max_tiles * 2 if high_res else _max_tiles
353 # Pre-compute per-track coordinates (avoids re-parsing GeoJSON in the inner loop)
354 per_track_coords = [_coords_from_geojson(r.get("geojson")) for r in track_rows]
355 all_coords: list[tuple[float, float]] = [c for tc in per_track_coords for c in tc]
357 if len(all_coords) < 2:
358 # Fallback: single blank frame
359 img = Image.new("RGB", (canvas_w, canvas_h), _BG_COLOUR)
360 buf = io.BytesIO()
361 img.save(buf, format="GIF")
362 return buf.getvalue()
364 font: Any # PIL font type varies across Pillow versions
365 font_sm: Any
366 try:
367 font = ImageFont.truetype(_font_path, _font_sz)
368 font_sm = ImageFont.truetype(_font_path, _font_sz_sm)
369 except (IOError, OSError):
370 font = font_sm = ImageFont.load_default()
372 def draw_shadow_text(draw: Any, text: str, font: Any) -> None:
373 draw.text(
374 (_txt_margin + 1, _txt_margin + 1), text, fill=(255, 255, 255), font=font
375 )
376 draw.text((_txt_margin, _txt_margin), text, fill=(40, 40, 60), font=font)
378 def draw_track(
379 draw: Any,
380 coords: list[tuple[float, float]],
381 colour: tuple[int, int, int],
382 width: int,
383 project_fn: Callable[[float, float], tuple[int, int]],
384 ) -> None:
385 pts = [project_fn(lon, lat) for lon, lat in coords]
386 if len(pts) >= 2:
387 draw.line(pts, fill=colour, width=width)
388 r = width + 2 * _q_scale
389 sx, sy = pts[0]
390 draw.ellipse([sx - r, sy - r, sx + r, sy + r], fill=(40, 160, 60))
391 ex, ey = pts[-1]
392 draw.ellipse([ex - r, ey - r, ex + r, ey + r], fill=(200, 40, 40))
394 # Shared tile cache: (z, tx, ty) → raw PNG bytes; ("opi", z, tx, ty) for OpenAIP.
395 # Avoids re-fetching tiles that appear in multiple frames at the same zoom level.
396 tile_cache: dict[Any, bytes] = {}
398 def _frame_bg(
399 project_fn: Callable[[float, float], tuple[int, int]],
400 min_lon: float,
401 max_lon: float,
402 min_lat: float,
403 max_lat: float,
404 ) -> Any:
405 fetch_lon_min, fetch_lat_min, fetch_lon_max, fetch_lat_max = (
406 min_lon,
407 min_lat,
408 max_lon,
409 max_lat,
410 )
411 if high_res:
412 fetch_lon_min, fetch_lat_min, fetch_lon_max, fetch_lat_max = (
413 _canvas_geo_bounds(
414 project_fn,
415 canvas_w,
416 canvas_h,
417 min_lon,
418 max_lon,
419 min_lat,
420 max_lat,
421 )
422 )
423 bg = _make_tile_background(
424 project_fn,
425 fetch_lon_min,
426 fetch_lon_max,
427 fetch_lat_min,
428 fetch_lat_max,
429 canvas_w,
430 canvas_h,
431 openaip_key=_openaip_key,
432 tile_cache=tile_cache,
433 max_tiles=_max_tiles_canvas,
434 )
435 if bg is None and high_res:
436 # Canvas extent still exceeded the raised cap; fall back to track bbox
437 bg = _make_tile_background(
438 project_fn,
439 min_lon,
440 max_lon,
441 min_lat,
442 max_lat,
443 canvas_w,
444 canvas_h,
445 openaip_key=_openaip_key,
446 tile_cache=tile_cache,
447 max_tiles=_max_tiles,
448 )
449 return (
450 bg.copy()
451 if bg is not None
452 else Image.new("RGB", (canvas_w, canvas_h), _BG_COLOUR)
453 )
455 frames: list[Any] = []
456 durations: list[int] = []
457 accumulated_coords: list[tuple[float, float]] = []
459 for frame_idx in range(len(track_rows)):
460 accumulated_coords.extend(per_track_coords[frame_idx])
461 proj_result = _build_gif_projection(
462 accumulated_coords, canvas_w=canvas_w, canvas_h=canvas_h, pad=_pad
463 )
464 if proj_result is None:
465 continue # not enough coords yet (e.g. leading rows with no geojson)
467 project_fn, (f_min_lon, f_min_lat, f_max_lon, f_max_lat) = proj_result
468 img = _frame_bg(project_fn, f_min_lon, f_max_lon, f_min_lat, f_max_lat)
469 draw = ImageDraw.Draw(img)
471 for i in range(frame_idx):
472 draw_track(
473 draw, per_track_coords[i], _HISTORY_COLOUR, _line_hist, project_fn
474 )
475 draw_track(
476 draw, per_track_coords[frame_idx], _TRACK_COLOUR, _line_curr, project_fn
477 )
479 row = track_rows[frame_idx]
480 label = f"{row.get('date', '')} {row.get('dep', '')} → {row.get('arr', '')}"
481 draw_shadow_text(draw, label, font)
482 draw.text(
483 (_txt_margin, canvas_h - _font_sz_sm - _txt_margin),
484 f"{frame_idx + 1} / {len(track_rows)}",
485 fill=(80, 80, 100),
486 font=font_sm,
487 )
489 frames.append(img)
490 durations.append(_GIF_STEP_MS)
492 # Final frame: all tracks at equal weight using the full bounding box, longer hold
493 if frames:
494 proj_result_final = _build_gif_projection(
495 all_coords, canvas_w=canvas_w, canvas_h=canvas_h, pad=_pad
496 )
497 # proj_result_final is guaranteed non-None: all_coords has ≥ 2 points (checked above)
498 project_final, (f_min_lon, f_min_lat, f_max_lon, f_max_lat) = proj_result_final # type: ignore[misc]
499 img = _frame_bg(project_final, f_min_lon, f_max_lon, f_min_lat, f_max_lat)
500 draw = ImageDraw.Draw(img)
501 for tc in per_track_coords:
502 draw_track(draw, tc, _TRACK_COLOUR, _line_hist, project_final)
503 draw_shadow_text(draw, f"All {len(track_rows)} tracks", font)
504 frames.append(img)
505 durations.append(_GIF_HOLD_MS)
507 # Quantise all frames to a shared 128-colour palette built from the last frame
508 # (widest view, most visually representative). A single global palette means
509 # identical background areas compress very well with GIF's LZW codec.
510 palette_src = frames[-1].quantize(colors=_q_colors, dither=Image.Dither.NONE)
511 quantized: list[Any] = [
512 f.quantize(palette=palette_src, dither=Image.Dither.NONE) for f in frames
513 ]
515 buf = io.BytesIO()
516 quantized[0].save(
517 buf,
518 format="GIF",
519 save_all=True,
520 append_images=quantized[1:],
521 loop=0,
522 duration=durations,
523 optimize=True,
524 )
525 return buf.getvalue()
528def generate_single_track_image(
529 geojson: dict[str, Any] | None,
530 date: str = "",
531 dep: str = "",
532 arr: str = "",
533 _font_path: str = "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
534 _openaip_key: str | None = None,
535 canvas_w: int = _GIF_W,
536 canvas_h: int = _GIF_H,
537 high_res: bool = False,
538) -> bytes:
539 """Render a single GPS track as a static PNG.
541 Returns raw PNG bytes. If the track has fewer than 2 points a blank
542 canvas is returned so the caller can always stream a valid image.
543 """
544 from PIL import Image, ImageDraw, ImageFont # pyright: ignore[reportMissingImports]
546 all_coords = _coords_from_geojson(geojson)
548 _q_scale = 2 if high_res else 1
549 _pad = _GIF_PAD * _q_scale
550 _line_w = 3 * _q_scale
551 _font_sz = 14 * _q_scale
552 _txt_margin = 8 * _q_scale
553 _max_tiles = 64 if high_res else 36
554 _max_tiles_canvas = _max_tiles * 2 if high_res else _max_tiles
556 def _blank_png() -> bytes:
557 buf = io.BytesIO()
558 Image.new("RGB", (canvas_w, canvas_h), _BG_COLOUR).save(buf, format="PNG")
559 return buf.getvalue()
561 if len(all_coords) < 2:
562 return _blank_png()
564 proj_result = _build_gif_projection(
565 all_coords, canvas_w=canvas_w, canvas_h=canvas_h, pad=_pad
566 )
567 if proj_result is None:
568 return _blank_png()
570 project_fn, (f_min_lon, f_min_lat, f_max_lon, f_max_lat) = proj_result
572 tile_cache: dict[Any, bytes] = {}
573 fetch_lon_min, fetch_lat_min, fetch_lon_max, fetch_lat_max = (
574 f_min_lon,
575 f_min_lat,
576 f_max_lon,
577 f_max_lat,
578 )
579 if high_res:
580 fetch_lon_min, fetch_lat_min, fetch_lon_max, fetch_lat_max = _canvas_geo_bounds(
581 project_fn, canvas_w, canvas_h, f_min_lon, f_max_lon, f_min_lat, f_max_lat
582 )
583 bg = _make_tile_background(
584 project_fn,
585 fetch_lon_min,
586 fetch_lon_max,
587 fetch_lat_min,
588 fetch_lat_max,
589 canvas_w,
590 canvas_h,
591 openaip_key=_openaip_key,
592 tile_cache=tile_cache,
593 max_tiles=_max_tiles_canvas,
594 )
595 if bg is None and high_res:
596 bg = _make_tile_background(
597 project_fn,
598 f_min_lon,
599 f_max_lon,
600 f_min_lat,
601 f_max_lat,
602 canvas_w,
603 canvas_h,
604 openaip_key=_openaip_key,
605 tile_cache=tile_cache,
606 max_tiles=_max_tiles,
607 )
608 img = (
609 bg.copy()
610 if bg is not None
611 else Image.new("RGB", (canvas_w, canvas_h), _BG_COLOUR)
612 )
614 draw = ImageDraw.Draw(img)
615 pts = [project_fn(lon, lat) for lon, lat in all_coords]
616 if len(pts) >= 2:
617 draw.line(pts, fill=_TRACK_COLOUR, width=_line_w)
618 r = _line_w + 2 * _q_scale
619 sx, sy = pts[0]
620 draw.ellipse([sx - r, sy - r, sx + r, sy + r], fill=(40, 160, 60))
621 ex, ey = pts[-1]
622 draw.ellipse([ex - r, ey - r, ex + r, ey + r], fill=(200, 40, 40))
624 label = f"{date} {dep} → {arr}".strip(" →").strip()
625 if label:
626 try:
627 font: Any = ImageFont.truetype(_font_path, _font_sz)
628 except (IOError, OSError):
629 font = ImageFont.load_default()
630 draw.text(
631 (_txt_margin + 1, _txt_margin + 1), label, fill=(255, 255, 255), font=font
632 )
633 draw.text((_txt_margin, _txt_margin), label, fill=(40, 40, 60), font=font)
635 buf = io.BytesIO()
636 img.save(buf, format="PNG")
637 return buf.getvalue()
640def generate_single_track_gif(
641 geojson: dict[str, Any] | None,
642 date: str = "",
643 dep: str = "",
644 arr: str = "",
645 _font_path: str = "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
646 _openaip_key: str | None = None,
647 canvas_w: int = _GIF_W,
648 canvas_h: int = _GIF_H,
649 high_res: bool = False,
650) -> bytes:
651 """Render an animated GIF of a single flight track drawn progressively.
653 The track's coordinate array is split into N equal chunks (3–10 depending
654 on track length). Each frame draws all chunks revealed so far, fitting the
655 map bounding box to the accumulated coords so the view zooms out as the
656 route grows — mirroring the per-flight behaviour of the web Animate button.
658 Falls back to a single-frame GIF wrapping the still PNG when the track has
659 fewer than 20 GPS points (too sparse for meaningful animation).
660 """
661 from PIL import Image, ImageDraw, ImageFont # pyright: ignore[reportMissingImports]
663 all_coords = _coords_from_geojson(geojson)
665 _q_scale = 2 if high_res else 1
666 _pad = _GIF_PAD * _q_scale
667 _line_curr = 3 * _q_scale
668 _line_hist = 2 * _q_scale
669 _font_sz = 14 * _q_scale
670 _font_sz_sm = 11 * _q_scale
671 _txt_margin = 8 * _q_scale
672 _q_colors = 256 if high_res else 128
673 _max_tiles = 64 if high_res else 36
674 _max_tiles_canvas = _max_tiles * 2 if high_res else _max_tiles
676 def _blank_gif() -> bytes:
677 buf = io.BytesIO()
678 Image.new("RGB", (canvas_w, canvas_h), _BG_COLOUR).save(buf, format="GIF")
679 return buf.getvalue()
681 if len(all_coords) < 2:
682 return _blank_gif()
684 # Sparse-track fallback: wrap the still image in a single-frame GIF
685 if len(all_coords) < 20:
686 png = generate_single_track_image(
687 geojson,
688 date=date,
689 dep=dep,
690 arr=arr,
691 _font_path=_font_path,
692 _openaip_key=_openaip_key,
693 canvas_w=canvas_w,
694 canvas_h=canvas_h,
695 high_res=high_res,
696 )
697 img = Image.open(io.BytesIO(png)).convert("RGB")
698 buf = io.BytesIO()
699 img.quantize(colors=_q_colors, dither=Image.Dither.NONE).save(buf, format="GIF")
700 return buf.getvalue()
702 # Split coords into N equal chunks (3–10)
703 n_chunks = max(3, min(10, len(all_coords) // 10))
704 chunk_size = len(all_coords) // n_chunks
705 chunks: list[list[tuple[float, float]]] = [
706 list(all_coords[i * chunk_size : (i + 1) * chunk_size])
707 for i in range(n_chunks - 1)
708 ]
709 chunks.append(
710 list(all_coords[(n_chunks - 1) * chunk_size :])
711 ) # last catches remainder
713 try:
714 font: Any = ImageFont.truetype(_font_path, _font_sz)
715 font_sm: Any = ImageFont.truetype(_font_path, _font_sz_sm)
716 except (IOError, OSError):
717 font = font_sm = ImageFont.load_default()
719 def draw_shadow_text(draw: Any, text: str, fnt: Any) -> None:
720 draw.text(
721 (_txt_margin + 1, _txt_margin + 1), text, fill=(255, 255, 255), font=fnt
722 )
723 draw.text((_txt_margin, _txt_margin), text, fill=(40, 40, 60), font=fnt)
725 def draw_track(
726 draw: Any,
727 coords: list[tuple[float, float]],
728 colour: tuple[int, int, int],
729 width: int,
730 project_fn: Callable[[float, float], tuple[int, int]],
731 ) -> None:
732 pts = [project_fn(lon, lat) for lon, lat in coords]
733 if len(pts) >= 2:
734 draw.line(pts, fill=colour, width=width)
735 r = width + 2 * _q_scale
736 sx, sy = pts[0]
737 draw.ellipse([sx - r, sy - r, sx + r, sy + r], fill=(40, 160, 60))
738 ex, ey = pts[-1]
739 draw.ellipse([ex - r, ey - r, ex + r, ey + r], fill=(200, 40, 40))
741 tile_cache: dict[Any, bytes] = {}
743 def _frame_bg(
744 project_fn: Callable[[float, float], tuple[int, int]],
745 min_lon: float,
746 max_lon: float,
747 min_lat: float,
748 max_lat: float,
749 ) -> Any:
750 fetch_lon_min, fetch_lat_min, fetch_lon_max, fetch_lat_max = (
751 min_lon,
752 min_lat,
753 max_lon,
754 max_lat,
755 )
756 if high_res:
757 fetch_lon_min, fetch_lat_min, fetch_lon_max, fetch_lat_max = (
758 _canvas_geo_bounds(
759 project_fn, canvas_w, canvas_h, min_lon, max_lon, min_lat, max_lat
760 )
761 )
762 bg = _make_tile_background(
763 project_fn,
764 fetch_lon_min,
765 fetch_lon_max,
766 fetch_lat_min,
767 fetch_lat_max,
768 canvas_w,
769 canvas_h,
770 openaip_key=_openaip_key,
771 tile_cache=tile_cache,
772 max_tiles=_max_tiles_canvas,
773 )
774 if bg is None and high_res:
775 bg = _make_tile_background(
776 project_fn,
777 min_lon,
778 max_lon,
779 min_lat,
780 max_lat,
781 canvas_w,
782 canvas_h,
783 openaip_key=_openaip_key,
784 tile_cache=tile_cache,
785 max_tiles=_max_tiles,
786 )
787 return (
788 bg.copy()
789 if bg is not None
790 else Image.new("RGB", (canvas_w, canvas_h), _BG_COLOUR)
791 )
793 label = f"{date} {dep} → {arr}".strip(" →").strip()
795 frames: list[Any] = []
796 durations: list[int] = []
797 accumulated: list[tuple[float, float]] = []
799 for chunk_idx, chunk in enumerate(chunks):
800 accumulated.extend(chunk)
801 proj_result = _build_gif_projection(
802 accumulated, canvas_w=canvas_w, canvas_h=canvas_h, pad=_pad
803 )
804 if proj_result is None:
805 continue
807 project_fn, (f_min_lon, f_min_lat, f_max_lon, f_max_lat) = proj_result
808 img = _frame_bg(project_fn, f_min_lon, f_max_lon, f_min_lat, f_max_lat)
809 draw = ImageDraw.Draw(img)
811 for i in range(chunk_idx):
812 draw_track(draw, chunks[i], _HISTORY_COLOUR, _line_hist, project_fn)
813 draw_track(draw, chunk, _TRACK_COLOUR, _line_curr, project_fn)
815 if label:
816 draw_shadow_text(draw, label, font)
817 draw.text(
818 (_txt_margin, canvas_h - _font_sz_sm - _txt_margin),
819 f"{chunk_idx + 1} / {n_chunks}",
820 fill=(80, 80, 100),
821 font=font_sm,
822 )
823 frames.append(img)
824 durations.append(_GIF_STEP_MS)
826 if not frames:
827 return _blank_gif()
829 # Final hold frame: full track, all chunks at equal weight
830 proj_result_final = _build_gif_projection(
831 list(all_coords), canvas_w=canvas_w, canvas_h=canvas_h, pad=_pad
832 )
833 # guaranteed non-None: all_coords has ≥ 20 points
834 project_final, (f_min_lon, f_min_lat, f_max_lon, f_max_lat) = proj_result_final # type: ignore[misc]
835 img = _frame_bg(project_final, f_min_lon, f_max_lon, f_min_lat, f_max_lat)
836 draw = ImageDraw.Draw(img)
837 for chunk in chunks:
838 draw_track(draw, chunk, _TRACK_COLOUR, _line_hist, project_final)
839 if label:
840 draw_shadow_text(draw, label, font)
841 frames.append(img)
842 durations.append(_GIF_HOLD_MS)
844 palette_src = frames[-1].quantize(colors=_q_colors, dither=Image.Dither.NONE)
845 quantized: list[Any] = [
846 f.quantize(palette=palette_src, dither=Image.Dither.NONE) for f in frames
847 ]
848 buf = io.BytesIO()
849 quantized[0].save(
850 buf,
851 format="GIF",
852 save_all=True,
853 append_images=quantized[1:],
854 loop=0,
855 duration=durations,
856 optimize=True,
857 )
858 return buf.getvalue()
861@functools.lru_cache(maxsize=1)
862def _load_aircraft_types() -> dict[str, tuple[str, str]]:
863 """Return {type_designator: (manufacturer, model)} from aircraft_types.csv."""
864 path = os.path.join(os.path.dirname(__file__), "data", "aircraft_types.csv")
865 result: dict[str, tuple[str, str]] = {}
866 try:
867 with open(path, newline="", encoding="utf-8") as f:
868 for row in csv.DictReader(f):
869 des = row.get("type_designator", "").strip().upper()
870 mfr = row.get("manufacturer", "").strip()
871 model = row.get("model", "").strip()
872 if des:
873 result[des] = (mfr, model)
874 except OSError as exc:
875 _log.warning("aircraft_types.csv not found: %s", exc)
876 return result
879def _load_aircraft_type_variants() -> list[tuple[str, str, str, str]]:
880 """Return all (type_designator, full_name, manufacturer, model) tuples — one per CSV row.
882 Unlike _load_aircraft_types(), duplicate designators are preserved so
883 the search endpoint can surface every variant (e.g. all PA-28-181 models
884 that share the P28A ICAO code). manufacturer and model are kept separate
885 so callers can pre-fill form fields for the exact selected variant rather
886 than an arbitrary one sharing the same ICAO code.
887 """
888 path = os.path.join(os.path.dirname(__file__), "data", "aircraft_types.csv")
889 result: list[tuple[str, str, str, str]] = []
890 seen: set[tuple[str, str]] = set()
891 try:
892 with open(path, newline="", encoding="utf-8") as f:
893 for row in csv.DictReader(f):
894 des = row.get("type_designator", "").strip().upper()
895 mfr = row.get("manufacturer", "").strip()
896 model = row.get("model", "").strip()
897 name = f"{mfr} {model}".strip()
898 if des and (des, name) not in seen:
899 result.append((des, name, mfr, model))
900 seen.add((des, name))
901 except OSError as exc:
902 _log.warning("aircraft_types.csv not found: %s", exc)
903 return result
906@functools.lru_cache(maxsize=1)
907def _load_aircraft_type_engine_data() -> dict[str, tuple[int, str]]:
908 """Return {type_designator: (engine_count, engine_type)} from aircraft_types.csv.
910 For designators with multiple rows (variants), uses data from the first row.
911 """
912 path = os.path.join(os.path.dirname(__file__), "data", "aircraft_types.csv")
913 result: dict[str, tuple[int, str]] = {}
914 try:
915 with open(path, newline="", encoding="utf-8") as f:
916 for row in csv.DictReader(f):
917 des = row.get("type_designator", "").strip().upper()
918 if not des or des in result:
919 continue
920 try:
921 ec = int(row.get("engine_count", "1"))
922 except ValueError:
923 ec = 1
924 et = row.get("engine_type", "").strip()
925 result[des] = (ec, et)
926 except OSError as exc:
927 _log.warning("aircraft_types.csv not found: %s", exc)
928 return result
931def get_aircraft_type_engine_info(icao_code: str) -> tuple[int, str] | None:
932 """Return (engine_count, engine_type) for the given ICAO code, or None."""
933 return _load_aircraft_type_engine_data().get(icao_code.strip().upper())
936@functools.lru_cache(maxsize=1)
937def _build_model_name_prefix_lookup() -> dict[str, str]:
938 """Return {compact_first_word_of_model: icao_code} for resolve_aircraft_type_icao fallback.
940 Enables matching pilot-logbook type strings like "DR401" or "PA28-161 TDI"
941 against ICAO codes like "DR40" or "P28A" via the CSV model-name column.
942 """
943 path = os.path.join(os.path.dirname(__file__), "data", "aircraft_types.csv")
944 result: dict[str, str] = {}
945 try:
946 with open(path, newline="", encoding="utf-8") as f:
947 for row in csv.DictReader(f):
948 des = row.get("type_designator", "").strip().upper()
949 model = row.get("model", "").strip()
950 if not des or not model:
951 continue
952 first_word = model.split()[0]
953 key = first_word.replace("-", "").replace(" ", "").upper()
954 if key and key not in result:
955 result[key] = des
956 except OSError as exc:
957 _log.warning("aircraft_types.csv not found: %s", exc)
958 return result
961@functools.lru_cache(maxsize=1)
962def _sorted_model_prefix_keys() -> list[str]:
963 """Keys of _build_model_name_prefix_lookup(), longest-first.
965 Cached separately from the lookup dict itself: found by fuzzing that
966 resolve_aircraft_type_icao() re-sorted these ~1-2k keys on *every* call
967 that reaches the fallback branch (i.e. almost every call — an exact ICAO
968 code is rarely what a user types or a logbook CSV column holds), even
969 though the lookup dict underneath never changes between calls.
970 """
971 return sorted(_build_model_name_prefix_lookup(), key=len, reverse=True)
974def resolve_aircraft_type_icao(aircraft_type: str | None) -> str | None:
975 """Return the matching ICAO type designator for *aircraft_type*, or None."""
976 if not aircraft_type:
977 return None
978 types = _load_aircraft_types()
979 norm = aircraft_type.strip().upper()
980 if norm in types:
981 return norm
982 # Try stripping hyphens and spaces (e.g. "PA-28" → "PA28")
983 compact = norm.replace("-", "").replace(" ", "")
984 if compact in types:
985 return compact
986 # Try matching against the first word of each CSV model name (longest key wins).
987 # e.g. "DR401" startswith "DR401" (from "DR-401 135CDI") → DR40
988 # "PA28161TDI" startswith "PA28161" (from "PA-28-161 Cherokee…") → P28A
989 # Guard: reject if the unmatched tail starts with a digit — that signals a
990 # different model number (e.g. "C172RG" wrongly matching key "C17").
991 prefix_lookup = _build_model_name_prefix_lookup()
992 for key in _sorted_model_prefix_keys():
993 if len(key) < 4:
994 break # keys are sorted longest-first; stop once they're too short
995 if compact.startswith(key):
996 tail = compact[len(key) :]
997 if tail and tail[0].isdigit():
998 continue # "C172RG" matching "C17" — tail "2RG" starts with digit
999 return prefix_lookup[key]
1000 return None
1003@functools.lru_cache(maxsize=1)
1004def _load_airport_names() -> dict[str, str]:
1005 """Return {ICAO ident: airport name} for all airports in airports.csv."""
1006 path = os.path.join(os.path.dirname(__file__), "data", "airports.csv")
1007 result: dict[str, str] = {}
1008 try:
1009 with open(path, newline="", encoding="utf-8") as f:
1010 for row in csv.DictReader(f):
1011 ident = row.get("ident", "").strip()
1012 name = row.get("name", "").strip()
1013 if ident and name:
1014 result[ident] = name
1015 except OSError as exc:
1016 _log.warning("airports.csv not found: %s", exc)
1017 return result
1020_alog = logging.getLogger("openhangar.activity")
1023def _sl(value: object) -> str:
1024 """Sanitize a value for log output — strips CR/LF to prevent log injection (CWE-117)."""
1025 return str(value).replace("\r\n", "").replace("\n", "").replace("\r", "")
1028def activity(event: str, **fields: object) -> None:
1029 """Emit a structured [ACTIVITY] log entry with user_id and ip automatically included."""
1030 from flask import request, session # noqa: PLC0415
1032 uid = session.get("user_id", "")
1033 ip = request.remote_addr or ""
1034 parts = [f"[ACTIVITY] {event}", f"user_id={_sl(uid)}", f"ip={_sl(ip)}"]
1035 parts.extend(f"{k}={_sl(v)}" for k, v in fields.items())
1036 _alog.info(" ".join(parts))
1039class AircraftRefConverter(BaseConverter):
1040 """URL converter for the ``aircraft_id`` slot: accepts either the numeric
1041 primary key (unchanged, always works) or the aircraft's registration
1042 (e.g. ``OO-GRN``), so routes like ``/aircraft/<aircraft_id>/flights``
1043 also resolve ``/aircraft/OO-GRN/flights``.
1045 Authorization is unaffected: this only resolves the URL segment to a
1046 primary key (or the reverse, for link generation) — every view still
1047 does its own tenant-scoped lookup exactly as before, so a registration
1048 belonging to another tenant 404s the same way a wrong numeric id does.
1050 Registrations containing '/' or spaces (rare, but not forbidden by the
1051 model) are sanitized the same way upload filenames already are
1052 elsewhere in this blueprint; such an aircraft simply isn't reachable via
1053 its pretty URL (only via the numeric id, which always works).
1054 """
1056 regex = r"[^/]+"
1058 def to_python(self, value: str) -> int:
1059 if value.isdigit():
1060 return int(value)
1062 from models import Aircraft, db
1064 needle = value.upper()
1065 ac = (
1066 Aircraft.query.filter(
1067 db.func.upper(
1068 db.func.replace(
1069 db.func.replace(Aircraft.registration, "/", "-"), " ", "-"
1070 )
1071 )
1072 == needle
1073 )
1074 .order_by(Aircraft.id)
1075 .first()
1076 )
1077 if ac is None:
1078 raise ValidationError()
1079 return int(ac.id)
1081 def to_url(self, value: Any) -> str:
1082 if isinstance(value, str) and not value.isdigit():
1083 return super().to_url(value)
1085 from models import Aircraft, db
1087 ac = db.session.get(Aircraft, int(value))
1088 reg = ac.registration if ac and ac.registration else str(value)
1089 safe_reg = reg.replace("/", "-").replace(" ", "-")
1090 return super().to_url(safe_reg)
1093def login_required(f: Callable[..., Any]) -> Callable[..., Any]:
1094 """Redirect unauthenticated users to the login page."""
1096 @wraps(f)
1097 def decorated(*args: Any, **kwargs: Any) -> Any:
1098 if not session.get("user_id"):
1099 return redirect(url_for("auth.login"))
1100 return f(*args, **kwargs)
1102 return decorated
1105def require_instance_admin(f: Callable[..., Any]) -> Callable[..., Any]:
1106 """Abort 403 unless the current user is the instance admin."""
1108 @wraps(f)
1109 def decorated(*args: Any, **kwargs: Any) -> Any:
1110 from models import User, db
1112 user_id = session.get("user_id")
1113 if not user_id:
1114 return redirect(url_for("auth.login"))
1115 user = db.session.get(User, user_id)
1116 if not user or not user.is_instance_admin:
1117 abort(403)
1118 return f(*args, **kwargs)
1120 return decorated
1123def current_user_role() -> str | None:
1124 """Return the Role of the current user in their tenant, or None."""
1125 from models import TenantUser
1127 user_id = session.get("user_id")
1128 if not user_id:
1129 return None
1130 tu = TenantUser.query.filter_by(user_id=user_id).first()
1131 return tu.role if tu else None
1134def check_update_available() -> bool:
1135 """Return True when a newer release is available than the running instance.
1137 Reads the ``update_available`` AppSetting written by the background
1138 version-check service (computed once per check cycle, not per request).
1139 Falls back to a live comparison against ``latest_version`` if the flag has
1140 not been written yet (fresh install before the first background check).
1141 Returns False on any error. Must be called inside a request (or
1142 application) context.
1143 """
1144 try:
1145 from models import AppSetting, db # pyright: ignore[reportMissingImports]
1147 flag = db.session.get(AppSetting, "update_available")
1148 if flag is not None:
1149 return bool(flag.value == "true")
1151 # Fallback for fresh installs: compute from latest_version directly.
1152 from packaging.version import Version # pyright: ignore[reportMissingImports]
1154 current = os.environ.get("OPENHANGAR_VERSION", "development")
1155 if current == "development":
1156 return False
1157 latest_s = db.session.get(AppSetting, "latest_version")
1158 latest = latest_s.value if latest_s else None
1159 return bool(latest and Version(latest) > Version(current))
1160 except Exception:
1161 return False
1164def check_legacy_logbook_data() -> bool:
1165 """Return True when pre-unification logbook/crew data was preserved
1166 but not migrated during the ``flight_crew``/``pilot_logbook_entries``
1167 table removal.
1169 Reads the ``legacy_logbook_data_present`` AppSetting written by the
1170 Alembic migration that dropped those two tables — set only when either
1171 table still had rows at upgrade time (both tables are dropped outright,
1172 with no flag set, when they were already empty). Returns False on any
1173 error. Must be called inside a request (or application) context.
1174 """
1175 try:
1176 from models import AppSetting, db # pyright: ignore[reportMissingImports]
1178 flag = db.session.get(AppSetting, "legacy_logbook_data_present")
1179 return bool(flag is not None and flag.value == "true")
1180 except Exception:
1181 return False
1184def require_role(*roles: str) -> Callable[..., Any]:
1185 """Decorator: abort 403 if the current user's role is not in *roles*."""
1187 def decorator(f: Callable[..., Any]) -> Callable[..., Any]:
1188 @wraps(f)
1189 def decorated(*args: Any, **kwargs: Any) -> Any:
1190 if current_user_role() not in roles:
1191 abort(403)
1192 return f(*args, **kwargs)
1194 return decorated
1196 return decorator
1199def require_pilot_access(f: Callable[..., Any]) -> Callable[..., Any]:
1200 """Decorator: abort 403 unless the user has pilot access.
1202 Pilot access is granted by ADMIN/OWNER/PILOT/STUDENT/INSTRUCTOR role,
1203 or by the per-user is_pilot capability flag.
1204 """
1206 @wraps(f)
1207 def decorated(*args: Any, **kwargs: Any) -> Any:
1208 from models import Role, User, db
1210 role = current_user_role()
1211 if role in (Role.ADMIN, Role.OWNER, Role.PILOT, Role.STUDENT, Role.INSTRUCTOR):
1212 return f(*args, **kwargs)
1213 uid = session.get("user_id")
1214 if uid:
1215 user = db.session.get(User, uid)
1216 if user and user.is_pilot:
1217 return f(*args, **kwargs)
1218 return abort(403)
1220 return decorated
1223def require_maint_access(f: Callable[..., Any]) -> Callable[..., Any]:
1224 """Decorator: abort 403 unless the user has maintenance access.
1226 Maintenance access is granted by ADMIN/OWNER/MAINTENANCE/INSTRUCTOR role,
1227 or by the per-user is_maintenance capability flag.
1228 """
1230 @wraps(f)
1231 def decorated(*args: Any, **kwargs: Any) -> Any:
1232 from models import Role, User, db
1234 role = current_user_role()
1235 if role in (Role.ADMIN, Role.OWNER, Role.MAINTENANCE, Role.INSTRUCTOR):
1236 return f(*args, **kwargs)
1237 uid = session.get("user_id")
1238 if uid:
1239 user = db.session.get(User, uid)
1240 if user and user.is_maintenance:
1241 return f(*args, **kwargs)
1242 return abort(403)
1244 return decorated
1247def user_can_access_aircraft(aircraft_id: int) -> bool:
1248 """Return True when the current user may access this aircraft.
1250 ADMIN and OWNER bypass the check entirely. Other roles need either a
1251 UserAllAircraftAccess row (all-planes grant) or a per-aircraft
1252 UserAircraftAccess row.
1253 """
1254 from models import Role, TenantUser, UserAircraftAccess, UserAllAircraftAccess
1256 role = current_user_role()
1257 if role in (Role.ADMIN, Role.OWNER):
1258 return True
1259 uid = session.get("user_id")
1260 if not uid:
1261 return False
1262 tu = TenantUser.query.filter_by(user_id=uid).first()
1263 if (
1264 tu
1265 and UserAllAircraftAccess.query.filter_by(
1266 user_id=uid, tenant_id=tu.tenant_id
1267 ).first()
1268 ):
1269 return True
1270 return (
1271 UserAircraftAccess.query.filter_by(user_id=uid, aircraft_id=aircraft_id).first()
1272 is not None
1273 )
1276def accessible_aircraft(tenant_id: int, include_archived: bool = False) -> Any:
1277 """Return a query of Aircraft the current user is allowed to see.
1279 ADMIN and OWNER see every aircraft in the tenant. A user with a
1280 UserAllAircraftAccess row for the tenant also sees all aircraft.
1281 Other roles see only aircraft granted via UserAircraftAccess.
1283 Archived aircraft are excluded unless include_archived is True — pass it
1284 for views that must keep showing an archived aircraft's history.
1285 """
1286 from models import Aircraft, Role, UserAircraftAccess, UserAllAircraftAccess
1288 base = Aircraft.query.filter_by(tenant_id=tenant_id).order_by(Aircraft.registration)
1289 if not include_archived:
1290 base = base.filter(Aircraft.archived_at.is_(None))
1291 role = current_user_role()
1292 if role in (Role.ADMIN, Role.OWNER):
1293 return base
1294 uid = session.get("user_id")
1295 if not uid:
1296 from sqlalchemy import false
1298 return base.filter(false())
1299 if UserAllAircraftAccess.query.filter_by(user_id=uid, tenant_id=tenant_id).first():
1300 return base
1301 ids = [
1302 row.aircraft_id
1303 for row in (
1304 UserAircraftAccess.query.filter_by(user_id=uid)
1305 .with_entities(UserAircraftAccess.aircraft_id)
1306 .all()
1307 )
1308 ]
1309 if not ids:
1310 from sqlalchemy import false
1312 return base.filter(false())
1313 return base.filter(Aircraft.id.in_(ids))
1316def compute_aircraft_statuses(
1317 aircraft_list: Any,
1318 triggers: Any,
1319 hobbs_by_id: Any,
1320 landings_by_id: Any = None,
1321 flight_hours_by_id: Any = None,
1322) -> dict[int, str]:
1323 """Return {aircraft_id: 'grounded'|'overdue'|'due_soon'|'ok'} for every aircraft.
1325 Grounded (expired insurance or unresolved grounding snag) takes priority.
1326 Among maintenance: overdue > due_soon > ok — maintenance triggers,
1327 component TBO / calendar life limits, and expiring insurance all count.
1329 ``landings_by_id``/``flight_hours_by_id`` default to an empty mapping —
1330 callers that don't (yet) pass them just get 'ok' from any landings- or
1331 flight-hours-basis trigger, same as passing no ``hobbs_by_id`` entry
1332 does for engine-hours-basis ones.
1333 """
1334 from services.component_limits import fleet_limit_statuses
1336 if landings_by_id is None:
1337 landings_by_id = {}
1338 if flight_hours_by_id is None:
1339 flight_hours_by_id = {}
1341 by_aircraft = defaultdict(list)
1342 for t in triggers:
1343 by_aircraft[t.aircraft_id].append(t)
1344 limit_status_by_ac = fleet_limit_statuses(aircraft_list)
1346 result = {}
1347 for ac in aircraft_list:
1348 if ac.is_grounded:
1349 result[ac.id] = "grounded"
1350 continue
1351 hobbs = hobbs_by_id.get(ac.id)
1352 landings = landings_by_id.get(ac.id)
1353 flight_hours = flight_hours_by_id.get(ac.id)
1354 statuses = [
1355 t.status(
1356 current_engine_hours=hobbs,
1357 current_landings=landings,
1358 current_flight_hours=flight_hours,
1359 )
1360 for t in by_aircraft.get(ac.id, [])
1361 ]
1362 statuses.append(limit_status_by_ac.get(ac.id, "ok"))
1363 ins = ac.insurance_status
1364 if ins == "expiring_soon":
1365 statuses.append("due_soon")
1366 if "overdue" in statuses:
1367 result[ac.id] = "overdue"
1368 elif "due_soon" in statuses:
1369 result[ac.id] = "due_soon"
1370 else:
1371 result[ac.id] = "ok"
1372 return result