Coverage for app/aircraft/routes.py: 100%
1019 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-06-28 23:33 +0000
1from typing import Any
3from flask import ( # pyright: ignore[reportMissingImports]
4 Blueprint,
5 abort,
6 current_app,
7 flash,
8 redirect,
9 render_template,
10 request,
11 session,
12 url_for,
13)
14from flask.typing import ResponseReturnValue # pyright: ignore[reportMissingImports]
16from flask_babel import gettext as _, ngettext # pyright: ignore[reportMissingImports]
17from werkzeug.utils import secure_filename # pyright: ignore[reportMissingImports]
19import json
20import os
21import uuid as _uuid_mod
23from models import (
24 Aircraft,
25 AircraftGpsImportBatch,
26 AircraftPhoto,
27 AirworthinessDocumentStatus,
28 AppSetting,
29 Component,
30 ComponentType,
31 DocType,
32 Document,
33 Expense,
34 ExpenseType,
35 FlightEntry,
36 FUEL_DENSITY,
37 GAL_TO_L,
38 GpsTrack,
39 MaintenanceTrigger,
40 PilotLogbookEntry,
41 Reservation,
42 ReservationStatus,
43 Role,
44 Snag,
45 TenantUser,
46 User,
47 WeightBalanceConfig,
48 WeightBalanceEntry,
49 WeightBalanceStation,
50 db,
51) # pyright: ignore[reportMissingImports]
52from aircraft.gps_import import ( # pyright: ignore[reportMissingImports]
53 detect_segments,
54 merge_and_sort,
55 parse_gps_file,
56 round_flight_time,
57)
58from utils import (
59 accessible_aircraft,
60 activity,
61 compute_aircraft_statuses,
62 get_aircraft_type_engine_info,
63 login_required,
64 require_role,
65 user_can_access_aircraft,
66) # pyright: ignore[reportMissingImports]
68aircraft_bp = Blueprint("aircraft", __name__, url_prefix="/aircraft")
70_OWNER_ROLES = (Role.ADMIN, Role.OWNER)
71_PILOT_ROLES = (Role.ADMIN, Role.OWNER, Role.PILOT)
74def _tenant_id() -> int:
75 """Return the tenant ID for the currently logged-in user."""
76 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
77 if not tu:
78 abort(403)
79 return int(tu.tenant_id)
82def _get_aircraft_or_404(aircraft_id: int) -> Aircraft:
83 """Fetch an aircraft that belongs to the current tenant and is accessible to the user."""
84 ac = db.session.get(Aircraft, aircraft_id)
85 if (
86 not ac
87 or ac.tenant_id != _tenant_id()
88 or not user_can_access_aircraft(aircraft_id)
89 ):
90 abort(404)
91 return ac
94def _get_component_or_404(aircraft: Aircraft, component_id: int) -> Component:
95 comp = db.session.get(Component, component_id)
96 if not comp or comp.aircraft_id != aircraft.id:
97 abort(404)
98 return comp
101# ── Aircraft list ─────────────────────────────────────────────────────────────
104@aircraft_bp.route("/")
105@login_required
106def list_aircraft() -> ResponseReturnValue:
107 from models import TenantProfile
109 if not request.args.get("list"):
110 tp = TenantProfile.query.filter_by(tenant_id=_tenant_id()).first()
111 if tp and tp.planned_aircraft_count == 1:
112 aircraft_list = accessible_aircraft(_tenant_id()).all()
113 if len(aircraft_list) == 1:
114 return redirect(
115 url_for("aircraft.detail", aircraft_id=aircraft_list[0].id)
116 )
118 aircraft = accessible_aircraft(_tenant_id()).all()
119 aircraft_ids = [ac.id for ac in aircraft]
120 hobbs_by_id = {ac.id: ac.total_engine_hours for ac in aircraft}
121 triggers = (
122 (
123 MaintenanceTrigger.query.filter(
124 MaintenanceTrigger.aircraft_id.in_(aircraft_ids)
125 ).all()
126 )
127 if aircraft_ids
128 else []
129 )
130 aircraft_status = compute_aircraft_statuses(aircraft, triggers, hobbs_by_id)
131 wb_configured_ids = {ac.id for ac in aircraft if ac.wb_config is not None}
132 cover_photos = (
133 {
134 p.aircraft_id: p
135 for p in AircraftPhoto.query.filter(
136 AircraftPhoto.aircraft_id.in_(aircraft_ids),
137 AircraftPhoto.sort_order == 1,
138 ).all()
139 }
140 if aircraft_ids
141 else {}
142 )
143 return render_template(
144 "aircraft/list.html",
145 aircraft=aircraft,
146 aircraft_status=aircraft_status,
147 wb_configured_ids=wb_configured_ids,
148 cover_photos=cover_photos,
149 )
152# ── Add aircraft ──────────────────────────────────────────────────────────────
155@aircraft_bp.route("/new", methods=["GET", "POST"])
156@login_required
157@require_role(*_OWNER_ROLES)
158def new_aircraft() -> ResponseReturnValue:
159 if request.method == "POST":
160 return _save_aircraft(None)
161 return render_template("aircraft/aircraft_form.html", aircraft=None)
164# ── Aircraft detail ───────────────────────────────────────────────────────────
167@aircraft_bp.route("/<int:aircraft_id>")
168@login_required
169def detail(aircraft_id: int) -> ResponseReturnValue:
170 from models import FlightEntry, MaintenanceTrigger
172 ac = _get_aircraft_or_404(aircraft_id)
173 components_by_type: dict[Any, list[Any]] = {}
174 for comp in sorted(ac.components, key=lambda c: (c.type, c.position or "")):
175 components_by_type.setdefault(comp.type, []).append(comp)
176 recent_flights = (
177 FlightEntry.query.filter_by(aircraft_id=ac.id)
178 .order_by(FlightEntry.date.desc(), FlightEntry.id.desc())
179 .limit(3)
180 .all()
181 )
182 current_hobbs = ac.total_engine_hours
183 triggers = MaintenanceTrigger.query.filter_by(aircraft_id=ac.id).all()
184 maintenance_summary = [(t, t.status(current_hobbs)) for t in triggers]
185 recent_expenses = (
186 Expense.query.filter_by(aircraft_id=ac.id)
187 .order_by(Expense.date.desc(), Expense.id.desc())
188 .limit(3)
189 .all()
190 )
191 recent_documents = (
192 Document.query.filter_by(aircraft_id=ac.id, is_sensitive=False)
193 .order_by(Document.uploaded_at.desc())
194 .limit(3)
195 .all()
196 )
197 document_count = Document.query.filter_by(aircraft_id=ac.id).count()
198 active_insurance_cert = (
199 Document.query.filter_by(
200 aircraft_id=ac.id,
201 doc_type=DocType.INSURANCE_CERT,
202 )
203 .filter(Document.superseded_by_id.is_(None))
204 .first()
205 )
206 open_snags = (
207 Snag.query.filter_by(aircraft_id=ac.id, resolved_at=None)
208 .order_by(Snag.is_grounding.desc(), Snag.reported_at.desc())
209 .all()
210 )
211 wb_cfg = ac.wb_config
212 last_wb_entry = None
213 if wb_cfg:
214 last_wb_entry = (
215 WeightBalanceEntry.query.filter_by(config_id=wb_cfg.id)
216 .order_by(WeightBalanceEntry.date.desc(), WeightBalanceEntry.id.desc())
217 .first()
218 )
219 from datetime import datetime, timezone as _tz
221 now = datetime.now(_tz.utc)
222 upcoming_reservations = (
223 Reservation.query.filter(
224 Reservation.aircraft_id == ac.id,
225 Reservation.status.in_(
226 [ReservationStatus.CONFIRMED, ReservationStatus.PENDING]
227 ),
228 Reservation.end_dt >= now,
229 )
230 .order_by(Reservation.start_dt)
231 .limit(5)
232 .all()
233 )
234 suggest_components = session.pop(f"suggest_components_{ac.id}", None)
235 photos = (
236 AircraftPhoto.query.filter_by(aircraft_id=ac.id)
237 .order_by(AircraftPhoto.sort_order)
238 .all()
239 )
240 aw_statuses = AirworthinessDocumentStatus.query.filter_by(aircraft_id=ac.id).all()
241 aw_counts: dict[str, int] = {}
242 for _st in aw_statuses:
243 aw_counts[_st.status] = aw_counts.get(_st.status, 0) + 1
244 aw_counts["total"] = len(aw_statuses)
245 _gps_entries = (
246 FlightEntry.query.filter_by(aircraft_id=ac.id)
247 .filter(FlightEntry.gps_track_id.isnot(None))
248 .order_by(FlightEntry.date.asc())
249 .all()
250 )
251 track_rows = [
252 {
253 "date": str(e.date),
254 "dep": e.departure_icao or "",
255 "arr": e.arrival_icao or "",
256 "time_str": f"{e.flight_time} h" if e.flight_time is not None else "",
257 "view_url": url_for(
258 "aircraft.flight_detail",
259 aircraft_id=aircraft_id,
260 flight_id=e.id,
261 ),
262 "geojson": e.gps_track.geojson if e.gps_track else None,
263 }
264 for e in _gps_entries
265 ]
266 _tile = db.session.get(AppSetting, "openaip_api_key")
267 openaip_key = _tile.value if _tile and _tile.value else None
268 return render_template(
269 "aircraft/detail.html",
270 aircraft=ac,
271 components_by_type=components_by_type,
272 component_types=ComponentType,
273 recent_flights=recent_flights,
274 suggest_components=suggest_components,
275 maintenance_summary=maintenance_summary,
276 recent_expenses=recent_expenses,
277 expense_type_labels=ExpenseType.LABELS,
278 recent_documents=recent_documents,
279 document_count=document_count,
280 active_insurance_cert=active_insurance_cert,
281 open_snags=open_snags,
282 wb_config=wb_cfg,
283 last_wb_entry=last_wb_entry,
284 upcoming_reservations=upcoming_reservations,
285 ReservationStatus=ReservationStatus,
286 photos=photos,
287 aw_counts=aw_counts,
288 track_rows=track_rows,
289 openaip_key=openaip_key,
290 )
293# ── Edit aircraft ─────────────────────────────────────────────────────────────
296@aircraft_bp.route("/<int:aircraft_id>/edit", methods=["GET", "POST"])
297@login_required
298@require_role(*_OWNER_ROLES)
299def edit_aircraft(aircraft_id: int) -> ResponseReturnValue:
300 ac = _get_aircraft_or_404(aircraft_id)
301 if request.method == "POST":
302 return _save_aircraft(ac)
303 return render_template("aircraft/aircraft_form.html", aircraft=ac)
306def _save_aircraft(ac: Aircraft | None) -> ResponseReturnValue:
307 is_new = ac is None
308 icao_type = request.form.get("aircraft_type_icao", "").strip().upper()
309 registration = request.form.get("registration", "").strip().upper()
310 make = request.form.get("make", "").strip()
311 model = request.form.get("model", "").strip()
312 year_raw = request.form.get("year", "").strip()
313 has_flight_counter = bool(request.form.get("has_flight_counter"))
314 flight_counter_offset_raw = request.form.get("flight_counter_offset", "0.3").strip()
315 fuel_flow_raw = request.form.get("fuel_flow", "").strip()
316 fuel_type = request.form.get("fuel_type", "avgas").strip()
317 if fuel_type not in ("avgas", "ul91", "mogas", "jet_a1"):
318 fuel_type = "avgas"
319 insurance_expiry_raw = request.form.get("insurance_expiry", "").strip()
320 logbook_time_precision = request.form.get(
321 "logbook_time_precision", "tenth_hour"
322 ).strip()
323 if logbook_time_precision not in ("tenth_hour", "minute"):
324 logbook_time_precision = "tenth_hour"
326 errors = []
327 if not registration:
328 errors.append(_("Registration is required."))
329 if not make:
330 errors.append(_("Manufacturer is required."))
331 if not model:
332 errors.append(_("Model is required."))
333 year = None
334 if year_raw:
335 try:
336 year = int(year_raw)
337 if not (1900 <= year <= 2100):
338 raise ValueError
339 except ValueError:
340 errors.append(_("Year must be a valid 4-digit year."))
342 flight_counter_offset = 0.3
343 if flight_counter_offset_raw:
344 try:
345 flight_counter_offset = float(flight_counter_offset_raw)
346 if flight_counter_offset < 0:
347 raise ValueError
348 except ValueError:
349 errors.append(_("Flight counter offset must be a non-negative number."))
351 fuel_flow = None
352 if fuel_flow_raw:
353 try:
354 fuel_flow = float(fuel_flow_raw)
355 if fuel_flow < 0:
356 raise ValueError
357 except ValueError:
358 errors.append(_("Fuel consumption must be a non-negative number."))
360 insurance_expiry = None
361 if insurance_expiry_raw:
362 from datetime import date as _date
364 try:
365 insurance_expiry = _date.fromisoformat(insurance_expiry_raw)
366 except ValueError:
367 errors.append(_("Insurance expiry must be a valid date (YYYY-MM-DD)."))
369 if errors:
370 for msg in errors:
371 flash(msg, "danger")
372 return render_template("aircraft/aircraft_form.html", aircraft=ac)
374 if ac is None:
375 ac = Aircraft(tenant_id=_tenant_id())
376 db.session.add(ac)
378 ac.registration = registration
379 ac.make = make
380 ac.model = model
381 ac.year = year
382 ac.has_flight_counter = has_flight_counter
383 ac.flight_counter_offset = flight_counter_offset
384 ac.fuel_flow = fuel_flow
385 ac.fuel_type = fuel_type
386 ac.insurance_expiry = insurance_expiry
387 ac.logbook_time_precision = logbook_time_precision
388 db.session.commit()
390 if is_new:
391 activity("aircraft.created", registration=ac.registration, aircraft_id=ac.id)
392 else:
393 activity("aircraft.updated", registration=ac.registration, aircraft_id=ac.id)
395 if is_new and icao_type:
396 engine_info = get_aircraft_type_engine_info(icao_type)
397 if engine_info:
398 ec, et = engine_info
399 if et == "Piston" and not ac.components:
400 session[f"suggest_components_{ac.id}"] = {"engine_count": ec}
402 flash(_("%(reg)s saved.", reg=ac.registration), "success")
403 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
406# ── Delete aircraft ───────────────────────────────────────────────────────────
409@aircraft_bp.route("/<int:aircraft_id>/delete", methods=["POST"])
410@login_required
411@require_role(*_OWNER_ROLES)
412def delete_aircraft(aircraft_id: int) -> ResponseReturnValue:
413 ac = _get_aircraft_or_404(aircraft_id)
414 reg = ac.registration
415 activity("aircraft.deleted", registration=reg, aircraft_id=aircraft_id)
416 db.session.delete(ac)
417 db.session.commit()
418 flash(_("%(reg)s and all its components have been deleted.", reg=reg), "success")
419 return redirect(url_for("aircraft.list_aircraft"))
422# ── Quick-add components from ICAO type suggestion ────────────────────────────
425@aircraft_bp.route("/<int:aircraft_id>/quick-add-components", methods=["POST"])
426@login_required
427@require_role(*_OWNER_ROLES)
428def quick_add_components(aircraft_id: int) -> ResponseReturnValue:
429 ac = _get_aircraft_or_404(aircraft_id)
430 try:
431 engine_count = max(1, min(int(request.form.get("engine_count", "1")), 4))
432 except ValueError:
433 engine_count = 1
434 for i in range(engine_count):
435 position = str(i + 1) if engine_count > 1 else None
436 db.session.add(
437 Component(
438 aircraft_id=ac.id,
439 type=ComponentType.ENGINE,
440 position=position,
441 make="",
442 model="",
443 )
444 )
445 db.session.add(
446 Component(
447 aircraft_id=ac.id,
448 type=ComponentType.PROPELLER,
449 position=position,
450 make="",
451 model="",
452 )
453 )
454 db.session.commit()
455 activity(
456 "component.quick_added",
457 aircraft_id=aircraft_id,
458 engine_count=engine_count,
459 )
460 flash(
461 ngettext(
462 "Engine and propeller added — fill in the details when ready.",
463 "%(n)s engines and propellers added — fill in the details when ready.",
464 engine_count,
465 n=engine_count,
466 ),
467 "success",
468 )
469 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
472# ── Add component ─────────────────────────────────────────────────────────────
475@aircraft_bp.route("/<int:aircraft_id>/components/new", methods=["GET", "POST"])
476@login_required
477@require_role(*_OWNER_ROLES)
478def new_component(aircraft_id: int) -> ResponseReturnValue:
479 ac = _get_aircraft_or_404(aircraft_id)
480 if request.method == "POST":
481 return _save_component(ac, None)
482 return render_template(
483 "aircraft/component_form.html",
484 aircraft=ac,
485 component=None,
486 component_types=ComponentType,
487 )
490# ── Edit component ────────────────────────────────────────────────────────────
493@aircraft_bp.route(
494 "/<int:aircraft_id>/components/<int:component_id>/edit", methods=["GET", "POST"]
495)
496@login_required
497@require_role(*_OWNER_ROLES)
498def edit_component(aircraft_id: int, component_id: int) -> ResponseReturnValue:
499 ac = _get_aircraft_or_404(aircraft_id)
500 comp = _get_component_or_404(ac, component_id)
501 if request.method == "POST":
502 return _save_component(ac, comp)
503 return render_template(
504 "aircraft/component_form.html",
505 aircraft=ac,
506 component=comp,
507 component_types=ComponentType,
508 )
511def _save_component(ac: Aircraft, comp: Component | None) -> ResponseReturnValue:
512 from datetime import date as _date
514 type_ = request.form.get("type", "").strip()
515 position = request.form.get("position", "").strip() or None
516 make = request.form.get("make", "").strip()
517 model = request.form.get("model", "").strip()
518 serial = request.form.get("serial_number", "").strip() or None
519 time_raw = request.form.get("time_at_install", "").strip()
520 installed_raw = request.form.get("installed_at", "").strip()
521 removed_raw = request.form.get("removed_at", "").strip()
523 errors = []
524 if not type_:
525 errors.append(_("Component type is required."))
526 if not make:
527 errors.append(_("Manufacturer is required."))
528 if not model:
529 errors.append(_("Model is required."))
531 time_at_install = None
532 if time_raw:
533 try:
534 time_at_install = float(time_raw)
535 if time_at_install < 0:
536 raise ValueError
537 except ValueError:
538 errors.append(_("Time at install must be a positive number."))
540 def _parse_date(raw: str, label: str) -> Any:
541 if not raw:
542 return None
543 try:
544 return _date.fromisoformat(raw)
545 except ValueError:
546 errors.append(
547 _("%(label)s must be a valid date (YYYY-MM-DD).", label=label)
548 )
549 return None
551 installed_at = _parse_date(installed_raw, "Install date")
552 removed_at = _parse_date(removed_raw, "Removal date")
554 if errors:
555 for msg in errors:
556 flash(msg, "danger")
557 return render_template(
558 "aircraft/component_form.html",
559 aircraft=ac,
560 component=comp,
561 component_types=ComponentType,
562 )
564 _comp_is_new = comp is None
565 if comp is None:
566 comp = Component(aircraft_id=ac.id)
567 db.session.add(comp)
569 comp.type = type_
570 comp.position = position
571 comp.make = make
572 comp.model = model
573 comp.serial_number = serial
574 comp.time_at_install = time_at_install
575 comp.installed_at = installed_at
576 comp.removed_at = removed_at
577 db.session.commit()
579 if _comp_is_new:
580 activity(
581 "component.added", type=comp.type, component_id=comp.id, aircraft_id=ac.id
582 )
583 else:
584 activity(
585 "component.updated", type=comp.type, component_id=comp.id, aircraft_id=ac.id
586 )
588 flash(_("%(make)s %(model)s saved.", make=comp.make, model=comp.model), "success")
589 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
592# ── Delete component ──────────────────────────────────────────────────────────
595@aircraft_bp.route(
596 "/<int:aircraft_id>/components/<int:component_id>/delete", methods=["POST"]
597)
598@login_required
599@require_role(*_OWNER_ROLES)
600def delete_component(aircraft_id: int, component_id: int) -> ResponseReturnValue:
601 ac = _get_aircraft_or_404(aircraft_id)
602 comp = _get_component_or_404(ac, component_id)
603 label = f"{comp.make} {comp.model}"
604 activity(
605 "component.deleted",
606 type=comp.type,
607 component_id=component_id,
608 aircraft_id=aircraft_id,
609 )
610 db.session.delete(comp)
611 db.session.commit()
612 flash(_("%(label)s removed.", label=label), "success")
613 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
616# ── Mass & Balance: helpers ───────────────────────────────────────────────────
619def _point_in_polygon(cg: float, weight: float, points: Any) -> bool:
620 """Ray-casting point-in-polygon test. points: list of [arm, weight] pairs."""
621 n = len(points)
622 inside = False
623 j = n - 1
624 for i in range(n):
625 xi, yi = float(points[i][0]), float(points[i][1])
626 xj, yj = float(points[j][0]), float(points[j][1])
627 if ((yi > weight) != (yj > weight)) and (
628 cg < (xj - xi) * (weight - yi) / (yj - yi) + xi
629 ):
630 inside = not inside
631 j = i
632 return inside
635# ── Mass & Balance: config ────────────────────────────────────────────────────
638@aircraft_bp.route("/<int:aircraft_id>/wb/config", methods=["GET", "POST"])
639@login_required
640@require_role(*_OWNER_ROLES)
641def wb_config(aircraft_id: int) -> ResponseReturnValue:
642 ac = _get_aircraft_or_404(aircraft_id)
643 cfg: WeightBalanceConfig | None = ac.wb_config # type: ignore[assignment]
645 if request.method == "POST":
646 errors = []
648 def _f(name: str) -> float | None:
649 try:
650 v = float(request.form.get(name, "").strip())
651 if v < 0:
652 raise ValueError
653 return v
654 except ValueError:
655 errors.append(_("%(field)s must be a positive number.", field=name))
656 return None
658 empty_weight = _f("empty_weight")
659 empty_cg_arm = _f("empty_cg_arm")
660 max_takeoff_weight = _f("max_takeoff_weight")
661 forward_cg_limit = _f("forward_cg_limit")
662 aft_cg_limit = _f("aft_cg_limit")
663 datum_note = request.form.get("datum_note", "").strip() or None
665 fuel_unit = request.form.get("fuel_unit", "L").strip()
666 if fuel_unit not in ("L", "gal"):
667 fuel_unit = "L"
669 # Stations: label[], arm[], station_limit[] (capacity for fuel, max_weight for non-fuel), is_fuel[]
670 labels = request.form.getlist("station_label[]")
671 arms = request.form.getlist("station_arm[]")
672 limits = request.form.getlist("station_limit[]")
673 is_fuels = request.form.getlist(
674 "station_is_fuel[]"
675 ) # index values of checked boxes
677 if not labels or all(lbl.strip() == "" for lbl in labels):
678 errors.append(_("At least one loading station is required."))
680 if errors:
681 for msg in errors:
682 flash(msg, "danger")
683 return render_template("aircraft/wb_config.html", aircraft=ac, config=cfg)
685 if cfg is None:
686 cfg = WeightBalanceConfig(aircraft_id=ac.id)
687 db.session.add(cfg)
689 # Optional envelope polygon: env_arm[], env_weight[]
690 env_arms = request.form.getlist("env_arm[]")
691 env_weights = request.form.getlist("env_weight[]")
692 envelope_points = []
693 for arm_s, w_s in zip(env_arms, env_weights):
694 try:
695 a = float(arm_s.strip())
696 w = float(w_s.strip())
697 if a >= 0 and w >= 0:
698 envelope_points.append([round(a, 4), round(w, 2)])
699 except (ValueError, AttributeError):
700 continue
702 cfg.empty_weight = empty_weight
703 cfg.empty_cg_arm = empty_cg_arm
704 cfg.max_takeoff_weight = max_takeoff_weight
705 cfg.forward_cg_limit = forward_cg_limit
706 cfg.aft_cg_limit = aft_cg_limit
707 cfg.fuel_unit = fuel_unit
708 cfg.datum_note = datum_note
709 cfg.envelope_points = envelope_points if len(envelope_points) >= 3 else None
711 # Replace stations
712 for s in list(cfg.stations):
713 db.session.delete(s)
714 db.session.flush()
716 for i, label in enumerate(labels):
717 label = label.strip()
718 if not label:
719 continue
720 try:
721 arm = float(arms[i])
722 except (ValueError, IndexError):
723 continue
724 limit_val = None
725 try:
726 lim_raw = limits[i].strip()
727 if lim_raw:
728 limit_val = float(lim_raw)
729 except (ValueError, IndexError):
730 limit_val = None
731 is_fuel = str(i) in is_fuels
732 db.session.add(
733 WeightBalanceStation(
734 config_id=cfg.id,
735 label=label,
736 arm=arm,
737 max_weight=None if is_fuel else limit_val,
738 capacity=limit_val if is_fuel else None,
739 is_fuel=is_fuel,
740 position=i,
741 )
742 )
744 db.session.commit()
745 flash(_("W&B configuration saved."), "success")
746 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
748 return render_template("aircraft/wb_config.html", aircraft=ac, config=cfg)
751# ── Mass & Balance: entry list ────────────────────────────────────────────────
754@aircraft_bp.route("/<int:aircraft_id>/wb/")
755@login_required
756def wb_list(aircraft_id: int) -> ResponseReturnValue:
757 ac = _get_aircraft_or_404(aircraft_id)
758 if not ac.wb_config:
759 flash(_("Configure W&B envelope first."), "warning")
760 return redirect(url_for("aircraft.wb_config", aircraft_id=ac.id))
761 entries = (
762 WeightBalanceEntry.query.filter_by(config_id=ac.wb_config.id)
763 .order_by(WeightBalanceEntry.date.desc(), WeightBalanceEntry.id.desc())
764 .all()
765 )
766 return render_template(
767 "aircraft/wb_list.html", aircraft=ac, config=ac.wb_config, entries=entries
768 )
771# ── Mass & Balance: new / edit entry ─────────────────────────────────────────
774@aircraft_bp.route("/<int:aircraft_id>/wb/new", methods=["GET", "POST"])
775@aircraft_bp.route("/<int:aircraft_id>/wb/<int:entry_id>/edit", methods=["GET", "POST"])
776@login_required
777@require_role(*_PILOT_ROLES)
778def wb_entry(aircraft_id: int, entry_id: int | None = None) -> ResponseReturnValue:
779 ac = _get_aircraft_or_404(aircraft_id)
780 if not ac.wb_config:
781 flash(_("Configure W&B envelope first."), "warning")
782 return redirect(url_for("aircraft.wb_config", aircraft_id=ac.id))
783 cfg = ac.wb_config
785 entry = None
786 if entry_id is not None:
787 entry = db.session.get(WeightBalanceEntry, entry_id)
788 if not entry or entry.config_id != cfg.id:
789 abort(404)
791 if request.method == "POST":
792 from datetime import date as _date
794 errors = []
795 date_raw = request.form.get("date", "").strip()
796 label = request.form.get("label", "").strip() or None
797 try:
798 entry_date = _date.fromisoformat(date_raw)
799 except ValueError:
800 errors.append(_("A valid date is required."))
801 entry_date = None
803 # Per-station values: fuel stations store volume (L/gal), non-fuel store kg
804 station_weights = {}
805 for st in cfg.stations:
806 if st.is_fuel:
807 raw = request.form.get(f"volume_{st.id}", "").strip()
808 try:
809 vol = float(raw) if raw else 0.0
810 if vol < 0:
811 raise ValueError
812 if st.capacity is not None and vol > float(st.capacity):
813 errors.append(
814 _(
815 "Volume for %(station)s exceeds tank capacity.",
816 station=st.label,
817 )
818 )
819 station_weights[str(st.id)] = vol
820 except ValueError:
821 errors.append(
822 _(
823 "Volume for %(station)s must be a non-negative number.",
824 station=st.label,
825 )
826 )
827 else:
828 raw = request.form.get(f"weight_{st.id}", "").strip()
829 try:
830 w = float(raw) if raw else 0.0
831 if w < 0:
832 raise ValueError
833 station_weights[str(st.id)] = w
834 except ValueError:
835 errors.append(
836 _(
837 "Weight for %(station)s must be a non-negative number.",
838 station=st.label,
839 )
840 )
842 # CG computation — fuel stations: convert volume → kg
843 empty_w = float(cfg.empty_weight)
844 empty_arm = float(cfg.empty_cg_arm)
845 total_moment = empty_w * empty_arm
846 total_weight = empty_w
847 fuel_density = FUEL_DENSITY.get(ac.fuel_type, 0.72)
848 gal_factor = GAL_TO_L if cfg.fuel_unit == "gal" else 1.0
849 for st in cfg.stations:
850 val = station_weights.get(str(st.id), 0.0)
851 w_kg = val * fuel_density * gal_factor if st.is_fuel else val
852 total_weight += w_kg
853 total_moment += w_kg * float(st.arm)
854 loaded_cg = total_moment / total_weight if total_weight else 0.0
856 if cfg.envelope_points and len(cfg.envelope_points) >= 3:
857 in_env = _point_in_polygon(loaded_cg, total_weight, cfg.envelope_points)
858 else:
859 mtow = float(cfg.max_takeoff_weight)
860 fwd = float(cfg.forward_cg_limit)
861 aft = float(cfg.aft_cg_limit)
862 in_env = total_weight <= mtow and fwd <= loaded_cg <= aft
864 if errors:
865 for msg in errors:
866 flash(msg, "danger")
867 return render_template(
868 "aircraft/wb_entry.html",
869 aircraft=ac,
870 config=cfg,
871 entry=entry,
872 fuel_density=FUEL_DENSITY,
873 )
875 if entry is None:
876 entry = WeightBalanceEntry(config_id=cfg.id)
877 db.session.add(entry)
879 entry.date = entry_date
880 entry.label = label
881 entry.total_weight = round(total_weight, 2)
882 entry.loaded_cg = round(loaded_cg, 2)
883 entry.is_in_envelope = in_env
884 entry.station_weights = station_weights
885 db.session.commit()
886 flash(_("W&B calculation saved."), "success")
887 return redirect(url_for("aircraft.wb_list", aircraft_id=ac.id))
889 return render_template(
890 "aircraft/wb_entry.html",
891 aircraft=ac,
892 config=cfg,
893 entry=entry,
894 fuel_density=FUEL_DENSITY,
895 )
898# ── Mass & Balance: delete entry ──────────────────────────────────────────────
901@aircraft_bp.route("/<int:aircraft_id>/wb/<int:entry_id>/delete", methods=["POST"])
902@login_required
903@require_role(*_PILOT_ROLES)
904def wb_entry_delete(aircraft_id: int, entry_id: int) -> ResponseReturnValue:
905 ac = _get_aircraft_or_404(aircraft_id)
906 if not ac.wb_config:
907 abort(404)
908 entry = db.session.get(WeightBalanceEntry, entry_id)
909 if not entry or entry.config_id != ac.wb_config.id:
910 abort(404)
911 db.session.delete(entry)
912 db.session.commit()
913 flash(_("W&B calculation deleted."), "success")
914 return redirect(url_for("aircraft.wb_list", aircraft_id=ac.id))
917# ── Phase 30: GPS Log Import ──────────────────────────────────────────────────
919_GPS_ALLOWED_EXTS = {".gpx", ".kml", ".csv"}
920_GPS_MAX_BYTES = 20 * 1024 * 1024 # 20 MB per file
923def _gps_tmp_dir() -> str:
924 """Return (and create if needed) the tmp directory for GPS uploads."""
925 upload_folder = current_app.config.get("UPLOAD_FOLDER", "/tmp")
926 d = os.path.join(upload_folder, "gps_import_tmp")
927 os.makedirs(d, exist_ok=True)
928 return d
931def _segment_to_dict(seg: Any, idx: int) -> dict[str, Any]:
932 """Serialise a FlightSegment for template rendering (includes track_geojson)."""
933 return {
934 "idx": idx,
935 "block_off_utc": seg.block_off_utc.isoformat(),
936 "block_on_utc": seg.block_on_utc.isoformat(),
937 "takeoff_utc": seg.takeoff_utc.isoformat() if seg.takeoff_utc else None,
938 "landing_utc": seg.landing_utc.isoformat() if seg.landing_utc else None,
939 "departure_icao": seg.departure_icao or "",
940 "arrival_icao": seg.arrival_icao or "",
941 "flight_time_raw_h": seg.flight_time_raw_h,
942 "flight_time_rounded_h": seg.flight_time_rounded_h,
943 "landing_count": seg.landing_count,
944 "is_ground_only": seg.is_ground_only,
945 "track_geojson": seg.track_geojson,
946 }
949def _linked_pilot_entries(flight_id: int, exclude_user_id: int) -> list[dict[str, Any]]:
950 """Return metadata about other users' PilotLogbookEntry records linked to a FlightEntry.
952 GPS track conflict policy (applied in _gps_import_create_segment and
953 pilot_gps_import_confirm_one):
954 - Airframe log (FlightEntry.gps_track_id): always replaced by the new upload.
955 The person uploading from the aircraft (e.g. owner with avionics data) is
956 considered authoritative for the airframe record.
957 - Other pilots' logbook entries (PilotLogbookEntry.gps_track_id): linked to
958 the new track ONLY if their entry currently has no GPS track (gps_track_id IS
959 NULL). If a pilot already linked their own track, that is preserved —
960 each pilot controls their own logbook. A discrepancy between the airframe
961 track and a pilot's track is acceptable; the review screen surfaces it so the
962 uploader is aware before confirming.
963 """
964 rows = PilotLogbookEntry.query.filter(
965 PilotLogbookEntry.flight_id == flight_id,
966 PilotLogbookEntry.pilot_user_id != exclude_user_id,
967 ).all()
968 result = []
969 for ple in rows:
970 user = db.session.get(User, ple.pilot_user_id)
971 result.append(
972 {
973 "user_id": ple.pilot_user_id,
974 "display_name": user.display_name
975 if user
976 else f"user #{ple.pilot_user_id}",
977 "has_existing_track": ple.gps_track_id is not None,
978 }
979 )
980 return result
983def _load_segment_geojson(seg: dict[str, Any]) -> Any:
984 """Read the GeoJSON dict back from the tmp file written by _segment_for_session."""
985 path = seg.get("geojson_path")
986 if not path or not os.path.exists(path):
987 return None
988 with open(path, encoding="utf-8") as fh:
989 return json.load(fh)
992def _segment_for_session(seg_dict: dict[str, Any], tmp_dir: str) -> dict[str, Any]:
993 """Return a copy of seg_dict safe for cookie-session storage.
995 track_geojson can be hundreds of KB — too large for Flask's 4 KB cookie
996 limit. We spill it to a tmp file and store the path instead.
997 """
998 s = {k: v for k, v in seg_dict.items() if k != "track_geojson"}
999 geojson = seg_dict.get("track_geojson")
1000 if geojson is not None:
1001 fname = f"seg_{seg_dict['idx']}_{_uuid_mod.uuid4().hex}.geojson"
1002 path = os.path.join(tmp_dir, fname)
1003 with open(path, "w", encoding="utf-8") as fh:
1004 json.dump(geojson, fh)
1005 s["geojson_path"] = path
1006 return s
1009@aircraft_bp.route("/<int:aircraft_id>/gps-import", methods=["GET", "POST"])
1010@login_required
1011@require_role(*_PILOT_ROLES)
1012def gps_import_upload(aircraft_id: int) -> ResponseReturnValue:
1013 ac = _get_aircraft_or_404(aircraft_id)
1015 if request.method == "GET":
1016 return render_template("aircraft/gps_import_upload.html", aircraft=ac)
1018 files = request.files.getlist("gps_files")
1019 if not files or all(f.filename == "" for f in files):
1020 flash(_("Please select at least one GPS log file."), "warning")
1021 return render_template("aircraft/gps_import_upload.html", aircraft=ac)
1023 tmp_dir = _gps_tmp_dir()
1024 parsed_meta: list[dict[str, Any]] = []
1025 errors: list[str] = []
1026 skipped_empty = 0
1027 formats: list[str] = []
1029 for f in files:
1030 if not f.filename:
1031 continue
1032 ext = os.path.splitext(f.filename.lower())[1]
1033 if ext not in _GPS_ALLOWED_EXTS:
1034 errors.append(
1035 _(
1036 "%(fn)s: unsupported file type (use .gpx, .kml, or .csv).",
1037 fn=f.filename,
1038 )
1039 )
1040 continue
1042 data = f.read(_GPS_MAX_BYTES + 1)
1043 if len(data) > _GPS_MAX_BYTES:
1044 errors.append(_("%(fn)s: file too large (20 MB limit).", fn=f.filename))
1045 continue
1047 try:
1048 parsed = parse_gps_file(data, f.filename)
1049 except ValueError as exc:
1050 errors.append(_("%(fn)s: %(err)s", fn=f.filename, err=str(exc)))
1051 continue
1053 if parsed.classification == "empty":
1054 skipped_empty += 1
1055 continue
1057 # Save raw bytes to tmp
1058 uid = _uuid_mod.uuid4().hex
1059 safe_name = f"{uid}_{secure_filename(f.filename)}"
1060 tmp_path = os.path.join(tmp_dir, safe_name)
1061 with open(tmp_path, "wb") as fh:
1062 fh.write(data)
1064 parsed_meta.append(
1065 {
1066 "tmp_path": tmp_path,
1067 "original_filename": f.filename,
1068 "format": parsed.format,
1069 "classification": parsed.classification,
1070 "trkpt_count": len(parsed.trackpoints),
1071 "hint_dep": parsed.hint_departure_icao,
1072 "hint_arr": parsed.hint_arrival_icao,
1073 "device_id": getattr(parsed, "device_id", None),
1074 }
1075 )
1076 formats.append(parsed.format)
1078 if errors:
1079 for e in errors:
1080 flash(e, "danger")
1081 if skipped_empty:
1082 flash(
1083 ngettext(
1084 "%(n)s file skipped — no movement detected.",
1085 "%(n)s files skipped — no movement detected.",
1086 skipped_empty,
1087 n=skipped_empty,
1088 ),
1089 "info",
1090 )
1092 if not parsed_meta:
1093 flash(_("No valid GPS files to import."), "warning")
1094 return render_template("aircraft/gps_import_upload.html", aircraft=ac)
1096 other_aircraft = request.form.get("other_aircraft") == "1"
1097 other_ac_make_model = request.form.get("other_ac_make_model", "").strip()
1098 other_ac_reg = request.form.get("other_ac_reg", "").strip().upper()
1100 session["gps_import"] = {
1101 "user_id": session["user_id"],
1102 "aircraft_id": aircraft_id,
1103 "files": parsed_meta,
1104 "skipped_empty": skipped_empty,
1105 "other_aircraft": other_aircraft,
1106 "other_ac_make_model": other_ac_make_model,
1107 "other_ac_reg": other_ac_reg,
1108 }
1109 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
1112@aircraft_bp.route("/<int:aircraft_id>/gps-import/review", methods=["GET"])
1113@login_required
1114@require_role(*_PILOT_ROLES)
1115def gps_import_review(aircraft_id: int) -> ResponseReturnValue:
1116 ac = _get_aircraft_or_404(aircraft_id)
1117 state = session.get("gps_import")
1118 if not state or state.get("aircraft_id") != aircraft_id:
1119 flash(_("Session expired — please upload your GPS files again."), "warning")
1120 return redirect(url_for("aircraft.gps_import_upload", aircraft_id=aircraft_id))
1122 file_metas = state["files"]
1124 # Re-parse each tmp file and build combined trackpoint list
1125 from aircraft.gps_import import ParsedGpsFile # pyright: ignore[reportMissingImports]
1127 all_parsed: list[ParsedGpsFile] = []
1128 for meta in file_metas:
1129 try:
1130 with open(meta["tmp_path"], "rb") as fh:
1131 data = fh.read()
1132 parsed = parse_gps_file(data, meta["original_filename"])
1133 parsed.hint_departure_icao = meta.get("hint_dep")
1134 parsed.hint_arrival_icao = meta.get("hint_arr")
1135 all_parsed.append(parsed)
1136 except (OSError, ValueError):
1137 flash(
1138 _(
1139 "Could not read %(fn)s — please upload again.",
1140 fn=meta["original_filename"],
1141 ),
1142 "warning",
1143 )
1144 return redirect(
1145 url_for("aircraft.gps_import_upload", aircraft_id=aircraft_id)
1146 )
1148 merged = merge_and_sort(all_parsed)
1150 # Collect ICAO hints from all files
1151 hint_dep = next(
1152 (p.hint_departure_icao for p in all_parsed if p.hint_departure_icao), None
1153 )
1154 hint_arr = next(
1155 (p.hint_arrival_icao for p in all_parsed if p.hint_arrival_icao), None
1156 )
1158 segments = detect_segments(
1159 merged,
1160 aircraft_precision=ac.logbook_time_precision,
1161 hint_dep=hint_dep,
1162 hint_arr=hint_arr,
1163 )
1165 # Build full dicts (with track_geojson) for template rendering.
1166 # Spill GeoJSON to tmp files for the session — track_geojson can be
1167 # hundreds of KB and silently overflows Flask's 4 KB cookie session.
1168 full_segs = [_segment_to_dict(seg, i) for i, seg in enumerate(segments)]
1170 # Duplicate detection: find existing FlightEntry records that overlap each segment.
1171 from datetime import datetime as _dt, timedelta as _td # noqa: PLC0415
1173 _BLOCK_TOLERANCE = _td(minutes=15)
1175 for seg in full_segs:
1176 block_off = _dt.fromisoformat(seg["block_off_utc"])
1177 block_on = _dt.fromisoformat(seg["block_on_utc"])
1178 matched = FlightEntry.query.filter(
1179 FlightEntry.aircraft_id == aircraft_id,
1180 FlightEntry.block_off_utc.isnot(None),
1181 FlightEntry.block_on_utc.isnot(None),
1182 FlightEntry.block_off_utc < block_on + _BLOCK_TOLERANCE,
1183 FlightEntry.block_on_utc > block_off - _BLOCK_TOLERANCE,
1184 ).first()
1185 if matched:
1186 seg["matched_flight_id"] = matched.id
1187 seg["matched_flight_str"] = (
1188 f"#{matched.id} — {matched.date} "
1189 f"{matched.departure_icao} → {matched.arrival_icao}"
1190 )
1191 seg["matched_has_existing_track"] = matched.gps_track_id is not None
1192 seg["linked_pilot_entries"] = _linked_pilot_entries(
1193 matched.id, int(session["user_id"])
1194 )
1195 else:
1196 seg["matched_flight_id"] = None
1197 seg["matched_flight_str"] = None
1198 seg["matched_has_existing_track"] = False
1199 seg["linked_pilot_entries"] = []
1201 tmp_dir = _gps_tmp_dir()
1202 session["gps_import"]["segments"] = [
1203 _segment_for_session(s, tmp_dir) for s in full_segs
1204 ]
1205 session.modified = True
1207 # Get OpenAIP API key for map tiles
1208 tile_setting = db.session.get(AppSetting, "openaip_api_key")
1209 openaip_key = tile_setting.value if tile_setting and tile_setting.value else None
1211 return render_template(
1212 "aircraft/gps_import_review.html",
1213 aircraft=ac,
1214 segments=full_segs,
1215 skipped_empty=state.get("skipped_empty", 0),
1216 openaip_key=openaip_key,
1217 other_aircraft=state.get("other_aircraft", False),
1218 other_ac_make_model=state.get("other_ac_make_model", ""),
1219 other_ac_reg=state.get("other_ac_reg", ""),
1220 confirmed_segments=state.get("confirmed_segments", {}),
1221 )
1224def _gps_import_create_segment(
1225 ac: Any,
1226 aircraft_id: int,
1227 seg: dict[str, Any],
1228 seg_idx: int,
1229 pilot_role: str,
1230 dep_icao: str,
1231 arr_icao: str,
1232 nature: str | None,
1233 remarks: str | None,
1234 batch: Any,
1235 file_metas: list[dict[str, Any]],
1236 linked_ids: list[int],
1237 pilot_display_name: str,
1238 other_aircraft: bool,
1239 other_ac_make_model: str,
1240 other_ac_reg: str,
1241 batch_device_id: str | None,
1242) -> tuple[Any, list[int]]:
1243 """Create FlightEntry + optionally PilotLogbookEntry for one GPS segment.
1245 Returns (entry_or_None, updated_linked_ids).
1246 """
1247 import decimal as _dec # noqa: PLC0415
1248 from datetime import datetime as _dt # noqa: PLC0415
1250 create_pilot_entries = pilot_role in ("pic", "dual")
1252 block_off = _dt.fromisoformat(seg["block_off_utc"])
1253 block_on = _dt.fromisoformat(seg["block_on_utc"])
1254 dep_time = block_off.time().replace(tzinfo=None)
1255 arr_time = block_on.time().replace(tzinfo=None)
1256 flight_time_h = round_flight_time(
1257 seg["flight_time_raw_h"], ac.logbook_time_precision
1258 )
1260 entry = None
1261 gps_track: GpsTrack | None = None
1263 if not other_aircraft:
1264 matched_id = seg.get("matched_flight_id")
1265 if matched_id:
1266 existing = db.session.get(FlightEntry, matched_id)
1267 if existing and existing.aircraft_id == aircraft_id:
1268 existing.block_off_utc = block_off
1269 existing.block_on_utc = block_on
1270 gps_track = GpsTrack(
1271 source_filename=file_metas[0]["original_filename"]
1272 if len(file_metas) == 1
1273 else None,
1274 device_id=batch_device_id,
1275 block_off_utc=block_off,
1276 block_on_utc=block_on,
1277 departure_icao=dep_icao,
1278 arrival_icao=arr_icao,
1279 geojson=_load_segment_geojson(seg),
1280 )
1281 db.session.add(gps_track)
1282 db.session.flush()
1283 existing.gps_track_id = gps_track.id
1284 linked_ids.append(existing.id)
1285 # Link track to other users' pilot logbook entries for this flight
1286 # (only when they have no existing GPS track — preserve their own data).
1287 current_uid = int(session["user_id"])
1288 for ple in PilotLogbookEntry.query.filter(
1289 PilotLogbookEntry.flight_id == existing.id,
1290 PilotLogbookEntry.pilot_user_id != current_uid,
1291 PilotLogbookEntry.gps_track_id.is_(None),
1292 ).all():
1293 ple.gps_track_id = gps_track.id
1294 db.session.flush()
1295 entry = existing
1296 else:
1297 matched_id = None
1299 if not matched_id:
1300 gps_track = GpsTrack(
1301 source_filename=file_metas[0]["original_filename"]
1302 if len(file_metas) == 1
1303 else None,
1304 block_off_utc=block_off,
1305 block_on_utc=block_on,
1306 departure_icao=dep_icao,
1307 arrival_icao=arr_icao,
1308 geojson=_load_segment_geojson(seg),
1309 )
1310 db.session.add(gps_track)
1311 db.session.flush()
1312 entry = FlightEntry(
1313 aircraft_id=aircraft_id,
1314 date=block_off.date(),
1315 departure_icao=dep_icao,
1316 arrival_icao=arr_icao,
1317 departure_time=dep_time,
1318 arrival_time=arr_time,
1319 flight_time=_dec.Decimal(str(flight_time_h)),
1320 landing_count=seg.get("landing_count") or 0,
1321 nature_of_flight=nature,
1322 source="gps_import",
1323 gps_import_batch_id=batch.id,
1324 block_off_utc=block_off,
1325 block_on_utc=block_on,
1326 gps_track_id=gps_track.id,
1327 )
1328 db.session.add(entry)
1329 db.session.flush()
1331 if create_pilot_entries:
1332 if other_aircraft:
1333 ac_type = other_ac_make_model or None
1334 ac_reg = other_ac_reg or None
1335 single_pilot_se: _dec.Decimal | None = _dec.Decimal(str(flight_time_h))
1336 single_pilot_me: _dec.Decimal | None = None
1337 flight_id_for_entry = None
1338 if gps_track is None:
1339 gps_track = GpsTrack(
1340 source_filename=file_metas[0]["original_filename"]
1341 if len(file_metas) == 1
1342 else None,
1343 device_id=batch_device_id,
1344 block_off_utc=block_off,
1345 block_on_utc=block_on,
1346 departure_icao=dep_icao,
1347 arrival_icao=arr_icao,
1348 geojson=_load_segment_geojson(seg),
1349 )
1350 db.session.add(gps_track)
1351 db.session.flush()
1352 else:
1353 ac_type = f"{ac.make} {ac.model}".strip()
1354 ac_reg = ac.registration
1355 ac_category = getattr(ac, "category", "SEP")
1356 single_pilot_se = (
1357 _dec.Decimal(str(flight_time_h))
1358 if ac_category in ("SEP", "SET", "")
1359 else None
1360 )
1361 single_pilot_me = (
1362 _dec.Decimal(str(flight_time_h))
1363 if ac_category in ("MEP", "MET")
1364 else None
1365 )
1366 flight_id_for_entry = entry.id if entry else None
1368 pentry = PilotLogbookEntry(
1369 pilot_user_id=int(session["user_id"]),
1370 flight_id=flight_id_for_entry,
1371 date=block_off.date(),
1372 aircraft_type=ac_type,
1373 aircraft_registration=ac_reg,
1374 departure_place=dep_icao,
1375 departure_time=dep_time,
1376 arrival_place=arr_icao,
1377 arrival_time=arr_time,
1378 pic_name=pilot_display_name,
1379 single_pilot_se=single_pilot_se,
1380 single_pilot_me=single_pilot_me,
1381 function_pic=_dec.Decimal(str(flight_time_h))
1382 if pilot_role == "pic"
1383 else None,
1384 function_dual=_dec.Decimal(str(flight_time_h))
1385 if pilot_role == "dual"
1386 else None,
1387 landings_day=seg.get("landing_count") or 0,
1388 remarks=remarks,
1389 source="gps_import",
1390 gps_batch_id=batch.id,
1391 gps_track_id=gps_track.id if gps_track else None,
1392 )
1393 db.session.add(pentry)
1395 return entry, linked_ids
1398def _gps_cleanup(state: dict[str, Any]) -> None:
1399 """Delete tmp GPS and GeoJSON files from a gps_import session state."""
1400 for meta in state.get("files", []):
1401 try:
1402 os.unlink(meta["tmp_path"])
1403 except OSError as exc:
1404 current_app.logger.debug("cleanup GPS tmp file: %s", exc)
1405 for seg in state.get("segments", []):
1406 gj_path = seg.get("geojson_path")
1407 if gj_path:
1408 try:
1409 os.unlink(gj_path)
1410 except OSError as exc:
1411 current_app.logger.debug("cleanup GPS geojson tmp: %s", exc)
1414@aircraft_bp.route("/<int:aircraft_id>/gps-import/confirm-one", methods=["POST"])
1415@login_required
1416@require_role(*_PILOT_ROLES)
1417def gps_import_confirm_one(aircraft_id: int) -> ResponseReturnValue:
1418 """Confirm a single GPS segment as-is and redirect back to the review page."""
1419 ac = _get_aircraft_or_404(aircraft_id)
1420 state = session.get("gps_import")
1421 if not state or state.get("aircraft_id") != aircraft_id:
1422 flash(_("Session expired — please upload your GPS files again."), "warning")
1423 return redirect(url_for("aircraft.gps_import_upload", aircraft_id=aircraft_id))
1425 segments_data: list[dict[str, Any]] = state.get("segments", [])
1426 if not segments_data:
1427 flash(_("No segments to import."), "warning")
1428 return redirect(url_for("aircraft.gps_import_upload", aircraft_id=aircraft_id))
1430 try:
1431 seg_idx = int(request.form.get("seg_idx", ""))
1432 except (ValueError, TypeError):
1433 flash(_("Invalid segment index."), "danger")
1434 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
1436 if seg_idx < 0 or seg_idx >= len(segments_data):
1437 flash(_("Invalid segment index."), "danger")
1438 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
1440 confirmed = state.get("confirmed_segments", {})
1441 if str(seg_idx) in confirmed:
1442 flash(_("This segment has already been confirmed."), "info")
1443 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
1445 pilot_role = request.form.get("pilot_role", "none")
1447 if request.form.get("skip") == "1":
1448 confirmed[str(seg_idx)] = "skip"
1449 state["confirmed_segments"] = confirmed
1450 session["gps_import"] = state
1451 session.modified = True
1453 all_handled = len(confirmed) == len(segments_data)
1454 if all_handled:
1455 _gps_cleanup(state)
1456 session.pop("gps_import", None)
1457 imported = sum(1 for v in confirmed.values() if v != "skip")
1458 skipped_count = len(segments_data) - imported
1459 if imported > 0:
1460 flash(
1461 ngettext(
1462 "%(n)s flight imported successfully.",
1463 "%(n)s flights imported successfully.",
1464 imported,
1465 n=imported,
1466 ),
1467 "success",
1468 )
1469 flash(
1470 ngettext(
1471 "%(n)s segment skipped.",
1472 "%(n)s segments skipped.",
1473 skipped_count,
1474 n=skipped_count,
1475 ),
1476 "info",
1477 )
1478 if pilot_role not in ("pic", "dual", "none"):
1479 pilot_role = "none"
1480 if imported > 0 and pilot_role in ("pic", "dual"):
1481 return redirect(url_for("pilots.logbook"))
1482 return redirect(url_for("flights.list_flights", aircraft_id=aircraft_id))
1484 flash(_("Segment skipped."), "info")
1485 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
1487 seg = segments_data[seg_idx]
1489 other_aircraft: bool = state.get("other_aircraft", False)
1490 other_ac_make_model: str = state.get("other_ac_make_model", "")
1491 other_ac_reg: str = state.get("other_ac_reg", "")
1493 if other_aircraft and pilot_role == "none":
1494 pilot_role = "pic"
1496 _pilot_user = db.session.get(User, int(session["user_id"]))
1497 pilot_display_name = _pilot_user.display_name if _pilot_user else ""
1498 file_metas = state["files"]
1500 dep_icao = (
1501 (request.form.get("dep_icao") or seg["departure_icao"] or "")
1502 .strip()
1503 .upper()[:4]
1504 )
1505 arr_icao = (
1506 (request.form.get("arr_icao") or seg["arrival_icao"] or "").strip().upper()[:4]
1507 )
1508 if not dep_icao:
1509 dep_icao = "????"
1510 if not arr_icao:
1511 arr_icao = "????"
1513 nature = (request.form.get("nature") or "").strip()[:100] or None
1514 remarks = (request.form.get("remarks") or "").strip() or None
1516 batch_device_id: str | None = next(
1517 (m.get("device_id") for m in file_metas if m.get("device_id")), None
1518 )
1520 # Get or create the shared batch record for this session
1521 batch_id = state.get("batch_id")
1522 batch: AircraftGpsImportBatch | None = (
1523 db.session.get(AircraftGpsImportBatch, batch_id) if batch_id else None
1524 )
1525 if batch is None:
1526 formats = {m["format"] for m in file_metas}
1527 format_label = formats.pop() if len(formats) == 1 else "mixed"
1528 batch = AircraftGpsImportBatch(
1529 aircraft_id=aircraft_id,
1530 pilot_user_id=int(session["user_id"]) if pilot_role != "none" else None,
1531 source_filenames=[m["original_filename"] for m in file_metas],
1532 format_detected=format_label,
1533 segments_found=len(segments_data),
1534 linked_flight_entry_ids=[],
1535 pilot_role=pilot_role,
1536 other_aircraft_make_model=other_ac_make_model or None,
1537 other_aircraft_registration=other_ac_reg or None,
1538 )
1539 db.session.add(batch)
1540 db.session.flush()
1541 state["batch_id"] = batch.id
1543 linked_ids: list[int] = list(batch.linked_flight_entry_ids or [])
1545 entry, linked_ids = _gps_import_create_segment(
1546 ac=ac,
1547 aircraft_id=aircraft_id,
1548 seg=seg,
1549 seg_idx=seg_idx,
1550 pilot_role=pilot_role,
1551 dep_icao=dep_icao,
1552 arr_icao=arr_icao,
1553 nature=nature,
1554 remarks=remarks,
1555 batch=batch,
1556 file_metas=file_metas,
1557 linked_ids=linked_ids,
1558 pilot_display_name=pilot_display_name,
1559 other_aircraft=other_aircraft,
1560 other_ac_make_model=other_ac_make_model,
1561 other_ac_reg=other_ac_reg,
1562 batch_device_id=batch_device_id,
1563 )
1565 batch.linked_flight_entry_ids = linked_ids
1566 batch.segments_imported = (batch.segments_imported or 0) + 1
1567 db.session.commit()
1569 confirmed[str(seg_idx)] = entry.id if entry else 0
1570 state["confirmed_segments"] = confirmed
1571 session["gps_import"] = state
1572 session.modified = True
1574 all_confirmed = len(confirmed) == len(segments_data)
1576 if all_confirmed:
1577 _gps_cleanup(state)
1578 session.pop("gps_import", None)
1579 total = len(confirmed)
1580 flash(
1581 ngettext(
1582 "%(n)s flight imported successfully.",
1583 "%(n)s flights imported successfully.",
1584 total,
1585 n=total,
1586 ),
1587 "success",
1588 )
1589 if pilot_role in ("pic", "dual"):
1590 return redirect(url_for("pilots.logbook"))
1591 return redirect(url_for("flights.list_flights", aircraft_id=aircraft_id))
1593 flash(_("Flight confirmed."), "success")
1594 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
1597@aircraft_bp.route(
1598 "/<int:aircraft_id>/gps-import/prefill-segment/<int:seg_idx>", methods=["GET"]
1599)
1600@login_required
1601@require_role(*_PILOT_ROLES)
1602def gps_import_prefill_segment(aircraft_id: int, seg_idx: int) -> ResponseReturnValue:
1603 """Store a batch segment as gps_prefill then redirect to /flights/new."""
1604 import json as _json # noqa: PLC0415
1605 from datetime import datetime as _dt # noqa: PLC0415
1607 _get_aircraft_or_404(aircraft_id)
1608 state = session.get("gps_import")
1609 if not state or state.get("aircraft_id") != aircraft_id:
1610 flash(_("Session expired — please upload your GPS files again."), "warning")
1611 return redirect(url_for("aircraft.gps_import_upload", aircraft_id=aircraft_id))
1613 segments_data: list[dict[str, Any]] = state.get("segments", [])
1614 if seg_idx < 0 or seg_idx >= len(segments_data):
1615 flash(_("Invalid segment index."), "danger")
1616 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
1618 seg = segments_data[seg_idx]
1619 block_off = _dt.fromisoformat(seg["block_off_utc"])
1620 block_on = _dt.fromisoformat(seg["block_on_utc"])
1621 file_metas = state.get("files", [])
1622 single_filename = file_metas[0]["original_filename"] if len(file_metas) == 1 else ""
1623 geojson_data = _load_segment_geojson(seg)
1624 geojson_str = _json.dumps(geojson_data) if geojson_data else ""
1626 session["gps_prefill"] = {
1627 "filename": single_filename,
1628 "date": block_off.date().isoformat(),
1629 "departure_icao": seg.get("departure_icao") or "",
1630 "arrival_icao": seg.get("arrival_icao") or "",
1631 "departure_time": block_off.strftime("%H:%M"),
1632 "arrival_time": block_on.strftime("%H:%M"),
1633 "flight_time_h": str(seg.get("flight_time_rounded_h") or 0),
1634 "block_off_utc": block_off.isoformat(),
1635 "block_on_utc": block_on.isoformat(),
1636 "geojson": geojson_str,
1637 "landing_count": seg.get("landing_count") or 0,
1638 }
1639 session.modified = True
1641 return redirect(
1642 url_for(
1643 "flights.log_flight",
1644 aircraft_id=aircraft_id,
1645 gps_review_return=aircraft_id,
1646 gps_seg=seg_idx,
1647 )
1648 )
1651@aircraft_bp.route("/<int:aircraft_id>/gps-import/history", methods=["GET"])
1652@login_required
1653@require_role(*_PILOT_ROLES)
1654def gps_import_history(aircraft_id: int) -> ResponseReturnValue:
1655 ac = _get_aircraft_or_404(aircraft_id)
1656 batches = (
1657 AircraftGpsImportBatch.query.filter_by(aircraft_id=aircraft_id)
1658 .order_by(AircraftGpsImportBatch.imported_at.desc())
1659 .all()
1660 )
1661 return render_template(
1662 "aircraft/gps_import_history.html", aircraft=ac, batches=batches
1663 )
1666@aircraft_bp.route(
1667 "/<int:aircraft_id>/gps-import/<int:batch_id>/rollback", methods=["POST"]
1668)
1669@login_required
1670@require_role(*_OWNER_ROLES)
1671def gps_import_rollback(aircraft_id: int, batch_id: int) -> ResponseReturnValue:
1672 _get_aircraft_or_404(aircraft_id)
1673 batch = db.session.get(AircraftGpsImportBatch, batch_id)
1674 if not batch or batch.aircraft_id != aircraft_id:
1675 abort(404)
1677 # Delete pilot logbook entries created by this batch.
1678 PilotLogbookEntry.query.filter_by(gps_batch_id=batch.id).delete(
1679 synchronize_session="fetch"
1680 )
1682 # Flights created by this batch — delete them entirely.
1683 FlightEntry.query.filter_by(gps_import_batch_id=batch.id).delete(
1684 synchronize_session="fetch"
1685 )
1687 # Flights that were pre-existing but got a GPS track linked — unlink only.
1688 linked_ids = batch.linked_flight_entry_ids or []
1689 if linked_ids:
1690 FlightEntry.query.filter(FlightEntry.id.in_(linked_ids)).update(
1691 {
1692 "gps_track_id": None,
1693 "block_off_utc": None,
1694 "block_on_utc": None,
1695 },
1696 synchronize_session="fetch",
1697 )
1699 db.session.delete(batch)
1700 db.session.commit()
1701 flash(
1702 _("GPS import batch rolled back and all linked flight entries removed."),
1703 "success",
1704 )
1705 return redirect(url_for("aircraft.gps_import_history", aircraft_id=aircraft_id))
1708@aircraft_bp.route("/<int:aircraft_id>/flights/<int:flight_id>", methods=["GET"])
1709@login_required
1710@require_role(*_PILOT_ROLES)
1711def flight_detail(aircraft_id: int, flight_id: int) -> ResponseReturnValue:
1712 ac = _get_aircraft_or_404(aircraft_id)
1713 entry = db.session.get(FlightEntry, flight_id)
1714 if not entry or entry.aircraft_id != aircraft_id:
1715 abort(404)
1717 uid = int(session["user_id"])
1718 pilot_entry = PilotLogbookEntry.query.filter_by(
1719 flight_id=flight_id, pilot_user_id=uid
1720 ).first()
1722 tile_setting = db.session.get(AppSetting, "openaip_api_key")
1723 openaip_key = tile_setting.value if tile_setting and tile_setting.value else None
1725 return render_template(
1726 "aircraft/flight_detail.html",
1727 aircraft=ac,
1728 entry=entry,
1729 pilot_entry=pilot_entry,
1730 openaip_key=openaip_key,
1731 )
1734@aircraft_bp.route("/<int:aircraft_id>/tracks", methods=["GET"])
1735@login_required
1736@require_role(*_PILOT_ROLES)
1737def flight_tracks(aircraft_id: int) -> ResponseReturnValue:
1738 from flask import url_for as _url_for
1740 ac = _get_aircraft_or_404(aircraft_id)
1741 entries_with_tracks = (
1742 FlightEntry.query.filter_by(aircraft_id=aircraft_id)
1743 .filter(FlightEntry.gps_track_id.isnot(None))
1744 .order_by(FlightEntry.date.asc())
1745 .all()
1746 )
1747 track_rows = [
1748 {
1749 "date": str(e.date),
1750 "dep": e.departure_icao,
1751 "arr": e.arrival_icao,
1752 "time_str": f"{e.flight_time} h" if e.flight_time is not None else "",
1753 "view_url": _url_for(
1754 "aircraft.flight_detail",
1755 aircraft_id=aircraft_id,
1756 flight_id=e.id,
1757 ),
1758 "geojson": e.gps_track.geojson if e.gps_track else None,
1759 }
1760 for e in entries_with_tracks
1761 ]
1763 tile_setting = db.session.get(AppSetting, "openaip_api_key")
1764 openaip_key = tile_setting.value if tile_setting and tile_setting.value else None
1766 return render_template(
1767 "aircraft/flight_tracks.html",
1768 aircraft=ac,
1769 track_rows=track_rows,
1770 openaip_key=openaip_key,
1771 )
1774@aircraft_bp.route("/<int:aircraft_id>/tracks/animation.gif")
1775@login_required
1776@require_role(*_PILOT_ROLES)
1777def flight_tracks_gif(aircraft_id: int) -> ResponseReturnValue:
1778 from utils import generate_tracks_gif, sort_tracks_oldest_first # pyright: ignore[reportMissingImports]
1780 ac = _get_aircraft_or_404(aircraft_id)
1781 entries = (
1782 FlightEntry.query.filter_by(aircraft_id=aircraft_id)
1783 .filter(FlightEntry.gps_track_id.isnot(None))
1784 .all()
1785 )
1786 track_rows = sort_tracks_oldest_first(
1787 [
1788 {
1789 "date": str(e.date),
1790 "dep": e.departure_icao or "",
1791 "arr": e.arrival_icao or "",
1792 "geojson": e.gps_track.geojson if e.gps_track else None,
1793 }
1794 for e in entries
1795 ]
1796 )
1797 tile_s = db.session.get(AppSetting, "openaip_api_key")
1798 portrait = request.args.get("orientation") == "portrait"
1799 hires = request.args.get("quality") == "hires"
1800 base_w, base_h = (480, 800) if portrait else (800, 480)
1801 mul = 2 if hires else 1
1802 canvas_w, canvas_h = base_w * mul, base_h * mul
1803 gif_bytes = generate_tracks_gif(
1804 track_rows,
1805 _openaip_key=tile_s.value if tile_s and tile_s.value else None,
1806 canvas_w=canvas_w,
1807 canvas_h=canvas_h,
1808 high_res=hires,
1809 )
1810 orient_sfx = "-portrait" if portrait else ""
1811 qual_sfx = "-hires" if hires else ""
1812 suffix = orient_sfx + qual_sfx
1813 filename = f"{ac.registration.lower().replace('-', '')}_tracks{suffix}.gif"
1814 from flask import Response # pyright: ignore[reportMissingImports]
1816 return Response(
1817 gif_bytes,
1818 mimetype="image/gif",
1819 headers={"Content-Disposition": f'attachment; filename="{filename}"'},
1820 )
1823# ── Aircraft photos ───────────────────────────────────────────────────────────
1825_PHOTO_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".heic"}
1828def _photo_folder(app: Any, tenant_slug: str, safe_reg: str) -> str:
1829 folder = app.config.get("UPLOAD_FOLDER", "/data/uploads")
1830 return os.path.join(folder, tenant_slug, safe_reg, "photos")
1833def _save_photo_file(
1834 file: Any,
1835 tenant_slug: str,
1836 safe_reg: str,
1837 sort_order: int,
1838) -> tuple[str, str]:
1839 """Save photo to canonical path; return (relpath, original_filename)."""
1841 original = secure_filename(file.filename or "photo.jpg")
1842 ext = os.path.splitext(original)[1].lower() or ".jpg"
1843 short_id = _uuid_mod.uuid4().hex[:6]
1844 fname = f"{sort_order:02d}-{short_id}{ext}"
1845 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
1846 dest_dir = os.path.join(folder, tenant_slug, safe_reg, "photos")
1847 os.makedirs(dest_dir, exist_ok=True)
1848 file.save(os.path.join(dest_dir, fname))
1849 relpath = os.path.join(tenant_slug, safe_reg, "photos", fname).replace("\\", "/")
1850 return relpath, original
1853def _trash_photo_file(filename: str) -> None:
1854 """Move photo file to _trash/ (same pattern as document deletion)."""
1855 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
1856 src = os.path.join(folder, filename)
1857 if not os.path.exists(src):
1858 return
1859 try:
1860 trash_dir = os.path.join(folder, "_trash")
1861 os.makedirs(trash_dir, exist_ok=True)
1862 base = os.path.basename(filename)
1863 dest = os.path.join(trash_dir, base)
1864 if os.path.exists(dest):
1865 stem, ext = os.path.splitext(base)
1866 dest = os.path.join(trash_dir, f"{stem}_{_uuid_mod.uuid4().hex[:6]}{ext}")
1867 os.rename(src, dest)
1868 except OSError:
1869 current_app.logger.debug("Could not trash photo: %s", filename)
1872def _renumber_photos(photos: list[Any], tenant_slug: str, safe_reg: str) -> None:
1873 """Assign sort_order 1..N, renaming files on disk to keep the numeric prefix."""
1874 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
1875 for new_order, photo in enumerate(photos, start=1):
1876 if photo.sort_order != new_order:
1877 old_full = os.path.join(folder, photo.filename)
1878 old_fname = os.path.basename(photo.filename)
1879 suffix = old_fname[
1880 2:
1881 ] # strip old "NN" prefix (e.g. "01-abc.jpg" → "-abc.jpg")
1882 new_fname = f"{new_order:02d}{suffix}"
1883 new_full = os.path.join(folder, tenant_slug, safe_reg, "photos", new_fname)
1884 if os.path.exists(old_full):
1885 try:
1886 os.rename(old_full, new_full)
1887 except OSError:
1888 current_app.logger.debug(
1889 "Could not renumber photo: %s", photo.filename
1890 )
1891 new_fname = old_fname # keep old name if rename fails
1892 new_rel = f"{tenant_slug}/{safe_reg}/photos/{new_fname}"
1893 photo.filename = new_rel
1894 photo.sort_order = new_order
1897@aircraft_bp.route("/<int:aircraft_id>/photos/upload", methods=["POST"])
1898@login_required
1899@require_role(*_OWNER_ROLES)
1900def upload_photo(aircraft_id: int) -> ResponseReturnValue:
1901 from documents.routes import _ensure_tenant_slug, _get_tenant # pyright: ignore[reportMissingImports]
1903 ac = _get_aircraft_or_404(aircraft_id)
1904 tenant = _get_tenant()
1906 files = request.files.getlist("photos")
1907 if not files or all(f.filename == "" for f in files):
1908 flash(_("No files selected."), "warning")
1909 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
1911 tenant_slug = _ensure_tenant_slug(tenant)
1912 safe_reg = ac.registration.replace("/", "-").replace(" ", "-").upper()
1913 next_order = (
1914 db.session.query(db.func.max(AircraftPhoto.sort_order))
1915 .filter_by(aircraft_id=ac.id)
1916 .scalar()
1917 or 0
1918 ) + 1
1920 uploaded = 0
1921 for f in files:
1922 if not f.filename:
1923 continue
1924 ext = os.path.splitext(secure_filename(f.filename))[1].lower()
1925 if ext not in _PHOTO_EXTS:
1926 flash(
1927 _(
1928 "%(name)s: unsupported format (use JPEG, PNG, WEBP or HEIC).",
1929 name=f.filename,
1930 ),
1931 "warning",
1932 )
1933 continue
1934 relpath, original = _save_photo_file(f, tenant_slug, safe_reg, next_order)
1935 photo = AircraftPhoto(
1936 aircraft_id=ac.id,
1937 filename=relpath,
1938 original_filename=original,
1939 sort_order=next_order,
1940 uploaded_by_user_id=session.get("user_id"),
1941 )
1942 db.session.add(photo)
1943 next_order += 1
1944 uploaded += 1
1946 if uploaded:
1947 db.session.commit()
1948 flash(
1949 ngettext(
1950 "%(n)s photo uploaded.", "%(n)s photos uploaded.", uploaded, n=uploaded
1951 ),
1952 "success",
1953 )
1954 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
1957@aircraft_bp.route("/<int:aircraft_id>/photos/<int:photo_id>/img")
1958@login_required
1959def serve_photo(aircraft_id: int, photo_id: int) -> ResponseReturnValue:
1960 from flask import send_from_directory # pyright: ignore[reportMissingImports]
1962 ac = _get_aircraft_or_404(aircraft_id)
1963 photo = db.session.get(AircraftPhoto, photo_id)
1964 if not photo or photo.aircraft_id != ac.id:
1965 abort(404)
1966 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
1967 directory = os.path.join(folder, os.path.dirname(photo.filename))
1968 fname = os.path.basename(photo.filename)
1969 return send_from_directory(directory, fname)
1972@aircraft_bp.route("/<int:aircraft_id>/photos/<int:photo_id>/delete", methods=["POST"])
1973@login_required
1974@require_role(*_OWNER_ROLES)
1975def delete_photo(aircraft_id: int, photo_id: int) -> ResponseReturnValue:
1976 from documents.routes import _get_tenant # pyright: ignore[reportMissingImports]
1978 ac = _get_aircraft_or_404(aircraft_id)
1979 photo = db.session.get(AircraftPhoto, photo_id)
1980 if not photo or photo.aircraft_id != ac.id:
1981 abort(404)
1983 _trash_photo_file(photo.filename)
1984 db.session.delete(photo)
1985 db.session.flush()
1987 # Renumber remaining photos
1988 remaining = (
1989 AircraftPhoto.query.filter_by(aircraft_id=ac.id)
1990 .order_by(AircraftPhoto.sort_order)
1991 .all()
1992 )
1993 tenant = _get_tenant()
1994 tenant_slug = tenant.slug or ""
1995 safe_reg = ac.registration.replace("/", "-").replace(" ", "-").upper()
1996 _renumber_photos(remaining, tenant_slug, safe_reg)
1997 db.session.commit()
1998 flash(_("Photo deleted."), "success")
1999 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
2002@aircraft_bp.route("/<int:aircraft_id>/photos/reorder", methods=["POST"])
2003@login_required
2004@require_role(*_OWNER_ROLES)
2005def reorder_photos(aircraft_id: int) -> ResponseReturnValue:
2006 from documents.routes import _get_tenant # pyright: ignore[reportMissingImports]
2008 ac = _get_aircraft_or_404(aircraft_id)
2010 ordered_ids: list[int] = []
2011 try:
2012 ordered_ids = [int(i) for i in request.form.getlist("photo_order[]")]
2013 except (ValueError, TypeError):
2014 abort(400)
2016 photos_by_id = {
2017 p.id: p for p in AircraftPhoto.query.filter_by(aircraft_id=ac.id).all()
2018 }
2019 if set(ordered_ids) != set(photos_by_id):
2020 abort(400)
2022 ordered_photos = [photos_by_id[pid] for pid in ordered_ids]
2023 tenant = _get_tenant()
2024 tenant_slug = tenant.slug or ""
2025 safe_reg = ac.registration.replace("/", "-").replace(" ", "-").upper()
2026 _renumber_photos(ordered_photos, tenant_slug, safe_reg)
2027 db.session.commit()
2028 return "", 204