Coverage for app/services/time_band_matching.py: 100%
30 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-31 10:13 +0000
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-31 10:13 +0000
1"""
2Shared graduated time-proximity scoring for near-match candidate detection,
3used by pilot-logbook import, airframe-logbook import, and GPS-track import
4review flows — one shared implementation instead of three that could drift.
6Two kinds of comparison show up across all three:
7 - "same-pair": a freshly parsed row's time against an existing row's time
8 of the *same* kind (pilot departure_time vs departure_time, airframe
9 takeoff_time vs takeoff_time) — expected to match closely, since it's
10 (if nothing changed) the same real-world instant reported twice.
11 - "cross-pair"/span: comparing across the pilot-log-facing
12 (departure_time/arrival_time, engine start/end) and airframe-log-facing
13 (takeoff_time/landing_time, wheels-up/down) pairs, which are never the
14 same instant. Used when reconciling an import against a placeholder
15 created from the other side (only the aircraft's own
16 flight_counter_offset is available as an estimate of the gap between
17 them), and by GPS matching (which spans both edges when known, since a
18 GPS track can start recording before engine start).
19"""
21from __future__ import annotations
23import math
24from datetime import time
26# Fallback for offset_ring_step_minutes when no aircraft-specific
27# flight_counter_offset is available (e.g. same-pair matching against a
28# standalone entry with no managed aircraft) — matches Aircraft.flight_counter_offset's
29# own column default in models.py.
30DEFAULT_OFFSET_HOURS = 0.3
33def offset_ring_step_minutes(offset_hours: float) -> int:
34 """Ring width for offset-derived graduated time bands: 1/3 of the
35 aircraft's flight_counter_offset, rounded up to a whole minute (minimum
36 1 minute, so a 0-offset aircraft still gets a usable band)."""
37 return max(1, math.ceil(offset_hours * 60 / 3))
40def _minutes(t: time) -> int:
41 return t.hour * 60 + t.minute
44def shift_time(t: time, delta_minutes: int) -> time:
45 """*t* plus/minus *delta_minutes*, wrapping around midnight."""
46 total = (_minutes(t) + delta_minutes) % 1440
47 return time(total // 60, total % 60)
50def time_band_score(
51 new_time: time | None,
52 band_edges: tuple[time, ...],
53 step_minutes: int,
54) -> float:
55 """Graduated proximity score for *new_time* against the inclusive
56 [min(band_edges), max(band_edges)] window.
58 Pass one edge for a zero-width/point band (same-pair matching, or
59 cross-pair matching centred on a reference time shifted by an offset),
60 or two for a real span (GPS matching between departure_time and
61 takeoff_time). *band_edges* may be empty when nothing on the existing
62 side is available to compare against.
64 Full credit (1.0) inside the window; degrades in two step_minutes-wide
65 rings (0.75, then 0.5) outside it; 0 beyond that.
67 Distance is computed in linear minutes-of-day, so a band that straddles
68 midnight isn't handled specially — a pre-existing limitation carried
69 over from the flat-tolerance check this replaces, and immaterial for
70 the short local flights these imports cover.
71 """
72 if new_time is None or not band_edges:
73 return 0.0
74 new_m = _minutes(new_time)
75 edge_ms = [_minutes(t) for t in band_edges]
76 lo, hi = min(edge_ms), max(edge_ms)
77 if lo <= new_m <= hi:
78 return 1.0
79 dist = lo - new_m if new_m < lo else new_m - hi
80 if dist <= step_minutes:
81 return 0.75
82 if dist <= step_minutes * 2:
83 return 0.5
84 return 0.0
87def widened_span_score(
88 new_time: time | None,
89 reference_edges: tuple[time, ...],
90 widen_minutes: int,
91) -> float:
92 """time_band_score, but the core credit window is *reference_edges*
93 widened outward by widen_minutes on each side first, before the two
94 further widen_minutes-wide rings are applied.
96 For GPS-track matching: reference_edges is whichever of
97 (departure_time, takeoff_time) — or (landing_time, arrival_time) — the
98 existing row has. A GPS track can start recording before engine start
99 (setting up a flight plan on a tablet) or after, so the credit window
100 needs to span both known edges rather than being anchored to one; if
101 only one edge is known, the window is still widened around it rather
102 than requiring an exact match, since a single GPS-detected instant is
103 less precise than a logged clock time either way. reference_edges may
104 be empty when the existing row has neither edge.
105 """
106 if not reference_edges:
107 return 0.0
108 widened = tuple(shift_time(t, -widen_minutes) for t in reference_edges) + tuple(
109 shift_time(t, widen_minutes) for t in reference_edges
110 )
111 return time_band_score(new_time, widened, widen_minutes)