Coverage for app/pilots/currency.py: 100%

1from datetime import date as _date, timedelta 

2from typing import Any 

3 

4WINDOW_DAYS = 90 

5PASSENGER_REQUIRED = 3 # FCL.060(b)(1): 3 take-offs/landings (any) in 90 days 

6NIGHT_REQUIRED = 1 # FCL.060(b)(2): 1 night landing in 90 days 

7EXPIRY_WARN_DAYS = 90 # warn when medical/SEP expires within this many days 

8CURRENCY_WARN_DAYS = 30 # warn when landing currency expires within this many days 

9 

10STATUS_OK = "ok" 

11STATUS_WARNING = "warning" 

12STATUS_EXPIRED = "expired" 

13STATUS_UNKNOWN = "unknown" 

14 

15 

16def _rolling_landing_currency( 

17 entries: Any, landing_fields: str | tuple[str, ...], required: int, today: _date 

18) -> dict[str, Any]: 

19 """Compute rolling currency for one or more landing fields summed together.""" 

20 if isinstance(landing_fields, str): 

21 landing_fields = (landing_fields,) 

22 

23 def _count(e: Any) -> int: 

24 return sum(getattr(e, f) or 0 for f in landing_fields) 

25 

26 window_start = today - timedelta(days=WINDOW_DAYS) 

27 qualifying = [e for e in entries if e.date >= window_start and _count(e) > 0] 

28 qualifying.sort(key=lambda e: (e.date, e.id), reverse=True) 

29 

30 total = sum(_count(e) for e in qualifying) 

31 shortfall = max(0, required - total) 

32 

33 if total >= required: 

34 cum = 0 

35 anchor_date: _date | None = None 

36 for e in qualifying: 

37 cum += _count(e) 

38 if cum >= required: 

39 anchor_date = e.date 

40 break 

41 assert anchor_date is not None # nosec B101 # mypy narrowing invariant 

42 expires_on = anchor_date + timedelta(days=WINDOW_DAYS) 

43 days_left = (expires_on - today).days 

44 status = STATUS_WARNING if days_left <= CURRENCY_WARN_DAYS else STATUS_OK 

45 else: 

46 expires_on = None 

47 days_left = None 

48 status = STATUS_EXPIRED if qualifying else STATUS_UNKNOWN 

49 

50 return { 

51 "count": total, 

52 "required": required, 

53 "status": status, 

54 "expires_on": expires_on, 

55 "days_left": days_left, 

56 "shortfall": shortfall, 

57 } 

58 

59 

60def _expiry_status( 

61 expiry_date: _date | None, today: _date, warn_days: int 

62) -> tuple[str, int | None]: 

63 if expiry_date is None: 

64 return STATUS_UNKNOWN, None 

65 days = (expiry_date - today).days 

66 if days < 0: 

67 return STATUS_EXPIRED, days 

68 if days <= warn_days: 

69 return STATUS_WARNING, days 

70 return STATUS_OK, days 

71 

72 

73def per_type_currency(entries: Any, today: _date | None = None) -> dict[str, Any]: 

74 """Rolling 90-day landing currency grouped by ICAO aircraft type. 

75 

76 EASA FCL.060 per type: 

77 - Passenger carry: 3 landings (day OR night combined) in 90 days. 

78 - Night passenger carry: 1 night landing in 90 days. 

79 

80 Entries whose aircraft_type_icao is blank are resolved on-the-fly from 

81 aircraft_type via resolve_aircraft_type_icao(). Those that still cannot be 

82 resolved are tallied in unresolved_count so the UI can surface a warning. 

83 

84 Returns:: 

85 

86 { 

87 "by_type": { 

88 "C172": {"passenger": {...}, "night": {...}, "status": "ok"}, 

89 "P28A": {"passenger": {...}, "night": {...}, "status": "warning"}, 

90 }, 

91 "unresolved_count": 3, 

92 } 

93 """ 

94 if today is None: 

95 today = _date.today() 

96 

97 from utils import resolve_aircraft_type_icao # pyright: ignore[reportMissingImports] 

98 

