STAR
Cal STAR — firmware, DAQ server, and engine design
Loading...
Searching...
No Matches
hardware.py
Go to the documentation of this file.
1"""The chamber and nozzle hardware along a Layer X burn (DATA-CONTRACT ``diagnostics.hardware``).
2
3Everything here is read off EngineDesign's erosion replay (:mod:`engine.layerx.replay`): the coupled
4time-varying solver's wall stations, the geometry it rebuilt from them at each replay point, and its
5nozzle state. Nothing is solved again except two table look-ups: the ideal vacuum thrust coefficient
6from the CEA cache for the Isp breakdown, and the as-built contour from the thermal geometry module.
7
8The block follows the replay's points, not the twin's steps (``t`` and ``index``, the series index
9of each point): the replay solves the engine at those points only, and everything between them is an
10interpolation (audit 5.3, "never imply 50 ms resolution").
11
12What it reports, per replay point:
13
14* the throat: diameter, area ratio to as built, ``eps = A_e / A_t(t)`` (A_e is fixed: LE4's nozzle
15 does not ablate), ``L* = V(t) / A_t(t)`` with V including the liner's recession, and the
16 contraction ratio ``A_c / A_t`` at the eroded barrel and throat;
17* the wall: the thinnest liner left over the liner stations, and the graphite insert's back-face
18 temperature, labelled an *adiabatic upper bound* when no case is declared (D17);
19* the contour as built, and the wall at each replay point: the as-built radius plus each station's
20 recession, interpolated between stations the way the solver integrates the volume;
21* nozzle separation against the ambient the nozzle actually exhausts into (Summerfield; Schmucker);
22* the delivered Isp as a waterfall: ideal, less the c* loss, less the nozzle loss;
23* the soak-back after shutdown, when :func:`engine.layerx.replay.soak_back` ran.
24
25``hardware_block`` never raises: a block it cannot build is ``{available: False, error}``.
26"""
27
28from __future__ import annotations
29
30import math
31from typing import Any, Dict, List, Optional, Sequence
32
33import numpy as np
34
35PSI = 6894.757293168361
36G0 = 9.80665
37
38#: Summerfield's criterion: an overexpanded nozzle runs full while p_e/p_a stays above ~0.4.
39#: Summerfield, M., Foster, C. R. & Swan, W. C., "Flow separation in overexpanded supersonic exhaust
40#: nozzles", Jet Propulsion 24 (1954). A correlation of their test data, not a law; the same number is
41#: the repo's existing threshold (engine/pipeline/handcheck.py, frontend plume.ts, diag/limits.py).
42SUMMERFIELD_PE_PA = 0.4
43#: Schmucker's criterion, p_sep/p_a = (1.88 M - 1)^-0.64, with M the Mach number just upstream of
44#: separation: Schmucker, R. H., 1973 (Technische Universitaet Muenchen), in the form given by the
45#: review Oestlund, J. & Muhammad-Klingmann, B., "Supersonic flow separation with application to rocket
46#: engine nozzles", Applied Mechanics Reviews 58(3), 2005 (the review is the reference checked here;
47#: Schmucker's original report is not). Applied at the exit plane, M = M_e.
48SCHMUCKER_A = 1.88
49SCHMUCKER_N = -0.64
50
51
52def _num(v: Any, scale: float = 1.0) -> Optional[float]:
53 try:
54 f = float(v)
55 except (TypeError, ValueError):
56 return None
57 return f * scale if math.isfinite(f) else None
58
59
60def _col(rp: Dict[str, Any], key: str) -> List[Optional[float]]:
61 return [_num(v) for v in (rp.get(key) or [])]
62
63
64def _r(v: Optional[float], digits: int = 6) -> Optional[float]:
65 return None if v is None else round(float(v), digits)
66
67
68def _config_of(prep_or_config: Any) -> Any:
69 """The engine the replay ran: the line-exit runner's config for a ``Prepared`` (its geometry is
70 the design's), the object itself otherwise."""
71 link = getattr(prep_or_config, "link", None)
72 sampler = getattr(link, "sampler", None) if link is not None else None
73 if sampler is not None and getattr(sampler, "runner", None) is not None:
74 return sampler.runner.config
75 if hasattr(prep_or_config, "chamber_geometry"):
76 return prep_or_config
77 raise ValueError("hardware_block needs a Prepared with an EngineDesign link, or an engine config")
78
79
80def _cea_of(prep_or_config: Any, config: Any, cea_cache: Any) -> Any:
81 if cea_cache is not None:
82 return cea_cache
83 link = getattr(prep_or_config, "link", None)
84 sampler = getattr(link, "sampler", None) if link is not None else None
85 runner = getattr(sampler, "runner", None) if sampler is not None else None
86 if runner is not None:
87 return runner.cea_cache
88 from engine.core.runner import PintleEngineRunner
89
90 return PintleEngineRunner(config).cea_cache
91
92
93def _provenance(obj: Any, field: str, path: str) -> str:
94 """Where a config number came from: the schema's default (left unset, or written out equal to
95 it), or a value the config sets. The config carries no source for either."""
96 if obj is None:
97 return f"absent: {path} not declared"
98 fields_set = getattr(obj, "model_fields_set", None)
99 if fields_set is not None and field not in fields_set:
100 return f"schema default ({path}); no source recorded"
101 info = (getattr(type(obj), "model_fields", None) or {}).get(field)
102 default = getattr(info, "default", None) if info is not None else None
103 value = getattr(obj, field, None)
104 if default is not None and value is not None and not callable(default):
105 try:
106 if float(value) == float(default):
107 return f"config {path}, equal to the schema default; no source recorded"
108 except (TypeError, ValueError):
109 pass
110 return f"config {path}; no source recorded"
111
112
113def _inp(value: Any, unit: str, provenance: str) -> Dict[str, Any]:
114 return {"value": value, "unit": unit, "provenance": provenance}
115
116
117# ------------------------------------------------------------------ the pieces
118
119
120def eroded_radius(x: np.ndarray, r0: np.ndarray, stations: Dict[str, Dict[str, Any]], k: int,
121 x_liner_end: float, x_insert_end: float = 0.0) -> np.ndarray:
122 """The wall at replay point ``k``: ``r0(x) + s(x)``, with ``s`` the stations' recession: along the
123 liner (``x <= x_liner_end``) linear between liner stations and held flat beyond the first and
124 last; over the whole graphite insert (``x_liner_end < x <= x_insert_end``) the throat station's.
125
126 The insert's downstream half recedes too, though ``TimeVaryingCoupledSolver._geometry`` counts
127 only the upstream half in the chamber volume: the solver's throat is ``D_t0 + 2 s_throat``, so
128 the drawn wall's narrowest point must be that diameter. Left at the as-built radius, the
129 downstream half kept the drawn throat at its as-built size while the run reported the throat
130 area growing 5.7 % (2026-10-03). Downstream of the insert nothing recedes (the solver moves
131 only the exit area, and only when the nozzle ablates). Lengths in the units of ``x``, ``r0``."""
132 s = np.zeros_like(r0, dtype=float)
133 liner = sorted((st["x_mm"], (st["recession_mm"] or [None])[k]) for name, st in stations.items()
134 if st.get("kind") == "liner" and name.startswith("liner") and st.get("x_mm") is not None)
135 liner = [(xs, ss) for xs, ss in liner if ss is not None]
136 if liner:
137 xs = np.array([p[0] for p in liner], float)
138 ss = np.array([p[1] for p in liner], float)
139 lined = x <= x_liner_end + 1e-9
140 s[lined] = np.interp(x[lined], xs, ss)
141 thr = stations.get("throat")
142 if thr is not None and (thr.get("recession_mm") or [None])[k] is not None:
143 insert = (x > x_liner_end + 1e-9) & (x <= max(x_insert_end, 0.0) + 1e-9)
144 s[insert] = float(thr["recession_mm"][k])
145 return r0 + s
146
147
148def separation(t: Sequence[float], p_exit_pa: Sequence[Optional[float]], ambient_pa: Sequence[Optional[float]],
149 M_exit: Sequence[Optional[float]], threshold: float = SUMMERFIELD_PE_PA) -> Dict[str, Any]:
150 """Nozzle separation at each point, from the exit pressure, the ambient and the exit Mach number.
151
152 * Summerfield: separated when ``p_e/p_a < threshold`` (~0.4).
153 * Schmucker: separated when ``p_e/p_a < (1.88 M_e - 1)^-0.64``. ``schmucker_pa_crit_psia`` is the
154 ambient above which Schmucker predicts separation at this ``p_e``, ``M_e``:
155 ``p_e / (1.88 M_e - 1)^-0.64``; ``schmucker_pe_sep_psia`` the exit pressure below which it does
156 at this ambient: ``p_a (1.88 M_e - 1)^-0.64``.
157 """
158 pe, pa, me = list(p_exit_pa), list(ambient_pa), list(M_exit)
159 ratio: List[Optional[float]] = []
160 crit: List[Optional[float]] = []
161 for p, a, m in zip(pe, pa, me):
162 ratio.append(p / a if p is not None and a else None)
163 crit.append((SCHMUCKER_A * m - 1.0) ** SCHMUCKER_N if m is not None and SCHMUCKER_A * m > 1.0 else None)
164 summer = [None if r is None else bool(r < threshold) for r in ratio]
165 schm = [None if r is None or c is None else bool(r < c) for r, c in zip(ratio, crit)]
166 ok = [j for j, r in enumerate(ratio) if r is not None]
167 j = min(ok, key=lambda j: ratio[j]) if ok else None # type: ignore[arg-type,return-value]
168 return {
169 "pe_pa": [_r(p, 3) for p in pe],
170 "pe_psia": [_r(p / PSI, 5) if p is not None else None for p in pe],
171 "ambient_psia": [_r(a / PSI, 5) if a is not None else None for a in pa],
172 "ratio": [_r(r, 6) for r in ratio],
173 "summerfield": summer,
174 "M_exit": [_r(m, 6) for m in me],
175 "schmucker_ratio": [_r(c, 6) for c in crit],
176 "schmucker": schm,
177 "schmucker_pa_crit_psia": [_r(p / c / PSI, 5) if p is not None and c else None for p, c in zip(pe, crit)],
178 "schmucker_pe_sep_psia": [_r(a * c / PSI, 5) if a is not None and c else None for a, c in zip(pa, crit)],
179 "flag": bool(any(summer[i] for i in ok) or any(schm[i] for i in ok if schm[i] is not None)),
180 "min_ratio": _r(ratio[j], 6) if j is not None else None,
181 "t_min": float(t[j]) if j is not None else None,
182 "margin": _r(ratio[j] / threshold, 6) if j is not None else None,
183 "model": {
184 "name": "nozzle flow separation, Summerfield and Schmucker criteria at the exit plane",
185 "source": ("Summerfield, Foster & Swan, Jet Propulsion 24 (1954): p_e/p_a < ~0.4; Schmucker 1973 "
186 "(TU Muenchen) p_sep/p_a = (1.88 M - 1)^-0.64, in the form given by Oestlund & "
187 "Muhammad-Klingmann, Appl. Mech. Rev. 58(3) (2005)"),
188 "assumptions": [
189 "steady, full-flowing nozzle at each replay point; start and shutdown transients are not "
190 "modelled (the replay's first point is already at full chamber pressure), so separation "
191 "during them is not assessed",
192 "p_e and M_e are CEA shifting-equilibrium exit values at the replay's O/F, nozzle "
193 "stagnation pressure and eps(t) (engine/core/nozzle.py); not a boundary-layer solution",
194 "Schmucker applied with the exit Mach number: it predicts whether separation would sit "
195 "inside the nozzle at the exit, not where",
196 "ambient is the air the nozzle exhausts into: the site on the pad, the trajectory's in "
197 "flight (delivered.ambient_psia), interpolated onto the replay points",
198 ],
199 "inputs": {
200 "summerfield_threshold": _inp(threshold, "", "Summerfield, Foster & Swan 1954 (~0.4, a data "
201 "correlation; unmeasured for this nozzle)"),
202 "schmucker_a": _inp(SCHMUCKER_A, "", "Schmucker 1973 correlation constant"),
203 "schmucker_n": _inp(SCHMUCKER_N, "", "Schmucker 1973 correlation exponent"),
204 },
205 },
206 }
207
208
209def isp_breakdown(t: Sequence[float], rp: Dict[str, Any], cea_cache: Any, ambient_pa: float,
210 zeta_n: float, zeta_provenance: str) -> Dict[str, Any]:
211 """Delivered Isp as ideal less two losses, at each replay point, closing exactly.
212
213 EngineDesign's thrust (engine/core/nozzle.py ``calculate_thrust``) is
214 ``F = zeta_n Cf_vac(P0, eps) P0 A_t - p_a A_e`` with ``P0 = Pc/kappa`` (the Rayleigh loss of a
215 finite-area chamber), and its flow is ``mdot = P0 A_t / c*`` with ``c* = eta_c* c*_ideal(Pc)``
216 (chamber_solver.py: c* is referenced to the nozzle's stagnation pressure). So, with
217 ``Isp g0 = F/mdot = eta_c* c*_ideal [zeta_n Cf_vac(P0) - kappa p_a eps / Pc]``:
218
219 * ``ideal_s``: the CEA shifting-equilibrium Isp at this point's Pc, O/F and eps(t), at the
220 replay's ambient: ``c*_ideal (Cf_vac(Pc, eps) - p_a eps / Pc) / g0``;
221 * ``cstar_loss_s = (1 - eta_c*) ideal_s``: at a fixed chamber pressure and throat the thrust is
222 unchanged and the flow rises by 1/eta_c*, so the whole ideal Isp scales by eta_c*, the
223 ambient term included;
224 * ``nozzle_loss_s``: what remains, ``eta_c* ideal_s - delivered_s``, which is exactly
225 ``zeta_n_loss_s = eta_c* c*_ideal (1 - zeta_n) Cf_vac(P0) / g0
226 = (1 - zeta_n)/zeta_n (Isp + p_a A_e/(mdot g0))`` plus
227 ``stagnation_loss_s = eta_c* c*_ideal [Cf_vac(Pc) - Cf_vac(P0) + (kappa - 1) p_a eps/Pc] / g0``.
228 Because c* is referenced to P0, the stagnation loss reaches Isp only through the ambient
229 term and the small shift of Cf_vac with pressure (~0.1 s on LE4); in vacuum it vanishes.
230 """
231 pc = [_num(v, PSI) for v in rp.get("pc_psia") or []]
232 mr, eps = _col(rp, "mr"), _col(rp, "eps")
233 eta, cs_id = _col(rp, "eta_cstar"), _col(rp, "cstar_ideal")
234 isp, ae = _col(rp, "isp_s"), _col(rp, "A_exit_m2")
235 mo, mf = _col(rp, "mdot_O"), _col(rp, "mdot_F")
236 out: Dict[str, List[Optional[float]]] = {k: [] for k in (
237 "ideal_s", "cstar_loss_s", "nozzle_loss_s", "delivered_s", "zeta_n_loss_s", "stagnation_loss_s",
238 "Cf_vac_ideal", "eta_cstar")}
239 clamped: List[str] = []
240 for k in range(len(t)):
241 vals = (pc[k], mr[k], eps[k], eta[k], cs_id[k], isp[k], ae[k], mo[k], mf[k])
242 if any(v is None for v in vals):
243 for key in out:
244 out[key].append(None)
245 continue
246 Pc, MR, e, et, c_id, isp_d, A_e, m_o, m_f = vals # type: ignore[misc]
247 props = cea_cache.eval(MR, Pc, ambient_pa, e)
248 clamped.extend(props.get("clamped") or ())
249 cf_vac = float(props["Cf_vac"])
250 ideal = c_id * (cf_vac - ambient_pa * e / Pc) / G0
251 cstar_loss = (1.0 - et) * ideal
252 nozzle = et * ideal - isp_d
253 zeta_loss = (1.0 - zeta_n) / zeta_n * (isp_d + ambient_pa * A_e / ((m_o + m_f) * G0)) if zeta_n > 0 else None
254 out["ideal_s"].append(ideal)
255 out["cstar_loss_s"].append(cstar_loss)
256 out["nozzle_loss_s"].append(nozzle)
257 out["delivered_s"].append(isp_d)
258 out["zeta_n_loss_s"].append(zeta_loss)
259 out["stagnation_loss_s"].append(nozzle - zeta_loss if zeta_loss is not None else None)
260 out["Cf_vac_ideal"].append(cf_vac)
261 out["eta_cstar"].append(et)
262 res: Dict[str, Any] = {k: [_r(v, 6) for v in vals] for k, vals in out.items()}
263 res["zeta_n"] = zeta_n
264 res["ambient_pa"] = float(ambient_pa)
265 res["basis"] = ("at the replay's own ambient (the site's: EngineDesign's delivered thrust is at site "
266 "ambient, in flight too); ideal = CEA shifting equilibrium at the instantaneous Pc, O/F "
267 "and eps(t); delivered - (ideal - c* loss - nozzle loss) = 0 by construction")
268 res["model"] = {
269 "name": "delivered Isp waterfall: ideal, c* loss, nozzle loss",
270 "source": ("EngineDesign thrust model (engine/core/nozzle.py calculate_thrust: F = zeta_n Cf_vac P0 A_t "
271 "- p_a A_e, the RPA delivered-thrust basis) and CEA tables (engine/pipeline/cea_cache.py)"),
272 "assumptions": [
273 "eta_c* is the chamber solve's eta_vap x eta_mix x eta_HL (combustion_eff.py); its breakdown is "
274 "in the replay (eta_vap, eta_mix, eta_heat_loss)",
275 "the nozzle loss is one lumped efficiency zeta_n plus the Rayleigh stagnation loss (which, with "
276 "c* referenced to P0, acts only through the ambient term); there is no separate divergence, "
277 "boundary-layer, kinetic or two-phase term in EngineDesign",
278 f"zeta_n = {zeta_n:g}: {zeta_provenance}",
279 "Cf_vac read from the CEA table at Pc (the ideal reference), EngineDesign's thrust reads it at "
280 "P0 = Pc/kappa; the difference is inside stagnation_loss_s",
281 ],
282 "inputs": {
283 "nozzle_efficiency": _inp(zeta_n, "", zeta_provenance),
284 "ambient": _inp(float(ambient_pa), "Pa", "the replay's ambient (prep.ambient_pa: site)"),
285 },
286 }
287 if clamped:
288 res["model"]["assumptions"].append("CEA table clamped on " + ", ".join(sorted(set(clamped))) + " at some points")
289 return res
290
291
292def _zeta(config: Any) -> tuple:
293 cg = config.chamber_geometry
294 zeta = float(cg.nozzle_efficiency)
295 ms = getattr(config, "measurements", None)
296 mv = getattr(ms, "nozzle_efficiency", None) if ms is not None else None
297 if mv is not None:
298 return zeta, f"measured: {mv.source}"
299 fields_set = getattr(cg, "model_fields_set", set())
300 if "nozzle_efficiency" not in fields_set:
301 return zeta, ("schema default (ChamberGeometryConfig.nozzle_efficiency = 0.95, 'typical 0.94-0.98'); "
302 "unsourced (audit D16)")
303 return zeta, "config chamber_geometry.nozzle_efficiency; no source recorded (audit D16)"
304
305
306def _interp_onto(t: Sequence[float], ts: Sequence[float], ys: Sequence[Any]) -> List[Optional[float]]:
307 pts = [(float(a), float(b)) for a, b in zip(ts, ys) if _num(b) is not None and _num(a) is not None]
308 if not pts:
309 return [None] * len(t)
310 xa, ya = zip(*pts)
311 return [float(v) for v in np.interp(np.asarray(t, float), xa, ya)]
312
313
314# ------------------------------------------------------------------ the block
315
316
317def hardware_block(prep_or_config: Any, replay_result: Optional[Dict[str, Any]],
318 delivered: Optional[Dict[str, Any]] = None, series: Optional[Dict[str, Any]] = None, *,
319 soak: Optional[Dict[str, Any]] = None, cea_cache: Any = None,
320 ambient_pa: Optional[float] = None) -> Dict[str, Any]:
321 """``diagnostics.hardware`` for one burn (DATA-CONTRACT section 3).
322
323 ``prep_or_config``: the run's ``Prepared`` (preferred: its runner's config and CEA cache are the
324 replay's own) or the engine config. ``replay_result``: :func:`engine.layerx.replay.replay` of the
325 final pass. ``delivered``: :func:`engine.layerx.replay.delivered` with ``ambient_psia`` set (the
326 trajectory's air in flight); separation falls back to the replay's ambient without it.
327 ``series``: the twin's series (only its ``t`` is read, to bound the block). ``soak``: the
328 :func:`engine.layerx.replay.soak_back` result, else ``replay_result["soak"]``. ``cea_cache``
329 overrides the CEA table (built from the config when neither it nor a runner is at hand).
330 ``ambient_pa`` is a fallback for a replay that does not carry its own ``ambient_pa``: the Isp
331 breakdown must use the ambient the replay's thrust was computed at, so the replay's wins.
332
333 Never raises: ``{available: False, error}`` on failure.
334 """
335 try:
336 return _hardware_block(prep_or_config, replay_result, delivered, series, soak, cea_cache, ambient_pa)
337 except Exception as exc: # noqa: BLE001 - a diagnostic that cannot be built is reported; the burn stands
338 return {"available": False, "error": f"{type(exc).__name__}: {exc}"}
339
340
341def _hardware_block(prep_or_config: Any, rp: Optional[Dict[str, Any]], delivered: Optional[Dict[str, Any]],
342 series: Optional[Dict[str, Any]], soak: Optional[Dict[str, Any]], cea_cache: Any,
343 ambient_pa: Optional[float]) -> Dict[str, Any]:
344 from engine.pipeline.config_schemas import ensure_chamber_geometry
345 from engine.pipeline.thermal import gas_side
346
347 if not rp or not rp.get("available"):
348 return {"available": False, "error": (rp or {}).get("error") or "no erosion replay for this run"}
349 if "stations" not in rp:
350 return {"available": False, "error": "this replay predates the kept wall stations (re-run the burn)"}
351 import copy as _copy
352
353 config = _config_of(prep_or_config)
354 t = [float(v) for v in rp["t"]]
355 n = len(t)
356 a_t = _col(rp, "A_throat_m2")
357 a_e = _col(rp, "A_exit_m2")
358 vol = _col(rp, "V_chamber_m3")
359 d_t = _col(rp, "D_throat_mm")
360 d_c = _col(rp, "D_chamber_mm")
361 st: Dict[str, Dict[str, Any]] = rp.get("stations") or {}
362
363 def ratio(a: Sequence[Optional[float]], b: Sequence[Optional[float]], p: float = 1.0) -> List[Optional[float]]:
364 return [(x / y) ** p if x is not None and y else None for x, y in zip(a, b)]
365
366 eps = ratio(a_e, a_t)
367 lstar = ratio(vol, a_t)
368 contraction = ratio(d_c, d_t, 2.0)
369 liners = [s for name, s in st.items() if s.get("kind") == "liner" and name.startswith("liner")]
370 liner_min = [min((s["remaining_mm"][k] for s in liners if s["remaining_mm"][k] is not None), default=None)
371 for k in range(n)] if liners else [None] * n
372 gr = getattr(config, "graphite_insert", None)
373 graphite = bool(gr is not None and getattr(gr, "enabled", False))
374 case = getattr(config, "stainless_steel_case", None)
375 case_declared = bool(case is not None and getattr(case, "enabled", False))
376 back_basis = ("insert back face (its interface with the backing behind it: the liner's phenolic out to the "
377 "case, assumed); the case's outer face adiabatic to the room (upper bound)" if case_declared
378 else "adiabatic upper bound")
379 thr = st.get("throat") or {}
380 # The insert's back face is the graphite layer's back: the station's back face when nothing is
381 # behind it (LE4), the graphite/case interface when a case is declared (the station's back face
382 # is then the case's outer face, not the insert's).
383 back_key = "T_interface_K" if case_declared else "T_back_K"
384 insert_back = list(thr.get(back_key) or [None] * n) if graphite and thr.get("kind") == "graphite" else [None] * n
385
386 # ---- the contour: as built, and the wall at each replay point
387 cg0 = ensure_chamber_geometry(_copy.deepcopy(config))
388 c0 = gas_side.contour_for(cg0)
389 x_mm = np.asarray(c0.x, float) * 1e3
390 r0_mm = np.asarray(c0.r, float) * 1e3
391 if graphite:
392 half = float(gr.axial_half_length or gr.axial_half_length_ratio * 2.0 * c0.R_t)
393 else:
394 half = 0.0
395 x_liner_end = -half * 1e3
396 frames = [eroded_radius(x_mm, r0_mm, st, k, x_liner_end, half * 1e3) for k in range(n)]
397 abl = getattr(config, "ablative_cooling", None)
398 lined = bool(abl is not None and getattr(abl, "enabled", False))
399 contour: Dict[str, Any] = {
400 "x_mm": [round(float(v), 4) for v in x_mm],
401 "r0_mm": [round(float(v), 4) for v in r0_mm],
402 "frames": {"t": t, "r_mm": [[round(float(v), 4) for v in f] for f in frames]},
403 "x_face_mm": round(float(c0.x_face) * 1e3, 4),
404 "x_liner_end_mm": round(x_liner_end, 4) if lined else None,
405 "x_insert_mm": [round(-half * 1e3, 4), round(half * 1e3, 4)] if graphite else None,
406 }
407 if lined:
408 t_l = float(abl.initial_thickness) * 1e3
409 contour["liner_r_mm"] = [round(float(r + t_l), 4) if x <= x_liner_end + 1e-9 else None
410 for x, r in zip(x_mm, r0_mm)]
411 if graphite:
412 t_g = float(gr.initial_thickness) * 1e3
413 contour["insert_r_mm"] = [round(float(r + t_g), 4) if -half * 1e3 - 1e-9 <= x <= half * 1e3 + 1e-9 else None
414 for x, r in zip(x_mm, r0_mm)]
415
416 # ---- separation, on the replay points, against the air the nozzle exhausts into
417 amb_rp = float(rp.get("ambient_pa") or ambient_pa or getattr(prep_or_config, "ambient_pa", 101325.0))
418 if delivered and delivered.get("ambient_psia") and delivered.get("t"):
419 amb_at = [None if a is None else a * PSI for a in _interp_onto(t, delivered["t"], delivered["ambient_psia"])]
420 amb_basis = "delivered.ambient_psia (site on the pad, the trajectory's air in flight)"
421 else:
422 amb_at = [amb_rp] * n
423 amb_basis = "the replay's ambient (site)"
424 pe = [_num(v, PSI) for v in rp.get("p_exit_psia") or [None] * n]
425 sep = separation(t, pe, amb_at, _col(rp, "M_exit"))
426 sep["ambient_basis"] = amb_basis
427
428 # ---- Isp waterfall
429 zeta, zeta_prov = _zeta(config)
430 cea = _cea_of(prep_or_config, config, cea_cache)
431 isp = isp_breakdown(t, rp, cea, amb_rp, zeta, zeta_prov)
432
433 # ---- soak-back
434 sk = soak if soak is not None else rp.get("soak")
435 if sk is None:
436 sk = {"available": False, "error": "soak-back not run for this replay (engine.layerx.replay.soak_back)"}
437 elif sk.get("available"):
438 sk = {**sk, "basis": sk.get("back_basis", back_basis)}
439
440 ar = _col(rp, "throat_area_ratio")
441 out: Dict[str, Any] = {
442 "available": True,
443 "t": t,
444 "index": list(rp.get("index") or []),
445 "replay_point": [True] * n,
446 "throat_d_mm": [_r(v, 5) for v in d_t],
447 "At_ratio": [_r(v, 7) for v in ar],
448 "eps": [_r(v, 6) for v in eps],
449 "Lstar_m": [_r(v, 6) for v in lstar],
450 "contraction": [_r(v, 6) for v in contraction],
451 "liner_min_mm": [_r(v, 5) for v in liner_min],
452 "insert_back_K": [_r(v, 3) for v in insert_back],
453 "insert_back_basis": back_basis if graphite else None,
454 "contour": contour,
455 "separation": sep,
456 "isp": isp,
457 "soak": sk,
458 "heatmap": "sidecar:axial",
459 "summary": {
460 "throat_growth": (ar[-1] - 1.0) if ar and ar[-1] is not None else None,
461 "throat_d_end_mm": d_t[-1] if d_t else None,
462 "eps_start": eps[0] if eps else None, "eps_end": eps[-1] if eps else None,
463 "Lstar_start_m": lstar[0] if lstar else None, "Lstar_end_m": lstar[-1] if lstar else None,
464 "liner_min_end_mm": liner_min[-1] if liner_min else None,
465 "insert_back_end_K": insert_back[-1] if insert_back else None,
466 # A_t at the first replay point over as built: the first step has already eroded (the
467 # per-pass throat_growth is measured from this point, delivered's from as built).
468 "first_point_area_ratio": ar[0] if ar else None,
469 },
470 }
471 if series is not None and series.get("t"):
472 out["series_span"] = [float(series["t"][0]), float(series["t"][-1])]
473 out["model"] = _model(config, rp, graphite, lined, case_declared, half, back_basis)
474 return out
475
476
477def _model(config: Any, rp: Dict[str, Any], graphite: bool, lined: bool, case_declared: bool, half: float,
478 back_basis: str) -> Dict[str, Any]:
479 abl = getattr(config, "ablative_cooling", None)
480 gr = getattr(config, "graphite_insert", None)
481 inputs: Dict[str, Any] = {}
482 if lined:
483 inputs["liner_thickness"] = _inp(float(abl.initial_thickness), "m",
484 _provenance(abl, "initial_thickness", "ablative_cooling.initial_thickness"))
485 inputs["liner_coverage_fraction"] = _inp(float(abl.coverage_fraction), "",
486 _provenance(abl, "coverage_fraction", "ablative_cooling.coverage_fraction")
487 + "; scales the volume change only")
488 for f, u in (("thermal_conductivity", "W/(m K)"), ("material_density", "kg/m^3"), ("specific_heat", "J/(kg K)"),
489 ("ablation_surface_temperature", "K"), ("heat_of_ablation", "J/kg")):
490 inputs[f"liner_{f}"] = _inp(float(getattr(abl, f)), u, _provenance(abl, f, f"ablative_cooling.{f}"))
491 if graphite:
492 inputs["insert_thickness"] = _inp(float(gr.initial_thickness), "m",
493 _provenance(gr, "initial_thickness", "graphite_insert.initial_thickness"))
494 inputs["insert_half_length"] = _inp(half, "m",
495 _provenance(gr, "axial_half_length", "graphite_insert.axial_half_length")
496 if gr.axial_half_length else
497 _provenance(gr, "axial_half_length_ratio",
498 "graphite_insert.axial_half_length_ratio") + " x D_t")
499 inputs["insert_conductivity"] = _inp(float(gr.thermal_conductivity), "W/(m K)",
500 _provenance(gr, "thermal_conductivity", "graphite_insert.thermal_conductivity"))
501 inputs["case_declared"] = _inp(case_declared, "", _provenance(config, "stainless_steel_case", "stainless_steel_case"))
502 return {
503 "name": "chamber and nozzle hardware along the burn (EngineDesign erosion replay)",
504 "source": ("engine/pipeline/time_varying_solver.py (wall stations, 1-D transient conduction per station; "
505 "geometry rebuilt from cumulative recession) and engine/pipeline/thermal/gas_side.py "
506 "wall_contour (barrel, 45 deg cone, 1.5 R_t arc, Rao bell)"),
507 "assumptions": [
508 "eroded frames are the as-built contour plus each station's recession, linear between the liner "
509 "stations and flat beyond them, the throat station's over the insert's upstream half: the "
510 "recession the solver integrates the volume from. The solver's own gas loads use a contour "
511 "redrawn from (A_t, bore, V) each step (gas_side.contour_for), which is not this drawing: at the "
512 "liner end the redraw moves the wall ~0.2 mm where the station receded ~2 mm (audit 5.4)",
513 "the insert's downstream half and the divergent nozzle do not recede in the model; A_e is fixed "
514 "unless the nozzle ablates, so eps = A_e/A_t(t) falls as the throat opens",
515 "liner and insert drawn as radial offsets of the as-built contour by their initial thickness",
516 "L* = V(t)/A_t(t), with V(t) the as-built volume plus the liner recession times its coverage "
517 "fraction plus the insert's upstream-half recession",
518 "contraction ratio = (D_barrel(t)/D_t(t))^2, at the barrel station",
519 f"insert back face: {back_basis}" + ("" if case_declared else
520 "; no case or backing is declared, so nothing behind the "
521 "insert takes heat (audit D17)"),
522 "values at the replay points only; the twin's steps between them are interpolations",
523 ],
524 "inputs": inputs,
525 }
#define min(a, b)
Definition Arduino.h:42
#define max(a, b)
Definition Arduino.h:43
Dict[str, Any] hardware_block(Any prep_or_config, Optional[Dict[str, Any]] replay_result, Optional[Dict[str, Any]] delivered=None, Optional[Dict[str, Any]] series=None, *, Optional[Dict[str, Any]] soak=None, Any cea_cache=None, Optional[float] ambient_pa=None)
Definition hardware.py:320
Optional[float] _num(Any v, float scale=1.0)
Definition hardware.py:52
Dict[str, Any] isp_breakdown(Sequence[float] t, Dict[str, Any] rp, Any cea_cache, float ambient_pa, float zeta_n, str zeta_provenance)
Definition hardware.py:210
List[Optional[float]] _interp_onto(Sequence[float] t, Sequence[float] ts, Sequence[Any] ys)
Definition hardware.py:306
Dict[str, Any] separation(Sequence[float] t, Sequence[Optional[float]] p_exit_pa, Sequence[Optional[float]] ambient_pa, Sequence[Optional[float]] M_exit, float threshold=SUMMERFIELD_PE_PA)
Definition hardware.py:149
Dict[str, Any] _model(Any config, Dict[str, Any] rp, bool graphite, bool lined, bool case_declared, float half, str back_basis)
Definition hardware.py:478
List[Optional[float]] _col(Dict[str, Any] rp, str key)
Definition hardware.py:60
str _provenance(Any obj, str field, str path)
Definition hardware.py:93
Dict[str, Any] _hardware_block(Any prep_or_config, Optional[Dict[str, Any]] rp, Optional[Dict[str, Any]] delivered, Optional[Dict[str, Any]] series, Optional[Dict[str, Any]] soak, Any cea_cache, Optional[float] ambient_pa)
Definition hardware.py:343
Any _config_of(Any prep_or_config)
Definition hardware.py:68
Dict[str, Any] _inp(Any value, str unit, str provenance)
Definition hardware.py:113
Optional[float] _r(Optional[float] v, int digits=6)
Definition hardware.py:64
Any _cea_of(Any prep_or_config, Any config, Any cea_cache)
Definition hardware.py:80
tuple _zeta(Any config)
Definition hardware.py:292
np.ndarray eroded_radius(np.ndarray x, np.ndarray r0, Dict[str, Dict[str, Any]] stations, int k, float x_liner_end, float x_insert_end=0.0)
Definition hardware.py:121