1"""Layer-1 design vector for impinging injectors (13 DOF: paired unlike doublets, vs 10 DOF pintle)."""
6from pathlib
import Path
7from unittest.mock
import MagicMock, patch
14 injector_dp_ratios_from_eval_result,
15 injector_dp_ratio_penalty_weighted,
18 TOTAL_WALL_THICKNESS_M,
19 _LAYER1_DEFAULT_DP_F_BAND,
20 _LAYER1_DEFAULT_DP_O_BAND,
21 _LAYER1_DEFAULT_EXIT_PRESSURE_INSIDE_QUAD,
22 _LAYER1_DEFAULT_MIN_STABILITY_MARGIN,
23 _compute_objective_value,
24 create_layer1_apply_x_to_config,
25 run_layer1_optimization,
26 _expected_geom_ao_af_for_unit_momentum_ratio,
27 _geom_ao_af_momentum_hint_squared,
28 _rescale_impinging_jet_diameters_for_mr_target,
29 _impinging_hard_geometry_blocks_eval,
30 _impinging_infeasibility_flow_capacity_terms,
31 _impinging_infeasibility_layout_terms,
32 _impinging_jet_pair_asymmetry_squared,
33 _layer1_exit_pressure_sq_term,
34 _impinging_momentum_band_violation_squared,
35 _impinging_momentum_hinge_squared,
36 _impinging_momentum_log_target_squared,
37 _layer1_cma_warmstart_candidates,
38 _merge_runner_eval_into_performance,
46ROOT = Path(__file__).resolve().parents[1]
50 """Loose bounds compatible with configs/impinging_smoke.yaml geometry.
52 Vector: throat, L*, eps, OD, n_doublets (int), LOX jets (d_jet, θ, spacing),
53 fuel jets (d_jet, θ, spacing), P_O_start, P_F_start.
61 (max_chamber_od * 0.5, max_chamber_od),
76 """Regression: x[11] is P_O, not fuel spacing."""
99 "min_stability_margin": 0.0,
100 "min_stability_score": 0.0,
101 "require_stable_state":
False,
102 "layer1_infeasibility_gate_eps": 1.0,
105 "injector_type":
"impinging",
108 "target_thrust": 8000.0,
110 "P_ambient": 101325.0,
111 "max_lox_P_psi": 900.0,
112 "max_fuel_P_psi": 900.0,
113 "TOTAL_WALL_THICKNESS_M": TOTAL_WALL_THICKNESS_M,
114 "max_nozzle_exit": 0.177,
115 "layer1_W_THRUST": 8000.0,
116 "layer1_W_OF": 200000.0,
117 "layer1_W_OF_low_MR_scale": 1.0,
118 "layer1_W_OF_high_MR_scale": 1.0,
127 "stability_results": {
128 "stability_state":
"stable",
129 "stability_score": 1.0,
130 "chugging": {
"stability_margin": 1.0},
131 "acoustic": {
"stability_margin": 1.0},
132 "feed_system": {
"stability_margin": 1.0},
134 "diagnostics": {
"momentum_ratio_R": 1.0},
137 obj_a = _compute_objective_value(result, x_base, requirements, constants)
138 obj_b = _compute_objective_value(result, x_alt, requirements, constants)
139 self.assertAlmostEqual(obj_a, obj_b, places=9)
142 """Zero hinge when |log R| <= log(1.1) i.e. R in [1/1.1, 1.1]."""
145 for R
in (1.0, tol_lo, tol_hi, 0.95, 1.05):
146 v = _impinging_momentum_hinge_squared(R)
147 self.assertAlmostEqual(v, 0.0, places=12, msg=f
"R={R}")
150 self.assertGreater(_impinging_momentum_hinge_squared(0.5), 0.0)
151 self.assertGreater(_impinging_momentum_hinge_squared(2.0), 0.0)
154 pairs = [(0.5, 2.0), (1.2, 1.0 / 1.2), (0.7, 1.0 / 0.7)]
156 self.assertAlmostEqual(
157 _impinging_momentum_hinge_squared(a),
158 _impinging_momentum_hinge_squared(b),
160 msg=f
"pair ({a}, {b})",
164 self.assertEqual(_impinging_momentum_hinge_squared(
None), 0.0)
165 self.assertEqual(_impinging_momentum_hinge_squared(-0.5), 0.0)
166 self.assertEqual(_impinging_momentum_hinge_squared(0.0), 0.0)
169 """Custom [0.85, 1.15] band: zero inside, quadratic outside."""
171 self.assertAlmostEqual(
172 _impinging_momentum_hinge_squared(1.0, r_band_lo=lo, r_band_hi=hi), 0.0, places=12
174 self.assertAlmostEqual(
175 _impinging_momentum_hinge_squared(0.85, r_band_lo=lo, r_band_hi=hi), 0.0, places=12
177 self.assertAlmostEqual(
178 _impinging_momentum_hinge_squared(1.15, r_band_lo=lo, r_band_hi=hi), 0.0, places=12
180 self.assertGreater(_impinging_momentum_hinge_squared(0.84, r_band_lo=lo, r_band_hi=hi), 0.0)
181 self.assertGreater(_impinging_momentum_hinge_squared(1.16, r_band_lo=lo, r_band_hi=hi), 0.0)
184 """Layer 1 default band [0.9, 1.1]: passing validation must not pay W_MOM."""
186 self.assertAlmostEqual(
187 _impinging_momentum_hinge_squared(0.948, r_band_lo=lo, r_band_hi=hi),
193 self.assertAlmostEqual(
194 _impinging_momentum_band_violation_squared(1.0, r_band_lo=0.8, r_band_hi=1.2),
200 v = _impinging_momentum_band_violation_squared(0.619, r_band_lo=0.8, r_band_hi=1.2)
201 self.assertGreater(v, 0.0)
204 self.assertAlmostEqual(
205 _impinging_momentum_log_target_squared(1.0, rel_deadband=0.008),
211 """Wide validation band used to zero the hinge; log target still penalizes R≠1."""
213 self.assertAlmostEqual(
214 _impinging_momentum_hinge_squared(1.05, r_band_lo=lo, r_band_hi=hi),
219 _impinging_momentum_log_target_squared(1.05, rel_deadband=0.008),
224 v_near = _impinging_momentum_log_target_squared(1.02, rel_deadband=0.008)
225 v_far = _impinging_momentum_log_target_squared(1.10, rel_deadband=0.008)
226 self.assertGreater(v_far, v_near)
232 exp = _expected_geom_ao_af_for_unit_momentum_ratio(mr, rho_o, rho_f)
233 self.assertAlmostEqual(exp, mr / math.sqrt(rho_o / rho_f), places=9)
239 exp = _expected_geom_ao_af_for_unit_momentum_ratio(mr, rho_o, rho_f)
242 sq, ao_af, exp2 = _geom_ao_af_momentum_hint_squared(A_o, A_f, mr, rho_o, rho_f)
243 self.assertAlmostEqual(sq, 0.0, places=12)
244 self.assertAlmostEqual(ao_af, exp, places=12)
245 self.assertAlmostEqual(exp2, exp, places=12)
250 for mr
in (2.8, 3.5):
251 exp_af = _expected_geom_ao_af_for_unit_momentum_ratio(mr, rho_o, rho_f)
252 d_o, d_f = _rescale_impinging_jet_diameters_for_mr_target(
261 a_o = 20 * math.pi * (d_o / 2.0) ** 2
262 a_f = 20 * math.pi * (d_f / 2.0) ** 2
263 self.assertAlmostEqual(a_o / a_f, exp_af, places=3)
269 exp_af = _expected_geom_ao_af_for_unit_momentum_ratio(mr, rho_o, rho_f)
271 d_o = math.sqrt(exp_af) * d_f
272 out_o, out_f = _rescale_impinging_jet_diameters_for_mr_target(
281 self.assertAlmostEqual(out_o, d_o, places=6)
282 self.assertAlmostEqual(out_f, d_f, places=6)
286 self.assertEqual(len(b), 13)
291 out = _snap_integer_dims(x, [4])
292 self.assertEqual(out[4], 12.0)
295 cfg_path = ROOT /
"configs" /
"impinging_smoke.yaml"
296 base = load_config(str(cfg_path))
297 self.assertEqual(base.injector.type,
"impinging")
299 cg = base.chamber_geometry
300 g = base.injector.geometry
301 D_outer = float(cg.chamber_diameter) + TOTAL_WALL_THICKNESS_M
303 n_d =
min(int(g.oxidizer.n_elements), int(g.fuel.n_elements))
308 float(cg.expansion_ratio),
311 float(g.oxidizer.d_jet),
312 float(g.oxidizer.impingement_angle),
313 float(g.oxidizer.spacing),
315 float(g.fuel.impingement_angle),
316 float(g.fuel.spacing),
317 float(base.lox_tank.initial_pressure_psi),
318 float(base.fuel_tank.initial_pressure_psi),
324 apply_fn = create_layer1_apply_x_to_config(
326 max_chamber_od=bounds[3][1],
327 max_nozzle_exit=0.15,
328 injector_type=
"impinging",
330 out_cfg, p_o, p_f = apply_fn(x, base)
331 self.assertEqual(out_cfg.injector.type,
"impinging")
332 nd_snap = int(round(x[4]))
333 self.assertEqual(out_cfg.injector.geometry.oxidizer.n_elements, nd_snap)
334 self.assertEqual(out_cfg.injector.geometry.fuel.n_elements, nd_snap)
335 self.assertAlmostEqual(p_o, float(np.clip(x[11], bounds[11][0], bounds[11][1])))
336 self.assertAlmostEqual(p_f, float(np.clip(x[12], bounds[12][0], bounds[12][1])))
339 cfg = load_config(str(ROOT /
"configs" /
"impinging_smoke.yaml"))
340 Po = cfg.lox_tank.initial_pressure_psi * 6894.76
341 Pf = cfg.fuel_tank.initial_pressure_psi * 6894.76
343 _mdot_o, _mdot_f, diag = flows(Po, Pf, Pc, cfg)
344 self.assertIn(
"u_O", diag)
345 self.assertIn(
"u_F", diag)
346 self.assertTrue(np.isfinite(diag[
"u_O"]))
347 self.assertTrue(np.isfinite(diag[
"u_F"]))
348 self.assertIn(
"turbulence_intensity_mix", diag)
349 self.assertTrue(np.isfinite(diag[
"turbulence_intensity_mix"]))
350 self.assertIn(
"momentum_ratio_R", diag)
351 self.assertTrue(np.isfinite(diag[
"momentum_ratio_R"]))
352 self.assertGreater(diag[
"momentum_ratio_R"], 0.0)
353 self.assertIn(
"v_O_bulk", diag)
354 self.assertIn(
"v_F_bulk", diag)
355 self.assertIn(
"rho_O_momentum", diag)
356 self.assertIn(
"rho_F_momentum", diag)
359 rho_O, rho_F = 1000.0, 500.0
361 v_F = v_O * np.sqrt(rho_O / rho_F)
362 R1 = momentum_ratio_R_from_bulk_velocities(rho_O, rho_F, v_O, v_F)
363 self.assertAlmostEqual(R1, 1.0, places=12)
365 R2 = momentum_ratio_R_from_bulk_velocities(1000.0, 1000.0, 20.0, 10.0)
366 self.assertAlmostEqual(R2, 2.0, places=12)
368 R3 = momentum_ratio_R_from_bulk_velocities(1000.0, 1000.0, 10.0, 20.0)
369 self.assertAlmostEqual(R3, 0.5, places=12)
372 cfg = load_config(str(ROOT /
"configs" /
"impinging_smoke.yaml"))
373 Po = cfg.lox_tank.initial_pressure_psi * 6894.76
374 Pf = cfg.fuel_tank.initial_pressure_psi * 6894.76
376 _mdot_o, _mdot_f, diag = flows(Po, Pf, Pc, cfg)
377 R_calc = momentum_ratio_R_from_bulk_velocities(
378 float(diag[
"rho_O_momentum"]),
379 float(diag[
"rho_F_momentum"]),
380 float(diag[
"v_O_bulk"]),
381 float(diag[
"v_F_bulk"]),
383 np.testing.assert_allclose(diag[
"momentum_ratio_R"], R_calc, rtol=1e-9, atol=0.0)
386 cfg = load_config(str(ROOT /
"configs" /
"impinging_smoke.yaml"))
388 Po = cfg.lox_tank.initial_pressure_psi * 6894.76
389 Pf = cfg.fuel_tank.initial_pressure_psi * 6894.76
390 res = r.evaluate(Po, Pf, P_ambient=101325.0, silent=
True)
391 self.assertGreater(float(res.get(
"Pc")
or 0.0), 0.0)
392 ip = res.get(
"injector_pressure")
or {}
393 self.assertIsNotNone(ip.get(
"P_injector_O"))
394 self.assertIsNotNone(ip.get(
"P_injector_F"))
395 dfull = res.get(
"diagnostics")
or {}
396 self.assertIn(
"momentum_ratio_R", dfull)
397 self.assertTrue(np.isfinite(dfull[
"momentum_ratio_R"]))
400 real_cfg = load_config(str(ROOT /
"configs" /
"impinging_smoke.yaml"))
401 mock_runner = MagicMock()
422 return copy.deepcopy(obj)
424 mock_engine = MagicMock()
425 mock_engine.evaluate.return_value = {
433 "stability_results": {},
437 "engine.optimizer.layers.layer1_static_optimization.copy.deepcopy",
440 "engine.optimizer.layers.layer1_static_optimization.PintleEngineRunner",
441 return_value=mock_engine,
443 "engine.optimizer.layers.layer1_static_optimization.cma.CMAEvolutionStrategy"
444 )
as mock_cma, patch(
445 "engine.optimizer.layers.layer1_static_optimization.minimize"
446 )
as mock_minimize, patch(
447 "engine.optimizer.layers.layer1_static_optimization.ProcessPoolExecutor"
449 mock_es = mock_cma.return_value
450 mock_es.stop.return_value =
True
451 mock_es.result.xbest = x13
452 mock_es.result.fbest = 0.1
454 mock_lbfgs_res = MagicMock()
455 mock_lbfgs_res.x = np.array(x13)
456 mock_lbfgs_res.fun = 0.05
457 mock_lbfgs_res.success =
True
458 mock_minimize.return_value = mock_lbfgs_res
461 "engine.optimizer.layers.layer1_static_optimization.create_layer1_apply_x_to_config"
462 )
as mock_create_apply:
463 mock_apply = MagicMock()
464 mock_apply.return_value = (real_cfg, 560.0, 660.0)
465 mock_create_apply.return_value = mock_apply
467 mock_update_progress = MagicMock()
468 mock_log_status = MagicMock()
469 requirements = real_cfg.design_requirements.model_dump()
471 optimized_config, _results = run_layer1_optimization(
474 requirements=requirements,
475 target_burn_time=10.0,
476 tolerances={
"thrust": 0.1,
"apogee": 0.15},
477 pressure_config={
"mode":
"optimizer_controlled"},
478 update_progress=mock_update_progress,
479 log_status=mock_log_status,
480 layer1_max_iterations=1,
481 layer1_cma_restarts=1,
484 self.assertIsNotNone(optimized_config.lox_tank)
485 self.assertIsNotNone(optimized_config.fuel_tank)
486 self.assertTrue(np.isfinite(optimized_config.lox_tank.initial_pressure_psi))
487 self.assertTrue(np.isfinite(optimized_config.fuel_tank.initial_pressure_psi))
488 self.assertLessEqual(optimized_config.lox_tank.initial_pressure_psi, 700.0)
489 self.assertLessEqual(optimized_config.fuel_tank.initial_pressure_psi, 850.0)
492 """Stored-validation path must merge diagnostics/injector_pressure like runner.evaluate."""
494 initial_performance: dict = {
"Pc": pc}
497 "momentum_ratio_R": 1.05,
498 "delta_p_injector_O": 0.26 * pc,
499 "delta_p_injector_F": 0.28 * pc,
501 "injector_pressure": {
"delta_p_injector_O": 0.26 * pc},
503 _merge_runner_eval_into_performance(initial_performance, eval_results)
504 ro, rf = injector_dp_ratios_from_eval_result(pc, initial_performance)
505 self.assertIsNotNone(ro)
506 self.assertIsNotNone(rf)
507 self.assertAlmostEqual(ro, 0.26, places=6)
508 self.assertAlmostEqual(rf, 0.28, places=6)
509 pen = injector_dp_ratio_penalty_weighted(ro, rf, 50.0, 50.0)
510 self.assertTrue(np.isfinite(pen))
511 mr = initial_performance[
"diagnostics"][
"momentum_ratio_R"]
512 self.assertAlmostEqual(mr, 1.05, places=6)
514 mbp = float(w_mom * _impinging_momentum_hinge_squared(mr))
515 self.assertTrue(np.isfinite(mbp))
518 """MR is mass-based O/F: must equal mdot_O/mdot_F everywhere (runner + chamber solver)."""
519 cfg = load_config(ROOT /
"configs" /
"impinging_smoke.yaml")
520 pc = cfg.pressure_curves
522 out = runner.evaluate(
523 float(pc.initial_lox_pressure_pa),
524 float(pc.initial_fuel_pressure_pa),
527 mo = float(out[
"mdot_O"])
528 mf = float(out[
"mdot_F"])
529 mr = float(out[
"MR"])
530 self.assertGreater(mf, 0.0)
531 self.assertAlmostEqual(mr, mo / mf, places=12)
534 """Warm-start with n_trials=0 must not call the executor (no fake objectives)."""
538 x, f = _layer1_cma_warmstart_candidates(
539 np.array([0.2, 0.5, 0.8]),
544 rng=np.random.default_rng(0),
549 make_cache_key_fn=
lambda a: tuple(np.round(a, 6)),
550 opt_state={
"function_evaluations": 0},
551 logger=logging.getLogger(
"test"),
553 ex.map.assert_not_called()
554 self.assertTrue(np.all(x >= lo)
and np.all(x <= hi))
555 self.assertTrue(np.isinf(f))
558 """Parallel workers score candidates with ``_compute_objective_value``; parent uses ``objective()``."""
559 self.assertEqual(_LAYER1_DEFAULT_MIN_STABILITY_MARGIN, 1.2)
560 self.assertEqual(_LAYER1_DEFAULT_DP_O_BAND, (0.20, 0.40))
561 self.assertEqual(_LAYER1_DEFAULT_DP_F_BAND, (0.20, 0.40))
562 self.assertEqual(_LAYER1_DEFAULT_EXIT_PRESSURE_INSIDE_QUAD, 0.35)
565 """Inside-rel quad steers toward P_target without changing the hinge outside the deadband."""
567 flat = _layer1_exit_pressure_sq_term(103_000.0, target, inside_rel_quad_scale=0.0)
568 self.assertAlmostEqual(flat, 0.0, places=15)
570 q = _layer1_exit_pressure_sq_term((1.0 + rel) * target, target, inside_rel_quad_scale=2.0)
571 self.assertAlmostEqual(q, 2.0 * rel * rel)
572 outs = _layer1_exit_pressure_sq_term(113_000.0, target, inside_rel_quad_scale=0.0)
573 self.assertAlmostEqual(outs, 0.08**2)
576 self.assertAlmostEqual(
577 _impinging_jet_pair_asymmetry_squared(45.0, 45.0, max_delta_deg=26.0),
580 dd = abs(62.43 - 16.79)
582 span =
max(26.0, 5.0)
583 self.assertAlmostEqual(
584 _impinging_jet_pair_asymmetry_squared(62.43, 16.79, max_delta_deg=26.0),
585 (excess / span) ** 2,
588 self.assertAlmostEqual(_impinging_jet_pair_asymmetry_squared(17.0, 42.0, max_delta_deg=28.0), 0.0)
591 """Unlike doublet layout: jet overlap + n·spacing/π pitch circle vs bore."""
592 score_ok = _impinging_infeasibility_layout_terms(
599 D_chamber_inner=0.15,
601 self.assertAlmostEqual(score_ok, 0.0, places=12)
603 score_overlap = _impinging_infeasibility_layout_terms(
610 D_chamber_inner=0.15,
612 self.assertGreater(score_overlap, 0.0)
615 score_pitch = _impinging_infeasibility_layout_terms(
622 D_chamber_inner=0.10,
624 self.assertGreater(score_pitch, 0.0)
628 _impinging_hard_geometry_blocks_eval(
633 D_chamber_inner=0.15,
635 A_chamber_check=0.017,
636 A_throat_check=1.0e-4,
640 _impinging_hard_geometry_blocks_eval(
645 D_chamber_inner=0.15,
647 A_chamber_check=0.017,
648 A_throat_check=1.0e-4,
654 self.assertAlmostEqual(
655 _impinging_infeasibility_flow_capacity_terms(
658 A_throat_check=A_throat,
663 over = _impinging_infeasibility_flow_capacity_terms(
666 A_throat_check=A_throat,
669 self.assertAlmostEqual(over, (1.2e-4 / A_throat - 0.95) ** 2, places=12)
672if __name__ ==
"__main__":
test_snap_integer_dims_index_4_only_n_doublets(self)
test_momentum_hinge_default_band_matches_validation_at_0948(self)
test_impinging_eval_mr_matches_mass_flow_ratio(self)
test_momentum_log_target_zero_at_unity(self)
test_impinging_flow_capacity_infeasibility(self)
test_cma_warmstart_zero_trials_skips_eval(self)
test_flows_includes_injection_velocities(self)
test_momentum_band_violation_squared_positive_outside(self)
test_momentum_hinge_skips_invalid_R(self)
test_impinging_layout_infeasibility_overlap_and_pitch(self)
test_momentum_log_target_active_inside_wide_band(self)
test_momentum_hinge_positive_outside_band(self)
test_momentum_hinge_symmetry_R_and_reciprocal(self)
test_impinging_jet_pair_asymmetry_squared(self)
test_flows_momentum_ratio_matches_bulk_formula(self)
test_momentum_ratio_R_matches_analytic(self)
test_geom_ao_af_hint_squared_zero_when_matching_expected(self)
test_design_vector_length_is_13_for_paired_impinging_doublets(self)
test_expected_geom_ao_af_matches_mr_over_sqrt_density_ratio(self)
test_apply_x_impinging_preserves_type_and_geometry(self)
test_momentum_hinge_zero_inside_log_deadband(self)
test_momentum_log_target_grows_away_from_one(self)
test_momentum_band_violation_squared_zero_inside(self)
test_layer1_impinging_mocked_one_iteration(self)
test_momentum_hinge_ratio_band_deadband(self)
test_rescale_impinging_jets_noop_when_already_matched(self)
test_exit_pressure_inside_band_quad_scale(self)
test_worker_objective_impinging_indices_do_not_alias_fuel_spacing_with_pressure(self)
test_min_stability_margin_default_worker_parent_aligned(self)
test_rescale_impinging_jets_matches_target_ao_af_for_mr(self)
test_merge_runner_eval_enables_dp_and_momentum_metrics(self)
test_runner_evaluate_impinging_smoke(self)
test_impinging_hard_geometry_blocks_eval(self)
_impinging_bounds_placeholder(float max_chamber_od=0.18)