Coverage for app/aircraft/gps

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 csv

12import functools

13import io

14import itertools

15import math

16import os

17import re

18from collections.abc import Callable

19from dataclasses import dataclass

20from datetime import UTC, datetime

21from typing import Any

22from xml.etree.ElementTree import ParseError as _ETParseError

24import defusedxml.ElementTree as ET # guards against XML bomb / entity expansion

26# ── Constants ─────────────────────────────────────────────────────────────────

28_MS_TO_KT = 1.94384 # m/s → knots

29_FT_TO_M = 0.3048 # ft → metres

30_KM_PER_NM = 1.852 # km per nautical mile

32_FLIGHT_SPEED_KT = 30.0 # sustained above this → airborne

33_GROUND_MOVE_KT = 5.0 # above this (but not 30kt for 30s) → ground movement

34_FLIGHT_SUSTAIN_S = 30.0 # seconds above 30kt required to classify as "flight"

35_SEGMENT_GAP_S = 300.0 # 5 min of slow speed or time gap → segment break

36_MAX_ICAO_DIST_KM = 5.0 # max distance for nearest-airport match

37_MAX_TRACK_POINTS = 500 # downsample threshold for GeoJSON storage

39# ── Data structures ───────────────────────────────────────────────────────────

42@dataclass

43class TrackPoint:

44 lat: float

45 lon: float

46 alt_m: float

47 speed_kt: float

48 utc_dt: datetime # always timezone-aware UTC

51@dataclass

52class FlightSegment:

53 trackpoints: list[TrackPoint]

54 block_off_utc: datetime

55 takeoff_utc: datetime | None

56 landing_utc: datetime | None

57 block_on_utc: datetime

58 departure_icao: str | None

59 arrival_icao: str | None

60 flight_time_raw_h: float # block_on − block_off in decimal hours

61 flight_time_rounded_h: float # rounded per aircraft precision setting

62 track_geojson: dict[str, Any] # GeoJSON Feature

63 landing_count: int

64 is_ground_only: bool # True when no airborne portion detected

65 hint_departure_icao: str | None

66 hint_arrival_icao: str | None

69@dataclass

70class ParsedGpsFile:

71 trackpoints: list[TrackPoint]

72 format: str # "gpx" | "kml" | "garmin_csv"

73 source_filename: str

74 classification: str # "flight" | "ground_movement" | "empty"

75 hint_departure_icao: str | None

76 hint_arrival_icao: str | None

77 device_id: str | None = None # avionics unit identifier (e.g. Garmin system_id)

80# ── Airport database ──────────────────────────────────────────────────────────

83@functools.lru_cache(maxsize=1)

84def _load_airports() -> dict[str, tuple[float, float]]:

85 """Load app/data/airports.csv once. Returns {icao: (lat, lon)}.

87 Only 4-letter ICAO codes are included. Returns an empty dict if the

88 data file is missing (ICAO lookup will return None for all queries).

89 """

90 data_path = os.path.join(os.path.dirname(__file__), "..", "data", "airports.csv")

91 airports: dict[str, tuple[float, float]] = {}

93 if os.path.exists(data_path):

94 with open(data_path, newline="", encoding="utf-8") as f:

95 reader = csv.DictReader(f)

96 for row in reader:

97 ident = row.get("ident", "").strip()

98 if not re.match(r"^[A-Z]{4}$", ident):

99 continue

100 try:

101 lat = float(row["latitude_deg"])

102 lon = float(row["longitude_deg"])

103 except (ValueError, KeyError):

104 continue

105 airports[ident] = (lat, lon)

106

107 return airports

108

109

110def _reset_airports_cache() -> None:

111 """Reset the airport cache (for testing)."""

112 _load_airports.cache_clear()

113

114

115# ── Haversine ─────────────────────────────────────────────────────────────────

116

117

118def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:

119 """Great-circle distance in kilometres."""

120 R = 6371.0

121 phi1 = math.radians(lat1)

