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

27 "fuel_added", 

28 "resolve_period", 

29] 

30 

31 

32def flights_in_period( 

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

34) -> list[Flight]: 

35 query = Flight.query.filter( 

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

37 ) 

38 if period_start is not None: 

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

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

41 

42 

43def engine_hours_flown( 

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

45) -> float: 

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

47 

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

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

50 flight_time.""" 

51 flights = flights_in_period(aircraft_id, period_start, period_end) 

52 return sum( 

53 float(f.engine_time) 

54 if f.engine_time is not None 

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

56 for f in flights 

57 if f.engine_time is not None 

58 or ( 

59 f.engine_time_counter_end is not None 

60 and f.engine_time_counter_start is not None 

61 ) 

62 ) 

63 

64 

65def fuel_added( 

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

67) -> dict[str, float]: 

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

69 

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

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

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

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

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

75 

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

77 if qty is None: 

78 return 

79 u = unit or "L" 

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

81 

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

83 _add(f.fuel_added_before_qty, f.fuel_added_before_unit) 

84 _add(f.fuel_added_after_qty, f.fuel_added_after_unit) 

85 

86 refuel_query = Refuel.query.filter( 

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

88 ) 

89 if period_start is not None: 

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

91 for r in refuel_query.all(): 

92 _add(r.quantity, r.unit) 

93 

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

95 

96 

97def _period_stats( 

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

99) -> dict[str, Any]: 

100 flights = flights_in_period(aircraft_id, period_start, period_end) 

101 flight_hours = sum( 

102 float(f.flight_time) 

103 if f.flight_time is not None 

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

105 for f in flights 

106 if f.flight_time is not None 

107 or ( 

108 f.flight_time_counter_end is not None 

109 and f.flight_time_counter_start is not None 

110 ) 

111 ) 

112 return { 

113 "period_start": period_start, 

114 "period_end": period_end, 

115 "flight_count": len(flights), 

116 "flight_hours": round(flight_hours, 1), 

117 "engine_hours": round( 

118 engine_hours_flown(aircraft_id, period_start, period_end), 1 

119 ), 

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

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

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

123 } 

124 

125 

126def compute_utilization_report( 

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

128) -> dict[str, Any]: 

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

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

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

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

133 without inventing a forecast.""" 

134 current = _period_stats(aircraft_id, period_start, period_end) 

135 

136 previous = None 

137 if period_start is not None: 

138 length_days = (period_end - period_start).days 

139 prev_end = period_start - timedelta(days=1) 

140 prev_start = prev_end - timedelta(days=length_days) 

141 previous = _period_stats(aircraft_id, prev_start, prev_end) 

142 

143 return { 

144 "period_start": period_start, 

145 "period_end": period_end, 

146 "current": current, 

147 "previous": previous, 

148 }