Spaces:
Sleeping
Sleeping
| """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 | |
| 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 | |
| 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 | |