Coverage for app/aircraft/gps_import.py: 100%
493 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 15:29 +0000
1"""GPS log file parsing for aircraft logbook import — Phase 30.
3Supported formats:
4- GPX 1.1 (SkyDemon, ForeFlight): speed in m/s, UTC timestamps
5- Garmin GTN/G1000 CSV: 3-row header, local time + UTC offset, GndSpd in kt
6- KML with gx:Track (SkyDemon): lon/lat/alt order, speed derived from consecutive points
7"""
9from __future__ import annotations
11import bisect
12import csv
13import functools
14import io
15import itertools
16import math
17import os
18import re
19from collections.abc import Callable
20from dataclasses import dataclass
21from datetime import UTC, date, datetime, time, timedelta
22from decimal import Decimal
23from typing import Any
24from xml.etree.ElementTree import ParseError as _ETParseError
26import defusedxml.ElementTree as ET # guards against XML bomb / entity expansion
28# ── Constants ─────────────────────────────────────────────────────────────────
30_MS_TO_KT = 1.94384 # m/s → knots
31_FT_TO_M = 0.3048 # ft → metres
32_KM_PER_NM = 1.852 # km per nautical mile
34_FLIGHT_SPEED_KT = 30.0 # sustained above this → airborne
35_GROUND_MOVE_KT = 5.0 # above this (but not 30kt for 30s) → ground movement
36_FLIGHT_SUSTAIN_S = 30.0 # seconds above 30kt required to classify as "flight"
37_SEGMENT_GAP_S = 300.0 # 5 min of slow speed or time gap → segment break
38_MAX_ICAO_DIST_KM = 5.0 # max distance for nearest-airport match
39_MAX_TRACK_POINTS = 500 # downsample threshold for GeoJSON storage
40# How many days from a GPS segment's date a hand-entered logbook date may be
41# off (a typo) and the flight still be offered as its possible match.
42NEAR_DATE_DAYS = 2
43# How far a GPS segment may start before / end after a logbook flight's
44# block times and still be part of what that flight records.
45_LOGGED_WINDOW_TOLERANCE = timedelta(minutes=15)
47# ── Data structures ───────────────────────────────────────────────────────────
50@dataclass
51class TrackPoint:
52 lat: float
53 lon: float
54 alt_m: float
55 speed_kt: float
56 utc_dt: datetime # always timezone-aware UTC
59@dataclass
60class FlightSegment:
61 trackpoints: list[TrackPoint]
62 block_off_utc: datetime
63 takeoff_utc: datetime | None
64 landing_utc: datetime | None
65 block_on_utc: datetime
66 departure_icao: str | None
67 arrival_icao: str | None
68 flight_time_raw_h: float # block_on − block_off in decimal hours
69 flight_time_rounded_h: float # rounded per aircraft precision setting
70 track_geojson: dict[str, Any] # GeoJSON Feature
71 landing_count: int
72 is_ground_only: bool # True when no airborne portion detected
73 hint_departure_icao: str | None
74 hint_arrival_icao: str | None
75 # Set when the segment joins several flights logged as one (see
76 # merge_segments_logged_as_one): their summed airborne time, which the
77 # takeoff-to-landing span would overstate by the time on the ground.
78 airborne_raw_h: float | None = None
81@dataclass
82class LoggedFlightWindow:
83 """A logbook flight's block span and route (UTC), used to recognise GPS
84 segments that it records as a single flight."""
86 start_utc: datetime
87 end_utc: datetime
88 departure_icao: str | None
89 arrival_icao: str | None
92@dataclass
93class ParsedGpsFile:
94 trackpoints: list[TrackPoint]
95 format: str # "gpx" | "kml" | "garmin_csv"
96 source_filename: str
97 classification: str # "flight" | "ground_movement" | "empty"
98 hint_departure_icao: str | None
99 hint_arrival_icao: str | None
100 device_id: str | None = None # avionics unit identifier (e.g. Garmin system_id)
103# ── Airport database ──────────────────────────────────────────────────────────
106@functools.lru_cache(maxsize=1)
107def _load_airports() -> dict[str, tuple[float, float]]:
108 """Load app/data/airports.csv once. Returns {icao: (lat, lon)}.
110 Only 4-letter ICAO codes are included. Returns an empty dict if the
111 data file is missing (ICAO lookup will return None for all queries).
112 """
113 data_path = os.path.join(os.path.dirname(__file__), "..", "data", "airports.csv")
114 airports: dict[str, tuple[float, float]] = {}
116 if os.path.exists(data_path):
117 with open(data_path, newline="", encoding="utf-8") as f:
118 reader = csv.DictReader(f)
119 for row in reader:
120 ident = row.get("ident", "").strip()
121 if not re.match(r"^[A-Z]{4}$", ident):
122 continue
123 try:
124 lat = float(row["latitude_deg"])
125 lon = float(row["longitude_deg"])
126 except (ValueError, KeyError):
127 continue
128 airports[ident] = (lat, lon)
130 return airports
133def _reset_airports_cache() -> None:
134 """Reset the airport cache (for testing)."""
135 _load_airports.cache_clear()
138# ── Haversine ─────────────────────────────────────────────────────────────────
141def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
142 """Great-circle distance in kilometres."""
143 R = 6371.0
144 phi1 = math.radians(lat1)
145 phi2 = math.radians(lat2)
146 dphi = math.radians(lat2 - lat1)
147 dlambda = math.radians(lon2 - lon1)
148 a = (
149 math.sin(dphi / 2) ** 2
150 + math.cos(phi1) * math.cos(phi2) * math.sin(dlambda / 2) ** 2
151 )
152 return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
155# ── Format detection ──────────────────────────────────────────────────────────
158def detect_format(data: bytes, filename: str) -> str:
159 """Return "gpx", "kml", or "garmin_csv". Raise ValueError for unknown format."""
160 ext = os.path.splitext(filename.lower())[1]
161 if ext == ".gpx":
162 return "gpx"
163 if ext == ".kml":
164 return "kml"
165 if ext == ".csv":
166 lines = data.decode("utf-8-sig", errors="replace").splitlines()
167 if lines and lines[0].startswith("#airframe_info"):
168 return "garmin_csv"
169 raise ValueError(f"Unsupported GPS file format: {filename!r}")
172# ── File classification ───────────────────────────────────────────────────────
175def classify_track(trackpoints: list[TrackPoint]) -> str:
176 """Return "flight", "ground_movement", or "empty"."""
177 if not trackpoints:
178 return "empty"
180 max_speed = max(tp.speed_kt for tp in trackpoints)
181 if max_speed <= _GROUND_MOVE_KT:
182 return "empty"
184 # Check for sustained window above 30kt
185 fast_window_s = 0.0
186 for prev, tp in itertools.pairwise(trackpoints):
187 if tp.speed_kt > _FLIGHT_SPEED_KT:
188 dt = (tp.utc_dt - prev.utc_dt).total_seconds()
189 fast_window_s += max(0.0, dt)
190 if fast_window_s >= _FLIGHT_SUSTAIN_S:
191 return "flight"
192 else:
193 fast_window_s = 0.0
195 return "ground_movement"
198# ── GPX parser ────────────────────────────────────────────────────────────────
200_GPX_NS = "http://www.topografix.com/GPX/1/1"
203def _extract_icao_hints(text: str) -> tuple[str | None, str | None]:
204 """Extract departure and arrival ICAO codes from a track name string."""
205 icao_matches = re.findall(r"\b([A-Z]{4})\b", text)
206 dep = icao_matches[0] if len(icao_matches) >= 1 else None
207 arr = icao_matches[-1] if len(icao_matches) >= 2 else None
208 return dep, arr
211def _parse_gpx(data: bytes, filename: str) -> ParsedGpsFile:
212 """Parse GPX 1.1 track. Speed field is in m/s; converted to kt."""
213 try:
214 root = ET.fromstring(data.decode("utf-8-sig", errors="replace"))
215 except _ETParseError as exc:
216 raise ValueError(f"Invalid GPX XML in {filename!r}: {exc}") from exc
218 hint_dep: str | None = None
219 hint_arr: str | None = None
220 name_el = root.find(f".//{{{_GPX_NS}}}name")
221 if name_el is not None and name_el.text:
222 hint_dep, hint_arr = _extract_icao_hints(name_el.text)
224 trackpoints: list[TrackPoint] = []
225 for trkpt in root.findall(f".//{{{_GPX_NS}}}trkpt"):
226 try:
227 lat = float(trkpt.get("lat", ""))
228 lon = float(trkpt.get("lon", ""))
229 except (ValueError, TypeError):
230 continue
232 ele_el = trkpt.find(f"{{{_GPX_NS}}}ele")
233 alt_m = float(ele_el.text) if ele_el is not None and ele_el.text else 0.0
235 speed_el = trkpt.find(f"{{{_GPX_NS}}}speed")
236 speed_kt = (
237 float(speed_el.text) * _MS_TO_KT
238 if speed_el is not None and speed_el.text
239 else 0.0
240 )
242 time_el = trkpt.find(f"{{{_GPX_NS}}}time")
243 if time_el is None or not time_el.text:
244 continue
245 try:
246 utc_dt = datetime.fromisoformat(time_el.text)
247 except ValueError:
248 continue
250 trackpoints.append(
251 TrackPoint(lat=lat, lon=lon, alt_m=alt_m, speed_kt=speed_kt, utc_dt=utc_dt)
252 )
254 return ParsedGpsFile(
255 trackpoints=trackpoints,
256 format="gpx",
257 source_filename=filename,
258 classification=classify_track(trackpoints),
259 hint_departure_icao=hint_dep,
260 hint_arrival_icao=hint_arr,
261 )
264# ── Garmin CSV parser ─────────────────────────────────────────────────────────
266_VALID_GPS_FIX = {"3D", "3DDiff"}
269def _parse_garmin_csv(data: bytes, filename: str) -> ParsedGpsFile:
270 """Parse Garmin GTN/G1000 CSV with 3-row header.
272 Row 0: #airframe_info metadata
273 Row 1: unit labels
274 Row 2: column names (Lcl Date, Lcl Time, UTCOfst, Latitude, Longitude, AltMSL, GndSpd, …, GPSfix, …)
275 Only rows with GPSfix in {"3D", "3DDiff"} are used.
276 Departure ICAO hint is extracted from filename: log_YYMMDD_HHMMSS_ICAO.csv
277 """
278 text = data.decode("utf-8-sig", errors="replace")
279 lines = text.splitlines()
281 if len(lines) < 4:
282 raise ValueError(f"Garmin CSV too short: {filename!r}")
284 # Device ID from #airframe_info header line
285 device_id: str | None = None
286 _did_match = re.search(r'system_id="([^"]+)"', lines[0])
287 if _did_match:
288 device_id = _did_match.group(1)
290 # Departure ICAO from filename pattern
291 hint_dep: str | None = None
292 base = os.path.splitext(os.path.basename(filename))[0]
293 parts = base.split("_")
294 if len(parts) >= 4:
295 candidate = parts[-1].strip()
296 if re.match(r"^[A-Z]{4}$", candidate):
297 hint_dep = candidate
299 # Skip rows 0–1 (metadata + units), use row 2 as header
300 csv_text = "\n".join(lines[2:])
301 reader = csv.DictReader(io.StringIO(csv_text))
302 if reader.fieldnames:
303 reader.fieldnames = [f.strip() for f in reader.fieldnames]
305 trackpoints: list[TrackPoint] = []
306 for row in reader:
307 gpsfx = row.get("GPSfix", "").strip()
308 if gpsfx not in _VALID_GPS_FIX:
309 continue
311 try:
312 lat = float(row["Latitude"].strip())
313 lon = float(row["Longitude"].strip())
314 except (ValueError, KeyError):
315 continue
317 try:
318 alt_m = float(row["AltMSL"].strip()) * _FT_TO_M
319 except (ValueError, KeyError):
320 alt_m = 0.0
322 try:
323 speed_kt = float(row["GndSpd"].strip())
324 except (ValueError, KeyError):
325 speed_kt = 0.0
327 try:
328 date_str = row["Lcl Date"].strip()
329 time_str = row["Lcl Time"].strip()
330 utc_off = row["UTCOfst"].strip()
331 local_dt = datetime.fromisoformat(f"{date_str}T{time_str}{utc_off}")
332 utc_dt = local_dt.astimezone(UTC)
333 except (ValueError, KeyError):
334 continue
336 trackpoints.append(
337 TrackPoint(lat=lat, lon=lon, alt_m=alt_m, speed_kt=speed_kt, utc_dt=utc_dt)
338 )
340 return ParsedGpsFile(
341 trackpoints=trackpoints,
342 format="garmin_csv",
343 source_filename=filename,
344 classification=classify_track(trackpoints),
345 hint_departure_icao=hint_dep,
346 hint_arrival_icao=None,
347 device_id=device_id,
348 )
351# ── KML parser ────────────────────────────────────────────────────────────────
353_KML_NS = "http://www.opengis.net/kml/2.2"
354_GX_NS = "http://www.google.com/kml/ext/2.2"
357def _parse_kml(data: bytes, filename: str) -> ParsedGpsFile:
358 """Parse SkyDemon KML with gx:Track.
360 Coordinate order is lon/lat/alt (note: reversed from GPX).
361 Speed is derived from consecutive point distance / time delta.
362 """
363 try:
364 root = ET.fromstring(data.decode("utf-8-sig", errors="replace"))
365 except _ETParseError as exc:
366 raise ValueError(f"Invalid KML XML in {filename!r}: {exc}") from exc
368 hint_dep: str | None = None
369 hint_arr: str | None = None
370 for pm in root.findall(f".//{{{_KML_NS}}}Placemark"):
371 name_el = pm.find(f"{{{_KML_NS}}}name")
372 if name_el is not None and name_el.text:
373 dep, arr = _extract_icao_hints(name_el.text)
374 if dep and arr:
375 hint_dep, hint_arr = dep, arr
376 break
378 track_el = root.find(f".//{{{_GX_NS}}}Track")
379 if track_el is None:
380 raise ValueError(f"No gx:Track element in KML: {filename!r}")
382 whens: list[datetime | None] = []
383 coords: list[tuple[float, float, float]] = []
385 for child in track_el:
386 if child.tag == f"{{{_KML_NS}}}when":
387 if child.text:
388 try:
389 dt = datetime.fromisoformat(child.text)
390 whens.append(dt.astimezone(UTC))
391 except ValueError:
392 whens.append(None)
393 else:
394 whens.append(None)
395 elif child.tag == f"{{{_GX_NS}}}coord":
396 if child.text:
397 parts = child.text.strip().split()
398 if len(parts) >= 3:
399 try:
400 lon_c = float(parts[0])
401 lat_c = float(parts[1])
402 alt_c = float(parts[2])
403 except ValueError:
404 lon_c, lat_c, alt_c = 0.0, 0.0, 0.0
405 coords.append((lon_c, lat_c, alt_c))
406 continue
407 coords.append((0.0, 0.0, 0.0))
409 if len(whens) != len(coords):
410 raise ValueError(
411 f"KML when/coord count mismatch in {filename!r}: "
412 f"{len(whens)} vs {len(coords)}"
413 )
415 trackpoints: list[TrackPoint] = []
416 for i, (when, (lon, lat, alt_m)) in enumerate(zip(whens, coords)):
417 if when is None:
418 continue
420 if trackpoints:
421 prev = trackpoints[-1]
422 dt_s = (when - prev.utc_dt).total_seconds()
423 dist_km = _haversine_km(prev.lat, prev.lon, lat, lon)
424 speed_kt = (dist_km / _KM_PER_NM * 3600.0 / dt_s) if dt_s > 0 else 0.0
425 else:
426 speed_kt = 0.0
428 trackpoints.append(
429 TrackPoint(lat=lat, lon=lon, alt_m=alt_m, speed_kt=speed_kt, utc_dt=when)
430 )
432 return ParsedGpsFile(
433 trackpoints=trackpoints,
434 format="kml",
435 source_filename=filename,
436 classification=classify_track(trackpoints),
437 hint_departure_icao=hint_dep,
438 hint_arrival_icao=hint_arr,
439 )
442# ── Entry point ───────────────────────────────────────────────────────────────
445def parse_gps_file(data: bytes, filename: str) -> ParsedGpsFile:
446 """Detect format and parse. Raises ValueError on unsupported or invalid data."""
447 fmt = detect_format(data, filename)
448 if fmt == "gpx":
449 return _parse_gpx(data, filename)
450 if fmt == "kml":
451 return _parse_kml(data, filename)
452 return _parse_garmin_csv(data, filename)
455# ── Track merge ───────────────────────────────────────────────────────────────
458def merge_and_sort(files: list[ParsedGpsFile]) -> list[TrackPoint]:
459 """Merge non-empty trackpoints from all files, sorted chronologically."""
460 all_pts: list[TrackPoint] = []
461 for f in files:
462 if f.classification != "empty":
463 all_pts.extend(f.trackpoints)
464 all_pts.sort(key=lambda tp: tp.utc_dt)
465 return all_pts
468# ── Segment detection ─────────────────────────────────────────────────────────
471def _split_into_raw_groups(trackpoints: list[TrackPoint]) -> list[list[TrackPoint]]:
472 """Split merged trackpoints into groups at slow/time gaps ≥ 5 min.
474 Only looks for breaks between the first and last fast (≥ 30kt) points, so
475 pre-flight taxi and post-landing taxi are preserved in the enclosing
476 segment — and at a break between two flights, each keeps its own taxi
477 (see _break_points).
478 """
479 n = len(trackpoints)
480 if n == 0:
481 return []
483 fast_indices = [
484 i for i, tp in enumerate(trackpoints) if tp.speed_kt >= _FLIGHT_SPEED_KT
485 ]
486 if not fast_indices:
487 return [trackpoints]
489 first_fast = fast_indices[0]
490 last_fast = fast_indices[-1]
492 groups: list[list[TrackPoint]] = []
493 current_start = 0
494 i = first_fast
496 while i < last_fast:
497 # Large time gap between consecutive points (gap between uploaded files)
498 time_gap = (trackpoints[i + 1].utc_dt - trackpoints[i].utc_dt).total_seconds()
499 if time_gap >= _SEGMENT_GAP_S:
500 groups.append(trackpoints[current_start : i + 1])
501 current_start = i + 1
502 i += 1
503 continue
505 # Slow run starting at i+1
506 if trackpoints[i + 1].speed_kt < _FLIGHT_SPEED_KT:
507 j = i + 2
508 while j <= last_fast and trackpoints[j].speed_kt < _FLIGHT_SPEED_KT:
509 j += 1
510 # Elapsed from the last fast point to the next one, so a
511 # recording gap at either end of the slow run counts too.
512 slow_dur = (trackpoints[j].utc_dt - trackpoints[i].utc_dt).total_seconds()
513 if slow_dur >= _SEGMENT_GAP_S:
514 # Real segment break between two flights — the taxi-in
515 # stays with the first, the taxi-out with the second.
516 end_prev, start_next = _break_points(trackpoints, i, j)
517 groups.append(trackpoints[current_start : end_prev + 1])
518 current_start = start_next
519 i = j
520 else:
521 i = j # short slow run — keep in current segment
522 else:
523 i += 1
525 groups.append(trackpoints[current_start:])
526 return [g for g in groups if g]
529def _break_points(
530 trackpoints: list[TrackPoint], last_fast: int, next_fast: int
531) -> tuple[int, int]:
532 """Where to split the slow stretch between two flights — the points
533 strictly between *last_fast* (the first flight's landing roll) and
534 *next_fast* (the second's takeoff roll). Returns (end of the first
535 segment, start of the second), both indices into *trackpoints*.
537 Splitting must keep each flight's taxi with it: the taxi-in carries the
538 first flight's landing (fast-to-slow transition, see _count_landings)
539 and its block-on, the taxi-out the second flight's block-off. In order:
540 - a recording gap (the GPS was off, e.g. one data-log file per
541 flight): the first flight ends at the first gap, the second starts
542 after the last one;
543 - otherwise the longest stop (below _GROUND_MOVE_KT — the aircraft
544 parked): the first flight ends where it came to rest, the second
545 starts where it moved off;
546 - otherwise (slow movement throughout, nothing to go by) the whole
547 slow stretch is left out of both.
548 """
549 gaps = [
550 k
551 for k in range(last_fast, next_fast)
552 if (trackpoints[k + 1].utc_dt - trackpoints[k].utc_dt).total_seconds()
553 >= _SEGMENT_GAP_S
554 ]
555 if gaps:
556 # Anything between the first and last gap is recording sessions
557 # with no flight in them (e.g. an engine ground run on another day).
558 return gaps[0], gaps[-1] + 1
560 best: tuple[float, int, int] | None = None
561 k = last_fast + 1
562 while k < next_fast:
563 if trackpoints[k].speed_kt >= _GROUND_MOVE_KT:
564 k += 1
565 continue
566 stop_start = k
567 while k + 1 < next_fast and trackpoints[k + 1].speed_kt < _GROUND_MOVE_KT:
568 k += 1
569 stop_s = (
570 trackpoints[k].utc_dt - trackpoints[stop_start].utc_dt
571 ).total_seconds()
572 if best is None or stop_s > best[0]:
573 best = (stop_s, stop_start, k)
574 k += 1
575 if best is not None:
576 return best[1], best[2]
577 return last_fast, next_fast
580def _count_landings(pts: list[TrackPoint]) -> int:
581 """Count transitions from airborne (≥30kt) to ground (<30kt)."""
582 count = 0
583 was_fast = False
584 for tp in pts:
585 is_fast = tp.speed_kt >= _FLIGHT_SPEED_KT
586 if was_fast and not is_fast:
587 count += 1
588 was_fast = is_fast
589 return count
592def detect_segments(
593 trackpoints: list[TrackPoint],
594 aircraft_precision: str = "tenth_hour",
595 hint_dep: str | None = None,
596 hint_arr: str | None = None,
597) -> list[FlightSegment]:
598 """Build FlightSegment objects from merged trackpoints.
600 hint_dep / hint_arr are optional ICAO codes from GPX/KML track names or
601 Garmin filename patterns, used as fallback when GPS-proximity lookup fails.
602 """
603 raw_groups = _split_into_raw_groups(trackpoints)
604 airports = _load_airports()
605 segments: list[FlightSegment] = []
607 for idx, pts in enumerate(raw_groups):
608 block_off = pts[0].utc_dt
609 block_on = pts[-1].utc_dt
611 takeoff_utc: datetime | None = None
612 landing_utc: datetime | None = None
613 for tp in pts:
614 if tp.speed_kt >= _FLIGHT_SPEED_KT:
615 if takeoff_utc is None:
616 takeoff_utc = tp.utc_dt
617 landing_utc = tp.utc_dt
619 is_ground_only = takeoff_utc is None
620 landing_count = _count_landings(pts)
622 raw_h = (block_on - block_off).total_seconds() / 3600.0
623 rounded_h = round_flight_time(raw_h, aircraft_precision)
625 dep_icao = resolve_icao(pts[0].lat, pts[0].lon, airports) or (
626 hint_dep if idx == 0 else None
627 )
628 arr_icao = resolve_icao(pts[-1].lat, pts[-1].lon, airports) or (
629 hint_arr if idx == len(raw_groups) - 1 else None
630 )
632 downsampled = downsample_track(pts)
633 geojson = build_geojson(downsampled)
635 segments.append(
636 FlightSegment(
637 trackpoints=pts,
638 block_off_utc=block_off,
639 takeoff_utc=takeoff_utc,
640 landing_utc=landing_utc,
641 block_on_utc=block_on,
642 departure_icao=dep_icao,
643 arrival_icao=arr_icao,
644 flight_time_raw_h=raw_h,
645 flight_time_rounded_h=rounded_h,
646 track_geojson=geojson,
647 landing_count=landing_count,
648 is_ground_only=is_ground_only,
649 hint_departure_icao=hint_dep if idx == 0 else None,
650 hint_arrival_icao=hint_arr if idx == len(raw_groups) - 1 else None,
651 )
652 )
654 return segments
657def logged_flight_windows(flights: list[Any]) -> list[LoggedFlightWindow]:
658 """LoggedFlightWindow for each Flight-like object whose block times
659 (departure/arrival, else takeoff/landing — all UTC times of day) are
660 both known; flights without them can't vouch for anything."""
661 windows: list[LoggedFlightWindow] = []
662 for f in flights:
663 start_t = f.departure_time or f.takeoff_time
664 end_t = f.arrival_time or f.landing_time
665 if f.date is None or start_t is None or end_t is None:
666 continue
667 start = datetime.combine(f.date, start_t, tzinfo=UTC)
668 end = datetime.combine(f.date, end_t, tzinfo=UTC)
669 if end < start: # past midnight UTC
670 end += timedelta(days=1)
671 windows.append(LoggedFlightWindow(start, end, f.departure_icao, f.arrival_icao))
672 return windows
675def _same_place(a: str | None, b: str | None) -> bool:
676 """Equal, or unknown on either side (nothing to contradict)."""
677 return not a or not b or a == b
680def _recorded_throughout(times: list[datetime], start: datetime, end: datetime) -> bool:
681 """True when the recording has no gap of _SEGMENT_GAP_S or more between
682 *start* and *end* — the avionics stayed on (one data-log file), so the
683 engine most likely kept running too."""
684 lo = bisect.bisect_left(times, start)
685 hi = bisect.bisect_right(times, end)
686 return all(
687 (times[k + 1] - times[k]).total_seconds() < _SEGMENT_GAP_S
688 for k in range(lo, hi - 1)
689 )
692def _logged_run_end(
693 segments: list[FlightSegment],
694 times: list[datetime],
695 windows: list[LoggedFlightWindow],
696 i: int,
697) -> int:
698 """Index of the last segment of the longest run starting at *i* that a
699 single logbook flight records as one, or *i* when there is none."""
700 best = i
701 for w in windows:
702 lo = w.start_utc - _LOGGED_WINDOW_TOLERANCE
703 hi = w.end_utc + _LOGGED_WINDOW_TOLERANCE
704 first = segments[i]
705 if (
706 first.is_ground_only
707 or first.block_off_utc < lo
708 or not _same_place(first.departure_icao, w.departure_icao)
709 ):
710 continue
711 k = i
712 while k + 1 < len(segments):
713 prev, nxt = segments[k], segments[k + 1]
714 if (
715 nxt.is_ground_only
716 or nxt.block_on_utc > hi
717 or not _same_place(prev.arrival_icao, nxt.departure_icao)
718 or not _recorded_throughout(times, prev.block_on_utc, nxt.block_off_utc)
719 ):
720 break
721 k += 1
722 if k > best and _same_place(nxt.arrival_icao, w.arrival_icao):
723 best = k
724 return best
727def _merge_run(
728 run: list[FlightSegment], trackpoints: list[TrackPoint], precision: str
729) -> FlightSegment:
730 """One FlightSegment spanning *run*, including the ground stops between
731 its legs."""
732 times = [tp.utc_dt for tp in trackpoints]
733 lo = bisect.bisect_left(times, run[0].trackpoints[0].utc_dt)
734 hi = bisect.bisect_right(times, run[-1].trackpoints[-1].utc_dt)
735 pts = trackpoints[lo:hi]
736 block_off = run[0].block_off_utc
737 block_on = run[-1].block_on_utc
738 raw_h = (block_on - block_off).total_seconds() / 3600.0
739 airborne_h = sum(
740 (s.landing_utc - s.takeoff_utc).total_seconds() / 3600.0
741 for s in run
742 if s.takeoff_utc and s.landing_utc
743 )
744 return FlightSegment(
745 trackpoints=pts,
746 block_off_utc=block_off,
747 takeoff_utc=run[0].takeoff_utc,
748 landing_utc=run[-1].landing_utc,
749 block_on_utc=block_on,
750 departure_icao=run[0].departure_icao,
751 arrival_icao=run[-1].arrival_icao,
752 flight_time_raw_h=raw_h,
753 flight_time_rounded_h=round_flight_time(raw_h, precision),
754 track_geojson=build_geojson(downsample_track(pts)),
755 landing_count=sum(s.landing_count for s in run),
756 is_ground_only=False,
757 hint_departure_icao=run[0].hint_departure_icao,
758 hint_arrival_icao=run[-1].hint_arrival_icao,
759 airborne_raw_h=airborne_h,
760 )
763def merge_segments_logged_as_one(
764 segments: list[FlightSegment],
765 trackpoints: list[TrackPoint],
766 windows: list[LoggedFlightWindow],
767 precision: str = "tenth_hour",
768) -> list[FlightSegment]:
769 """Join consecutive segments that one logbook flight records as a single
770 flight — e.g. a landing to check something with the engine running,
771 then on to the destination, logged as one flight with two landings.
773 Segment detection alone splits at any 5-minute stop; only the logbook
774 can say the pilot counted it as one flight. Segments are joined when
775 they all fall within that flight's block times, chain from its
776 departure to its arrival, and the recording never stopped in between
777 (a gap means the avionics, and most likely the engine, were switched
778 off — a separate flight whatever the logbook says).
779 """
780 if len(segments) < 2 or not windows:
781 return segments
782 times = [tp.utc_dt for tp in trackpoints]
783 out: list[FlightSegment] = []
784 i = 0
785 while i < len(segments):
786 k = _logged_run_end(segments, times, windows, i)
787 out.append(
788 segments[i]
789 if k == i
790 else _merge_run(segments[i : k + 1], trackpoints, precision)
791 )
792 i = k + 1
793 return out
796# ── ICAO resolution ───────────────────────────────────────────────────────────
799def resolve_icao(
800 lat: float,
801 lon: float,
802 airports: dict[str, tuple[float, float]] | None = None,
803) -> str | None:
804 """Return the nearest ICAO code within 5 km, or None if none is close enough."""
805 if airports is None:
806 airports = _load_airports()
808 best_code: str | None = None
809 best_dist = _MAX_ICAO_DIST_KM
811 for code, (ap_lat, ap_lon) in airports.items():
812 d = _haversine_km(lat, lon, ap_lat, ap_lon)
813 if d < best_dist:
814 best_dist = d
815 best_code = code
817 return best_code
820# ── Time rounding ─────────────────────────────────────────────────────────────
823def round_flight_time(raw_hours: float, precision: str) -> float:
824 """Round raw_hours up to the nearest precision boundary.
826 precision="tenth_hour": round up to nearest 0.1 h (6-min boundary).
827 precision="minute": round up to nearest 1/60 h (1-min boundary).
828 """
829 if raw_hours <= 0:
830 return 0.0
831 if precision == "minute":
832 minutes = math.ceil(raw_hours * 60)
833 return round(minutes / 60, 4)
834 # tenth_hour: ceiling to nearest 0.1
835 return round(math.ceil(raw_hours * 10) / 10, 1)
838# ── GeoJSON / downsampling ────────────────────────────────────────────────────
841@dataclass
842class SegmentTimes:
843 """A confirmed GPS segment's two duration tracks (see models.Flight's
844 docstring): block-off to block-on is the engine side (block time,
845 feeds the pilot logbook), the detected takeoff-to-landing span is the
846 airborne side the airframe log's flight_time tracks."""
848 block_time: Decimal
849 airborne_time: Decimal | None
850 takeoff_time: time | None
851 landing_time: time | None
854def segment_times(seg: dict[str, Any], precision: str) -> SegmentTimes:
855 """SegmentTimes for a serialised segment dict (as kept between the
856 review and confirm steps), rounded to the aircraft's *precision*."""
857 takeoff = (
858 datetime.fromisoformat(seg["takeoff_utc"]) if seg.get("takeoff_utc") else None
859 )
860 landing = (
861 datetime.fromisoformat(seg["landing_utc"]) if seg.get("landing_utc") else None
862 )
863 airborne: Decimal | None = None
864 if seg.get("airborne_raw_h") is not None:
865 airborne = Decimal(str(round_flight_time(seg["airborne_raw_h"], precision)))
866 elif takeoff is not None and landing is not None:
867 hours = (landing - takeoff).total_seconds() / 3600.0
868 airborne = Decimal(str(round_flight_time(hours, precision)))
869 return SegmentTimes(
870 block_time=Decimal(
871 str(round_flight_time(seg.get("flight_time_raw_h", 0), precision))
872 ),
873 airborne_time=airborne,
874 takeoff_time=takeoff.time().replace(tzinfo=None) if takeoff else None,
875 landing_time=landing.time().replace(tzinfo=None) if landing else None,
876 )
879def downsample_track(
880 trackpoints: list[TrackPoint], max_points: int = _MAX_TRACK_POINTS
881) -> list[TrackPoint]:
882 """Return ≤ max_points trackpoints using uniform stride; first and last preserved."""
883 n = len(trackpoints)
884 if n <= max_points:
885 return trackpoints
887 stride = n / max_points
888 indices: set[int] = {round(i * stride) for i in range(max_points)}
889 indices.add(0)
890 indices.add(n - 1)
891 return [trackpoints[i] for i in sorted(indices) if i < n]
894def build_geojson(trackpoints: list[TrackPoint]) -> dict[str, Any]:
895 """Return a GeoJSON Feature with a LineString geometry.
897 Coordinates: [lon, lat, alt_m] per GeoJSON spec (RFC 7946).
898 Properties carry parallel arrays of altitudes_m and speeds_kt for
899 colour-gradient rendering in Leaflet.
900 """
901 coords = [
902 [round(tp.lon, 6), round(tp.lat, 6), round(tp.alt_m, 1)] for tp in trackpoints
903 ]
904 return {
905 "type": "Feature",
906 "geometry": {"type": "LineString", "coordinates": coords},
907 "properties": {
908 "altitudes_m": [round(tp.alt_m, 1) for tp in trackpoints],
909 "speeds_kt": [round(tp.speed_kt, 1) for tp in trackpoints],
910 },
911 }
914# ── Near-match scoring against existing Flight rows without GPS block data ───
917def score_gps_candidates(
918 fields: dict[str, Any],
919 candidates: list[Any],
920 scorer: Callable[[dict[str, Any], Any], float],
921 min_score: int,
922) -> list[Any]:
923 """Rank *candidates* (Flight rows) against a parsed GPS segment's
924 *fields* dict using *scorer*, best match first.
926 A GPS segment's block_off_utc/block_on_utc only ever exists on rows
927 that themselves came from a GPS import — a flight logged manually or
928 via CSV import (airframe or pilot logbook) has neither, so the exact
929 block-time overlap check the review routes try first can never find
930 it. This is the fallback: reuse the exact same near-match scorers the
931 CSV-import review flows already use where possible
932 (_score_gps_candidate below, built on
933 flights.airframe_import._score_airframe_non_time_signals, for
934 aircraft-linked rows; pilots.logbook_import._score_candidate directly
935 for standalone ones) so a previously-imported flight is recognised
936 instead of silently duplicated — rather than inventing a second scoring
937 implementation that could drift from the first.
938 """
939 scored = [
940 (score, c) for c in candidates if (score := scorer(fields, c)) >= min_score
941 ]
942 scored.sort(key=lambda t: -t[0])
943 return [c for _score, c in scored]
946def near_date_range(gps_date: date) -> tuple[date, date]:
947 """First and last logbook date a near-date match may carry."""
948 span = timedelta(days=NEAR_DATE_DAYS)
949 return gps_date - span, gps_date + span
952def rank_near_date_candidates(
953 fields: dict[str, Any],
954 gps_date: date,
955 candidates: list[Any],
956 scorer: Callable[[dict[str, Any], Any], float],
957 min_score: int,
958) -> list[Any]:
959 """Flights logged on another date close to a GPS segment's — a typo in
960 a hand-entered logbook date — ranked closest date first, then by score.
962 The GPS date is the reliable one, so this is only a fallback for a
963 segment nothing matched on its own date, and stricter than a same-day
964 match: the route must be known and identical at both ends, and the
965 date at most NEAR_DATE_DAYS days off. A flight years away never
966 qualifies, whatever else it shares.
967 """
968 dep = fields.get("departure_icao")
969 arr = fields.get("arrival_icao")
970 if not dep or not arr:
971 return []
972 pool = [
973 c
974 for c in candidates
975 if c.date != gps_date
976 and abs((c.date - gps_date).days) <= NEAR_DATE_DAYS
977 and c.departure_icao == dep
978 and c.arrival_icao == arr
979 ]
980 ranked = score_gps_candidates(fields, pool, scorer, min_score)
981 return sorted(ranked, key=lambda c: abs((c.date - gps_date).days))
984def apply_gps_date(flight: Any, seg: dict[str, Any], block_off: datetime) -> None:
985 """Confirming a near-date match: the logbook date was the typo — take
986 the GPS one."""
987 if seg.get("matched_near_date"):
988 flight.date = block_off.date()
991def _score_gps_candidate(
992 fields: dict[str, Any], existing: Any, offset_hours: float
993) -> float:
994 """Score how likely *existing* (a Flight row) is the same real-world
995 flight as *fields* (a parsed GPS segment) — the 5 non-time signals from
996 flights.airframe_import._score_airframe_non_time_signals, plus a
997 GPS-specific comparison for the other 2.
999 A GPS track's block_off/block_on record a single observed instant per
1000 end, and — unlike a same-source CSV re-import — there's no guarantee it
1001 lines up tightly with either departure_time (engine start) or
1002 takeoff_time (wheels-up): recording can start before the engine does
1003 (setting up a flight plan on a tablet) and keep running a little after
1004 landing. So instead of preferring one time field over the other, the
1005 full-credit window spans whichever of departure_time/takeoff_time (and
1006 landing_time/arrival_time) the existing row has, widened further on
1007 each side — see services.time_band_matching.widened_span_score.
1008 """
1009 from flights.airframe_import import ( # pyright: ignore[reportMissingImports]
1010 _score_airframe_non_time_signals,
1011 )
1012 from services.time_band_matching import ( # pyright: ignore[reportMissingImports]
1013 offset_ring_step_minutes,
1014 widened_span_score,
1015 )
1017 score = _score_airframe_non_time_signals(fields, existing)
1018 widen_minutes = offset_ring_step_minutes(offset_hours)
1020 takeoff_edges = tuple(
1021 t for t in (existing.departure_time, existing.takeoff_time) if t is not None
1022 )
1023 score += widened_span_score(
1024 fields.get("takeoff_time"), takeoff_edges, widen_minutes
1025 )
1027 landing_edges = tuple(
1028 t for t in (existing.landing_time, existing.arrival_time) if t is not None
1029 )
1030 score += widened_span_score(
1031 fields.get("landing_time"), landing_edges, widen_minutes
1032 )
1034 return score