Coverage for app/aircraft/gps_import.py: 35%
620 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-13 03:30 +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_NOT_FOR_AEROPLANES = {"heliport", "closed", "balloonport"}
109@functools.lru_cache(maxsize=1)
110def _load_airports() -> dict[str, tuple[float, float]]:
111 """Load app/data/airports.csv once. Returns {icao: (lat, lon)}.
113 Only 4-letter ICAO codes are included, and only places an aeroplane
114 lands: not heliports, closed fields or balloon ports — a heliport's
115 reference point can be closer to an airport's runway than the airport's
116 own (Oostende Heliport EBNH, 1.6 km from EBOS), so it would name the
117 wrong field. Returns an empty dict if the data file is missing (ICAO
118 lookup will return None for all queries).
119 """
120 data_path = os.path.join(os.path.dirname(__file__), "..", "data", "airports.csv")
121 airports: dict[str, tuple[float, float]] = {}
123 if os.path.exists(data_path):
124 with open(data_path, newline="", encoding="utf-8") as f:
125 reader = csv.DictReader(f)
126 for row in reader:
127 ident = row.get("ident", "").strip()
128 if not re.match(r"^[A-Z]{4}$", ident):
129 continue
130 if row.get("type", "") in _NOT_FOR_AEROPLANES:
131 continue
132 try:
133 lat = float(row["latitude_deg"])
134 lon = float(row["longitude_deg"])
135 except (ValueError, KeyError):
136 continue
137 airports[ident] = (lat, lon)
139 return airports
142def _reset_airports_cache() -> None:
143 """Reset the airport cache (for testing)."""
144 _load_airports.cache_clear()
147# ── Haversine ─────────────────────────────────────────────────────────────────
150def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
151 """Great-circle distance in kilometres."""
152 R = 6371.0
153 phi1 = math.radians(lat1)
154 phi2 = math.radians(lat2)
155 dphi = math.radians(lat2 - lat1)
156 dlambda = math.radians(lon2 - lon1)
157 a = (
158 math.sin(dphi / 2) ** 2
159 + math.cos(phi1) * math.cos(phi2) * math.sin(dlambda / 2) ** 2
160 )
161 return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
164# ── Format detection ──────────────────────────────────────────────────────────
167# File name extensions a GPS upload may have. ".bin": SkyDemon for Android
168# exports its GPX and KML as "….gpx.bin" / "….kml.bin" — what's inside
169# decides (detect_format).
170GPS_UPLOAD_EXTENSIONS = (".gpx", ".kml", ".csv", ".bin")
173def _xml_root_name(text: str) -> str:
174 """The name of an XML document's first element, past its declaration,
175 comments and doctype ("" when there's none). A plain left-to-right scan
176 rather than a regular expression: one with nested repeats over those
177 parts backtracks exponentially on crafted input (CodeQL py/redos)."""
178 pos, end = 0, len(text)
179 while True:
180 while pos < end and text[pos].isspace():
181 pos += 1
182 if text.startswith("<?", pos):
183 close = text.find("?>", pos + 2)
184 skip = 2
185 elif text.startswith("<!--", pos):
186 close = text.find("-->", pos + 4)
187 skip = 3
188 elif text.startswith("<!", pos):
189 close = text.find(">", pos + 2)
190 skip = 1
191 elif text.startswith("<", pos):
192 start = pos = pos + 1
193 while pos < end and (text[pos].isalnum() or text[pos] in "_:.-"):
194 pos += 1
195 return text[start:pos]
196 else:
197 return ""
198 if close < 0:
199 return ""
200 pos = close + skip
203def is_gps_upload_name(filename: str) -> bool:
204 """Whether *filename*'s extension is one a GPS upload may have."""
205 return os.path.splitext(filename.lower())[1] in GPS_UPLOAD_EXTENSIONS
208# A Garmin data log row: "2026-05-18, 21:38:31, +02:00, ...".
209_GARMIN_ROW_RE = re.compile(r"\d{4}-\d{2}-\d{2},\s*\d{2}:\d{2}:\d{2},")
212def detect_format(data: bytes, filename: str) -> str:
213 """Return "gpx", "kml", or "garmin_csv". Raise ValueError for unknown format.
215 By extension for .gpx / .kml / .csv; otherwise (e.g. SkyDemon's
216 ".gpx.bin") by content — an XML document whose root element is <gpx>
217 or <kml>."""
218 ext = os.path.splitext(filename.lower())[1]
219 if ext == ".gpx":
220 return "gpx"
221 if ext == ".kml":
222 return "kml"
223 if ext == ".csv":
224 lines = data.decode("utf-8-sig", errors="replace").splitlines()
225 if lines and lines[0].startswith("#airframe_info"):
226 return "garmin_csv"
227 if any(_GARMIN_ROW_RE.match(line) for line in lines[:10]):
228 # Rows of a Garmin data log without its header: a fragment the
229 # logger writes while powering up or down — seconds to a few
230 # minutes, usually without a GPS position, never a flight.
231 raise ValueError(
232 f"{filename!r} is a fragment of a Garmin data log without its "
233 "header (written while the logger was powering up or down) — "
234 "nothing to import."
235 )
236 else:
237 head = data[:4096].decode("utf-8-sig", errors="replace")
238 root = _xml_root_name(head).rsplit(":", 1)[-1].lower()
239 if root in ("gpx", "kml"):
240 return root
241 if filename.lower().endswith(".flightlog.bin"):
242 raise ValueError(
243 f"{filename!r} is SkyDemon's own log format — export the track "
244 "as GPX (….gpx.bin) or KML from SkyDemon instead."
245 )
246 raise ValueError(f"Unsupported GPS file format: {filename!r}")
249# ── File classification ───────────────────────────────────────────────────────
252def classify_track(trackpoints: list[TrackPoint]) -> str:
253 """Return "flight", "ground_movement", or "empty"."""
254 if not trackpoints:
255 return "empty"
257 max_speed = max(tp.speed_kt for tp in trackpoints)
258 if max_speed <= _GROUND_MOVE_KT:
259 return "empty"
261 # Check for sustained window above 30kt
262 fast_window_s = 0.0
263 for prev, tp in itertools.pairwise(trackpoints):
264 if tp.speed_kt > _FLIGHT_SPEED_KT:
265 dt = (tp.utc_dt - prev.utc_dt).total_seconds()
266 fast_window_s += max(0.0, dt)
267 if fast_window_s >= _FLIGHT_SUSTAIN_S:
268 return "flight"
269 else:
270 fast_window_s = 0.0
272 return "ground_movement"
275# ── GPX parser ────────────────────────────────────────────────────────────────
278def _extract_icao_hints(text: str) -> tuple[str | None, str | None]:
279 """Extract departure and arrival ICAO codes from a track name string."""
280 icao_matches = re.findall(r"\b([A-Z]{4})\b", text)
281 dep = icao_matches[0] if len(icao_matches) >= 1 else None
282 arr = icao_matches[-1] if len(icao_matches) >= 2 else None
283 return dep, arr
286def _parse_xml_up_to_error(text: str) -> Any:
287 """The XML document's root element — or, when the document breaks off
288 (seen on real GPX exports missing their closing </gpx> tag), what was
289 read of it before the error. Raises the parse error when nothing was."""
290 try:
291 return ET.fromstring(text)
292 except _ETParseError as exc:
293 error = exc
294 # Read again incrementally, keeping the root element as soon as it's
295 # seen; the same error then ends the reading where the document breaks.
296 root = None
297 try:
298 for _event, element in ET.iterparse(
299 io.BytesIO(text.encode("utf-8")), events=("start",)
300 ):
301 if root is None:
302 root = element
303 except _ETParseError:
304 if root is not None:
305 return root
306 raise error from None
309# A jump between two consecutive points implying more than this is a bad
310# fix, not flight — no light aircraft flies it, even across a recording gap.
311_MAX_PLAUSIBLE_KT = 600.0
312# …and covering at least this distance (a stale fix jumps tens of km).
313_MIN_JUMP_NM = 0.5
314# What's left between two such jumps, if shorter than this, is the bad fix.
315_MIN_FIX_RUN_S = 60.0
318def _drop_bad_fixes(
319 points: list[tuple[float, float, float, float | None, datetime]],
320) -> list[tuple[float, float, float, float | None, datetime]]:
321 """Points (lat, lon, alt, speed, time) in time order, without the bad
322 fixes cameras and phones record while acquiring the GPS signal: no fix
323 at all (0°, 0° — often with an unset clock, e.g. a date in another
324 month), or a stale last-known position the track then jumps away from
325 (seen on GoPro exports: 41 km in one second). The track is cut at every
326 implausible jump; pieces under _MIN_FIX_RUN_S are then dropped."""
327 ordered = sorted(
328 (p for p in points if not (p[0] == 0.0 and p[1] == 0.0)),
329 key=lambda p: p[4],
330 )
331 runs: list[list[tuple[float, float, float, float | None, datetime]]] = []
332 for point in ordered:
333 if runs:
334 prev = runs[-1][-1]
335 dt_s = (point[4] - prev[4]).total_seconds()
336 dist_nm = _haversine_km(prev[0], prev[1], point[0], point[1]) / _KM_PER_NM
337 # Both far and fast: a few metres of jitter between two
338 # timestamps a fraction of a second apart (GoPro exports have
339 # those) computes as hundreds of knots too.
340 jump = dist_nm > _MIN_JUMP_NM and (
341 dt_s <= 0 or dist_nm * 3600.0 / dt_s > _MAX_PLAUSIBLE_KT
342 )
343 if not jump:
344 runs[-1].append(point)
345 continue
346 runs.append([point])
347 if len(runs) <= 1: # no implausible jump: nothing to tell apart
348 return ordered
349 return [
350 p
351 for run in runs
352 if (run[-1][4] - run[0][4]).total_seconds() >= _MIN_FIX_RUN_S
353 for p in run
354 ]
357# Speed from positions is measured over at least this long: GPS positions
358# jitter by a few metres, so point-to-point at 1 Hz turns a parked aircraft's
359# noise — or one bad fix — into spurious knots.
360_DERIVED_SPEED_SPAN_S = 5.0
363def _speeds_from_positions(points: list[tuple[float, float, datetime]]) -> list[float]:
364 """Ground speed in knots at each (lat, lon, time) point, from the
365 distance covered since the latest point at least _DERIVED_SPEED_SPAN_S
366 earlier (the previous point when none is that old yet)."""
367 speeds: list[float] = []
368 j = 0
369 for i, (lat, lon, when) in enumerate(points):
370 if i == 0:
371 speeds.append(0.0)
372 continue
373 while (
374 j + 1 < i
375 and (when - points[j + 1][2]).total_seconds() >= _DERIVED_SPEED_SPAN_S
376 ):
377 j += 1
378 k = (
379 j
380 if (when - points[j][2]).total_seconds() >= _DERIVED_SPEED_SPAN_S
381 else i - 1
382 )
383 dt_s = (when - points[k][2]).total_seconds()
384 dist_km = _haversine_km(points[k][0], points[k][1], lat, lon)
385 speeds.append(dist_km / _KM_PER_NM * 3600.0 / dt_s if dt_s > 0 else 0.0)
386 return speeds
389def _parse_gpx(data: bytes, filename: str) -> ParsedGpsFile:
390 """Parse GPX 1.1 track. Speed field is in m/s; converted to kt — or,
391 when the file has none, derived from the positions."""
392 try:
393 root = _parse_xml_up_to_error(data.decode("utf-8-sig", errors="replace"))
394 except _ETParseError as exc:
395 raise ValueError(f"Invalid GPX XML in {filename!r}: {exc}") from exc
397 # The namespace the file declares: GPX 1.1, GPX 1.0 (same structure, e.g.
398 # ExifTool exports) — or none at all.
399 ns = root.tag[1 : root.tag.index("}")] if root.tag.startswith("{") else ""
401 def q(name: str) -> str:
402 return f"{{{ns}}}{name}" if ns else name
404 hint_dep: str | None = None
405 hint_arr: str | None = None
406 name_el = root.find(f".//{q('name')}")
407 if name_el is not None and name_el.text:
408 hint_dep, hint_arr = _extract_icao_hints(name_el.text)
410 points: list[tuple[float, float, float, float | None, datetime]] = []
411 for trkpt in root.findall(f".//{q('trkpt')}"):
412 try:
413 lat = float(trkpt.get("lat", ""))
414 lon = float(trkpt.get("lon", ""))
415 except (ValueError, TypeError):
416 continue
418 ele_el = trkpt.find(q("ele"))
419 alt_m = float(ele_el.text) if ele_el is not None and ele_el.text else 0.0
421 speed_el = trkpt.find(q("speed"))
422 speed_kt = (
423 float(speed_el.text) * _MS_TO_KT
424 if speed_el is not None and speed_el.text
425 else None
426 )
428 time_el = trkpt.find(q("time"))
429 if time_el is None or not time_el.text:
430 continue
431 try:
432 utc_dt = datetime.fromisoformat(time_el.text)
433 except ValueError:
434 continue
436 points.append((lat, lon, alt_m, speed_kt, utc_dt))
438 points = _drop_bad_fixes(points)
439 # GPX 1.1 has no speed element: many writers (e.g. ExifTool exporting
440 # a GoPro's GPS) leave it out — derive it from the positions then.
441 if points and all(p[3] is None for p in points):
442 speeds = _speeds_from_positions([(p[0], p[1], p[4]) for p in points])
443 else:
444 speeds = [p[3] or 0.0 for p in points]
445 trackpoints = [
446 TrackPoint(lat=lat, lon=lon, alt_m=alt_m, speed_kt=kt, utc_dt=when)
447 for (lat, lon, alt_m, _speed, when), kt in zip(points, speeds, strict=True)
448 ]
450 return ParsedGpsFile(
451 trackpoints=trackpoints,
452 format="gpx",
453 source_filename=filename,
454 classification=classify_track(trackpoints),
455 hint_departure_icao=hint_dep,
456 hint_arrival_icao=hint_arr,
457 )
460# ── Garmin CSV parser ─────────────────────────────────────────────────────────
462_VALID_GPS_FIX = {"3D", "3DDiff"}
465def _parse_garmin_csv(data: bytes, filename: str) -> ParsedGpsFile:
466 """Parse Garmin GTN/G1000 CSV with 3-row header.
468 Row 0: #airframe_info metadata
469 Row 1: unit labels
470 Row 2: column names (Lcl Date, Lcl Time, UTCOfst, Latitude, Longitude, AltMSL, GndSpd, …, GPSfix, …)
471 Only rows with GPSfix in {"3D", "3DDiff"} are used.
472 Departure ICAO hint is extracted from filename: log_YYMMDD_HHMMSS_ICAO.csv
473 """
474 text = data.decode("utf-8-sig", errors="replace")
475 lines = text.splitlines()
477 if len(lines) < 4:
478 raise ValueError(f"Garmin CSV too short: {filename!r}")
480 # Device ID from #airframe_info header line
481 device_id: str | None = None
482 _did_match = re.search(r'system_id="([^"]+)"', lines[0])
483 if _did_match:
484 device_id = _did_match.group(1)
486 # Departure ICAO from filename pattern
487 hint_dep: str | None = None
488 base = os.path.splitext(os.path.basename(filename))[0]
489 parts = base.split("_")
490 if len(parts) >= 4:
491 candidate = parts[-1].strip()
492 if re.match(r"^[A-Z]{4}$", candidate):
493 hint_dep = candidate
495 # Skip rows 0–1 (metadata + units), use row 2 as header
496 csv_text = "\n".join(lines[2:])
497 reader = csv.DictReader(io.StringIO(csv_text))
498 if reader.fieldnames:
499 reader.fieldnames = [f.strip() for f in reader.fieldnames]
501 trackpoints: list[TrackPoint] = []
502 for row in reader:
503 gpsfx = row.get("GPSfix", "").strip()
504 if gpsfx not in _VALID_GPS_FIX:
505 continue
507 try:
508 lat = float(row["Latitude"].strip())
509 lon = float(row["Longitude"].strip())
510 except (ValueError, KeyError):
511 continue
513 try:
514 alt_m = float(row["AltMSL"].strip()) * _FT_TO_M
515 except (ValueError, KeyError):
516 alt_m = 0.0
518 try:
519 speed_kt = float(row["GndSpd"].strip())
520 except (ValueError, KeyError):
521 speed_kt = 0.0
523 try:
524 date_str = row["Lcl Date"].strip()
525 time_str = row["Lcl Time"].strip()
526 utc_off = row["UTCOfst"].strip()
527 local_dt = datetime.fromisoformat(f"{date_str}T{time_str}{utc_off}")
528 utc_dt = local_dt.astimezone(UTC)
529 except (ValueError, KeyError):
530 continue
532 trackpoints.append(
533 TrackPoint(lat=lat, lon=lon, alt_m=alt_m, speed_kt=speed_kt, utc_dt=utc_dt)
534 )
536 return ParsedGpsFile(
537 trackpoints=trackpoints,
538 format="garmin_csv",
539 source_filename=filename,
540 classification=classify_track(trackpoints),
541 hint_departure_icao=hint_dep,
542 hint_arrival_icao=None,
543 device_id=device_id,
544 )
547# ── KML parser ────────────────────────────────────────────────────────────────
549_KML_NS = "http://www.opengis.net/kml/2.2"
550_GX_NS = "http://www.google.com/kml/ext/2.2"
553def _parse_kml(data: bytes, filename: str) -> ParsedGpsFile:
554 """Parse SkyDemon KML with gx:Track.
556 Coordinate order is lon/lat/alt (note: reversed from GPX).
557 Speed is derived from consecutive point distance / time delta.
558 """
559 try:
560 root = ET.fromstring(data.decode("utf-8-sig", errors="replace"))
561 except _ETParseError as exc:
562 raise ValueError(f"Invalid KML XML in {filename!r}: {exc}") from exc
564 hint_dep: str | None = None
565 hint_arr: str | None = None
566 for pm in root.findall(f".//{{{_KML_NS}}}Placemark"):
567 name_el = pm.find(f"{{{_KML_NS}}}name")
568 if name_el is not None and name_el.text:
569 dep, arr = _extract_icao_hints(name_el.text)
570 if dep and arr:
571 hint_dep, hint_arr = dep, arr
572 break
574 track_el = root.find(f".//{{{_GX_NS}}}Track")
575 if track_el is None:
576 raise ValueError(f"No gx:Track element in KML: {filename!r}")
578 whens: list[datetime | None] = []
579 coords: list[tuple[float, float, float]] = []
581 for child in track_el:
582 if child.tag == f"{{{_KML_NS}}}when":
583 if child.text:
584 try:
585 dt = datetime.fromisoformat(child.text)
586 whens.append(dt.astimezone(UTC))
587 except ValueError:
588 whens.append(None)
589 else:
590 whens.append(None)
591 elif child.tag == f"{{{_GX_NS}}}coord":
592 if child.text:
593 parts = child.text.strip().split()
594 if len(parts) >= 3:
595 try:
596 lon_c = float(parts[0])
597 lat_c = float(parts[1])
598 alt_c = float(parts[2])
599 except ValueError:
600 lon_c, lat_c, alt_c = 0.0, 0.0, 0.0
601 coords.append((lon_c, lat_c, alt_c))
602 continue
603 coords.append((0.0, 0.0, 0.0))
605 if len(whens) != len(coords):
606 raise ValueError(
607 f"KML when/coord count mismatch in {filename!r}: "
608 f"{len(whens)} vs {len(coords)}"
609 )
611 # KML carries no speed: derived from the positions, like a GPX file
612 # without one — with the same bad-fix filtering and smoothing.
613 points = _drop_bad_fixes(
614 [
615 (lat, lon, alt_m, None, when)
616 for when, (lon, lat, alt_m) in zip(whens, coords, strict=True)
617 if when is not None
618 ]
619 )
620 speeds = _speeds_from_positions([(p[0], p[1], p[4]) for p in points])
621 trackpoints = [
622 TrackPoint(lat=lat, lon=lon, alt_m=alt_m, speed_kt=kt, utc_dt=when)
623 for (lat, lon, alt_m, _speed, when), kt in zip(points, speeds, strict=True)
624 ]
626 return ParsedGpsFile(
627 trackpoints=trackpoints,
628 format="kml",
629 source_filename=filename,
630 classification=classify_track(trackpoints),
631 hint_departure_icao=hint_dep,
632 hint_arrival_icao=hint_arr,
633 )
636# ── Entry point ───────────────────────────────────────────────────────────────
639def parse_gps_file(data: bytes, filename: str) -> ParsedGpsFile:
640 """Detect format and parse. Raises ValueError on unsupported or invalid data."""
641 fmt = detect_format(data, filename)
642 if fmt == "gpx":
643 return _parse_gpx(data, filename)
644 if fmt == "kml":
645 return _parse_kml(data, filename)
646 return _parse_garmin_csv(data, filename)
649# ── Track merge ───────────────────────────────────────────────────────────────
652def merge_and_sort(files: list[ParsedGpsFile]) -> list[TrackPoint]:
653 """Merge non-empty trackpoints from all files, sorted chronologically."""
654 all_pts: list[TrackPoint] = []
655 for f in files:
656 if f.classification != "empty":
657 all_pts.extend(f.trackpoints)
658 all_pts.sort(key=lambda tp: tp.utc_dt)
659 return all_pts
662# ── Segment detection ─────────────────────────────────────────────────────────
665def _split_into_raw_groups(trackpoints: list[TrackPoint]) -> list[list[TrackPoint]]:
666 """Split merged trackpoints into groups at slow/time gaps ≥ 5 min.
668 Only looks for breaks between the first and last fast (≥ 30kt) points, so
669 pre-flight taxi and post-landing taxi are preserved in the enclosing
670 segment — and at a break between two flights, each keeps its own taxi
671 (see _break_points).
672 """
673 n = len(trackpoints)
674 if n == 0:
675 return []
677 fast_indices = [
678 i for i, tp in enumerate(trackpoints) if tp.speed_kt >= _FLIGHT_SPEED_KT
679 ]
680 if not fast_indices:
681 # Ground runs only: still one group per recording session.
682 sessions: list[list[TrackPoint]] = []
683 start = 0
684 for k in range(n - 1):
685 if _recording_gap(trackpoints, k):
686 sessions.append(trackpoints[start : k + 1])
687 start = k + 1
688 sessions.append(trackpoints[start:])
689 return sessions
691 first_fast = fast_indices[0]
692 last_fast = fast_indices[-1]
694 groups: list[list[TrackPoint]] = []
695 current_start = 0
696 # Before the first flight and after the last one, a recording gap still
697 # separates sessions — a ground-only file from days earlier or later
698 # (e.g. engine run at the parking spot) isn't that flight's taxi.
699 for k in range(first_fast):
700 if _recording_gap(trackpoints, k):
701 groups.append(trackpoints[current_start : k + 1])
702 current_start = k + 1
703 i = first_fast
705 while i < last_fast:
706 # Large time gap between consecutive points (gap between uploaded files)
707 time_gap = (trackpoints[i + 1].utc_dt - trackpoints[i].utc_dt).total_seconds()
708 if time_gap >= _SEGMENT_GAP_S:
709 groups.append(trackpoints[current_start : i + 1])
710 current_start = i + 1
711 i += 1
712 continue
714 # Slow run starting at i+1
715 if trackpoints[i + 1].speed_kt < _FLIGHT_SPEED_KT:
716 j = i + 2
717 while j <= last_fast and trackpoints[j].speed_kt < _FLIGHT_SPEED_KT:
718 j += 1
719 # Elapsed from the last fast point to the next one, so a
720 # recording gap at either end of the slow run counts too.
721 slow_dur = (trackpoints[j].utc_dt - trackpoints[i].utc_dt).total_seconds()
722 if slow_dur >= _SEGMENT_GAP_S:
723 # Real segment break between two flights — the taxi-in
724 # stays with the first, the taxi-out with the second.
725 end_prev, start_next = _break_points(trackpoints, i, j)
726 groups.append(trackpoints[current_start : end_prev + 1])
727 # Recording sessions in between, with no flight in them (a
728 # taxi to the fuel station, an engine run on another day):
729 # each its own ground-only group.
730 if _recording_gap(trackpoints, end_prev):
731 session_start = end_prev + 1
732 for k in range(end_prev + 1, start_next):
733 if _recording_gap(trackpoints, k):
734 groups.append(trackpoints[session_start : k + 1])
735 session_start = k + 1
736 current_start = start_next
737 i = j
738 else:
739 i = j # short slow run — keep in current segment
740 else:
741 i += 1
743 for k in range(last_fast, n - 1):
744 if _recording_gap(trackpoints, k):
745 groups.append(trackpoints[current_start : k + 1])
746 current_start = k + 1
747 groups.append(trackpoints[current_start:])
748 return [g for g in groups if g]
751def _recording_gap(trackpoints: list[TrackPoint], k: int) -> bool:
752 """Whether the recording stopped for _SEGMENT_GAP_S or more after point k."""
753 gap = (trackpoints[k + 1].utc_dt - trackpoints[k].utc_dt).total_seconds()
754 return gap >= _SEGMENT_GAP_S
757def _break_points(
758 trackpoints: list[TrackPoint], last_fast: int, next_fast: int
759) -> tuple[int, int]:
760 """Where to split the slow stretch between two flights — the points
761 strictly between *last_fast* (the first flight's landing roll) and
762 *next_fast* (the second's takeoff roll). Returns (end of the first
763 segment, start of the second), both indices into *trackpoints*.
765 Splitting must keep each flight's taxi with it: the taxi-in carries the
766 first flight's landing (fast-to-slow transition, see _count_landings)
767 and its block-on, the taxi-out the second flight's block-off. In order:
768 - a recording gap (the GPS was off, e.g. one data-log file per
769 flight): the first flight ends at the first gap, the second starts
770 after the last one;
771 - otherwise the longest stop (below _GROUND_MOVE_KT — the aircraft
772 parked): the first flight ends where it came to rest, the second
773 starts where it moved off;
774 - otherwise (slow movement throughout, nothing to go by) the whole
775 slow stretch is left out of both.
776 """
777 gaps = [
778 k
779 for k in range(last_fast, next_fast)
780 if (trackpoints[k + 1].utc_dt - trackpoints[k].utc_dt).total_seconds()
781 >= _SEGMENT_GAP_S
782 ]
783 if gaps:
784 # Anything between the first and last gap is recording sessions
785 # with no flight in them (e.g. an engine ground run on another day)
786 # — _split_into_raw_groups keeps each as a ground-only group.
787 return gaps[0], gaps[-1] + 1
789 best: tuple[float, int, int] | None = None
790 k = last_fast + 1
791 while k < next_fast:
792 if trackpoints[k].speed_kt >= _GROUND_MOVE_KT:
793 k += 1
794 continue
795 stop_start = k
796 while k + 1 < next_fast and trackpoints[k + 1].speed_kt < _GROUND_MOVE_KT:
797 k += 1
798 stop_s = (
799 trackpoints[k].utc_dt - trackpoints[stop_start].utc_dt
800 ).total_seconds()
801 if best is None or stop_s > best[0]:
802 best = (stop_s, stop_start, k)
803 k += 1
804 if best is not None:
805 return best[1], best[2]
806 return last_fast, next_fast
809# A segment whose flying starts (ends) this close to its first (last)
810# point, and that high above its lowest part (≈ the airfield), was recorded
811# from (until) the air — not from a takeoff roll (until a landing roll),
812# which is fast too but on the ground (SkyDemon may start logging mid-roll).
813_IN_FLIGHT_EDGE_S = 10.0
814_IN_FLIGHT_HEIGHT_M = 150.0
815_EDGE_POINTS = 10
818def _median(values: list[float]) -> float:
819 ordered = sorted(values)
820 return ordered[len(ordered) // 2]
823def _high_above_ground(pts: list[TrackPoint], edge: list[TrackPoint]) -> bool:
824 """Whether the *edge* points (a segment's first or last few) are well
825 above the segment's low part (its 10th percentile altitude — robust to
826 the noisy altitudes of cameras and phones)."""
827 altitudes = sorted(p.alt_m for p in pts)
828 low = altitudes[len(altitudes) // 10]
829 return _median([p.alt_m for p in edge]) - low > _IN_FLIGHT_HEIGHT_M
832# Above _FLIGHT_SPEED_KT for at least this long is flying, below it for at
833# least this long after flying is a landing: a noisy GPS (a camera's, a
834# phone's) briefly crosses the threshold while taxiing or in flight.
835_SUSTAIN_S = 30.0
838def _airborne_periods(pts: list[TrackPoint]) -> list[tuple[int, int]]:
839 """(first, last) point indices of each stretch flown above
840 _FLIGHT_SPEED_KT for at least _SUSTAIN_S, merging stretches separated
841 by less than _SUSTAIN_S below it (noise, not a landing)."""
842 runs: list[tuple[int, int]] = []
843 i = 0
844 while i < len(pts):
845 if pts[i].speed_kt < _FLIGHT_SPEED_KT:
846 i += 1
847 continue
848 j = i
849 while j + 1 < len(pts) and pts[j + 1].speed_kt >= _FLIGHT_SPEED_KT:
850 j += 1
851 if (pts[j].utc_dt - pts[i].utc_dt).total_seconds() >= _SUSTAIN_S:
852 runs.append((i, j))
853 i = j + 1
854 periods: list[tuple[int, int]] = []
855 for first, last in runs:
856 if (
857 periods
858 and (pts[first].utc_dt - pts[periods[-1][1]].utc_dt).total_seconds()
859 < _SUSTAIN_S
860 ):
861 periods[-1] = (periods[-1][0], last)
862 else:
863 periods.append((first, last))
864 return periods
867def _count_landings(pts: list[TrackPoint]) -> int:
868 """Full-stop landings: airborne periods (see _airborne_periods) the
869 recording carries on after — i.e. followed by slow points."""
870 return sum(1 for _first, last in _airborne_periods(pts) if last + 1 < len(pts))
873def detect_segments(
874 trackpoints: list[TrackPoint],
875 aircraft_precision: str = "tenth_hour",
876 hint_dep: str | None = None,
877 hint_arr: str | None = None,
878) -> list[FlightSegment]:
879 """Build FlightSegment objects from merged trackpoints.
881 hint_dep / hint_arr are optional ICAO codes from GPX/KML track names or
882 Garmin filename patterns, used as fallback when GPS-proximity lookup fails.
883 """
884 raw_groups = _split_into_raw_groups(trackpoints)
885 airports = _load_airports()
886 segments: list[FlightSegment] = []
888 for idx, pts in enumerate(raw_groups):
889 block_off = pts[0].utc_dt
890 block_on = pts[-1].utc_dt
892 periods = _airborne_periods(pts)
893 takeoff_utc: datetime | None = pts[periods[0][0]].utc_dt if periods else None
894 landing_utc: datetime | None = pts[periods[-1][1]].utc_dt if periods else None
896 is_ground_only = takeoff_utc is None
897 landing_count = _count_landings(pts)
899 raw_h = (block_on - block_off).total_seconds() / 3600.0
900 rounded_h = round_flight_time(raw_h, aircraft_precision)
902 # A recording that starts (or ends) already flying — the camera or
903 # logger switched on in flight — has no departure (arrival) airfield
904 # to tell: whatever it's flying over isn't one.
905 started_flying = bool(
906 periods
907 and (pts[periods[0][0]].utc_dt - pts[0].utc_dt).total_seconds()
908 < _IN_FLIGHT_EDGE_S
909 and _high_above_ground(pts, pts[:_EDGE_POINTS])
910 )
911 ended_flying = bool(
912 periods
913 and (pts[-1].utc_dt - pts[periods[-1][1]].utc_dt).total_seconds()
914 < _IN_FLIGHT_EDGE_S
915 and _high_above_ground(pts, pts[-_EDGE_POINTS:])
916 )
917 # (The file's own hint — a Garmin file name, a GPX track name —
918 # still counts: it's the recorder's own record of the field.)
919 dep_icao = (
920 None if started_flying else resolve_icao(pts[0].lat, pts[0].lon, airports)
921 ) or (hint_dep if idx == 0 else None)
922 arr_icao = (
923 None if ended_flying else resolve_icao(pts[-1].lat, pts[-1].lon, airports)
924 ) or (hint_arr if idx == len(raw_groups) - 1 else None)
926 downsampled = downsample_track(pts)
927 geojson = build_geojson(downsampled)
929 segments.append(
930 FlightSegment(
931 trackpoints=pts,
932 block_off_utc=block_off,
933 takeoff_utc=takeoff_utc,
934 landing_utc=landing_utc,
935 block_on_utc=block_on,
936 departure_icao=dep_icao,
937 arrival_icao=arr_icao,
938 flight_time_raw_h=raw_h,
939 flight_time_rounded_h=rounded_h,
940 track_geojson=geojson,
941 landing_count=landing_count,
942 is_ground_only=is_ground_only,
943 hint_departure_icao=hint_dep if idx == 0 else None,
944 hint_arrival_icao=hint_arr if idx == len(raw_groups) - 1 else None,
945 )
946 )
948 return segments
951def logged_flight_windows(flights: list[Any]) -> list[LoggedFlightWindow]:
952 """LoggedFlightWindow for each Flight-like object whose block times
953 (departure/arrival, else takeoff/landing — all UTC times of day) are
954 both known; flights without them can't vouch for anything."""
955 windows: list[LoggedFlightWindow] = []
956 for f in flights:
957 start_t = f.departure_time or f.takeoff_time
958 end_t = f.arrival_time or f.landing_time
959 if f.date is None or start_t is None or end_t is None:
960 continue
961 start = datetime.combine(f.date, start_t, tzinfo=UTC)
962 end = datetime.combine(f.date, end_t, tzinfo=UTC)
963 if end < start: # past midnight UTC
964 end += timedelta(days=1)
965 windows.append(LoggedFlightWindow(start, end, f.departure_icao, f.arrival_icao))
966 return windows
969def _same_place(a: str | None, b: str | None) -> bool:
970 """Equal, or unknown on either side (nothing to contradict)."""
971 return not a or not b or a == b
974def _recorded_throughout(times: list[datetime], start: datetime, end: datetime) -> bool:
975 """True when the recording has no gap of _SEGMENT_GAP_S or more between
976 *start* and *end* — the avionics stayed on (one data-log file), so the
977 engine most likely kept running too."""
978 lo = bisect.bisect_left(times, start)
979 hi = bisect.bisect_right(times, end)
980 return all(
981 (times[k + 1] - times[k]).total_seconds() < _SEGMENT_GAP_S
982 for k in range(lo, hi - 1)
983 )
986def _logged_run_end(
987 segments: list[FlightSegment],
988 times: list[datetime],
989 windows: list[LoggedFlightWindow],
990 i: int,
991) -> int:
992 """Index of the last segment of the longest run starting at *i* that a
993 single logbook flight records as one, or *i* when there is none."""
994 best = i
995 for w in windows:
996 lo = w.start_utc - _LOGGED_WINDOW_TOLERANCE
997 hi = w.end_utc + _LOGGED_WINDOW_TOLERANCE
998 first = segments[i]
999 if (
1000 first.is_ground_only
1001 or first.block_off_utc < lo
1002 or not _same_place(first.departure_icao, w.departure_icao)
1003 ):
1004 continue
1005 k = i
1006 while k + 1 < len(segments):
1007 prev, nxt = segments[k], segments[k + 1]
1008 if (
1009 nxt.is_ground_only
1010 or nxt.block_on_utc > hi
1011 or not _same_place(prev.arrival_icao, nxt.departure_icao)
1012 or not _recorded_throughout(times, prev.block_on_utc, nxt.block_off_utc)
1013 ):
1014 break
1015 k += 1
1016 if k > best and _same_place(nxt.arrival_icao, w.arrival_icao):
1017 best = k
1018 return best
1021def _merge_run(
1022 run: list[FlightSegment], trackpoints: list[TrackPoint], precision: str
1023) -> FlightSegment:
1024 """One FlightSegment spanning *run*, including the ground stops between
1025 its legs."""
1026 times = [tp.utc_dt for tp in trackpoints]
1027 lo = bisect.bisect_left(times, run[0].trackpoints[0].utc_dt)
1028 hi = bisect.bisect_right(times, run[-1].trackpoints[-1].utc_dt)
1029 pts = trackpoints[lo:hi]
1030 block_off = run[0].block_off_utc
1031 block_on = run[-1].block_on_utc
1032 raw_h = (block_on - block_off).total_seconds() / 3600.0
1033 airborne_h = sum(
1034 (s.landing_utc - s.takeoff_utc).total_seconds() / 3600.0
1035 for s in run
1036 if s.takeoff_utc and s.landing_utc
1037 )
1038 return FlightSegment(
1039 trackpoints=pts,
1040 block_off_utc=block_off,
1041 takeoff_utc=run[0].takeoff_utc,
1042 landing_utc=run[-1].landing_utc,
1043 block_on_utc=block_on,
1044 departure_icao=run[0].departure_icao,
1045 arrival_icao=run[-1].arrival_icao,
1046 flight_time_raw_h=raw_h,
1047 flight_time_rounded_h=round_flight_time(raw_h, precision),
1048 track_geojson=build_geojson(downsample_track(pts)),
1049 landing_count=sum(s.landing_count for s in run),
1050 is_ground_only=False,
1051 hint_departure_icao=run[0].hint_departure_icao,
1052 hint_arrival_icao=run[-1].hint_arrival_icao,
1053 airborne_raw_h=airborne_h,
1054 )
1057def merge_segments_logged_as_one(
1058 segments: list[FlightSegment],
1059 trackpoints: list[TrackPoint],
1060 windows: list[LoggedFlightWindow],
1061 precision: str = "tenth_hour",
1062) -> list[FlightSegment]:
1063 """Join consecutive segments that one logbook flight records as a single
1064 flight — e.g. a landing to check something with the engine running,
1065 then on to the destination, logged as one flight with two landings.
1067 Segment detection alone splits at any 5-minute stop; only the logbook
1068 can say the pilot counted it as one flight. Segments are joined when
1069 they all fall within that flight's block times, chain from its
1070 departure to its arrival, and the recording never stopped in between
1071 (a gap means the avionics, and most likely the engine, were switched
1072 off — a separate flight whatever the logbook says).
1073 """
1074 if len(segments) < 2 or not windows:
1075 return segments
1076 times = [tp.utc_dt for tp in trackpoints]
1077 out: list[FlightSegment] = []
1078 i = 0
1079 while i < len(segments):
1080 k = _logged_run_end(segments, times, windows, i)
1081 out.append(
1082 segments[i]
1083 if k == i
1084 else _merge_run(segments[i : k + 1], trackpoints, precision)
1085 )
1086 i = k + 1
1087 return out
1090# ── ICAO resolution ───────────────────────────────────────────────────────────
1093def resolve_icao(
1094 lat: float,
1095 lon: float,
1096 airports: dict[str, tuple[float, float]] | None = None,
1097) -> str | None:
1098 """Return the nearest ICAO code within 5 km, or None if none is close enough."""
1099 if airports is None:
1100 airports = _load_airports()
1102 best_code: str | None = None
1103 best_dist = _MAX_ICAO_DIST_KM
1105 for code, (ap_lat, ap_lon) in airports.items():
1106 d = _haversine_km(lat, lon, ap_lat, ap_lon)
1107 if d < best_dist:
1108 best_dist = d
1109 best_code = code
1111 return best_code
1114# ── Time rounding ─────────────────────────────────────────────────────────────
1117def round_flight_time(raw_hours: float, precision: str) -> float:
1118 """Round raw_hours up to the nearest precision boundary.
1120 precision="tenth_hour": round up to nearest 0.1 h (6-min boundary).
1121 precision="minute": round up to nearest 1/60 h (1-min boundary).
1122 """
1123 if raw_hours <= 0:
1124 return 0.0
1125 if precision == "minute":
1126 minutes = math.ceil(raw_hours * 60)
1127 return round(minutes / 60, 4)
1128 # tenth_hour: ceiling to nearest 0.1
1129 return round(math.ceil(raw_hours * 10) / 10, 1)
1132# ── GeoJSON / downsampling ────────────────────────────────────────────────────
1135@dataclass
1136class SegmentTimes:
1137 """A confirmed GPS segment's two duration tracks (see models.Flight's
1138 docstring): block-off to block-on is the engine side (block time,
1139 feeds the pilot logbook), the detected takeoff-to-landing span is the
1140 airborne side the airframe log's flight_time tracks."""
1142 block_time: Decimal
1143 airborne_time: Decimal | None
1144 takeoff_time: time | None
1145 landing_time: time | None
1148def segment_times(seg: dict[str, Any], precision: str) -> SegmentTimes:
1149 """SegmentTimes for a serialised segment dict (as kept between the
1150 review and confirm steps), rounded to the aircraft's *precision*."""
1151 takeoff = (
1152 datetime.fromisoformat(seg["takeoff_utc"]) if seg.get("takeoff_utc") else None
1153 )
1154 landing = (
1155 datetime.fromisoformat(seg["landing_utc"]) if seg.get("landing_utc") else None
1156 )
1157 airborne: Decimal | None = None
1158 if seg.get("airborne_raw_h") is not None:
1159 airborne = Decimal(str(round_flight_time(seg["airborne_raw_h"], precision)))
1160 elif takeoff is not None and landing is not None:
1161 hours = (landing - takeoff).total_seconds() / 3600.0
1162 airborne = Decimal(str(round_flight_time(hours, precision)))
1163 return SegmentTimes(
1164 block_time=Decimal(
1165 str(round_flight_time(seg.get("flight_time_raw_h", 0), precision))
1166 ),
1167 airborne_time=airborne,
1168 takeoff_time=takeoff.time().replace(tzinfo=None) if takeoff else None,
1169 landing_time=landing.time().replace(tzinfo=None) if landing else None,
1170 )
1173def downsample_track(
1174 trackpoints: list[TrackPoint], max_points: int = _MAX_TRACK_POINTS
1175) -> list[TrackPoint]:
1176 """Return ≤ max_points trackpoints using uniform stride; first and last preserved."""
1177 n = len(trackpoints)
1178 if n <= max_points:
1179 return trackpoints
1181 stride = n / max_points
1182 indices: set[int] = {round(i * stride) for i in range(max_points)}
1183 indices.add(0)
1184 indices.add(n - 1)
1185 return [trackpoints[i] for i in sorted(indices) if i < n]
1188def build_geojson(trackpoints: list[TrackPoint]) -> dict[str, Any]:
1189 """Return a GeoJSON Feature with a LineString geometry.
1191 Coordinates: [lon, lat, alt_m] per GeoJSON spec (RFC 7946).
1192 Properties carry parallel arrays of altitudes_m and speeds_kt for
1193 colour-gradient rendering in Leaflet.
1194 """
1195 coords = [
1196 [round(tp.lon, 6), round(tp.lat, 6), round(tp.alt_m, 1)] for tp in trackpoints
1197 ]
1198 return {
1199 "type": "Feature",
1200 "geometry": {"type": "LineString", "coordinates": coords},
1201 "properties": {
1202 "altitudes_m": [round(tp.alt_m, 1) for tp in trackpoints],
1203 "speeds_kt": [round(tp.speed_kt, 1) for tp in trackpoints],
1204 },
1205 }
1208# ── Near-match scoring against existing Flight rows without GPS block data ───
1211def score_gps_candidates(
1212 fields: dict[str, Any],
1213 candidates: list[Any],
1214 scorer: Callable[[dict[str, Any], Any], float],
1215 min_score: int,
1216) -> list[Any]:
1217 """Rank *candidates* (Flight rows) against a parsed GPS segment's
1218 *fields* dict using *scorer*, best match first.
1220 A GPS segment's block_off_utc/block_on_utc only ever exists on rows
1221 that themselves came from a GPS import — a flight logged manually or
1222 via CSV import (airframe or pilot logbook) has neither, so the exact
1223 block-time overlap check the review routes try first can never find
1224 it. This is the fallback: reuse the exact same near-match scorers the
1225 CSV-import review flows already use where possible
1226 (_score_gps_candidate below, built on
1227 flights.airframe_import._score_airframe_non_time_signals, for
1228 aircraft-linked rows; pilots.logbook_import._score_candidate directly
1229 for standalone ones) so a previously-imported flight is recognised
1230 instead of silently duplicated — rather than inventing a second scoring
1231 implementation that could drift from the first.
1232 """
1233 scored = [
1234 (score, c) for c in candidates if (score := scorer(fields, c)) >= min_score
1235 ]
1236 scored.sort(key=lambda t: -t[0])
1237 return [c for _score, c in scored]
1240def near_date_range(gps_date: date) -> tuple[date, date]:
1241 """First and last logbook date a near-date match may carry."""
1242 span = timedelta(days=NEAR_DATE_DAYS)
1243 return gps_date - span, gps_date + span
1246def rank_near_date_candidates(
1247 fields: dict[str, Any],
1248 gps_date: date,
1249 candidates: list[Any],
1250 scorer: Callable[[dict[str, Any], Any], float],
1251 min_score: int,
1252) -> list[Any]:
1253 """Flights logged on another date close to a GPS segment's — a typo in
1254 a hand-entered logbook date — ranked closest date first, then by score.
1256 The GPS date is the reliable one, so this is only a fallback for a
1257 segment nothing matched on its own date, and stricter than a same-day
1258 match: the route must be known and identical at both ends, and the
1259 date at most NEAR_DATE_DAYS days off. A flight years away never
1260 qualifies, whatever else it shares.
1261 """
1262 dep = fields.get("departure_icao")
1263 arr = fields.get("arrival_icao")
1264 if not dep or not arr:
1265 return []
1266 pool = [
1267 c
1268 for c in candidates
1269 if c.date != gps_date
1270 and abs((c.date - gps_date).days) <= NEAR_DATE_DAYS
1271 and c.departure_icao == dep
1272 and c.arrival_icao == arr
1273 ]
1274 ranked = score_gps_candidates(fields, pool, scorer, min_score)
1275 return sorted(ranked, key=lambda c: abs((c.date - gps_date).days))
1278def _score_gps_candidate(
1279 fields: dict[str, Any], existing: Any, offset_hours: float
1280) -> float:
1281 """Score how likely *existing* (a Flight row) is the same real-world
1282 flight as *fields* (a parsed GPS segment) — the 5 non-time signals from
1283 flights.airframe_import._score_airframe_non_time_signals, plus a
1284 GPS-specific comparison for the other 2.
1286 A GPS track's block_off/block_on record a single observed instant per
1287 end, and — unlike a same-source CSV re-import — there's no guarantee it
1288 lines up tightly with either departure_time (engine start) or
1289 takeoff_time (wheels-up): recording can start before the engine does
1290 (setting up a flight plan on a tablet) and keep running a little after
1291 landing. So instead of preferring one time field over the other, the
1292 full-credit window spans whichever of departure_time/takeoff_time (and
1293 landing_time/arrival_time) the existing row has, widened further on
1294 each side — see services.time_band_matching.widened_span_score.
1295 """
1296 from flights.airframe_import import ( # pyright: ignore[reportMissingImports]
1297 _score_airframe_non_time_signals,
1298 )
1299 from services.time_band_matching import ( # pyright: ignore[reportMissingImports]
1300 offset_ring_step_minutes,
1301 widened_span_score,
1302 )
1304 score = _score_airframe_non_time_signals(fields, existing)
1305 widen_minutes = offset_ring_step_minutes(offset_hours)
1307 takeoff_edges = tuple(
1308 t for t in (existing.departure_time, existing.takeoff_time) if t is not None
1309 )
1310 score += widened_span_score(
1311 fields.get("takeoff_time"), takeoff_edges, widen_minutes
1312 )
1314 landing_edges = tuple(
1315 t for t in (existing.landing_time, existing.arrival_time) if t is not None
1316 )
1317 score += widened_span_score(
1318 fields.get("landing_time"), landing_edges, widen_minutes
1319 )
1321 return score