Coverage for app/aircraft/routes.py: 100%
1450 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-31 10:13 +0000
1import json
2import math
3import os
4import uuid as _uuid_mod
5from datetime import UTC
6from typing import Any, cast
8from flask import ( # pyright: ignore[reportMissingImports]
9 Blueprint,
10 abort,
11 current_app,
12 flash,
13 redirect,
14 render_template,
15 request,
16 session,
17 url_for,
18)
19from flask.typing import ResponseReturnValue # pyright: ignore[reportMissingImports]
20from flask_babel import gettext as _ # pyright: ignore[reportMissingImports]
21from flask_babel import ngettext
22from models import (
23 FUEL_DENSITY,
24 GAL_TO_L,
25 Aircraft,
26 AircraftFuelTank,
27 AircraftGpsImportBatch,
28 AircraftOwner,
29 AircraftPhoto,
30 AirworthinessDocumentStatus,
31 AppSetting,
32 Component,
33 ComponentType,
34 DocType,
35 Document,
36 Expense,
37 ExpenseType,
38 Flight,
39 GpsTrack,
40 MaintenanceTrigger,
41 OperatingModel,
42 Refuel,
43 Reservation,
44 ReservationStatus,
45 Role,
46 Snag,
47 TenantProfile,
48 TenantUser,
49 User,
50 WeightBalanceConfig,
51 WeightBalanceEntry,
52 WeightBalanceStation,
53 db,
54) # pyright: ignore[reportMissingImports]
55from utils import (
56 accessible_aircraft,
57 activity,
58 compute_aircraft_statuses,
59 get_aircraft_type_engine_info,
60 login_required,
61 require_role,
62 user_can_access_aircraft,
63) # pyright: ignore[reportMissingImports]
64from werkzeug.utils import secure_filename # pyright: ignore[reportMissingImports]
66from aircraft.co_owner_form_parsing import ( # pyright: ignore[reportMissingImports]
67 parse_owners_form,
68 parse_reserve_fields,
69)
70from aircraft.gps_import import ( # pyright: ignore[reportMissingImports]
71 detect_segments,
72 merge_and_sort,
73 parse_gps_file,
74 round_flight_time,
75)
77aircraft_bp = Blueprint("aircraft", __name__, url_prefix="/aircraft")
79_OWNER_ROLES = (Role.ADMIN, Role.OWNER)
80_PILOT_ROLES = (Role.ADMIN, Role.OWNER, Role.PILOT)
83def _tenant_id() -> int:
84 """Return the tenant ID for the currently logged-in user."""
85 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
86 if not tu:
87 abort(403)
88 return int(tu.tenant_id)
91def _get_aircraft_or_404(aircraft_id: int) -> Aircraft:
92 """Fetch an aircraft that belongs to the current tenant and is accessible to the user."""
93 ac = db.session.get(Aircraft, aircraft_id)
94 if (
95 not ac
96 or ac.tenant_id != _tenant_id()
97 or not user_can_access_aircraft(aircraft_id)
98 ):
99 abort(404)
100 return ac
103def _get_tenant_aircraft_or_404(aircraft_id: int) -> Aircraft:
104 """Like _get_aircraft_or_404 but without the operational
105 user_can_access_aircraft gate — for self-service co-owner routes
106 (my_share/my_share_statement_csv) where the AircraftOwner row checked
107 right after this call is the real authorization boundary. A co-owner may
108 have no operational role on the aircraft at all (e.g. a silent investor
109 with no UserAircraftAccess grant) and must still be able to view their
110 own share."""
111 ac = db.session.get(Aircraft, aircraft_id)
112 if not ac or ac.tenant_id != _tenant_id():
113 abort(404)
114 return ac
117def _shared_ownership_enabled(tenant_id: int, aircraft_id: int) -> bool:
118 """True when the tenant runs the shared_ownership operating model, OR
119 when this aircraft already has AircraftOwner rows (legacy data after a
120 model switch — the pages must stay reachable so an admin can view
121 accounts and clear the owner set; without rows they 404, keeping zero
122 trace on instances that never used the feature)."""
123 profile = TenantProfile.query.filter_by(tenant_id=tenant_id).first()
124 if profile and profile.operating_model == OperatingModel.SHARED_OWNERSHIP:
125 return True
126 return bool(
127 db.session.query(
128 AircraftOwner.query.filter_by(aircraft_id=aircraft_id).exists()
129 ).scalar()
130 )
133def _registration_taken(
134 registration: str, tenant_id: int, exclude_aircraft_id: int | None = None
135) -> bool:
136 """True if another aircraft in this tenant already has this registration —
137 compared the same way AircraftRefConverter (utils.py) sanitizes for its
138 URL slot ('/' and ' ' -> '-', case-insensitive), so two registrations
139 that only differ that way don't end up silently sharing a URL."""
140 needle = registration.replace("/", "-").replace(" ", "-").upper()
141 query = Aircraft.query.filter(
142 Aircraft.tenant_id == tenant_id,
143 db.func.upper(
144 db.func.replace(db.func.replace(Aircraft.registration, "/", "-"), " ", "-")
145 )
146 == needle,
147 )
148 if exclude_aircraft_id is not None:
149 query = query.filter(Aircraft.id != exclude_aircraft_id)
150 return query.first() is not None
153def _get_component_or_404(aircraft: Aircraft, component_id: int) -> Component:
154 comp = db.session.get(Component, component_id)
155 if not comp or comp.aircraft_id != aircraft.id:
156 abort(404)
157 return comp
160# ── Aircraft list ─────────────────────────────────────────────────────────────
163@aircraft_bp.route("/")
164@login_required
165def list_aircraft() -> ResponseReturnValue:
166 from models import TenantProfile
168 if not request.args.get("list"):
169 tp = TenantProfile.query.filter_by(tenant_id=_tenant_id()).first()
170 if tp and tp.planned_aircraft_count == 1:
171 aircraft_list = accessible_aircraft(_tenant_id()).all()
172 if len(aircraft_list) == 1:
173 return redirect(
174 url_for("aircraft.detail", aircraft_id=aircraft_list[0].id)
175 )
177 show_archived = request.args.get("archived") == "1"
178 all_accessible = accessible_aircraft(_tenant_id(), include_archived=True).all()
179 archived_count = sum(1 for ac in all_accessible if ac.is_archived)
180 aircraft = (
181 all_accessible
182 if show_archived
183 else [ac for ac in all_accessible if not ac.is_archived]
184 )
185 aircraft_ids = [ac.id for ac in aircraft]
186 hobbs_by_id = Aircraft.engine_hours_by_id(aircraft_ids)
187 landings_by_id = Aircraft.landings_by_id(aircraft_ids)
188 flight_hours_by_id = Aircraft.flight_hours_by_id(aircraft_ids)
189 triggers = (
190 (
191 MaintenanceTrigger.query.filter(
192 MaintenanceTrigger.aircraft_id.in_(aircraft_ids)
193 ).all()
194 )
195 if aircraft_ids
196 else []
197 )
198 aircraft_status = compute_aircraft_statuses(
199 aircraft, triggers, hobbs_by_id, landings_by_id, flight_hours_by_id
200 )
201 wb_configured_ids = {ac.id for ac in aircraft if ac.wb_config is not None}
202 cover_photos = (
203 {
204 p.aircraft_id: p
205 for p in AircraftPhoto.query.filter(
206 AircraftPhoto.aircraft_id.in_(aircraft_ids),
207 AircraftPhoto.sort_order == 1,
208 ).all()
209 }
210 if aircraft_ids
211 else {}
212 )
213 return render_template(
214 "aircraft/list.html",
215 aircraft=aircraft,
216 aircraft_status=aircraft_status,
217 wb_configured_ids=wb_configured_ids,
218 cover_photos=cover_photos,
219 show_archived=show_archived,
220 archived_count=archived_count,
221 )
224# ── Add aircraft ──────────────────────────────────────────────────────────────
227@aircraft_bp.route("/new", methods=["GET", "POST"])
228@login_required
229@require_role(*_OWNER_ROLES)
230def new_aircraft() -> ResponseReturnValue:
231 if request.method == "POST":
232 return _save_aircraft(None)
233 return render_template("aircraft/aircraft_form.html", aircraft=None)
236# ── Aircraft detail ───────────────────────────────────────────────────────────
239@aircraft_bp.route("/<aircraft_ref:aircraft_id>")
240@login_required
241def detail(aircraft_id: int) -> ResponseReturnValue:
242 from models import Flight, MaintenanceTrigger
244 ac = _get_aircraft_or_404(aircraft_id)
245 components_by_type: dict[Any, list[Any]] = {}
246 for comp in sorted(ac.components, key=lambda c: (c.type, c.position or "")):
247 components_by_type.setdefault(comp.type, []).append(comp)
248 recent_flights = (
249 Flight.query.filter_by(aircraft_id=ac.id)
250 .order_by(
251 Flight.date.desc(),
252 Flight.departure_time.desc().nullslast(),
253 Flight.id.desc(),
254 )
255 .limit(3)
256 .all()
257 )
258 current_hobbs = ac.total_engine_hours
259 current_landings = ac.total_landings
260 current_flight_hours = ac.total_flight_hours
261 triggers = MaintenanceTrigger.query.filter_by(aircraft_id=ac.id).all()
262 maintenance_summary = [
263 (
264 t,
265 t.status(
266 current_engine_hours=current_hobbs,
267 current_landings=current_landings,
268 current_flight_hours=current_flight_hours,
269 ),
270 )
271 for t in triggers
272 ]
273 recent_expenses = (
274 Expense.query.filter_by(aircraft_id=ac.id)
275 .order_by(Expense.date.desc(), Expense.id.desc())
276 .limit(3)
277 .all()
278 )
279 recent_refuels = (
280 Refuel.query.filter_by(aircraft_id=ac.id)
281 .order_by(Refuel.date.desc(), Refuel.id.desc())
282 .limit(3)
283 .all()
284 )
285 recent_documents = (
286 Document.query.filter_by(aircraft_id=ac.id, is_sensitive=False)
287 .order_by(Document.uploaded_at.desc())
288 .limit(3)
289 .all()
290 )
291 document_count = Document.query.filter_by(aircraft_id=ac.id).count()
292 from datetime import date as _cert_date
294 from documents.routes import ( # pyright: ignore[reportMissingImports]
295 active_document_for,
296 )
298 active_insurance_cert = active_document_for(ac.id, DocType.INSURANCE_CERT)
299 active_arc_cert = active_document_for(ac.id, DocType.ARC)
300 upcoming_insurance_cert = (
301 Document.query.filter(
302 Document.aircraft_id == ac.id,
303 Document.doc_type == DocType.INSURANCE_CERT,
304 Document.component_id.is_(None),
305 Document.valid_from > _cert_date.today(),
306 )
307 .order_by(Document.valid_from.asc())
308 .first()
309 )
310 upcoming_arc_cert = (
311 Document.query.filter(
312 Document.aircraft_id == ac.id,
313 Document.doc_type == DocType.ARC,
314 Document.component_id.is_(None),
315 Document.valid_from > _cert_date.today(),
316 )
317 .order_by(Document.valid_from.asc())
318 .first()
319 )
320 open_snags = (
321 Snag.query.filter_by(aircraft_id=ac.id, resolved_at=None)
322 .order_by(Snag.is_grounding.desc(), Snag.reported_at.desc())
323 .all()
324 )
325 wb_cfg = ac.wb_config
326 last_wb_entry = None
327 if wb_cfg:
328 last_wb_entry = (
329 WeightBalanceEntry.query.filter_by(config_id=wb_cfg.id)
330 .order_by(WeightBalanceEntry.date.desc(), WeightBalanceEntry.id.desc())
331 .first()
332 )
333 from datetime import datetime
335 now = datetime.now(UTC)
336 upcoming_reservations = (
337 Reservation.query.filter(
338 Reservation.aircraft_id == ac.id,
339 Reservation.status.in_(
340 [ReservationStatus.CONFIRMED, ReservationStatus.PENDING]
341 ),
342 Reservation.end_dt >= now,
343 )
344 .order_by(Reservation.start_dt)
345 .limit(5)
346 .all()
347 )
348 suggest_components = session.pop(f"suggest_components_{ac.id}", None)
349 photos = (
350 AircraftPhoto.query.filter_by(aircraft_id=ac.id)
351 .order_by(AircraftPhoto.sort_order)
352 .all()
353 )
354 aw_statuses = AirworthinessDocumentStatus.query.filter_by(aircraft_id=ac.id).all()
355 aw_counts: dict[str, int] = {}
356 for _st in aw_statuses:
357 aw_counts[_st.status] = aw_counts.get(_st.status, 0) + 1
358 aw_counts["total"] = len(aw_statuses)
359 _gps_entries = (
360 Flight.query.filter_by(aircraft_id=ac.id)
361 .filter(Flight.gps_track_id.isnot(None))
362 .order_by(Flight.date.asc())
363 .all()
364 )
365 track_rows = [
366 {
367 "date": str(e.date),
368 "dep": e.departure_icao or "",
369 "arr": e.arrival_icao or "",
370 "time_str": f"{e.flight_time} h" if e.flight_time is not None else "",
371 "view_url": url_for(
372 "aircraft.flight_detail",
373 aircraft_id=aircraft_id,
374 flight_id=e.id,
375 ),
376 "geojson": e.gps_track.geojson if e.gps_track else None,
377 }
378 for e in _gps_entries
379 ]
380 _tile = db.session.get(AppSetting, "openaip_api_key")
381 openaip_key = _tile.value if _tile and _tile.value else None
382 from services.component_limits import (
383 aircraft_limit_infos, # pyright: ignore[reportMissingImports]
384 )
386 component_limit_by_id = {
387 info["component"].id: info for info in aircraft_limit_infos(ac)
388 }
389 from datetime import date as _today_date
391 shared_ownership_enabled = _shared_ownership_enabled(ac.tenant_id, ac.id)
392 is_current_user_co_owner = any(
393 o.user_id == session["user_id"] for o in list(ac.owners)
394 )
396 return render_template(
397 "aircraft/detail.html",
398 aircraft=ac,
399 components_by_type=components_by_type,
400 component_limit_by_id=component_limit_by_id,
401 today_date=_today_date.today(),
402 component_types=ComponentType,
403 recent_flights=recent_flights,
404 suggest_components=suggest_components,
405 maintenance_summary=maintenance_summary,
406 recent_expenses=recent_expenses,
407 expense_type_labels=ExpenseType.LABELS,
408 recent_refuels=recent_refuels,
409 recent_documents=recent_documents,
410 document_count=document_count,
411 active_insurance_cert=active_insurance_cert,
412 active_arc_cert=active_arc_cert,
413 upcoming_insurance_cert=upcoming_insurance_cert,
414 upcoming_arc_cert=upcoming_arc_cert,
415 open_snags=open_snags,
416 wb_config=wb_cfg,
417 last_wb_entry=last_wb_entry,
418 upcoming_reservations=upcoming_reservations,
419 ReservationStatus=ReservationStatus,
420 photos=photos,
421 aw_counts=aw_counts,
422 track_rows=track_rows,
423 openaip_key=openaip_key,
424 shared_ownership_enabled=shared_ownership_enabled,
425 owners=ac.owners if shared_ownership_enabled else [],
426 is_current_user_co_owner=is_current_user_co_owner,
427 )
430# ── Edit aircraft ─────────────────────────────────────────────────────────────
433@aircraft_bp.route("/<aircraft_ref:aircraft_id>/edit", methods=["GET", "POST"])
434@login_required
435@require_role(*_OWNER_ROLES)
436def edit_aircraft(aircraft_id: int) -> ResponseReturnValue:
437 ac = _get_aircraft_or_404(aircraft_id)
438 if request.method == "POST":
439 return _save_aircraft(ac)
440 return render_template("aircraft/aircraft_form.html", aircraft=ac)
443def _save_aircraft(ac: Aircraft | None) -> ResponseReturnValue:
444 is_new = ac is None
445 icao_type = request.form.get("aircraft_type_icao", "").strip().upper()
446 registration = request.form.get("registration", "").strip().upper()
447 make = request.form.get("make", "").strip()
448 model = request.form.get("model", "").strip()
449 year_raw = request.form.get("year", "").strip()
450 has_flight_counter = bool(request.form.get("has_flight_counter"))
451 flight_counter_offset_raw = request.form.get("flight_counter_offset", "0.3").strip()
452 fuel_flow_raw = request.form.get("fuel_flow", "").strip()
453 fuel_type = request.form.get("fuel_type", "avgas").strip()
454 if fuel_type not in ("avgas", "ul91", "mogas", "jet_a1"):
455 fuel_type = "avgas"
456 reserve_hourly_rate_raw = request.form.get("reserve_hourly_rate", "").strip()
457 oil_warning_lph_raw = request.form.get("oil_warning_lph", "").strip()
458 showcase_blurb = request.form.get("showcase_blurb", "").strip() or None
459 logbook_time_precision = request.form.get(
460 "logbook_time_precision", "tenth_hour"
461 ).strip()
462 if logbook_time_precision not in ("tenth_hour", "minute"):
463 logbook_time_precision = "tenth_hour"
465 errors = []
466 if not registration:
467 errors.append(_("Registration is required."))
468 elif _registration_taken(
469 registration, _tenant_id(), exclude_aircraft_id=ac.id if ac else None
470 ):
471 errors.append(
472 _(
473 "%(reg)s is already used by another aircraft in this hangar "
474 "(registrations that only differ by spaces or slashes still "
475 "count as the same, since they'd otherwise share a URL).",
476 reg=registration,
477 )
478 )
479 if not make:
480 errors.append(_("Manufacturer is required."))
481 if not model:
482 errors.append(_("Model is required."))
483 year = None
484 if year_raw:
485 try:
486 year = int(year_raw)
487 if not (1900 <= year <= 2100):
488 raise ValueError
489 except ValueError:
490 errors.append(_("Year must be a valid 4-digit year."))
492 flight_counter_offset = 0.3
493 if flight_counter_offset_raw:
494 try:
495 flight_counter_offset = float(flight_counter_offset_raw)
496 if flight_counter_offset < 0:
497 raise ValueError
498 except ValueError:
499 errors.append(_("Flight counter offset must be a non-negative number."))
501 fuel_flow = None
502 if fuel_flow_raw:
503 try:
504 fuel_flow = float(fuel_flow_raw)
505 if fuel_flow < 0:
506 raise ValueError
507 except ValueError:
508 errors.append(_("Fuel consumption must be a non-negative number."))
510 reserve_hourly_rate = None
511 if reserve_hourly_rate_raw:
512 try:
513 reserve_hourly_rate = float(reserve_hourly_rate_raw)
514 if reserve_hourly_rate < 0:
515 raise ValueError
516 except ValueError:
517 errors.append(_("Overhaul reserve accrual must be a non-negative number."))
519 oil_warning_lph = None
520 if oil_warning_lph_raw:
521 try:
522 oil_warning_lph = float(oil_warning_lph_raw)
523 if oil_warning_lph < 0:
524 raise ValueError
525 except ValueError:
526 errors.append(
527 _("Oil consumption warning threshold must be a non-negative number.")
528 )
530 # Fuel tanks (see AircraftFuelTank docstring). Same "replace all rows"
531 # pattern as the W&B stations form: blank-name rows are silently
532 # dropped, a non-positive/invalid capacity is an error.
533 tank_names = request.form.getlist("tank_name[]")
534 tank_capacities_raw = request.form.getlist("tank_capacity[]")
535 tanks_data: list[tuple[str, float]] = []
536 for i, tname in enumerate(tank_names):
537 tname = tname.strip()
538 if not tname:
539 continue
540 try:
541 tcap = float(tank_capacities_raw[i])
542 if tcap <= 0:
543 raise ValueError
544 except (ValueError, IndexError):
545 errors.append(_("Fuel tank capacity must be a positive number."))
546 continue
547 tanks_data.append((tname, tcap))
549 if errors:
550 for msg in errors:
551 flash(msg, "danger")
552 return render_template("aircraft/aircraft_form.html", aircraft=ac)
554 if ac is None:
555 ac = Aircraft(tenant_id=_tenant_id())
556 db.session.add(ac)
558 ac.registration = registration
559 ac.make = make
560 ac.model = model
561 ac.year = year
562 ac.has_flight_counter = has_flight_counter
563 ac.flight_counter_offset = flight_counter_offset
564 ac.fuel_flow = fuel_flow
565 ac.fuel_type = fuel_type
566 ac.logbook_time_precision = logbook_time_precision
567 ac.reserve_hourly_rate = reserve_hourly_rate
568 ac.oil_warning_lph = oil_warning_lph
569 ac.showcase_blurb = showcase_blurb
570 db.session.commit()
572 # Replace fuel tanks (same "delete all, recreate" strategy as W&B stations)
573 for t in list(ac.fuel_tanks):
574 db.session.delete(t)
575 for i, (tname, tcap) in enumerate(tanks_data):
576 db.session.add(
577 AircraftFuelTank(
578 aircraft_id=ac.id, name=tname, capacity_liters=tcap, sort_order=i
579 )
580 )
581 db.session.commit()
583 if is_new:
584 activity("aircraft.created", registration=ac.registration, aircraft_id=ac.id)
585 else:
586 activity("aircraft.updated", registration=ac.registration, aircraft_id=ac.id)
588 if is_new and icao_type:
589 engine_info = get_aircraft_type_engine_info(icao_type)
590 if engine_info:
591 ec, et = engine_info
592 if et == "Piston" and not ac.components:
593 session[f"suggest_components_{ac.id}"] = {"engine_count": ec}
595 flash(_("%(reg)s saved.", reg=ac.registration), "success")
596 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
599# ── Shared ownership (Phase 39a) ──────────────────────────────────────────────
602@aircraft_bp.route("/<aircraft_ref:aircraft_id>/owners", methods=["GET", "POST"])
603@login_required
604@require_role(*_OWNER_ROLES)
605def manage_owners(aircraft_id: int) -> ResponseReturnValue:
606 ac = _get_aircraft_or_404(aircraft_id)
607 if not _shared_ownership_enabled(ac.tenant_id, ac.id):
608 abort(404)
610 tenant_users = (
611 User.query.join(TenantUser, TenantUser.user_id == User.id)
612 .filter(TenantUser.tenant_id == ac.tenant_id, User.is_active.is_(True))
613 .order_by(User.name)
614 .all()
615 )
617 if request.method == "POST":
618 from datetime import date as _date
620 rows, billing_start, hourly_rate, errors = parse_owners_form(request.form)
621 reserve_hourly, reserve_monthly, reserve_errors = parse_reserve_fields(
622 request.form
623 )
624 errors = errors + reserve_errors
626 if errors:
627 for msg in errors:
628 flash(msg, "danger")
629 return render_template(
630 "aircraft/manage_owners.html",
631 aircraft=ac,
632 tenant_users=tenant_users,
633 )
635 submitted_user_ids = {r["user_id"] for r in rows}
636 existing_by_user = {o.user_id: o for o in list(ac.owners)}
637 is_first_save = not existing_by_user
639 for uid, existing in existing_by_user.items():
640 if uid not in submitted_user_ids:
641 db.session.delete(existing)
643 for row in rows:
644 existing = existing_by_user.get(row["user_id"])
645 if existing is not None:
646 existing.share_pct = row["share_pct"]
647 existing.buy_in_amount = row["buy_in_amount"]
648 else:
649 db.session.add(
650 AircraftOwner(
651 aircraft_id=ac.id,
652 user_id=row["user_id"],
653 share_pct=row["share_pct"],
654 buy_in_amount=row["buy_in_amount"],
655 )
656 )
658 ac.co_owner_hourly_rate = hourly_rate
659 ac.reserve_contribution_hourly = reserve_hourly
660 ac.reserve_contribution_monthly = reserve_monthly
661 if billing_start is not None:
662 ac.co_owner_billing_start = billing_start
663 elif is_first_save and rows and ac.co_owner_billing_start is None:
664 ac.co_owner_billing_start = _date.today()
665 db.session.commit()
667 flash(_("Ownership updated."), "success")
668 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
670 return render_template(
671 "aircraft/manage_owners.html",
672 aircraft=ac,
673 tenant_users=tenant_users,
674 )
677def _co_owner_billing_period(period_months_raw: str | None) -> tuple[Any, Any]:
678 from datetime import date as _date
679 from datetime import timedelta as _timedelta
681 try:
682 period_months = int(period_months_raw) if period_months_raw else 12
683 except ValueError:
684 period_months = 12
685 if period_months <= 0:
686 period_months = 12
687 end = _date.today()
688 start = end - _timedelta(days=period_months * 30)
689 return start, end
692@aircraft_bp.route("/<aircraft_ref:aircraft_id>/owners/billing")
693@login_required
694@require_role(*_OWNER_ROLES)
695def owners_billing(aircraft_id: int) -> ResponseReturnValue:
696 from decimal import Decimal
697 from itertools import groupby
699 from models import (
700 BillingAccountKind,
701 CoOwnerValuationSnapshot,
702 LedgerEntryType,
703 LogbookEntryType,
704 TenantProfile,
705 )
706 from services.billing import BillingService
707 from services.co_owner_billing import (
708 overdue_since,
709 reserve_fund_balance,
710 run_co_owner_billing_pass,
711 )
713 ac = _get_aircraft_or_404(aircraft_id)
714 if not _shared_ownership_enabled(ac.tenant_id, ac.id):
715 abort(404)
717 run_co_owner_billing_pass(ac)
718 db.session.commit()
720 profile = TenantProfile.query.filter_by(tenant_id=ac.tenant_id).first()
721 overdue_days = profile.co_owner_overdue_days if profile else 30
723 start, end = _co_owner_billing_period(request.args.get("period"))
725 from datetime import date as _date
727 from models import LedgerEntry
729 today = _date.today()
730 owners = list(ac.owners)
731 rows = []
732 for owner in owners:
733 account = BillingService.get_or_create_account(
734 ac.tenant_id, owner.user_id, BillingAccountKind.CO_OWNER, aircraft_id=ac.id
735 )
736 statement = BillingService.statement(account, start, end)
737 already_reversed_ids = {
738 row[0]
739 for row in db.session.query(LedgerEntry.reverses_id)
740 .filter(
741 LedgerEntry.account_id == account.id,
742 LedgerEntry.reverses_id.isnot(None),
743 )
744 .all()
745 }
747 hours = Decimal(0)
748 fixed_liability = Decimal(0)
749 operating_liability = Decimal(0)
750 reserve_liability = Decimal(0)
751 payments = Decimal(0)
752 for line in statement.lines:
753 entry = line.entry
754 if entry.source_type == "flight_usage":
755 operating_liability += Decimal(entry.amount)
756 flight = db.session.get(Flight, entry.source_id)
757 if flight is not None and flight.flight_time is not None:
758 hours += Decimal(flight.flight_time)
759 elif entry.source_type == "expense_share":
760 fixed_liability += Decimal(entry.amount)
761 elif entry.source_type == "reserve_contribution":
762 reserve_liability += Decimal(entry.amount)
763 elif entry.entry_type == LedgerEntryType.PAYMENT:
764 payments += -Decimal(entry.amount)
766 lines = []
767 for line in reversed(statement.lines):
768 entry = line.entry
769 can_reverse = (
770 entry.entry_type == LedgerEntryType.PAYMENT
771 and entry.reverses_id is None
772 and entry.id not in already_reversed_ids
773 )
774 lines.append({"line": line, "can_reverse": can_reverse})
776 since = overdue_since(account)
777 rows.append(
778 {
779 "owner": owner,
780 "hours": hours,
781 "fixed_liability": fixed_liability,
782 "operating_liability": operating_liability,
783 "reserve_liability": reserve_liability,
784 "payments": payments,
785 "capital_balance": -BillingService.balance(account),
786 "overdue_since": since,
787 "overdue_flagged": since is not None
788 and (today - since).days > overdue_days,
789 "lines": lines,
790 }
791 )
793 owner_user_ids = {o.user_id for o in owners}
794 unattributed_flights: list[Any] = []
795 if ac.co_owner_billing_start is not None:
796 unattributed_flights = [
797 f
798 for f in Flight.query.filter(
799 Flight.aircraft_id == ac.id,
800 Flight.entry_type == LogbookEntryType.FLIGHT,
801 Flight.date >= ac.co_owner_billing_start,
802 Flight.flight_time.isnot(None),
803 Flight.flight_time > 0,
804 ).all()
805 if f.pic_user_id not in owner_user_ids
806 ]
807 unattributed_hours = sum(
808 (Decimal(f.flight_time) for f in unattributed_flights), Decimal(0)
809 )
811 snapshots = (
812 CoOwnerValuationSnapshot.query.filter_by(aircraft_id=ac.id)
813 .order_by(
814 CoOwnerValuationSnapshot.valuation_date.desc(),
815 CoOwnerValuationSnapshot.id.desc(),
816 )
817 .all()
818 )
819 snapshot_groups = [
820 (valuation_date, list(items))
821 for valuation_date, items in groupby(snapshots, key=lambda s: s.valuation_date)
822 ]
824 return render_template(
825 "aircraft/owners_billing.html",
826 aircraft=ac,
827 rows=rows,
828 period=request.args.get("period") or "12",
829 snapshot_groups=snapshot_groups,
830 unattributed_flights=unattributed_flights,
831 unattributed_hours=unattributed_hours,
832 today=today.isoformat(),
833 reserve_fund_balance=reserve_fund_balance(ac),
834 )
837def _get_current_owner_or_404(aircraft: Aircraft, user_id: int) -> AircraftOwner:
838 owner = AircraftOwner.query.filter_by(
839 aircraft_id=aircraft.id, user_id=user_id
840 ).first()
841 if owner is None:
842 abort(404)
843 return cast(AircraftOwner, owner)
846@aircraft_bp.route(
847 "/<aircraft_ref:aircraft_id>/owners/<int:user_id>/payment", methods=["POST"]
848)
849@login_required
850@require_role(*_OWNER_ROLES)
851def owner_record_payment(aircraft_id: int, user_id: int) -> ResponseReturnValue:
852 from datetime import date as _date_cls
854 from models import BillingAccountKind, LedgerEntryType
855 from services.billing import BillingService
857 ac = _get_aircraft_or_404(aircraft_id)
858 if not _shared_ownership_enabled(ac.tenant_id, ac.id):
859 abort(404)
860 _get_current_owner_or_404(ac, user_id)
862 amount_raw = request.form.get("amount", "").strip()
863 date_raw = request.form.get("date", "").strip()
864 note = request.form.get("note", "").strip() or None
866 errors = []
867 try:
868 amount = float(amount_raw)
869 if amount <= 0:
870 errors.append(_("Payment amount must be positive."))
871 except ValueError:
872 amount = 0.0
873 errors.append(_("Invalid payment amount."))
874 try:
875 payment_date = (
876 _date_cls.fromisoformat(date_raw) if date_raw else _date_cls.today()
877 )
878 except ValueError:
879 payment_date = _date_cls.today()
880 errors.append(_("Invalid payment date."))
882 if errors:
883 for msg in errors:
884 flash(msg, "danger")
885 return redirect(url_for("aircraft.owners_billing", aircraft_id=ac.id))
887 account = BillingService.get_or_create_account(
888 ac.tenant_id, user_id, BillingAccountKind.CO_OWNER, aircraft_id=ac.id
889 )
890 recorder = db.session.get(User, session["user_id"])
891 description = str(_("Payment — %(note)s", note=note)) if note else str(_("Payment"))
892 BillingService.post(
893 account,
894 LedgerEntryType.PAYMENT,
895 -amount,
896 description,
897 payment_date,
898 source_type="payment",
899 created_by=recorder,
900 )
901 db.session.commit()
902 flash(_("Payment recorded."), "success")
903 return redirect(url_for("aircraft.owners_billing", aircraft_id=ac.id))
906@aircraft_bp.route(
907 "/<aircraft_ref:aircraft_id>/owners/<int:user_id>/entries/<int:entry_id>/reverse",
908 methods=["POST"],
909)
910@login_required
911@require_role(*_OWNER_ROLES)
912def owner_reverse_entry(
913 aircraft_id: int, user_id: int, entry_id: int
914) -> ResponseReturnValue:
915 from models import BillingAccountKind, LedgerEntry
916 from services.billing import BillingService
918 ac = _get_aircraft_or_404(aircraft_id)
919 if not _shared_ownership_enabled(ac.tenant_id, ac.id):
920 abort(404)
921 _get_current_owner_or_404(ac, user_id)
923 account = BillingService.get_or_create_account(
924 ac.tenant_id, user_id, BillingAccountKind.CO_OWNER, aircraft_id=ac.id
925 )
926 entry = db.session.get(LedgerEntry, entry_id)
927 if entry is None or entry.account_id != account.id:
928 abort(404)
930 recorder = db.session.get(User, session["user_id"])
931 try:
932 BillingService.reverse(entry, recorder, str(_("Payment reversed")))
933 db.session.commit()
934 flash(_("Entry reversed."), "success")
935 except ValueError as exc:
936 flash(str(exc), "danger")
937 return redirect(url_for("aircraft.owners_billing", aircraft_id=ac.id))
940@aircraft_bp.route("/<aircraft_ref:aircraft_id>/owners/valuation", methods=["POST"])
941@login_required
942@require_role(*_OWNER_ROLES)
943def record_valuation_snapshot(aircraft_id: int) -> ResponseReturnValue:
944 from datetime import date as _date_cls
946 from models import BillingAccountKind, CoOwnerValuationSnapshot
947 from services.billing import BillingService
948 from services.co_owner_billing import run_co_owner_billing_pass
950 ac = _get_aircraft_or_404(aircraft_id)
951 if not _shared_ownership_enabled(ac.tenant_id, ac.id):
952 abort(404)
954 date_raw = request.form.get("date", "").strip()
955 note = request.form.get("note", "").strip() or None
956 try:
957 valuation_date = (
958 _date_cls.fromisoformat(date_raw) if date_raw else _date_cls.today()
959 )
960 except ValueError:
961 valuation_date = _date_cls.today()
963 run_co_owner_billing_pass(ac)
964 db.session.commit()
966 recorder = db.session.get(User, session["user_id"])
967 for owner in list(ac.owners):
968 account = BillingService.get_or_create_account(
969 ac.tenant_id, owner.user_id, BillingAccountKind.CO_OWNER, aircraft_id=ac.id
970 )
971 balance = BillingService.balance(account, as_of=valuation_date)
972 db.session.add(
973 CoOwnerValuationSnapshot(
974 aircraft_id=ac.id,
975 user_id=owner.user_id,
976 valuation_date=valuation_date,
977 share_pct=owner.share_pct,
978 capital_balance=-balance,
979 note=note,
980 created_by_id=recorder.id if recorder else None,
981 )
982 )
983 db.session.commit()
984 flash(_("Valuation snapshot recorded."), "success")
985 return redirect(url_for("aircraft.owners_billing", aircraft_id=ac.id))
988@aircraft_bp.route("/<aircraft_ref:aircraft_id>/owners/<int:user_id>/account")
989@login_required
990@require_role(*_OWNER_ROLES)
991def owner_account(aircraft_id: int, user_id: int) -> ResponseReturnValue:
992 from models import BillingAccountKind
993 from services.billing import BillingService
995 ac = _get_aircraft_or_404(aircraft_id)
996 if not _shared_ownership_enabled(ac.tenant_id, ac.id):
997 abort(404)
998 owner = _get_current_owner_or_404(ac, user_id)
1000 account = BillingService.get_or_create_account(
1001 ac.tenant_id, user_id, BillingAccountKind.CO_OWNER, aircraft_id=ac.id
1002 )
1003 db.session.commit()
1004 start, end = _co_owner_billing_period(request.args.get("period"))
1005 statement = BillingService.statement(account, start, end)
1007 return render_template(
1008 "aircraft/owner_account.html",
1009 aircraft=ac,
1010 owner=owner,
1011 account=account,
1012 statement=statement,
1013 balance=BillingService.balance(account),
1014 is_owner_view=True,
1015 csv_url=url_for(
1016 "aircraft.owner_statement_csv", aircraft_id=ac.id, user_id=user_id
1017 ),
1018 )
1021@aircraft_bp.route(
1022 "/<aircraft_ref:aircraft_id>/owners/<int:user_id>/account/statement.csv"
1023)
1024@login_required
1025@require_role(*_OWNER_ROLES)
1026def owner_statement_csv(aircraft_id: int, user_id: int) -> ResponseReturnValue:
1027 from flask import Response
1028 from models import BillingAccountKind
1029 from services.billing import BillingService
1031 ac = _get_aircraft_or_404(aircraft_id)
1032 if not _shared_ownership_enabled(ac.tenant_id, ac.id):
1033 abort(404)
1034 _get_current_owner_or_404(ac, user_id)
1036 account = BillingService.get_or_create_account(
1037 ac.tenant_id, user_id, BillingAccountKind.CO_OWNER, aircraft_id=ac.id
1038 )
1039 db.session.commit()
1040 start, end = _co_owner_billing_period(request.args.get("period"))
1041 statement = BillingService.statement(account, start, end)
1042 exporter = db.session.get(User, session["user_id"])
1043 csv_text = BillingService.statement_csv(statement, exported_by=exporter)
1044 filename = (
1045 f"co_owner_statement_{ac.registration}_{user_id}_"
1046 f"{start.isoformat()}_{end.isoformat()}.csv"
1047 )
1048 return Response(
1049 csv_text,
1050 mimetype="text/csv",
1051 headers={"Content-Disposition": f"attachment; filename={filename}"},
1052 )
1055@aircraft_bp.route("/<aircraft_ref:aircraft_id>/my-share")
1056@login_required
1057def my_share(aircraft_id: int) -> ResponseReturnValue:
1058 from models import BillingAccountKind
1059 from services.billing import BillingService
1061 ac = _get_tenant_aircraft_or_404(aircraft_id)
1062 uid = session["user_id"]
1063 owner = AircraftOwner.query.filter_by(aircraft_id=ac.id, user_id=uid).first()
1064 if owner is None:
1065 abort(404)
1067 account = BillingService.get_or_create_account(
1068 ac.tenant_id, uid, BillingAccountKind.CO_OWNER, aircraft_id=ac.id
1069 )
1070 db.session.commit()
1071 start, end = _co_owner_billing_period(request.args.get("period"))
1072 statement = BillingService.statement(account, start, end)
1074 return render_template(
1075 "aircraft/owner_account.html",
1076 aircraft=ac,
1077 owner=owner,
1078 account=account,
1079 statement=statement,
1080 balance=BillingService.balance(account),
1081 is_owner_view=False,
1082 csv_url=url_for("aircraft.my_share_statement_csv", aircraft_id=ac.id),
1083 )
1086@aircraft_bp.route("/<aircraft_ref:aircraft_id>/my-share/statement.csv")
1087@login_required
1088def my_share_statement_csv(aircraft_id: int) -> ResponseReturnValue:
1089 from flask import Response
1090 from models import BillingAccountKind
1091 from services.billing import BillingService
1093 ac = _get_tenant_aircraft_or_404(aircraft_id)
1094 uid = session["user_id"]
1095 owner = AircraftOwner.query.filter_by(aircraft_id=ac.id, user_id=uid).first()
1096 if owner is None:
1097 abort(404)
1099 account = BillingService.get_or_create_account(
1100 ac.tenant_id, uid, BillingAccountKind.CO_OWNER, aircraft_id=ac.id
1101 )
1102 db.session.commit()
1103 start, end = _co_owner_billing_period(request.args.get("period"))
1104 statement = BillingService.statement(account, start, end)
1105 exporter = db.session.get(User, uid)
1106 csv_text = BillingService.statement_csv(statement, exported_by=exporter)
1107 filename = f"my_share_statement_{ac.registration}_{start.isoformat()}_{end.isoformat()}.csv"
1108 return Response(
1109 csv_text,
1110 mimetype="text/csv",
1111 headers={"Content-Disposition": f"attachment; filename={filename}"},
1112 )
1115# ── Delete aircraft ───────────────────────────────────────────────────────────
1118@aircraft_bp.route("/<aircraft_ref:aircraft_id>/archive", methods=["POST"])
1119@login_required
1120@require_role(*_OWNER_ROLES)
1121def archive_aircraft(aircraft_id: int) -> ResponseReturnValue:
1122 from datetime import datetime
1124 ac = _get_aircraft_or_404(aircraft_id)
1125 if not ac.is_archived:
1126 ac.archived_at = datetime.now(UTC)
1127 db.session.commit()
1128 activity("aircraft.archived", registration=ac.registration, aircraft_id=ac.id)
1129 flash(
1130 _(
1131 "%(reg)s has been archived — its history stays available, but it no "
1132 "longer appears in the active fleet.",
1133 reg=ac.registration,
1134 ),
1135 "success",
1136 )
1137 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
1140@aircraft_bp.route("/<aircraft_ref:aircraft_id>/unarchive", methods=["POST"])
1141@login_required
1142@require_role(*_OWNER_ROLES)
1143def unarchive_aircraft(aircraft_id: int) -> ResponseReturnValue:
1144 ac = _get_aircraft_or_404(aircraft_id)
1145 if ac.is_archived:
1146 ac.archived_at = None
1147 db.session.commit()
1148 activity("aircraft.unarchived", registration=ac.registration, aircraft_id=ac.id)
1149 flash(_("%(reg)s is back in the active fleet.", reg=ac.registration), "success")
1150 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
1153@aircraft_bp.route("/<aircraft_ref:aircraft_id>/delete", methods=["POST"])
1154@login_required
1155@require_role(*_OWNER_ROLES)
1156def delete_aircraft(aircraft_id: int) -> ResponseReturnValue:
1157 ac = _get_aircraft_or_404(aircraft_id)
1158 reg = ac.registration
1159 activity("aircraft.deleted", registration=reg, aircraft_id=aircraft_id)
1160 db.session.delete(ac)
1161 db.session.commit()
1162 flash(_("%(reg)s and all its components have been deleted.", reg=reg), "success")
1163 return redirect(url_for("aircraft.list_aircraft"))
1166# ── Quick-add components from ICAO type suggestion ────────────────────────────
1169@aircraft_bp.route("/<aircraft_ref:aircraft_id>/quick-add-components", methods=["POST"])
1170@login_required
1171@require_role(*_OWNER_ROLES)
1172def quick_add_components(aircraft_id: int) -> ResponseReturnValue:
1173 ac = _get_aircraft_or_404(aircraft_id)
1174 try:
1175 engine_count = max(1, min(int(request.form.get("engine_count", "1")), 4))
1176 except ValueError:
1177 engine_count = 1
1178 for i in range(engine_count):
1179 position = str(i + 1) if engine_count > 1 else None
1180 db.session.add(
1181 Component(
1182 aircraft_id=ac.id,
1183 type=ComponentType.ENGINE,
1184 position=position,
1185 make="",
1186 model="",
1187 )
1188 )
1189 db.session.add(
1190 Component(
1191 aircraft_id=ac.id,
1192 type=ComponentType.PROPELLER,
1193 position=position,
1194 make="",
1195 model="",
1196 )
1197 )
1198 db.session.commit()
1199 activity(
1200 "component.quick_added",
1201 aircraft_id=aircraft_id,
1202 engine_count=engine_count,
1203 )
1204 flash(
1205 ngettext(
1206 "Engine and propeller added — fill in the details when ready.",
1207 "%(n)s engines and propellers added — fill in the details when ready.",
1208 engine_count,
1209 n=engine_count,
1210 ),
1211 "success",
1212 )
1213 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
1216# ── Add component ─────────────────────────────────────────────────────────────
1219@aircraft_bp.route(
1220 "/<aircraft_ref:aircraft_id>/components/new", methods=["GET", "POST"]
1221)
1222@login_required
1223@require_role(*_OWNER_ROLES)
1224def new_component(aircraft_id: int) -> ResponseReturnValue:
1225 ac = _get_aircraft_or_404(aircraft_id)
1226 if request.method == "POST":
1227 return _save_component(ac, None)
1228 return render_template(
1229 "aircraft/component_form.html",
1230 aircraft=ac,
1231 component=None,
1232 component_types=ComponentType,
1233 )
1236# ── Edit component ────────────────────────────────────────────────────────────
1239@aircraft_bp.route(
1240 "/<aircraft_ref:aircraft_id>/components/<int:component_id>/edit",
1241 methods=["GET", "POST"],
1242)
1243@login_required
1244@require_role(*_OWNER_ROLES)
1245def edit_component(aircraft_id: int, component_id: int) -> ResponseReturnValue:
1246 ac = _get_aircraft_or_404(aircraft_id)
1247 comp = _get_component_or_404(ac, component_id)
1248 if request.method == "POST":
1249 return _save_component(ac, comp)
1250 return render_template(
1251 "aircraft/component_form.html",
1252 aircraft=ac,
1253 component=comp,
1254 component_types=ComponentType,
1255 )
1258def _save_component(ac: Aircraft, comp: Component | None) -> ResponseReturnValue:
1259 from datetime import date as _date
1261 type_ = request.form.get("type", "").strip()
1262 position = request.form.get("position", "").strip() or None
1263 make = request.form.get("make", "").strip()
1264 model = request.form.get("model", "").strip()
1265 serial = request.form.get("serial_number", "").strip() or None
1266 time_raw = request.form.get("time_at_install", "").strip()
1267 installed_raw = request.form.get("installed_at", "").strip()
1268 removed_raw = request.form.get("removed_at", "").strip()
1269 tbo_raw = request.form.get("tbo_hours", "").strip()
1270 life_limit_raw = request.form.get("life_limit_date", "").strip()
1271 overhauled_at_raw = request.form.get("overhauled_at_hours", "").strip()
1272 overhauled_on_raw = request.form.get("overhauled_on", "").strip()
1274 errors = []
1275 if not type_:
1276 errors.append(_("Component type is required."))
1277 if not make:
1278 errors.append(_("Manufacturer is required."))
1279 if not model:
1280 errors.append(_("Model is required."))
1282 time_at_install = None
1283 if time_raw:
1284 try:
1285 time_at_install = float(time_raw)
1286 if time_at_install < 0:
1287 raise ValueError
1288 except ValueError:
1289 errors.append(_("Time at install must be a positive number."))
1291 tbo_hours = None
1292 if tbo_raw:
1293 try:
1294 tbo_hours = float(tbo_raw)
1295 if tbo_hours <= 0:
1296 raise ValueError
1297 except ValueError:
1298 errors.append(_("TBO must be a positive number of hours."))
1300 overhauled_at_hours = None
1301 if overhauled_at_raw:
1302 try:
1303 overhauled_at_hours = float(overhauled_at_raw)
1304 if overhauled_at_hours < 0:
1305 raise ValueError
1306 except ValueError:
1307 errors.append(_("Last overhaul hours must be a non-negative number."))
1309 def _parse_date(raw: str, label: str) -> Any:
1310 if not raw:
1311 return None
1312 try:
1313 return _date.fromisoformat(raw)
1314 except ValueError:
1315 errors.append(
1316 _("%(label)s must be a valid date (YYYY-MM-DD).", label=label)
1317 )
1318 return None
1320 installed_at = _parse_date(installed_raw, "Install date")
1321 removed_at = _parse_date(removed_raw, "Removal date")
1322 life_limit_date = _parse_date(life_limit_raw, "Calendar life limit")
1323 overhauled_on = _parse_date(overhauled_on_raw, "Last overhaul date")
1325 if errors:
1326 for msg in errors:
1327 flash(msg, "danger")
1328 return render_template(
1329 "aircraft/component_form.html",
1330 aircraft=ac,
1331 component=comp,
1332 component_types=ComponentType,
1333 )
1335 _comp_is_new = comp is None
1336 if comp is None:
1337 comp = Component(aircraft_id=ac.id)
1338 db.session.add(comp)
1340 comp.type = type_
1341 comp.position = position
1342 comp.make = make
1343 comp.model = model
1344 comp.serial_number = serial
1345 comp.time_at_install = time_at_install
1346 comp.installed_at = installed_at
1347 comp.removed_at = removed_at
1348 comp.tbo_hours = tbo_hours
1349 comp.life_limit_date = life_limit_date
1350 comp.overhauled_at_hours = overhauled_at_hours
1351 comp.overhauled_on = overhauled_on
1352 db.session.commit()
1354 if _comp_is_new:
1355 activity(
1356 "component.added", type=comp.type, component_id=comp.id, aircraft_id=ac.id
1357 )
1358 else:
1359 activity(
1360 "component.updated", type=comp.type, component_id=comp.id, aircraft_id=ac.id
1361 )
1363 flash(_("%(make)s %(model)s saved.", make=comp.make, model=comp.model), "success")
1364 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
1367# ── Delete component ──────────────────────────────────────────────────────────
1370@aircraft_bp.route(
1371 "/<aircraft_ref:aircraft_id>/components/<int:component_id>/delete", methods=["POST"]
1372)
1373@login_required
1374@require_role(*_OWNER_ROLES)
1375def delete_component(aircraft_id: int, component_id: int) -> ResponseReturnValue:
1376 ac = _get_aircraft_or_404(aircraft_id)
1377 comp = _get_component_or_404(ac, component_id)
1378 label = f"{comp.make} {comp.model}"
1379 activity(
1380 "component.deleted",
1381 type=comp.type,
1382 component_id=component_id,
1383 aircraft_id=aircraft_id,
1384 )
1385 db.session.delete(comp)
1386 db.session.commit()
1387 flash(_("%(label)s removed.", label=label), "success")
1388 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
1391# ── Mass & Balance: helpers ───────────────────────────────────────────────────
1394def _point_in_polygon(cg: float, weight: float, points: Any) -> bool:
1395 """Ray-casting point-in-polygon test. points: list of [arm, weight] pairs.
1397 Returns False (outside) for malformed point data rather than raising —
1398 envelope_points is a DB-stored JSON field with no enforced schema, so a
1399 corrupted or malformed entry should degrade to "envelope check
1400 unavailable" (the conservative, fail-safe answer for a W&B calculation),
1401 not crash the entry page with an unhandled 500. Non-finite coordinates
1402 (e.g. "Infinity"/"NaN" — accepted by float() but not a valid arm/weight
1403 value) are rejected the same way.
1404 """
1405 try:
1406 coords = [(float(p[0]), float(p[1])) for p in points]
1407 except (IndexError, ValueError, TypeError):
1408 return False
1409 if any(not (math.isfinite(x) and math.isfinite(y)) for x, y in coords):
1410 return False
1411 n = len(coords)
1412 inside = False
1413 j = n - 1
1414 for i in range(n):
1415 xi, yi = coords[i]
1416 xj, yj = coords[j]
1417 if ((yi > weight) != (yj > weight)) and (
1418 cg < (xj - xi) * (weight - yi) / (yj - yi) + xi
1419 ):
1420 inside = not inside
1421 j = i
1422 return inside
1425# ── Mass & Balance: config ────────────────────────────────────────────────────
1428@aircraft_bp.route("/<aircraft_ref:aircraft_id>/wb/config", methods=["GET", "POST"])
1429@login_required
1430@require_role(*_OWNER_ROLES)
1431def wb_config(aircraft_id: int) -> ResponseReturnValue:
1432 ac = _get_aircraft_or_404(aircraft_id)
1433 cfg: WeightBalanceConfig | None = ac.wb_config # type: ignore[assignment]
1435 if request.method == "POST":
1436 errors = []
1438 def _f(name: str) -> float | None:
1439 try:
1440 v = float(request.form.get(name, "").strip())
1441 if v < 0:
1442 raise ValueError
1443 return v
1444 except ValueError:
1445 errors.append(_("%(field)s must be a positive number.", field=name))
1446 return None
1448 empty_weight = _f("empty_weight")
1449 empty_cg_arm = _f("empty_cg_arm")
1450 max_takeoff_weight = _f("max_takeoff_weight")
1451 forward_cg_limit = _f("forward_cg_limit")
1452 aft_cg_limit = _f("aft_cg_limit")
1453 datum_note = request.form.get("datum_note", "").strip() or None
1455 fuel_unit = request.form.get("fuel_unit", "L").strip()
1456 if fuel_unit not in ("L", "gal"):
1457 fuel_unit = "L"
1459 # Stations: label[], arm[], station_limit[] (capacity for fuel, max_weight for non-fuel), is_fuel[]
1460 labels = request.form.getlist("station_label[]")
1461 arms = request.form.getlist("station_arm[]")
1462 limits = request.form.getlist("station_limit[]")
1463 is_fuels = request.form.getlist(
1464 "station_is_fuel[]"
1465 ) # index values of checked boxes
1467 if not labels or all(lbl.strip() == "" for lbl in labels):
1468 errors.append(_("At least one loading station is required."))
1470 if errors:
1471 for msg in errors:
1472 flash(msg, "danger")
1473 return render_template("aircraft/wb_config.html", aircraft=ac, config=cfg)
1475 if cfg is None:
1476 cfg = WeightBalanceConfig(aircraft_id=ac.id)
1477 db.session.add(cfg)
1479 # Optional envelope polygon: env_arm[], env_weight[]
1480 env_arms = request.form.getlist("env_arm[]")
1481 env_weights = request.form.getlist("env_weight[]")
1482 envelope_points = []
1483 for arm_s, w_s in zip(env_arms, env_weights):
1484 try:
1485 a = float(arm_s.strip())
1486 w = float(w_s.strip())
1487 if a >= 0 and w >= 0:
1488 envelope_points.append([round(a, 4), round(w, 2)])
1489 except (ValueError, AttributeError):
1490 continue
1492 cfg.empty_weight = empty_weight
1493 cfg.empty_cg_arm = empty_cg_arm
1494 cfg.max_takeoff_weight = max_takeoff_weight
1495 cfg.forward_cg_limit = forward_cg_limit
1496 cfg.aft_cg_limit = aft_cg_limit
1497 cfg.fuel_unit = fuel_unit
1498 cfg.datum_note = datum_note
1499 cfg.envelope_points = envelope_points if len(envelope_points) >= 3 else None
1501 # Replace stations
1502 for s in list(cfg.stations):
1503 db.session.delete(s)
1504 db.session.flush()
1506 for i, label in enumerate(labels):
1507 label = label.strip()
1508 if not label:
1509 continue
1510 try:
1511 arm = float(arms[i])
1512 except (ValueError, IndexError):
1513 continue
1514 limit_val = None
1515 try:
1516 lim_raw = limits[i].strip()
1517 if lim_raw:
1518 limit_val = float(lim_raw)
1519 except (ValueError, IndexError):
1520 limit_val = None
1521 is_fuel = str(i) in is_fuels
1522 db.session.add(
1523 WeightBalanceStation(
1524 config_id=cfg.id,
1525 label=label,
1526 arm=arm,
1527 max_weight=None if is_fuel else limit_val,
1528 capacity=limit_val if is_fuel else None,
1529 is_fuel=is_fuel,
1530 position=i,
1531 )
1532 )
1534 db.session.commit()
1535 flash(_("W&B configuration saved."), "success")
1536 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
1538 return render_template("aircraft/wb_config.html", aircraft=ac, config=cfg)
1541# ── Mass & Balance: entry list ────────────────────────────────────────────────
1544@aircraft_bp.route("/<aircraft_ref:aircraft_id>/wb/")
1545@login_required
1546def wb_list(aircraft_id: int) -> ResponseReturnValue:
1547 ac = _get_aircraft_or_404(aircraft_id)
1548 if not ac.wb_config:
1549 flash(_("Configure W&B envelope first."), "warning")
1550 return redirect(url_for("aircraft.wb_config", aircraft_id=ac.id))
1551 entries = (
1552 WeightBalanceEntry.query.filter_by(config_id=ac.wb_config.id)
1553 .order_by(WeightBalanceEntry.date.desc(), WeightBalanceEntry.id.desc())
1554 .all()
1555 )
1556 return render_template(
1557 "aircraft/wb_list.html", aircraft=ac, config=ac.wb_config, entries=entries
1558 )
1561# ── Mass & Balance: new / edit entry ─────────────────────────────────────────
1564@aircraft_bp.route("/<aircraft_ref:aircraft_id>/wb/new", methods=["GET", "POST"])
1565@aircraft_bp.route(
1566 "/<aircraft_ref:aircraft_id>/wb/<int:entry_id>/edit", methods=["GET", "POST"]
1567)
1568@login_required
1569@require_role(*_PILOT_ROLES)
1570def wb_entry(aircraft_id: int, entry_id: int | None = None) -> ResponseReturnValue:
1571 ac = _get_aircraft_or_404(aircraft_id)
1572 if not ac.wb_config:
1573 flash(_("Configure W&B envelope first."), "warning")
1574 return redirect(url_for("aircraft.wb_config", aircraft_id=ac.id))
1575 cfg = ac.wb_config
1577 entry = None
1578 if entry_id is not None:
1579 entry = db.session.get(WeightBalanceEntry, entry_id)
1580 if not entry or entry.config_id != cfg.id:
1581 abort(404)
1583 if request.method == "POST":
1584 from datetime import date as _date
1586 errors = []
1587 date_raw = request.form.get("date", "").strip()
1588 label = request.form.get("label", "").strip() or None
1589 try:
1590 entry_date = _date.fromisoformat(date_raw)
1591 except ValueError:
1592 errors.append(_("A valid date is required."))
1593 entry_date = None
1595 # Per-station values: fuel stations store volume (L/gal), non-fuel store kg
1596 station_weights = {}
1597 for st in cfg.stations:
1598 if st.is_fuel:
1599 raw = request.form.get(f"volume_{st.id}", "").strip()
1600 try:
1601 vol = float(raw) if raw else 0.0
1602 if vol < 0:
1603 raise ValueError
1604 if st.capacity is not None and vol > float(st.capacity):
1605 errors.append(
1606 _(
1607 "Volume for %(station)s exceeds tank capacity.",
1608 station=st.label,
1609 )
1610 )
1611 station_weights[str(st.id)] = vol
1612 except ValueError:
1613 errors.append(
1614 _(
1615 "Volume for %(station)s must be a non-negative number.",
1616 station=st.label,
1617 )
1618 )
1619 else:
1620 raw = request.form.get(f"weight_{st.id}", "").strip()
1621 try:
1622 w = float(raw) if raw else 0.0
1623 if w < 0:
1624 raise ValueError
1625 station_weights[str(st.id)] = w
1626 except ValueError:
1627 errors.append(
1628 _(
1629 "Weight for %(station)s must be a non-negative number.",
1630 station=st.label,
1631 )
1632 )
1634 # CG computation — fuel stations: convert volume → kg
1635 empty_w = float(cfg.empty_weight)
1636 empty_arm = float(cfg.empty_cg_arm)
1637 total_moment = empty_w * empty_arm
1638 total_weight = empty_w
1639 fuel_density = FUEL_DENSITY.get(ac.fuel_type, 0.72)
1640 gal_factor = GAL_TO_L if cfg.fuel_unit == "gal" else 1.0
1641 for st in cfg.stations:
1642 val = station_weights.get(str(st.id), 0.0)
1643 w_kg = val * fuel_density * gal_factor if st.is_fuel else val
1644 total_weight += w_kg
1645 total_moment += w_kg * float(st.arm)
1646 loaded_cg = total_moment / total_weight if total_weight else 0.0
1648 if cfg.envelope_points and len(cfg.envelope_points) >= 3:
1649 in_env = _point_in_polygon(loaded_cg, total_weight, cfg.envelope_points)
1650 else:
1651 mtow = float(cfg.max_takeoff_weight)
1652 fwd = float(cfg.forward_cg_limit)
1653 aft = float(cfg.aft_cg_limit)
1654 in_env = total_weight <= mtow and fwd <= loaded_cg <= aft
1656 if errors:
1657 for msg in errors:
1658 flash(msg, "danger")
1659 return render_template(
1660 "aircraft/wb_entry.html",
1661 aircraft=ac,
1662 config=cfg,
1663 entry=entry,
1664 fuel_density=FUEL_DENSITY,
1665 )
1667 if entry is None:
1668 entry = WeightBalanceEntry(config_id=cfg.id)
1669 db.session.add(entry)
1671 entry.date = entry_date
1672 entry.label = label
1673 entry.total_weight = round(total_weight, 2)
1674 entry.loaded_cg = round(loaded_cg, 2)
1675 entry.is_in_envelope = in_env
1676 entry.station_weights = station_weights
1677 db.session.commit()
1678 flash(_("W&B calculation saved."), "success")
1679 return redirect(url_for("aircraft.wb_list", aircraft_id=ac.id))
1681 return render_template(
1682 "aircraft/wb_entry.html",
1683 aircraft=ac,
1684 config=cfg,
1685 entry=entry,
1686 fuel_density=FUEL_DENSITY,
1687 )
1690# ── Mass & Balance: delete entry ──────────────────────────────────────────────
1693@aircraft_bp.route(
1694 "/<aircraft_ref:aircraft_id>/wb/<int:entry_id>/delete", methods=["POST"]
1695)
1696@login_required
1697@require_role(*_PILOT_ROLES)
1698def wb_entry_delete(aircraft_id: int, entry_id: int) -> ResponseReturnValue:
1699 ac = _get_aircraft_or_404(aircraft_id)
1700 if not ac.wb_config:
1701 abort(404)
1702 entry = db.session.get(WeightBalanceEntry, entry_id)
1703 if not entry or entry.config_id != ac.wb_config.id:
1704 abort(404)
1705 db.session.delete(entry)
1706 db.session.commit()
1707 flash(_("W&B calculation deleted."), "success")
1708 return redirect(url_for("aircraft.wb_list", aircraft_id=ac.id))
1711# ── Phase 30: GPS Log Import ──────────────────────────────────────────────────
1713_GPS_ALLOWED_EXTS = {".gpx", ".kml", ".csv"}
1714_GPS_MAX_BYTES = 20 * 1024 * 1024 # 20 MB per file
1717def _gps_tmp_dir() -> str:
1718 """Return (and create if needed) the tmp directory for GPS uploads."""
1719 upload_folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
1720 d = os.path.join(upload_folder, "gps_import_tmp")
1721 os.makedirs(d, exist_ok=True)
1722 return d
1725def _segment_to_dict(seg: Any, idx: int) -> dict[str, Any]:
1726 """Serialise a FlightSegment for template rendering (includes track_geojson)."""
1727 return {
1728 "idx": idx,
1729 "block_off_utc": seg.block_off_utc.isoformat(),
1730 "block_on_utc": seg.block_on_utc.isoformat(),
1731 "takeoff_utc": seg.takeoff_utc.isoformat() if seg.takeoff_utc else None,
1732 "landing_utc": seg.landing_utc.isoformat() if seg.landing_utc else None,
1733 "departure_icao": seg.departure_icao or "",
1734 "arrival_icao": seg.arrival_icao or "",
1735 "flight_time_raw_h": seg.flight_time_raw_h,
1736 "flight_time_rounded_h": seg.flight_time_rounded_h,
1737 "landing_count": seg.landing_count,
1738 "is_ground_only": seg.is_ground_only,
1739 "track_geojson": seg.track_geojson,
1740 }
1743def _linked_pilot_entries(flight_id: int, exclude_user_id: int) -> list[dict[str, Any]]:
1744 """Return metadata about the *other* crew member (if any) on this Flight
1745 row, for the GPS-track-replacement notice shown before an upload
1746 overwrites the shared airframe/GPS data.
1748 Unified model note: there's only one gps_track_id now, shared by both
1749 crew slots — the old per-pilot "preserve their own track if they already
1750 have one" policy doesn't apply any more (there's nothing to preserve
1751 separately); this is now purely informational — surfacing that another
1752 named pilot is also linked to this flight and will see the replaced
1753 track too.
1754 """
1755 fe = db.session.get(Flight, flight_id)
1756 assert fe is not None, "caller passes a Flight id it just queried"
1757 other_user_id = None
1758 if fe.pic_user_id and fe.pic_user_id != exclude_user_id:
1759 other_user_id = fe.pic_user_id
1760 elif fe.second_crew_user_id and fe.second_crew_user_id != exclude_user_id:
1761 other_user_id = fe.second_crew_user_id
1762 if other_user_id is None:
1763 return []
1764 user = db.session.get(User, other_user_id)
1765 return [
1766 {
1767 "user_id": other_user_id,
1768 "display_name": user.display_name if user else f"user #{other_user_id}",
1769 "has_existing_track": fe.gps_track_id is not None,
1770 }
1771 ]
1774def _gps_candidate_str(fe: Any) -> str:
1775 """One-line summary of an existing Flight row for a GPS-import match
1776 picker — includes whatever the row already has recorded (times,
1777 duration) so a human can tell it apart from a similar-looking flight
1778 without needing to open it first."""
1779 s = f"#{fe.id} — {fe.date} {fe.departure_icao} → {fe.arrival_icao}"
1780 if fe.departure_time and fe.arrival_time:
1781 s += (
1782 f" ({fe.departure_time.strftime('%H:%M')}"
1783 f"–{fe.arrival_time.strftime('%H:%M')})"
1784 )
1785 if fe.total_flight_time is not None:
1786 s += f", {fe.total_flight_time:.1f} h"
1787 return s
1790def _gps_candidate_dict(fe: Any) -> dict[str, Any]:
1791 """Serialise one match candidate for the review template/session."""
1792 return {
1793 "id": fe.id,
1794 "str": _gps_candidate_str(fe),
1795 "aircraft_id": fe.aircraft_id,
1796 "aircraft_reg": fe.aircraft.registration if fe.aircraft else "?",
1797 "has_existing_track": fe.gps_track_id is not None,
1798 }
1801def _load_segment_geojson(seg: dict[str, Any]) -> Any:
1802 """Read the GeoJSON dict back from the tmp file written by _segment_for_session."""
1803 path = seg.get("geojson_path")
1804 if not path or not os.path.exists(path):
1805 return None
1806 with open(path, encoding="utf-8") as fh:
1807 return json.load(fh)
1810def _segment_for_session(seg_dict: dict[str, Any], tmp_dir: str) -> dict[str, Any]:
1811 """Return a copy of seg_dict safe for cookie-session storage.
1813 track_geojson can be hundreds of KB — too large for Flask's 4 KB cookie
1814 limit. We spill it to a tmp file and store the path instead.
1815 """
1816 s = {k: v for k, v in seg_dict.items() if k != "track_geojson"}
1817 geojson = seg_dict.get("track_geojson")
1818 if geojson is not None:
1819 fname = f"seg_{seg_dict['idx']}_{_uuid_mod.uuid4().hex}.geojson"
1820 path = os.path.join(tmp_dir, fname)
1821 with open(path, "w", encoding="utf-8") as fh:
1822 json.dump(geojson, fh)
1823 s["geojson_path"] = path
1824 return s
1827@aircraft_bp.route("/<aircraft_ref:aircraft_id>/gps-import", methods=["GET", "POST"])
1828@login_required
1829@require_role(*_PILOT_ROLES)
1830def gps_import_upload(aircraft_id: int) -> ResponseReturnValue:
1831 ac = _get_aircraft_or_404(aircraft_id)
1833 if request.method == "GET":
1834 return render_template("aircraft/gps_import_upload.html", aircraft=ac)
1836 files = request.files.getlist("gps_files")
1837 if not files or all(f.filename == "" for f in files):
1838 flash(_("Please select at least one GPS log file."), "warning")
1839 return render_template("aircraft/gps_import_upload.html", aircraft=ac)
1841 tmp_dir = _gps_tmp_dir()
1842 parsed_meta: list[dict[str, Any]] = []
1843 errors: list[str] = []
1844 skipped_empty = 0
1845 formats: list[str] = []
1847 for f in files:
1848 if not f.filename:
1849 continue
1850 ext = os.path.splitext(f.filename.lower())[1]
1851 if ext not in _GPS_ALLOWED_EXTS:
1852 errors.append(
1853 _(
1854 "%(fn)s: unsupported file type (use .gpx, .kml, or .csv).",
1855 fn=f.filename,
1856 )
1857 )
1858 continue
1860 data = f.read(_GPS_MAX_BYTES + 1)
1861 if len(data) > _GPS_MAX_BYTES:
1862 errors.append(_("%(fn)s: file too large (20 MB limit).", fn=f.filename))
1863 continue
1865 try:
1866 parsed = parse_gps_file(data, f.filename)
1867 except ValueError as exc:
1868 errors.append(_("%(fn)s: %(err)s", fn=f.filename, err=str(exc)))
1869 continue
1871 if parsed.classification == "empty":
1872 skipped_empty += 1
1873 continue
1875 # Save raw bytes to tmp
1876 uid = _uuid_mod.uuid4().hex
1877 safe_name = f"{uid}_{secure_filename(f.filename)}"
1878 tmp_path = os.path.join(tmp_dir, safe_name)
1879 with open(tmp_path, "wb") as fh:
1880 fh.write(data)
1882 parsed_meta.append(
1883 {
1884 "tmp_path": tmp_path,
1885 "original_filename": f.filename,
1886 "format": parsed.format,
1887 "classification": parsed.classification,
1888 "trkpt_count": len(parsed.trackpoints),
1889 "hint_dep": parsed.hint_departure_icao,
1890 "hint_arr": parsed.hint_arrival_icao,
1891 "device_id": getattr(parsed, "device_id", None),
1892 }
1893 )
1894 formats.append(parsed.format)
1896 if errors:
1897 for e in errors:
1898 flash(e, "danger")
1899 if skipped_empty:
1900 flash(
1901 ngettext(
1902 "%(n)s file skipped — no movement detected.",
1903 "%(n)s files skipped — no movement detected.",
1904 skipped_empty,
1905 n=skipped_empty,
1906 ),
1907 "info",
1908 )
1910 if not parsed_meta:
1911 flash(_("No valid GPS files to import."), "warning")
1912 return render_template("aircraft/gps_import_upload.html", aircraft=ac)
1914 other_aircraft = request.form.get("other_aircraft") == "1"
1915 other_ac_make_model = request.form.get("other_ac_make_model", "").strip()
1916 other_ac_reg = request.form.get("other_ac_reg", "").strip().upper()
1918 session["gps_import"] = {
1919 "user_id": session["user_id"],
1920 "aircraft_id": aircraft_id,
1921 "files": parsed_meta,
1922 "skipped_empty": skipped_empty,
1923 "other_aircraft": other_aircraft,
1924 "other_ac_make_model": other_ac_make_model,
1925 "other_ac_reg": other_ac_reg,
1926 }
1927 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
1930@aircraft_bp.route("/<aircraft_ref:aircraft_id>/gps-import/review", methods=["GET"])
1931@login_required
1932@require_role(*_PILOT_ROLES)
1933def gps_import_review(aircraft_id: int) -> ResponseReturnValue:
1934 ac = _get_aircraft_or_404(aircraft_id)
1935 state = session.get("gps_import")
1936 if not state or state.get("aircraft_id") != aircraft_id:
1937 flash(_("Session expired — please upload your GPS files again."), "warning")
1938 return redirect(url_for("aircraft.gps_import_upload", aircraft_id=aircraft_id))
1940 file_metas = state["files"]
1942 # Re-parse each tmp file and build combined trackpoint list
1943 from aircraft.gps_import import (
1944 ParsedGpsFile, # pyright: ignore[reportMissingImports]
1945 )
1947 all_parsed: list[ParsedGpsFile] = []
1948 for meta in file_metas:
1949 try:
1950 with open(meta["tmp_path"], "rb") as fh:
1951 data = fh.read()
1952 parsed = parse_gps_file(data, meta["original_filename"])
1953 parsed.hint_departure_icao = meta.get("hint_dep")
1954 parsed.hint_arrival_icao = meta.get("hint_arr")
1955 all_parsed.append(parsed)
1956 except (OSError, ValueError):
1957 flash(
1958 _(
1959 "Could not read %(fn)s — please upload again.",
1960 fn=meta["original_filename"],
1961 ),
1962 "warning",
1963 )
1964 return redirect(
1965 url_for("aircraft.gps_import_upload", aircraft_id=aircraft_id)
1966 )
1968 merged = merge_and_sort(all_parsed)
1970 # Collect ICAO hints from all files
1971 hint_dep = next(
1972 (p.hint_departure_icao for p in all_parsed if p.hint_departure_icao), None
1973 )
1974 hint_arr = next(
1975 (p.hint_arrival_icao for p in all_parsed if p.hint_arrival_icao), None
1976 )
1978 segments = detect_segments(
1979 merged,
1980 aircraft_precision=ac.logbook_time_precision,
1981 hint_dep=hint_dep,
1982 hint_arr=hint_arr,
1983 )
1985 # Build full dicts (with track_geojson) for template rendering.
1986 # Spill GeoJSON to tmp files for the session — track_geojson can be
1987 # hundreds of KB and silently overflows Flask's 4 KB cookie session.
1988 full_segs = [_segment_to_dict(seg, i) for i, seg in enumerate(segments)]
1990 # Duplicate detection: find existing Flight records that overlap each segment.
1991 from datetime import datetime as _dt
1992 from datetime import timedelta as _td
1993 from functools import partial
1995 from flights.airframe_import import _CANDIDATE_MIN_SCORE
1997 from aircraft.gps_import import _score_gps_candidate, score_gps_candidates
1999 _BLOCK_TOLERANCE = _td(minutes=15)
2001 for seg in full_segs:
2002 block_off = _dt.fromisoformat(seg["block_off_utc"])
2003 block_on = _dt.fromisoformat(seg["block_on_utc"])
2004 matched = Flight.query.filter(
2005 Flight.aircraft_id == aircraft_id,
2006 Flight.block_off_utc.isnot(None),
2007 Flight.block_on_utc.isnot(None),
2008 Flight.block_off_utc < block_on + _BLOCK_TOLERANCE,
2009 Flight.block_on_utc > block_off - _BLOCK_TOLERANCE,
2010 ).first()
2011 seg["matched_ambiguous"] = False
2012 seg["matched_candidates"] = []
2013 if matched:
2014 seg["matched_flight_id"] = matched.id
2015 seg["matched_flight_str"] = (
2016 f"#{matched.id} — {matched.date} "
2017 f"{matched.departure_icao} → {matched.arrival_icao}"
2018 )
2019 seg["matched_has_existing_track"] = matched.gps_track_id is not None
2020 seg["linked_pilot_entries"] = _linked_pilot_entries(
2021 matched.id, int(session["user_id"])
2022 )
2023 continue
2025 # No exact GPS-block overlap — fall back to a fuzzy date/route/
2026 # duration/landings match against this aircraft's flights that
2027 # don't yet have real GPS block data (typically logged manually
2028 # or imported from a CSV, airframe or pilot logbook — see
2029 # docs/backlog.md's GPS-vs-CSV reconciliation note). _score_gps_candidate
2030 # reuses the airframe-import review's non-time signals, with its own
2031 # wider time-of-day matching (see its docstring).
2032 # takeoff_time/landing_time here just labels *which* GPS-observed
2033 # instant this is (block-off vs block-on) — _score_gps_candidate
2034 # itself compares each against whichever of the existing row's
2035 # departure_time/takeoff_time (or landing_time/arrival_time) is set.
2036 fields = {
2037 "departure_icao": seg["departure_icao"],
2038 "arrival_icao": seg["arrival_icao"],
2039 "takeoff_time": block_off.time(),
2040 "landing_time": block_on.time(),
2041 "flight_time": seg["flight_time_rounded_h"],
2042 "landing_count": seg["landing_count"],
2043 }
2044 # Only rows with no real GPS block data yet — a flight that
2045 # already has its own track is a different real flight (e.g. two
2046 # short hops the same day on the same route), not the one this
2047 # exact-match check above already ruled out.
2048 same_day = Flight.query.filter(
2049 Flight.aircraft_id == aircraft_id,
2050 Flight.date == block_off.date(),
2051 Flight.block_off_utc.is_(None),
2052 ).all()
2053 candidates = score_gps_candidates(
2054 fields,
2055 same_day,
2056 partial(_score_gps_candidate, offset_hours=float(ac.flight_counter_offset)),
2057 _CANDIDATE_MIN_SCORE,
2058 )
2059 if candidates:
2060 seg["matched_flight_id"] = candidates[0].id
2061 seg["matched_flight_str"] = None
2062 seg["matched_has_existing_track"] = False
2063 seg["linked_pilot_entries"] = []
2064 seg["matched_ambiguous"] = True
2065 seg["matched_candidates"] = [_gps_candidate_dict(c) for c in candidates]
2066 else:
2067 seg["matched_flight_id"] = None
2068 seg["matched_flight_str"] = None
2069 seg["matched_has_existing_track"] = False
2070 seg["linked_pilot_entries"] = []
2072 tmp_dir = _gps_tmp_dir()
2073 session["gps_import"]["segments"] = [
2074 _segment_for_session(s, tmp_dir) for s in full_segs
2075 ]
2076 session.modified = True
2078 # Get OpenAIP API key for map tiles
2079 tile_setting = db.session.get(AppSetting, "openaip_api_key")
2080 openaip_key = tile_setting.value if tile_setting and tile_setting.value else None
2082 return render_template(
2083 "aircraft/gps_import_review.html",
2084 aircraft=ac,
2085 segments=full_segs,
2086 skipped_empty=state.get("skipped_empty", 0),
2087 openaip_key=openaip_key,
2088 other_aircraft=state.get("other_aircraft", False),
2089 other_ac_make_model=state.get("other_ac_make_model", ""),
2090 other_ac_reg=state.get("other_ac_reg", ""),
2091 confirmed_segments=state.get("confirmed_segments", {}),
2092 )
2095def _gps_import_create_segment(
2096 ac: Any,
2097 aircraft_id: int,
2098 seg: dict[str, Any],
2099 seg_idx: int,
2100 pilot_role: str,
2101 dep_icao: str,
2102 arr_icao: str,
2103 nature: str | None,
2104 remarks: str | None,
2105 batch: Any,
2106 file_metas: list[dict[str, Any]],
2107 linked_ids: list[int],
2108 pilot_display_name: str,
2109 other_aircraft: bool,
2110 other_ac_make_model: str,
2111 other_ac_reg: str,
2112 batch_device_id: str | None,
2113) -> tuple[Any, list[int]]:
2114 """Create Flight + optionally PilotLogbookEntry for one GPS segment.
2116 Returns (entry_or_None, updated_linked_ids).
2117 """
2118 import decimal as _dec
2119 from datetime import datetime as _dt
2121 create_pilot_entries = pilot_role in ("pic", "dual")
2123 block_off = _dt.fromisoformat(seg["block_off_utc"])
2124 block_on = _dt.fromisoformat(seg["block_on_utc"])
2125 dep_time = block_off.time().replace(tzinfo=None)
2126 arr_time = block_on.time().replace(tzinfo=None)
2127 flight_time_h = round_flight_time(
2128 seg["flight_time_raw_h"], ac.logbook_time_precision
2129 )
2131 entry = None
2132 gps_track: GpsTrack | None = None
2134 if not other_aircraft:
2135 matched_id = seg.get("matched_flight_id")
2136 if matched_id:
2137 existing = db.session.get(Flight, matched_id)
2138 if existing and existing.aircraft_id == aircraft_id:
2139 old_track_id = existing.gps_track_id
2140 existing.block_off_utc = block_off
2141 existing.block_on_utc = block_on
2142 gps_track = GpsTrack(
2143 source_filename=file_metas[0]["original_filename"]
2144 if len(file_metas) == 1
2145 else None,
2146 device_id=batch_device_id,
2147 block_off_utc=block_off,
2148 block_on_utc=block_on,
2149 departure_icao=dep_icao,
2150 arrival_icao=arr_icao,
2151 geojson=_load_segment_geojson(seg),
2152 )
2153 db.session.add(gps_track)
2154 db.session.flush()
2155 existing.gps_track_id = gps_track.id
2156 if old_track_id and old_track_id != gps_track.id:
2157 # Re-confirming an already-linked match (the review page
2158 # warns this replaces the track) — drop the superseded
2159 # row instead of leaving it as a permanent orphan.
2160 old_track = db.session.get(GpsTrack, old_track_id)
2161 if old_track is not None:
2162 db.session.delete(old_track)
2163 linked_ids.append(existing.id)
2164 # Unified model: this Flight row already covers both crew
2165 # slots, so setting gps_track_id above already applies to
2166 # whichever other pilot occupies the other slot too — no
2167 # separate per-pilot row left to update.
2168 db.session.flush()
2169 entry = existing
2170 else:
2171 matched_id = None
2173 if not matched_id:
2174 gps_track = GpsTrack(
2175 source_filename=file_metas[0]["original_filename"]
2176 if len(file_metas) == 1
2177 else None,
2178 block_off_utc=block_off,
2179 block_on_utc=block_on,
2180 departure_icao=dep_icao,
2181 arrival_icao=arr_icao,
2182 geojson=_load_segment_geojson(seg),
2183 )
2184 db.session.add(gps_track)
2185 db.session.flush()
2186 entry = Flight(
2187 aircraft_id=aircraft_id,
2188 date=block_off.date(),
2189 departure_icao=dep_icao,
2190 arrival_icao=arr_icao,
2191 departure_time=dep_time,
2192 arrival_time=arr_time,
2193 flight_time=_dec.Decimal(str(flight_time_h)),
2194 landing_count=seg.get("landing_count") or 0,
2195 nature_of_flight=nature,
2196 source="gps_import",
2197 gps_import_batch_id=batch.id,
2198 block_off_utc=block_off,
2199 block_on_utc=block_on,
2200 gps_track_id=gps_track.id,
2201 )
2202 db.session.add(entry)
2203 db.session.flush()
2205 if create_pilot_entries:
2206 from flights.routes import apply_pilot_identity
2208 if entry is None:
2209 # Other/external aircraft — standalone row (no airframe side).
2210 if gps_track is None:
2211 gps_track = GpsTrack(
2212 source_filename=file_metas[0]["original_filename"]
2213 if len(file_metas) == 1
2214 else None,
2215 device_id=batch_device_id,
2216 block_off_utc=block_off,
2217 block_on_utc=block_on,
2218 departure_icao=dep_icao,
2219 arrival_icao=arr_icao,
2220 geojson=_load_segment_geojson(seg),
2221 )
2222 db.session.add(gps_track)
2223 db.session.flush()
2224 entry = Flight(
2225 date=block_off.date(),
2226 other_aircraft_type=other_ac_make_model or None,
2227 other_aircraft_registration=other_ac_reg or None,
2228 departure_icao=dep_icao,
2229 departure_time=dep_time,
2230 arrival_icao=arr_icao,
2231 arrival_time=arr_time,
2232 source="gps_import",
2233 gps_import_batch_id=batch.id,
2234 gps_track_id=gps_track.id,
2235 )
2236 db.session.add(entry)
2237 db.session.flush()
2239 entry.flight_time = _dec.Decimal(str(flight_time_h))
2240 # GPS-derived landing count has no day/night split — treat them all
2241 # as day landings, same simplification the old standalone pilot
2242 # entry made.
2243 entry.landings_day = seg.get("landing_count") or 0
2244 if remarks:
2245 entry.notes = remarks
2246 entry_ac: Aircraft | None = None if other_aircraft else ac
2247 apply_pilot_identity(entry, entry_ac, int(session["user_id"]), pilot_role)
2249 return entry, linked_ids
2252def _gps_cleanup(state: dict[str, Any]) -> None:
2253 """Delete tmp GPS and GeoJSON files from a gps_import session state."""
2254 for meta in state.get("files", []):
2255 try:
2256 os.unlink(meta["tmp_path"])
2257 except OSError as exc:
2258 current_app.logger.debug("cleanup GPS tmp file: %s", exc)
2259 for seg in state.get("segments", []):
2260 gj_path = seg.get("geojson_path")
2261 if gj_path:
2262 try:
2263 os.unlink(gj_path)
2264 except OSError as exc:
2265 current_app.logger.debug("cleanup GPS geojson tmp: %s", exc)
2268@aircraft_bp.route(
2269 "/<aircraft_ref:aircraft_id>/gps-import/confirm-one", methods=["POST"]
2270)
2271@login_required
2272@require_role(*_PILOT_ROLES)
2273def gps_import_confirm_one(aircraft_id: int) -> ResponseReturnValue:
2274 """Confirm a single GPS segment as-is and redirect back to the review page."""
2275 ac = _get_aircraft_or_404(aircraft_id)
2276 state = session.get("gps_import")
2277 if not state or state.get("aircraft_id") != aircraft_id:
2278 flash(_("Session expired — please upload your GPS files again."), "warning")
2279 return redirect(url_for("aircraft.gps_import_upload", aircraft_id=aircraft_id))
2281 segments_data: list[dict[str, Any]] = state.get("segments", [])
2282 if not segments_data:
2283 flash(_("No segments to import."), "warning")
2284 return redirect(url_for("aircraft.gps_import_upload", aircraft_id=aircraft_id))
2286 try:
2287 seg_idx = int(request.form.get("seg_idx", ""))
2288 except (ValueError, TypeError):
2289 flash(_("Invalid segment index."), "danger")
2290 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
2292 if seg_idx < 0 or seg_idx >= len(segments_data):
2293 flash(_("Invalid segment index."), "danger")
2294 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
2296 confirmed = state.get("confirmed_segments", {})
2297 if str(seg_idx) in confirmed:
2298 flash(_("This segment has already been confirmed."), "info")
2299 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
2301 pilot_role = request.form.get("pilot_role", "none")
2303 if request.form.get("skip") == "1":
2304 confirmed[str(seg_idx)] = "skip"
2305 state["confirmed_segments"] = confirmed
2306 session["gps_import"] = state
2307 session.modified = True
2309 all_handled = len(confirmed) == len(segments_data)
2310 if all_handled:
2311 _gps_cleanup(state)
2312 session.pop("gps_import", None)
2313 imported = sum(1 for v in confirmed.values() if v != "skip")
2314 skipped_count = len(segments_data) - imported
2315 if imported > 0:
2316 flash(
2317 ngettext(
2318 "%(n)s flight imported successfully.",
2319 "%(n)s flights imported successfully.",
2320 imported,
2321 n=imported,
2322 ),
2323 "success",
2324 )
2325 flash(
2326 ngettext(
2327 "%(n)s segment skipped.",
2328 "%(n)s segments skipped.",
2329 skipped_count,
2330 n=skipped_count,
2331 ),
2332 "info",
2333 )
2334 if pilot_role not in ("pic", "dual", "none"):
2335 pilot_role = "none"
2336 if imported > 0 and pilot_role in ("pic", "dual"):
2337 return redirect(url_for("pilots.logbook"))
2338 return redirect(url_for("flights.list_flights", aircraft_id=aircraft_id))
2340 flash(_("Segment skipped."), "info")
2341 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
2343 seg = segments_data[seg_idx]
2345 if seg.get("matched_ambiguous"):
2346 # A fuzzy (non-exact) match picker was rendered — respect the
2347 # human's explicit choice instead of always taking the top-scored
2348 # candidate, including "none of these" (submitted as "").
2349 picked = request.form.get("matched_flight_id", "")
2350 seg["matched_flight_id"] = int(picked) if picked.strip().isdigit() else None
2352 other_aircraft: bool = state.get("other_aircraft", False)
2353 other_ac_make_model: str = state.get("other_ac_make_model", "")
2354 other_ac_reg: str = state.get("other_ac_reg", "")
2356 if other_aircraft and pilot_role == "none":
2357 pilot_role = "pic"
2359 _pilot_user = db.session.get(User, int(session["user_id"]))
2360 pilot_display_name = _pilot_user.display_name if _pilot_user else ""
2361 file_metas = state["files"]
2363 dep_icao = (
2364 (request.form.get("dep_icao") or seg["departure_icao"] or "")
2365 .strip()
2366 .upper()[:4]
2367 )
2368 arr_icao = (
2369 (request.form.get("arr_icao") or seg["arrival_icao"] or "").strip().upper()[:4]
2370 )
2371 if not dep_icao:
2372 dep_icao = "????"
2373 if not arr_icao:
2374 arr_icao = "????"
2376 nature = (request.form.get("nature") or "").strip()[:100] or None
2377 remarks = (request.form.get("remarks") or "").strip() or None
2379 batch_device_id: str | None = next(
2380 (m.get("device_id") for m in file_metas if m.get("device_id")), None
2381 )
2383 # Get or create the shared batch record for this session
2384 batch_id = state.get("batch_id")
2385 batch: AircraftGpsImportBatch | None = (
2386 db.session.get(AircraftGpsImportBatch, batch_id) if batch_id else None
2387 )
2388 if batch is None:
2389 formats = {m["format"] for m in file_metas}
2390 format_label = formats.pop() if len(formats) == 1 else "mixed"
2391 batch = AircraftGpsImportBatch(
2392 aircraft_id=aircraft_id,
2393 pilot_user_id=int(session["user_id"]) if pilot_role != "none" else None,
2394 source_filenames=[m["original_filename"] for m in file_metas],
2395 format_detected=format_label,
2396 segments_found=len(segments_data),
2397 linked_flight_entry_ids=[],
2398 pilot_role=pilot_role,
2399 other_aircraft_make_model=other_ac_make_model or None,
2400 other_aircraft_registration=other_ac_reg or None,
2401 )
2402 db.session.add(batch)
2403 db.session.flush()
2404 state["batch_id"] = batch.id
2406 linked_ids: list[int] = list(batch.linked_flight_entry_ids or [])
2408 entry, linked_ids = _gps_import_create_segment(
2409 ac=ac,
2410 aircraft_id=aircraft_id,
2411 seg=seg,
2412 seg_idx=seg_idx,
2413 pilot_role=pilot_role,
2414 dep_icao=dep_icao,
2415 arr_icao=arr_icao,
2416 nature=nature,
2417 remarks=remarks,
2418 batch=batch,
2419 file_metas=file_metas,
2420 linked_ids=linked_ids,
2421 pilot_display_name=pilot_display_name,
2422 other_aircraft=other_aircraft,
2423 other_ac_make_model=other_ac_make_model,
2424 other_ac_reg=other_ac_reg,
2425 batch_device_id=batch_device_id,
2426 )
2428 batch.linked_flight_entry_ids = linked_ids
2429 batch.segments_imported = (batch.segments_imported or 0) + 1
2430 db.session.commit()
2432 confirmed[str(seg_idx)] = entry.id if entry else 0
2433 state["confirmed_segments"] = confirmed
2434 session["gps_import"] = state
2435 session.modified = True
2437 all_confirmed = len(confirmed) == len(segments_data)
2439 if all_confirmed:
2440 _gps_cleanup(state)
2441 session.pop("gps_import", None)
2442 total = len(confirmed)
2443 flash(
2444 ngettext(
2445 "%(n)s flight imported successfully.",
2446 "%(n)s flights imported successfully.",
2447 total,
2448 n=total,
2449 ),
2450 "success",
2451 )
2452 if pilot_role in ("pic", "dual"):
2453 return redirect(url_for("pilots.logbook"))
2454 return redirect(url_for("flights.list_flights", aircraft_id=aircraft_id))
2456 flash(_("Flight confirmed."), "success")
2457 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
2460@aircraft_bp.route(
2461 "/<aircraft_ref:aircraft_id>/gps-import/prefill-segment/<int:seg_idx>",
2462 methods=["GET"],
2463)
2464@login_required
2465@require_role(*_PILOT_ROLES)
2466def gps_import_prefill_segment(aircraft_id: int, seg_idx: int) -> ResponseReturnValue:
2467 """Store a batch segment as gps_prefill then redirect to /flights/new."""
2468 import json as _json
2469 from datetime import datetime as _dt
2471 _get_aircraft_or_404(aircraft_id)
2472 state = session.get("gps_import")
2473 if not state or state.get("aircraft_id") != aircraft_id:
2474 flash(_("Session expired — please upload your GPS files again."), "warning")
2475 return redirect(url_for("aircraft.gps_import_upload", aircraft_id=aircraft_id))
2477 segments_data: list[dict[str, Any]] = state.get("segments", [])
2478 if seg_idx < 0 or seg_idx >= len(segments_data):
2479 flash(_("Invalid segment index."), "danger")
2480 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
2482 seg = segments_data[seg_idx]
2483 block_off = _dt.fromisoformat(seg["block_off_utc"])
2484 block_on = _dt.fromisoformat(seg["block_on_utc"])
2485 file_metas = state.get("files", [])
2486 single_filename = file_metas[0]["original_filename"] if len(file_metas) == 1 else ""
2487 geojson_data = _load_segment_geojson(seg)
2488 geojson_str = _json.dumps(geojson_data) if geojson_data else ""
2490 session["gps_prefill"] = {
2491 "filename": single_filename,
2492 "date": block_off.date().isoformat(),
2493 "departure_icao": seg.get("departure_icao") or "",
2494 "arrival_icao": seg.get("arrival_icao") or "",
2495 "departure_time": block_off.strftime("%H:%M"),
2496 "arrival_time": block_on.strftime("%H:%M"),
2497 "flight_time_h": str(seg.get("flight_time_rounded_h") or 0),
2498 "block_off_utc": block_off.isoformat(),
2499 "block_on_utc": block_on.isoformat(),
2500 "geojson": geojson_str,
2501 "landing_count": seg.get("landing_count") or 0,
2502 }
2503 session.modified = True
2505 return redirect(
2506 url_for(
2507 "flights.log_flight",
2508 aircraft_id=aircraft_id,
2509 gps_review_return=aircraft_id,
2510 gps_seg=seg_idx,
2511 )
2512 )
2515@aircraft_bp.route("/<aircraft_ref:aircraft_id>/gps-import/history", methods=["GET"])
2516@login_required
2517@require_role(*_PILOT_ROLES)
2518def gps_import_history(aircraft_id: int) -> ResponseReturnValue:
2519 ac = _get_aircraft_or_404(aircraft_id)
2520 batches = (
2521 AircraftGpsImportBatch.query.filter_by(aircraft_id=aircraft_id)
2522 .order_by(AircraftGpsImportBatch.imported_at.desc())
2523 .all()
2524 )
2525 return render_template(
2526 "aircraft/gps_import_history.html", aircraft=ac, batches=batches
2527 )
2530@aircraft_bp.route(
2531 "/<aircraft_ref:aircraft_id>/gps-import/<int:batch_id>/rollback", methods=["POST"]
2532)
2533@login_required
2534@require_role(*_OWNER_ROLES)
2535def gps_import_rollback(aircraft_id: int, batch_id: int) -> ResponseReturnValue:
2536 _get_aircraft_or_404(aircraft_id)
2537 batch = db.session.get(AircraftGpsImportBatch, batch_id)
2538 if not batch or batch.aircraft_id != aircraft_id:
2539 abort(404)
2541 # Flights created by this batch — delete them entirely. Unified model:
2542 # this already covers what used to be a separate PilotLogbookEntry
2543 # deletion pass (gps_batch_id and gps_import_batch_id were merged into
2544 # one column on the unified row).
2545 Flight.query.filter_by(gps_import_batch_id=batch.id).delete(
2546 synchronize_session="fetch"
2547 )
2549 # Flights that were pre-existing but got a GPS track linked — unlink only.
2550 linked_ids = batch.linked_flight_entry_ids or []
2551 if linked_ids:
2552 Flight.query.filter(Flight.id.in_(linked_ids)).update(
2553 {
2554 "gps_track_id": None,
2555 "block_off_utc": None,
2556 "block_on_utc": None,
2557 },
2558 synchronize_session="fetch",
2559 )
2561 db.session.delete(batch)
2562 db.session.commit()
2563 flash(
2564 _("GPS import batch rolled back and all linked flight entries removed."),
2565 "success",
2566 )
2567 return redirect(url_for("aircraft.gps_import_history", aircraft_id=aircraft_id))
2570@aircraft_bp.route(
2571 "/<aircraft_ref:aircraft_id>/flights/<int:flight_id>", methods=["GET"]
2572)
2573@login_required
2574@require_role(*_PILOT_ROLES)
2575def flight_detail(aircraft_id: int, flight_id: int) -> ResponseReturnValue:
2576 ac = _get_aircraft_or_404(aircraft_id)
2577 entry = db.session.get(Flight, flight_id)
2578 if not entry or entry.aircraft_id != aircraft_id:
2579 abort(404)
2581 # Unified model: every pilot-log field now lives on `entry` itself — no
2582 # separate PilotLogbookEntry to fetch.
2583 tile_setting = db.session.get(AppSetting, "openaip_api_key")
2584 openaip_key = tile_setting.value if tile_setting and tile_setting.value else None
2586 return render_template(
2587 "aircraft/flight_detail.html",
2588 aircraft=ac,
2589 entry=entry,
2590 pilot_entry=None,
2591 openaip_key=openaip_key,
2592 )
2595@aircraft_bp.route("/<aircraft_ref:aircraft_id>/tracks", methods=["GET"])
2596@login_required
2597@require_role(*_PILOT_ROLES)
2598def flight_tracks(aircraft_id: int) -> ResponseReturnValue:
2599 from flask import url_for as _url_for
2601 ac = _get_aircraft_or_404(aircraft_id)
2602 entries_with_tracks = (
2603 Flight.query.filter_by(aircraft_id=aircraft_id)
2604 .filter(Flight.gps_track_id.isnot(None))
2605 .order_by(Flight.date.asc())
2606 .all()
2607 )
2608 track_rows = [
2609 {
2610 "date": str(e.date),
2611 "dep": e.departure_icao,
2612 "arr": e.arrival_icao,
2613 "time_str": f"{e.flight_time} h" if e.flight_time is not None else "",
2614 "view_url": _url_for(
2615 "aircraft.flight_detail",
2616 aircraft_id=aircraft_id,
2617 flight_id=e.id,
2618 ),
2619 "geojson": e.gps_track.geojson if e.gps_track else None,
2620 }
2621 for e in entries_with_tracks
2622 ]
2624 tile_setting = db.session.get(AppSetting, "openaip_api_key")
2625 openaip_key = tile_setting.value if tile_setting and tile_setting.value else None
2627 return render_template(
2628 "aircraft/flight_tracks.html",
2629 aircraft=ac,
2630 track_rows=track_rows,
2631 openaip_key=openaip_key,
2632 )
2635@aircraft_bp.route("/<aircraft_ref:aircraft_id>/tracks/animation.gif")
2636@login_required
2637@require_role(*_PILOT_ROLES)
2638def flight_tracks_gif(aircraft_id: int) -> ResponseReturnValue:
2639 from utils import ( # pyright: ignore[reportMissingImports]
2640 generate_tracks_gif,
2641 sort_tracks_oldest_first,
2642 )
2644 ac = _get_aircraft_or_404(aircraft_id)
2645 entries = (
2646 Flight.query.filter_by(aircraft_id=aircraft_id)
2647 .filter(Flight.gps_track_id.isnot(None))
2648 .all()
2649 )
2650 track_rows = sort_tracks_oldest_first(
2651 [
2652 {
2653 "date": str(e.date),
2654 "dep": e.departure_icao or "",
2655 "arr": e.arrival_icao or "",
2656 "geojson": e.gps_track.geojson if e.gps_track else None,
2657 }
2658 for e in entries
2659 ]
2660 )
2661 tile_s = db.session.get(AppSetting, "openaip_api_key")
2662 portrait = request.args.get("orientation") == "portrait"
2663 hires = request.args.get("quality") == "hires"
2664 base_w, base_h = (480, 800) if portrait else (800, 480)
2665 mul = 2 if hires else 1
2666 canvas_w, canvas_h = base_w * mul, base_h * mul
2667 gif_bytes = generate_tracks_gif(
2668 track_rows,
2669 _openaip_key=tile_s.value if tile_s and tile_s.value else None,
2670 canvas_w=canvas_w,
2671 canvas_h=canvas_h,
2672 high_res=hires,
2673 )
2674 orient_sfx = "-portrait" if portrait else ""
2675 qual_sfx = "-hires" if hires else ""
2676 suffix = orient_sfx + qual_sfx
2677 filename = f"{ac.registration.lower().replace('-', '')}_tracks{suffix}.gif"
2678 from flask import Response # pyright: ignore[reportMissingImports]
2680 return Response(
2681 gif_bytes,
2682 mimetype="image/gif",
2683 headers={"Content-Disposition": f'attachment; filename="{filename}"'},
2684 )
2687# ── Aircraft photos ───────────────────────────────────────────────────────────
2689_PHOTO_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".heic"}
2692def _photo_folder(app: Any, tenant_slug: str, safe_reg: str) -> str:
2693 folder = app.config.get("UPLOAD_FOLDER", "/data/uploads")
2694 return os.path.join(folder, tenant_slug, safe_reg, "photos")
2697def _save_photo_file(
2698 file: Any,
2699 tenant_slug: str,
2700 safe_reg: str,
2701 sort_order: int,
2702) -> tuple[str, str]:
2703 """Save photo to canonical path; return (relpath, original_filename)."""
2705 original = secure_filename(file.filename or "photo.jpg")
2706 ext = os.path.splitext(original)[1].lower() or ".jpg"
2707 short_id = _uuid_mod.uuid4().hex[:6]
2708 fname = f"{sort_order:02d}-{short_id}{ext}"
2709 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
2710 dest_dir = os.path.join(folder, tenant_slug, safe_reg, "photos")
2711 os.makedirs(dest_dir, exist_ok=True)
2712 file.save(os.path.join(dest_dir, fname))
2713 relpath = os.path.join(tenant_slug, safe_reg, "photos", fname).replace("\\", "/")
2714 return relpath, original
2717def _trash_photo_file(filename: str) -> None:
2718 """Move photo file to _trash/ (same pattern as document deletion)."""
2719 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
2720 src = os.path.join(folder, filename)
2721 if not os.path.exists(src):
2722 return
2723 try:
2724 trash_dir = os.path.join(folder, "_trash")
2725 os.makedirs(trash_dir, exist_ok=True)
2726 base = os.path.basename(filename)
2727 dest = os.path.join(trash_dir, base)
2728 if os.path.exists(dest):
2729 stem, ext = os.path.splitext(base)
2730 dest = os.path.join(trash_dir, f"{stem}_{_uuid_mod.uuid4().hex[:6]}{ext}")
2731 os.rename(src, dest)
2732 except OSError:
2733 current_app.logger.debug("Could not trash photo: %s", filename)
2736def _renumber_photos(photos: list[Any], tenant_slug: str, safe_reg: str) -> None:
2737 """Assign sort_order 1..N, renaming files on disk to keep the numeric prefix."""
2738 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
2739 for new_order, photo in enumerate(photos, start=1):
2740 if photo.sort_order != new_order:
2741 old_full = os.path.join(folder, photo.filename)
2742 old_fname = os.path.basename(photo.filename)
2743 suffix = old_fname[
2744 2:
2745 ] # strip old "NN" prefix (e.g. "01-abc.jpg" → "-abc.jpg")
2746 new_fname = f"{new_order:02d}{suffix}"
2747 new_full = os.path.join(folder, tenant_slug, safe_reg, "photos", new_fname)
2748 if os.path.exists(old_full):
2749 try:
2750 os.rename(old_full, new_full)
2751 except OSError:
2752 current_app.logger.debug(
2753 "Could not renumber photo: %s", photo.filename
2754 )
2755 new_fname = old_fname # keep old name if rename fails
2756 new_rel = f"{tenant_slug}/{safe_reg}/photos/{new_fname}"
2757 photo.filename = new_rel
2758 photo.sort_order = new_order
2761@aircraft_bp.route("/<aircraft_ref:aircraft_id>/photos/upload", methods=["POST"])
2762@login_required
2763@require_role(*_OWNER_ROLES)
2764def upload_photo(aircraft_id: int) -> ResponseReturnValue:
2765 from documents.routes import ( # pyright: ignore[reportMissingImports]
2766 _ensure_tenant_slug,
2767 _get_tenant,
2768 )
2770 ac = _get_aircraft_or_404(aircraft_id)
2771 tenant = _get_tenant()
2773 files = request.files.getlist("photos")
2774 if not files or all(f.filename == "" for f in files):
2775 flash(_("No files selected."), "warning")
2776 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
2778 tenant_slug = _ensure_tenant_slug(tenant)
2779 safe_reg = ac.registration.replace("/", "-").replace(" ", "-").upper()
2780 next_order = (
2781 db.session.query(db.func.max(AircraftPhoto.sort_order))
2782 .filter_by(aircraft_id=ac.id)
2783 .scalar()
2784 or 0
2785 ) + 1
2787 uploaded = 0
2788 for f in files:
2789 if not f.filename:
2790 continue
2791 ext = os.path.splitext(secure_filename(f.filename))[1].lower()
2792 if ext not in _PHOTO_EXTS:
2793 flash(
2794 _(
2795 "%(name)s: unsupported format (use JPEG, PNG, WEBP or HEIC).",
2796 name=f.filename,
2797 ),
2798 "warning",
2799 )
2800 continue
2801 relpath, original = _save_photo_file(f, tenant_slug, safe_reg, next_order)
2802 photo = AircraftPhoto(
2803 aircraft_id=ac.id,
2804 filename=relpath,
2805 original_filename=original,
2806 sort_order=next_order,
2807 uploaded_by_user_id=session.get("user_id"),
2808 )
2809 db.session.add(photo)
2810 next_order += 1
2811 uploaded += 1
2813 if uploaded:
2814 db.session.commit()
2815 flash(
2816 ngettext(
2817 "%(n)s photo uploaded.", "%(n)s photos uploaded.", uploaded, n=uploaded
2818 ),
2819 "success",
2820 )
2821 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
2824@aircraft_bp.route("/<aircraft_ref:aircraft_id>/photos/<int:photo_id>/img")
2825@login_required
2826def serve_photo(aircraft_id: int, photo_id: int) -> ResponseReturnValue:
2827 from flask import send_from_directory # pyright: ignore[reportMissingImports]
2829 ac = _get_aircraft_or_404(aircraft_id)
2830 photo = db.session.get(AircraftPhoto, photo_id)
2831 if not photo or photo.aircraft_id != ac.id:
2832 abort(404)
2833 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
2834 directory = os.path.join(folder, os.path.dirname(photo.filename))
2835 fname = os.path.basename(photo.filename)
2836 # photo_id -> file is immutable for the row's lifetime (reordering only
2837 # changes sort_order; a replacement upload gets a new id) — safe to
2838 # cache aggressively so the dashboard/list prefetch hints actually pay
2839 # off instead of being revalidated away on the next navigation.
2840 response = send_from_directory(directory, fname, max_age=31536000)
2841 response.headers["Cache-Control"] = "private, max-age=31536000, immutable"
2842 return response
2845@aircraft_bp.route(
2846 "/<aircraft_ref:aircraft_id>/photos/<int:photo_id>/delete", methods=["POST"]
2847)
2848@login_required
2849@require_role(*_OWNER_ROLES)
2850def delete_photo(aircraft_id: int, photo_id: int) -> ResponseReturnValue:
2851 from documents.routes import _get_tenant # pyright: ignore[reportMissingImports]
2853 ac = _get_aircraft_or_404(aircraft_id)
2854 photo = db.session.get(AircraftPhoto, photo_id)
2855 if not photo or photo.aircraft_id != ac.id:
2856 abort(404)
2858 _trash_photo_file(photo.filename)
2859 db.session.delete(photo)
2860 db.session.flush()
2862 # Renumber remaining photos
2863 remaining = (
2864 AircraftPhoto.query.filter_by(aircraft_id=ac.id)
2865 .order_by(AircraftPhoto.sort_order)
2866 .all()
2867 )
2868 tenant = _get_tenant()
2869 tenant_slug = tenant.slug or ""
2870 safe_reg = ac.registration.replace("/", "-").replace(" ", "-").upper()
2871 _renumber_photos(remaining, tenant_slug, safe_reg)
2872 db.session.commit()
2873 flash(_("Photo deleted."), "success")
2874 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
2877@aircraft_bp.route("/<aircraft_ref:aircraft_id>/photos/reorder", methods=["POST"])
2878@login_required
2879@require_role(*_OWNER_ROLES)
2880def reorder_photos(aircraft_id: int) -> ResponseReturnValue:
2881 from documents.routes import _get_tenant # pyright: ignore[reportMissingImports]
2883 ac = _get_aircraft_or_404(aircraft_id)
2885 ordered_ids: list[int] = []
2886 try:
2887 ordered_ids = [int(i) for i in request.form.getlist("photo_order[]")]
2888 except (ValueError, TypeError):
2889 abort(400)
2891 photos_by_id = {
2892 p.id: p for p in AircraftPhoto.query.filter_by(aircraft_id=ac.id).all()
2893 }
2894 if set(ordered_ids) != set(photos_by_id):
2895 abort(400)
2897 ordered_photos = [photos_by_id[pid] for pid in ordered_ids]
2898 tenant = _get_tenant()
2899 tenant_slug = tenant.slug or ""
2900 safe_reg = ac.registration.replace("/", "-").replace(" ", "-").upper()
2901 _renumber_photos(ordered_photos, tenant_slug, safe_reg)
2902 db.session.commit()
2903 return "", 204