122 phi2 = math.radians(lat2)

123 dphi = math.radians(lat2 - lat1)

124 dlambda = math.radians(lon2 - lon1)

125 a = (

126 math.sin(dphi / 2) ** 2

127 + math.cos(phi1) * math.cos(phi2) * math.sin(dlambda / 2) ** 2

128 )

129 return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))

130

131

132# ── Format detection ──────────────────────────────────────────────────────────

133

134

135def detect_format(data: bytes, filename: str) -> str:

136 """Return "gpx", "kml", or "garmin_csv". Raise ValueError for unknown format."""

137 ext = os.path.splitext(filename.lower())[1]

138 if ext == ".gpx":

139 return "gpx"

140 if ext == ".kml":

141 return "kml"

142 if ext == ".csv":

143 lines = data.decode("utf-8-sig", errors="replace").splitlines()

144 if lines and lines[0].startswith("#airframe_info"):

145 return "garmin_csv"

146 raise ValueError(f"Unsupported GPS file format: {filename!r}")

147

148

149# ── File classification ───────────────────────────────────────────────────────

150

151

152def classify_track(trackpoints: list[TrackPoint]) -> str:

153 """Return "flight", "ground_movement", or "empty"."""

154 if not trackpoints:

155 return "empty"

156

157 max_speed = max(tp.speed_kt for tp in trackpoints)

158 if max_speed <= _GROUND_MOVE_KT:

159 return "empty"

160

161 # Check for sustained window above 30kt

162 fast_window_s = 0.0

163 for prev, tp in itertools.pairwise(trackpoints):

164 if tp.speed_kt > _FLIGHT_SPEED_KT:

165 dt = (tp.utc_dt - prev.utc_dt).total_seconds()

166 fast_window_s += max(0.0, dt)

167 if fast_window_s >= _FLIGHT_SUSTAIN_S:

168 return "flight"

169 else:

170 fast_window_s = 0.0

171

172 return "ground_movement"

173

174

175# ── GPX parser ────────────────────────────────────────────────────────────────

176

177_GPX_NS = "http://www.topografix.com/GPX/1/1"

178

179

180def _extract_icao_hints(text: str) -> tuple[str | None, str | None]:

181 """Extract departure and arrival ICAO codes from a track name string."""

182 icao_matches = re.findall(r"\b([A-Z]{4})\b", text)

183 dep = icao_matches[0] if len(icao_matches) >= 1 else None

184 arr = icao_matches[-1] if len(icao_matches) >= 2 else None

185 return dep, arr

186

187

188def _parse_gpx(data: bytes, filename: str) -> ParsedGpsFile:

189 """Parse GPX 1.1 track. Speed field is in m/s; converted to kt."""

190 try:

191 root = ET.fromstring(data.decode("utf-8-sig", errors="replace"))

192 except _ETParseError as exc:

193 raise ValueError(f"Invalid GPX XML in {filename!r}: {exc}") from exc

194

195 hint_dep: str | None = None

196 hint_arr: str | None = None

197 name_el = root.find(f".//{{{_GPX_NS}}}name")

198 if name_el is not None and name_el.text:

199 hint_dep, hint_arr = _extract_icao_hints(name_el.text)

200

201 trackpoints: list[TrackPoint] = []

202 for trkpt in root.findall(f".//{{{_GPX_NS}}}trkpt"):

203 try:

204 lat = float(trkpt.get("lat", ""))

205 lon = float(trkpt.get("lon", ""))

206 except (ValueError, TypeError):

207 continue

208

209 ele_el = trkpt.find(f"{{{_GPX_NS}}}ele")

210 alt_m = float(ele_el.text) if ele_el is not None and ele_el.text else 0.0

211

212 speed_el = trkpt.find(f"{{{_GPX_NS}}}speed")

213 speed_kt = (

214 float(speed_el.text) * _MS_TO_KT

215 if speed_el is not None and speed_el.text

216 else 0.0

217 )

218

