Coverage for app/aircraft/routes.py: 100%
1578 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 15:29 +0000
1import json
2import logging
3import math
4import os
5import uuid as _uuid_mod
6from datetime import UTC
7from typing import Any, cast
9from flask import ( # pyright: ignore[reportMissingImports]
10 Blueprint,
11 abort,
12 current_app,
13 flash,
14 redirect,
15 render_template,
16 request,
17 session,
18 url_for,
19)
20from flask.typing import ResponseReturnValue # pyright: ignore[reportMissingImports]
21from flask_babel import gettext as _ # pyright: ignore[reportMissingImports]
22from flask_babel import ngettext
23from flights.crew_removal import ( # pyright: ignore[reportMissingImports]
24 detach_from_aircraft,
25 flash_kept_in_pilot_logbooks,
26 remove_from_aircraft_log,
27)
28from models import (
29 FUEL_DENSITY,
30 GAL_TO_L,
31 Aircraft,
32 AircraftFuelTank,
33 AircraftGpsImportBatch,
34 AircraftOwner,
35 AircraftPhoto,
36 AirworthinessDocumentStatus,
37 AppSetting,
38 Component,
39 ComponentType,
40 DocType,
41 Document,
42 Expense,
43 ExpenseType,
44 Flight,
45 GpsTrack,
46 MaintenanceTrigger,
47 OperatingModel,
48 Refuel,
49 Reservation,
50 ReservationStatus,
51 Role,
52 Snag,
53 Tenant,
54 TenantProfile,
55 TenantUser,
56 User,
57 WeightBalanceConfig,
58 WeightBalanceEntry,
59 WeightBalanceStation,
60 db,
61) # pyright: ignore[reportMissingImports]
62from services.gps_archive import ( # pyright: ignore[reportMissingImports]
63 archive_gps_file,
64 pilot_archive_parts,
65)
66from services.upload_layout import ( # pyright: ignore[reportMissingImports]
67 GPS_ARCHIVE_DIRNAME,
68)
69from utils import (
70 accessible_aircraft,
71 activity,
72 compute_aircraft_statuses,
73 get_aircraft_type_engine_info,
74 login_required,
75 require_role,
76 user_can_access_aircraft,
77) # pyright: ignore[reportMissingImports]
78from werkzeug.utils import secure_filename # pyright: ignore[reportMissingImports]
80from aircraft.co_owner_form_parsing import ( # pyright: ignore[reportMissingImports]
81 parse_owners_form,
82 parse_reserve_fields,
83)
84from aircraft.gps_import import ( # pyright: ignore[reportMissingImports]
85 apply_gps_date,
86 detect_segments,
87 logged_flight_windows,
88 merge_and_sort,
89 merge_segments_logged_as_one,
90 parse_gps_file,
91 segment_times,
92)
94aircraft_bp = Blueprint("aircraft", __name__, url_prefix="/aircraft")
95log = logging.getLogger(__name__)
97_OWNER_ROLES = (Role.ADMIN, Role.OWNER)
98_PILOT_ROLES = (Role.ADMIN, Role.OWNER, Role.PILOT)
101def _tenant_id() -> int:
102 """Return the tenant ID for the currently logged-in user."""
103 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
104 if not tu:
105 abort(403)
106 return int(tu.tenant_id)
109def _get_aircraft_or_404(aircraft_id: int) -> Aircraft:
110 """Fetch an aircraft that belongs to the current tenant and is accessible to the user."""
111 ac = db.session.get(Aircraft, aircraft_id)
112 if (
113 not ac
114 or ac.tenant_id != _tenant_id()
115 or not user_can_access_aircraft(aircraft_id)
116 ):
117 abort(404)
118 return ac
121def _get_tenant_aircraft_or_404(aircraft_id: int) -> Aircraft:
122 """Like _get_aircraft_or_404 but without the operational
123 user_can_access_aircraft gate — for self-service co-owner routes
124 (my_share/my_share_statement_csv) where the AircraftOwner row checked
125 right after this call is the real authorization boundary. A co-owner may
126 have no operational role on the aircraft at all (e.g. a silent investor
127 with no UserAircraftAccess grant) and must still be able to view their
128 own share."""
129 ac = db.session.get(Aircraft, aircraft_id)
130 if not ac or ac.tenant_id != _tenant_id():
131 abort(404)
132 return ac
135def _shared_ownership_enabled(tenant_id: int, aircraft_id: int) -> bool:
136 """True when the tenant runs the shared_ownership operating model, OR
137 when this aircraft already has AircraftOwner rows (legacy data after a
138 model switch — the pages must stay reachable so an admin can view
139 accounts and clear the owner set; without rows they 404, keeping zero
140 trace on instances that never used the feature)."""
141 profile = TenantProfile.query.filter_by(tenant_id=tenant_id).first()
142 if profile and profile.operating_model == OperatingModel.SHARED_OWNERSHIP:
143 return True
144 return bool(
145 db.session.query(
146 AircraftOwner.query.filter_by(aircraft_id=aircraft_id).exists()
147 ).scalar()
148 )
151def _registration_taken(
152 registration: str, tenant_id: int, exclude_aircraft_id: int | None = None
153) -> bool:
154 """True if another aircraft in this tenant already has this registration —
155 compared the same way AircraftRefConverter (utils.py) sanitizes for its
156 URL slot ('/' and ' ' -> '-', case-insensitive), so two registrations
157 that only differ that way don't end up silently sharing a URL."""
158 needle = registration.replace("/", "-").replace(" ", "-").upper()
159 query = Aircraft.query.filter(
160 Aircraft.tenant_id == tenant_id,
161 db.func.upper(
162 db.func.replace(db.func.replace(Aircraft.registration, "/", "-"), " ", "-")
163 )
164 == needle,
165 )
166 if exclude_aircraft_id is not None:
167 query = query.filter(Aircraft.id != exclude_aircraft_id)
168 return query.first() is not None
171def _get_component_or_404(aircraft: Aircraft, component_id: int) -> Component:
172 comp = db.session.get(Component, component_id)
173 if not comp or comp.aircraft_id != aircraft.id:
174 abort(404)
175 return comp
178# ── Aircraft list ─────────────────────────────────────────────────────────────
181@aircraft_bp.route("/")
182@login_required
183def list_aircraft() -> ResponseReturnValue:
184 from models import TenantProfile
186 if not request.args.get("list"):
187 tp = TenantProfile.query.filter_by(tenant_id=_tenant_id()).first()
188 if tp and tp.planned_aircraft_count == 1:
189 aircraft_list = accessible_aircraft(_tenant_id()).all()
190 if len(aircraft_list) == 1:
191 return redirect(
192 url_for("aircraft.detail", aircraft_id=aircraft_list[0].id)
193 )
195 show_archived = request.args.get("archived") == "1"
196 all_accessible = accessible_aircraft(_tenant_id(), include_archived=True).all()
197 archived_count = sum(1 for ac in all_accessible if ac.is_archived)
198 aircraft = (
199 all_accessible
200 if show_archived
201 else [ac for ac in all_accessible if not ac.is_archived]
202 )
203 aircraft_ids = [ac.id for ac in aircraft]
204 hobbs_by_id = Aircraft.engine_hours_by_id(aircraft_ids)
205 landings_by_id = Aircraft.landings_by_id(aircraft_ids)
206 flight_hours_by_id = Aircraft.flight_hours_by_id(aircraft_ids)
207 triggers = (
208 (
209 MaintenanceTrigger.query.filter(
210 MaintenanceTrigger.aircraft_id.in_(aircraft_ids)
211 ).all()
212 )
213 if aircraft_ids
214 else []
215 )
216 aircraft_status = compute_aircraft_statuses(
217 aircraft, triggers, hobbs_by_id, landings_by_id, flight_hours_by_id
218 )
219 wb_configured_ids = {ac.id for ac in aircraft if ac.wb_config is not None}
220 cover_photos = (
221 {
222 p.aircraft_id: p
223 for p in AircraftPhoto.query.filter(
224 AircraftPhoto.aircraft_id.in_(aircraft_ids),
225 AircraftPhoto.sort_order == 1,
226 ).all()
227 }
228 if aircraft_ids
229 else {}
230 )
231 return render_template(
232 "aircraft/list.html",
233 aircraft=aircraft,
234 aircraft_status=aircraft_status,
235 wb_configured_ids=wb_configured_ids,
236 cover_photos=cover_photos,
237 show_archived=show_archived,
238 archived_count=archived_count,
239 )
242# ── Add aircraft ──────────────────────────────────────────────────────────────
245@aircraft_bp.route("/new", methods=["GET", "POST"])
246@login_required
247@require_role(*_OWNER_ROLES)
248def new_aircraft() -> ResponseReturnValue:
249 if request.method == "POST":
250 return _save_aircraft(None)
251 return render_template("aircraft/aircraft_form.html", aircraft=None)
254# ── Aircraft detail ───────────────────────────────────────────────────────────
257@aircraft_bp.route("/<aircraft_ref:aircraft_id>")
258@login_required
259def detail(aircraft_id: int) -> ResponseReturnValue:
260 from models import Flight, MaintenanceTrigger
262 ac = _get_aircraft_or_404(aircraft_id)
263 components_by_type: dict[Any, list[Any]] = {}
264 for comp in sorted(ac.components, key=lambda c: (c.type, c.position or "")):
265 components_by_type.setdefault(comp.type, []).append(comp)
266 recent_flights = (
267 Flight.query.filter_by(aircraft_id=ac.id)
268 .order_by(
269 Flight.date.desc(),
270 Flight.departure_time.desc().nullslast(),
271 Flight.id.desc(),
272 )
273 .limit(3)
274 .all()
275 )
276 current_hobbs = ac.total_engine_hours
277 current_landings = ac.total_landings
278 current_flight_hours = ac.total_flight_hours
279 triggers = MaintenanceTrigger.query.filter_by(aircraft_id=ac.id).all()
280 maintenance_summary = [
281 (
282 t,
283 t.status(
284 current_engine_hours=current_hobbs,
285 current_landings=current_landings,
286 current_flight_hours=current_flight_hours,
287 ),
288 )
289 for t in triggers
290 ]
291 recent_expenses = (
292 Expense.query.filter_by(aircraft_id=ac.id)
293 .order_by(Expense.date.desc(), Expense.id.desc())
294 .limit(3)
295 .all()
296 )
297 recent_refuels = (
298 Refuel.query.filter_by(aircraft_id=ac.id)
299 .order_by(Refuel.date.desc(), Refuel.id.desc())
300 .limit(3)
301 .all()
302 )
303 recent_documents = (
304 Document.query.filter_by(aircraft_id=ac.id, is_sensitive=False)
305 .order_by(Document.uploaded_at.desc())
306 .limit(3)
307 .all()
308 )
309 document_count = Document.query.filter_by(aircraft_id=ac.id).count()
310 from datetime import date as _cert_date
312 from documents.routes import ( # pyright: ignore[reportMissingImports]
313 active_document_for,
314 effective_coverage_until,
315 )
317 active_insurance_cert = active_document_for(ac.id, DocType.INSURANCE_CERT)
318 active_arc_cert = active_document_for(ac.id, DocType.ARC)
319 # Chain-aware "actually covered until" -- accounts for a renewal already
320 # on file (e.g. next quarter's cert uploaded a month ahead), unlike the
321 # raw insurance_expiry/arc_expiry cache fields, which only reflect the
322 # currently-active document's own expiry. Falls back to the cache field
323 # itself when there's no backing Document (e.g. dev/demo seed data) --
324 # see the matching fallback in services.notification_service.
325 insurance_covered_until = (
326 effective_coverage_until(ac.id, DocType.INSURANCE_CERT) or ac.insurance_expiry
327 )
328 arc_covered_until = effective_coverage_until(ac.id, DocType.ARC) or ac.arc_expiry
329 # Status badges use the chain-aware date too, so a renewal already on
330 # file also stops the compliance section itself claiming "expiring
331 # soon" -- not just the (separate) notification email.
332 insurance_display_status = Aircraft._expiry_status(insurance_covered_until)
333 arc_display_status = Aircraft._expiry_status(arc_covered_until)
334 upcoming_insurance_cert = (
335 Document.query.filter(
336 Document.aircraft_id == ac.id,
337 Document.doc_type == DocType.INSURANCE_CERT,
338 Document.component_id.is_(None),
339 Document.valid_from > _cert_date.today(),
340 )
341 .order_by(Document.valid_from.asc())
342 .first()
343 )
344 upcoming_arc_cert = (
345 Document.query.filter(
346 Document.aircraft_id == ac.id,
347 Document.doc_type == DocType.ARC,
348 Document.component_id.is_(None),
349 Document.valid_from > _cert_date.today(),
350 )
351 .order_by(Document.valid_from.asc())
352 .first()
353 )
354 open_snags = (
355 Snag.query.filter_by(aircraft_id=ac.id, resolved_at=None)
356 .order_by(Snag.is_grounding.desc(), Snag.reported_at.desc())
357 .all()
358 )
359 wb_cfg = ac.wb_config
360 last_wb_entry = None
361 if wb_cfg:
362 last_wb_entry = (
363 WeightBalanceEntry.query.filter_by(config_id=wb_cfg.id)
364 .order_by(WeightBalanceEntry.date.desc(), WeightBalanceEntry.id.desc())
365 .first()
366 )
367 from datetime import datetime
369 now = datetime.now(UTC)
370 upcoming_reservations = (
371 Reservation.query.filter(
372 Reservation.aircraft_id == ac.id,
373 Reservation.status.in_(
374 [ReservationStatus.CONFIRMED, ReservationStatus.PENDING]
375 ),
376 Reservation.end_dt >= now,
377 )
378 .order_by(Reservation.start_dt)
379 .limit(5)
380 .all()
381 )
382 suggest_components = session.pop(f"suggest_components_{ac.id}", None)
383 photos = (
384 AircraftPhoto.query.filter_by(aircraft_id=ac.id)
385 .order_by(AircraftPhoto.sort_order)
386 .all()
387 )
388 aw_statuses = AirworthinessDocumentStatus.query.filter_by(aircraft_id=ac.id).all()
389 aw_counts: dict[str, int] = {}
390 for _st in aw_statuses:
391 aw_counts[_st.status] = aw_counts.get(_st.status, 0) + 1
392 aw_counts["total"] = len(aw_statuses)
393 _gps_entries = (
394 Flight.query.filter_by(aircraft_id=ac.id)
395 .filter(Flight.gps_track_id.isnot(None))
396 .order_by(Flight.date.asc())
397 .all()
398 )
399 track_rows = [
400 {
401 "date": str(e.date),
402 "dep": e.departure_icao or "",
403 "arr": e.arrival_icao or "",
404 "time_str": f"{e.flight_time} h" if e.flight_time is not None else "",
405 "view_url": url_for(
406 "aircraft.flight_detail",
407 aircraft_id=aircraft_id,
408 flight_id=e.id,
409 ),
410 "geojson": e.gps_track.geojson if e.gps_track else None,
411 }
412 for e in _gps_entries
413 ]
414 _tile = db.session.get(AppSetting, "openaip_api_key")
415 openaip_key = _tile.value if _tile and _tile.value else None
416 from services.component_limits import (
417 aircraft_limit_infos, # pyright: ignore[reportMissingImports]
418 )
420 component_limit_by_id = {
421 info["component"].id: info for info in aircraft_limit_infos(ac)
422 }
423 from datetime import date as _today_date
425 shared_ownership_enabled = _shared_ownership_enabled(ac.tenant_id, ac.id)
426 is_current_user_co_owner = any(
427 o.user_id == session["user_id"] for o in list(ac.owners)
428 )
430 return render_template(
431 "aircraft/detail.html",
432 aircraft=ac,
433 components_by_type=components_by_type,
434 component_limit_by_id=component_limit_by_id,
435 today_date=_today_date.today(),
436 component_types=ComponentType,
437 recent_flights=recent_flights,
438 suggest_components=suggest_components,
439 maintenance_summary=maintenance_summary,
440 recent_expenses=recent_expenses,
441 expense_type_labels=ExpenseType.LABELS,
442 recent_refuels=recent_refuels,
443 recent_documents=recent_documents,
444 document_count=document_count,
445 active_insurance_cert=active_insurance_cert,
446 active_arc_cert=active_arc_cert,
447 upcoming_insurance_cert=upcoming_insurance_cert,
448 upcoming_arc_cert=upcoming_arc_cert,
449 insurance_covered_until=insurance_covered_until,
450 arc_covered_until=arc_covered_until,
451 insurance_display_status=insurance_display_status,
452 arc_display_status=arc_display_status,
453 open_snags=open_snags,
454 wb_config=wb_cfg,
455 last_wb_entry=last_wb_entry,
456 upcoming_reservations=upcoming_reservations,
457 ReservationStatus=ReservationStatus,
458 photos=photos,
459 aw_counts=aw_counts,
460 track_rows=track_rows,
461 openaip_key=openaip_key,
462 shared_ownership_enabled=shared_ownership_enabled,
463 owners=ac.owners if shared_ownership_enabled else [],
464 is_current_user_co_owner=is_current_user_co_owner,
465 )
468# ── Edit aircraft ─────────────────────────────────────────────────────────────
471@aircraft_bp.route("/<aircraft_ref:aircraft_id>/edit", methods=["GET", "POST"])
472@login_required
473@require_role(*_OWNER_ROLES)
474def edit_aircraft(aircraft_id: int) -> ResponseReturnValue:
475 ac = _get_aircraft_or_404(aircraft_id)
476 if request.method == "POST":
477 return _save_aircraft(ac)
478 return render_template("aircraft/aircraft_form.html", aircraft=ac)
481def _save_aircraft(ac: Aircraft | None) -> ResponseReturnValue:
482 is_new = ac is None
483 icao_type = request.form.get("aircraft_type_icao", "").strip().upper()
484 registration = request.form.get("registration", "").strip().upper()
485 make = request.form.get("make", "").strip()
486 model = request.form.get("model", "").strip()
487 year_raw = request.form.get("year", "").strip()
488 has_flight_counter = bool(request.form.get("has_flight_counter"))
489 flight_counter_offset_raw = request.form.get("flight_counter_offset", "0.3").strip()
490 fuel_flow_raw = request.form.get("fuel_flow", "").strip()
491 fuel_type = request.form.get("fuel_type", "avgas").strip()
492 if fuel_type not in ("avgas", "ul91", "mogas", "jet_a1"):
493 fuel_type = "avgas"
494 reserve_hourly_rate_raw = request.form.get("reserve_hourly_rate", "").strip()
495 oil_warning_lph_raw = request.form.get("oil_warning_lph", "").strip()
496 showcase_blurb = request.form.get("showcase_blurb", "").strip() or None
497 logbook_time_precision = request.form.get(
498 "logbook_time_precision", "tenth_hour"
499 ).strip()
500 if logbook_time_precision not in ("tenth_hour", "minute"):
501 logbook_time_precision = "tenth_hour"
503 errors = []
504 if not registration:
505 errors.append(_("Registration is required."))
506 elif _registration_taken(
507 registration, _tenant_id(), exclude_aircraft_id=ac.id if ac else None
508 ):
509 errors.append(
510 _(
511 "%(reg)s is already used by another aircraft in this hangar "
512 "(registrations that only differ by spaces or slashes still "
513 "count as the same, since they'd otherwise share a URL).",
514 reg=registration,
515 )
516 )
517 if not make:
518 errors.append(_("Manufacturer is required."))
519 if not model:
520 errors.append(_("Model is required."))
521 year = None
522 if year_raw:
523 try:
524 year = int(year_raw)
525 if not (1900 <= year <= 2100):
526 raise ValueError
527 except ValueError:
528 errors.append(_("Year must be a valid 4-digit year."))
530 flight_counter_offset = 0.3
531 if flight_counter_offset_raw:
532 try:
533 flight_counter_offset = float(flight_counter_offset_raw)
534 if flight_counter_offset < 0:
535 raise ValueError
536 except ValueError:
537 errors.append(_("Flight counter offset must be a non-negative number."))
539 fuel_flow = None
540 if fuel_flow_raw:
541 try:
542 fuel_flow = float(fuel_flow_raw)
543 if fuel_flow < 0:
544 raise ValueError
545 except ValueError:
546 errors.append(_("Fuel consumption must be a non-negative number."))
548 reserve_hourly_rate = None
549 if reserve_hourly_rate_raw:
550 try:
551 reserve_hourly_rate = float(reserve_hourly_rate_raw)
552 if reserve_hourly_rate < 0:
553 raise ValueError
554 except ValueError:
555 errors.append(_("Overhaul reserve accrual must be a non-negative number."))
557 oil_warning_lph = None
558 if oil_warning_lph_raw:
559 try:
560 oil_warning_lph = float(oil_warning_lph_raw)
561 if oil_warning_lph < 0:
562 raise ValueError
563 except ValueError:
564 errors.append(
565 _("Oil consumption warning threshold must be a non-negative number.")
566 )
568 # Fuel tanks (see AircraftFuelTank docstring). Same "replace all rows"
569 # pattern as the W&B stations form: blank-name rows are silently
570 # dropped, a non-positive/invalid capacity is an error.
571 tank_names = request.form.getlist("tank_name[]")
572 tank_capacities_raw = request.form.getlist("tank_capacity[]")
573 tanks_data: list[tuple[str, float]] = []
574 for i, tname in enumerate(tank_names):
575 tname = tname.strip()
576 if not tname:
577 continue
578 try:
579 tcap = float(tank_capacities_raw[i])
580 if tcap <= 0:
581 raise ValueError
582 except (ValueError, IndexError):
583 errors.append(_("Fuel tank capacity must be a positive number."))
584 continue
585 tanks_data.append((tname, tcap))
587 if errors:
588 for msg in errors:
589 flash(msg, "danger")
590 return render_template("aircraft/aircraft_form.html", aircraft=ac)
592 if ac is None:
593 ac = Aircraft(tenant_id=_tenant_id())
594 db.session.add(ac)
596 ac.registration = registration
597 ac.make = make
598 ac.model = model
599 ac.year = year
600 ac.has_flight_counter = has_flight_counter
601 ac.flight_counter_offset = flight_counter_offset
602 ac.fuel_flow = fuel_flow
603 ac.fuel_type = fuel_type
604 ac.logbook_time_precision = logbook_time_precision
605 ac.reserve_hourly_rate = reserve_hourly_rate
606 ac.oil_warning_lph = oil_warning_lph
607 ac.showcase_blurb = showcase_blurb
608 db.session.commit()
610 # Replace fuel tanks (same "delete all, recreate" strategy as W&B stations)
611 for t in list(ac.fuel_tanks):
612 db.session.delete(t)
613 for i, (tname, tcap) in enumerate(tanks_data):
614 db.session.add(
615 AircraftFuelTank(
616 aircraft_id=ac.id, name=tname, capacity_liters=tcap, sort_order=i
617 )
618 )
619 db.session.commit()
621 if is_new:
622 activity("aircraft.created", registration=ac.registration, aircraft_id=ac.id)
623 else:
624 activity("aircraft.updated", registration=ac.registration, aircraft_id=ac.id)
626 if is_new and icao_type:
627 engine_info = get_aircraft_type_engine_info(icao_type)
628 if engine_info:
629 ec, et = engine_info
630 if et == "Piston" and not ac.components:
631 session[f"suggest_components_{ac.id}"] = {"engine_count": ec}
633 flash(_("%(reg)s saved.", reg=ac.registration), "success")
634 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
637# ── Shared ownership (Phase 39a) ──────────────────────────────────────────────
640@aircraft_bp.route("/<aircraft_ref:aircraft_id>/owners", methods=["GET", "POST"])
641@login_required
642@require_role(*_OWNER_ROLES)
643def manage_owners(aircraft_id: int) -> ResponseReturnValue:
644 ac = _get_aircraft_or_404(aircraft_id)
645 if not _shared_ownership_enabled(ac.tenant_id, ac.id):
646 abort(404)
648 tenant_users = (
649 User.query.join(TenantUser, TenantUser.user_id == User.id)
650 .filter(TenantUser.tenant_id == ac.tenant_id, User.is_active.is_(True))
651 .order_by(User.name)
652 .all()
653 )
655 if request.method == "POST":
656 from datetime import date as _date
658 rows, billing_start, hourly_rate, errors = parse_owners_form(request.form)
659 reserve_hourly, reserve_monthly, reserve_errors = parse_reserve_fields(
660 request.form
661 )
662 errors = errors + reserve_errors
664 if errors:
665 for msg in errors:
666 flash(msg, "danger")
667 return render_template(
668 "aircraft/manage_owners.html",
669 aircraft=ac,
670 tenant_users=tenant_users,
671 )
673 submitted_user_ids = {r["user_id"] for r in rows}
674 existing_by_user = {o.user_id: o for o in list(ac.owners)}
675 is_first_save = not existing_by_user
677 for uid, existing in existing_by_user.items():
678 if uid not in submitted_user_ids:
679 db.session.delete(existing)
681 for row in rows:
682 existing = existing_by_user.get(row["user_id"])
683 if existing is not None:
684 existing.share_pct = row["share_pct"]
685 existing.buy_in_amount = row["buy_in_amount"]
686 else:
687 db.session.add(
688 AircraftOwner(
689 aircraft_id=ac.id,
690 user_id=row["user_id"],
691 share_pct=row["share_pct"],
692 buy_in_amount=row["buy_in_amount"],
693 )
694 )
696 ac.co_owner_hourly_rate = hourly_rate
697 ac.reserve_contribution_hourly = reserve_hourly
698 ac.reserve_contribution_monthly = reserve_monthly
699 if billing_start is not None:
700 ac.co_owner_billing_start = billing_start
701 elif is_first_save and rows and ac.co_owner_billing_start is None:
702 ac.co_owner_billing_start = _date.today()
703 db.session.commit()
705 flash(_("Ownership updated."), "success")
706 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
708 return render_template(
709 "aircraft/manage_owners.html",
710 aircraft=ac,
711 tenant_users=tenant_users,
712 )
715def _co_owner_billing_period(period_months_raw: str | None) -> tuple[Any, Any]:
716 from datetime import date as _date
717 from datetime import timedelta as _timedelta
719 try:
720 period_months = int(period_months_raw) if period_months_raw else 12
721 except ValueError:
722 period_months = 12
723 if period_months <= 0:
724 period_months = 12
725 end = _date.today()
726 start = end - _timedelta(days=period_months * 30)
727 return start, end
730@aircraft_bp.route("/<aircraft_ref:aircraft_id>/owners/billing")
731@login_required
732@require_role(*_OWNER_ROLES)
733def owners_billing(aircraft_id: int) -> ResponseReturnValue:
734 from decimal import Decimal
735 from itertools import groupby
737 from models import (
738 BillingAccountKind,
739 CoOwnerValuationSnapshot,
740 LedgerEntryType,
741 LogbookEntryType,
742 TenantProfile,
743 )
744 from services.billing import BillingService
745 from services.co_owner_billing import (
746 overdue_since,
747 reserve_fund_balance,
748 run_co_owner_billing_pass,
749 )
751 ac = _get_aircraft_or_404(aircraft_id)
752 if not _shared_ownership_enabled(ac.tenant_id, ac.id):
753 abort(404)
755 run_co_owner_billing_pass(ac)
756 db.session.commit()
758 profile = TenantProfile.query.filter_by(tenant_id=ac.tenant_id).first()
759 overdue_days = profile.co_owner_overdue_days if profile else 30
761 start, end = _co_owner_billing_period(request.args.get("period"))
763 from datetime import date as _date
765 from models import LedgerEntry
767 today = _date.today()
768 owners = list(ac.owners)
769 rows = []
770 for owner in owners:
771 account = BillingService.get_or_create_account(
772 ac.tenant_id, owner.user_id, BillingAccountKind.CO_OWNER, aircraft_id=ac.id
773 )
774 statement = BillingService.statement(account, start, end)
775 already_reversed_ids = {
776 row[0]
777 for row in db.session.query(LedgerEntry.reverses_id)
778 .filter(
779 LedgerEntry.account_id == account.id,
780 LedgerEntry.reverses_id.isnot(None),
781 )
782 .all()
783 }
785 hours = Decimal(0)
786 fixed_liability = Decimal(0)
787 operating_liability = Decimal(0)
788 reserve_liability = Decimal(0)
789 payments = Decimal(0)
790 for line in statement.lines:
791 entry = line.entry
792 if entry.source_type == "flight_usage":
793 operating_liability += Decimal(entry.amount)
794 flight = db.session.get(Flight, entry.source_id)
795 if flight is not None and flight.flight_time is not None:
796 hours += Decimal(flight.flight_time)
797 elif entry.source_type == "expense_share":
798 fixed_liability += Decimal(entry.amount)
799 elif entry.source_type == "reserve_contribution":
800 reserve_liability += Decimal(entry.amount)
801 elif entry.entry_type == LedgerEntryType.PAYMENT:
802 payments += -Decimal(entry.amount)
804 lines = []
805 for line in reversed(statement.lines):
806 entry = line.entry
807 can_reverse = (
808 entry.entry_type == LedgerEntryType.PAYMENT
809 and entry.reverses_id is None
810 and entry.id not in already_reversed_ids
811 )
812 lines.append({"line": line, "can_reverse": can_reverse})
814 since = overdue_since(account)
815 rows.append(
816 {
817 "owner": owner,
818 "hours": hours,
819 "fixed_liability": fixed_liability,
820 "operating_liability": operating_liability,
821 "reserve_liability": reserve_liability,
822 "payments": payments,
823 "capital_balance": -BillingService.balance(account),
824 "overdue_since": since,
825 "overdue_flagged": since is not None
826 and (today - since).days > overdue_days,
827 "lines": lines,
828 }
829 )
831 owner_user_ids = {o.user_id for o in owners}
832 unattributed_flights: list[Any] = []
833 if ac.co_owner_billing_start is not None:
834 unattributed_flights = [
835 f
836 for f in Flight.query.filter(
837 Flight.aircraft_id == ac.id,
838 Flight.entry_type == LogbookEntryType.FLIGHT,
839 Flight.date >= ac.co_owner_billing_start,
840 Flight.flight_time.isnot(None),
841 Flight.flight_time > 0,
842 ).all()
843 if f.pic_user_id not in owner_user_ids
844 ]
845 unattributed_hours = sum(
846 (Decimal(f.flight_time) for f in unattributed_flights), Decimal(0)
847 )
849 snapshots = (
850 CoOwnerValuationSnapshot.query.filter_by(aircraft_id=ac.id)
851 .order_by(
852 CoOwnerValuationSnapshot.valuation_date.desc(),
853 CoOwnerValuationSnapshot.id.desc(),
854 )
855 .all()
856 )
857 snapshot_groups = [
858 (valuation_date, list(items))
859 for valuation_date, items in groupby(snapshots, key=lambda s: s.valuation_date)
860 ]
862 return render_template(
863 "aircraft/owners_billing.html",
864 aircraft=ac,
865 rows=rows,
866 period=request.args.get("period") or "12",
867 snapshot_groups=snapshot_groups,
868 unattributed_flights=unattributed_flights,
869 unattributed_hours=unattributed_hours,
870 today=today.isoformat(),
871 reserve_fund_balance=reserve_fund_balance(ac),
872 )
875def _get_current_owner_or_404(aircraft: Aircraft, user_id: int) -> AircraftOwner:
876 owner = AircraftOwner.query.filter_by(
877 aircraft_id=aircraft.id, user_id=user_id
878 ).first()
879 if owner is None:
880 abort(404)
881 return cast(AircraftOwner, owner)
884@aircraft_bp.route(
885 "/<aircraft_ref:aircraft_id>/owners/<int:user_id>/payment", methods=["POST"]
886)
887@login_required
888@require_role(*_OWNER_ROLES)
889def owner_record_payment(aircraft_id: int, user_id: int) -> ResponseReturnValue:
890 from datetime import date as _date_cls
892 from models import BillingAccountKind, LedgerEntryType
893 from services.billing import BillingService
895 ac = _get_aircraft_or_404(aircraft_id)
896 if not _shared_ownership_enabled(ac.tenant_id, ac.id):
897 abort(404)
898 _get_current_owner_or_404(ac, user_id)
900 amount_raw = request.form.get("amount", "").strip()
901 date_raw = request.form.get("date", "").strip()
902 note = request.form.get("note", "").strip() or None
904 errors = []
905 try:
906 amount = float(amount_raw)
907 if amount <= 0:
908 errors.append(_("Payment amount must be positive."))
909 except ValueError:
910 amount = 0.0
911 errors.append(_("Invalid payment amount."))
912 try:
913 payment_date = (
914 _date_cls.fromisoformat(date_raw) if date_raw else _date_cls.today()
915 )
916 except ValueError:
917 payment_date = _date_cls.today()
918 errors.append(_("Invalid payment date."))
920 if errors:
921 for msg in errors:
922 flash(msg, "danger")
923 return redirect(url_for("aircraft.owners_billing", aircraft_id=ac.id))
925 account = BillingService.get_or_create_account(
926 ac.tenant_id, user_id, BillingAccountKind.CO_OWNER, aircraft_id=ac.id
927 )
928 recorder = db.session.get(User, session["user_id"])
929 description = str(_("Payment — %(note)s", note=note)) if note else str(_("Payment"))
930 BillingService.post(
931 account,
932 LedgerEntryType.PAYMENT,
933 -amount,
934 description,
935 payment_date,
936 source_type="payment",
937 created_by=recorder,
938 )
939 db.session.commit()
940 flash(_("Payment recorded."), "success")
941 return redirect(url_for("aircraft.owners_billing", aircraft_id=ac.id))
944@aircraft_bp.route(
945 "/<aircraft_ref:aircraft_id>/owners/<int:user_id>/entries/<int:entry_id>/reverse",
946 methods=["POST"],
947)
948@login_required
949@require_role(*_OWNER_ROLES)
950def owner_reverse_entry(
951 aircraft_id: int, user_id: int, entry_id: int
952) -> ResponseReturnValue:
953 from models import BillingAccountKind, LedgerEntry
954 from services.billing import BillingService
956 ac = _get_aircraft_or_404(aircraft_id)
957 if not _shared_ownership_enabled(ac.tenant_id, ac.id):
958 abort(404)
959 _get_current_owner_or_404(ac, user_id)
961 account = BillingService.get_or_create_account(
962 ac.tenant_id, user_id, BillingAccountKind.CO_OWNER, aircraft_id=ac.id
963 )
964 entry = db.session.get(LedgerEntry, entry_id)
965 if entry is None or entry.account_id != account.id:
966 abort(404)
968 recorder = db.session.get(User, session["user_id"])
969 try:
970 BillingService.reverse(entry, recorder, str(_("Payment reversed")))
971 db.session.commit()
972 flash(_("Entry reversed."), "success")
973 except ValueError as exc:
974 flash(str(exc), "danger")
975 return redirect(url_for("aircraft.owners_billing", aircraft_id=ac.id))
978@aircraft_bp.route("/<aircraft_ref:aircraft_id>/owners/valuation", methods=["POST"])
979@login_required
980@require_role(*_OWNER_ROLES)
981def record_valuation_snapshot(aircraft_id: int) -> ResponseReturnValue:
982 from datetime import date as _date_cls
984 from models import BillingAccountKind, CoOwnerValuationSnapshot
985 from services.billing import BillingService
986 from services.co_owner_billing import run_co_owner_billing_pass
988 ac = _get_aircraft_or_404(aircraft_id)
989 if not _shared_ownership_enabled(ac.tenant_id, ac.id):
990 abort(404)
992 date_raw = request.form.get("date", "").strip()
993 note = request.form.get("note", "").strip() or None
994 try:
995 valuation_date = (
996 _date_cls.fromisoformat(date_raw) if date_raw else _date_cls.today()
997 )
998 except ValueError:
999 valuation_date = _date_cls.today()
1001 run_co_owner_billing_pass(ac)
1002 db.session.commit()
1004 recorder = db.session.get(User, session["user_id"])
1005 for owner in list(ac.owners):
1006 account = BillingService.get_or_create_account(
1007 ac.tenant_id, owner.user_id, BillingAccountKind.CO_OWNER, aircraft_id=ac.id
1008 )
1009 balance = BillingService.balance(account, as_of=valuation_date)
1010 db.session.add(
1011 CoOwnerValuationSnapshot(
1012 aircraft_id=ac.id,
1013 user_id=owner.user_id,
1014 valuation_date=valuation_date,
1015 share_pct=owner.share_pct,
1016 capital_balance=-balance,
1017 note=note,
1018 created_by_id=recorder.id if recorder else None,
1019 )
1020 )
1021 db.session.commit()
1022 flash(_("Valuation snapshot recorded."), "success")
1023 return redirect(url_for("aircraft.owners_billing", aircraft_id=ac.id))
1026@aircraft_bp.route("/<aircraft_ref:aircraft_id>/owners/<int:user_id>/account")
1027@login_required
1028@require_role(*_OWNER_ROLES)
1029def owner_account(aircraft_id: int, user_id: int) -> ResponseReturnValue:
1030 from models import BillingAccountKind
1031 from services.billing import BillingService
1033 ac = _get_aircraft_or_404(aircraft_id)
1034 if not _shared_ownership_enabled(ac.tenant_id, ac.id):
1035 abort(404)
1036 owner = _get_current_owner_or_404(ac, user_id)
1038 account = BillingService.get_or_create_account(
1039 ac.tenant_id, user_id, BillingAccountKind.CO_OWNER, aircraft_id=ac.id
1040 )
1041 db.session.commit()
1042 start, end = _co_owner_billing_period(request.args.get("period"))
1043 statement = BillingService.statement(account, start, end)
1045 return render_template(
1046 "aircraft/owner_account.html",
1047 aircraft=ac,
1048 owner=owner,
1049 account=account,
1050 statement=statement,
1051 balance=BillingService.balance(account),
1052 is_owner_view=True,
1053 csv_url=url_for(
1054 "aircraft.owner_statement_csv", aircraft_id=ac.id, user_id=user_id
1055 ),
1056 )
1059@aircraft_bp.route(
1060 "/<aircraft_ref:aircraft_id>/owners/<int:user_id>/account/statement.csv"
1061)
1062@login_required
1063@require_role(*_OWNER_ROLES)
1064def owner_statement_csv(aircraft_id: int, user_id: int) -> ResponseReturnValue:
1065 from flask import Response
1066 from models import BillingAccountKind
1067 from services.billing import BillingService
1069 ac = _get_aircraft_or_404(aircraft_id)
1070 if not _shared_ownership_enabled(ac.tenant_id, ac.id):
1071 abort(404)
1072 _get_current_owner_or_404(ac, user_id)
1074 account = BillingService.get_or_create_account(
1075 ac.tenant_id, user_id, BillingAccountKind.CO_OWNER, aircraft_id=ac.id
1076 )
1077 db.session.commit()
1078 start, end = _co_owner_billing_period(request.args.get("period"))
1079 statement = BillingService.statement(account, start, end)
1080 exporter = db.session.get(User, session["user_id"])
1081 csv_text = BillingService.statement_csv(statement, exported_by=exporter)
1082 filename = (
1083 f"co_owner_statement_{ac.registration}_{user_id}_"
1084 f"{start.isoformat()}_{end.isoformat()}.csv"
1085 )
1086 return Response(
1087 csv_text,
1088 mimetype="text/csv",
1089 headers={"Content-Disposition": f"attachment; filename={filename}"},
1090 )
1093@aircraft_bp.route("/<aircraft_ref:aircraft_id>/my-share")
1094@login_required
1095def my_share(aircraft_id: int) -> ResponseReturnValue:
1096 from models import BillingAccountKind
1097 from services.billing import BillingService
1099 ac = _get_tenant_aircraft_or_404(aircraft_id)
1100 uid = session["user_id"]
1101 owner = AircraftOwner.query.filter_by(aircraft_id=ac.id, user_id=uid).first()
1102 if owner is None:
1103 abort(404)
1105 account = BillingService.get_or_create_account(
1106 ac.tenant_id, uid, BillingAccountKind.CO_OWNER, aircraft_id=ac.id
1107 )
1108 db.session.commit()
1109 start, end = _co_owner_billing_period(request.args.get("period"))
1110 statement = BillingService.statement(account, start, end)
1112 return render_template(
1113 "aircraft/owner_account.html",
1114 aircraft=ac,
1115 owner=owner,
1116 account=account,
1117 statement=statement,
1118 balance=BillingService.balance(account),
1119 is_owner_view=False,
1120 csv_url=url_for("aircraft.my_share_statement_csv", aircraft_id=ac.id),
1121 )
1124@aircraft_bp.route("/<aircraft_ref:aircraft_id>/my-share/statement.csv")
1125@login_required
1126def my_share_statement_csv(aircraft_id: int) -> ResponseReturnValue:
1127 from flask import Response
1128 from models import BillingAccountKind
1129 from services.billing import BillingService
1131 ac = _get_tenant_aircraft_or_404(aircraft_id)
1132 uid = session["user_id"]
1133 owner = AircraftOwner.query.filter_by(aircraft_id=ac.id, user_id=uid).first()
1134 if owner is None:
1135 abort(404)
1137 account = BillingService.get_or_create_account(
1138 ac.tenant_id, uid, BillingAccountKind.CO_OWNER, aircraft_id=ac.id
1139 )
1140 db.session.commit()
1141 start, end = _co_owner_billing_period(request.args.get("period"))
1142 statement = BillingService.statement(account, start, end)
1143 exporter = db.session.get(User, uid)
1144 csv_text = BillingService.statement_csv(statement, exported_by=exporter)
1145 filename = f"my_share_statement_{ac.registration}_{start.isoformat()}_{end.isoformat()}.csv"
1146 return Response(
1147 csv_text,
1148 mimetype="text/csv",
1149 headers={"Content-Disposition": f"attachment; filename={filename}"},
1150 )
1153# ── Delete aircraft ───────────────────────────────────────────────────────────
1156@aircraft_bp.route("/<aircraft_ref:aircraft_id>/archive", methods=["POST"])
1157@login_required
1158@require_role(*_OWNER_ROLES)
1159def archive_aircraft(aircraft_id: int) -> ResponseReturnValue:
1160 from datetime import datetime
1162 ac = _get_aircraft_or_404(aircraft_id)
1163 if not ac.is_archived:
1164 ac.archived_at = datetime.now(UTC)
1165 db.session.commit()
1166 activity("aircraft.archived", registration=ac.registration, aircraft_id=ac.id)
1167 flash(
1168 _(
1169 "%(reg)s has been archived — its history stays available, but it no "
1170 "longer appears in the active fleet.",
1171 reg=ac.registration,
1172 ),
1173 "success",
1174 )
1175 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
1178@aircraft_bp.route("/<aircraft_ref:aircraft_id>/unarchive", methods=["POST"])
1179@login_required
1180@require_role(*_OWNER_ROLES)
1181def unarchive_aircraft(aircraft_id: int) -> ResponseReturnValue:
1182 ac = _get_aircraft_or_404(aircraft_id)
1183 if ac.is_archived:
1184 ac.archived_at = None
1185 db.session.commit()
1186 activity("aircraft.unarchived", registration=ac.registration, aircraft_id=ac.id)
1187 flash(_("%(reg)s is back in the active fleet.", reg=ac.registration), "success")
1188 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
1191@aircraft_bp.route("/<aircraft_ref:aircraft_id>/delete", methods=["POST"])
1192@login_required
1193@require_role(*_OWNER_ROLES)
1194def delete_aircraft(aircraft_id: int) -> ResponseReturnValue:
1195 ac = _get_aircraft_or_404(aircraft_id)
1196 reg = ac.registration
1197 activity("aircraft.deleted", registration=reg, aircraft_id=aircraft_id)
1198 # Pilot hours outlive the aircraft: every flight linked to a pilot account
1199 # (the owner's own included) is detached and kept as an "other aircraft"
1200 # flight; only flights nobody has in their logbook are deleted with it.
1201 kept = 0
1202 for fe in Flight.query.filter_by(aircraft_id=ac.id).all():
1203 if fe.pic_user_id is not None or fe.second_crew_user_id is not None:
1204 detach_from_aircraft(fe)
1205 kept += 1
1206 db.session.flush()
1207 # The detached rows must not be swept up by Aircraft.flights'
1208 # delete-orphan cascade if that collection was already loaded.
1209 db.session.expire(ac, ["flights"])
1210 db.session.delete(ac)
1211 db.session.commit()
1212 flash(_("%(reg)s and all its components have been deleted.", reg=reg), "success")
1213 flash_kept_in_pilot_logbooks(kept)
1214 return redirect(url_for("aircraft.list_aircraft"))
1217# ── Quick-add components from ICAO type suggestion ────────────────────────────
1220@aircraft_bp.route("/<aircraft_ref:aircraft_id>/quick-add-components", methods=["POST"])
1221@login_required
1222@require_role(*_OWNER_ROLES)
1223def quick_add_components(aircraft_id: int) -> ResponseReturnValue:
1224 ac = _get_aircraft_or_404(aircraft_id)
1225 try:
1226 engine_count = max(1, min(int(request.form.get("engine_count", "1")), 4))
1227 except ValueError:
1228 engine_count = 1
1229 for i in range(engine_count):
1230 position = str(i + 1) if engine_count > 1 else None
1231 db.session.add(
1232 Component(
1233 aircraft_id=ac.id,
1234 type=ComponentType.ENGINE,
1235 position=position,
1236 make="",
1237 model="",
1238 )
1239 )
1240 db.session.add(
1241 Component(
1242 aircraft_id=ac.id,
1243 type=ComponentType.PROPELLER,
1244 position=position,
1245 make="",
1246 model="",
1247 )
1248 )
1249 db.session.commit()
1250 activity(
1251 "component.quick_added",
1252 aircraft_id=aircraft_id,
1253 engine_count=engine_count,
1254 )
1255 flash(
1256 ngettext(
1257 "Engine and propeller added — fill in the details when ready.",
1258 "%(n)s engines and propellers added — fill in the details when ready.",
1259 engine_count,
1260 n=engine_count,
1261 ),
1262 "success",
1263 )
1264 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
1267# ── Add component ─────────────────────────────────────────────────────────────
1270@aircraft_bp.route(
1271 "/<aircraft_ref:aircraft_id>/components/new", methods=["GET", "POST"]
1272)
1273@login_required
1274@require_role(*_OWNER_ROLES)
1275def new_component(aircraft_id: int) -> ResponseReturnValue:
1276 ac = _get_aircraft_or_404(aircraft_id)
1277 if request.method == "POST":
1278 return _save_component(ac, None)
1279 return render_template(
1280 "aircraft/component_form.html",
1281 aircraft=ac,
1282 component=None,
1283 component_types=ComponentType,
1284 )
1287# ── Edit component ────────────────────────────────────────────────────────────
1290@aircraft_bp.route(
1291 "/<aircraft_ref:aircraft_id>/components/<int:component_id>/edit",
1292 methods=["GET", "POST"],
1293)
1294@login_required
1295@require_role(*_OWNER_ROLES)
1296def edit_component(aircraft_id: int, component_id: int) -> ResponseReturnValue:
1297 ac = _get_aircraft_or_404(aircraft_id)
1298 comp = _get_component_or_404(ac, component_id)
1299 if request.method == "POST":
1300 return _save_component(ac, comp)
1301 return render_template(
1302 "aircraft/component_form.html",
1303 aircraft=ac,
1304 component=comp,
1305 component_types=ComponentType,
1306 )
1309def _save_component(ac: Aircraft, comp: Component | None) -> ResponseReturnValue:
1310 from datetime import date as _date
1312 type_ = request.form.get("type", "").strip()
1313 position = request.form.get("position", "").strip() or None
1314 make = request.form.get("make", "").strip()
1315 model = request.form.get("model", "").strip()
1316 serial = request.form.get("serial_number", "").strip() or None
1317 time_raw = request.form.get("time_at_install", "").strip()
1318 installed_raw = request.form.get("installed_at", "").strip()
1319 removed_raw = request.form.get("removed_at", "").strip()
1320 tbo_raw = request.form.get("tbo_hours", "").strip()
1321 life_limit_raw = request.form.get("life_limit_date", "").strip()
1322 overhauled_at_raw = request.form.get("overhauled_at_hours", "").strip()
1323 overhauled_on_raw = request.form.get("overhauled_on", "").strip()
1325 errors = []
1326 if not type_:
1327 errors.append(_("Component type is required."))
1328 if not make:
1329 errors.append(_("Manufacturer is required."))
1330 if not model:
1331 errors.append(_("Model is required."))
1333 time_at_install = None
1334 if time_raw:
1335 try:
1336 time_at_install = float(time_raw)
1337 if time_at_install < 0:
1338 raise ValueError
1339 except ValueError:
1340 errors.append(_("Time at install must be a positive number."))
1342 tbo_hours = None
1343 if tbo_raw:
1344 try:
1345 tbo_hours = float(tbo_raw)
1346 if tbo_hours <= 0:
1347 raise ValueError
1348 except ValueError:
1349 errors.append(_("TBO must be a positive number of hours."))
1351 overhauled_at_hours = None
1352 if overhauled_at_raw:
1353 try:
1354 overhauled_at_hours = float(overhauled_at_raw)
1355 if overhauled_at_hours < 0:
1356 raise ValueError
1357 except ValueError:
1358 errors.append(_("Last overhaul hours must be a non-negative number."))
1360 def _parse_date(raw: str, label: str) -> Any:
1361 if not raw:
1362 return None
1363 try:
1364 return _date.fromisoformat(raw)
1365 except ValueError:
1366 errors.append(
1367 _("%(label)s must be a valid date (YYYY-MM-DD).", label=label)
1368 )
1369 return None
1371 installed_at = _parse_date(installed_raw, "Install date")
1372 removed_at = _parse_date(removed_raw, "Removal date")
1373 life_limit_date = _parse_date(life_limit_raw, "Calendar life limit")
1374 overhauled_on = _parse_date(overhauled_on_raw, "Last overhaul date")
1376 if errors:
1377 for msg in errors:
1378 flash(msg, "danger")
1379 return render_template(
1380 "aircraft/component_form.html",
1381 aircraft=ac,
1382 component=comp,
1383 component_types=ComponentType,
1384 )
1386 _comp_is_new = comp is None
1387 if comp is None:
1388 comp = Component(aircraft_id=ac.id)
1389 db.session.add(comp)
1391 comp.type = type_
1392 comp.position = position
1393 comp.make = make
1394 comp.model = model
1395 comp.serial_number = serial
1396 comp.time_at_install = time_at_install
1397 comp.installed_at = installed_at
1398 comp.removed_at = removed_at
1399 comp.tbo_hours = tbo_hours
1400 comp.life_limit_date = life_limit_date
1401 comp.overhauled_at_hours = overhauled_at_hours
1402 comp.overhauled_on = overhauled_on
1403 db.session.commit()
1405 if _comp_is_new:
1406 activity(
1407 "component.added", type=comp.type, component_id=comp.id, aircraft_id=ac.id
1408 )
1409 else:
1410 activity(
1411 "component.updated", type=comp.type, component_id=comp.id, aircraft_id=ac.id
1412 )
1414 flash(_("%(make)s %(model)s saved.", make=comp.make, model=comp.model), "success")
1415 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
1418# ── Delete component ──────────────────────────────────────────────────────────
1421@aircraft_bp.route(
1422 "/<aircraft_ref:aircraft_id>/components/<int:component_id>/delete", methods=["POST"]
1423)
1424@login_required
1425@require_role(*_OWNER_ROLES)
1426def delete_component(aircraft_id: int, component_id: int) -> ResponseReturnValue:
1427 ac = _get_aircraft_or_404(aircraft_id)
1428 comp = _get_component_or_404(ac, component_id)
1429 label = f"{comp.make} {comp.model}"
1430 activity(
1431 "component.deleted",
1432 type=comp.type,
1433 component_id=component_id,
1434 aircraft_id=aircraft_id,
1435 )
1436 db.session.delete(comp)
1437 db.session.commit()
1438 flash(_("%(label)s removed.", label=label), "success")
1439 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
1442# ── Mass & Balance: helpers ───────────────────────────────────────────────────
1445def _point_in_polygon(cg: float, weight: float, points: Any) -> bool:
1446 """Ray-casting point-in-polygon test. points: list of [arm, weight] pairs.
1448 Returns False (outside) for malformed point data rather than raising —
1449 envelope_points is a DB-stored JSON field with no enforced schema, so a
1450 corrupted or malformed entry should degrade to "envelope check
1451 unavailable" (the conservative, fail-safe answer for a W&B calculation),
1452 not crash the entry page with an unhandled 500. Non-finite coordinates
1453 (e.g. "Infinity"/"NaN" — accepted by float() but not a valid arm/weight
1454 value) are rejected the same way.
1455 """
1456 try:
1457 coords = [(float(p[0]), float(p[1])) for p in points]
1458 except (IndexError, ValueError, TypeError):
1459 return False
1460 if any(not (math.isfinite(x) and math.isfinite(y)) for x, y in coords):
1461 return False
1462 n = len(coords)
1463 inside = False
1464 j = n - 1
1465 for i in range(n):
1466 xi, yi = coords[i]
1467 xj, yj = coords[j]
1468 if ((yi > weight) != (yj > weight)) and (
1469 cg < (xj - xi) * (weight - yi) / (yj - yi) + xi
1470 ):
1471 inside = not inside
1472 j = i
1473 return inside
1476# ── Mass & Balance: config ────────────────────────────────────────────────────
1479@aircraft_bp.route("/<aircraft_ref:aircraft_id>/wb/config", methods=["GET", "POST"])
1480@login_required
1481@require_role(*_OWNER_ROLES)
1482def wb_config(aircraft_id: int) -> ResponseReturnValue:
1483 ac = _get_aircraft_or_404(aircraft_id)
1484 cfg: WeightBalanceConfig | None = ac.wb_config # type: ignore[assignment]
1486 if request.method == "POST":
1487 errors = []
1489 def _f(name: str) -> float | None:
1490 try:
1491 v = float(request.form.get(name, "").strip())
1492 if v < 0:
1493 raise ValueError
1494 return v
1495 except ValueError:
1496 errors.append(_("%(field)s must be a positive number.", field=name))
1497 return None
1499 empty_weight = _f("empty_weight")
1500 empty_cg_arm = _f("empty_cg_arm")
1501 max_takeoff_weight = _f("max_takeoff_weight")
1502 forward_cg_limit = _f("forward_cg_limit")
1503 aft_cg_limit = _f("aft_cg_limit")
1504 datum_note = request.form.get("datum_note", "").strip() or None
1506 fuel_unit = request.form.get("fuel_unit", "L").strip()
1507 if fuel_unit not in ("L", "gal"):
1508 fuel_unit = "L"
1510 # Stations: label[], arm[], station_limit[] (capacity for fuel, max_weight for non-fuel), is_fuel[]
1511 labels = request.form.getlist("station_label[]")
1512 arms = request.form.getlist("station_arm[]")
1513 limits = request.form.getlist("station_limit[]")
1514 is_fuels = request.form.getlist(
1515 "station_is_fuel[]"
1516 ) # index values of checked boxes
1518 if not labels or all(lbl.strip() == "" for lbl in labels):
1519 errors.append(_("At least one loading station is required."))
1521 if errors:
1522 for msg in errors:
1523 flash(msg, "danger")
1524 return render_template("aircraft/wb_config.html", aircraft=ac, config=cfg)
1526 if cfg is None:
1527 cfg = WeightBalanceConfig(aircraft_id=ac.id)
1528 db.session.add(cfg)
1530 # Optional envelope polygon: env_arm[], env_weight[]
1531 env_arms = request.form.getlist("env_arm[]")
1532 env_weights = request.form.getlist("env_weight[]")
1533 envelope_points = []
1534 for arm_s, w_s in zip(env_arms, env_weights):
1535 try:
1536 a = float(arm_s.strip())
1537 w = float(w_s.strip())
1538 if a >= 0 and w >= 0:
1539 envelope_points.append([round(a, 4), round(w, 2)])
1540 except (ValueError, AttributeError):
1541 continue
1543 cfg.empty_weight = empty_weight
1544 cfg.empty_cg_arm = empty_cg_arm
1545 cfg.max_takeoff_weight = max_takeoff_weight
1546 cfg.forward_cg_limit = forward_cg_limit
1547 cfg.aft_cg_limit = aft_cg_limit
1548 cfg.fuel_unit = fuel_unit
1549 cfg.datum_note = datum_note
1550 cfg.envelope_points = envelope_points if len(envelope_points) >= 3 else None
1552 # Replace stations
1553 for s in list(cfg.stations):
1554 db.session.delete(s)
1555 db.session.flush()
1557 for i, label in enumerate(labels):
1558 label = label.strip()
1559 if not label:
1560 continue
1561 try:
1562 arm = float(arms[i])
1563 except (ValueError, IndexError):
1564 continue
1565 limit_val = None
1566 try:
1567 lim_raw = limits[i].strip()
1568 if lim_raw:
1569 limit_val = float(lim_raw)
1570 except (ValueError, IndexError):
1571 limit_val = None
1572 is_fuel = str(i) in is_fuels
1573 db.session.add(
1574 WeightBalanceStation(
1575 config_id=cfg.id,
1576 label=label,
1577 arm=arm,
1578 max_weight=None if is_fuel else limit_val,
1579 capacity=limit_val if is_fuel else None,
1580 is_fuel=is_fuel,
1581 position=i,
1582 )
1583 )
1585 db.session.commit()
1586 flash(_("W&B configuration saved."), "success")
1587 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
1589 return render_template("aircraft/wb_config.html", aircraft=ac, config=cfg)
1592# ── Mass & Balance: entry list ────────────────────────────────────────────────
1595@aircraft_bp.route("/<aircraft_ref:aircraft_id>/wb/")
1596@login_required
1597def wb_list(aircraft_id: int) -> ResponseReturnValue:
1598 ac = _get_aircraft_or_404(aircraft_id)
1599 if not ac.wb_config:
1600 flash(_("Configure W&B envelope first."), "warning")
1601 return redirect(url_for("aircraft.wb_config", aircraft_id=ac.id))
1602 entries = (
1603 WeightBalanceEntry.query.filter_by(config_id=ac.wb_config.id)
1604 .order_by(WeightBalanceEntry.date.desc(), WeightBalanceEntry.id.desc())
1605 .all()
1606 )
1607 return render_template(
1608 "aircraft/wb_list.html", aircraft=ac, config=ac.wb_config, entries=entries
1609 )
1612# ── Mass & Balance: new / edit entry ─────────────────────────────────────────
1615@aircraft_bp.route("/<aircraft_ref:aircraft_id>/wb/new", methods=["GET", "POST"])
1616@aircraft_bp.route(
1617 "/<aircraft_ref:aircraft_id>/wb/<int:entry_id>/edit", methods=["GET", "POST"]
1618)
1619@login_required
1620@require_role(*_PILOT_ROLES)
1621def wb_entry(aircraft_id: int, entry_id: int | None = None) -> ResponseReturnValue:
1622 ac = _get_aircraft_or_404(aircraft_id)
1623 if not ac.wb_config:
1624 flash(_("Configure W&B envelope first."), "warning")
1625 return redirect(url_for("aircraft.wb_config", aircraft_id=ac.id))
1626 cfg = ac.wb_config
1628 entry = None
1629 if entry_id is not None:
1630 entry = db.session.get(WeightBalanceEntry, entry_id)
1631 if not entry or entry.config_id != cfg.id:
1632 abort(404)
1634 if request.method == "POST":
1635 from datetime import date as _date
1637 errors = []
1638 date_raw = request.form.get("date", "").strip()
1639 label = request.form.get("label", "").strip() or None
1640 try:
1641 entry_date = _date.fromisoformat(date_raw)
1642 except ValueError:
1643 errors.append(_("A valid date is required."))
1644 entry_date = None
1646 # Per-station values: fuel stations store volume (L/gal), non-fuel store kg
1647 station_weights = {}
1648 for st in cfg.stations:
1649 if st.is_fuel:
1650 raw = request.form.get(f"volume_{st.id}", "").strip()
1651 try:
1652 vol = float(raw) if raw else 0.0
1653 if vol < 0:
1654 raise ValueError
1655 if st.capacity is not None and vol > float(st.capacity):
1656 errors.append(
1657 _(
1658 "Volume for %(station)s exceeds tank capacity.",
1659 station=st.label,
1660 )
1661 )
1662 station_weights[str(st.id)] = vol
1663 except ValueError:
1664 errors.append(
1665 _(
1666 "Volume for %(station)s must be a non-negative number.",
1667 station=st.label,
1668 )
1669 )
1670 else:
1671 raw = request.form.get(f"weight_{st.id}", "").strip()
1672 try:
1673 w = float(raw) if raw else 0.0
1674 if w < 0:
1675 raise ValueError
1676 station_weights[str(st.id)] = w
1677 except ValueError:
1678 errors.append(
1679 _(
1680 "Weight for %(station)s must be a non-negative number.",
1681 station=st.label,
1682 )
1683 )
1685 # CG computation — fuel stations: convert volume → kg
1686 empty_w = float(cfg.empty_weight)
1687 empty_arm = float(cfg.empty_cg_arm)
1688 total_moment = empty_w * empty_arm
1689 total_weight = empty_w
1690 fuel_density = FUEL_DENSITY.get(ac.fuel_type, 0.72)
1691 gal_factor = GAL_TO_L if cfg.fuel_unit == "gal" else 1.0
1692 for st in cfg.stations:
1693 val = station_weights.get(str(st.id), 0.0)
1694 w_kg = val * fuel_density * gal_factor if st.is_fuel else val
1695 total_weight += w_kg
1696 total_moment += w_kg * float(st.arm)
1697 loaded_cg = total_moment / total_weight if total_weight else 0.0
1699 if cfg.envelope_points and len(cfg.envelope_points) >= 3:
1700 in_env = _point_in_polygon(loaded_cg, total_weight, cfg.envelope_points)
1701 else:
1702 mtow = float(cfg.max_takeoff_weight)
1703 fwd = float(cfg.forward_cg_limit)
1704 aft = float(cfg.aft_cg_limit)
1705 in_env = total_weight <= mtow and fwd <= loaded_cg <= aft
1707 if errors:
1708 for msg in errors:
1709 flash(msg, "danger")
1710 return render_template(
1711 "aircraft/wb_entry.html",
1712 aircraft=ac,
1713 config=cfg,
1714 entry=entry,
1715 fuel_density=FUEL_DENSITY,
1716 )
1718 if entry is None:
1719 entry = WeightBalanceEntry(config_id=cfg.id)
1720 db.session.add(entry)
1722 entry.date = entry_date
1723 entry.label = label
1724 entry.total_weight = round(total_weight, 2)
1725 entry.loaded_cg = round(loaded_cg, 2)
1726 entry.is_in_envelope = in_env
1727 entry.station_weights = station_weights
1728 db.session.commit()
1729 flash(_("W&B calculation saved."), "success")
1730 return redirect(url_for("aircraft.wb_list", aircraft_id=ac.id))
1732 return render_template(
1733 "aircraft/wb_entry.html",
1734 aircraft=ac,
1735 config=cfg,
1736 entry=entry,
1737 fuel_density=FUEL_DENSITY,
1738 )
1741# ── Mass & Balance: delete entry ──────────────────────────────────────────────
1744@aircraft_bp.route(
1745 "/<aircraft_ref:aircraft_id>/wb/<int:entry_id>/delete", methods=["POST"]
1746)
1747@login_required
1748@require_role(*_PILOT_ROLES)
1749def wb_entry_delete(aircraft_id: int, entry_id: int) -> ResponseReturnValue:
1750 ac = _get_aircraft_or_404(aircraft_id)
1751 if not ac.wb_config:
1752 abort(404)
1753 entry = db.session.get(WeightBalanceEntry, entry_id)
1754 if not entry or entry.config_id != ac.wb_config.id:
1755 abort(404)
1756 db.session.delete(entry)
1757 db.session.commit()
1758 flash(_("W&B calculation deleted."), "success")
1759 return redirect(url_for("aircraft.wb_list", aircraft_id=ac.id))
1762# ── Phase 30: GPS Log Import ──────────────────────────────────────────────────
1764_GPS_ALLOWED_EXTS = {".gpx", ".kml", ".csv"}
1765_GPS_MAX_BYTES = 20 * 1024 * 1024 # 20 MB per file
1768def _gps_archive_dir(ac: Aircraft, other_aircraft: bool) -> str:
1769 """Where this upload's original files are kept (services.gps_archive):
1770 with the aircraft's own files, or — for an upload recorded on an
1771 aircraft not managed here — in the uploading pilot's own archive."""
1772 from documents.routes import ( # pyright: ignore[reportMissingImports]
1773 _ensure_tenant_slug,
1774 _safe_join,
1775 )
1777 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
1778 if other_aircraft:
1779 return _safe_join(folder, *pilot_archive_parts(int(session["user_id"])))
1780 tenant = db.session.get(Tenant, ac.tenant_id)
1781 assert tenant is not None, "every aircraft belongs to a tenant"
1782 slug = _ensure_tenant_slug(tenant)
1783 db.session.commit() # a freshly generated slug must outlive this request
1784 safe_reg = ac.registration.replace("/", "-").replace(" ", "-").upper()
1785 return _safe_join(folder, slug, safe_reg, GPS_ARCHIVE_DIRNAME)
1788def _gps_tmp_dir() -> str:
1789 """Return (and create if needed) the tmp directory for GPS uploads."""
1790 upload_folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
1791 d = os.path.join(upload_folder, "gps_import_tmp")
1792 os.makedirs(d, exist_ok=True)
1793 return d
1796def _segment_to_dict(seg: Any, idx: int) -> dict[str, Any]:
1797 """Serialise a FlightSegment for template rendering (includes track_geojson)."""
1798 return {
1799 "idx": idx,
1800 "block_off_utc": seg.block_off_utc.isoformat(),
1801 "block_on_utc": seg.block_on_utc.isoformat(),
1802 "takeoff_utc": seg.takeoff_utc.isoformat() if seg.takeoff_utc else None,
1803 "landing_utc": seg.landing_utc.isoformat() if seg.landing_utc else None,
1804 "departure_icao": seg.departure_icao or "",
1805 "arrival_icao": seg.arrival_icao or "",
1806 "flight_time_raw_h": seg.flight_time_raw_h,
1807 "flight_time_rounded_h": seg.flight_time_rounded_h,
1808 "landing_count": seg.landing_count,
1809 "is_ground_only": seg.is_ground_only,
1810 "airborne_raw_h": seg.airborne_raw_h,
1811 "track_geojson": seg.track_geojson,
1812 }
1815def _linked_pilot_entries(flight_id: int, exclude_user_id: int) -> list[dict[str, Any]]:
1816 """Return metadata about the *other* crew member (if any) on this Flight
1817 row, for the GPS-track-replacement notice shown before an upload
1818 overwrites the shared airframe/GPS data.
1820 Unified model note: there's only one gps_track_id now, shared by both
1821 crew slots — the old per-pilot "preserve their own track if they already
1822 have one" policy doesn't apply any more (there's nothing to preserve
1823 separately); this is now purely informational — surfacing that another
1824 named pilot is also linked to this flight and will see the replaced
1825 track too.
1826 """
1827 fe = db.session.get(Flight, flight_id)
1828 assert fe is not None, "caller passes a Flight id it just queried"
1829 other_user_id = None
1830 if fe.pic_user_id and fe.pic_user_id != exclude_user_id:
1831 other_user_id = fe.pic_user_id
1832 elif fe.second_crew_user_id and fe.second_crew_user_id != exclude_user_id:
1833 other_user_id = fe.second_crew_user_id
1834 if other_user_id is None:
1835 return []
1836 user = db.session.get(User, other_user_id)
1837 return [
1838 {
1839 "user_id": other_user_id,
1840 "display_name": user.display_name if user else f"user #{other_user_id}",
1841 "has_existing_track": fe.gps_track_id is not None,
1842 }
1843 ]
1846def _gps_candidate_str(fe: Any) -> str:
1847 """One-line summary of an existing Flight row for a GPS-import match
1848 picker — includes whatever the row already has recorded (times,
1849 duration) so a human can tell it apart from a similar-looking flight
1850 without needing to open it first."""
1851 s = f"#{fe.id} — {fe.date} {fe.departure_icao} → {fe.arrival_icao}"
1852 if fe.departure_time and fe.arrival_time:
1853 s += (
1854 f" ({fe.departure_time.strftime('%H:%M')}"
1855 f"–{fe.arrival_time.strftime('%H:%M')})"
1856 )
1857 if fe.total_flight_time is not None:
1858 s += f", {fe.total_flight_time:.1f} h"
1859 return s
1862def _gps_candidate_dict(fe: Any) -> dict[str, Any]:
1863 """Serialise one match candidate for the review template/session."""
1864 return {
1865 "id": fe.id,
1866 "str": _gps_candidate_str(fe),
1867 "aircraft_id": fe.aircraft_id,
1868 "aircraft_reg": fe.aircraft.registration if fe.aircraft else "?",
1869 "has_existing_track": fe.gps_track_id is not None,
1870 }
1873def _load_segment_geojson(seg: dict[str, Any]) -> Any:
1874 """Read the GeoJSON dict back from the tmp file written by _segment_for_session."""
1875 path = seg.get("geojson_path")
1876 if not path or not os.path.exists(path):
1877 return None
1878 with open(path, encoding="utf-8") as fh:
1879 return json.load(fh)
1882# The per-file and per-segment lists of a GPS import grow with the number
1883# of files/flights in one upload — a season of Garmin data logs easily
1884# makes 20+ segments, past Flask's ~4 KB cookie-session limit (browsers
1885# then silently drop the cookie and the next step reports the session as
1886# expired). They live in a server-side JSON file instead; the signed
1887# session only keeps its path.
1888_GPS_STATE_SPILL_KEYS = ("files", "segments")
1891def _gps_state_load() -> dict[str, Any] | None:
1892 """The current GPS import state, with its spilled lists merged back in.
1893 None when there's no import in progress, or its state file is gone."""
1894 state = session.get("gps_import")
1895 if not state or "spill_path" not in state:
1896 return state
1897 try:
1898 with open(state["spill_path"], encoding="utf-8") as fh:
1899 spilled = json.load(fh)
1900 except (OSError, ValueError):
1901 return None
1902 return {**state, **spilled}
1905def _gps_state_save(state: dict[str, Any]) -> None:
1906 spilled = {k: state[k] for k in _GPS_STATE_SPILL_KEYS if k in state}
1907 slim = {k: v for k, v in state.items() if k not in _GPS_STATE_SPILL_KEYS}
1908 path = slim.get("spill_path") or os.path.join(
1909 _gps_tmp_dir(), f"state_{_uuid_mod.uuid4().hex}.json"
1910 )
1911 with open(path, "w", encoding="utf-8") as fh:
1912 json.dump(spilled, fh)
1913 slim["spill_path"] = path
1914 session["gps_import"] = slim
1915 session.modified = True
1918def _segment_for_session(seg_dict: dict[str, Any], tmp_dir: str) -> dict[str, Any]:
1919 """Return a copy of seg_dict safe for cookie-session storage.
1921 track_geojson can be hundreds of KB — too large for Flask's 4 KB cookie
1922 limit. We spill it to a tmp file and store the path instead.
1923 """
1924 s = {k: v for k, v in seg_dict.items() if k != "track_geojson"}
1925 geojson = seg_dict.get("track_geojson")
1926 if geojson is not None:
1927 fname = f"seg_{seg_dict['idx']}_{_uuid_mod.uuid4().hex}.geojson"
1928 path = os.path.join(tmp_dir, fname)
1929 with open(path, "w", encoding="utf-8") as fh:
1930 json.dump(geojson, fh)
1931 s["geojson_path"] = path
1932 return s
1935@aircraft_bp.route("/<aircraft_ref:aircraft_id>/gps-import", methods=["GET", "POST"])
1936@login_required
1937@require_role(*_PILOT_ROLES)
1938def gps_import_upload(aircraft_id: int) -> ResponseReturnValue:
1939 ac = _get_aircraft_or_404(aircraft_id)
1941 if request.method == "GET":
1942 return render_template("aircraft/gps_import_upload.html", aircraft=ac)
1944 files = request.files.getlist("gps_files")
1945 if not files or all(f.filename == "" for f in files):
1946 flash(_("Please select at least one GPS log file."), "warning")
1947 return render_template("aircraft/gps_import_upload.html", aircraft=ac)
1949 tmp_dir = _gps_tmp_dir()
1950 archive_dir = _gps_archive_dir(ac, request.form.get("other_aircraft") == "1")
1951 parsed_meta: list[dict[str, Any]] = []
1952 errors: list[str] = []
1953 skipped_empty = 0
1954 formats: list[str] = []
1956 for f in files:
1957 if not f.filename:
1958 continue
1959 ext = os.path.splitext(f.filename.lower())[1]
1960 if ext not in _GPS_ALLOWED_EXTS:
1961 errors.append(
1962 _(
1963 "%(fn)s: unsupported file type (use .gpx, .kml, or .csv).",
1964 fn=f.filename,
1965 )
1966 )
1967 continue
1969 data = f.read(_GPS_MAX_BYTES + 1)
1970 if len(data) > _GPS_MAX_BYTES:
1971 errors.append(_("%(fn)s: file too large (20 MB limit).", fn=f.filename))
1972 continue
1974 # Kept whatever the parser makes of it — see services.gps_archive.
1975 archive_gps_file(data, f.filename, archive_dir)
1977 try:
1978 parsed = parse_gps_file(data, f.filename)
1979 except ValueError as exc:
1980 errors.append(_("%(fn)s: %(err)s", fn=f.filename, err=str(exc)))
1981 continue
1983 if parsed.classification == "empty":
1984 skipped_empty += 1
1985 continue
1987 # Save raw bytes to tmp
1988 uid = _uuid_mod.uuid4().hex
1989 safe_name = f"{uid}_{secure_filename(f.filename)}"
1990 tmp_path = os.path.join(tmp_dir, safe_name)
1991 with open(tmp_path, "wb") as fh:
1992 fh.write(data)
1994 parsed_meta.append(
1995 {
1996 "tmp_path": tmp_path,
1997 "original_filename": f.filename,
1998 "format": parsed.format,
1999 "classification": parsed.classification,
2000 "trkpt_count": len(parsed.trackpoints),
2001 "hint_dep": parsed.hint_departure_icao,
2002 "hint_arr": parsed.hint_arrival_icao,
2003 "device_id": getattr(parsed, "device_id", None),
2004 }
2005 )
2006 formats.append(parsed.format)
2008 if errors:
2009 for e in errors:
2010 flash(e, "danger")
2011 if skipped_empty:
2012 flash(
2013 ngettext(
2014 "one file skipped — no movement detected.",
2015 "%(n)s files skipped — no movement detected.",
2016 skipped_empty,
2017 n=skipped_empty,
2018 ),
2019 "info",
2020 )
2022 if not parsed_meta:
2023 flash(_("No valid GPS files to import."), "warning")
2024 return render_template("aircraft/gps_import_upload.html", aircraft=ac)
2026 other_aircraft = request.form.get("other_aircraft") == "1"
2027 other_ac_make_model = request.form.get("other_ac_make_model", "").strip()
2028 other_ac_reg = request.form.get("other_ac_reg", "").strip().upper()
2030 _gps_state_save(
2031 {
2032 "user_id": session["user_id"],
2033 "aircraft_id": aircraft_id,
2034 "files": parsed_meta,
2035 "skipped_empty": skipped_empty,
2036 "other_aircraft": other_aircraft,
2037 "other_ac_make_model": other_ac_make_model,
2038 "other_ac_reg": other_ac_reg,
2039 }
2040 )
2041 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
2044@aircraft_bp.route("/<aircraft_ref:aircraft_id>/gps-import/review", methods=["GET"])
2045@login_required
2046@require_role(*_PILOT_ROLES)
2047def gps_import_review(aircraft_id: int) -> ResponseReturnValue:
2048 ac = _get_aircraft_or_404(aircraft_id)
2049 state = _gps_state_load()
2050 if not state or state.get("aircraft_id") != aircraft_id:
2051 flash(_("Session expired — please upload your GPS files again."), "warning")
2052 return redirect(url_for("aircraft.gps_import_upload", aircraft_id=aircraft_id))
2054 file_metas = state["files"]
2056 # Re-parse each tmp file and build combined trackpoint list
2057 from aircraft.gps_import import (
2058 ParsedGpsFile, # pyright: ignore[reportMissingImports]
2059 )
2061 all_parsed: list[ParsedGpsFile] = []
2062 for meta in file_metas:
2063 try:
2064 with open(meta["tmp_path"], "rb") as fh:
2065 data = fh.read()
2066 parsed = parse_gps_file(data, meta["original_filename"])
2067 parsed.hint_departure_icao = meta.get("hint_dep")
2068 parsed.hint_arrival_icao = meta.get("hint_arr")
2069 all_parsed.append(parsed)
2070 except (OSError, ValueError):
2071 flash(
2072 _(
2073 "Could not read %(fn)s — please upload again.",
2074 fn=meta["original_filename"],
2075 ),
2076 "warning",
2077 )
2078 return redirect(
2079 url_for("aircraft.gps_import_upload", aircraft_id=aircraft_id)
2080 )
2082 merged = merge_and_sort(all_parsed)
2084 # Collect ICAO hints from all files
2085 hint_dep = next(
2086 (p.hint_departure_icao for p in all_parsed if p.hint_departure_icao), None
2087 )
2088 hint_arr = next(
2089 (p.hint_arrival_icao for p in all_parsed if p.hint_arrival_icao), None
2090 )
2092 segments = detect_segments(
2093 merged,
2094 aircraft_precision=ac.logbook_time_precision,
2095 hint_dep=hint_dep,
2096 hint_arr=hint_arr,
2097 )
2098 seg_dates = {seg.block_off_utc.date() for seg in segments}
2099 segments = merge_segments_logged_as_one(
2100 segments,
2101 merged,
2102 logged_flight_windows(
2103 Flight.query.filter(
2104 Flight.aircraft_id == aircraft_id, Flight.date.in_(seg_dates)
2105 ).all()
2106 ),
2107 ac.logbook_time_precision,
2108 )
2110 # Build full dicts (with track_geojson) for template rendering.
2111 # Spill GeoJSON to tmp files for the session — track_geojson can be
2112 # hundreds of KB and silently overflows Flask's 4 KB cookie session.
2113 full_segs = [_segment_to_dict(seg, i) for i, seg in enumerate(segments)]
2115 # Duplicate detection: find existing Flight records that overlap each segment.
2116 from datetime import datetime as _dt
2117 from datetime import timedelta as _td
2118 from functools import partial
2120 from flights.airframe_import import _CANDIDATE_MIN_SCORE
2122 from aircraft.gps_import import (
2123 _score_gps_candidate,
2124 near_date_range,
2125 rank_near_date_candidates,
2126 score_gps_candidates,
2127 )
2129 _BLOCK_TOLERANCE = _td(minutes=15)
2130 scorer = partial(_score_gps_candidate, offset_hours=float(ac.flight_counter_offset))
2131 unmatched: list[tuple[dict[str, Any], dict[str, Any], Any]] = []
2133 for seg in full_segs:
2134 block_off = _dt.fromisoformat(seg["block_off_utc"])
2135 block_on = _dt.fromisoformat(seg["block_on_utc"])
2136 matched = Flight.query.filter(
2137 Flight.aircraft_id == aircraft_id,
2138 Flight.block_off_utc.isnot(None),
2139 Flight.block_on_utc.isnot(None),
2140 Flight.block_off_utc < block_on + _BLOCK_TOLERANCE,
2141 Flight.block_on_utc > block_off - _BLOCK_TOLERANCE,
2142 ).first()
2143 seg["matched_ambiguous"] = False
2144 seg["matched_candidates"] = []
2145 if matched:
2146 seg["matched_flight_id"] = matched.id
2147 seg["matched_flight_str"] = (
2148 f"#{matched.id} — {matched.date} "
2149 f"{matched.departure_icao} → {matched.arrival_icao}"
2150 )
2151 seg["matched_has_existing_track"] = matched.gps_track_id is not None
2152 seg["linked_pilot_entries"] = _linked_pilot_entries(
2153 matched.id, int(session["user_id"])
2154 )
2155 continue
2157 # No exact GPS-block overlap — fall back to a fuzzy date/route/
2158 # duration/landings match against this aircraft's flights that
2159 # don't yet have real GPS block data (typically logged manually
2160 # or imported from a CSV, airframe or pilot logbook — see
2161 # docs/backlog.md's GPS-vs-CSV reconciliation note). _score_gps_candidate
2162 # reuses the airframe-import review's non-time signals, with its own
2163 # wider time-of-day matching (see its docstring).
2164 # takeoff_time/landing_time here just labels *which* GPS-observed
2165 # instant this is (block-off vs block-on) — _score_gps_candidate
2166 # itself compares each against whichever of the existing row's
2167 # departure_time/takeoff_time (or landing_time/arrival_time) is set.
2168 fields = {
2169 "departure_icao": seg["departure_icao"],
2170 "arrival_icao": seg["arrival_icao"],
2171 "takeoff_time": block_off.time(),
2172 "landing_time": block_on.time(),
2173 "flight_time": seg["flight_time_rounded_h"],
2174 "landing_count": seg["landing_count"],
2175 }
2176 # Only rows with no real GPS block data yet — a flight that
2177 # already has its own track is a different real flight (e.g. two
2178 # short hops the same day on the same route), not the one this
2179 # exact-match check above already ruled out.
2180 same_day = Flight.query.filter(
2181 Flight.aircraft_id == aircraft_id,
2182 Flight.date == block_off.date(),
2183 Flight.block_off_utc.is_(None),
2184 ).all()
2185 candidates = score_gps_candidates(
2186 fields, same_day, scorer, _CANDIDATE_MIN_SCORE
2187 )
2188 seg["matched_near_date"] = False
2189 if candidates:
2190 seg["matched_flight_id"] = candidates[0].id
2191 seg["matched_flight_str"] = None
2192 seg["matched_has_existing_track"] = False
2193 seg["linked_pilot_entries"] = []
2194 seg["matched_ambiguous"] = True
2195 seg["matched_candidates"] = [_gps_candidate_dict(c) for c in candidates]
2196 else:
2197 seg["matched_flight_id"] = None
2198 seg["matched_flight_str"] = None
2199 seg["matched_has_existing_track"] = False
2200 seg["linked_pilot_entries"] = []
2201 if not seg["is_ground_only"]:
2202 unmatched.append((seg, fields, block_off.date()))
2204 # Nothing on the GPS date: a flight on the same route logged a day or
2205 # two off may carry a date typo — offered for an explicit choice, never
2206 # matched on its own, and not one another segment already matches.
2207 claimed = {s["matched_flight_id"] for s in full_segs if s["matched_flight_id"]}
2208 for seg, fields, gps_date in unmatched:
2209 first_day, last_day = near_date_range(gps_date)
2210 near = rank_near_date_candidates(
2211 fields,
2212 gps_date,
2213 Flight.query.filter(
2214 Flight.aircraft_id == aircraft_id,
2215 Flight.date.between(first_day, last_day),
2216 Flight.block_off_utc.is_(None),
2217 Flight.id.notin_(claimed),
2218 ).all(),
2219 scorer,
2220 _CANDIDATE_MIN_SCORE,
2221 )
2222 if near:
2223 seg["matched_flight_id"] = near[0].id
2224 seg["matched_ambiguous"] = True
2225 seg["matched_near_date"] = True
2226 seg["matched_candidates"] = [_gps_candidate_dict(c) for c in near]
2228 tmp_dir = _gps_tmp_dir()
2229 state["segments"] = [_segment_for_session(s, tmp_dir) for s in full_segs]
2230 _gps_state_save(state)
2232 # Get OpenAIP API key for map tiles
2233 tile_setting = db.session.get(AppSetting, "openaip_api_key")
2234 openaip_key = tile_setting.value if tile_setting and tile_setting.value else None
2236 return render_template(
2237 "aircraft/gps_import_review.html",
2238 aircraft=ac,
2239 segments=full_segs,
2240 skipped_empty=state.get("skipped_empty", 0),
2241 openaip_key=openaip_key,
2242 other_aircraft=state.get("other_aircraft", False),
2243 other_ac_make_model=state.get("other_ac_make_model", ""),
2244 other_ac_reg=state.get("other_ac_reg", ""),
2245 confirmed_segments=state.get("confirmed_segments", {}),
2246 )
2249def _gps_import_create_segment(
2250 ac: Any,
2251 aircraft_id: int,
2252 seg: dict[str, Any],
2253 seg_idx: int,
2254 pilot_role: str,
2255 dep_icao: str,
2256 arr_icao: str,
2257 nature: str | None,
2258 remarks: str | None,
2259 batch: Any,
2260 file_metas: list[dict[str, Any]],
2261 linked_ids: list[int],
2262 pilot_display_name: str,
2263 other_aircraft: bool,
2264 other_ac_make_model: str,
2265 other_ac_reg: str,
2266 batch_device_id: str | None,
2267) -> tuple[Any, list[int]]:
2268 """Create Flight + optionally PilotLogbookEntry for one GPS segment.
2270 Returns (entry_or_None, updated_linked_ids).
2271 """
2272 from datetime import datetime as _dt
2274 create_pilot_entries = pilot_role in ("pic", "dual")
2276 block_off = _dt.fromisoformat(seg["block_off_utc"])
2277 block_on = _dt.fromisoformat(seg["block_on_utc"])
2278 dep_time = block_off.time().replace(tzinfo=None)
2279 arr_time = block_on.time().replace(tzinfo=None)
2280 times = segment_times(seg, ac.logbook_time_precision)
2282 entry = None
2283 gps_track: GpsTrack | None = None
2285 if not other_aircraft:
2286 matched_id = seg.get("matched_flight_id")
2287 if matched_id:
2288 existing = db.session.get(Flight, matched_id)
2289 if existing and existing.aircraft_id == aircraft_id:
2290 old_track_id = existing.gps_track_id
2291 existing.block_off_utc = block_off
2292 existing.block_on_utc = block_on
2293 apply_gps_date(existing, seg, block_off)
2294 # Block and airborne times are facts about the aircraft,
2295 # whoever flew it (e.g. a previous owner's flights, confirmed
2296 # with no pilot role) — fill whichever the logbook left
2297 # empty, never replacing what it recorded.
2298 if existing.engine_time is None:
2299 existing.engine_time = times.block_time
2300 if existing.flight_time is None:
2301 existing.flight_time = times.airborne_time
2302 gps_track = GpsTrack(
2303 source_filename=file_metas[0]["original_filename"]
2304 if len(file_metas) == 1
2305 else None,
2306 device_id=batch_device_id,
2307 block_off_utc=block_off,
2308 block_on_utc=block_on,
2309 departure_icao=dep_icao,
2310 arrival_icao=arr_icao,
2311 geojson=_load_segment_geojson(seg),
2312 )
2313 db.session.add(gps_track)
2314 db.session.flush()
2315 existing.gps_track_id = gps_track.id
2316 if old_track_id and old_track_id != gps_track.id:
2317 # Re-confirming an already-linked match (the review page
2318 # warns this replaces the track) — drop the superseded
2319 # row instead of leaving it as a permanent orphan.
2320 old_track = db.session.get(GpsTrack, old_track_id)
2321 if old_track is not None:
2322 db.session.delete(old_track)
2323 linked_ids.append(existing.id)
2324 # Unified model: this Flight row already covers both crew
2325 # slots, so setting gps_track_id above already applies to
2326 # whichever other pilot occupies the other slot too — no
2327 # separate per-pilot row left to update.
2328 db.session.flush()
2329 entry = existing
2330 else:
2331 matched_id = None
2333 if not matched_id:
2334 gps_track = GpsTrack(
2335 source_filename=file_metas[0]["original_filename"]
2336 if len(file_metas) == 1
2337 else None,
2338 block_off_utc=block_off,
2339 block_on_utc=block_on,
2340 departure_icao=dep_icao,
2341 arrival_icao=arr_icao,
2342 geojson=_load_segment_geojson(seg),
2343 )
2344 db.session.add(gps_track)
2345 db.session.flush()
2346 entry = Flight(
2347 aircraft_id=aircraft_id,
2348 date=block_off.date(),
2349 departure_icao=dep_icao,
2350 arrival_icao=arr_icao,
2351 departure_time=dep_time,
2352 arrival_time=arr_time,
2353 takeoff_time=times.takeoff_time,
2354 landing_time=times.landing_time,
2355 flight_time=times.airborne_time,
2356 engine_time=times.block_time,
2357 landing_count=seg.get("landing_count") or 0,
2358 nature_of_flight=nature,
2359 source="gps_import",
2360 gps_import_batch_id=batch.id,
2361 block_off_utc=block_off,
2362 block_on_utc=block_on,
2363 gps_track_id=gps_track.id,
2364 )
2365 db.session.add(entry)
2366 db.session.flush()
2368 if create_pilot_entries:
2369 from flights.routes import apply_pilot_identity
2371 if entry is None:
2372 # Other/external aircraft — standalone row (no airframe side).
2373 if gps_track is None:
2374 gps_track = GpsTrack(
2375 source_filename=file_metas[0]["original_filename"]
2376 if len(file_metas) == 1
2377 else None,
2378 device_id=batch_device_id,
2379 block_off_utc=block_off,
2380 block_on_utc=block_on,
2381 departure_icao=dep_icao,
2382 arrival_icao=arr_icao,
2383 geojson=_load_segment_geojson(seg),
2384 )
2385 db.session.add(gps_track)
2386 db.session.flush()
2387 entry = Flight(
2388 date=block_off.date(),
2389 other_aircraft_type=other_ac_make_model or None,
2390 other_aircraft_registration=other_ac_reg or None,
2391 departure_icao=dep_icao,
2392 departure_time=dep_time,
2393 arrival_icao=arr_icao,
2394 arrival_time=arr_time,
2395 source="gps_import",
2396 gps_import_batch_id=batch.id,
2397 gps_track_id=gps_track.id,
2398 )
2399 db.session.add(entry)
2400 db.session.flush()
2402 # The pilot's figures come from the block time (see
2403 # flights.routes.pilot_time) — recorded as engine_time unless the
2404 # flight already has one (e.g. from its engine counters). A matched
2405 # flight's own airborne flight_time is never replaced by it.
2406 if entry.engine_time is None:
2407 entry.engine_time = times.block_time
2408 if entry.flight_time is None:
2409 entry.flight_time = times.airborne_time
2410 # GPS-derived landing count has no day/night split — treat them all
2411 # as day landings, same simplification the old standalone pilot
2412 # entry made.
2413 entry.landings_day = seg.get("landing_count") or 0
2414 if remarks:
2415 entry.notes = remarks
2416 entry_ac: Aircraft | None = None if other_aircraft else ac
2417 apply_pilot_identity(entry, entry_ac, int(session["user_id"]), pilot_role)
2419 return entry, linked_ids
2422def _gps_cleanup(state: dict[str, Any]) -> None:
2423 """Delete tmp GPS and GeoJSON files (and the spilled state file) of a
2424 gps_import session state."""
2425 spill_path = state.get("spill_path")
2426 if spill_path:
2427 try:
2428 os.unlink(spill_path)
2429 except OSError as exc:
2430 current_app.logger.debug("cleanup GPS state file: %s", exc)
2431 for meta in state.get("files", []):
2432 try:
2433 os.unlink(meta["tmp_path"])
2434 except OSError as exc:
2435 current_app.logger.debug("cleanup GPS tmp file: %s", exc)
2436 for seg in state.get("segments", []):
2437 gj_path = seg.get("geojson_path")
2438 if gj_path:
2439 try:
2440 os.unlink(gj_path)
2441 except OSError as exc:
2442 current_app.logger.debug("cleanup GPS geojson tmp: %s", exc)
2445@aircraft_bp.route(
2446 "/<aircraft_ref:aircraft_id>/gps-import/confirm-one", methods=["POST"]
2447)
2448@login_required
2449@require_role(*_PILOT_ROLES)
2450def gps_import_confirm_one(aircraft_id: int) -> ResponseReturnValue:
2451 """Confirm a single GPS segment as-is and redirect back to the review page."""
2452 ac = _get_aircraft_or_404(aircraft_id)
2453 state = _gps_state_load()
2454 if not state or state.get("aircraft_id") != aircraft_id:
2455 flash(_("Session expired — please upload your GPS files again."), "warning")
2456 return redirect(url_for("aircraft.gps_import_upload", aircraft_id=aircraft_id))
2458 segments_data: list[dict[str, Any]] = state.get("segments", [])
2459 if not segments_data:
2460 flash(_("No segments to import."), "warning")
2461 return redirect(url_for("aircraft.gps_import_upload", aircraft_id=aircraft_id))
2463 try:
2464 seg_idx = int(request.form.get("seg_idx", ""))
2465 except (ValueError, TypeError):
2466 flash(_("Invalid segment index."), "danger")
2467 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
2469 if seg_idx < 0 or seg_idx >= len(segments_data):
2470 flash(_("Invalid segment index."), "danger")
2471 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
2473 confirmed = state.get("confirmed_segments", {})
2474 if str(seg_idx) in confirmed:
2475 flash(_("This segment has already been confirmed."), "info")
2476 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
2478 pilot_role = request.form.get("pilot_role", "none")
2480 if request.form.get("skip") == "1":
2481 confirmed[str(seg_idx)] = "skip"
2482 state["confirmed_segments"] = confirmed
2483 _gps_state_save(state)
2485 all_handled = len(confirmed) == len(segments_data)
2486 if all_handled:
2487 _gps_cleanup(state)
2488 session.pop("gps_import", None)
2489 imported = sum(1 for v in confirmed.values() if v != "skip")
2490 skipped_count = len(segments_data) - imported
2491 if imported > 0:
2492 flash(
2493 ngettext(
2494 "one flight imported successfully.",
2495 "%(n)s flights imported successfully.",
2496 imported,
2497 n=imported,
2498 ),
2499 "success",
2500 )
2501 flash(
2502 ngettext(
2503 "one segment skipped.",
2504 "%(n)s segments skipped.",
2505 skipped_count,
2506 n=skipped_count,
2507 ),
2508 "info",
2509 )
2510 if pilot_role not in ("pic", "dual", "none"):
2511 pilot_role = "none"
2512 if imported > 0 and pilot_role in ("pic", "dual"):
2513 return redirect(url_for("pilots.logbook"))
2514 return redirect(url_for("flights.list_flights", aircraft_id=aircraft_id))
2516 flash(_("Segment skipped."), "info")
2517 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
2519 seg = segments_data[seg_idx]
2521 if seg.get("matched_ambiguous"):
2522 # A fuzzy (non-exact) match picker was rendered — respect the
2523 # human's explicit choice instead of always taking the top-scored
2524 # candidate, including "none of these" (submitted as "").
2525 picked = request.form.get("matched_flight_id", "")
2526 seg["matched_flight_id"] = int(picked) if picked.strip().isdigit() else None
2528 other_aircraft: bool = state.get("other_aircraft", False)
2529 other_ac_make_model: str = state.get("other_ac_make_model", "")
2530 other_ac_reg: str = state.get("other_ac_reg", "")
2532 if other_aircraft and pilot_role == "none":
2533 pilot_role = "pic"
2535 _pilot_user = db.session.get(User, int(session["user_id"]))
2536 pilot_display_name = _pilot_user.display_name if _pilot_user else ""
2537 file_metas = state["files"]
2539 dep_icao = (
2540 (request.form.get("dep_icao") or seg["departure_icao"] or "")
2541 .strip()
2542 .upper()[:4]
2543 )
2544 arr_icao = (
2545 (request.form.get("arr_icao") or seg["arrival_icao"] or "").strip().upper()[:4]
2546 )
2547 if not dep_icao:
2548 dep_icao = "????"
2549 if not arr_icao:
2550 arr_icao = "????"
2552 nature = (request.form.get("nature") or "").strip()[:100] or None
2553 remarks = (request.form.get("remarks") or "").strip() or None
2555 batch_device_id: str | None = next(
2556 (m.get("device_id") for m in file_metas if m.get("device_id")), None
2557 )
2559 # Get or create the shared batch record for this session
2560 batch_id = state.get("batch_id")
2561 batch: AircraftGpsImportBatch | None = (
2562 db.session.get(AircraftGpsImportBatch, batch_id) if batch_id else None
2563 )
2564 if batch is None:
2565 formats = {m["format"] for m in file_metas}
2566 format_label = formats.pop() if len(formats) == 1 else "mixed"
2567 batch = AircraftGpsImportBatch(
2568 aircraft_id=aircraft_id,
2569 pilot_user_id=int(session["user_id"]) if pilot_role != "none" else None,
2570 source_filenames=[m["original_filename"] for m in file_metas],
2571 format_detected=format_label,
2572 segments_found=len(segments_data),
2573 linked_flight_entry_ids=[],
2574 pilot_role=pilot_role,
2575 other_aircraft_make_model=other_ac_make_model or None,
2576 other_aircraft_registration=other_ac_reg or None,
2577 )
2578 db.session.add(batch)
2579 db.session.flush()
2580 state["batch_id"] = batch.id
2582 linked_ids: list[int] = list(batch.linked_flight_entry_ids or [])
2584 entry, linked_ids = _gps_import_create_segment(
2585 ac=ac,
2586 aircraft_id=aircraft_id,
2587 seg=seg,
2588 seg_idx=seg_idx,
2589 pilot_role=pilot_role,
2590 dep_icao=dep_icao,
2591 arr_icao=arr_icao,
2592 nature=nature,
2593 remarks=remarks,
2594 batch=batch,
2595 file_metas=file_metas,
2596 linked_ids=linked_ids,
2597 pilot_display_name=pilot_display_name,
2598 other_aircraft=other_aircraft,
2599 other_ac_make_model=other_ac_make_model,
2600 other_ac_reg=other_ac_reg,
2601 batch_device_id=batch_device_id,
2602 )
2604 batch.linked_flight_entry_ids = linked_ids
2605 batch.segments_imported = (batch.segments_imported or 0) + 1
2606 db.session.commit()
2608 confirmed[str(seg_idx)] = entry.id if entry else 0
2609 state["confirmed_segments"] = confirmed
2610 _gps_state_save(state)
2612 all_confirmed = len(confirmed) == len(segments_data)
2614 if all_confirmed:
2615 _gps_cleanup(state)
2616 session.pop("gps_import", None)
2617 total = sum(1 for v in confirmed.values() if v != "skip")
2618 flash(
2619 ngettext(
2620 "one flight imported successfully.",
2621 "%(n)s flights imported successfully.",
2622 total,
2623 n=total,
2624 ),
2625 "success",
2626 )
2627 skipped_count = len(confirmed) - total
2628 if skipped_count:
2629 flash(
2630 ngettext(
2631 "one segment skipped.",
2632 "%(n)s segments skipped.",
2633 skipped_count,
2634 n=skipped_count,
2635 ),
2636 "info",
2637 )
2638 if pilot_role in ("pic", "dual"):
2639 return redirect(url_for("pilots.logbook"))
2640 return redirect(url_for("flights.list_flights", aircraft_id=aircraft_id))
2642 flash(_("Flight confirmed."), "success")
2643 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
2646@aircraft_bp.route(
2647 "/<aircraft_ref:aircraft_id>/gps-import/prefill-segment/<int:seg_idx>",
2648 methods=["GET"],
2649)
2650@login_required
2651@require_role(*_PILOT_ROLES)
2652def gps_import_prefill_segment(aircraft_id: int, seg_idx: int) -> ResponseReturnValue:
2653 """Store a batch segment as gps_prefill then redirect to /flights/new."""
2654 import json as _json
2655 from datetime import datetime as _dt
2657 _get_aircraft_or_404(aircraft_id)
2658 state = _gps_state_load()
2659 if not state or state.get("aircraft_id") != aircraft_id:
2660 flash(_("Session expired — please upload your GPS files again."), "warning")
2661 return redirect(url_for("aircraft.gps_import_upload", aircraft_id=aircraft_id))
2663 segments_data: list[dict[str, Any]] = state.get("segments", [])
2664 if seg_idx < 0 or seg_idx >= len(segments_data):
2665 flash(_("Invalid segment index."), "danger")
2666 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
2668 seg = segments_data[seg_idx]
2669 block_off = _dt.fromisoformat(seg["block_off_utc"])
2670 block_on = _dt.fromisoformat(seg["block_on_utc"])
2671 file_metas = state.get("files", [])
2672 single_filename = file_metas[0]["original_filename"] if len(file_metas) == 1 else ""
2673 geojson_data = _load_segment_geojson(seg)
2674 geojson_str = _json.dumps(geojson_data) if geojson_data else ""
2676 session["gps_prefill"] = {
2677 "filename": single_filename,
2678 "date": block_off.date().isoformat(),
2679 "departure_icao": seg.get("departure_icao") or "",
2680 "arrival_icao": seg.get("arrival_icao") or "",
2681 "departure_time": block_off.strftime("%H:%M"),
2682 "arrival_time": block_on.strftime("%H:%M"),
2683 "flight_time_h": str(seg.get("flight_time_rounded_h") or 0),
2684 "block_off_utc": block_off.isoformat(),
2685 "block_on_utc": block_on.isoformat(),
2686 "geojson": geojson_str,
2687 "landing_count": seg.get("landing_count") or 0,
2688 }
2689 session.modified = True
2691 return redirect(
2692 url_for(
2693 "flights.log_flight",
2694 aircraft_id=aircraft_id,
2695 gps_review_return=aircraft_id,
2696 gps_seg=seg_idx,
2697 )
2698 )
2701@aircraft_bp.route("/<aircraft_ref:aircraft_id>/gps-import/history", methods=["GET"])
2702@login_required
2703@require_role(*_PILOT_ROLES)
2704def gps_import_history(aircraft_id: int) -> ResponseReturnValue:
2705 ac = _get_aircraft_or_404(aircraft_id)
2706 batches = (
2707 AircraftGpsImportBatch.query.filter_by(aircraft_id=aircraft_id)
2708 .order_by(AircraftGpsImportBatch.imported_at.desc())
2709 .all()
2710 )
2711 return render_template(
2712 "aircraft/gps_import_history.html", aircraft=ac, batches=batches
2713 )
2716@aircraft_bp.route(
2717 "/<aircraft_ref:aircraft_id>/gps-import/<int:batch_id>/rollback", methods=["POST"]
2718)
2719@login_required
2720@require_role(*_OWNER_ROLES)
2721def gps_import_rollback(aircraft_id: int, batch_id: int) -> ResponseReturnValue:
2722 _get_aircraft_or_404(aircraft_id)
2723 batch = db.session.get(AircraftGpsImportBatch, batch_id)
2724 if not batch or batch.aircraft_id != aircraft_id:
2725 abort(404)
2727 # Flights created by this batch — delete them entirely. Unified model:
2728 # this already covers what used to be a separate PilotLogbookEntry
2729 # deletion pass (gps_batch_id and gps_import_batch_id were merged into
2730 # one column on the unified row). A flight another pilot has in their
2731 # logbook is detached from the aircraft instead of deleted
2732 # (flights/crew_removal.py).
2733 actor_id = session.get("user_id")
2734 kept = sum(
2735 not remove_from_aircraft_log(fe, actor_id)
2736 for fe in Flight.query.filter_by(gps_import_batch_id=batch.id).all()
2737 )
2739 # Flights that were pre-existing but got a GPS track linked — unlink only.
2740 linked_ids = batch.linked_flight_entry_ids or []
2741 if linked_ids:
2742 Flight.query.filter(Flight.id.in_(linked_ids)).update(
2743 {
2744 "gps_track_id": None,
2745 "block_off_utc": None,
2746 "block_on_utc": None,
2747 },
2748 synchronize_session="fetch",
2749 )
2751 db.session.delete(batch)
2752 db.session.commit()
2753 flash(
2754 _("GPS import batch rolled back and all linked flight entries removed."),
2755 "success",
2756 )
2757 flash_kept_in_pilot_logbooks(kept)
2758 return redirect(url_for("aircraft.gps_import_history", aircraft_id=aircraft_id))
2761@aircraft_bp.route(
2762 "/<aircraft_ref:aircraft_id>/flights/<int:flight_id>", methods=["GET"]
2763)
2764@login_required
2765@require_role(*_PILOT_ROLES)
2766def flight_detail(aircraft_id: int, flight_id: int) -> ResponseReturnValue:
2767 ac = _get_aircraft_or_404(aircraft_id)
2768 entry = db.session.get(Flight, flight_id)
2769 if not entry or entry.aircraft_id != aircraft_id:
2770 abort(404)
2772 # Unified model: every pilot-log field now lives on `entry` itself — no
2773 # separate PilotLogbookEntry to fetch.
2774 tile_setting = db.session.get(AppSetting, "openaip_api_key")
2775 openaip_key = tile_setting.value if tile_setting and tile_setting.value else None
2777 return render_template(
2778 "aircraft/flight_detail.html",
2779 aircraft=ac,
2780 entry=entry,
2781 pilot_entry=None,
2782 openaip_key=openaip_key,
2783 )
2786@aircraft_bp.route("/<aircraft_ref:aircraft_id>/tracks", methods=["GET"])
2787@login_required
2788@require_role(*_PILOT_ROLES)
2789def flight_tracks(aircraft_id: int) -> ResponseReturnValue:
2790 from flask import url_for as _url_for
2792 ac = _get_aircraft_or_404(aircraft_id)
2793 entries_with_tracks = (
2794 Flight.query.filter_by(aircraft_id=aircraft_id)
2795 .filter(Flight.gps_track_id.isnot(None))
2796 .order_by(Flight.date.asc())
2797 .all()
2798 )
2799 track_rows = [
2800 {
2801 "date": str(e.date),
2802 "dep": e.departure_icao,
2803 "arr": e.arrival_icao,
2804 "time_str": f"{e.flight_time} h" if e.flight_time is not None else "",
2805 "view_url": _url_for(
2806 "aircraft.flight_detail",
2807 aircraft_id=aircraft_id,
2808 flight_id=e.id,
2809 ),
2810 "geojson": e.gps_track.geojson if e.gps_track else None,
2811 }
2812 for e in entries_with_tracks
2813 ]
2815 tile_setting = db.session.get(AppSetting, "openaip_api_key")
2816 openaip_key = tile_setting.value if tile_setting and tile_setting.value else None
2818 return render_template(
2819 "aircraft/flight_tracks.html",
2820 aircraft=ac,
2821 track_rows=track_rows,
2822 openaip_key=openaip_key,
2823 )
2826@aircraft_bp.route("/<aircraft_ref:aircraft_id>/tracks/animation.gif")
2827@login_required
2828@require_role(*_PILOT_ROLES)
2829def flight_tracks_gif(aircraft_id: int) -> ResponseReturnValue:
2830 from utils import ( # pyright: ignore[reportMissingImports]
2831 generate_tracks_gif,
2832 sort_tracks_oldest_first,
2833 )
2835 ac = _get_aircraft_or_404(aircraft_id)
2836 entries = (
2837 Flight.query.filter_by(aircraft_id=aircraft_id)
2838 .filter(Flight.gps_track_id.isnot(None))
2839 .all()
2840 )
2841 track_rows = sort_tracks_oldest_first(
2842 [
2843 {
2844 "date": str(e.date),
2845 "dep": e.departure_icao or "",
2846 "arr": e.arrival_icao or "",
2847 "geojson": e.gps_track.geojson if e.gps_track else None,
2848 }
2849 for e in entries
2850 ]
2851 )
2852 tile_s = db.session.get(AppSetting, "openaip_api_key")
2853 portrait = request.args.get("orientation") == "portrait"
2854 hires = request.args.get("quality") == "hires"
2855 base_w, base_h = (480, 800) if portrait else (800, 480)
2856 mul = 2 if hires else 1
2857 canvas_w, canvas_h = base_w * mul, base_h * mul
2858 gif_bytes = generate_tracks_gif(
2859 track_rows,
2860 _openaip_key=tile_s.value if tile_s and tile_s.value else None,
2861 canvas_w=canvas_w,
2862 canvas_h=canvas_h,
2863 high_res=hires,
2864 )
2865 orient_sfx = "-portrait" if portrait else ""
2866 qual_sfx = "-hires" if hires else ""
2867 suffix = orient_sfx + qual_sfx
2868 filename = f"{ac.registration.lower().replace('-', '')}_tracks{suffix}.gif"
2869 from flask import Response # pyright: ignore[reportMissingImports]
2871 return Response(
2872 gif_bytes,
2873 mimetype="image/gif",
2874 headers={"Content-Disposition": f'attachment; filename="{filename}"'},
2875 )
2878# ── Aircraft photos ───────────────────────────────────────────────────────────
2880_PHOTO_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".heic"}
2883def _photo_folder(app: Any, tenant_slug: str, safe_reg: str) -> str:
2884 folder = app.config.get("UPLOAD_FOLDER", "/data/uploads")
2885 return os.path.join(folder, tenant_slug, safe_reg, "photos")
2888_PHOTO_THUMB_MAX_SIZE = (
2889 800 # px, long edge — see thumb_path_for()/_generate_photo_thumbnail
2890)
2893def thumb_path_for(relpath: str) -> str:
2894 """Map a photo's stored relpath to its thumbnail's -- same directory's
2895 thumbs/ subfolder, always .jpg regardless of the original's format.
2896 A pure path computation (no filesystem access) so both the upload
2897 path and serve_photo_thumb derive the same location from one place."""
2898 photo_dir, fname = os.path.split(relpath)
2899 base = os.path.splitext(fname)[0]
2900 return os.path.join(photo_dir, "thumbs", base + ".jpg").replace("\\", "/")
2903def _generate_photo_thumbnail(original_path: str, thumb_path: str) -> None:
2904 """Best-effort resized-JPEG thumbnail next to an already-saved photo.
2906 Every failure (corrupt upload, exotic format Pillow can't decode, disk
2907 error) is swallowed: serve_photo_thumb falls back to the full original
2908 whenever no thumbnail file exists, so a thumbnail-generation problem
2909 degrades to "not optimized yet" rather than blocking the upload or
2910 breaking the photo's display.
2911 """
2912 try:
2913 from PIL import ( # pyright: ignore[reportMissingImports]
2914 Image,
2915 ImageOps,
2916 )
2918 with Image.open(original_path) as img_file:
2919 upright = ImageOps.exif_transpose(img_file) or img_file
2920 rgb = upright if upright.mode in ("RGB", "L") else upright.convert("RGB")
2921 rgb.thumbnail(
2922 (_PHOTO_THUMB_MAX_SIZE, _PHOTO_THUMB_MAX_SIZE),
2923 Image.Resampling.LANCZOS,
2924 )
2925 os.makedirs(os.path.dirname(thumb_path), exist_ok=True)
2926 rgb.save(thumb_path, "JPEG", quality=85, optimize=True)
2927 except Exception:
2928 log.warning(
2929 "Photo thumbnail generation failed for %s", original_path, exc_info=True
2930 )
2933def _save_photo_file(
2934 file: Any,
2935 tenant_slug: str,
2936 safe_reg: str,
2937 sort_order: int,
2938) -> tuple[str, str]:
2939 """Save photo to canonical path; return (relpath, original_filename)."""
2941 original = secure_filename(file.filename or "photo.jpg")
2942 ext = os.path.splitext(original)[1].lower() or ".jpg"
2943 short_id = _uuid_mod.uuid4().hex[:6]
2944 fname = f"{sort_order:02d}-{short_id}{ext}"
2945 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
2946 dest_dir = os.path.join(folder, tenant_slug, safe_reg, "photos")
2947 os.makedirs(dest_dir, exist_ok=True)
2948 dest_path = os.path.join(dest_dir, fname)
2949 file.save(dest_path)
2950 relpath = os.path.join(tenant_slug, safe_reg, "photos", fname).replace("\\", "/")
2951 _generate_photo_thumbnail(dest_path, os.path.join(folder, thumb_path_for(relpath)))
2952 return relpath, original
2955def _trash_one_file(folder: str, relpath: str) -> None:
2956 src = os.path.join(folder, relpath)
2957 if not os.path.exists(src):
2958 return
2959 try:
2960 trash_dir = os.path.join(folder, "_trash")
2961 os.makedirs(trash_dir, exist_ok=True)
2962 base = os.path.basename(relpath)
2963 dest = os.path.join(trash_dir, base)
2964 if os.path.exists(dest):
2965 stem, ext = os.path.splitext(base)
2966 dest = os.path.join(trash_dir, f"{stem}_{_uuid_mod.uuid4().hex[:6]}{ext}")
2967 os.rename(src, dest)
2968 except OSError:
2969 current_app.logger.debug("Could not trash photo: %s", relpath)
2972def _trash_photo_file(filename: str) -> None:
2973 """Move a photo file, and its thumbnail if one exists, to _trash/ (same
2974 pattern as document deletion)."""
2975 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
2976 _trash_one_file(folder, filename)
2977 _trash_one_file(folder, thumb_path_for(filename))
2980def _renumber_photos(photos: list[Any], tenant_slug: str, safe_reg: str) -> None:
2981 """Assign sort_order 1..N, renaming files on disk to keep the numeric prefix."""
2982 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
2983 for new_order, photo in enumerate(photos, start=1):
2984 if photo.sort_order != new_order:
2985 old_full = os.path.join(folder, photo.filename)
2986 old_fname = os.path.basename(photo.filename)
2987 suffix = old_fname[
2988 2:
2989 ] # strip old "NN" prefix (e.g. "01-abc.jpg" → "-abc.jpg")
2990 new_fname = f"{new_order:02d}{suffix}"
2991 new_full = os.path.join(folder, tenant_slug, safe_reg, "photos", new_fname)
2992 if os.path.exists(old_full):
2993 try:
2994 os.rename(old_full, new_full)
2995 except OSError:
2996 current_app.logger.debug(
2997 "Could not renumber photo: %s", photo.filename
2998 )
2999 new_fname = old_fname # keep old name if rename fails
3000 new_rel = f"{tenant_slug}/{safe_reg}/photos/{new_fname}"
3001 # Keep the thumbnail's name in sync so it's still found by
3002 # thumb_path_for(new filename) after the rename above — without
3003 # this, every reordered photo would silently fall back to
3004 # serving its full-size original until next re-uploaded.
3005 old_thumb_full = os.path.join(folder, thumb_path_for(photo.filename))
3006 new_thumb_full = os.path.join(folder, thumb_path_for(new_rel))
3007 if new_fname != old_fname and os.path.exists(old_thumb_full):
3008 try:
3009 os.makedirs(os.path.dirname(new_thumb_full), exist_ok=True)
3010 os.rename(old_thumb_full, new_thumb_full)
3011 except OSError:
3012 current_app.logger.debug(
3013 "Could not renumber photo thumbnail: %s", photo.filename
3014 )
3015 photo.filename = new_rel
3016 photo.sort_order = new_order
3019@aircraft_bp.route("/<aircraft_ref:aircraft_id>/photos/upload", methods=["POST"])
3020@login_required
3021@require_role(*_OWNER_ROLES)
3022def upload_photo(aircraft_id: int) -> ResponseReturnValue:
3023 from documents.routes import ( # pyright: ignore[reportMissingImports]
3024 _ensure_tenant_slug,
3025 _get_tenant,
3026 )
3028 ac = _get_aircraft_or_404(aircraft_id)
3029 tenant = _get_tenant()
3031 files = request.files.getlist("photos")
3032 if not files or all(f.filename == "" for f in files):
3033 flash(_("No files selected."), "warning")
3034 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
3036 tenant_slug = _ensure_tenant_slug(tenant)
3037 safe_reg = ac.registration.replace("/", "-").replace(" ", "-").upper()
3038 next_order = (
3039 db.session.query(db.func.max(AircraftPhoto.sort_order))
3040 .filter_by(aircraft_id=ac.id)
3041 .scalar()
3042 or 0
3043 ) + 1
3045 uploaded = 0
3046 for f in files:
3047 if not f.filename:
3048 continue
3049 ext = os.path.splitext(secure_filename(f.filename))[1].lower()
3050 if ext not in _PHOTO_EXTS:
3051 flash(
3052 _(
3053 "%(name)s: unsupported format (use JPEG, PNG, WEBP or HEIC).",
3054 name=f.filename,
3055 ),
3056 "warning",
3057 )
3058 continue
3059 relpath, original = _save_photo_file(f, tenant_slug, safe_reg, next_order)
3060 photo = AircraftPhoto(
3061 aircraft_id=ac.id,
3062 filename=relpath,
3063 original_filename=original,
3064 sort_order=next_order,
3065 uploaded_by_user_id=session.get("user_id"),
3066 )
3067 db.session.add(photo)
3068 next_order += 1
3069 uploaded += 1
3071 if uploaded:
3072 db.session.commit()
3073 flash(
3074 ngettext(
3075 "one photo uploaded.", "%(n)s photos uploaded.", uploaded, n=uploaded
3076 ),
3077 "success",
3078 )
3079 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
3082@aircraft_bp.route("/<aircraft_ref:aircraft_id>/photos/<int:photo_id>/img")
3083@login_required
3084def serve_photo(aircraft_id: int, photo_id: int) -> ResponseReturnValue:
3085 from flask import send_from_directory # pyright: ignore[reportMissingImports]
3087 ac = _get_aircraft_or_404(aircraft_id)
3088 photo = db.session.get(AircraftPhoto, photo_id)
3089 if not photo or photo.aircraft_id != ac.id:
3090 abort(404)
3091 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
3092 directory = os.path.join(folder, os.path.dirname(photo.filename))
3093 fname = os.path.basename(photo.filename)
3094 # photo_id -> file is immutable for the row's lifetime (reordering only
3095 # changes sort_order; a replacement upload gets a new id) — safe to
3096 # cache aggressively so the dashboard/list prefetch hints actually pay
3097 # off instead of being revalidated away on the next navigation.
3098 response = send_from_directory(directory, fname, max_age=31536000)
3099 response.headers["Cache-Control"] = "private, max-age=31536000, immutable"
3100 return response
3103@aircraft_bp.route("/<aircraft_ref:aircraft_id>/photos/<int:photo_id>/thumb")
3104@login_required
3105def serve_photo_thumb(aircraft_id: int, photo_id: int) -> ResponseReturnValue:
3106 """Resized-JPEG version of serve_photo for the small display contexts
3107 (dashboard fleet row, single-aircraft cover, photo gallery grid) that
3108 never need the full original — see _generate_photo_thumbnail. Falls
3109 back to serving the original when no thumbnail file exists yet (a
3110 photo uploaded before this existed, or thumbnail generation failed at
3111 upload time), so every photo still displays either way."""
3112 from flask import send_from_directory # pyright: ignore[reportMissingImports]
3114 ac = _get_aircraft_or_404(aircraft_id)
3115 photo = db.session.get(AircraftPhoto, photo_id)
3116 if not photo or photo.aircraft_id != ac.id:
3117 abort(404)
3118 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
3119 thumb_relpath = thumb_path_for(photo.filename)
3120 if not os.path.isfile(os.path.join(folder, thumb_relpath)):
3121 thumb_relpath = photo.filename
3122 directory = os.path.join(folder, os.path.dirname(thumb_relpath))
3123 fname = os.path.basename(thumb_relpath)
3124 response = send_from_directory(directory, fname, max_age=31536000)
3125 response.headers["Cache-Control"] = "private, max-age=31536000, immutable"
3126 return response
3129@aircraft_bp.route(
3130 "/<aircraft_ref:aircraft_id>/photos/<int:photo_id>/delete", methods=["POST"]
3131)
3132@login_required
3133@require_role(*_OWNER_ROLES)
3134def delete_photo(aircraft_id: int, photo_id: int) -> ResponseReturnValue:
3135 from documents.routes import _get_tenant # pyright: ignore[reportMissingImports]
3137 ac = _get_aircraft_or_404(aircraft_id)
3138 photo = db.session.get(AircraftPhoto, photo_id)
3139 if not photo or photo.aircraft_id != ac.id:
3140 abort(404)
3142 _trash_photo_file(photo.filename)
3143 db.session.delete(photo)
3144 db.session.flush()
3146 # Renumber remaining photos
3147 remaining = (
3148 AircraftPhoto.query.filter_by(aircraft_id=ac.id)
3149 .order_by(AircraftPhoto.sort_order)
3150 .all()
3151 )
3152 tenant = _get_tenant()
3153 tenant_slug = tenant.slug or ""
3154 safe_reg = ac.registration.replace("/", "-").replace(" ", "-").upper()
3155 _renumber_photos(remaining, tenant_slug, safe_reg)
3156 db.session.commit()
3157 flash(_("Photo deleted."), "success")
3158 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
3161@aircraft_bp.route("/<aircraft_ref:aircraft_id>/photos/reorder", methods=["POST"])
3162@login_required
3163@require_role(*_OWNER_ROLES)
3164def reorder_photos(aircraft_id: int) -> ResponseReturnValue:
3165 from documents.routes import _get_tenant # pyright: ignore[reportMissingImports]
3167 ac = _get_aircraft_or_404(aircraft_id)
3169 ordered_ids: list[int] = []
3170 try:
3171 ordered_ids = [int(i) for i in request.form.getlist("photo_order[]")]
3172 except (ValueError, TypeError):
3173 abort(400)
3175 photos_by_id = {
3176 p.id: p for p in AircraftPhoto.query.filter_by(aircraft_id=ac.id).all()
3177 }
3178 if set(ordered_ids) != set(photos_by_id):
3179 abort(400)
3181 ordered_photos = [photos_by_id[pid] for pid in ordered_ids]
3182 tenant = _get_tenant()
3183 tenant_slug = tenant.slug or ""
3184 safe_reg = ac.registration.replace("/", "-").replace(" ", "-").upper()
3185 _renumber_photos(ordered_photos, tenant_slug, safe_reg)
3186 db.session.commit()
3187 return "", 204