Coverage for app/services/component_limits.py: 100%

1"""Engine / propeller TBO and life-limited component tracking. 

2 

3A component may carry an hours limit (tbo_hours — time between overhauls or 

4an hours life limit) and/or a calendar limit (life_limit_date, e.g. 12-year 

5rubber hoses). This module computes, per component: 

6 

7 - total component hours (time_at_install + the aircraft's flight-counter 

8 deltas inside the installation window) 

9 - hours since the last recorded overhaul (overhauled_at_hours resets the 

10 reference point) 

11 - a status: 'overdue' | 'due_soon' | 'ok' 

12 

13Hours limits warn inside the last 10 % of the interval; calendar limits warn 

1490 days ahead (mirroring the airworthiness document window). 

15""" 

16 

17from datetime import date as _date 

18from datetime import timedelta 

19from typing import Any 

20 

21CALENDAR_WARN_DAYS = 90 

22HOURS_WARN_FRACTION = 0.1 

23 

24 

25def component_hours(comp: Any) -> float: 

26 """Total hours on the component: time at install + engine-hours deltas 

27 since. 

28 

29 Engine/propeller TBO and life limits are an engine-hours metric, not 

30 flight (airborne) time — sums each flight's directly-logged engine_time 

31 (set by the aircraft-log form/offline sync) over the engine counter 

32 delta, matching how duration is resolved everywhere else an engine-hours 

33 total is displayed. A flight with no engine counters logged would 

34 otherwise silently not count towards the component's hours at all. 

35 """ 

36 from models import Flight, db # pyright: ignore[reportMissingImports] 

37 from sqlalchemy import case # pyright: ignore[reportMissingImports] 

38 

39 hours = case( 

40 (Flight.engine_time.isnot(None), Flight.engine_time), 

41 ( 

42 db.and_( 

43 Flight.engine_time_counter_end.isnot(None), 

44 Flight.engine_time_counter_start.isnot(None), 

45 ), 

46 Flight.engine_time_counter_end - Flight.engine_time_counter_start, 

47 ), 

48 else_=0, 

49 ) 

50 query = db.session.query(db.func.sum(hours)).filter( 

51 Flight.aircraft_id == comp.aircraft_id, 

52 ) 

53 if comp.installed_at: 

54 query = query.filter(Flight.date >= comp.installed_at) 

55 if comp.removed_at: 

56 query = query.filter(Flight.date <= comp.removed_at) 

57 flown = float(query.scalar() or 0) 

58 return round(float(comp.time_at_install or 0) + flown, 1) 

59 

60 

61def component_limit_info( 

62 comp: Any, today: "_date | None" = None 

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

64 """Limit status for one component, or None when it has no limits set. 

65 

66 Returns a dict with: component, total_hours, since_overhaul, tbo_hours, 

67 tbo_remaining, life_limit_date, status. 

68 """ 

69 tbo = float(comp.tbo_hours) if comp.tbo_hours is not None else None 

70 limit_date = comp.life_limit_date 

71 if tbo is None and limit_date is None: 

72 return None 

73 if today is None: 

74 today = _date.today() 

75 

76 statuses = [] 

77 total_hours = component_hours(comp) 

78 since_overhaul = round(total_hours - float(comp.overhauled_at_hours or 0), 1) 

79 tbo_remaining = None 

80 if tbo is not None: 

81 tbo_remaining = round(tbo - since_overhaul, 1) 

82 if tbo_remaining <= 0: 

83 statuses.append("overdue") 

84 elif tbo_remaining <= tbo * HOURS_WARN_FRACTION: 

85 statuses.append("due_soon") 

86 if limit_date is not None: 

87 if limit_date < today: 

88 statuses.append("overdue") 

89 elif limit_date <= today + timedelta(days=CALENDAR_WARN_DAYS): 

90 statuses.append("due_soon") 

91 

92 if "overdue" in statuses: 

93 status = "overdue" 

94 elif "due_soon" in statuses: 

95 status = "due_soon" 

96 else: 

97 status = "ok" 

98 return { 

99 "component": comp, 

100 "total_hours": total_hours, 

101 "since_overhaul": since_overhaul, 

102 "tbo_hours": tbo, 

103 "tbo_remaining": tbo_remaining, 

104 "life_limit_date": limit_date, 

105 "status": status, 

106 } 

107 

108 

109def aircraft_limit_infos( 

110 ac: Any, today: "_date | None" = None 

111) -> "list[dict[str, Any]]": 

112 """Limit info for every currently installed, limited component of ac.""" 

113 infos = [] 

114 for comp in ac.components: 

115 if comp.removed_at is not None: 

116 continue 

117 info = component_limit_info(comp, today) 

118 if info is not None: 

119 infos.append(info) 

120 return infos 

121 

122 

123def fleet_limit_statuses( 

124 aircraft_list: Any, today: "_date | None" = None 

125) -> "dict[int, str]": 

126 """Worst component-limit status per aircraft ('overdue'|'due_soon'|'ok').""" 

127 result: dict[int, str] = {} 

128 for ac in aircraft_list: 

129 statuses = [info["status"] for info in aircraft_limit_infos(ac, today)] 

130 if "overdue" in statuses: 

131 result[ac.id] = "overdue" 

132 elif "due_soon" in statuses: 

133 result[ac.id] = "due_soon" 

134 else: 

135 result[ac.id] = "ok" 

136 return result