219 time_el = trkpt.find(f"{{{_GPX_NS}}}time")

220 if time_el is None or not time_el.text:

221 continue

222 try:

223 utc_dt = datetime.fromisoformat(time_el.text)

224 except ValueError:

225 continue

226

227 trackpoints.append(

228 TrackPoint(lat=lat, lon=lon, alt_m=alt_m, speed_kt=speed_kt, utc_dt=utc_dt)

229 )

230

231 return ParsedGpsFile(

232 trackpoints=trackpoints,

233 format="gpx",

234 source_filename=filename,

235 classification=classify_track(trackpoints),

236 hint_departure_icao=hint_dep,

237 hint_arrival_icao=hint_arr,

238 )

239

240

241# ── Garmin CSV parser ─────────────────────────────────────────────────────────

242

243_VALID_GPS_FIX = {"3D", "3DDiff"}

244

245

246def _parse_garmin_csv(data: bytes, filename: str) -> ParsedGpsFile:

247 """Parse Garmin GTN/G1000 CSV with 3-row header.

248

249 Row 0: #airframe_info metadata

250 Row 1: unit labels

251 Row 2: column names (Lcl Date, Lcl Time, UTCOfst, Latitude, Longitude, AltMSL, GndSpd, …, GPSfix, …)

252 Only rows with GPSfix in {"3D", "3DDiff"} are used.

253 Departure ICAO hint is extracted from filename: log_YYMMDD_HHMMSS_ICAO.csv

254 """

255 text = data.decode("utf-8-sig", errors="replace")

256 lines = text.splitlines()

257

258 if len(lines) < 4:

259 raise ValueError(f"Garmin CSV too short: {filename!r}")

260

261 # Device ID from #airframe_info header line

262 device_id: str | None = None

263 _did_match = re.search(r'system_id="([^"]+)"', lines[0])

264 if _did_match:

265 device_id = _did_match.group(1)

266

267 # Departure ICAO from filename pattern

268 hint_dep: str | None = None

269 base = os.path.splitext(os.path.basename(filename))[0]

270 parts = base.split("_")

271 if len(parts) >= 4:

272 candidate = parts[-1].strip()

273 if re.match(r"^[A-Z]{4}$", candidate):

274 hint_dep = candidate

275

276 # Skip rows 0–1 (metadata + units), use row 2 as header

277 csv_text = "\n".join(lines[2:])

278 reader = csv.DictReader(io.StringIO(csv_text))

279 if reader.fieldnames:

280 reader.fieldnames = [f.strip() for f in reader.fieldnames]

281

282 trackpoints: list[TrackPoint] = []

283 for row in reader:

284 gpsfx = row.get("GPSfix", "").strip()

285 if gpsfx not in _VALID_GPS_FIX:

286 continue

287

288 try:

289 lat = float(row["Latitude"].strip())

290 lon = float(row["Longitude"].strip())

291 except (ValueError, KeyError):

292 continue

293

294 try:

295 alt_m = float(row["AltMSL"].strip()) * _FT_TO_M

296 except (ValueError, KeyError):

297 alt_m = 0.0

298

299 try:

300 speed_kt = float(row["GndSpd"].strip())

301 except (ValueError, KeyError):

302 speed_kt = 0.0

303

304 try:

305 date_str = row["Lcl Date"].strip()

306 time_str = row["Lcl Time"].strip()

307 utc_off = row["UTCOfst"].strip()

308 local_dt = datetime.fromisoformat(f"{date_str}T{time_str}{utc_off}")

309 utc_dt = local_dt.astimezone(UTC)

310 except (ValueError, KeyError):

311 continue

312

313 trackpoints.append(

314 TrackPoint(lat=lat, lon=lon, alt_m=alt_m, speed_kt=speed_kt, utc_dt=utc_dt)

315 )

316

317 return ParsedGpsFile(

318 trackpoints=trackpoints,

319 format="garmin_csv",

320 source_filename=filename,

321 classification=classify_track(trackpoints),

322 hint_departure_icao=hint_dep,

323 hint_arrival_icao=None,

324 device_id=device_id,

325 )