99 buckets: dict[str, list[Any]] = {} 

100 unresolved_count = 0 

101 

102 for entry in entries: 

103 icao: str | None = getattr(entry, "aircraft_type_icao", None) or None 

104 if not icao: 

105 icao = resolve_aircraft_type_icao(getattr(entry, "aircraft_type", None)) 

106 if not icao: 

107 unresolved_count += 1 

108 continue 

109 buckets.setdefault(icao, []).append(entry) 

110 

111 by_type: dict[str, dict[str, Any]] = {} 

112 for icao, type_entries in sorted(buckets.items()): 

113 pax = _rolling_landing_currency( 

114 type_entries, 

115 ("landings_day", "landings_night"), 

116 PASSENGER_REQUIRED, 

117 today, 

118 ) 

119 night = _rolling_landing_currency( 

120 type_entries, "landings_night", NIGHT_REQUIRED, today 

121 ) 

122 statuses = [pax["status"], night["status"]] 

123 if STATUS_EXPIRED in statuses: 

124 status = STATUS_EXPIRED 

125 elif STATUS_WARNING in statuses: 

126 status = STATUS_WARNING 

127 elif STATUS_UNKNOWN in statuses: 

128 status = STATUS_UNKNOWN 

129 else: 

130 status = STATUS_OK 

131 by_type[icao] = {"passenger": pax, "night": night, "status": status} 

132 

133 return {"by_type": by_type, "unresolved_count": unresolved_count} 

134 

135 

136def passenger_currency(entries: Any, today: _date | None = None) -> dict[str, Any]: 

137 """3 landings (day or night) in rolling 90-day window (EASA FCL.060 passenger carry).""" 

138 if today is None: 

139 today = _date.today() 

140 return _rolling_landing_currency( 

141 entries, ("landings_day", "landings_night"), PASSENGER_REQUIRED, today 

142 ) 

143 

144 

145def night_currency(entries: Any, today: _date | None = None) -> dict[str, Any]: 

146 """1 night landing in rolling 90-day window (EASA FCL.060 night passenger carry).""" 

147 if today is None: 

148 today = _date.today() 

149 return _rolling_landing_currency(entries, "landings_night", NIGHT_REQUIRED, today) 

150 

151 

152def medical_status(profile: Any, today: _date | None = None) -> dict[str, Any]: 

153 if today is None: 

154 today = _date.today() 

155 expiry = profile.medical_expiry if profile else None 

156 status, days = _expiry_status(expiry, today, EXPIRY_WARN_DAYS) 

157 return {"expiry": expiry, "status": status, "days_remaining": days} 

158 

159 

160def sep_status(profile: Any, today: _date | None = None) -> dict[str, Any]: 

161 if today is None: 

162 today = _date.today() 

163 expiry = profile.sep_expiry if profile else None 

164 status, days = _expiry_status(expiry, today, EXPIRY_WARN_DAYS) 

165 return {"expiry": expiry, "status": status, "days_remaining": days} 

166 

167 

168def currency_summary( 

169 profile: Any, entries: Any, today: _date | None = None 

170) -> dict[str, Any] | None: 

171 """ 

172 Aggregate all currency checks. Returns None if profile is None. 

173 """ 

174 if profile is None: 

175 return None 

176 if today is None: 

177 today = _date.today() 

178 

179 pax = passenger_currency(entries, today) 

180 nite = night_currency(entries, today) 

181 med = medical_status(profile, today) 

182 sep = sep_status(profile, today) 

183 per_type = per_type_currency(entries, today) 

184 

185 statuses = [pax["status"], nite["status"], med["status"], sep["status"]] 

186 if STATUS_EXPIRED in statuses: 

187 overall = STATUS_EXPIRED 

188 elif STATUS_WARNING in statuses or STATUS_UNKNOWN in statuses: 

189 overall = STATUS_WARNING 

190 else: 

191 overall = STATUS_OK 

192 

193 return { 

194 "passenger": pax, 

195 "night": nite, 

196 "medical": med, 

197 "sep": sep, 

198 "per_type": per_type, 

199 "overall": overall, 

200 }