Coverage for app/services/time_band_matching.py: 100%

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. 

5 

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""" 

20 

21from __future__ import annotations 

22 

23import math 

24from datetime import time 

25 

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 

31 

32 

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)) 

38 

39 

40def _minutes(t: time) -> int: 

41 return t.hour * 60 + t.minute 

42 

43 

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) 

48 

49 

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. 

57 

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. 

63 

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. 

66 

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 

85 

86 

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. 

95 

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) 

112 

113 

114def taxi_out_score( 

115 block_off: time | None, takeoff: time | None, offset_hours: float 

116) -> float: 

117 """Cross-pair score for a block-off time (pilot-log departure_time) 

118 against a takeoff time (airframe-log takeoff_time) of the same flight. 

119 

120 The aircraft's flight_counter_offset is the *total* block-to-airborne 

121 gap (taxi-out plus taxi-in), so block-off falls anywhere from the whole 

122 offset before takeoff up to takeoff itself: full credit inside 

123 [takeoff - offset, takeoff], degrading in time_band_score's rings 

124 outside it. (Shifting each edge by the whole offset instead — as this 

125 used to — counts that gap twice, so a real flight with an ordinary 

126 6-minute taxi each way ended up two rings out on both edges.) 

127 Symmetric: the same window, seen from the takeoff side, is 

128 [block_off, block_off + offset]. 

129 """ 

130 if block_off is None or takeoff is None: 

131 return 0.0 

132 offset_minutes = round(offset_hours * 60) 

133 return time_band_score( 

134 block_off, 

135 (shift_time(takeoff, -offset_minutes), takeoff), 

136 offset_ring_step_minutes(offset_hours), 

137 ) 

138 

139 

140def taxi_in_score( 

141 landing: time | None, block_on: time | None, offset_hours: float 

142) -> float: 

143 """Cross-pair score for a landing time (airframe log) against a block-on 

144 time (pilot-log arrival_time): full credit inside 

145 [landing, landing + offset] — see taxi_out_score.""" 

146 if landing is None or block_on is None: 

147 return 0.0 

148 offset_minutes = round(offset_hours * 60) 

149 return time_band_score( 

150 block_on, 

151 (landing, shift_time(landing, offset_minutes)), 

152 offset_ring_step_minutes(offset_hours), 

153 )