326

327

328# ── KML parser ────────────────────────────────────────────────────────────────

329

330_KML_NS = "http://www.opengis.net/kml/2.2"

331_GX_NS = "http://www.google.com/kml/ext/2.2"

332

333

334def _parse_kml(data: bytes, filename: str) -> ParsedGpsFile:

335 """Parse SkyDemon KML with gx:Track.

336

337 Coordinate order is lon/lat/alt (note: reversed from GPX).

338 Speed is derived from consecutive point distance / time delta.

339 """

340 try:

341 root = ET.fromstring(data.decode("utf-8-sig", errors="replace"))

342 except _ETParseError as exc:

343 raise ValueError(f"Invalid KML XML in {filename!r}: {exc}") from exc

344

345 hint_dep: str | None = None

346 hint_arr: str | None = None

347 for pm in root.findall(f".//{{{_KML_NS}}}Placemark"):

348 name_el = pm.find(f"{{{_KML_NS}}}name")

349 if name_el is not None and name_el.text:

350 dep, arr = _extract_icao_hints(name_el.text)

351 if dep and arr:

352 hint_dep, hint_arr = dep, arr

353 break

354

355 track_el = root.find(f".//{{{_GX_NS}}}Track")

356 if track_el is None:

357 raise ValueError(f"No gx:Track element in KML: {filename!r}")

358

359 whens: list[datetime | None] = []

360 coords: list[tuple[float, float, float]] = []

361

362 for child in track_el:

363 if child.tag == f"{{{_KML_NS}}}when":

364 if child.text:

365 try:

366 dt = datetime.fromisoformat(child.text)

367 whens.append(dt.astimezone(UTC))

368 except ValueError:

369 whens.append(None)

370 else:

371 whens.append(None)

372 elif child.tag == f"{{{_GX_NS}}}coord":

373 if child.text:

374 parts = child.text.strip().split()

375 if len(parts) >= 3:

376 try:

377 lon_c = float(parts[0])

378 lat_c = float(parts[1])

379 alt_c = float(parts[2])

380 except ValueError:

381 lon_c, lat_c, alt_c = 0.0, 0.0, 0.0

382 coords.append((lon_c, lat_c, alt_c))

383 continue

384 coords.append((0.0, 0.0, 0.0))

385

386 if len(whens) != len(coords):

387 raise ValueError(

388 f"KML when/coord count mismatch in {filename!r}: "

389 f"{len(whens)} vs {len(coords)}"

390 )

391

392 trackpoints: list[TrackPoint] = []

393 for i, (when, (lon, lat, alt_m)) in enumerate(zip(whens, coords)):

394 if when is None:

395 continue

396

397 if trackpoints:

398 prev = trackpoints[-1]

399 dt_s = (when - prev.utc_dt).total_seconds()

400 dist_km = _haversine_km(prev.lat, prev.lon, lat, lon)

401 speed_kt = (dist_km / _KM_PER_NM * 3600.0 / dt_s) if dt_s > 0 else 0.0

402 else:

403 speed_kt = 0.0

404

405 trackpoints.append(

406 TrackPoint(lat=lat, lon=lon, alt_m=alt_m, speed_kt=speed_kt, utc_dt=when)

407 )

408

409 return ParsedGpsFile(

410 trackpoints=trackpoints,

411 format="kml",

412 source_filename=filename,

413 classification=classify_track(trackpoints),

414 hint_departure_icao=hint_dep,

415 hint_arrival_icao=hint_arr,

416 )

417

418

419# ── Entry point ───────────────────────────────────────────────────────────────

420

421

422def parse_gps_file(data: bytes, filename: str) -> ParsedGpsFile:

423 """Detect format and parse. Raises ValueError on unsupported or invalid data."""

424 fmt = detect_format(data, filename)

425 if fmt == "gpx":

