Coverage for app/maintenance/amp_import.py: 100%
140 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-31 10:13 +0000
1"""AMP task-list spreadsheet import (Phase 40).
3Parses the OpenHangar "AMP task list" import template — documented in
4docs/maintenance_import.md — into structured rows ready for review. Column
5order and leading unrelated rows/sheets don't matter: the header row is
6located by matching known column names, not a fixed position.
7"""
9from __future__ import annotations
11import re
12from dataclasses import dataclass
13from datetime import date as _date
14from datetime import timedelta
15from typing import TYPE_CHECKING, Any
17from models import ( # pyright: ignore[reportMissingImports]
18 AmpCategory,
19 ComponentType,
20 HoursBasis,
21)
23if TYPE_CHECKING:
24 import openpyxl
26_MAX_HEADER_SCAN_ROWS = 30
28# Canonical column key -> normalised header text it matches.
29_HEADER_ALIASES = {
30 "category": "category",
31 "task description": "task_description",
32 "reference": "reference",
33 "action": "action",
34 "interval": "interval",
35 "part number": "part_number",
36 "serial number": "serial_number",
37 "notes": "notes",
38}
39_REQUIRED_HEADER_KEYS = {"task_description", "interval"}
42def _norm(raw: Any) -> str:
43 return " ".join(str(raw or "").strip().lower().split())
46@dataclass
47class HeaderLocation:
48 sheet_name: str
49 header_row_index: int # 0-based, within the sheet
50 col_index: dict[str, int] # canonical key -> column index
53def find_header(wb: openpyxl.Workbook) -> HeaderLocation:
54 """Scan every sheet's first _MAX_HEADER_SCAN_ROWS rows for one whose
55 cells include at least 'Task description' and 'Interval' (matched
56 case/whitespace-insensitively); returns the first match found, sheets
57 in workbook order."""
58 for ws in wb.worksheets:
59 for row_idx, row in enumerate(
60 ws.iter_rows(min_row=1, max_row=_MAX_HEADER_SCAN_ROWS, values_only=True)
61 ):
62 col_index: dict[str, int] = {}
63 for col_idx, cell in enumerate(row):
64 key = _HEADER_ALIASES.get(_norm(cell))
65 if key and key not in col_index:
66 col_index[key] = col_idx
67 if _REQUIRED_HEADER_KEYS <= col_index.keys():
68 return HeaderLocation(ws.title, row_idx, col_index)
69 raise ValueError(
70 "No matching header row found. Expected a sheet with at least "
71 "'Task description' and 'Interval' columns."
72 )
75_INTERVAL_FH_RE = re.compile(r"^(\d+(?:\.\d+)?)\s*FH$", re.IGNORECASE)
76_INTERVAL_CAL_RE = re.compile(r"^(\d+)\s*(DY|MO|YR)$", re.IGNORECASE)
77_CAL_UNIT_DAYS = {"DY": 1, "MO": 30, "YR": 365}
80@dataclass
81class IntervalParseResult:
82 interval_hours: float | None = None
83 interval_days: int | None = None
84 needs_review: bool = False
87def parse_interval(raw: Any) -> IntervalParseResult:
88 """Parse the Interval mini-syntax: one or two '/'-separated parts, each
89 either '<n>FH' or '<n>(DY|MO|YR)'. Both parts populate together for a
90 combined "whichever comes first" interval. Empty, 'PENDING', or any
91 unparseable part makes the whole interval needs_review — never a
92 partial/silent result."""
93 text = str(raw or "").strip()
94 if not text or text.upper() == "PENDING":
95 return IntervalParseResult(needs_review=True)
97 interval_hours: float | None = None
98 interval_days: int | None = None
99 for part in text.split("/"):
100 part = part.strip()
101 m = _INTERVAL_FH_RE.match(part)
102 if m:
103 interval_hours = float(m.group(1))
104 continue
105 m = _INTERVAL_CAL_RE.match(part)
106 if m:
107 n = int(m.group(1))
108 interval_days = n * _CAL_UNIT_DAYS[m.group(2).upper()]
109 continue
110 return IntervalParseResult(needs_review=True)
112 # Every part matched (or we'd have returned above), so at least one of
113 # interval_hours/interval_days is set here — text.split("/") can never
114 # yield zero parts.
115 return IntervalParseResult(
116 interval_hours=interval_hours, interval_days=interval_days
117 )
120def match_category(raw: Any) -> str | None:
121 """Case/whitespace-tolerant exact match against the 9 canonical
122 AmpCategory values. No match (blank, admin/routine text, typo) -> None,
123 the row's category is simply left unset — not rejected."""
124 norm = _norm(raw)
125 if not norm:
126 return None
127 for canonical in AmpCategory.ALL:
128 if _norm(canonical) == norm:
129 return canonical
130 return None
133def suggest_component_id(
134 category_raw: Any, task_description_raw: Any, components: list[Any]
135) -> int | None:
136 """Heuristic: a mention of 'propeller' or 'engine' in the category or
137 task description text suggests the aircraft's matching installed
138 component. No mention, or no installed component of that type, leaves
139 the row unscoped — corrected per-row on the review screen."""
140 text = f"{category_raw or ''} {task_description_raw or ''}".lower()
141 if "propeller" in text:
142 comp_type = ComponentType.PROPELLER
143 elif "engine" in text:
144 comp_type = ComponentType.ENGINE
145 else:
146 return None
147 for c in components:
148 if c.removed_at is None and c.type == comp_type:
149 return c.id # type: ignore[no-any-return]
150 return None
153def hours_basis_for_component(component: Any) -> str:
154 """Engine/propeller components track TBO-style intervals in engine
155 hours; everything else (including unscoped) is flight-hours based —
156 mirrors the same default used by the trigger form's component picker."""
157 if component is not None and component.type in (
158 ComponentType.ENGINE,
159 ComponentType.PROPELLER,
160 ):
161 return HoursBasis.ENGINE
162 return HoursBasis.FLIGHT
165@dataclass
166class ParsedAmpRow:
167 row_number: int # 1-based spreadsheet row, for display only
168 name: str
169 category_raw: str | None
170 category: str | None
171 reference: str | None
172 action: str | None
173 part_number: str | None
174 serial_number: str | None
175 notes: str | None
176 interval_raw: str | None
177 interval_hours: float | None
178 interval_days: int | None
179 needs_review: bool
180 suggested_component_id: int | None
183def _cell(row: tuple[Any, ...], col_index: dict[str, int], key: str) -> Any:
184 idx = col_index.get(key)
185 if idx is None or idx >= len(row):
186 return None
187 return row[idx]
190def _text_or_none(v: Any) -> str | None:
191 if v is None:
192 return None
193 s = str(v).strip()
194 return s or None
197def parse_amp_rows(wb: openpyxl.Workbook, components: list[Any]) -> list[ParsedAmpRow]:
198 """Parse every data row of the located header's sheet. A row with no
199 task description is a blank/trailing row and is silently skipped (not
200 counted, not an error) — everything else always produces a row, even
201 when its interval is unparseable (see parse_interval)."""
202 header = find_header(wb)
203 ws = wb[header.sheet_name]
204 rows: list[ParsedAmpRow] = []
205 for row_idx, row in enumerate(
206 ws.iter_rows(min_row=header.header_row_index + 2, values_only=True),
207 start=header.header_row_index + 2,
208 ):
209 name = _text_or_none(_cell(row, header.col_index, "task_description"))
210 if not name:
211 continue
212 category_raw = _text_or_none(_cell(row, header.col_index, "category"))
213 interval_raw = _text_or_none(_cell(row, header.col_index, "interval"))
214 interval = parse_interval(interval_raw)
215 rows.append(
216 ParsedAmpRow(
217 row_number=row_idx,
218 name=name,
219 category_raw=category_raw,
220 category=match_category(category_raw),
221 reference=_text_or_none(_cell(row, header.col_index, "reference")),
222 action=_text_or_none(_cell(row, header.col_index, "action")),
223 part_number=_text_or_none(_cell(row, header.col_index, "part_number")),
224 serial_number=_text_or_none(
225 _cell(row, header.col_index, "serial_number")
226 ),
227 notes=_text_or_none(_cell(row, header.col_index, "notes")),
228 interval_raw=interval_raw,
229 interval_hours=interval.interval_hours,
230 interval_days=interval.interval_days,
231 needs_review=interval.needs_review,
232 suggested_component_id=suggest_component_id(
233 category_raw, name, components
234 ),
235 )
236 )
237 return rows
240def format_interval(interval_hours: float | None, interval_days: int | None) -> str:
241 """Inverse of parse_interval: reconstruct the "<n>FH / <n>MO"-style
242 display text from stored interval_hours/interval_days, for Appendix B
243 export. Days are expressed in whichever of YR/MO/DY the value divides
244 evenly by (365/30/1 days respectively — the same units parse_interval
245 converts from), preferring the largest, so a value imported as "12MO"
246 round-trips to "12MO" rather than some equivalent-but-different text."""
247 parts = []
248 if interval_hours is not None:
249 h = float(interval_hours)
250 parts.append(f"{h:.0f}FH" if h == int(h) else f"{h}FH")
251 if interval_days is not None:
252 d = int(interval_days)
253 if d % 365 == 0:
254 parts.append(f"{d // 365}YR")
255 elif d % 30 == 0:
256 parts.append(f"{d // 30}MO")
257 else:
258 parts.append(f"{d}DY")
259 return " / ".join(parts)
262def compute_due_fields(
263 interval_hours: float | None,
264 interval_days: int | None,
265 hours_basis: str,
266 current_engine_hours: float | None,
267 current_flight_hours: float | None,
268 today: _date | None = None,
269) -> tuple[float | None, _date | None]:
270 """Initial due_engine_hours/due_date for a freshly-imported trigger:
271 the current reading plus the interval (hours side needs a current
272 reading to anchor to; with none available yet, that side is simply
273 left unset — the same as a manually-entered trigger with no due value
274 yet, correctable once flight history exists)."""
275 due_engine_hours = None
276 if interval_hours is not None:
277 current = (
278 current_flight_hours
279 if hours_basis == HoursBasis.FLIGHT
280 else current_engine_hours
281 )
282 if current is not None:
283 due_engine_hours = current + interval_hours
284 due_date = None
285 if interval_days is not None:
286 due_date = (today or _date.today()) + timedelta(days=interval_days)
287 return due_engine_hours, due_date