433def solve(evaluate: Evaluate, *, start_lockup: float, start_fill: float, max_fill: float, target: float,
434 request: SetpointRequest, bounds: Tuple[float, float], first_guess: Optional[float] =
None,
435 say: Progress =
lambda s, f:
None, cancelled: Callable[[], bool] =
lambda:
False,
436 seed: Optional[List[Dict[str, Any]]] =
None) -> Dict[str, Any]:
437 """The root, by whole burns through ``evaluate``. Pure in everything but ``evaluate``, so it is
438 tested against a closed-form stand.
440 ``evaluate([(lockup, fill, replay), ...])`` burns the points (in parallel when it can) and returns
441 one :func:`burn_point`-shaped dict each. ``first_guess`` is the second lockup of the opening pair
442 (burned alongside the starting point); ``None``: proportional to the thrust missing. ``seed``: burns
443 already made at the starting point (Hardware mode's candidate), used instead of burning it again;
444 they do not count against ``max_burns``."""
449 replay = bool(request.replay)
450 tol_F = request.thrust_tol_rel * target
451 seeded = [{**p,
"stage":
"seed",
"asked": {
"lockup_psia": p.get(
"lockup_psia"),
"fill_psig": p.get(
"fill_psig")}}
452 for p
in (seed
or [])]
453 st.history.extend(seeded)
454 budget =
max(1, int(request.max_burns)) + len(seeded)
456 def burn(points: List[Tuple[float, float]], stage: str, *, rp: Optional[bool] =
None) -> List[Dict[str, Any]]:
459 left = budget - len(st.history)
462 points = points[:left]
463 out = evaluate([(L, F, replay
if rp
is None else rp)
for L, F
in points])
464 for (L, F), p
in zip(points, out):
465 p = {**p,
"stage": stage,
"asked": {
"lockup_psia": L,
"fill_psig": F}}
467 return st.history[-len(points):]
469 def thrust_err(p: Dict[str, Any]) -> float:
470 return float(p[
"figures"][
"mean_thrust_N"]) - target
474 half_band = 0.5 * float(request.margin_tol_psi)
475 spare_aim = float(request.margin_psi) + half_band
477 def spare_err(p: Dict[str, Any]) -> Optional[float]:
478 s = p[
"figures"].get(
"copv_spare_psi")
479 return None if s
is None else float(s) - spare_aim
481 def clamp(L: float, ref: float) -> float:
483 cap = MAX_LOCKUP_STEP * ref
485 L = ref + math.copysign(cap, step)
486 return min(
max(L, lo), hi)
489 L0, F0 =
min(
max(start_lockup, lo), hi), start_fill
490 if first_guess
is None or not math.isfinite(first_guess):
491 first_guess = L0 * 1.03
492 L1 = clamp(first_guess, L0)
493 if abs(L1 - L0) < 0.5:
494 L1 = clamp(L0 * 1.02, L0)
495 say(
"Burning the current settings and a first guess", 0.05)
496 pair = seeded + burn([(L1, F0)],
"bracket")
if seeded
else burn([(L0, F0), (L1, F0)],
"bracket")
497 before = pair[0]
if pair
else None
498 good = [p
for p
in pair
if usable(p)]
500 errs =
"; ".join(p.get(
"error")
or "; ".join(p.get(
"preflight")
or [])
or "tripped" for p
in pair)
501 raise ValueError(f
"neither opening burn could be used: {errs}")
503 def at_fill(F: float) -> List[Dict[str, Any]]:
504 return [p
for p
in st.history
if usable(p)
and abs(float(p[
"fill_psig"]) - F) < 0.5
and p[
"replay"] == replay]
506 def best_of(ps: List[Dict[str, Any]]) -> Dict[str, Any]:
507 return min(ps, key=
lambda p: abs(thrust_err(p)))
509 current = best_of(good)
511 while abs(thrust_err(current)) > tol_F
and len(st.history) < budget:
513 slope =
_fit_slope([(float(p[
"lockup_psia"]), float(p[
"figures"][
"mean_thrust_N"]))
for p
in same])
514 if slope
is None or slope <= 0:
515 if slope
is not None:
516 st.notes.append(f
"Mean thrust does not rise with lockup between the burns (slope {slope:.2f} N/psi); "
517 "the solve stopped.")
519 slope = float(current[
"figures"][
"mean_thrust_N"]) / float(current[
"lockup_psia"])
520 anchor = best_of(same)
521 L_a = float(anchor[
"lockup_psia"])
522 L_new = clamp(L_a - thrust_err(anchor) / slope, L_a)
525 failed = [float(p[
"asked"][
"lockup_psia"])
for p
in st.history
526 if not usable(p)
and abs(float(p[
"asked"][
"fill_psig"]) - F0) < 0.5]
527 above = [L
for L
in failed
if L > L_a]
528 below = [L
for L
in failed
if L < L_a]
529 if above
and L_new >=
min(above):
530 L_new = 0.5 * (L_a +
min(above))
531 if below
and L_new <=
max(below):
532 L_new = 0.5 * (L_a +
max(below))
533 if any(abs(float(p[
"asked"][
"lockup_psia"]) - L_new) < 1e-6
and abs(float(p[
"asked"][
"fill_psig"]) - F0) < 0.5
534 for p
in st.history):
535 st.notes.append(
"The next lockup is one already burned (a bound, or the slope's resolution): stopped.")
537 say(f
"Lockup {L_new:.1f} psia (burn {len(st.history) + 1})", 0.1 + 0.5 * len(st.history) / budget)
538 got = burn([(L_new, F0)],
"lockup")
542 current = best_of(at_fill(F0))
545 st.notes.append(f
"The burn at {L_new:.1f} psia could not be used ({got[0].get('error') or 'tripped'}); "
546 "stepping back toward the last good lockup.")
549 lockup_done = abs(thrust_err(current)) <= tol_F
553 if request.solve_fill
and lockup_done
and spare_err(current)
is not None:
554 x = [float(current[
"lockup_psia"]), float(current[
"fill_psig"])]
555 r = [thrust_err(current), float(spare_err(current))]
557 a =
_fit_slope([(float(p[
"lockup_psia"]), float(p[
"figures"][
"mean_thrust_N"]))
for p
in same])
558 c =
_fit_slope([(float(p[
"lockup_psia"]), float(p[
"figures"][
"copv_spare_psi"]))
for p
in same
559 if p[
"figures"].get(
"copv_spare_psi")
is not None])
560 J = [[a
if a
and a > 0
else float(current[
"figures"][
"mean_thrust_N"]) / x[0], D_THRUST_D_FILL_GUESS],
561 [c
if c
is not None else -1.0, D_SPARE_D_FILL_GUESS]]
562 fill_status =
"solving"
564 while not (abs(r[0]) <= tol_F
and abs(r[1]) <= half_band)
and len(st.history) < budget:
565 det = J[0][0] * J[1][1] - J[0][1] * J[1][0]
566 if not math.isfinite(det)
or abs(det) < 1e-12:
567 st.notes.append(
"The fill and lockup no longer move thrust and spare independently: stopped.")
569 dx = [-damp * (J[1][1] * r[0] - J[0][1] * r[1]) / det, -damp * (-J[1][0] * r[0] + J[0][0] * r[1]) / det]
571 f_floor = x[0] + request.margin_psi
573 if x[1] >= max_fill - 0.5
and r[1] + half_band < 0:
574 fill_status =
"bottle_short"
575 st.notes.append(f
"A full bottle ({max_fill:.0f} psig, as drawn) leaves {r[1] + spare_aim:.0f} psi "
576 f
"at burnout, under the {request.margin_psi:.0f} psi margin: the drawn bottle cannot "
577 "keep the margin at this thrust.")
581 F_new = 0.5 * (x[1] + f_floor)
582 L_new = clamp(x[0] + dx[0], x[0])
583 say(f
"Lockup {L_new:.1f} psia, fill {F_new:.0f} psig (burn {len(st.history) + 1})",
584 0.1 + 0.8 * len(st.history) / budget)
585 got = burn([(L_new, F_new)],
"lockup+fill")
589 if not usable(p)
or spare_err(p)
is None:
591 st.notes.append(f
"The burn at {L_new:.1f} psia / {F_new:.0f} psig could not be used "
592 f
"({p.get('error') or 'tripped'}); halving the step.")
598 x_new = [float(p[
"lockup_psia"]), float(p[
"fill_psig"])]
599 r_new = [thrust_err(p), float(spare_err(p))]
600 s = [x_new[0] - x[0], x_new[1] - x[1]]
601 ss = s[0] ** 2 + s[1] ** 2
603 y = [r_new[0] - r[0], r_new[1] - r[1]]
604 Js = [J[0][0] * s[0] + J[0][1] * s[1], J[1][0] * s[0] + J[1][1] * s[1]]
607 J[i][j] += (y[i] - Js[i]) * s[j] / ss
608 x, r, current = x_new, r_new, p
609 if fill_status ==
"solving":
610 fill_status =
"solved" if abs(r[1]) <= half_band
and abs(r[0]) <= tol_F
else "unsolved"
614 if not replay
and request.verify
and usable(current):
615 say(
"Verifying the answer with the erosion replay", 0.92)
616 L, F = float(current[
"lockup_psia"]), float(current[
"fill_psig"])
617 got = burn([(L, F)],
"verify", rp=
True)
618 if got
and usable(got[0]):
620 offset = thrust_err(verified)
621 slope =
_fit_slope([(float(p[
"lockup_psia"]), float(p[
"figures"][
"mean_thrust_N"]))
622 for p
in at_fill(float(current[
"fill_psig"]))])
or \
623 float(current[
"figures"][
"mean_thrust_N"]) / L
624 if abs(offset) > tol_F
and slope > 0:
626 got = burn([(clamp(L - offset / slope, L), F)],
"verify (corrected)", rp=
True)
627 if got
and usable(got[0]):
632 converged = abs(thrust_err(final)) <= tol_F
and (
not request.solve_fill
or fill_status
in (
"solved",
"bottle_short"))
633 if len(st.history) >= budget
and not converged:
634 st.notes.append(f
"Stopped at the {budget}-burn budget before both targets were met; the last burn is shown.")
635 return {
"final": final,
"before": before,
"history": st.history,
"notes": st.notes,
"converged": converged,
636 "lockup_converged": lockup_done
or abs(thrust_err(final)) <= tol_F,
"fill_status": fill_status}
705def _changes(before: Dict[str, Any], final: Dict[str, Any], start: Dict[str, Any], history_n: int) -> List[Dict[str, Any]]:
708 ids = (final.get(
"derived")
or {}).get(
"stand_ids")
or {}
709 prov = f
"solved: Layer X set point, {history_n} whole burns; the last one is this setting"
711 for target, component, field_, unit, b, a, node
in (
712 (
"op:dome_psig",
"Dome loader (the dial)",
"dome_psig",
"psig", before.get(
"dome_psig"), final.get(
"dome_psig"),
714 (
"op:lockup_psia",
"Tank lockup (dome-loaded regulator)",
"lockup_psia",
"psia",
715 before.get(
"lockup_psia"), final.get(
"lockup_psia"), ids.get(
"regulator")),
716 (
"op:copv_psig",
"Bottle fill",
"copv_psig",
"psig", before.get(
"fill_psig"), final.get(
"fill_psig"),
719 if a
is None or (isinstance(b, (int, float))
and abs(float(a) - float(b)) < 0.05):
721 rows.append(diff.change(component=component, field=field_, before=
None if b
is None else round(float(b), 2),
722 after=round(float(a), 2), unit=unit, provenance=prov, cad_impact=
"setting only",
723 target=target, pid_node_id=node, source=
"solved",
724 before_provenance=start.get(
"source")
if field_ ==
"lockup_psia" else
725 start.get(
"fill_source")
if field_ ==
"copv_psig" else "solved for the starting lockup"))
736def run_setpoint(config: Any, drawing: Any, settings: Any, overrides: List[Any], request: SetpointRequest, *,
737 progress: Optional[Progress] =
None, cancelled: Callable[[], bool] =
lambda:
False,
738 workers: Optional[int] =
None, evaluate: Optional[Evaluate] =
None,
739 stand: Optional[Dict[str, float]] =
None) -> Dict[str, Any]:
740 """The set point for ``request`` on this drawing and design.
742 ``evaluate`` replaces the burns, and ``stand`` the preflight's readings of the drawing
743 (``lockup_psia``, ``fill_psig``, ``max_fill_psig``, ``reference_thrust_N``, ``mawp_cap_psia``):
744 both for tests, which hand it a closed-form stand."""
748 say = progress
or (
lambda stage, fraction:
None)
749 started = time.perf_counter()
752 grading = {
"margin_psi": request.margin_psi,
"meop_psi": dict(request.meop_psi
or {}),
"design_of": design_of}
754 say(
"Preparing", 0.01)
755 lockup0 = getattr(settings,
"tank_pressure_psia",
None)
756 cfg_lockup = float(getattr(getattr(config,
"lox_tank",
None),
"initial_pressure_psi", 0.0)
or 0.0)
758 fill_psia0 = getattr(settings,
"copv_pressure_psia",
None)
759 fill_psig0 = fill_psia0 - 101325.0 / PSI
if fill_psia0
is not None else getattr(settings,
"copv_pressure_psig",
None)
760 start = {
"lockup_psia": lockup0
or cfg_lockup,
761 "source":
"the rail's tank pressure" if lockup0
else "config lox_tank.initial_pressure_psi",
762 "fill_psig": fill_psig0,
763 "fill_source":
"the rail's bottle pressure" if fill_psig0
is not None else "the drawing's bottle pressure"}
770 nominal = prepare(config,
None, drawing, replace(settings, card_center_psia=
None), overrides)
772 raise ValueError(
"Preflight failed: " +
"; ".join(c.label
for c
in nominal.checks
if c.status ==
"fail"))
774 start[
"lockup_psia"] = float(d[
"target_lockup_psia"])
776 start[
"source"] = {
"set on the rail":
"the lockup the rail's dome dial gives",
777 "solved for the rail's tank pressure":
"the rail's tank pressure",
778 "as drawn":
"the lockup the drawing's dome dial gives",
779 "solved for the design's tank pressure":
"config lox_tank.initial_pressure_psi",
780 }.get(str(d.get(
"dome_source")), start[
"source"])
781 start[
"fill_psig"] = float(d[
"copv_psig"])
782 max_fill = float(d.get(
"copv_drawn_psig")
or d[
"copv_psig"])
783 reference_F = float((nominal.link.reference
or {}).get(
"F")
or 0.0)
if nominal.link
else None
786 start[
"lockup_psia"] = float(stand.get(
"lockup_psia")
or start[
"lockup_psia"])
787 if stand.get(
"fill_psig")
is not None:
788 start[
"fill_psig"] = float(stand[
"fill_psig"])
789 max_fill = stand.get(
"max_fill_psig")
790 reference_F = stand.get(
"reference_thrust_N")
791 mawp_cap = float(stand.get(
"mawp_cap_psia")
or math.inf)
792 if start[
"fill_psig"]
is None:
793 raise ValueError(
"No bottle pressure: the rail and the drawing state none.")
794 max_fill =
max(max_fill
or float(start[
"fill_psig"]), float(start[
"fill_psig"]))
795 L0 = float(start[
"lockup_psia"])
797 raise ValueError(
"No starting tank pressure: the rail and the config state none.")
798 lo, hi = request.lockup_bounds_psia
or (0.5 * L0, 1.5 * L0)
799 hi =
min(hi, mawp_cap)
802 guess = L0 * target / reference_F
if reference_F
else None
806 n_workers = default_workers(2, workers)
807 pool = WorkerPool(n_workers)
808 base_args = {
"config": config,
"drawing": drawing,
"settings": settings,
"overrides": list(overrides),
812 def evaluate(points: List[Tuple[float, Optional[float], bool]]) -> List[Dict[str, Any]]:
814 for L, F, rp
in points:
815 args.append({**base_args,
"lockup_psia": L,
"fill_psig": F,
"replay": rp,
"index": counter[0]})
817 return pool.map(burn_point, args, cancelled=cancelled)
821 out =
solve(evaluate, start_lockup=L0, start_fill=float(start[
"fill_psig"]), max_fill=max_fill, target=target,
822 request=request, bounds=(lo, hi), first_guess=guess, say=say, cancelled=cancelled)
825 pool.close(cancelled())
826 final, before, history = out[
"final"], out[
"before"], out[
"history"]
827 notes = list(out[
"notes"])
830 raise ValueError(
"no usable burn: " +
"; ".join(p.get(
"error")
or "" for p
in history))
831 fig = final[
"figures"]
832 of = fig.get(
"of_mean")
834 groups: Dict[float, List[Tuple[float, float]]] = {}
836 if usable(p)
and p[
"figures"].get(
"of_mean")
is not None:
837 groups.setdefault(round(float(p[
"fill_psig"]), 0), []).append(
838 (float(p[
"lockup_psia"]), float(p[
"figures"][
"of_mean"])))
839 lockup_pts =
max(groups.values(), key=len)
if groups
else []
841 "of_mean": of,
"design_of": design_of,
"design_of_source": of_source,
842 "offset_rel": (of / design_of - 1.0)
if of
and design_of
else None,
843 "per_100psi_lockup": (
_fit_slope(lockup_pts)
or 0.0) * 100.0
if len(lockup_pts) >= 2
else None,
844 "settable_here":
False,
845 "note": (
"One dome regulator presses both tanks through one manifold, so lockup and fill move O/F only "
846 "through second-order effects. O/F is set by hardware: the holes (Injector holes), the lines, or a "
847 "trim orifice (Hardware mode)."),
852 bad = [r[
"label"]
for r
in final.get(
"limits")
or []
if r.get(
"grade") ==
"bad"]
853 notes.append(
"The set point breaks " +
", ".join(bad) +
": the target cannot be met within the limits "
854 "on this drawing (see `binding` for where each limit is reached, estimated).")
855 tripped_above = [float(p[
"asked"][
"lockup_psia"])
for p
in history
856 if p.get(
"tripped")
and float(p[
"asked"][
"lockup_psia"]) > float(final[
"lockup_psia"])]
857 if tripped_above
and not out[
"converged"]:
858 notes.append(f
"The target lies beyond a lockup that trips the stand ({min(tripped_above):.1f} psia trips): "
859 f
"the highest usable lockup found, {float(final['lockup_psia']):.1f} psia, is shown.")
860 if out[
"fill_status"] ==
"held" and request.solve_fill:
861 notes.append(
"The fill was not solved: the lockup did not converge first.")
862 if not request.replay:
863 notes.append(
"The search burned without the erosion replay" + (
864 "; the answer was verified with it." if request.verify
else "; its numbers exclude the eroding throat."))
865 if fig.get(
"card_outside_steps"):
866 notes.append(
"The answer's burn asked the engine card about points outside its fitted box.")
867 dome = final.get(
"dome_psig")
868 dome_rate = final.get(
"dome_per_1000psi_fill")
869 derived = final.get(
"derived")
or {}
872 "lockup_psia": final.get(
"lockup_psia"),
873 "copv_fill_psig": final.get(
"fill_psig"),
874 "dome_per_1000psi_fill": dome_rate,
875 "dome_at_full_bottle_psig": (dome + dome_rate * (max_fill - float(final[
"fill_psig"])) / 1000.0
876 if dome
is not None and dome_rate
is not None else None),
877 "full_bottle_psig": max_fill,
878 "fuel_lead_s": getattr(settings,
"fuel_lead_s",
None),
879 "fuel_lead": (
"the rail's, burned as given (not solved)" if getattr(settings,
"fuel_lead_s",
None)
is not None
880 else "not modelled: both mains open together at Fire"),
881 "pressurant": derived.get(
"pressurant_gas"),
882 "drawing": derived.get(
"drawing"),
883 "config_sha256": derived.get(
"config_sha256"),
884 "ids": derived.get(
"stand_ids"),
886 changes =
_changes(before
or {}, final, start, len(history))
887 before_fig = (before
or {}).get(
"figures")
if usable(before)
else None
888 if before_fig
is not None and before
is not None:
889 before_fig = {**before_fig,
"dome_psig": before.get(
"dome_psig"),
"lockup_psia": before.get(
"lockup_psia")}
890 same_basis = before
is not None and bool(before.get(
"replay")) == bool(final.get(
"replay"))
891 change_list = diff.build(
"setpoint", changes, before=before_fig
if same_basis
else None, after=fig,
892 limits=final.get(
"limits"),
893 basis={
"config_sha256": derived.get(
"config_sha256"),
"drawing": derived.get(
"drawing"),
894 "target_mean_thrust_N": target,
"replay": final.get(
"replay")},
895 notes=[]
if same_basis
else [
"The current settings burned on another basis: no effect "
897 change_list[
"exports"] = {
"settings_patch": diff.settings_patch(change_list),
"design_write":
None,
898 "pid_designer":
None}
902 "target": {
"mean_thrust_N": target,
"source": target_source,
"tol_rel": request.thrust_tol_rel,
903 "margin_psi": request.margin_psi
if request.solve_fill
else None},
904 "converged": out[
"converged"],
905 "fill_status": out[
"fill_status"],
906 "feasible": is_feasible,
907 "settings_card": card,
909 "before": before_fig,
911 "limits": final.get(
"limits"),
912 "limits_basis": final.get(
"limits_basis"),
914 "history": [{k: v
for k, v
in p.items()
if k
not in (
"derived",)}
for p
in history],
915 "change_list": change_list,
916 "model":
_model_block(target, target_source, request, start, max_fill, design_of, of_source),
917 "unmeasured": UNMEASURED,
919 "burns": len(history),
920 "workers": n_workers,
921 "wall_s": time.perf_counter() - started,
922 "summary": {
"mean_thrust_N": fig.get(
"mean_thrust_N"),
"total_impulse_Ns": fig.get(
"total_impulse_Ns"),
923 "burn_time_s": fig.get(
"burn_time_s"),
"of_mean": of,
"lockup_psia": final.get(
"lockup_psia"),
924 "dome_psig": dome,
"copv_psig": final.get(
"fill_psig"),
"feasible": is_feasible},
925 "basis": (
"Whole Layer X burns with the rail's settings" + (
", the erosion replay on" if final.get(
"replay")
else "")
926 +
": a secant on lockup for the mean thrust, then Broyden on lockup and fill for the bottle margin. "
927 "The answer is the last burn."),
Dict[str, Any] solve(Evaluate evaluate, *, float start_lockup, float start_fill, float max_fill, float target, SetpointRequest request, Tuple[float, float] bounds, Optional[float] first_guess=None, Progress say=lambda s, None f, Callable[[], bool] cancelled=lambda:False, Optional[List[Dict[str, Any]]] seed=None)