426 return _parse_gpx(data, filename)

427 if fmt == "kml":

428 return _parse_kml(data, filename)

429 return _parse_garmin_csv(data, filename)

430

431

432# ── Track merge ───────────────────────────────────────────────────────────────

433

434

435def merge_and_sort(files: list[ParsedGpsFile]) -> list[TrackPoint]:

436 """Merge non-empty trackpoints from all files, sorted chronologically."""

437 all_pts: list[TrackPoint] = []

438 for f in files:

439 if f.classification != "empty":

440 all_pts.extend(f.trackpoints)

441 all_pts.sort(key=lambda tp: tp.utc_dt)

442 return all_pts

443

444

445# ── Segment detection ─────────────────────────────────────────────────────────

446

447

448def _split_into_raw_groups(trackpoints: list[TrackPoint]) -> list[list[TrackPoint]]:

449 """Split merged trackpoints into groups at slow/time gaps ≥ 5 min.

450

451 Only looks for breaks between the first and last fast (≥ 30kt) points, so

452 pre-flight taxi and post-landing taxi are preserved in the enclosing segment.

453 """

454 n = len(trackpoints)

455 if n == 0:

456 return []

457

458 fast_indices = [

459 i for i, tp in enumerate(trackpoints) if tp.speed_kt >= _FLIGHT_SPEED_KT

460 ]

461 if not fast_indices:

462 return [trackpoints]

463

464 first_fast = fast_indices[0]

465 last_fast = fast_indices[-1]

466

467 groups: list[list[TrackPoint]] = []

468 current_start = 0

469 i = first_fast

470

471 while i < last_fast:

472 # Large time gap between consecutive points (gap between uploaded files)

473 time_gap = (trackpoints[i + 1].utc_dt - trackpoints[i].utc_dt).total_seconds()

474 if time_gap >= _SEGMENT_GAP_S:

475 groups.append(trackpoints[current_start : i + 1])

476 current_start = i + 1

477 i += 1

478 continue

479

480 # Slow run starting at i+1

481 if trackpoints[i + 1].speed_kt < _FLIGHT_SPEED_KT:

482 j = i + 2

483 while j <= last_fast and trackpoints[j].speed_kt < _FLIGHT_SPEED_KT:

484 j += 1

485 slow_dur = (

486 trackpoints[j - 1].utc_dt - trackpoints[i + 1].utc_dt

487 ).total_seconds()

488 if slow_dur >= _SEGMENT_GAP_S:

489 # Real segment break — exclude slow gap from both segments

490 groups.append(trackpoints[current_start : i + 1])

491 current_start = j

492 i = j

493 else:

494 i = j # short slow run — keep in current segment

495 else:

496 i += 1

497

498 groups.append(trackpoints[current_start:])

499 return [g for g in groups if g]

500

501

502def _count_landings(pts: list[TrackPoint]) -> int:

503 """Count transitions from airborne (≥30kt) to ground (<30kt)."""

504 count = 0

505 was_fast = False

506 for tp in pts:

507 is_fast = tp.speed_kt >= _FLIGHT_SPEED_KT

508 if was_fast and not is_fast:

509 count += 1

510 was_fast = is_fast

511 return count

512

513

514def detect_segments(

515 trackpoints: list[TrackPoint],

516 aircraft_precision: str = "tenth_hour",

517 hint_dep: str | None = None,

518 hint_arr: str | None = None,

519) -> list[FlightSegment]:

520 """Build FlightSegment objects from merged trackpoints.

521

522 hint_dep / hint_arr are optional ICAO codes from GPX/KML track names or

523 Garmin filename patterns, used as fallback when GPS-proximity lookup fails.

524 """

525 raw_groups = _split_into_raw_groups(trackpoints)

526 airports = _load_airports()

527 segments: list[FlightSegment] = []

528

529 for idx, pts in enumerate(raw_groups):

530 block_off = pts[0].utc_dt

531 block_on = pts[-1].utc_dt

532

