STAR
Cal STAR — firmware, DAQ server, and engine design
Loading...
Searching...
No Matches
limits.py
Go to the documentation of this file.
1"""Every limit a Layer X burn is graded against, in one place (DATA-CONTRACT section 1).
2
3Before this module the thresholds lived three times: the UI's ``VERDICT`` and ``verdictItems``
4(``frontend/src/components/layerx/format.ts``, ``LayerXResult.tsx``), the optimiser's ``grade``
5(``engine/layerx/optimize.py``) and the uncertainty sweep's crossings
6(``engine/layerx/uncertainty.py``). They agreed by copy. :data:`THRESHOLDS` holds each number once,
7with the reasoning that the UI kept in its hint, and :func:`grade` turns a result into the graded
8list the GUI draws as margin bars.
9
10What each entry carries (one dict per limit):
11
12* ``key``, ``label``, ``group`` (stability | injector | tanks | pressurant | propellant | flight |
13 hardware | model);
14* ``value`` and ``unit``; ``limit`` (the red line) and ``warn`` (the amber edge), either of which
15 may be ``None`` when the limit has only one; ``direction`` ``"min"`` (stay above) or ``"max"``
16 (stay below);
17* ``grade``: ``ok`` | ``warn`` | ``bad`` | ``info``. ``info`` is reported and never counts;
18* ``t_worst`` and ``index_worst``: the time [s from Fire] and the ``series`` index of the worst
19 point, ``None`` when the quantity is not time-resolved;
20* ``series_ref``: the dotted path of its history in the result, for the margin bar to jump to;
21* ``basis`` (what the value is computed from) and ``hint`` (one sentence: why this threshold).
22
23A limit whose data is absent from the result is skipped, never guessed. An old saved run therefore
24grades exactly as the UI graded it, and a run carrying the new ``diagnostics`` blocks grades them too.
25
26Nothing here burns, solves or reads a file: it reads the result. It never raises on a malformed
27block; the entry is skipped and the reason goes in a ``model`` health entry instead.
28"""
29
30from __future__ import annotations
31
32import math
33from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple
34
35PSI = 6894.757293168361
36"""One psi in pascals."""
37
38#: The threshold for each limit, with what it is. None is a requirement from the config unless it
39#: says so; they are judgement, stated so a reader can disagree. Ported from the UI's ``VERDICT``
40#: (``format.ts``), ``optimize.grade`` and ``uncertainty.SPARE_PSI``, which agreed.
41THRESHOLDS: Dict[str, Dict[str, Any]] = {
42 "copv_headroom_psi": {
43 "value": 100.0, "unit": "psi",
44 "why": "Bottle above tank lockup at burnout: the dome regulator needs supply over its outlet to keep "
45 "regulating. Assumed (UI VERDICT.copvHeadroomPsi, optimize dropout_margin_psi, uncertainty "
46 "SPARE_PSI); replace with the regulator's measured dropout. Amber at twice it.",
47 },
48 "droop_psi": {
49 "value": (30.0, 60.0), "unit": "psi",
50 "why": "Tank droop from lockup while firing: amber, red. Assumed (UI VERDICT.droopPsi).",
51 },
52 "residual_kg": {
53 "value": 0.2, "unit": "kg",
54 "why": "Propellant stranded in the other tank above which to look. The team's figure (2026-10-03: "
55 "\"if it's within 0.2 kg it's fine\"); about 2 % of the LE4 load.",
56 },
57 "stiffness_floor": {
58 "value": 0.15, "unit": "",
59 "why": "Injector dP/Pc floor when the config states no band: a common chug rule of thumb (UI "
60 "VERDICT.stiffnessFloor; uncertainty's floor fallback).",
61 },
62 "chug_margin_red": {
63 "value": 1.0, "unit": "",
64 "why": "Gain margin 1 is neutral: below it the feed-coupled chug loop is predicted unstable.",
65 },
66 "chug_margin_amber": {
67 "value": 1.2, "unit": "",
68 "why": "Judgement (UI VERDICT.chugMarginWarn): the margin rests on an unmeasured mixing-lag band "
69 "(AUDIT D7: nominal 2.18 against the gating 1.33) and moves ~0.1 with the feed basis. The "
70 "amber edge is max(this, design_requirements.min_stability_margin).",
71 },
72 "rating_use": {
73 "value": 0.8, "unit": "",
74 "why": "Peak pressure across a vessel wall, as a fraction of its rating, above which to look. "
75 "Judgement (UI VERDICT.ratingUse).",
76 },
77 "depletion_tie": {
78 "value": 0.03, "unit": "",
79 "why": "Left-over propellant, as a fraction of the load, under which which tank runs dry first is "
80 "inside the orifice Cd's +-3 % scatter (UI VERDICT.depletionTie).",
81 },
82 "replay_agreement": {
83 "value": (0.005, 0.02), "unit": "",
84 "why": "Twin against EngineDesign's replay: fine below the first, worth a look below the second, "
85 "red above. The card's own fit is ~0.02 % (UI VERDICT.replayAgreement).",
86 },
87 "summerfield_pe_pa": {
88 "value": 0.4, "unit": "",
89 "why": "Summerfield criterion: an overexpanded nozzle separates when p_exit/p_ambient falls to ~0.4 "
90 "(Sutton & Biblarz, Rocket Propulsion Elements, sec. 3.3; engine/pipeline/handcheck.py).",
91 },
92 "saturation_margin_psi": {
93 "value": (0.0, 25.0), "unit": "psi",
94 "why": "Local static pressure over the liquid's vapour pressure: red at 0 (the liquid boils or flashes); "
95 "amber under 25 psi, assumed: about one LOX-line dynamic head at full flow (AUDIT 9.6 B4), room for "
96 "the transients the twin does not resolve (valve opening, line acoustics). The saturation block's "
97 "own flag_margin_psi input, when it states one, sets the amber edge instead.",
98 },
99 "cavitation_margin": {
100 "value": (1.0, 1.2), "unit": "",
101 "why": "Injector cavitation number over Nurick's critical value (K/K_crit; Nurick, J. Fluids Eng. 98(4), "
102 "1976): red under 1, where the orifice cavitates and its flow is Cc sqrt(K), not the single-phase "
103 "Cd the engine card uses, so the burn's flows are not valid there; amber under 1.2, assumed (K_crit "
104 "is the model's onset and incipient cavitation appears above it). The cavitation block's own "
105 "near_margin input, when it states one, sets the amber edge instead.",
106 },
107 "conservation_mass_pct": {
108 "value": (0.1, 1.0), "unit": "%",
109 "why": "Propellant the burn cannot account for (diag/vv.py), amber, red. The recorded flows are end-of-step "
110 "samples of a burn the twin integrates on finer sub-steps, so a sound burn sits at that quadrature "
111 "error, first order in dt: 0.002-0.004 % on the flown GN2 runs of 2026-10-02, 0.035 % (2 g) on the "
112 "LE4 he_pad baseline at dt 0.05 s. Amber at 0.1 % (~11 g on LE4), red at 1 %, the audit's "
113 "threshold for a result change that matters (AUDIT 8).",
114 },
115 "conservation_pressurant_pct": {
116 "value": (0.5, 2.0), "unit": "%",
117 "why": "Pressurant the burn cannot account for (diag/vv.py), amber, red. The check re-prices each ullage "
118 "with CoolProp from the recorded pressure, temperature and fill; it closes to ~1e-12 % on the flown "
119 "GN2 runs of 2026-10-02 and 1.6e-7 % on the LE4 he_pad baseline, so its floor is property agreement, "
120 "not bookkeeping. Amber at 0.5 % and red at 2 % are assumed (judgement): over three orders of "
121 "magnitude above that floor, and below the ~2 % helium compressibility (Z ~ 1.02 at 578 psia, "
122 "293 K, CoolProp) that an ideal-gas slip in the bookkeeping would show.",
123 },
124 "regulator_use": {
125 "value": (0.8, 1.0), "unit": "",
126 "why": "Regulator flow over its capacity at the seat: amber above 0.8 (judgement: droop grows "
127 "steeply near full lift), red at 1 (wide open, the tanks follow the bottle).",
128 },
129 # The competition the vehicle flies (FAR-OUT Rules and Requirements 2024-25, rev 2024-10-02). Graded
130 # only where the design states no requirement of its own: a design's own number always wins. Before
131 # these, a design that left them blank showed a 22 m/s rail exit under "Within limits".
132 "farout_rail_exit": {
133 "value": 85.0 * 0.3048, "unit": "m/s",
134 "why": "FAR-OUT 2024-25 rule 5.1.4: the vehicle leaves the rail above 85 ft/s (25.9 m/s); the rules "
135 "recommend designing to 100 ft/s. Applied because the design states no "
136 "design_requirements.min_rail_exit_velocity_m_s.",
137 },
138 "farout_static_margin": {
139 "value": 1.5, "unit": "cal",
140 "why": "FAR-OUT 2024-25 rule 5.1.5: a static margin of at least 1.5 calibres throughout the ascent. "
141 "Applied because the design states no design_requirements.min_static_margin_cal.",
142 },
143}
144
145
146# ---------------------------------------------------------------------- small helpers
147
148
149def _num(v: Any) -> Optional[float]:
150 """A finite float, or None."""
151 if v is None or isinstance(v, bool):
152 return None
153 try:
154 f = float(v)
155 except (TypeError, ValueError):
156 return None
157 return f if math.isfinite(f) else None
158
159
160def _dig(d: Any, path: str) -> Any:
161 for part in path.split("."):
162 if not isinstance(d, Mapping):
163 return None
164 d = d.get(part)
165 return d
166
167
168def _grade(value: Optional[float], limit: Optional[float], warn: Optional[float], direction: str) -> str:
169 """ok | warn | bad, strict comparisons (the UI's): equal to the limit is not over it."""
170 if value is None:
171 return "warn"
172 if direction == "min":
173 if limit is not None and value < limit:
174 return "bad"
175 if warn is not None and value < warn:
176 return "warn"
177 return "ok"
178 if limit is not None and value > limit:
179 return "bad"
180 if warn is not None and value > warn:
181 return "warn"
182 return "ok"
183
184
185def _entry(key: str, label: str, group: str, value: Optional[float], unit: str, limit: Optional[float],
186 warn: Optional[float], direction: str, grade: str, t_worst: Optional[float],
187 index_worst: Optional[int], series_ref: Optional[str], basis: str, hint: str,
188 **extra: Any) -> Dict[str, Any]:
189 out = {"key": key, "label": label, "group": group, "value": value, "unit": unit, "limit": limit,
190 "warn": warn, "direction": direction, "grade": grade, "t_worst": t_worst,
191 "index_worst": index_worst, "series_ref": series_ref, "basis": basis, "hint": hint}
192 out.update(extra)
193 return out
194
195
196class _Run:
197 """Read-only views of a result that every grader needs: the series' clock and its firing steps."""
198
199 def __init__(self, result: Mapping[str, Any]) -> None:
200 self.result = result
201 self.series = result.get("series") or {}
202 self.summary = result.get("summary") or {}
203 self.t: List[float] = [float(v) for v in (self.series.get("t") or [])]
204 firing = self.series.get("firing") or []
205 self.fire = [i for i, f in enumerate(firing) if f]
206 # Summary statistics leave out the first 0.2 s on the burn's clock (analysis.reduce_trace).
207 self.settled = [i for i in self.fire if self.t[i] >= 0.2] or list(self.fire)
208
209 def at(self, i: Optional[int]) -> Tuple[Optional[float], Optional[int]]:
210 if i is None or not 0 <= i < len(self.t):
211 return None, None
212 return self.t[i], i
213
214 def index_of(self, t: Optional[float]) -> Optional[int]:
215 """The series step nearest ``t`` (the step whose end is ``t`` when it is on the grid)."""
216 if t is None or not self.t:
217 return None
218 return min(range(len(self.t)), key=lambda i: abs(self.t[i] - t))
219
220 def column(self, path: str) -> List[Optional[float]]:
221 col = _dig(self.series, path)
222 return [_num(v) for v in col] if isinstance(col, list) else []
223
224 def argext(self, values: Sequence[Optional[float]], idx: Iterable[int], fn: Any) -> Optional[int]:
225 pairs = [(values[i], i) for i in idx if i < len(values) and values[i] is not None]
226 if not pairs:
227 return None
228 return fn(pairs, key=lambda p: p[0])[1]
229
230 def firing_to_series(self, k: Optional[int]) -> Optional[int]:
231 """A ``delivered`` index (firing steps only) as a series index."""
232 if k is None or not 0 <= k < len(self.fire):
233 return None
234 return self.fire[k]
235
236
237def _derived(result: Mapping[str, Any], prep: Any) -> Dict[str, Any]:
238 d = _dig(result, "provenance.derived")
239 if isinstance(d, Mapping) and d:
240 return dict(d)
241 return dict(getattr(prep, "derived", None) or {})
242
243
244def requirements(result: Mapping[str, Any], config: Any = None) -> Dict[str, Any]:
245 """The design's ``design_requirements`` as a plain dict: from ``config`` when given, else from
246 the YAML the run recorded (``provenance.reproduce.design_yaml``), else empty."""
247 req = getattr(config, "design_requirements", None) if config is not None else None
248 if req is not None:
249 if hasattr(req, "model_dump"):
250 return dict(req.model_dump())
251 if isinstance(req, Mapping):
252 return dict(req)
253 return {k: getattr(req, k) for k in dir(req) if not k.startswith("_")}
254 text = _dig(result, "provenance.reproduce.design_yaml")
255 if isinstance(text, str) and text.strip():
256 try:
257 import yaml
258
259 doc = yaml.safe_load(text)
260 dr = doc.get("design_requirements") if isinstance(doc, dict) else None
261 return dict(dr) if isinstance(dr, dict) else {}
262 except Exception: # noqa: BLE001 - an unreadable record grades without the design's caps
263 return {}
264 return {}
265
266
267def _defaulted(config: Any) -> frozenset:
268 """The ``design_requirements`` keys that hold the schema's default because the design never
269 stated them (pydantic's ``model_fields_set``). Such a number has no source, so a limit built on
270 it is reported, not graded. Empty when it cannot be told (a plain mapping, or the recorded YAML,
271 which holds only what the design stated)."""
272 req = getattr(config, "design_requirements", None) if config is not None else None
273 fields = getattr(type(req), "model_fields", None) if req is not None else None
274 stated = getattr(req, "model_fields_set", None) if req is not None else None
275 if not isinstance(fields, Mapping) or stated is None:
276 return frozenset()
277 return frozenset(k for k in fields if k not in stated)
278
279
280# ---------------------------------------------------------------------- the graders
281
282
283def _chug(run: _Run, req: Mapping[str, Any], defaulted: frozenset = frozenset(),
284 derived: Optional[Mapping[str, Any]] = None) -> List[Dict[str, Any]]:
285 msm = _num(req.get("min_stability_margin"))
286 amber = THRESHOLDS["chug_margin_amber"]["value"]
287 warn = max(amber, msm) if msm is not None else amber
288 red = THRESHOLDS["chug_margin_red"]["value"]
289 said = ("the config schema's default for design_requirements.min_stability_margin, which the design does "
290 "not state" if "min_stability_margin" in defaulted else "design_requirements.min_stability_margin")
291 why = (f"Below {red:g} the feed-coupled loop is predicted unstable (red). Amber under {warn:g}: "
292 + (f"the design requires {msm:g} ({said}), but the amber edge "
293 f"stays at {amber:g} or above because the margin rests on an unmeasured mixing-lag band and "
294 "moves ~0.1 with the feed basis (AUDIT D7)." if msm is not None else
295 "judgement, the margin rests on an unmeasured mixing-lag band (AUDIT D7)."))
296 out: List[Dict[str, Any]] = []
297 st = _dig(run.result, "diagnostics.stability")
298 if isinstance(st, Mapping) and st.get("available", True) is not False and (
299 _num(_dig(st, "settled_min.margin")) is not None or _num(_dig(st, "worst.margin")) is not None):
300 basis_name = st.get("basis") or "config"
301 settled = _num(_dig(st, "settled_min.margin"))
302 worst = _num(_dig(st, "worst.margin"))
303 if settled is not None:
304 value, t, idx = settled, _num(_dig(st, "settled_min.t")), _dig(st, "settled_min.index")
305 # The frequency of the point that is graded, not the start window's (AUDIT 5.1: the
306 # reported 34.6 Hz was not the gating ~22 Hz).
307 freq = _num(_dig(st, "settled_min.frequency_hz"))
308 window = _num(st.get("start_window_s"))
309 scope = (f"from the first full-flow step (after the first {window:.3f} s)" if window is not None
310 else "from the first full-flow step")
311 else:
312 value, t, scope = worst, _num(_dig(st, "worst.t")), "over the whole burn, start included"
313 idx = _dig(st, "worst.index")
314 freq = _num(_dig(st, "worst.frequency_hz"))
315 i = idx if isinstance(idx, int) and 0 <= idx < len(run.t) else run.index_of(t)
316 # The gaps in the drawing this margin stands on, beside it rather than buried in the block's
317 # flags (W14: LE4 (3)'s LOX path is undrawn and its fuel tank line has a 0 mm bore).
318 gaps = [str(f) for f in (st.get("flags") or [])
319 if isinstance(f, str) and ("no usable length/bore" in f or "declares no line" in f)]
320 out.append(_entry(
321 "chug_margin", "Chug margin", "stability", value, "", red, warn, "min", _grade(value, red, warn, "min"),
322 t, i, "diagnostics.stability.margin",
323 f"EngineDesign double-time-lag chug model, {basis_name} feed basis, {scope}",
324 f"Feed-coupled gain margin, worst over the mixing-lag band. {why}"
325 + (f" It rests on gaps in the drawing: {'; '.join(gaps)}." if gaps else ""),
326 frequency_hz=freq, **({"drawing_gaps": gaps} if gaps else {})))
327 if settled is not None and worst is not None and worst < settled:
328 tw = _num(_dig(st, "worst.t"))
329 iw = _dig(st, "worst.index")
330 out.append(_entry(
331 "chug_margin_start", "Chug margin, start window", "stability", worst, "", red, warn, "min", "info",
332 tw, iw if isinstance(iw, int) and 0 <= iw < len(run.t) else run.index_of(tw),
333 "diagnostics.stability.margin",
334 f"EngineDesign double-time-lag chug model, {basis_name} feed basis, inside the start window",
335 "Reported, not graded: the start transient (valve ramp, priming, ignition delay) is not modelled "
336 "here, and the minimum inside it moves with the time step (AUDIT 5.1, #8).",
337 frequency_hz=_num(_dig(st, "worst.frequency_hz"))))
338 other = st.get("other_basis")
339 if isinstance(other, Mapping) and _num(other.get("margin_min")) is not None:
340 to = _num(other.get("t"))
341 v = _num(other.get("margin_min"))
342 out.append(_entry(
343 "chug_margin_other_basis", f"Chug margin, {other.get('basis') or 'other'} basis", "stability", v, "",
344 red, warn, "min", "info", to, run.index_of(to),
345 "diagnostics.stability.margin_other" if isinstance(st.get("margin_other"), list) else None,
346 f"EngineDesign double-time-lag chug model, {other.get('basis') or 'other'} feed basis",
347 "The same margin on the other feed basis, for comparison; the graded one is above.",
348 frequency_hz=_num(other.get("frequency_hz"))))
349 return out
350 dv = run.result.get("delivered") or {}
351 s = dv.get("summary") or {}
352 value = _num(s.get("chug_margin_min"))
353 if value is None:
354 return out
355 t = _num(s.get("chug_margin_min_t"))
356 col = [_num(v) for v in (dv.get("chug_margin") or [])]
357 k = run.argext(col, range(len(col)), min) if col else None
358 i = run.firing_to_series(k) if k is not None else run.index_of(t)
359 basis = ("EngineDesign double-time-lag chug model at the erosion replay's points (config feed basis, A_t and L* "
360 "at the design point), interpolated onto the twin's firing steps; lowest over every firing step, the "
361 "first included, which lies inside the valve ramp, so the minimum moves with the time step "
362 "(AUDIT 5.1, #12)")
363 if _dig(derived or {}, "options.chug_basis") == "drawing":
364 # The run is graded on the drawing's feed (decided 2026-10-03: "physics, not options"), and its
365 # stability block is absent (a set-point, Hardware, reconcile or sweep burn runs no diagnostics).
366 # The replay's figure is EngineDesign's own feed_system: it read 1.21 'ok' on LE4 (3) where the
367 # drawing basis grades 1.004 'warn'. Shown for what it is, never graded.
368 out.append(_entry(
369 "chug_margin", "Chug margin, config basis (not graded)", "stability", value, "", red, warn, "min", "info",
370 t, i, "delivered.chug_margin", basis,
371 "EngineDesign's own feed_system, not the drawing's lines the run is graded on: that margin is the "
372 "stability diagnostic's, which this burn did not run (a Burn-tab run does) or could not compute, so "
373 f"its chug is not graded. Feed-coupled gain margin, worst over the mixing-lag band. {why}",
374 feed_basis="config"))
375 return out
376 out.append(_entry(
377 "chug_margin", "Chug margin", "stability", value, "", red, warn, "min", _grade(value, red, warn, "min"),
378 t, i, "delivered.chug_margin", basis,
379 f"Feed-coupled gain margin, worst over the mixing-lag band. {why}"))
380 return out
381
382
383def _stiffness(run: _Run, band: Mapping[str, Any]) -> List[Dict[str, Any]]:
384 floor = THRESHOLDS["stiffness_floor"]["value"]
385 out = []
386 for side, short, name in (("oxidiser", "ox", "LOX"), ("fuel", "fuel", "Fuel")):
387 s = run.summary.get(short) or {}
388 if not s and not run.series.get(short):
389 continue
390 value = _num(s.get("stiffness_min"))
391 b = band.get(side) if isinstance(band, Mapping) else None
392 b = [float(b[0]), float(b[1])] if isinstance(b, (list, tuple)) and len(b) == 2 and None not in b else None
393 if value is None:
394 grade = "warn"
395 elif b:
396 grade = "bad" if value < b[0] else "warn" if value > b[1] else "ok"
397 else:
398 grade = "bad" if value < floor else "ok"
399 col = run.column(f"{short}.stiffness")
400 i = run.argext(col, run.settled, min)
401 t, i = run.at(i)
402 band_text = (f"the design's band is {b[0] * 100:g}-{b[1] * 100:g} %: red below it (chug), amber above it "
403 "(tank pressure spent on the injector)" if b else
404 f"the design sets no band, so it is graded against {floor * 100:g} % "
405 "(a common chug rule of thumb)")
406 out.append(_entry(
407 f"stiffness_{short}", f"{name} injector ΔP/Pc", "injector", value, "",
408 b[0] if b else floor, None, "min", grade, t, i, f"series.{short}.stiffness",
409 "Twin: manifold (line exit less the Borda dump) minus Pc, over Pc; lowest over firing steps from "
410 "t = 0.2 s (the start transient is not modelled)",
411 f"Lowest injector pressure drop over chamber pressure through the burn; {band_text}.",
412 band=b))
413 ign = _num(s.get("stiffness_min_ignition"))
414 if ign is not None:
415 early = [k for k in run.fire if k not in set(run.settled)]
416 j = run.argext(col, early, min)
417 tj, j = run.at(j)
418 out.append(_entry(
419 f"stiffness_{short}_ignition", f"{name} injector ΔP/Pc, first 0.2 s", "injector", ign, "",
420 b[0] if b else floor, None, "min", "info", tj, j, f"series.{short}.stiffness",
421 "Twin, firing steps before t = 0.2 s",
422 "Reported, not graded: the start transient (valve ramp, priming, ignition) is not modelled, "
423 "so the figure moves with the time step (AUDIT 5.1, #13).", band=b))
424 return out
425
426
427def _tanks(run: _Run, derived: Mapping[str, Any], req: Mapping[str, Any],
428 meop_psi: Optional[Mapping[str, float]], defaulted: frozenset = frozenset()) -> List[Dict[str, Any]]:
429 out: List[Dict[str, Any]] = []
430 roles = derived.get("roles") or {}
431 mawps = derived.get("tank_mawp_psi") or {}
432 ratings = derived.get("vessel_ratings") or {}
433 ratings = ratings if isinstance(ratings, Mapping) else {}
434 ambient_pa = _num(derived.get("ambient_pa"))
435 ambient_psia = ambient_pa / PSI if ambient_pa is not None else 14.6959
436 gauge_pa = _num(derived.get("gauge_zero_pa"))
437 gauge_zero_psia = gauge_pa / PSI if gauge_pa is not None else 101325.0 / PSI
438 meops = dict(derived.get("tank_meop_psi") or {})
439 use = THRESHOLDS["rating_use"]["value"]
440 for side, short, name, cap_key in (("oxidiser", "ox", "LOX", "max_lox_tank_pressure_psi"),
441 ("fuel", "fuel", "Fuel", "max_fuel_tank_pressure_psi")):
442 s = run.summary.get(short) or {}
443 peak = _num(s.get("peak_psia"))
444 if peak is None:
445 continue
446 col = run.column(f"{short}.tank_psia")
447 i = run.argext(col, range(len(col)), max)
448 t, i = run.at(i)
449 across = peak - ambient_psia
450 tank_id = roles.get(side)
451 trip = _num(mawps.get(tank_id)) if tank_id else None
452 if trip is not None and trip > gauge_zero_psia:
453 # ``tank_mawp_psi`` is the absolute pressure the twin trips the stand at (prepare.vessel_trip;
454 # MAWP.si on runs before burst ratings were read, absolute too). Graded on the gauge the
455 # drawing states its rating on, so the red line is exactly the trip and the percentage is
456 # of the drawn psig. It used to subtract the site's atmosphere from the peak and hold that
457 # against the absolute figure: one atmosphere and more lenient, the % reading low.
458 rating = ratings.get(tank_id) if isinstance(ratings.get(tank_id), Mapping) else {}
459 words = rating.get("words") or f"{trip:.1f} psia (the drawing's MAWP)"
460 vessel = rating.get("label") or tank_id
461 peak_g, rating_g = peak - gauge_zero_psia, trip - gauge_zero_psia
462 out.append(_entry(
463 # "psi": a pressure across the wall is a difference (ADR 0007: psia to people, a difference
464 # plain psi); the page draws "psi" as a gap and has no "psig" kind (contract.limitKindOf).
465 f"tank_mawp_{short}", f"{name} tank peak vs rating", "tanks", peak_g, "psi", rating_g, use * rating_g,
466 "max",
467 # Information for now (the team, 2026-10-03: "don't worry about tank limits for now"); a
468 # vessel over its rating still trips the stand (vessel_trip, the twin's own physics).
469 "info", t, i, f"series.{short}.tank_psia",
470 f"Ullage pressure, highest over the recorded lead-in and the burn, less the gauge zero "
471 f"({gauge_zero_psia:.2f} psia) the drawing's rating is stated on, against the {words} {vessel} trips "
472 "the stand at",
473 f"Highest pressure across the wall: {peak_g / rating_g * 100:.0f} % of the rating the twin trips at "
474 f"(amber above {use * 100:.0f} %, a judgement). The recorded lead-in is the last 0.5 s before Fire, "
475 "not the pad hold.", rating_source=str(rating.get("basis") or "drawing")))
476 meop = None
477 for src in (meop_psi or {}, meops):
478 meop = _num(src.get(side)) if src.get(side) is not None else _num(src.get(tank_id)) if tank_id else None
479 if meop is not None:
480 break
481 if meop is not None:
482 out.append(_entry(
483 f"tank_meop_{short}", f"{name} tank peak vs MEOP", "tanks", across, "psi", meop, None, "max",
484 "info", t, i, f"series.{short}.tank_psia",
485 f"Ullage pressure less the site's atmosphere ({ambient_psia:.2f} psia), against the stated maximum "
486 "expected operating pressure",
487 "MEOP is the most the vessel is meant to see in operation: above it the burn has left the envelope "
488 "the vessel was qualified for.", rating_source="restated MEOP"))
489 cap = _num(req.get(cap_key))
490 if cap is not None:
491 schema_default = cap_key in defaulted
492 out.append(_entry(
493 f"tank_cap_{short}", f"{name} tank peak vs design cap", "tanks", peak, "psia", cap, cap, "max",
494 "info", t, i, f"series.{short}.tank_psia",
495 f"Ullage pressure, highest over the recorded lead-in and the burn, against "
496 f"design_requirements.{cap_key}, read as absolute (as EngineDesign's tank pressures and the "
497 "optimiser's lockup bound read it)"
498 + ("; the design does not state it, so this is the config schema's default (no source)"
499 if schema_default else ""),
500 ("Reported, not graded: the cap is the schema's default, not a number the design states. "
501 if schema_default else "")
502 + "Graded amber, not red, until it is decided whether the design's tank cap applies at T-0 or at "
503 "the peak (AUDIT D11): the regulator's supply-pressure effect lifts the tanks ~40 psi during a "
504 "burn.",
505 cap_source="schema default" if schema_default else "design_requirements"))
506 return out
507
508
509def _bottle(run: _Run) -> List[Dict[str, Any]]:
510 s = run.summary
511 if "copv_end_psia" not in s:
512 return []
513 ox, fu = s.get("ox") or {}, s.get("fuel") or {}
514 lockups = [v for v in (_num(ox.get("t0_psia")), _num(fu.get("t0_psia"))) if v is not None]
515 end = _num(s.get("copv_end_psia"))
516 value = end - max(lockups) if end is not None and lockups else None
517 h = THRESHOLDS["copv_headroom_psi"]["value"]
518 t, i = run.at(len(run.t) - 1 if run.t else None)
519 # The bottle is read off its gauge everywhere else on the page (psig, at the run's gauge zero):
520 # quote it the same way here, or one bottle shows as two numbers.
521 derived = (run.result.get("provenance") or {}).get("derived") or {}
522 zero_pa = _num(derived.get("gauge_zero_pa"))
523 zero_psia = zero_pa / PSI if zero_pa is not None else 14.6959
524 return [_entry(
525 "bottle_margin", "Bottle over lockup at burnout", "pressurant", value, "psi", h, 2.0 * h, "min",
526 _grade(value, h, 2.0 * h, "min"), t, i, "series.copv_psia",
527 "Bottle vessel pressure at the last step less the higher tank's pressure at T-0 (the lockup)",
528 "How far the bottle is above the tanks' lockup when the burn ends"
529 + (f" (the bottle reads {end - zero_psia:,.0f} psig)" if end is not None else "")
530 + f". Under {h:g} psi the regulator stops holding tank pressure (an assumed dropout); amber under "
531 f"{2 * h:g}.")]
532
533
534def _sag(run: _Run) -> List[Dict[str, Any]]:
535 best: Optional[Tuple[float, str]] = None
536 for short in ("ox", "fuel"):
537 s = run.summary.get(short) or {}
538 t0, lo = _num(s.get("t0_psia")), _num(s.get("min_psia"))
539 if t0 is None:
540 continue
541 d = t0 - (lo if lo is not None else t0)
542 if best is None or d > best[0]:
543 best = (d, short)
544 if best is None:
545 return []
546 droop, short = best
547 warn, red = THRESHOLDS["droop_psi"]["value"]
548 i = run.argext(run.column(f"{short}.tank_psia"), run.fire, min)
549 t, i = run.at(i)
550 return [_entry(
551 "tank_sag", "Tank pressure sag", "tanks", droop, "psi", red, warn, "max", _grade(droop, red, warn, "max"),
552 t, i, f"series.{short}.tank_psia",
553 f"T-0 tank pressure less the lowest while firing, the deeper of the two tanks ({'LOX' if short == 'ox' else 'fuel'})",
554 f"Deepest dip below the set tank pressure while firing (amber over {warn:g}, red over {red:g} psi; "
555 "assumed).")]
556
557
558def _propellant(run: _Run) -> List[Dict[str, Any]]:
559 s = run.summary
560 side = s.get("depleted_side")
561 ox, fu = s.get("ox") or {}, s.get("fuel") or {}
562 out: List[Dict[str, Any]] = []
563 if side not in ("oxidiser", "fuel"):
564 if run.fire and s:
565 t, i = run.at(run.fire[-1])
566 why = ("The burn stopped at a vessel trip with propellant in both tanks (vessel_trip): its totals end "
567 "at the trip and are not the load's." if isinstance(run.result.get("tripped"), Mapping) else
568 "The burn reached its horizon with propellant in both tanks: a regulator that dropped out or a "
569 "horizon too short. Its totals are not the load's.")
570 out.append(_entry(
571 "depletion", "A tank runs dry", "model", None, "", None, None, "min", "warn", t, i, None,
572 "The burn's end condition", why))
573 return out
574 residual = _num((fu if side == "oxidiser" else ox).get("residual_kg"))
575 loads = [_num(ox.get("loaded_kg")), _num(fu.get("loaded_kg"))]
576 first = "LOX" if side == "oxidiser" else "Fuel"
577 t_dry = _num(s.get("burn_time_s"))
578 # The step the burn ended on (the burn cuts its last step to land on depletion, so its time is
579 # the depletion time); without firing steps, the summary's.
580 t, i = run.at(run.fire[-1]) if run.fire else (t_dry, None)
581 if residual is None:
582 return out
583 if None not in loads:
584 frac = THRESHOLDS["depletion_tie"]["value"]
585 tie = frac * (loads[0] + loads[1])
586 other = "fuel" if side == "oxidiser" else "LOX"
587 # Information, never a warning: one tank always empties first, and which one inside the Cd
588 # scatter is not something a person acts on (the team, 2026-10-03). What is worth a look is the
589 # propellant left over, graded below.
590 out.append(_entry(
591 "depletion_tie", "Runs dry first", "propellant", residual, "kg", None, None, "min",
592 "info", t, i, f"series.{'fuel' if side == 'oxidiser' else 'ox'}.liquid_kg",
593 f"Liquid left in the {other} tank when the {first} tank ran dry",
594 f"Within {frac * 100:.0f} % of the load ({tie:.2f} kg) the orifice Cd's +-3 % scatter decides which tank "
595 "empties first: on the stand it may be the other one.",
596 detail=f"{first} runs dry first, {residual:.3f} kg of {other} left", side=side))
597 r = THRESHOLDS["residual_kg"]["value"]
598 out.append(_entry(
599 "residual", "Propellant left over", "propellant", residual, "kg", None, r, "max",
600 "warn" if residual > r else "ok", t, i, None,
601 "Liquid left in the tank that did not run dry, at burnout",
602 f"Propellant carried to burnout and never burned: dead mass in flight. Amber above {r:g} kg (the team's "
603 "figure)."))
604 return out
605
606
607def _saturation(run: _Run) -> List[Dict[str, Any]]:
608 sat = _dig(run.result, "diagnostics.saturation")
609 if not isinstance(sat, Mapping) or sat.get("available") is False:
610 return []
611 red, warn = THRESHOLDS["saturation_margin_psi"]["value"]
612 stated = _num(sat.get("flag_margin_psi"))
613 warn = stated if stated is not None else warn
614 out = []
615 for n, node in enumerate(sat.get("nodes") or []):
616 if not isinstance(node, Mapping):
617 continue
618 # The static margin (total less the dynamic head) is what boils a moving liquid; the total
619 # margin when the block could not price the head.
620 static = _num(node.get("min_static_psi")) is not None
621 key, col_key, t_key = (("min_static_psi", "margin_static_psi", "t_min_static") if static
622 else ("min_psi", "margin_psi", "t_min"))
623 margins = [_num(v) for v in (node.get(col_key) or [])] if isinstance(node.get(col_key), list) else []
624 value = _num(node.get(key))
625 k = run.argext(margins, run.fire or range(len(margins)), min) if margins else None
626 if value is None and k is not None:
627 value = margins[k]
628 if value is None:
629 continue
630 t = _num(node.get(t_key))
631 i = k if k is not None and len(margins) == len(run.t) else run.index_of(t)
632 if t is None and i is not None:
633 t = run.t[i]
634 nid = str(node.get("id") or n)
635 out.append(_entry(
636 f"saturation_{nid}", f"Saturation margin, {node.get('label') or nid}", "propellant", value, "psi",
637 red, warn, "min", _grade(value, red, warn, "min"), t, i, f"diagnostics.saturation.nodes.{n}.{col_key}",
638 ("Local static pressure (node total less the line's dynamic head)" if static else "Local node pressure")
639 + " less the liquid's vapour pressure at its temperature, lowest over the firing steps "
640 "(diagnostics.saturation)",
641 THRESHOLDS["saturation_margin_psi"]["why"], side=node.get("side")))
642 return out
643
644
645def _cavitation(run: _Run) -> List[Dict[str, Any]]:
646 cav = _dig(run.result, "diagnostics.cavitation")
647 if not isinstance(cav, Mapping) or cav.get("available") is False:
648 return []
649 red, warn = THRESHOLDS["cavitation_margin"]["value"]
650 near = _num(_dig(cav, "model.inputs.near_margin.value"))
651 warn = red + near if near is not None else warn
652 out = []
653 for short, name in (("ox", "LOX"), ("fuel", "Fuel")):
654 c = cav.get(short)
655 if not isinstance(c, Mapping):
656 continue
657 ratio = [_num(v) for v in (c.get("margin") or [])] if isinstance(c.get("margin"), list) else []
658 value = _num(c.get("min_margin"))
659 k = run.argext(ratio, range(len(ratio)), min) if ratio else None
660 if value is None and k is not None:
661 value = ratio[k]
662 if value is None:
663 # A block with only K: graded against its incipient K.
664 K, inc = _num(c.get("min_K")), _num(c.get("K_incipient"))
665 value = K / inc if K is not None and inc else None
666 if value is None:
667 continue
668 t = _num(c.get("t_min"))
669 i = k if k is not None and len(ratio) == len(run.t) else run.index_of(t)
670 if t is None and i is not None:
671 t = run.t[i]
672 flip = bool(c.get("flip_risk"))
673 grade = "bad" if flip else _grade(value, red, warn, "min")
674 out.append(_entry(
675 f"cavitation_{short}", f"{name} injector cavitation", "injector", value, "", red, warn, "min",
676 grade, t, i, f"diagnostics.cavitation.{short}.margin",
677 "Cavitation number K = (p_manifold - p_vapour)/(p_manifold - Pc) over Nurick's critical K, lowest over "
678 "the firing steps (diagnostics.cavitation)",
679 THRESHOLDS["cavitation_margin"]["why"] + (" Hydraulic-flip risk flagged: the jet may detach from the "
680 "orifice wall." if flip else ""),
681 flip_risk=flip, K=_num(c.get("min_K")), K_incipient=_num(c.get("K_incipient"))))
682 return out
683
684
685def _water_hammer(run: _Run) -> List[Dict[str, Any]]:
686 """The surge when the mains open at Fire, as information. The mains open for Fire and stay open
687 (the team, 2026-10-03): the burn ends on depletion, never on a closing valve, so the closing
688 surge the diagnostic also computes is not a case the stand sees and is not listed."""
689 wh = _dig(run.result, "diagnostics.water_hammer")
690 if not isinstance(wh, list):
691 return []
692 out = []
693 for n, w in enumerate(wh):
694 if not isinstance(w, Mapping):
695 continue
696 opening = w.get("opening") if isinstance(w.get("opening"), Mapping) else {}
697 peak, rating = _num(opening.get("peak_psia")), _num(w.get("rating_psia"))
698 if peak is None:
699 continue
700 line = str(w.get("line") or n)
701 out.append(_entry(
702 f"water_hammer_{line}", f"Opening surge, {line}", "hardware", peak, "psia", rating, None, "max",
703 "info", None, None, None,
704 "Peak line pressure when the main valve opens at Fire (diagnostics.water_hammer.opening): the "
705 "start model's arrival flow meeting the orifices, the gas cushion ignored (an upper bound)",
706 "Information: the mains open once and stay open, so there is no closing surge to grade.",
707 side=w.get("side")))
708 return out
709
710
711def _separation(run: _Run, derived: Mapping[str, Any]) -> List[Dict[str, Any]]:
712 crit = THRESHOLDS["summerfield_pe_pa"]["value"]
713 sep = _dig(run.result, "diagnostics.hardware.separation")
714 hw_t = _dig(run.result, "diagnostics.hardware.t")
715 ambient_pa = _num(derived.get("ambient_pa"))
716 ratio: List[Optional[float]] = []
717 times: List[float] = []
718 schmucker = False
719 if isinstance(sep, Mapping) and isinstance(sep.get("ratio"), list):
720 ratio = [_num(v) for v in sep["ratio"]]
721 times = [float(v) for v in hw_t] if isinstance(hw_t, list) and len(hw_t) == len(ratio) else []
722 schmucker = any(bool(v) for v in (sep.get("schmucker") or []))
723 basis = ("Exit pressure over the ambient the nozzle exhausts into (diagnostics.hardware.separation, at the "
724 "erosion replay's points)")
725 ref = "diagnostics.hardware.separation.ratio"
726 else:
727 dv = run.result.get("delivered") or {}
728 pe = [_num(v) for v in (dv.get("p_exit_psia") or [])]
729 amb = [_num(v) for v in (dv.get("ambient_psia") or [])]
730 if not pe:
731 return []
732 if amb and len(amb) == len(pe):
733 ratio = [p / a if p is not None and a else None for p, a in zip(pe, amb)]
734 elif ambient_pa:
735 ratio = [p * PSI / ambient_pa if p is not None else None for p in pe]
736 times = [float(v) for v in (dv.get("t") or [])]
737 basis = ("EngineDesign's exit pressure from the erosion replay (delivered.p_exit_psia, interpolated onto the "
738 "firing steps) over the ambient the nozzle exhausts into")
739 ref = "delivered.p_exit_psia"
740 live = [j for j in range(len(ratio)) if ratio[j] is not None]
741 if not live:
742 return []
743 k = min(live, key=lambda j: ratio[j]) # type: ignore[arg-type,return-value]
744 value = ratio[k]
745 t = times[k] if k < len(times) else None
746 grade = _grade(value, crit, None, "min")
747 if grade == "ok" and schmucker:
748 grade = "warn"
749 return [_entry(
750 "separation", "Nozzle exit over ambient", "hardware", value, "", crit, None, "min", grade, t,
751 run.index_of(t), ref, basis + "; lowest over the burn",
752 THRESHOLDS["summerfield_pe_pa"]["why"] + (" Amber: Schmucker's criterion already predicts separation."
753 if schmucker else ""))]
754
755
756def _flight(run: _Run, req: Mapping[str, Any]) -> List[Dict[str, Any]]:
757 f = run.result.get("flight")
758 if not isinstance(f, Mapping) or not f.get("ok"):
759 return []
760 st = f.get("stability") if isinstance(f.get("stability"), Mapping) else {}
761 out: List[Dict[str, Any]] = []
762 need = _num(req.get("min_rail_exit_velocity_m_s"))
763 if need is None:
764 need = _num(st.get("rail_exit_required_m_s"))
765 rail_why = ("design_requirements.min_rail_exit_velocity_m_s: below it the fins have too little airspeed to "
766 "stabilise the vehicle off the rail.")
767 if need is None:
768 need, rail_why = THRESHOLDS["farout_rail_exit"]["value"], THRESHOLDS["farout_rail_exit"]["why"]
769 v = _num(st.get("rail_exit_m_s"))
770 if v is None:
771 v = _num(f.get("rail_exit_velocity_m_s"))
772 if v is not None:
773 t = _num(st.get("rail_exit_t"))
774 t = t if t is not None else _num(f.get("rail_exit_time_s"))
775 out.append(_entry(
776 "rail_exit", "Rail exit speed", "flight", v, "m/s", need, None, "min", _grade(v, need, None, "min"),
777 t, run.index_of(t), None, "RocketPy Flight.out_of_rail_velocity", rail_why))
778 lo, hi = _num(req.get("min_static_margin_cal")), _num(req.get("max_static_margin_cal"))
779 margin_why = (f"design_requirements.min_static_margin_cal ({lo:g} cal)" if lo is not None
780 else THRESHOLDS["farout_static_margin"]["why"])
781 if lo is None:
782 lo = THRESHOLDS["farout_static_margin"]["value"]
783 liftoff = _num(st.get("liftoff_static_margin_cal"))
784 liftoff = liftoff if liftoff is not None else _num(st.get("static_margin_liftoff_cal"))
785 smin = _num(st.get("min_static_margin_cal"))
786 smin_t = _num(st.get("min_static_margin_t"))
787 if smin is None:
788 smin, smin_t = _num(st.get("min_stability_margin_cal")), _num(st.get("min_stability_margin_time_s"))
789 for key, label, value, t in (("static_margin_liftoff", "Static margin at liftoff", liftoff, 0.0),
790 ("static_margin_min", "Static margin, lowest", smin, smin_t)):
791 if value is None:
792 continue
793 grade = _grade(value, lo, None, "min")
794 if hi is not None and value > hi:
795 grade = "warn"
796 out.append(_entry(
797 key, label, "flight", value, "cal", lo, None, "min", grade, t, run.index_of(t),
798 "flight.stability.static_margin_cal" if isinstance(st.get("static_margin_cal"), list) else None,
799 "RocketPy Barrowman static margin (CG less CP over the body diameter)",
800 margin_why + (f"; amber above max_static_margin_cal ({hi:g} cal), where the vehicle weathercocks"
801 if hi is not None else "")))
802 q = _num(st.get("max_q_pa"))
803 if q is not None:
804 tq = _num(st.get("max_q_t"))
805 out.append(_entry(
806 "max_q", "Max dynamic pressure", "flight", q, "Pa", None, None, "max", "info", tq, run.index_of(tq),
807 None, "RocketPy Flight.max_dynamic_pressure", "Reported for the airframe's loads; not a Layer X limit."))
808 ceiling = f.get("ceiling") if isinstance(f.get("ceiling"), Mapping) else None
809 apogee = _num(f.get("apogee_agl_m"))
810 if ceiling and apogee is not None and _num(ceiling.get("limit_agl_m")) is not None:
811 lim = _num(ceiling.get("limit_agl_m"))
812 out.append(_entry(
813 "apogee_ceiling", "Apogee under the ceiling", "flight", apogee, "m", lim, None, "max",
814 _grade(apogee, lim, None, "max"), None, None, None, "RocketPy apogee above the pad",
815 "design_requirements.max_apogee_m: the waiver ceiling (optimize.grade's 'ceiling')."))
816 return out
817
818
819def _regulator(run: _Run) -> List[Dict[str, Any]]:
820 reg = _dig(run.result, "diagnostics.regulator")
821 if not isinstance(reg, Mapping) or reg.get("available") is False:
822 return []
823 use = [_num(v) for v in (reg.get("use_frac") or [])]
824 wide = [bool(v) for v in (reg.get("wide_open") or [])]
825 times = [float(v) for v in (reg.get("t") or run.t)]
826 if not use and not wide:
827 return []
828 warn, red = THRESHOLDS["regulator_use"]["value"]
829 k = run.argext(use, range(len(use)), max) if use else None
830 value = use[k] if k is not None else None
831 if any(wide):
832 k = wide.index(True)
833 t = times[k] if k is not None and k < len(times) else None
834 grade = "bad" if any(wide) else _grade(value, red, warn, "max") if value is not None else "info"
835 return [_entry(
836 "regulator_wide_open", "Regulator use of capacity", "pressurant", value, "", red, warn, "max", grade, t,
837 run.index_of(t), "diagnostics.regulator.use_frac",
838 "Dome regulator flow over its capacity at the seat (diagnostics.regulator)",
839 THRESHOLDS["regulator_use"]["why"])]
840
841
842def _health(run: _Run) -> List[Dict[str, Any]]:
843 r, s = run.result, run.summary
844 out: List[Dict[str, Any]] = []
845 last_t, last_i = run.at(len(run.t) - 1 if run.t else None)
846 trip = r.get("tripped")
847 if isinstance(trip, Mapping):
848 p, m = _num(trip.get("p_psia")), _num(trip.get("mawp_psia"))
849 tt = _num(trip.get("t"))
850 out.append(_entry(
851 "vessel_trip", f"Vessel trip ({trip.get('vessel') or 'vessel'})", "model", p, "psia", m, None, "max",
852 "bad", tt, run.index_of(tt), None,
853 "The twin's MAWP trip: the stand shut down at this instant and the burn stopped there",
854 "A vessel passed the MAWP its drawing declares; everything after this instant was not burned."))
855 if r.get("converged") is False:
856 out.append(_entry(
857 "unsettled", "Model settled", "model", None, "", None, None, "min", "warn", last_t, last_i, None,
858 "The erosion replay and flight coupling loop (result.passes)",
859 "The erosion replay or the flight coupling did not settle, or the replay failed: thrust and Pc may be "
860 "the as-built throat's. See the events."))
861 outside = s.get("card_outside_steps")
862 if isinstance(outside, (int, float)) and not isinstance(outside, bool):
863 out.append(_entry(
864 "card_outside", "Engine table", "model", float(outside), "steps", None, 0.0, "max",
865 "warn" if outside > 0 else "ok", None, None, None,
866 "Firing steps whose injector or chamber point lay outside the engine card's sample hull (a count; the "
867 "steps are not recorded)",
868 "Steps outside what the card was fitted to are extrapolated."))
869 ec = r.get("engine_check")
870 if isinstance(ec, Mapping) and ec.get("available") and isinstance(ec.get("worst"), Mapping):
871 w = ec["worst"]
872 keys = ["pc", "mdot_O", "mdot_F"] + ([] if ec.get("against") == "replay" else ["thrust"])
873 gaps = [(abs(_num(w.get(k)) or 0.0), k) for k in keys]
874 gap, which = max(gaps)
875 a, b = THRESHOLDS["replay_agreement"]["value"]
876 grade = "ok" if gap < a else "warn" if gap < b else "bad"
877 t_w = None
878 for row in ec.get("rows") or []:
879 rel = (row or {}).get("rel") or {}
880 v = _num(rel.get(which))
881 if v is not None and abs(v) == gap:
882 t_w = _num(row.get("t"))
883 break
884 out.append(_entry(
885 "engine_fit", "Engine fit", "model", gap, "", b, a, "max", grade, t_w, run.index_of(t_w),
886 None, f"Twin against EngineDesign at the same line-exit pressures (engine_check, worst of {', '.join(keys)})",
887 THRESHOLDS["replay_agreement"]["why"], worst_of=which))
888 failed = s.get("failed_steps")
889 settled_t0 = s.get("t0_settled")
890 if failed is not None or settled_t0 is not None:
891 conv = r.get("series", {}).get("converged") or []
892 first_bad = next((j for j, c in enumerate(conv) if c is False), None)
893 tb, ib = run.at(first_bad)
894 bad = bool(failed) or settled_t0 is False
895 out.append(_entry(
896 "solver", "Solver", "model", float(failed or 0), "steps", None, 0.0, "max", "warn" if bad else "ok", tb, ib,
897 "series.converged", "Steps the network solve could not close, and whether T-0 settled",
898 "A step the solver could not close holds its last good flows." + (
899 " T-0 did not settle: the trace opens off its datum." if settled_t0 is False else "")))
900 vv = _dig(r, "diagnostics.vv")
901 if isinstance(vv, Mapping):
902 out.extend(_conservation(vv, last_t, last_i))
903 return out
904
905
906def _conservation(vv: Mapping[str, Any], last_t: Optional[float], last_i: Optional[int]) -> List[Dict[str, Any]]:
907 out = []
908 rows: List[Tuple[str, str, Any, str]] = []
909 for side, name in (("ox", "LOX"), ("fuel", "Fuel")):
910 rows.append((f"conservation_mass_{side}", f"{name} mass balance", _dig(vv, f"mass.{side}.error_pct"), "mass"))
911 rows.append(("conservation_pressurant", "Pressurant mass balance", _dig(vv, "pressurant.error_pct"), "pressurant"))
912 for key, label, err, kind in rows:
913 e = _num(err)
914 if e is None:
915 continue
916 warn, red = THRESHOLDS[f"conservation_{kind}_pct"]["value"]
917 out.append(_entry(
918 key, label, "model", abs(e), "%", red, warn, "max", _grade(abs(e), red, warn, "max"), last_t, last_i,
919 f"diagnostics.vv.{'mass.' + key.rsplit('_', 1)[-1] if kind == 'mass' else 'pressurant'}",
920 "Conservation check (engine/layerx/diag/vv.py): relative error of the balance over the burn",
921 THRESHOLDS[f"conservation_{kind}_pct"]["why"]))
922 e = _num(_dig(vv, "energy.error_pct"))
923 if e is not None:
924 out.append(_entry(
925 "conservation_energy", "Gas energy balance", "model", e, "%", None, None, "max", "info", last_t, last_i,
926 "diagnostics.vv.energy", str(_dig(vv, "energy.basis") or "gas-side first law"),
927 "Reported, not graded: the residual includes the heat the tank walls, the liquid surface and the lines "
928 "exchange with the gas, which the series does not record (see the basis)."))
929 return out
930
931
932# ---------------------------------------------------------------------- the entry point
933
934
935def grade(result: Mapping[str, Any], prep: Any = None, config: Any = None, *,
936 meop_psi: Optional[Mapping[str, float]] = None) -> List[Dict[str, Any]]:
937 """Every limit the burn in ``result`` is graded against, most consequential group first.
938
939 ``prep`` (a :class:`engine.layerx.prepare.Prepared`) is read only when the result carries no
940 ``provenance.derived``; ``config`` supplies ``design_requirements`` (otherwise read from the YAML
941 the run recorded). ``meop_psi`` is a maximum expected operating pressure across the wall per
942 side (``{"oxidiser": psi, "fuel": psi}``), as the set-point request states it; a drawing's or a
943 restatement's ``derived.tank_meop_psi`` (per tank id or side) is read too.
944 """
945 run = _Run(result or {})
946 if not run.t:
947 return []
948 derived = _derived(result, prep)
949 req = requirements(result, config)
950 defaulted = _defaulted(config)
951 band = derived.get("stiffness_band") or {}
952 sections = (
953 lambda: _chug(run, req, defaulted, derived),
954 lambda: _stiffness(run, band),
955 lambda: _tanks(run, derived, req, meop_psi, defaulted),
956 lambda: _bottle(run),
957 lambda: _regulator(run),
958 lambda: _sag(run),
959 lambda: _propellant(run),
960 lambda: _saturation(run),
961 lambda: _cavitation(run),
962 lambda: _water_hammer(run),
963 lambda: _separation(run, derived),
964 lambda: _flight(run, req),
965 lambda: _health(run),
966 )
967 out: List[Dict[str, Any]] = []
968 for build in sections:
969 try:
970 out.extend(build())
971 except Exception as exc: # noqa: BLE001 - one malformed block never costs the others their grades
972 out.append(_entry(
973 "limits_error", "Limits", "model", None, "", None, None, "max", "info", None, None, None,
974 f"{type(exc).__name__}: {exc}", "A block of the result could not be graded; the others were."))
975 return out
976
977
978def overall(entries: Sequence[Mapping[str, Any]]) -> str:
979 """bad if any limit is broken, warn if any is worth a look, else ok. ``info`` never counts."""
980 grades = {e.get("grade") for e in entries}
981 return "bad" if "bad" in grades else "warn" if "warn" in grades else "ok"
982
983
984def breaks(entries: Sequence[Mapping[str, Any]]) -> List[str]:
985 """The broken limits as one line each, for a sweep's crossing list."""
986 out = []
987 for e in entries:
988 if e.get("grade") != "bad":
989 continue
990 v = e.get("value")
991 lim = e.get("limit")
992 unit = e.get("unit") or ""
993 rel = "<" if e.get("direction") == "min" else ">"
994 out.append(f"{e.get('label')} {v:.4g}{(' ' + unit) if unit else ''} {rel} {lim:.4g}"
995 if isinstance(v, (int, float)) and isinstance(lim, (int, float)) else str(e.get("label")))
996 return out
#define min(a, b)
Definition Arduino.h:42
#define max(a, b)
Definition Arduino.h:43
T & at(const key_type &key)
Definition json.hpp:19110