Coverage for app/utils.py: 100%
672 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-31 10:13 +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_alog = logging.getLogger("openhangar.activity")
1030def _sl(value: object) -> str:
1031 """Sanitize a value for log output — strips CR/LF to prevent log injection (CWE-117)."""
1032 return str(value).replace("\r\n", "").replace("\n", "").replace("\r", "")
1035def activity(event: str, **fields: object) -> None:
1036 """Emit a structured [ACTIVITY] log entry with user_id and ip automatically included."""
1037 from flask import request, session
1039 uid = session.get("user_id", "")
1040 ip = request.remote_addr or ""
1041 parts = [f"[ACTIVITY] {event}", f"user_id={_sl(uid)}", f"ip={_sl(ip)}"]
1042 parts.extend(f"{k}={_sl(v)}" for k, v in fields.items())
1043 _alog.info(" ".join(parts))
1046class AircraftRefConverter(BaseConverter):
1047 """URL converter for the ``aircraft_id`` slot: accepts either the numeric
1048 primary key (unchanged, always works) or the aircraft's registration
1049 (e.g. ``OO-GRN``), so routes like ``/aircraft/<aircraft_id>/flights``
1050 also resolve ``/aircraft/OO-GRN/flights``.
1052 Authorization is unaffected: this only resolves the URL segment to a
1053 primary key (or the reverse, for link generation) — every view still
1054 does its own tenant-scoped lookup exactly as before, so a registration
1055 belonging to another tenant 404s the same way a wrong numeric id does.
1057 Registrations containing '/' or spaces (rare, but not forbidden by the
1058 model) are sanitized the same way upload filenames already are
1059 elsewhere in this blueprint; such an aircraft simply isn't reachable via
1060 its pretty URL (only via the numeric id, which always works).
1061 """
1063 regex = r"[^/]+"
1065 def to_python(self, value: str) -> int:
1066 if value.isdigit():
1067 return int(value)
1069 from models import Aircraft, db
1071 needle = value.upper()
1072 ac = (
1073 Aircraft.query.filter(
1074 db.func.upper(
1075 db.func.replace(
1076 db.func.replace(Aircraft.registration, "/", "-"), " ", "-"
1077 )
1078 )
1079 == needle
1080 )
1081 .order_by(Aircraft.id)
1082 .first()
1083 )
1084 if ac is None:
1085 raise ValidationError()
1086 return int(ac.id)
1088 def to_url(self, value: Any) -> str:
1089 if isinstance(value, str) and not value.isdigit():
1090 return super().to_url(value)
1092 from models import Aircraft, db
1094 ac = db.session.get(Aircraft, int(value))
1095 reg = ac.registration if ac and ac.registration else str(value)
1096 safe_reg = reg.replace("/", "-").replace(" ", "-")
1097 return super().to_url(safe_reg)
1100def login_required(f: Callable[..., Any]) -> Callable[..., Any]:
1101 """Redirect unauthenticated users to the login page."""
1103 @wraps(f)
1104 def decorated(*args: Any, **kwargs: Any) -> Any:
1105 if not session.get("user_id"):
1106 return redirect(url_for("auth.login"))
1107 return f(*args, **kwargs)
1109 return decorated
1112def require_instance_admin(f: Callable[..., Any]) -> Callable[..., Any]:
1113 """Abort 403 unless the current user is the instance admin."""
1115 @wraps(f)
1116 def decorated(*args: Any, **kwargs: Any) -> Any:
1117 from models import User, db
1119 user_id = session.get("user_id")
1120 if not user_id:
1121 return redirect(url_for("auth.login"))
1122 user = db.session.get(User, user_id)
1123 if not user or not user.is_instance_admin:
1124 abort(403)
1125 return f(*args, **kwargs)
1127 return decorated
1130def current_user_role() -> str | None:
1131 """Return the Role of the current user in their tenant, or None."""
1132 from models import TenantUser
1134 user_id = session.get("user_id")
1135 if not user_id:
1136 return None
1137 tu = TenantUser.query.filter_by(user_id=user_id).first()
1138 return tu.role if tu else None
1141def check_update_available() -> bool:
1142 """Return True when a newer release is available than the running instance.
1144 Reads the ``update_available`` AppSetting written by the background
1145 version-check service (computed once per check cycle, not per request).
1146 Falls back to a live comparison against ``latest_version`` if the flag has
1147 not been written yet (fresh install before the first background check).
1148 Returns False on any error. Must be called inside a request (or
1149 application) context.
1150 """
1151 try:
1152 from models import AppSetting, db # pyright: ignore[reportMissingImports]
1154 flag = db.session.get(AppSetting, "update_available")
1155 if flag is not None:
1156 return bool(flag.value == "true")
1158 # Fallback for fresh installs: compute from latest_version directly.
1159 from packaging.version import Version # pyright: ignore[reportMissingImports]
1161 current = os.environ.get("OPENHANGAR_VERSION", "development")
1162 if current == "development":
1163 return False
1164 latest_s = db.session.get(AppSetting, "latest_version")
1165 latest = latest_s.value if latest_s else None
1166 return bool(latest and Version(latest) > Version(current))
1167 except Exception: # noqa: BLE001 -- cosmetic nav-badge hint, degrade to "no update" on any error
1168 return False
1171def check_legacy_logbook_data() -> bool:
1172 """Return True when pre-unification logbook/crew data was preserved
1173 but not migrated during the ``flight_crew``/``pilot_logbook_entries``
1174 table removal.
1176 Reads the ``legacy_logbook_data_present`` AppSetting written by the
1177 Alembic migration that dropped those two tables — set only when either
1178 table still had rows at upgrade time (both tables are dropped outright,
1179 with no flag set, when they were already empty). Returns False on any
1180 error. Must be called inside a request (or application) context.
1181 """
1182 try:
1183 from models import AppSetting, db # pyright: ignore[reportMissingImports]
1185 flag = db.session.get(AppSetting, "legacy_logbook_data_present")
1186 return bool(flag is not None and flag.value == "true")
1187 except Exception: # noqa: BLE001 -- cosmetic nav-badge hint, degrade to "no legacy data" on any error
1188 return False
1191def require_role(*roles: str) -> Callable[..., Any]:
1192 """Decorator: abort 403 if the current user's role is not in *roles*."""
1194 def decorator(f: Callable[..., Any]) -> Callable[..., Any]:
1195 @wraps(f)
1196 def decorated(*args: Any, **kwargs: Any) -> Any:
1197 if current_user_role() not in roles:
1198 abort(403)
1199 return f(*args, **kwargs)
1201 return decorated
1203 return decorator
1206def require_pilot_access(f: Callable[..., Any]) -> Callable[..., Any]:
1207 """Decorator: abort 403 unless the user has pilot access.
1209 Pilot access is granted by ADMIN/OWNER/PILOT/STUDENT/INSTRUCTOR role,
1210 or by the per-user is_pilot capability flag.
1211 """
1213 @wraps(f)
1214 def decorated(*args: Any, **kwargs: Any) -> Any:
1215 from models import Role, User, db
1217 role = current_user_role()
1218 if role in (Role.ADMIN, Role.OWNER, Role.PILOT, Role.STUDENT, Role.INSTRUCTOR):
1219 return f(*args, **kwargs)
1220 uid = session.get("user_id")
1221 if uid:
1222 user = db.session.get(User, uid)
1223 if user and user.is_pilot:
1224 return f(*args, **kwargs)
1225 return abort(403)
1227 return decorated
1230def require_maint_access(f: Callable[..., Any]) -> Callable[..., Any]:
1231 """Decorator: abort 403 unless the user has maintenance access.
1233 Maintenance access is granted by ADMIN/OWNER/MAINTENANCE/INSTRUCTOR role,
1234 or by the per-user is_maintenance capability flag.
1235 """
1237 @wraps(f)
1238 def decorated(*args: Any, **kwargs: Any) -> Any:
1239 from models import Role, User, db
1241 role = current_user_role()
1242 if role in (Role.ADMIN, Role.OWNER, Role.MAINTENANCE, Role.INSTRUCTOR):
1243 return f(*args, **kwargs)
1244 uid = session.get("user_id")
1245 if uid:
1246 user = db.session.get(User, uid)
1247 if user and user.is_maintenance:
1248 return f(*args, **kwargs)
1249 return abort(403)
1251 return decorated
1254def user_can_access_aircraft(aircraft_id: int) -> bool:
1255 """Return True when the current user may access this aircraft.
1257 ADMIN and OWNER bypass the check entirely. Other roles need either a
1258 UserAllAircraftAccess row (all-planes grant) or a per-aircraft
1259 UserAircraftAccess row.
1260 """
1261 from models import Role, TenantUser, UserAircraftAccess, UserAllAircraftAccess
1263 role = current_user_role()
1264 if role in (Role.ADMIN, Role.OWNER):
1265 return True
1266 uid = session.get("user_id")
1267 if not uid:
1268 return False
1269 tu = TenantUser.query.filter_by(user_id=uid).first()
1270 if (
1271 tu
1272 and UserAllAircraftAccess.query.filter_by(
1273 user_id=uid, tenant_id=tu.tenant_id
1274 ).first()
1275 ):
1276 return True
1277 return (
1278 UserAircraftAccess.query.filter_by(user_id=uid, aircraft_id=aircraft_id).first()
1279 is not None
1280 )
1283def accessible_aircraft(tenant_id: int, include_archived: bool = False) -> Any:
1284 """Return a query of Aircraft the current user is allowed to see.
1286 ADMIN and OWNER see every aircraft in the tenant. A user with a
1287 UserAllAircraftAccess row for the tenant also sees all aircraft.
1288 Other roles see only aircraft granted via UserAircraftAccess.
1290 Archived aircraft are excluded unless include_archived is True — pass it
1291 for views that must keep showing an archived aircraft's history.
1292 """
1293 from models import Aircraft, Role, UserAircraftAccess, UserAllAircraftAccess
1295 base = Aircraft.query.filter_by(tenant_id=tenant_id).order_by(Aircraft.registration)
1296 if not include_archived:
1297 base = base.filter(Aircraft.archived_at.is_(None))
1298 role = current_user_role()
1299 if role in (Role.ADMIN, Role.OWNER):
1300 return base
1301 uid = session.get("user_id")
1302 if not uid:
1303 from sqlalchemy import false
1305 return base.filter(false())
1306 if UserAllAircraftAccess.query.filter_by(user_id=uid, tenant_id=tenant_id).first():
1307 return base
1308 ids = [
1309 row.aircraft_id
1310 for row in (
1311 UserAircraftAccess.query.filter_by(user_id=uid)
1312 .with_entities(UserAircraftAccess.aircraft_id)
1313 .all()
1314 )
1315 ]
1316 if not ids:
1317 from sqlalchemy import false
1319 return base.filter(false())
1320 return base.filter(Aircraft.id.in_(ids))
1323def compute_aircraft_statuses(
1324 aircraft_list: Any,
1325 triggers: Any,
1326 hobbs_by_id: Any,
1327 landings_by_id: Any = None,
1328 flight_hours_by_id: Any = None,
1329) -> dict[int, str]:
1330 """Return {aircraft_id: 'grounded'|'overdue'|'due_soon'|'ok'} for every aircraft.
1332 Grounded (expired insurance or unresolved grounding snag) takes priority.
1333 Among maintenance: overdue > due_soon > ok — maintenance triggers,
1334 component TBO / calendar life limits, and expiring insurance all count.
1336 ``landings_by_id``/``flight_hours_by_id`` default to an empty mapping —
1337 callers that don't (yet) pass them just get 'ok' from any landings- or
1338 flight-hours-basis trigger, same as passing no ``hobbs_by_id`` entry
1339 does for engine-hours-basis ones.
1340 """
1341 from services.component_limits import fleet_limit_statuses
1343 if landings_by_id is None:
1344 landings_by_id = {}
1345 if flight_hours_by_id is None:
1346 flight_hours_by_id = {}
1348 by_aircraft = defaultdict(list)
1349 for t in triggers:
1350 by_aircraft[t.aircraft_id].append(t)
1351 limit_status_by_ac = fleet_limit_statuses(aircraft_list)
1353 result = {}
1354 for ac in aircraft_list:
1355 if ac.is_grounded:
1356 result[ac.id] = "grounded"
1357 continue
1358 hobbs = hobbs_by_id.get(ac.id)
1359 landings = landings_by_id.get(ac.id)
1360 flight_hours = flight_hours_by_id.get(ac.id)
1361 statuses = [
1362 t.status(
1363 current_engine_hours=hobbs,
1364 current_landings=landings,
1365 current_flight_hours=flight_hours,
1366 )
1367 for t in by_aircraft.get(ac.id, [])
1368 ]
1369 statuses.append(limit_status_by_ac.get(ac.id, "ok"))
1370 ins = ac.insurance_status
1371 if ins == "expiring_soon":
1372 statuses.append("due_soon")
1373 if "overdue" in statuses:
1374 result[ac.id] = "overdue"
1375 elif "due_soon" in statuses:
1376 result[ac.id] = "due_soon"
1377 else:
1378 result[ac.id] = "ok"
1379 return result