533 takeoff_utc: datetime | None = None

534 landing_utc: datetime | None = None

535 for tp in pts:

536 if tp.speed_kt >= _FLIGHT_SPEED_KT:

537 if takeoff_utc is None:

538 takeoff_utc = tp.utc_dt

539 landing_utc = tp.utc_dt

540

541 is_ground_only = takeoff_utc is None

542 landing_count = _count_landings(pts)

543

544 raw_h = (block_on - block_off).total_seconds() / 3600.0

545 rounded_h = round_flight_time(raw_h, aircraft_precision)

546

547 dep_icao = resolve_icao(pts[0].lat, pts[0].lon, airports) or (

548 hint_dep if idx == 0 else None

549 )

550 arr_icao = resolve_icao(pts[-1].lat, pts[-1].lon, airports) or (

551 hint_arr if idx == len(raw_groups) - 1 else None

552 )

553

554 downsampled = downsample_track(pts)

555 geojson = build_geojson(downsampled)

556

557 segments.append(

558 FlightSegment(

559 trackpoints=pts,

560 block_off_utc=block_off,

561 takeoff_utc=takeoff_utc,

562 landing_utc=landing_utc,

563 block_on_utc=block_on,

564 departure_icao=dep_icao,

565 arrival_icao=arr_icao,

566 flight_time_raw_h=raw_h,

567 flight_time_rounded_h=rounded_h,

568 track_geojson=geojson,

569 landing_count=landing_count,

570 is_ground_only=is_ground_only,

571 hint_departure_icao=hint_dep if idx == 0 else None,

572 hint_arrival_icao=hint_arr if idx == len(raw_groups) - 1 else None,

573 )

574 )

575

576 return segments

577

578

579# ── ICAO resolution ───────────────────────────────────────────────────────────

580

581

582def resolve_icao(

583 lat: float,

584 lon: float,

585 airports: dict[str, tuple[float, float]] | None = None,

586) -> str | None:

587 """Return the nearest ICAO code within 5 km, or None if none is close enough."""

588 if airports is None:

589 airports = _load_airports()

590

591 best_code: str | None = None

592 best_dist = _MAX_ICAO_DIST_KM

593

594 for code, (ap_lat, ap_lon) in airports.items():

595 d = _haversine_km(lat, lon, ap_lat, ap_lon)

596 if d < best_dist:

597 best_dist = d

598 best_code = code

599

600 return best_code

601

602

603# ── Time rounding ─────────────────────────────────────────────────────────────

604

605

606def round_flight_time(raw_hours: float, precision: str) -> float:

607 """Round raw_hours up to the nearest precision boundary.

608

609 precision="tenth_hour": round up to nearest 0.1 h (6-min boundary).

610 precision="minute": round up to nearest 1/60 h (1-min boundary).

611 """

612 if raw_hours <= 0:

613 return 0.0

614 if precision == "minute":

615 minutes = math.ceil(raw_hours * 60)

616 return round(minutes / 60, 4)

617 # tenth_hour: ceiling to nearest 0.1

618 return round(math.ceil(raw_hours * 10) / 10, 1)

619

620

621# ── GeoJSON / downsampling ────────────────────────────────────────────────────

622

623

624def downsample_track(

625 trackpoints: list[TrackPoint], max_points: int = _MAX_TRACK_POINTS

626) -> list[TrackPoint]:

627 """Return ≤ max_points trackpoints using uniform stride; first and last preserved."""

628 n = len(trackpoints)

629 if n <= max_points:

630 return trackpoints

631

632 stride = n / max_points

633 indices: set[int] = {round(i * stride) for i in range(max_points)}

634 indices.add(0)

635 indices.add(n - 1)

636 return [trackpoints[i] for i in sorted(indices) if i < n]

637

638

639def build_geojson(trackpoints: list[TrackPoint]) -> dict[str, Any]:

