STAR
Cal STAR — firmware, DAQ server, and engine design
Loading...
Searching...
No Matches
test_layerx_optimize_modes.py
Go to the documentation of this file.
1"""Optimize rebuilt (docs/layerx/AUDIT.md D6, 9.10): the trade study's removal, Hardware mode, the
2catalogues it draws from, and the one change-list format Optimize and Injector holes share.
3
4Set point has its own file (tests/test_layerx_setpoint.py). Nothing here burns: the burn is the
5function every mode is handed, and the tests hand it a closed-form stand.
6"""
7
8from __future__ import annotations
9
10import json
11import os
12import sys
13
14import pytest
15
16sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
17
18from fastapi import FastAPI # noqa: E402
19from fastapi.testclient import TestClient # noqa: E402
20
21from backend.routers import layerx as lx # noqa: E402
22from backend.session import UserSession, get_session # noqa: E402
23
24
25@pytest.fixture()
26def api(tmp_path, monkeypatch):
27 monkeypatch.setenv("USERDATA_DIR", str(tmp_path))
28 session = UserSession(f"optmodes-{tmp_path.name}")
29 app = FastAPI()
30 app.include_router(lx.router)
31 app.dependency_overrides[get_session] = lambda: session
32 yield TestClient(app), session
33 with lx._JOBS_LOCK:
34 for job_id in [j.id for j in lx._JOBS.values() if j.user == session.user]:
35 lx._JOBS.pop(job_id, None)
36
37
38def _saved(session, run_id: str, kind: str, result=None) -> None:
39 job = lx.Job(session.user, lx.Settings(drawing_id="d"), "design", kind=kind)
40 job.id = run_id
41 job.status, job.finished = "done", 1.0
42 job.result = result or {"summary": {"burn_time_s": 3.5, "total_impulse_Ns": 24000.0}}
43 lx._persist(lx._runs_dir(session), job)
44
45
46# ------------------------------------------------------------------ the trade study is gone
47
48
50 client, session = api
51 _saved(session, "20261002-093657-c71a29", "trade", {"x": {"key": "lockup_psia"}, "points": [{"ok": True}]})
52 _saved(session, "20261002-100000-aaaaaa", "run")
53 listed = {r["id"]: r for r in client.get("/api/layerx/runs").json()}
54 assert listed["20261002-093657-c71a29"]["legacy"] is True
55 assert "legacy" not in listed["20261002-100000-aaaaaa"]
56 got = client.get("/api/layerx/runs/20261002-093657-c71a29").json()
57 assert got["legacy"] is True and got["kind"] == "trade" and got["result"]["points"] == [{"ok": True}]
58 # Marked at read time only: the files on disk are as they were written.
59 index = json.loads((lx._runs_dir(session) / "_index" / "20261002-093657-c71a29.json").read_text())
60 assert "legacy" not in index
61
62
64 import importlib.util
65
66 client, _ = api
67 assert importlib.util.find_spec("engine.layerx.trade") is None
68 for path in ("/api/layerx/trade", "/api/layerx/trade/axes"):
69 assert client.post(path, json={}).status_code in (404, 405)
70 # What the sweep shares with it stays.
71 from engine.layerx.pool import Cancelled, WorkerPool, default_workers # noqa: F401
72
73
74# ------------------------------------------------------------------ the catalogues
75
76
78 from engine.layerx import catalog as cat
79
80 cs = cat.load_catalogs(None)
81 no51 = next(r for r in cs["drills"] if r["id"] == "drill-no51")
82 assert cat.param_si(no51, "d") == pytest.approx(0.0670 * 25.4) # ASME B94.11M #51 = 0.0670 in
83 assert "B94.11M" in no51["provenance"]
84 # Seeded with the named parts only: the Aqua 1092-50 at Cv 0.8 and the 1.7 Cv press solenoid.
85 by = {r["id"]: r for r in cs["valves"]}
86 assert cat.param_si(by["aqua-1092-50"], "Cv") == 0.8 and cat.param_si(by["press-solenoid-cv1.7"], "Cv") == 1.7
87 for kind in cat.FILE_KINDS:
88 for r in cs[kind]:
89 assert r["provenance"].strip() and r["source"] in cat.SOURCES, r["id"]
90 # A person's row replaces a shipped one by id; a row with no provenance is refused, and said so.
91 udir = tmp_path / "layerx" / "catalog"
92 udir.mkdir(parents=True)
93 (udir / "valves.json").write_text(json.dumps({"schema": cat.SCHEMA, "kind": "valves", "rows": [
94 {"id": "press-solenoid-cv1.7", "label": "x", "applies_to": ["SOL"], "params": {"Cv": {"value": 1.75, "unit": "Cv"}},
95 "source": "measured", "provenance": "flow bench 2026-10-04"}]}))
96 mine = {r["id"]: r for r in cat.load_catalogs(tmp_path)["valves"]}
97 assert cat.param_si(mine["press-solenoid-cv1.7"], "Cv") == 1.75 and mine["press-solenoid-cv1.7"]["origin"] == "user"
98 (udir / "tubes.json").write_text(json.dumps({"schema": cat.SCHEMA, "kind": "tubes", "rows": [
99 {"id": "t", "label": "x", "applies_to": ["line"], "params": {"bore": {"value": 4.0, "unit": "mm"}},
100 "source": "manufacturer", "provenance": ""}]}))
101 out = cat.load_catalogs(tmp_path)
102 assert any("no provenance" in p for p in out["problems"]) and out["tubes"] # shipped rows stand
103
104
105# ------------------------------------------------------------------ Hardware mode, closed-form stand
106
107
109 """D 10.92 mm (l_ox1), Do 9.70 mm, C 0.6: beta 0.888278, beta^4 0.622582;
110 K = [sqrt(1 - 0.622582 x 0.64) / (0.6 x 0.789038) - 1]^2 = 0.407389 (ISO 5167-2 permanent loss)."""
111 from engine.layerx.optimize import trim_orifice_C, trim_orifice_K
112
113 assert trim_orifice_K(10.92e-3, 9.70e-3, 0.6) == pytest.approx(0.4073891, rel=1e-6)
114 # Reader-Harris/Gallagher, corner taps, at the LOX line's flow: the audit's 0.542 (AUDIT 5.3).
115 assert trim_orifice_C(10.92e-3, 9.70e-3, 1142.1, 1.9e-4, 1.826) == pytest.approx(0.542, abs=1e-3)
116
117
119 from engine.layerx.sources import shipped_drawings_dir
120
121 path = shipped_drawings_dir() / "copv_study_he.json"
122 if not path.is_file():
123 pytest.skip("feed-twin's shipped drawings are not next to this checkout")
124 return json.loads(path.read_text())
125
126
127class _Net:
128 def conditions(self, node, p, signals=None):
129 from types import SimpleNamespace
130
131 return SimpleNamespace(rho=1142.1, mu=1.9e-4)
132
133
134def _prep(payload):
135 from types import SimpleNamespace
136
137 return SimpleNamespace(
138 drawing=SimpleNamespace(payload=payload, name="copv_study_he", id="he", sha256="s"), measurements=[],
139 derived={"species": {"oxidiser": "oxygen", "fuel": "ethanol"}, "target_lockup_psia": 578.0,
140 "inlet_nodes": {"oxidiser": "ox_in", "fuel": "fu_in"}},
141 inlet_nodes={"oxidiser": "ox_in", "fuel": "fu_in"},
142 model=SimpleNamespace(built=SimpleNamespace(network=_Net())), config_sha256="a" * 64)
143
144
145D_O0, D_F0 = 1.6318401623160582e-3, 1.4715307322406754e-3
146
147
148def _hw_stand(calls=None):
149 """O/F = 1.5212 (dO/dO0)^2 (dF0/dF)^2 / sqrt(1 + 0.03 dK_lox); mean thrust 7013 + 10.1 (L - 578);
150 spread 6.4 - 0.3 (Cv_lox_press - 1.7) %."""
151
152 def evaluate(args_list):
153 out = []
154 for a in args_list:
155 if calls is not None:
156 calls.append(a)
157 g = a["config"].injector.geometry
158 ov = {(o.target, o.parameter): o.value for o in a.get("extra_overrides") or []}
159 dK = ov.get(("edge:l_ox1", "K_minor"), 0.5) - 0.5
160 cv = ov.get(("node:SV_LOX_PRESS", "Cv"), 1.7)
161 L = a.get("lockup_psia") or 578.0
162 of = 1.5212 * (g.oxidizer.d_jet / D_O0) ** 2 * (D_F0 / g.fuel.d_jet) ** 2 / (1.0 + 0.03 * dK) ** 0.5
163 F = 7013.0 + 10.1 * (L - 578.0)
164 figs = {"mean_thrust_N": F, "of_mean": of, "thrust_spread_pct": 6.4 - 0.3 * (cv - 1.7),
165 "total_impulse_Ns": F * 3.45, "burn_time_s": 3.45, "copv_spare_psi": 900.0, "ox_used_kg": 6.3,
166 "fuel_used_kg": 6.3 / of, "lockup_psia": L, "pc_mean_psia": 0.69 * L}
167 out.append({"ok": True, "preflight": [], "lockup_psia": L, "fill_psig": 4500.0, "replay": True,
168 "dome_psig": L - 64.4, "figures": figs, "limits": [], "tripped": None,
169 "derived": {"config_sha256": "a" * 64, "drawing": {"id": "he"}}})
170 return out
171
172 return evaluate
173
174
175@pytest.fixture(scope="module")
176def le4():
177 pytest.importorskip("feedtwin")
178 from engine.pipeline.io import load_config
179
180 # The undrilled holes the hand numbers below are worked from, not the live
181 # design: configs/ethalox_6800N.yaml has since been drilled to 1.65 mm.
182 return load_config(os.path.join(os.path.dirname(os.path.abspath(__file__)), "fixtures",
183 "ethalox_6800N_undrilled_2026-10-04.yaml"))
184
185
186def _hardware(config, payload, **req):
187 from types import SimpleNamespace
188
189 from engine.layerx import catalog as cat
190 from engine.layerx.optimize import HardwareRequest, run_hardware
191
192 return run_hardware(config, None, SimpleNamespace(replay=True), [], HardwareRequest.from_dict(req),
193 catalogs=cat.load_catalogs(None), evaluate=_hw_stand(), prep=_prep(payload))
194
195
197 """LOX holes 1.6318 mm, O/F 1.5212 against 1.5. Neighbour drills and the stand's O/F by hand:
198 #52 1.6129 mm -> 1.4861 (0.93 % off); 1.60 -> 1.4624 (2.5 %); 1.65 -> 1.5552 (3.7 %);
199 1.70 -> 1.6509 (10 %); drawn 1.4 % off. #52 wins; a smaller hole is a new plate."""
200 out = _hardware(le4, _he_payload(), components=[{"target": "design:oxidizer.d_jet"}], objective="of_error",
201 target_thrust_N=7200.0, design_of=1.5, neighbours=2)
202 assert out["improves"] and out["winner"]["summary"].startswith("LOX holes #52")
203 assert out["winner"]["figures"]["of_mean"] == pytest.approx(1.5212 * (0.0635 * 25.4e-3 / D_O0) ** 2, rel=1e-9)
204 assert len(out["candidates"]) == 4 and out["candidates"][0]["rank"] is not None
205 by = {c["field"]: c for c in out["change_list"]["changes"]}
206 assert by["d_jet"]["after"] == pytest.approx(1.6129, abs=1e-4) and by["d_jet"]["cad_impact"] == "new plate"
207 assert by["d_jet"]["target"] == "design:injector.geometry.oxidizer.d_jet" and by["d_jet"]["drill"]["drill"] == "#52"
208 assert by["orifice_l_over_d"]["after"] == pytest.approx(5.0 * D_O0 / 1.6129e-3, rel=1e-4)
209 assert by["lockup_psia"]["cad_impact"] == "setting only" # the set point, re-solved
210 assert out["final"]["figures"]["mean_thrust_N"] == pytest.approx(7200.0, abs=7.2)
211 w = out["change_list"]["exports"]["design_write"]
212 assert w["method"] == "PUT" and w["query"] == {"expect_sha256": "a" * 64} and w["requires_confirmation"]
213 assert w["body"]["injector"]["geometry"]["oxidizer"]["d_jet"] == pytest.approx(1.6129e-3, abs=1e-7)
214 assert out["change_list"]["exports"]["pid_designer"] is None # nothing on the drawing moved
215
216
218 """Baseline: O/F 1.5212, lockup 578, Pc 398.82 psia. A trim taking O/F to 1.5 at a fixed drop
219 needs (1 - (1.5/1.5212)^2) x 179.18 psi = 4.959 psi of permanent loss in l_ox1."""
220 from engine.layerx.optimize import trim_orifice_C, trim_orifice_K
221
222 payload = _he_payload()
223 before = json.dumps(payload, sort_keys=True)
224 out = _hardware(le4, payload, components=[{"target": "edge:l_ox1", "kind": "trim_orifice"}], objective="of_error",
225 design_of=1.5, neighbours=2, verify=False)
226 trims = [c["trim"] for c in out["candidates"]]
227 assert trims and all(t["sized_for_psi"] == pytest.approx(4.9594, abs=1e-3) for t in trims)
228 mdot = 6.3 / 3.45
229 for t in trims:
230 C = trim_orifice_C(10.92e-3, t["bore_mm"] * 1e-3, 1142.1, 1.9e-4, mdot)
231 assert t["K"] == pytest.approx(trim_orifice_K(10.92e-3, t["bore_mm"] * 1e-3, C), rel=1e-9)
232 # The drills bracket the bore that gives exactly the loss: one smaller (more loss), one larger.
233 v = mdot / (1142.1 * 3.141592653589793 * 10.92e-3 ** 2 / 4)
234 dps = sorted(t["K"] * 0.5 * 1142.1 * v * v / 6894.757293168361 for t in trims)
235 assert dps[0] < 4.9594 < dps[-1]
236 assert out["needs_pid_designer"][0]["target"] == "edge:l_ox1"
237 (ch,) = [c for c in out["change_list"]["changes"] if c["field"] == "K_minor"]
238 assert ch["domain"] == "drawing" and ch["cad_impact"] == "new part" and ch["before"] == 0.5
239 exp = out["change_list"]["exports"]["pid_designer"]
240 l_ox1 = next(e for e in exp["edges"] if e["id"] == "l_ox1")
241 assert l_ox1["data"]["params"]["K_minor"]["value"] == pytest.approx(ch["after"])
242 assert exp["needs_pid_designer"] and json.dumps(payload, sort_keys=True) == before # the drawing is untouched
243 assert out["trim_model"]["inputs"]["C"]["provenance"].startswith("Reader-Harris")
244
245
247 """SV_LOX_PRESS is 1.7 Cv; the catalogue's other solenoid rows are 1.2, 3.8 and 26.1. Spread
248 6.4 - 0.3 (Cv - 1.7): the 26.1 Cv ball valve flattens most (estimated, and said so)."""
249 out = _hardware(le4, _he_payload(), components=[{"target": "node:SV_LOX_PRESS"}], objective="thrust_flatness",
250 neighbours=2, verify=False)
251 assert sorted(c["summary"] for c in out["candidates"]) == sorted([
252 "SV-LOX-PRESS: Solenoid, Cv 1.2 (ethalox_stand drawing)", "SV-LOX-PRESS: Vent solenoid, Cv 3.8 (calibrated)",
253 "SV-LOX-PRESS: Main valve, 1/2 in full-port ball valve"])
254 cv = next(c for c in out["change_list"]["changes"] if c["field"] == "Cv")
255 assert cv["after"] == 26.1 and cv["before"] == 1.7 and cv["source"] == "estimated"
256 assert cv["catalog"]["id"] == "main-ball-valve-half-inch" and "Crane TP-410" in cv["provenance"]
257 assert cv["pid_node_id"] == "SV_LOX_PRESS" and cv["cad_impact"] == "new part"
258 # Each number keeps its own provenance: the ball valve's Cv is estimated (Crane), its 1/2 in bore is
259 # the manufacturer's -- not the row's one tag for both.
260 bore = next(c for c in out["change_list"]["changes"] if c["field"] == "bore")
261 assert (bore["before"], bore["after"], bore["source"]) == (6.35, 12.7, "manufacturer")
262 exp = out["change_list"]["exports"]["pid_designer"]
263 sv = next(n for n in exp["nodes"] if n["id"] == "SV_LOX_PRESS")["data"]["params"]
264 assert (sv["Cv"]["source"], sv["bore"]["source"]) == ("estimated", "manufacturer")
265
266
268 """KB1 on the drawings: volume measured, service pressure manufacturer, wall mass and MAWP
269 estimated. The catalogue row carries each, and a wrong per-parameter tag is refused."""
270 from engine.layerx import catalog as cat
271
272 (row,) = cat.load_catalogs(None)["bottles"]
273 assert {k: p["source"] for k, p in row["params"].items()} == {
274 "volume": "measured", "pressure": "manufacturer", "wall_mass": "estimated", "MAWP": "estimated"}
275 with pytest.raises(ValueError, match="source 'guessed'"):
276 cat._check_row("bottles", {**row, "params": {"volume": {"value": 5.0, "unit": "L", "source": "guessed"}}}, "x")
277
278
280 """Every candidate breaks a limit the drawn hardware keeps (the stand marks it). #52 still ranks
281 first among them on O/F, but it is not reported as an improvement."""
282 from engine.layerx import catalog as cat
283 from engine.layerx.optimize import HardwareRequest, run_hardware
284 from types import SimpleNamespace
285
286 stand = _hw_stand()
287
288 def evaluate(args_list):
289 out = stand(args_list)
290 for a, b in zip(args_list, out):
291 if a.get("tag") != "baseline":
292 b["limits"] = [{"key": "stiffness_ox", "label": "LOX injector dP/Pc", "grade": "bad"}]
293 return out
294
295 out = run_hardware(le4, None, SimpleNamespace(replay=True), [],
296 HardwareRequest.from_dict({"components": [{"target": "design:oxidizer.d_jet"}],
297 "objective": "of_error", "design_of": 1.5, "verify": False}),
298 catalogs=cat.load_catalogs(None), evaluate=evaluate, prep=_prep(_he_payload()))
299 assert out["winner"]["summary"].startswith("LOX holes #52") and out["winner"]["limits_bad"]
300 assert out["improves"] is False
301 assert any("within the limits" in n for n in out["notes"])
302
303
305 from types import SimpleNamespace
306
307 from engine.layerx import catalog as cat
308 from engine.layerx.optimize import HardwareRequest, plan_candidates, run_hardware
309
310 no_of = SimpleNamespace(design_requirements=SimpleNamespace(optimal_of_ratio=None, target_thrust=7200.0))
311 with pytest.raises(ValueError, match="needs a design O/F"):
312 run_hardware(no_of, None, SimpleNamespace(replay=True), [],
313 HardwareRequest.from_dict({"components": [{"target": "node:SV_LOX_PRESS"}]}),
314 catalogs=cat.load_catalogs(None), evaluate=_hw_stand(), prep=_prep(_he_payload()))
315 # A line itemised into segments: feedtwin drops its own bore and K_minor, so a restated tube or trim
316 # would burn as drawn. Refused, not burned as a no-op.
317 payload = _he_payload()
318 edge = next(e for e in payload["edges"] if e["id"] == "l_ox1")
319 edge["data"]["segments"] = [{"kind": "straight", "length": {"value": 0.07, "unit": "m", "source": "measured"}}]
320 for comp in ({"target": "edge:l_ox1"}, {"target": "edge:l_ox1", "kind": "trim_orifice"}):
321 with pytest.raises(ValueError, match="itemised into 1 segment"):
322 plan_candidates(_prep(payload), le4, HardwareRequest.from_dict({"components": [comp]}), cat.load_catalogs(None))
323
324
326 from engine.layerx import catalog as cat
327 from engine.layerx.optimize import HardwareRequest, plan_candidates
328
329 prep = _prep(_he_payload())
330 with pytest.raises(ValueError, match="no node:NOPE"):
331 plan_candidates(prep, le4, HardwareRequest.from_dict({"components": [{"target": "node:NOPE"}]}),
332 cat.load_catalogs(None))
333 with pytest.raises(ValueError, match="trim orifice goes in a line"):
334 HardwareRequest.from_dict({"components": [{"target": "node:SV_LOX_PRESS", "kind": "trim_orifice"}]})
335 # Row ids as a bare string would be matched by substring: refused.
336 with pytest.raises(ValueError, match="a list of catalogue row ids"):
337 HardwareRequest.from_dict({"components": [{"target": "edge:l_ox1", "rows": "tube-6.35-bore"}]})
338 with pytest.raises(ValueError, match="no catalogue row"):
339 plan_candidates(prep, le4, HardwareRequest.from_dict({"components": [{"target": "node:KB1"}]}),
340 cat.load_catalogs(None))
341
342
343# ------------------------------------------------------------------ Injector holes in the same format
344
345
347 """A saved reconcile's numbers (AUDIT 9.10 5: 1.6318 -> 1.6633 mm LOX, 1.4715 -> 1.5112 mm fuel; #51/#53
348 gave O/F 1.5501, the 1.65/1.50 mm pair 1.4982 against 1.5)."""
349 from engine.layerx import diff
350
351 result = {
352 "config_sha256": "b" * 64, "condition": "in flight", "lockup_psia": 578.0,
353 "target": {"thrust_N": 6800.0, "of": 1.5, "source": "custom"},
354 "before": {"d_O_mm": 1.6318, "d_F_mm": 1.4715},
355 "changes": [
356 {"item": "LOX orifice diameter", "unit": "mm", "from": 1.6318, "to": 1.6633, "fabrication": "re-drill larger"},
357 {"item": "LOX passage L/d", "unit": "", "from": 5.0, "to": 4.905, "fabrication": "passage stays 8.16 mm long"},
358 {"item": "Fuel orifice diameter", "unit": "mm", "from": 1.4715, "to": 1.5112, "fabrication": "re-drill larger"}],
359 "design_update": {"injector": {"geometry": {"oxidizer": {"d_jet": 1.6633e-3}, "fuel": {"d_jet": 1.5112e-3}}},
360 "discharge": {"oxidizer": {"orifice_l_over_d": 4.905}},
361 "feed_system": {"oxidizer": {"K0": 1.1104, "derived_from": {"by": "Layer X injector reconcile"}}}},
362 "drill_options": {"oxidizer": [{"drill": "#51", "d_mm": 1.7018}], "fuel": [{"drill": "#53", "d_mm": 1.5113}]},
363 "drill_grid": [
364 {"oxidizer": "#51", "fuel": "#53", "d_O_mm": 1.7018, "d_F_mm": 1.5113, "thrust_N": 6902.0, "of": 1.5501},
365 {"oxidizer": "1.65 mm", "fuel": "1.50 mm", "d_O_mm": 1.65, "d_F_mm": 1.50, "thrust_N": 6758.0, "of": 1.4982},
366 # On thrust alone this pair would win (made up for the test: thrust exact, O/F 4 % off).
367 {"oxidizer": "#50", "fuel": "1.45 mm", "d_O_mm": 1.778, "d_F_mm": 1.45, "thrust_N": 6800.0, "of": 1.56},
368 # On O/F alone this one would (made up: O/F exact, thrust 5.9 % short).
369 {"oxidizer": "1.60 mm", "fuel": "1.45 mm", "d_O_mm": 1.60, "d_F_mm": 1.45, "thrust_N": 6400.0, "of": 1.5}],
370 "before_burn": {"mean_thrust_N": 6600.0, "of_mean": 1.53}, "after_burn": {"mean_thrust_N": 6861.0, "of_mean": 1.501},
371 "notes": [],
372 }
373 keys = set(result)
374 cl = diff.from_reconcile(result)
375 assert set(result) == keys # the old keys stay, unchanged
376 by = {(c["target"]): c for c in cl["changes"]}
377 lox = by["design:injector.geometry.oxidizer.d_jet"]
378 assert (lox["before"], lox["after"], lox["cad_impact"], lox["source"]) == (1.6318, 1.6633, "re-drill", "solved")
379 # The pair nearest in thrust and O/F together is 1.65/1.50 (0.62 % + 0.12 %): not the nearest drills
380 # (#51/#53: 1.5 % + 3.3 %), not the exact-thrust pair (0 % + 4 %), not the exact-O/F one (5.9 % + 0 %).
381 assert lox["drill"]["picked"]["pair"] == {"oxidizer": "1.65 mm", "fuel": "1.50 mm"}
382 assert lox["effect"]["mean_thrust_N"] == pytest.approx(261.0)
383 assert by["design:discharge.oxidizer.orifice_l_over_d"]["cad_impact"] == "none"
384 assert by["model:feed_system.oxidizer.K0"]["domain"] == "model"
385 w = cl["exports"]["design_write"]
386 assert w["query"] == {"expect_sha256": "b" * 64} and w["body"] == result["design_update"]
387 for c in cl["changes"]:
388 assert set(c) >= {"component", "pid_node_id", "field", "before", "after", "unit", "provenance", "effect", "cad_impact"}
389
390
392 client, session = api
393 settings = {"drawing_id": "d"}
394 assert client.post("/api/layerx/hardware", json={"settings": settings, "components": []}).status_code == 422
395 assert client.post("/api/layerx/hardware", json={"settings": settings, "components": [{"target": "node:X"}],
396 "objective": "apogee"}).status_code == 422
397 r = client.post("/api/layerx/hardware", json={"settings": settings, "components": [{"target": "node:X"}]})
398 assert r.status_code in (400, 404)
399 with lx._JOBS_LOCK:
400 assert not [j for j in lx._JOBS.values() if j.user == session.user]
401
402
404 """The wiring only: preflight, the module's refusal check, the job's kind and its result. The
405 burns are stubbed (the modules' own tests burn closed-form stands)."""
406 import time
407 from types import SimpleNamespace
408
409 from engine.layerx import optimize as opt
410 from engine.layerx import setpoint as sp
411
412 client, session = api
413 prep = SimpleNamespace(ok=True, checks=[], drawing="D", settings="S", measurements=[])
414 config = SimpleNamespace(design_requirements=SimpleNamespace(target_thrust=7200.0))
415 monkeypatch.setattr(lx, "_prepare", lambda s, settings: (prep, None, config))
416 seen = {}
417
418 def fake_setpoint(cfg, drawing, settings, overrides, request, progress=None, cancelled=None):
419 seen["setpoint"] = request
420 return {"mode": "setpoint", "summary": {"mean_thrust_N": 7200.0}}
421
422 def fake_hardware(cfg, drawing, settings, overrides, request, catalogs=None, progress=None, cancelled=None):
423 seen["hardware"] = (request, sorted(catalogs))
424 return {"mode": "hardware", "summary": {"objective": request.objective}}
425
426 monkeypatch.setattr(sp, "run_setpoint", fake_setpoint)
427 monkeypatch.setattr(opt, "run_hardware", fake_hardware)
428 monkeypatch.setattr(opt, "plan_candidates", lambda *a, **k: {"candidates": []})
429
430 def finish(r):
431 assert r.status_code == 200, r.text
432 for _ in range(100):
433 got = client.get(f"/api/layerx/runs/{r.json()['id']}").json()
434 if got["status"] in ("done", "failed"):
435 return got
436 time.sleep(0.02)
437 raise AssertionError("job did not finish")
438
439 got = finish(client.post("/api/layerx/setpoint", json={"settings": {"drawing_id": "d"}, "meop_psi": {"fuel": 750},
440 "target_thrust_N": 7100}))
441 assert got["kind"] == "setpoint" and got["status"] == "done" and got["result"]["mode"] == "setpoint"
442 assert seen["setpoint"].target_thrust_N == 7100 and seen["setpoint"].meop_psi == {"fuel": 750.0}
443 got = finish(client.post("/api/layerx/hardware", json={"settings": {"drawing_id": "d"},
444 "components": [{"target": "node:SV_LOX_PRESS"}]}))
445 assert got["kind"] == "hardware" and got["status"] == "done"
446 request, kinds = seen["hardware"]
447 assert request.objective == "of_error" and "valves" in kinds and "drills" in kinds
448 assert "legacy" not in got
449 # Neither new kind is legacy in the listing.
450 assert all(not r.get("legacy") for r in client.get("/api/layerx/runs").json())
451 # With no target thrust in the design and none sent, Set point is refused before a slot is taken.
452 monkeypatch.setattr(lx, "_prepare", lambda s, settings: (prep, None, SimpleNamespace(design_requirements=None)))
453 r = client.post("/api/layerx/setpoint", json={"settings": {"drawing_id": "d"}})
454 assert r.status_code == 422 and "target thrust" in r.json()["detail"]["message"]
a class to store JSON values
Definition json.hpp:19401
conditions(self, node, p, signals=None)
static emmc_status_t finish(uint32_t budget_ms)
Definition emmc_stm32.c:100
test_a_trim_orifice_is_sized_from_the_baseline_and_listed_for_pid_designer(le4)
test_drills_come_from_the_reconcile_table_and_rows_need_provenance(tmp_path)
None _saved(session, str run_id, str kind, result=None)
test_hardware_picks_the_drill_that_brings_of_home_and_re_solves_the_set_point(le4)
test_setpoint_and_hardware_routes_start_their_jobs(api, monkeypatch)