Coverage for app/expenses/cost_dashboard.py: 100%
71 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-31 10:13 +0000
1"""Phase 36: aircraft operating cost dashboard.
3Pure calculation functions, kept independent of Flask routing so they can be
4unit-tested deterministically (mirrors the style of app/pilots/currency.py).
5"""
7from datetime import date as _date
8from datetime import timedelta
9from typing import Any
11from models import Aircraft, Expense, ExpenseCategory, ExpenseType, Flight, db
13DEFAULT_PERIOD_MONTHS = 12
14PERIOD_OPTIONS = (3, 6, 12, 24, 0) # 0 = all time
17def resolve_period(period_months: int, today: _date) -> tuple[_date | None, _date]:
18 """Return (period_start, period_end) for a given period-in-months selector.
20 period_months <= 0 means "all time" (period_start is None). Otherwise the
21 window is `round(period_months * 365 / 12)` days ending today, so the
22 default 12-month option is exactly a rolling 365-day window.
23 """
24 period_end = today
25 if period_months <= 0:
26 return None, period_end
27 days = round(period_months * 365 / 12)
28 return period_end - timedelta(days=days), period_end
31def hours_flown(
32 aircraft_id: int, period_start: _date | None, period_end: _date
33) -> float:
34 """Sum of flight hours for flights within [period_start, period_end].
36 Prefers each flight's directly-logged flight_time over the counter
37 delta, matching how duration is resolved everywhere else a Flight's
38 hours are shown (a counters-only sum would silently exclude flights
39 logged without a Hobbs reading, e.g. GPS/logbook imports)."""
40 query = Flight.query.filter(
41 Flight.aircraft_id == aircraft_id,
42 Flight.date <= period_end,
43 )
44 if period_start is not None:
45 query = query.filter(Flight.date >= period_start)
46 flights = query.all()
47 return sum(
48 float(f.flight_time)
49 if f.flight_time is not None
50 else float(f.flight_time_counter_end) - float(f.flight_time_counter_start)
51 for f in flights
52 if f.flight_time is not None
53 or (
54 f.flight_time_counter_end is not None
55 and f.flight_time_counter_start is not None
56 )
57 )
60def oil_added(aircraft_id: int, period_start: _date | None, period_end: _date) -> float:
61 """Sum of oil top-ups (L) logged on flights within [period_start, period_end],
62 combining independent before-flight and after-flight top-ups."""
63 query = Flight.query.filter(
64 Flight.aircraft_id == aircraft_id,
65 Flight.date <= period_end,
66 db.or_(
67 Flight.oil_added_before_l.isnot(None), Flight.oil_added_after_l.isnot(None)
68 ),
69 )
70 if period_start is not None:
71 query = query.filter(Flight.date >= period_start)
72 total = sum(
73 (float(f.oil_added_before_l) if f.oil_added_before_l is not None else 0.0)
74 + (float(f.oil_added_after_l) if f.oil_added_after_l is not None else 0.0)
75 for f in query.all()
76 )
77 return round(total, 2)
80def _in_period(expense: Expense, period_start: _date | None, period_end: _date) -> bool:
81 """True if the expense (or its coverage span) overlaps [period_start, period_end]."""
82 if expense.coverage_start and expense.coverage_end:
83 if expense.coverage_start > period_end:
84 return False
85 return not (period_start is not None and expense.coverage_end < period_start)
86 if expense.date > period_end:
87 return False
88 return not (period_start is not None and expense.date < period_start)
91def _prorated_amount(
92 expense: Expense, period_start: _date | None, period_end: _date
93) -> float:
94 """Amount attributable to the report period, pro-rated by coverage-span overlap."""
95 amount = float(expense.amount)
96 coverage_start: _date | None = expense.coverage_start
97 coverage_end: _date | None = expense.coverage_end
98 if coverage_start is None or coverage_end is None:
99 return amount
101 coverage_days = (coverage_end - coverage_start).days + 1
102 if coverage_days <= 0:
103 return amount
105 overlap_start = (
106 coverage_start if period_start is None else max(coverage_start, period_start)
107 )
108 overlap_end = min(coverage_end, period_end)
109 overlap_days = (overlap_end - overlap_start).days + 1
110 if overlap_days <= 0:
111 return 0.0
112 return round(amount * overlap_days / coverage_days, 2)
115def compute_cost_dashboard(
116 aircraft: Aircraft, period_months: int, today: _date | None = None
117) -> dict[str, Any]:
118 """Compute the fixed/operating/reserve cost breakdown for one aircraft."""
119 if today is None:
120 today = _date.today()
121 period_start, period_end = resolve_period(period_months, today)
123 all_expenses = Expense.query.filter_by(aircraft_id=aircraft.id).all()
124 in_period = [e for e in all_expenses if _in_period(e, period_start, period_end)]
126 per_flight = [e for e in in_period if e.flight_entry_id is not None]
127 counted = [e for e in in_period if e.flight_entry_id is None]
129 fixed_total = round(
130 sum(
131 _prorated_amount(e, period_start, period_end)
132 for e in counted
133 if e.expense_category == ExpenseCategory.FIXED
134 ),
135 2,
136 )
137 operating = [e for e in counted if e.expense_category != ExpenseCategory.FIXED]
138 operating_total = round(
139 sum(_prorated_amount(e, period_start, period_end) for e in operating), 2
140 )
141 fuel_total = round(
142 sum(
143 _prorated_amount(e, period_start, period_end)
144 for e in operating
145 if e.expense_type == ExpenseType.FUEL
146 ),
147 2,
148 )
149 maintenance_total = round(
150 sum(
151 _prorated_amount(e, period_start, period_end)
152 for e in operating
153 if e.expense_type == ExpenseType.PARTS
154 ),
155 2,
156 )
157 excluded_per_flight_total = round(sum(float(e.amount) for e in per_flight), 2)
159 hours = hours_flown(aircraft.id, period_start, period_end)
161 oil_total_l = oil_added(aircraft.id, period_start, period_end)
162 oil_per_hour = (
163 round(oil_total_l / hours, 2) if hours > 0 and oil_total_l > 0 else None
164 )
165 oil_warning = (
166 aircraft.oil_warning_lph is not None
167 and oil_per_hour is not None
168 and oil_per_hour > float(aircraft.oil_warning_lph)
169 )
171 reserve_rate = (
172 float(aircraft.reserve_hourly_rate)
173 if aircraft.reserve_hourly_rate is not None
174 else None
175 )
176 reserve_total = round(reserve_rate * hours, 2) if reserve_rate is not None else 0.0
177 wet_total = round(fixed_total + operating_total + reserve_total, 2)
179 def _per_hour(total: float) -> float | None:
180 return round(total / hours, 2) if hours > 0 else None
182 return {
183 "period_start": period_start,
184 "period_end": period_end,
185 "hours_flown": hours,
186 "fixed_total": fixed_total,
187 "fixed_per_hour": _per_hour(fixed_total),
188 "operating_total": operating_total,
189 "operating_per_hour": _per_hour(operating_total),
190 "fuel_total": fuel_total,
191 "fuel_per_hour": _per_hour(fuel_total),
192 "maintenance_total": maintenance_total,
193 "maintenance_per_hour": _per_hour(maintenance_total),
194 "reserve_per_hour": reserve_rate,
195 "reserve_total": reserve_total,
196 "wet_total": wet_total,
197 "wet_per_hour": _per_hour(wet_total),
198 "excluded_per_flight_total": excluded_per_flight_total,
199 "oil_total_l": oil_total_l,
200 "oil_per_hour": oil_per_hour,
201 "oil_warning": oil_warning,
202 }