640 """Return a GeoJSON Feature with a LineString geometry.

641

642 Coordinates: [lon, lat, alt_m] per GeoJSON spec (RFC 7946).

643 Properties carry parallel arrays of altitudes_m and speeds_kt for

644 colour-gradient rendering in Leaflet.

645 """

646 coords = [

647 [round(tp.lon, 6), round(tp.lat, 6), round(tp.alt_m, 1)] for tp in trackpoints

648 ]

649 return {

650 "type": "Feature",

651 "geometry": {"type": "LineString", "coordinates": coords},

652 "properties": {

653 "altitudes_m": [round(tp.alt_m, 1) for tp in trackpoints],

654 "speeds_kt": [round(tp.speed_kt, 1) for tp in trackpoints],

655 },

656 }

657

658

659# ── Near-match scoring against existing Flight rows without GPS block data ───

660

661

662def score_gps_candidates(

663 fields: dict[str, Any],

664 candidates: list[Any],

665 scorer: Callable[[dict[str, Any], Any], float],

666 min_score: int,

667) -> list[Any]:

668 """Rank *candidates* (Flight rows) against a parsed GPS segment's

669 *fields* dict using *scorer*, best match first.

670

671 A GPS segment's block_off_utc/block_on_utc only ever exists on rows

672 that themselves came from a GPS import — a flight logged manually or

673 via CSV import (airframe or pilot logbook) has neither, so the exact

674 block-time overlap check the review routes try first can never find

675 it. This is the fallback: reuse the exact same near-match scorers the

676 CSV-import review flows already use where possible

677 (_score_gps_candidate below, built on

678 flights.airframe_import._score_airframe_non_time_signals, for

679 aircraft-linked rows; pilots.logbook_import._score_candidate directly

680 for standalone ones) so a previously-imported flight is recognised

681 instead of silently duplicated — rather than inventing a second scoring

682 implementation that could drift from the first.

683 """

684 scored = [

685 (score, c) for c in candidates if (score := scorer(fields, c)) >= min_score

686 ]

687 scored.sort(key=lambda t: -t[0])

688 return [c for _score, c in scored]

689

690

691def _score_gps_candidate(

692 fields: dict[str, Any], existing: Any, offset_hours: float

693) -> float:

694 """Score how likely *existing* (a Flight row) is the same real-world

695 flight as *fields* (a parsed GPS segment) — the 5 non-time signals from

696 flights.airframe_import._score_airframe_non_time_signals, plus a

697 GPS-specific comparison for the other 2.

698

699 A GPS track's block_off/block_on record a single observed instant per

700 end, and — unlike a same-source CSV re-import — there's no guarantee it

701 lines up tightly with either departure_time (engine start) or

702 takeoff_time (wheels-up): recording can start before the engine does

703 (setting up a flight plan on a tablet) and keep running a little after

704 landing. So instead of preferring one time field over the other, the

705 full-credit window spans whichever of departure_time/takeoff_time (and

706 landing_time/arrival_time) the existing row has, widened further on

707 each side — see services.time_band_matching.widened_span_score.

708 """

709 from flights.airframe_import import ( # pyright: ignore[reportMissingImports]

710 _score_airframe_non_time_signals,

711 )

712 from services.time_band_matching import ( # pyright: ignore[reportMissingImports]

713 offset_ring_step_minutes,

714 widened_span_score,

715 )

716

717 score = _score_airframe_non_time_signals(fields, existing)

718 widen_minutes = offset_ring_step_minutes(offset_hours)

719

720 takeoff_edges = tuple(

721 t for t in (existing.departure_time, existing.takeoff_time) if t is not None

722 )

723 score += widened_span_score(

724 fields.get("takeoff_time"), takeoff_edges, widen_minutes

725 )

726

727 landing_edges = tuple(

728 t for t in (existing.landing_time, existing.arrival_time) if t is not None

729 )

730 score += widened_span_score(

731 fields.get("landing_time"), landing_edges, widen_minutes

732 )

733

734 return score

Coverage for app/aircraft/gps_import.py: 100%

374 statements