Coverage for app/reports/utilization.py: 100%

1"""Backlog: annual utilization & insurance-renewal summary. 

2 

3Pure calculation functions, kept independent of Flask routing so they can be 

4unit-tested deterministically (mirrors the style of expenses/cost_dashboard.py, 

5whose resolve_period() this module reuses directly). 

6""" 

7 

8from datetime import date as _date 

9from datetime import timedelta 

10from typing import Any 

11 

12from expenses.cost_dashboard import ( # pyright: ignore[reportMissingImports] 

13 oil_added, 

14 resolve_period, 

15) 

16from models import Flight, Refuel # pyright: ignore[reportMissingImports] 

17 

18DEFAULT_PERIOD_MONTHS = 12 

19PERIOD_OPTIONS = (3, 6, 12, 24, 0) # months; 0 = all time 

20 

21__all__ = [ 

22 "DEFAULT_PERIOD_MONTHS", 

23 "PERIOD_OPTIONS", 

24 "compute_utilization_report", 

25 "engine_hours_flown", 

26 "flight_hours_flown", 

27 "flights_in_period", 

28 "fuel_added", 

29 "resolve_period", 

30] 

31 

32 

33def flights_in_period( 

34 aircraft_id: int, period_start: _date | None, period_end: _date 

35) -> list[Flight]: 

36 query = Flight.query.filter( 

37 Flight.aircraft_id == aircraft_id, Flight.date <= period_end 

38 ) 

39 if period_start is not None: 

40 query = query.filter(Flight.date >= period_start) 

41 return query.all() # type: ignore[no-any-return] 

42 

43 

44def engine_hours_flown( 

45 aircraft_id: int, period_start: _date | None, period_end: _date 

46) -> float: 

47 """Sum of engine hours for flights within [period_start, period_end]. 

48 

49 Mirrors cost_dashboard.hours_flown()'s "prefer the directly-logged 

50 figure over the counter delta" rule, applied to engine_time instead of 

51 flight_time.""" 

52 flights = flights_in_period(aircraft_id, period_start, period_end) 

53 return sum( 

54 float(f.engine_time) 

55 if f.engine_time is not None 

56 else float(f.engine_time_counter_end) - float(f.engine_time_counter_start) 

57 for f in flights 

58 if f.engine_time is not None 

59 or ( 

60 f.engine_time_counter_end is not None 

61 and f.engine_time_counter_start is not None 

62 ) 

63 ) 

64 

65 

66def flight_hours_flown( 

67 aircraft_id: int, period_start: _date | None, period_end: _date 

68) -> float: 

69 """Sum of flight hours for flights within [period_start, period_end]. 

70 

71 Mirrors engine_hours_flown()'s "prefer the directly-logged figure over 

72 the counter delta" rule, applied to flight_time instead of engine_time. 

73 """ 

74 flights = flights_in_period(aircraft_id, period_start, period_end) 

75 return sum( 

76 float(f.flight_time) 

77 if f.flight_time is not None 

78 else float(f.flight_time_counter_end) - float(f.flight_time_counter_start) 

79 for f in flights 

80 if f.flight_time is not None 

81 or ( 

82 f.flight_time_counter_end is not None 

83 and f.flight_time_counter_start is not None 

84 ) 

85 ) 

86 

87 

88def fuel_added( 

89 aircraft_id: int, period_start: _date | None, period_end: _date 

90) -> dict[str, float]: 

91 """Total fuel added within [period_start, period_end], keyed by unit. 

92 

93 Combines a flight's independent before/after top-ups with standalone 

94 Refuel records (not tied to any flight) — both are equally "fuel 

95 added" for a utilization report. Kept per-unit rather than converted, 

96 since a mix of L and gal entries has no unambiguous single total.""" 

97 totals: dict[str, float] = {} 

98 

99 def _add(qty: Any, unit: str | None) -> None: 

100 if qty is None: 

101 return 

102 u = unit or "L" 

103 totals[u] = totals.get(u, 0.0) + float(qty) 

104 

105 for f in flights_in_period(aircraft_id, period_start, period_end): 

106 _add(f.fuel_added_before_qty, f.fuel_added_before_unit) 

107 _add(f.fuel_added_after_qty, f.fuel_added_after_unit) 

108 

109 refuel_query = Refuel.query.filter( 

110 Refuel.aircraft_id == aircraft_id, Refuel.date <= period_end 

111 ) 

112 if period_start is not None: 

113 refuel_query = refuel_query.filter(Refuel.date >= period_start) 

114 for r in refuel_query.all(): 

115 _add(r.quantity, r.unit) 

116 

117 return {u: round(v, 1) for u, v in totals.items()} 

118 

119 

120def _period_stats( 

121 aircraft_id: int, period_start: _date | None, period_end: _date 

122) -> dict[str, Any]: 

123 flights = flights_in_period(aircraft_id, period_start, period_end) 

124 return { 

125 "period_start": period_start, 

126 "period_end": period_end, 

127 "flight_count": len(flights), 

128 "flight_hours": round( 

129 flight_hours_flown(aircraft_id, period_start, period_end), 1 

130 ), 

131 "engine_hours": round( 

132 engine_hours_flown(aircraft_id, period_start, period_end), 1 

133 ), 

134 "landings": sum(f.landing_count or 0 for f in flights), 

135 "fuel_added": fuel_added(aircraft_id, period_start, period_end), 

136 "oil_added_l": oil_added(aircraft_id, period_start, period_end), 

137 } 

138 

139 

140def compute_utilization_report( 

141 aircraft_id: int, period_start: _date | None, period_end: _date 

142) -> dict[str, Any]: 

143 """Utilization for [period_start, period_end], plus the immediately 

144 preceding period of the same length for comparison — insurers commonly 

145 ask for both hours flown in the past policy year and expected hours 

146 for the next; the prior-period figure is the honest baseline for that 

147 without inventing a forecast.""" 

148 current = _period_stats(aircraft_id, period_start, period_end) 

149 

150 previous = None 

151 if period_start is not None: 

152 length_days = (period_end - period_start).days 

153 prev_end = period_start - timedelta(days=1) 

154 prev_start = prev_end - timedelta(days=length_days) 

155 previous = _period_stats(aircraft_id, prev_start, prev_end) 

156 

157 return { 

158 "period_start": period_start, 

159 "period_end": period_end, 

160 "current": current, 

161 "previous": previous, 

162 }