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

1from datetime import date as _date 

2from datetime import timedelta 

3from typing import Any 

4 

5WINDOW_DAYS = 90 

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

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

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

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

10 

11STATUS_OK = "ok" 

12STATUS_WARNING = "warning" 

13STATUS_EXPIRED = "expired" 

14STATUS_UNKNOWN = "unknown" 

15 

16 

17def _rolling_landing_currency( 

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

19) -> dict[str, Any]: 

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

21 if isinstance(landing_fields, str): 

22 landing_fields = (landing_fields,) 

23 

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

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

26 

27 window_start = today - timedelta(days=WINDOW_DAYS) 

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

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

30 

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

32 shortfall = max(0, required - total) 

33 

34 if total >= required: 

35 cum = 0 

36 anchor_date: _date | None = None 

37 for e in qualifying: 

38 cum += _count(e) 

39 if cum >= required: 

40 anchor_date = e.date 

41 break 

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

43 expires_on = anchor_date + timedelta(days=WINDOW_DAYS) 

44 days_left = (expires_on - today).days 

45 status = STATUS_WARNING if days_left <= CURRENCY_WARN_DAYS else STATUS_OK 

46 else: 

47 expires_on = None 

48 days_left = None 

49 status = STATUS_EXPIRED if qualifying else STATUS_UNKNOWN 

50 

51 return { 

52 "count": total, 

53 "required": required, 

54 "status": status, 

55 "expires_on": expires_on, 

56 "days_left": days_left, 

57 "shortfall": shortfall, 

58 } 

59 

60 

61def _expiry_status( 

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

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

64 if expiry_date is None: 

65 return STATUS_UNKNOWN, None 

66 days = (expiry_date - today).days 

67 if days < 0: 

68 return STATUS_EXPIRED, days 

69 if days <= warn_days: 

70 return STATUS_WARNING, days 

71 return STATUS_OK, days 

72 

73 

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

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

76 

77 EASA FCL.060 per type: 

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

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

80 

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

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

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

84 

85 Returns:: 

86 

87 { 

88 "by_type": { 

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

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

91 }, 

92 "unresolved_count": 3, 

93 } 

94 """ 

95 if today is None: 

96 today = _date.today() 

97 

98 from utils import ( 

99 resolve_aircraft_type_icao, # pyright: ignore[reportMissingImports] 

100 ) 

101 

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

103 unresolved_count = 0 

104 

105 for entry in entries: 

106 icao: str | None = getattr(entry, "display_aircraft_type_icao", None) or None 

107 if not icao: 

108 icao = resolve_aircraft_type_icao( 

109 getattr(entry, "display_aircraft_type", None) 

110 ) 

111 if not icao: 

112 unresolved_count += 1 

113 continue 

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

115 

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

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

118 pax = _rolling_landing_currency( 

119 type_entries, 

120 ("landings_day", "landings_night"), 

121 PASSENGER_REQUIRED, 

122 today, 

123 ) 

124 night = _rolling_landing_currency( 

125 type_entries, "landings_night", NIGHT_REQUIRED, today 

126 ) 

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

128 if STATUS_EXPIRED in statuses: 

129 status = STATUS_EXPIRED 

130 elif STATUS_WARNING in statuses: 

131 status = STATUS_WARNING 

132 elif STATUS_UNKNOWN in statuses: 

133 status = STATUS_UNKNOWN 

134 else: 

135 status = STATUS_OK 

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

137 

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

139 

140 

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

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

143 if today is None: 

144 today = _date.today() 

145 return _rolling_landing_currency( 

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

147 ) 

148 

149 

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

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

152 if today is None: 

153 today = _date.today() 

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

155 

156 

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

158 if today is None: 

159 today = _date.today() 

160 expiry = profile.medical_expiry if profile else None 

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

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

163 

164 

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

166 if today is None: 

167 today = _date.today() 

168 expiry = profile.sep_expiry if profile else None 

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

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

171 

172 

173def currency_summary( 

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

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

176 """ 

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

178 """ 

179 if profile is None: 

180 return None 

181 if today is None: 

182 today = _date.today() 

183 

184 pax = passenger_currency(entries, today) 

185 nite = night_currency(entries, today) 

186 med = medical_status(profile, today) 

187 sep = sep_status(profile, today) 

188 per_type = per_type_currency(entries, today) 

189 

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

191 if STATUS_EXPIRED in statuses: 

192 overall = STATUS_EXPIRED 

193 elif STATUS_WARNING in statuses or STATUS_UNKNOWN in statuses: 

194 overall = STATUS_WARNING 

195 else: 

196 overall = STATUS_OK 

197 

198 return { 

199 "passenger": pax, 

200 "night": nite, 

201 "medical": med, 

202 "sep": sep, 

203 "per_type": per_type, 

204 "overall": overall, 

205 }