"""Transient downstream piping simulation — Euler integration. Components (in flow order): 1. Cryogenic reciprocating pump (2-D LUT: motor% x back-pressure -> mdot, T) 2. Snubber volume (pulsation dampener, single state = mass) 3. Control valve 2 (variable 0-100%, liquid/dense-fluid equation) 4. Pipe 2 (vents to atmosphere or discharges into tank) Ported from HuggingFace Space csh2/process_sim. """ from __future__ import annotations import math import numpy as np try: import CoolProp.CoolProp as CP except ImportError as e: raise ImportError("CoolProp is required. Install with: pip install CoolProp") from e # ── Constants ──────────────────────────────────────────────────────────────── FLUID = "Hydrogen" R_H2 = 4124.2 BAR2PA = 1.0e5 IN2M = 0.0254 FT2M = 0.3048 PSI2PA = 6894.76 MAX_STEPS = 500_000 # hard cap SIM_MODE_VALVE = "Pressure Build through Valve" SIM_MODE_FILL = "Simulated Fill" SIM_MODES = [SIM_MODE_VALVE, SIM_MODE_FILL] # ── Density → Pressure LUT (1-D, fixed T) ─────────────────────────────────── class _RhoToPressureLUT: """Pre-built 1-D lookup: density → pressure at fixed temperature. Replaces per-step CoolProp calls with np.interp (~1000x faster). Built once before the integration loop. """ N_PTS = 2000 def __init__(self, T: float, rho_min: float, rho_max: float, fluid: str = FLUID): rho_min = max(rho_min * 0.1, 1e-4) rho_max = rho_max * 1.5 self.rho_pts = np.geomspace(rho_min, rho_max, self.N_PTS) self.P_pts = np.empty(self.N_PTS) for i, rho in enumerate(self.rho_pts): try: self.P_pts[i] = CP.PropsSI("P", "D", rho, "T", T, fluid) except Exception: self.P_pts[i] = rho * R_H2 * T def __call__(self, rho: float) -> float: return float(np.interp(rho, self.rho_pts, self.P_pts)) # ── Simulation Engine ──────────────────────────────────────────────────────── class Simulation: """Euler integration engine — mass-only state variable with 1-D P(rho) LUT.""" def __init__(self, p: dict): self.p = p def _friction(self, Re: float, D: float, eps: float) -> float: if Re < 1.0: return 0.0 if Re < 2300.0: return 64.0 / Re term = eps / (3.7 * D) + 5.74 / Re**0.9 if term <= 0: return 0.02 return 0.25 / math.log10(term) ** 2 def _pipe_dp(self, mdot, rho, mu, D, L, A_pipe, fLD_cache, eps): if mdot <= 0 or rho < 1e-10: return 0.0 v = mdot / (rho * A_pipe) Re = rho * abs(v) * D / mu if mu > 1e-15 else 1e6 f = self._friction(Re, D, eps) return f * fLD_cache * 0.5 * rho * v * v @staticmethod def _valve_mdot_liq(P1, rho1, P2, Cv_eff): if P1 <= P2 or Cv_eff <= 0: return 0.0 dP_psi = (P1 - P2) / PSI2PA SG = rho1 / 999.0 if SG <= 0: return 0.0 q_gpm = Cv_eff * math.sqrt(dP_psi / SG) return q_gpm * rho1 * 6.309e-5 @staticmethod def _profile_lookup(t, profile_times, profile_values, default): if profile_times is None or len(profile_times) == 0: return default idx = np.searchsorted(profile_times, t, side="right") - 1 idx = max(0, min(idx, len(profile_values) - 1)) return profile_values[idx] def run(self, cb=None) -> dict: p = self.p dt = p["dt"] dur = p["duration"] N = int(dur / dt) P_atm = p["P_atm_bar"] * BAR2PA use_profiles = p.get("use_profiles", False) prof_times = p.get("profile_times", None) prof_motor = p.get("profile_motor", None) prof_valve2 = p.get("profile_valve2", None) pump_lut = p.get("pump_lut", None) sim_mode = p.get("sim_mode", SIM_MODE_VALVE) fill_mode = sim_mode == SIM_MODE_FILL D2 = p["p2_id_in"] * IN2M L2 = p["p2_len_ft"] * FT2M eps = p["roughness_mm"] * 1e-3 A2 = math.pi / 4.0 * D2 * D2 fLD2 = L2 / D2 static_motor_pct = p["motor_pct"] static_v2_pct = p["v2_pct"] T_out = p["pump_T_out"] V_snub = p["snub_V_L"] * 1e-3 V_snub_inv = 1.0 / V_snub P_snub0 = p["snub_P0_bar"] * BAR2PA T_snub = p["snub_T0_K"] try: rho_s0 = CP.PropsSI("D", "P", P_snub0, "T", T_snub, FLUID) except Exception: rho_s0 = P_snub0 / (R_H2 * T_snub) m_snub = rho_s0 * V_snub Cv2 = p["v2_cv"] P_ref = max(P_snub0, 2.0 * P_atm) try: mu_g = CP.PropsSI("V", "P", P_ref, "T", T_out, FLUID) except Exception: mu_g = 5.0e-6 pulsation = p["pulsation"] cpm_max = p.get("pump_cpm_max", 500.0) omega_max = 2.0 * math.pi * cpm_max / 60.0 pi_val = math.pi if cb: cb(0.0) if pump_lut is not None: mdot_max_est = float(np.nanmax(pump_lut.mdot_grid)) else: mdot_max_est = 0.05 rho_max_est = max((m_snub + mdot_max_est * dur) * V_snub_inv, 200.0) rho_min_est = max(rho_s0 * 0.1, 1e-4) P_lut = _RhoToPressureLUT(T_out, rho_min_est, rho_max_est) # ── Tank setup (Fill mode only) ── if fill_mode: V_tank = p["tank_V_L"] * 1e-3 V_tank_inv = 1.0 / V_tank P_tank0 = p["tank_P0_bar"] * BAR2PA tank_dT = p.get("tank_dT_K", 20.0) if pump_lut is not None: if use_profiles and prof_motor is not None and len(prof_motor) > 0: init_motor = float(prof_motor[0]) else: init_motor = static_motor_pct try: _, T_pump_init = pump_lut.lookup(init_motor, p["snub_P0_bar"]) except Exception: T_pump_init = T_out else: T_pump_init = T_out T_tank = T_pump_init + tank_dT try: rho_t0 = CP.PropsSI("D", "P", P_tank0, "T", T_tank, FLUID) except Exception: rho_t0 = P_tank0 / (R_H2 * T_tank) m_tank = rho_t0 * V_tank rho_tank_max_est = max( (m_tank + mdot_max_est * dur) * V_tank_inv, rho_t0 * 5.0 ) rho_tank_min_est = max(rho_t0 * 0.1, 1e-4) P_tank_lut = _RhoToPressureLUT(T_tank, rho_tank_min_est, rho_tank_max_est) else: V_tank = V_tank_inv = P_tank0 = T_tank = 0.0 m_tank = 0.0 P_tank_lut = None if cb: cb(0.05) t_arr = np.linspace(0, dur, N + 1) P_snub_arr = np.zeros(N + 1) mdot_pump_arr = np.zeros(N + 1) mdot_out_arr = np.zeros(N + 1) mdot_net_arr = np.zeros(N + 1) motor_pct_arr = np.zeros(N + 1) valve2_pct_arr = np.zeros(N + 1) T_pump_arr = np.zeros(N + 1) P_tank_arr = np.zeros(N + 1) if fill_mode else None m_tank_arr = np.zeros(N + 1) if fill_mode else None rpt = max(N // 100, 1) mdot_out_prev = 0.0 _sin = math.sin _max = max _min = min for i in range(N + 1): t = i * dt if use_profiles and prof_times is not None: motor_pct = self._profile_lookup(t, prof_times, prof_motor, static_motor_pct) v2_pct = self._profile_lookup(t, prof_times, prof_valve2, static_v2_pct) else: motor_pct = static_motor_pct v2_pct = static_v2_pct speed_frac = _max(0.0, _min(motor_pct * 0.01, 1.0)) f2 = _max(0.0, _min(v2_pct * 0.01, 1.0)) rho_s = _max(m_snub * V_snub_inv, 1e-6) P_snub = P_lut(rho_s) P_snub_bar = P_snub / BAR2PA # Downstream boundary if fill_mode: rho_t = _max(m_tank * V_tank_inv, 1e-6) P_tank = P_tank_lut(rho_t) P_downstream = P_tank else: P_tank = 0.0 P_downstream = P_atm if pump_lut is not None and speed_frac > 0: mdot_avg, T_pump = pump_lut.lookup(motor_pct, P_snub_bar) if pulsation: omega = speed_frac * omega_max phase = _sin(omega * t) mdot_p = mdot_avg * pi_val * _max(phase, 0.0) else: mdot_p = mdot_avg else: mdot_p = 0.0 T_pump = T_out Cv2_eff = Cv2 * f2 mdot_o = mdot_out_prev for _iter in range(3): P_avg2 = _max((P_snub + P_downstream) * 0.5, P_downstream) rho_g2 = P_avg2 / (R_H2 * T_out) dp_pipe2 = self._pipe_dp(mdot_o, rho_g2, mu_g, D2, L2, A2, fLD2, eps) P_back = P_downstream + dp_pipe2 mdot_o = self._valve_mdot_liq(P_snub, rho_s, P_back, Cv2_eff) P_snub_arr[i] = P_snub mdot_pump_arr[i] = mdot_p mdot_out_arr[i] = mdot_o mdot_net_arr[i] = mdot_p - mdot_o motor_pct_arr[i] = motor_pct valve2_pct_arr[i] = v2_pct T_pump_arr[i] = T_pump if fill_mode: P_tank_arr[i] = P_tank m_tank_arr[i] = m_tank if i < N: m_snub += (mdot_p - mdot_o) * dt m_snub = _max(m_snub, 1e-15) if fill_mode: m_tank += mdot_o * dt m_tank = _max(m_tank, 1e-15) mdot_out_prev = mdot_o if cb and i % rpt == 0: cb(0.05 + 0.95 * i / N) if cb: cb(1.0) out = { "time": t_arr, "P_snub_bar": P_snub_arr / BAR2PA, "mdot_pump": mdot_pump_arr * 60.0, "mdot_out": mdot_out_arr * 60.0, "mdot_net": mdot_net_arr * 60.0, "motor_pct": motor_pct_arr, "valve2_pct": valve2_pct_arr, "T_pump_K": T_pump_arr, "sim_mode": sim_mode, } if fill_mode: out["P_tank_bar"] = P_tank_arr / BAR2PA out["m_tank_kg"] = m_tank_arr out["T_tank_K"] = T_tank return out