""" Scenario Analyzer — LLM-driven diagnostic and optimization toolkit for the CSH2 cryogenic pump cycle simulator. Wraps cycle2mdot_v2.cycle_sim() to run parameter sweeps, diagnose results against physics-based checks, compare against real pump data from TimescaleDB, and find optimal parameter values. Usage: python scenario_analyzer.py single --Pexit 900 --speed 0.8 python scenario_analyzer.py sweep --param ICVport_mm --n 10 python scenario_analyzer.py compare --presets 900,500,350 python scenario_analyzer.py baseline --pressure-range 300 400 """ import copy import os import time import json import argparse import traceback import numpy as np from typing import Dict, List, Optional, Tuple, Any try: from sweep_cache import SweepCache _cache = SweepCache() except Exception: _cache = None def _db_conn_params() -> dict: """Build TimescaleDB connection params from env vars. No hardcoded credentials.""" host = os.environ.get('TSDB_HOST', '') if not host: return {} return { 'host': host, 'port': int(os.environ.get('TSDB_PORT', '32576')), 'database': os.environ.get('TSDB_DATABASE', 'tsdb'), 'user': os.environ.get('TSDB_USER', ''), 'password': os.environ.get('TSDB_PASSWORD', ''), 'sslmode': os.environ.get('TSDB_SSLMODE', 'require'), } # --------------------------------------------------------------------------- # Validated MVP 1.0 Simplex defaults (from hf_space/app.py:52-103) # These produce mdot ~1.78 kg/min at 900 barg. DO NOT use the GUI dataclass # defaults (ICV port=6.0 mm) — those are a different hardware spec. # --------------------------------------------------------------------------- DEFAULT_OPERATING = dict(Pexit_barg=900.0, speed_f=0.8, Ptank_barg=7.0, Psat_barg=2.0) # Array order per cycle2mdot_fast.py lines 62-102 DEFAULT_ICVparam = [20.5, 48, 5, 779.3, 33.4, 3.75, 0.0000736, 1.0, 100] DEFAULT_DCVparam = [8.3, 25, 3.79, 78.54, 7.8, 1.053, 0.0000026, 0.8, 200] DEFAULT_pump_geom = [40.3, 60, 120, 300, 0.8, 0.8, 0.026, 15, 37, 760000, 500, 0.01] DEFAULT_proc_param = [300, 10, 4, 30, 0.006, 0, 0.5, 0.2] # Parameter name → array, index, default, sweep range, tier # array='_operating' → inject as run_single() kwarg (Pexit_barg, speed_f, etc.) # array='_kwarg' → inject as extra kwarg to run_single() (flash_eff) SWEEP_PARAMS = { # ── Tier 1: Highest impact ── 'Kv_BB': {'array': 'proc_param', 'index': 4, 'default': 0.006, 'range': (0.0001, 0.05), 'tier': 1}, 'dvf': {'array': 'pump_geom', 'index': 11, 'default': 0.01, 'range': (0.005, 0.10), 'tier': 1}, 'flash_eff': {'array': '_kwarg', 'kwarg': 'flash_eff', 'default': 0.02, 'range': (0.0, 0.10), 'tier': 1}, 'ICVFs_N': {'array': 'ICVparam', 'index': 4, 'default': 33.4, 'range': (5.0, 80.0), 'tier': 1}, 'DCVFs_N': {'array': 'DCVparam', 'index': 4, 'default': 7.8, 'range': (1.0, 30.0), 'tier': 1}, 'speed_f': {'array': '_operating', 'kwarg': 'speed_f', 'default': 0.8, 'range': (0.1, 1.0), 'tier': 1}, 'Pexit_barg': {'array': '_operating', 'kwarg': 'Pexit_barg', 'default': 900.0, 'range': (50.0, 1100.0), 'tier': 1}, 'ICVport_mm': {'array': 'ICVparam', 'index': 0, 'default': 20.5, 'range': (8.0, 30.0), 'tier': 1}, 'DCVport_mm': {'array': 'DCVparam', 'index': 0, 'default': 8.3, 'range': (4.0, 16.0), 'tier': 1}, # ── Tier 2: Valve dynamics ── 'ICVleakKv': {'array': 'ICVparam', 'index': 6, 'default': 0.0000736, 'range': (1e-7, 0.01), 'tier': 2}, 'DCVleakKv': {'array': 'DCVparam', 'index': 6, 'default': 0.0000026, 'range': (1e-7, 0.01), 'tier': 2}, # ── Tier 3: Process/thermal ── 'Exp_eff': {'array': 'proc_param', 'index': 7, 'default': 0.2, 'range': (0.0, 1.0), 'tier': 3}, 'ICVcomp_eff': {'array': 'ICVparam', 'index': 7, 'default': 1.0, 'range': (0.5, 1.0), 'tier': 3}, 'DCVcomp_eff': {'array': 'DCVparam', 'index': 7, 'default': 0.8, 'range': (0.3, 1.0), 'tier': 3}, 'fric2chamber':{'array': 'proc_param', 'index': 6, 'default': 0.5, 'range': (0.0, 2.0), 'tier': 3}, 'Ptank_barg': {'array': '_operating', 'kwarg': 'Ptank_barg', 'default': 7.0, 'range': (1.0, 20.0), 'tier': 3}, 'Psat_barg': {'array': '_operating', 'kwarg': 'Psat_barg', 'default': 2.0, 'range': (0.5, 10.0), 'tier': 3}, } # Diagnostic checks: field, operator, threshold, severity, physics rationale DIAGNOSTIC_CHECKS = [ {'name': 'low_flow', 'field': 'mdot_kgpm', 'op': '<', 'threshold': 0.5, 'severity': 'CRITICAL', 'rationale': 'Insufficient filling — ICV may be stuck or excessive blowby'}, {'name': 'high_temp', 'field': 'Tc_peak_K', 'op': '>', 'threshold': 200, 'severity': 'WARNING', 'rationale': 'Blowby heating or insufficient cooling'}, {'name': 'over_pressure', 'field': 'pc_peak_barg', 'op': '>', 'threshold': 960, 'severity': 'CRITICAL', 'rationale': 'Peak chamber pressure exceeds MAWP (960 barg)'}, {'name': 'low_efficiency', 'field': 'mass_eff', 'op': '<', 'threshold': 0.5, 'severity': 'WARNING', 'rationale': 'Valve leakage or dead volume too large'}, {'name': 'icv_stuck', 'field': 'ICVmax_open_frac','op': '<', 'threshold': 0.01, 'severity': 'CRITICAL', 'rationale': 'Spring preload exceeds pressure differential — ICV never opens'}, {'name': 'dcv_slow', 'field': 'DCV_ct_s', 'op': '>', 'threshold': 0.02, 'severity': 'WARNING', 'rationale': 'DCV closure too slow — poppet mass too high or spring too soft'}, {'name': 'excessive_blowby','field': 'mass_eff', 'op': '<', 'threshold': 0.3, 'severity': 'CRITICAL', 'rationale': 'Piston seal failure — Kv_BB too high'}, ] # Phase 1 benchmarks (from CLAUDE.md) — fallback when no DB data available PHASE1_BENCHMARKS = { 'mdot_kgpm': 1.19, # midpoint of 1.13-1.25 range 'discharge_pressure_bar': 370, 'delivery_temp_K': 55, 'specific_energy_kWh_kg': 0.26, } # =========================================================================== # SimulationRunner # =========================================================================== class SimulationRunner: """Thin wrapper around cycle_sim() with error handling and structured output.""" def __init__(self, engine: str = 'fast'): from engine_bridge import cycle_sim, multi_cycle_sim, ENGINES if engine not in ENGINES and engine not in ('fast', 'fast_v2'): raise ValueError(f"Unknown engine: {engine!r}. Available: {ENGINES}") self._cycle_sim = cycle_sim self._multi_cycle_sim = multi_cycle_sim self._engine = engine def run_single(self, Pexit_barg: float = None, speed_f: float = None, Ptank_barg: float = None, Psat_barg: float = None, ICVparam: list = None, DCVparam: list = None, pump_geom: list = None, proc_param: list = None, fluid: str = 'h2', keep_history: bool = True, flash_eff: float = 0.0, exit_param=None, num_cycles: int = 1) -> dict: """Run one simulation and return a structured result dict. Any parameter left as None defaults to the MVP 1.0 Simplex preset. exit_param: list[5] or None — downstream volume parameters for V2 engine. num_cycles: int — number of pump cycles to simulate (requires exit_param for >1). """ Pexit_barg = Pexit_barg if Pexit_barg is not None else DEFAULT_OPERATING['Pexit_barg'] speed_f = speed_f if speed_f is not None else DEFAULT_OPERATING['speed_f'] Ptank_barg = Ptank_barg if Ptank_barg is not None else DEFAULT_OPERATING['Ptank_barg'] Psat_barg = Psat_barg if Psat_barg is not None else DEFAULT_OPERATING['Psat_barg'] ICVparam = ICVparam if ICVparam is not None else list(DEFAULT_ICVparam) DCVparam = DCVparam if DCVparam is not None else list(DEFAULT_DCVparam) pump_geom = pump_geom if pump_geom is not None else list(DEFAULT_pump_geom) proc_param = proc_param if proc_param is not None else list(DEFAULT_proc_param) inputs = dict(Pexit_barg=Pexit_barg, speed_f=speed_f, Ptank_barg=Ptank_barg, Psat_barg=Psat_barg) t0 = time.time() try: if num_cycles > 1 and exit_param is not None: out, hist = self._multi_cycle_sim( num_cycles, Pexit_barg, speed_f, Ptank_barg, Psat_barg, ICVparam, DCVparam, pump_geom, proc_param, fluid=fluid, prtMode=0, flash_eff=flash_eff, exit_param=exit_param, engine=self._engine, ) else: out, hist = self._cycle_sim( Pexit_barg, speed_f, Ptank_barg, Psat_barg, ICVparam, DCVparam, pump_geom, proc_param, fluid=fluid, prtMode=0, flash_eff=flash_eff, exit_param=exit_param, engine=self._engine, ) wall = time.time() - t0 scalars = { 'mdot_kgpm': float(hist['mdot_kgpm']), 'mass_eff': float(out[1, 0]), 'Tc_peak_K': float(np.max(hist['Tc_K'])), 'pc_peak_barg': float(np.max(hist['pc'])), 'DCV_ct_s': float(out[2, 0]), 'ICVmax_open_frac':float(out[4, 0]), 'm_discharged_kg': float(out[9, 0]), 'cycle_mass_eff': float(out[10, 0]), 'kWh_retract': float(out[13, 0]), 'kWh_extend': float(out[14, 0]), # Valve force / timing / velocity (match Tkinter GUI) 'cpu_time_s': float(out[0, 0]), 'ICV_opens_at_frac': float(out[3, 0]), 'max_DCV_force_N': float(out[5, 0]), 'max_ICV_open_force_N': float(out[6, 0]), 'max_ICV_close_force_N':float(out[7, 0]), 'max_ICV_velocity_mps': float(out[8, 0]), 'DCV_opens_at_frac': float(out[11, 0]), 'ICV_closure_s': float(out[12, 0]), } # Downstream volume scalars (V2 engine) if 'psnub' in hist: scalars['psnub_peak'] = float(np.max(hist['psnub'])) if 'pchss' in hist: scalars['pchss_final'] = float(hist['pchss'][-1]) if len(hist['pchss']) > 0 else 0.0 history = None if keep_history: history = {k: v for k, v in hist.items() if isinstance(v, np.ndarray)} return { 'success': True, 'error': None, 'inputs': inputs, 'scalars': scalars, 'history': history, 'wall_time_s': wall, 'steps': int(hist['steps']), } except Exception as e: wall = time.time() - t0 nan_scalars = {k: float('nan') for k in [ 'mdot_kgpm', 'mass_eff', 'Tc_peak_K', 'pc_peak_barg', 'DCV_ct_s', 'ICVmax_open_frac', 'm_discharged_kg', 'cycle_mass_eff', 'kWh_retract', 'kWh_extend', 'cpu_time_s', 'ICV_opens_at_frac', 'max_DCV_force_N', 'max_ICV_open_force_N', 'max_ICV_close_force_N', 'max_ICV_velocity_mps', 'DCV_opens_at_frac', 'ICV_closure_s']} return { 'success': False, 'error': f"{type(e).__name__}: {e}", 'inputs': inputs, 'scalars': nan_scalars, 'history': None, 'wall_time_s': wall, 'steps': 0, } def run_batch(self, conditions: List[dict], progress_cb=None) -> List[dict]: """Run multiple conditions serially. Each entry is kwargs for run_single.""" results = [] for i, cond in enumerate(conditions): r = self.run_single(**cond) results.append(r) if progress_cb: progress_cb(i + 1, len(conditions), r) return results # =========================================================================== # Diagnostician # =========================================================================== class Diagnostician: """Evaluate simulation outputs against physics-based diagnostic checks.""" def evaluate(self, result: dict) -> dict: """Run all diagnostic checks on a simulation result. Returns dict with findings, counts, and overall_status. """ if not result.get('success'): return { 'findings': [{'check': 'sim_failed', 'severity': 'CRITICAL', 'message': f"Simulation failed: {result.get('error', 'unknown')}", 'rationale': 'CoolProp crash — likely near H2 saturation boundary'}], 'n_critical': 1, 'n_warning': 0, 'overall_status': 'CRITICAL', } scalars = result['scalars'] findings = [] for chk in DIAGNOSTIC_CHECKS: val = scalars.get(chk['field']) if val is None or np.isnan(val): continue triggered = (val < chk['threshold']) if chk['op'] == '<' else (val > chk['threshold']) if triggered: findings.append({ 'check': chk['name'], 'severity': chk['severity'], 'value': round(val, 6), 'threshold': chk['threshold'], 'message': f"{chk['field']} = {val:.4f} {'below' if chk['op'] == '<' else 'above'} " f"threshold {chk['threshold']}", 'rationale': chk['rationale'], }) n_crit = sum(1 for f in findings if f['severity'] == 'CRITICAL') n_warn = sum(1 for f in findings if f['severity'] == 'WARNING') status = 'CRITICAL' if n_crit > 0 else ('WARNING' if n_warn > 0 else 'OK') return { 'findings': findings, 'n_critical': n_crit, 'n_warning': n_warn, 'overall_status': status, } def physics_interpretation(self, findings: list) -> str: """Map finding combinations to physics-grounded explanations.""" names = {f['check'] for f in findings} lines = [] if 'icv_stuck' in names and 'low_flow' in names: lines.append( "D1-like fault: ICV spring preload (ICVFs_N) exceeds the pressure " "differential across the valve. The poppet never lifts, so no fluid " "enters the chamber. Reduce ICVFs_N or increase Ptank_barg." ) if 'low_flow' in names and 'low_efficiency' in names and 'icv_stuck' not in names: lines.append( "D3-like fault: ICV leaking in reverse — hot, high-pressure gas flows " "back through the inlet valve, reducing net inflow and heating the " "chamber. Reduce ICVleakKv." ) if 'over_pressure' in names: lines.append( "Over-pressure condition: Peak chamber pressure exceeds MAWP (960 barg). " "The DCV may be undersized (increase DCVport_mm) or the DCV spring is " "too stiff (reduce DCVFs_N). Also check if Pexit exceeds safe range." ) if 'high_temp' in names and 'excessive_blowby' in names: lines.append( "D8-like efficiency degradation: Excessive blowby (high Kv_BB) causes " "mass loss AND frictional heating of the chamber. Reduce Kv_BB to " "simulate tighter piston seals." ) if 'dcv_slow' in names: lines.append( "DCV dynamics issue: Discharge valve takes too long to close, allowing " "backflow from the discharge manifold. Consider reducing DCVmass_g or " "increasing DCVSC_Npmm (stiffer spring)." ) if not lines: if findings: lines.append("Findings detected but no known fault-mode combination matched. " "Review individual findings above.") else: lines.append("All checks passed — simulation is within normal operating bounds.") return "\n".join(lines) # =========================================================================== # IdealBaseline # =========================================================================== class IdealBaseline: """Fetch real pump performance data from TimescaleDB to define targets.""" def __init__(self): self._conn_params = _db_conn_params() self._available = None def _check_available(self) -> bool: if self._available is not None: return self._available try: import psycopg2 conn = psycopg2.connect(**self._conn_params, connect_timeout=10) conn.close() self._available = True except Exception: self._available = False return self._available def _query(self, sql: str, params: tuple = None): import psycopg2 import pandas as pd with psycopg2.connect(**self._conn_params, connect_timeout=30) as conn: return pd.read_sql_query(sql, conn, params=params) def get_steady_state_benchmarks(self, pressure_lo: float = 300, pressure_hi: float = 400) -> dict: """Query real-world flow rate and temperature at a given pressure band. Returns dict with mean/std for mdot, pressure, temperature, specific energy. Falls back to Phase 1 benchmarks if DB is unavailable. """ if not self._check_available(): return self._fallback_benchmarks(pressure_lo, pressure_hi) try: sql = """ WITH big5 AS ( SELECT utc_full_timestamp, tagindex, val FROM procdatafloattable_utc_15sec WHERE tagindex IN (7, 18, 27, 13, 68, 2) AND utc_full_timestamp BETWEEN '2025-05-02' AND '2025-09-26' ), wide AS ( SELECT utc_full_timestamp, MAX(CASE WHEN tagindex=18 THEN val END) AS pt130, MAX(CASE WHEN tagindex=7 THEN val END) / 11.4 AS ft140_lh2, MAX(CASE WHEN tagindex=27 THEN val END) AS tt130, COALESCE( MAX(CASE WHEN tagindex=68 THEN val END), MAX(CASE WHEN tagindex=13 THEN val END) ) AS vfd, MAX(CASE WHEN tagindex=2 THEN val END) AS power_kw FROM big5 GROUP BY utc_full_timestamp ) SELECT COUNT(*) AS n_samples, AVG(ft140_lh2) * 60 AS mdot_kgpm_mean, STDDEV(ft140_lh2) * 60 AS mdot_kgpm_std, AVG(pt130) AS pt130_mean, STDDEV(pt130) AS pt130_std, AVG(tt130) AS tt130_mean, AVG(power_kw) AS power_kw_mean, CASE WHEN AVG(ft140_lh2) > 0 THEN AVG(power_kw) / (AVG(ft140_lh2) * 3600) ELSE NULL END AS kwh_per_kg FROM wide WHERE pt130 BETWEEN %s AND %s AND vfd > 10 AND ft140_lh2 > 0; """ df = self._query(sql, (pressure_lo, pressure_hi)) if df.empty or df.iloc[0]['n_samples'] == 0: return self._fallback_benchmarks(pressure_lo, pressure_hi) row = df.iloc[0] def _safe_float(val): """Convert to float, returning None for NULL/NaN.""" if val is None: return None try: f = float(val) return None if np.isnan(f) else f except (TypeError, ValueError): return None return { 'source': 'database', 'pressure_band': (pressure_lo, pressure_hi), 'n_samples': int(row['n_samples']), 'mdot_kgpm_mean': _safe_float(row['mdot_kgpm_mean']), 'mdot_kgpm_std': _safe_float(row['mdot_kgpm_std']), 'pt130_mean': _safe_float(row['pt130_mean']), 'tt130_mean': _safe_float(row['tt130_mean']), 'power_kw_mean': _safe_float(row['power_kw_mean']), 'kWh_per_kg': _safe_float(row['kwh_per_kg']), } except Exception as e: return {**self._fallback_benchmarks(pressure_lo, pressure_hi), 'db_error': str(e)} def _fallback_benchmarks(self, pressure_lo: float, pressure_hi: float) -> dict: return { 'source': 'phase1_benchmarks', 'pressure_band': (pressure_lo, pressure_hi), 'n_samples': 0, 'mdot_kgpm_mean': PHASE1_BENCHMARKS['mdot_kgpm'], 'mdot_kgpm_std': None, 'pt130_mean': PHASE1_BENCHMARKS['discharge_pressure_bar'], 'tt130_mean': PHASE1_BENCHMARKS['delivery_temp_K'], 'power_kw_mean': None, 'kWh_per_kg': PHASE1_BENCHMARKS['specific_energy_kWh_kg'], } def build_comparison_targets(self, Pexit_barg: float, band_width: float = 50) -> dict: """Return target values for comparison at a given Pexit.""" return self.get_steady_state_benchmarks( Pexit_barg - band_width / 2, Pexit_barg + band_width / 2) # =========================================================================== # CycleBaseline # =========================================================================== # H2 speed scale factor — calibrated value from fill_simulator pipeline. # Motor speed (%) → simulation speed_f: speed_f = motor_pct/100 * H2_SPEED_SCALE H2_SPEED_SCALE = 0.663504 class CycleBaseline: """Query individual pump cycles from TimescaleDB for real-vs-sim comparison. Uses the ``cycle_periods`` table for fill metadata and ``procdatafloattable_utc_1sec`` for per-second sensor time-series. """ def __init__(self): self._conn_params = _db_conn_params() def _query(self, sql: str, params: tuple = None): import psycopg2 import pandas as pd with psycopg2.connect(**self._conn_params, connect_timeout=30) as conn: return pd.read_sql_query(sql, conn, params=params) # ------------------------------------------------------------------ # 1. List cycles # ------------------------------------------------------------------ def get_cycles(self, pressure_lo: float = 0, pressure_hi: float = 2000, min_duration_s: float = 60) -> 'pd.DataFrame': """Return a DataFrame of pump cycles filtered by peak discharge pressure. Columns: id, cycle_start, duration_sec, max_motor_speed, max_discharge_pressure, avg_flow_rate, total_kg_dispensed, cryotank_start_pressure, min_fill_temperature, max_fill_temperature, total_pump_strokes """ import pandas as pd sql = """ SELECT id, cycle_start, cycle_end, duration_sec, max_motor_speed, max_flow_rate, avg_flow_rate, total_kg_dispensed, max_discharge_pressure, min_fill_temperature, max_fill_temperature, cryotank_start_pressure, total_pump_strokes FROM cycle_periods WHERE COALESCE(max_discharge_pressure, 0) BETWEEN %s AND %s AND COALESCE(duration_sec, 0) >= %s ORDER BY max_discharge_pressure DESC """ try: df = self._query(sql, (pressure_lo, pressure_hi, min_duration_s)) # Convert Decimal columns to float for col in df.select_dtypes(include=['object']).columns: try: df[col] = df[col].astype(float) except (ValueError, TypeError): pass numeric_cols = ['duration_sec', 'max_motor_speed', 'max_flow_rate', 'avg_flow_rate', 'total_kg_dispensed', 'max_discharge_pressure', 'min_fill_temperature', 'max_fill_temperature', 'cryotank_start_pressure', 'total_pump_strokes'] for col in numeric_cols: if col in df.columns: df[col] = pd.to_numeric(df[col], errors='coerce') return df except Exception as e: print(f"CycleBaseline.get_cycles error: {e}") return pd.DataFrame() # ------------------------------------------------------------------ # 2. Sensor time-series for a cycle # ------------------------------------------------------------------ def get_cycle_sensors(self, cycle_start, cycle_end) -> 'pd.DataFrame': """Query Big-5 + FT140 sensor data for a cycle window. Tries 1-second aggregate first, falls back to 15-second, then raw table. Returns wide DataFrame indexed by timestamp with columns: PT110, PT130, TT110, TT130, VFD, FT140, Power """ import pandas as pd tags_filter = "AND tagindex IN (7, 13, 17, 18, 24, 27, 68, 2)" tables = [ 'procdatafloattable_utc_1sec', 'procdatafloattable_utc_15sec', ] df = pd.DataFrame() for table in tables: sql = f""" SELECT utc_full_timestamp AS ts, tagindex, val FROM {table} WHERE utc_full_timestamp BETWEEN %s AND %s {tags_filter} ORDER BY utc_full_timestamp, tagindex """ try: df = self._query(sql, (cycle_start, cycle_end)) if not df.empty: print(f" Sensor data from {table}: {len(df)} rows") break except Exception: continue try: if df.empty: return pd.DataFrame() tag_map = { 17: 'PT110', 18: 'PT130', 24: 'TT110', 27: 'TT130', 7: 'FT140', 2: 'Power', } # Combine VFD tags 13 + 68 → VFD df['tag_name'] = df['tagindex'].map(tag_map) vfd_mask = df['tagindex'].isin([13, 68]) df.loc[vfd_mask, 'tag_name'] = 'VFD' df = df.dropna(subset=['tag_name']) df['val'] = pd.to_numeric(df['val'], errors='coerce') wide = df.pivot_table(index='ts', columns='tag_name', values='val', aggfunc='first') wide = wide.sort_index() return wide except Exception as e: print(f"CycleBaseline.get_cycle_sensors error: {e}") return pd.DataFrame() # ------------------------------------------------------------------ # 3. Build sim conditions from cycle metadata # ------------------------------------------------------------------ @staticmethod def cycle_to_sim_conditions(cycle_row: dict) -> dict: """Convert cycle_periods row into ICV_open() operating conditions. Applies H2 SPEED_SCALE_FACTOR to convert motor speed % → speed_f. Uses default tank pressure of 7.0 barg if not available. """ pmax = float(cycle_row.get('max_discharge_pressure') or 500) speed_pct = float(cycle_row.get('max_motor_speed') or 30) ptank = float(cycle_row.get('cryotank_start_pressure') or 7.0) # If ptank is 0 or unreasonable, default to 7.0 if ptank < 1.0: ptank = 7.0 # Convert motor speed % to simulation speed_f speed_f = speed_pct / 100.0 * H2_SPEED_SCALE return { 'Pexit_barg': pmax, 'speed_f': speed_f, 'Ptank_barg': ptank, 'Psat_barg': 2.0, } # =========================================================================== # SweepAnalyzer # =========================================================================== class SweepAnalyzer: """Parameter sweep execution, optimization, comparison, and plotting.""" def __init__(self, runner: SimulationRunner = None, diagnostician: Diagnostician = None): self.runner = runner or SimulationRunner() self.diag = diagnostician or Diagnostician() @staticmethod def _inject_param(pdef, val, arrays, operating, kwargs): """Inject a swept param value into the correct location. Handles three types: - '_operating' → operating conditions kwarg (Pexit_barg, speed_f, etc.) - '_kwarg' → extra kwarg to run_single() (flash_eff) - else → array[index] (ICVparam, DCVparam, pump_geom, proc_param) """ if pdef['array'] == '_operating': operating[pdef['kwarg']] = val elif pdef['array'] == '_kwarg': kwargs[pdef['kwarg']] = val else: arrays[pdef['array']][pdef['index']] = val def sweep_single_param(self, param_name: str, n_points: int = 10, Pexit_barg: float = None, speed_f: float = None, Ptank_barg: float = None, Psat_barg: float = None, flash_eff: float = 0.0, custom_range: tuple = None, ICVparam: list = None, DCVparam: list = None, pump_geom: list = None, proc_param: list = None, exit_param=None) -> dict: """Sweep one parameter across its range, holding everything else fixed. If ICVparam/DCVparam/pump_geom/proc_param are provided, they are used as the base values instead of the built-in defaults. Returns structured dict with per-point results, diagnostics, and optimal values. """ if param_name not in SWEEP_PARAMS: raise ValueError(f"Unknown sweep parameter: {param_name!r}. " f"Valid: {list(SWEEP_PARAMS.keys())}") pdef = SWEEP_PARAMS[param_name] lo, hi = custom_range or pdef['range'] values = np.linspace(lo, hi, n_points).tolist() # Assemble base arrays — use provided values or fall back to defaults arrays = { 'ICVparam': list(ICVparam) if ICVparam is not None else list(DEFAULT_ICVparam), 'DCVparam': list(DCVparam) if DCVparam is not None else list(DEFAULT_DCVparam), 'pump_geom': list(pump_geom) if pump_geom is not None else list(DEFAULT_pump_geom), 'proc_param':list(proc_param)if proc_param is not None else list(DEFAULT_proc_param), } operating = dict(DEFAULT_OPERATING) if Pexit_barg is not None: operating['Pexit_barg'] = Pexit_barg if speed_f is not None: operating['speed_f'] = speed_f if Ptank_barg is not None: operating['Ptank_barg'] = Ptank_barg if Psat_barg is not None: operating['Psat_barg'] = Psat_barg extra_kwargs = {'flash_eff': flash_eff, 'exit_param': exit_param} results = [] failures = [] t0 = time.time() for i, val in enumerate(values): # Deep copy and inject the swept value run_arrays = {k: list(v) for k, v in arrays.items()} run_operating = dict(operating) run_kwargs = dict(extra_kwargs) self._inject_param(pdef, val, run_arrays, run_operating, run_kwargs) # Check cache before running simulation cache_params = dict( Pexit_barg=run_operating['Pexit_barg'], speed_f=run_operating['speed_f'], Ptank_barg=run_operating['Ptank_barg'], Psat_barg=run_operating['Psat_barg'], ICVparam=run_arrays['ICVparam'], DCVparam=run_arrays['DCVparam'], pump_geom=run_arrays['pump_geom'], proc_param=run_arrays['proc_param'], flash_eff=run_kwargs.get('flash_eff', 0.0), ) cached = _cache.get(**cache_params) if _cache else None if cached is not None: scalars, wt = cached r = {'success': True, 'scalars': scalars, 'wall_time_s': wt, 'steps': 0, 'error': None, 'history': None, 'inputs': run_operating} else: r = self.runner.run_single( Pexit_barg=run_operating['Pexit_barg'], speed_f=run_operating['speed_f'], Ptank_barg=run_operating['Ptank_barg'], Psat_barg=run_operating['Psat_barg'], ICVparam=run_arrays['ICVparam'], DCVparam=run_arrays['DCVparam'], pump_geom=run_arrays['pump_geom'], proc_param=run_arrays['proc_param'], keep_history=False, **run_kwargs, ) if r['success'] and _cache: _cache.put(r['scalars'], r['wall_time_s'], **cache_params) diag = self.diag.evaluate(r) entry = {'param_value': val, **r['scalars'], 'success': r['success'], 'wall_time_s': r['wall_time_s'], 'status': diag['overall_status']} results.append(entry) if not r['success']: failures.append({'value': val, 'error': r['error']}) print(f" [{i+1}/{n_points}] {param_name}={val:.4g} " f"mdot={r['scalars']['mdot_kgpm']:.4f} " f"eff={r['scalars']['mass_eff']:.4f} " f"status={diag['overall_status']}") total_time = time.time() - t0 # Find optimal successful = [r for r in results if r['success'] and not np.isnan(r['mdot_kgpm'])] optimal = {} if successful: best_mdot = max(successful, key=lambda x: x['mdot_kgpm']) best_eff = max(successful, key=lambda x: x['mass_eff']) optimal = { 'by_mdot': {'value': best_mdot['param_value'], 'mdot_kgpm': best_mdot['mdot_kgpm']}, 'by_efficiency': {'value': best_eff['param_value'], 'mass_eff': best_eff['mass_eff']}, } return { 'param_name': param_name, 'param_values': values, 'operating_conditions': operating, 'results': results, 'optimal': optimal, 'failures': failures, 'n_runs': len(results), 'n_failures': len(failures), 'total_time_s': total_time, } def sweep_two_params(self, param_x: str, param_y: str, n_x: int = 8, n_y: int = 8, Pexit_barg: float = None, speed_f: float = None, Ptank_barg: float = None, Psat_barg: float = None, flash_eff: float = 0.0, range_x: tuple = None, range_y: tuple = None, ICVparam: list = None, DCVparam: list = None, pump_geom: list = None, proc_param: list = None, exit_param=None) -> dict: """2D parameter sweep. Returns structured dict with NxM grid of results. If ICVparam/DCVparam/pump_geom/proc_param are provided, they are used as the base values instead of the built-in defaults. """ if param_x == param_y: raise ValueError(f"X and Y parameters must differ, got both = {param_x!r}") for p in (param_x, param_y): if p not in SWEEP_PARAMS: raise ValueError(f"Unknown sweep param: {p!r}") pdef_x, pdef_y = SWEEP_PARAMS[param_x], SWEEP_PARAMS[param_y] lx, hx = range_x or pdef_x['range'] ly, hy = range_y or pdef_y['range'] vals_x = np.linspace(lx, hx, n_x).tolist() vals_y = np.linspace(ly, hy, n_y).tolist() # Assemble base arrays — use provided values or fall back to defaults arrays = { 'ICVparam': list(ICVparam) if ICVparam is not None else list(DEFAULT_ICVparam), 'DCVparam': list(DCVparam) if DCVparam is not None else list(DEFAULT_DCVparam), 'pump_geom': list(pump_geom) if pump_geom is not None else list(DEFAULT_pump_geom), 'proc_param':list(proc_param)if proc_param is not None else list(DEFAULT_proc_param), } operating = dict(DEFAULT_OPERATING) if Pexit_barg is not None: operating['Pexit_barg'] = Pexit_barg if speed_f is not None: operating['speed_f'] = speed_f if Ptank_barg is not None: operating['Ptank_barg'] = Ptank_barg if Psat_barg is not None: operating['Psat_barg'] = Psat_barg extra_kwargs = {'flash_eff': flash_eff, 'exit_param': exit_param} grid_keys = ['mdot_kgpm', 'mass_eff', 'Tc_peak_K', 'pc_peak_barg'] grid = {k: np.full((n_y, n_x), np.nan) for k in grid_keys} results_flat = [] t0 = time.time() total = n_x * n_y done = 0 for iy, vy in enumerate(vals_y): for ix, vx in enumerate(vals_x): run_arrays = {k: list(v) for k, v in arrays.items()} run_operating = dict(operating) run_kwargs = dict(extra_kwargs) self._inject_param(pdef_x, vx, run_arrays, run_operating, run_kwargs) self._inject_param(pdef_y, vy, run_arrays, run_operating, run_kwargs) cache_params = dict( Pexit_barg=run_operating['Pexit_barg'], speed_f=run_operating['speed_f'], Ptank_barg=run_operating['Ptank_barg'], Psat_barg=run_operating['Psat_barg'], ICVparam=run_arrays['ICVparam'], DCVparam=run_arrays['DCVparam'], pump_geom=run_arrays['pump_geom'], proc_param=run_arrays['proc_param'], flash_eff=run_kwargs.get('flash_eff', 0.0), ) cached = _cache.get(**cache_params) if _cache else None if cached is not None: scalars, wt = cached r = {'success': True, 'scalars': scalars, 'wall_time_s': wt} else: r = self.runner.run_single( Pexit_barg=run_operating['Pexit_barg'], speed_f=run_operating['speed_f'], Ptank_barg=run_operating['Ptank_barg'], Psat_barg=run_operating['Psat_barg'], ICVparam=run_arrays['ICVparam'], DCVparam=run_arrays['DCVparam'], pump_geom=run_arrays['pump_geom'], proc_param=run_arrays['proc_param'], keep_history=False, **run_kwargs, ) if r['success'] and _cache: _cache.put(r['scalars'], r['wall_time_s'], **cache_params) scalars = r['scalars'] for k in grid_keys: grid[k][iy, ix] = scalars.get(k, np.nan) done += 1 results_flat.append({ 'x': vx, 'y': vy, **{k: scalars.get(k, np.nan) for k in grid_keys}, 'success': r['success'], 'wall_time_s': r['wall_time_s'], }) print(f" [{done}/{total}] {param_x}={vx:.4g}, {param_y}={vy:.4g} " f"mdot={scalars.get('mdot_kgpm', 0):.4f}") return { 'param_x': param_x, 'param_y': param_y, 'vals_x': vals_x, 'vals_y': vals_y, 'grid': grid, 'results': results_flat, 'total_time_s': time.time() - t0, } def find_optimal(self, sweep_result: dict, objective: str = 'mdot_kgpm', constraints: dict = None) -> dict: """Find optimal parameter value from a sweep, subject to constraints. constraints: {'pc_peak_barg': ('<', 875), 'mass_eff': ('>', 0.5)} """ candidates = [r for r in sweep_result['results'] if r['success'] and not np.isnan(r.get(objective, float('nan')))] if constraints: for field, (op, thresh) in constraints.items(): if op == '<': candidates = [r for r in candidates if r.get(field, float('inf')) < thresh] elif op == '>': candidates = [r for r in candidates if r.get(field, float('-inf')) > thresh] if not candidates: return {'found': False, 'reason': 'No feasible points after constraints'} best = max(candidates, key=lambda x: x[objective]) return { 'found': True, 'param_name': sweep_result['param_name'], 'optimal_value': best['param_value'], 'objective': objective, 'objective_value': best[objective], 'all_scalars': {k: v for k, v in best.items() if k not in ('success', 'wall_time_s', 'status', 'param_value')}, } def compare_scenarios(self, scenarios: List[dict]) -> dict: """Compare named scenarios. Each scenario: {'name': str, 'Pexit_barg': float, ...overrides...} """ results = [] for sc in scenarios: name = sc.pop('name', f"Scenario_{len(results)+1}") r = self.runner.run_single(**sc, keep_history=False) d = self.diag.evaluate(r) results.append({ 'name': name, 'inputs': r['inputs'], 'scalars': r['scalars'], 'diagnostics': d, 'success': r['success'], 'wall_time_s': r['wall_time_s'], }) # Build comparison table metrics = ['mdot_kgpm', 'mass_eff', 'Tc_peak_K', 'pc_peak_barg', 'DCV_ct_s'] comparison = {} for m in metrics: vals = [r['scalars'].get(m, float('nan')) for r in results] comparison[m] = { 'values': {r['name']: v for r, v in zip(results, vals)}, 'best': results[int(np.nanargmax(vals) if m != 'Tc_peak_K' else np.nanargmin(vals))]['name'] if any(not np.isnan(v) for v in vals) else None, } return {'scenarios': results, 'comparison': comparison} def plot_sweep(self, sweep_result: dict, save_path: str = None): """Generate a 3-panel sweep visualization and save to PNG.""" import matplotlib.pyplot as plt sr = sweep_result vals = [r['param_value'] for r in sr['results'] if r['success']] mdots = [r['mdot_kgpm'] for r in sr['results'] if r['success']] effs = [r['mass_eff'] for r in sr['results'] if r['success']] temps = [r['Tc_peak_K'] for r in sr['results'] if r['success']] press = [r['pc_peak_barg'] for r in sr['results'] if r['success']] if not vals: print("No successful runs to plot.") return fig, axes = plt.subplots(3, 1, figsize=(10, 10), sharex=True) fig.suptitle(f"Parameter Sweep: {sr['param_name']}\n" f"Pexit={sr['operating_conditions']['Pexit_barg']} barg, " f"speed_f={sr['operating_conditions']['speed_f']}", fontsize=13) # Panel 1: mdot axes[0].plot(vals, mdots, 'b-o', linewidth=2, markersize=5) axes[0].set_ylabel('Mass Flow Rate (kg/min)', color='blue') axes[0].axvline(SWEEP_PARAMS[sr['param_name']]['default'], color='gray', linestyle='--', alpha=0.7, label='Default') axes[0].legend(loc='upper left') axes[0].grid(True, alpha=0.3) # Panel 2: efficiency + temperature ax2a = axes[1] ax2b = ax2a.twinx() ax2a.plot(vals, effs, 'g-s', linewidth=2, markersize=5, label='Mass Efficiency') ax2b.plot(vals, temps, 'orange', linestyle='-', marker='^', linewidth=2, markersize=5, label='Tc_peak (K)') ax2a.set_ylabel('Mass Efficiency', color='green') ax2b.set_ylabel('Peak Chamber Temp (K)', color='orange') ax2a.axhline(0.5, color='green', linestyle=':', alpha=0.5, label='Eff threshold') ax2b.axhline(200, color='orange', linestyle=':', alpha=0.5, label='Temp threshold') lines1, labels1 = ax2a.get_legend_handles_labels() lines2, labels2 = ax2b.get_legend_handles_labels() ax2a.legend(lines1 + lines2, labels1 + labels2, loc='upper left', fontsize=8) ax2a.grid(True, alpha=0.3) # Panel 3: peak pressure with MAWP line axes[2].plot(vals, press, 'r-D', linewidth=2, markersize=5) axes[2].axhline(960, color='darkred', linestyle='--', linewidth=2, alpha=0.8, label='MAWP (960 barg)') axes[2].set_ylabel('Peak Chamber Pressure (barg)', color='red') axes[2].set_xlabel(sr['param_name']) axes[2].legend(loc='upper left') axes[2].grid(True, alpha=0.3) plt.tight_layout() path = save_path or f"sweep_{sr['param_name']}.png" fig.savefig(path, dpi=150, bbox_inches='tight') plt.close(fig) print(f" Plot saved: {path}") def to_dataframe(self, sweep_result: dict): """Convert sweep results to a pandas DataFrame.""" import pandas as pd rows = [] for r in sweep_result['results']: row = {'param_value': r['param_value'], 'success': r['success'], 'status': r.get('status', ''), 'wall_time_s': r.get('wall_time_s', 0)} for k in ['mdot_kgpm', 'mass_eff', 'Tc_peak_K', 'pc_peak_barg', 'DCV_ct_s', 'ICVmax_open_frac', 'm_discharged_kg', 'cycle_mass_eff', 'kWh_retract', 'kWh_extend']: row[k] = r.get(k, float('nan')) rows.append(row) df = pd.DataFrame(rows) df.columns.name = sweep_result['param_name'] return df def save_results(self, sweep_result: dict, path: str): """Save sweep results to CSV.""" df = self.to_dataframe(sweep_result) df.to_csv(path, index=False, float_format='%.6g') print(f" Results saved: {path}") # =========================================================================== # CLI # =========================================================================== def _print_single_result(result: dict, diag: dict, interp: str): """Pretty-print a single simulation result.""" s = result['scalars'] print("\n" + "=" * 65) print(f" Simulation Result ({result['wall_time_s']:.2f}s, {result['steps']} steps)") print("=" * 65) inp = result['inputs'] print(f" Pexit={inp['Pexit_barg']} barg speed_f={inp['speed_f']} " f"Ptank={inp['Ptank_barg']} barg Psat={inp['Psat_barg']} barg") print("-" * 65) print(f" Mass flow rate: {s['mdot_kgpm']:.4f} kg/min") print(f" Mass efficiency: {s['mass_eff']:.4f}") print(f" Peak chamber temp: {s['Tc_peak_K']:.1f} K") print(f" Peak chamber press: {s['pc_peak_barg']:.1f} barg") print(f" DCV closure time: {s['DCV_ct_s']:.6f} s") print(f" ICV max open frac: {s['ICVmax_open_frac']:.4f}") print(f" Mass discharged: {s['m_discharged_kg']:.6f} kg/cycle") print(f" Cycle mass eff: {s['cycle_mass_eff']:.4f}") print(f" kWh retract: {s.get('kWh_retract', float('nan')):.6f} kWh/kg") print(f" kWh extend: {s.get('kWh_extend', float('nan')):.6f} kWh/kg") print("-" * 65) status = diag['overall_status'] tag = {'OK': 'OK', 'WARNING': 'WARNING', 'CRITICAL': 'CRITICAL'}[status] print(f" Status: [{tag}] ({diag['n_critical']} critical, {diag['n_warning']} warnings)") for f in diag['findings']: print(f" [{f['severity']}] {f['check']}: {f['message']}") print("-" * 65) print(f" Interpretation:\n {interp.replace(chr(10), chr(10) + ' ')}") print("=" * 65) def main(): parser = argparse.ArgumentParser( description='CSH2 Cryogenic Pump Scenario Analyzer') sub = parser.add_subparsers(dest='command') # single p_single = sub.add_parser('single', help='Run a single simulation') p_single.add_argument('--Pexit', type=float, default=900.0) p_single.add_argument('--speed', type=float, default=0.8) p_single.add_argument('--Ptank', type=float, default=7.0) p_single.add_argument('--Psat', type=float, default=2.0) # sweep p_sweep = sub.add_parser('sweep', help='Sweep one parameter') p_sweep.add_argument('--param', required=True, choices=list(SWEEP_PARAMS.keys())) p_sweep.add_argument('--n', type=int, default=10) p_sweep.add_argument('--Pexit', type=float, default=900.0) p_sweep.add_argument('--speed', type=float, default=0.8) p_sweep.add_argument('--save-csv', type=str, default=None) p_sweep.add_argument('--save-plot', type=str, default=None) # compare p_comp = sub.add_parser('compare', help='Compare pressure presets') p_comp.add_argument('--presets', type=str, default='900,500,350', help='Comma-separated Pexit values') # baseline p_base = sub.add_parser('baseline', help='Query real data baseline') p_base.add_argument('--pressure-range', nargs=2, type=float, default=[300, 400], metavar=('LO', 'HI')) args = parser.parse_args() if args.command == 'single': runner = SimulationRunner() diag = Diagnostician() r = runner.run_single(Pexit_barg=args.Pexit, speed_f=args.speed, Ptank_barg=args.Ptank, Psat_barg=args.Psat) d = diag.evaluate(r) interp = diag.physics_interpretation(d['findings']) _print_single_result(r, d, interp) elif args.command == 'sweep': analyzer = SweepAnalyzer() print(f"\nSweeping {args.param} ({args.n} points) at Pexit={args.Pexit} barg...") sr = analyzer.sweep_single_param(args.param, n_points=args.n, Pexit_barg=args.Pexit, speed_f=args.speed) print(f"\nSweep complete: {sr['n_runs']} runs, {sr['n_failures']} failures, " f"{sr['total_time_s']:.1f}s total") if sr['optimal']: o = sr['optimal'] print(f" Best mdot: {args.param}={o['by_mdot']['value']:.4g} → " f"mdot={o['by_mdot']['mdot_kgpm']:.4f} kg/min") print(f" Best eff: {args.param}={o['by_efficiency']['value']:.4g} → " f"eff={o['by_efficiency']['mass_eff']:.4f}") # Constrained optimal opt = analyzer.find_optimal(sr, objective='mdot_kgpm', constraints={'pc_peak_barg': ('<', 960), 'mass_eff': ('>', 0.3)}) if opt['found']: print(f" Constrained optimal (pc<960, eff>0.3): " f"{args.param}={opt['optimal_value']:.4g} → " f"mdot={opt['objective_value']:.4f} kg/min") csv_path = args.save_csv or f"sweep_{args.param}.csv" analyzer.save_results(sr, csv_path) plot_path = args.save_plot or f"sweep_{args.param}.png" analyzer.plot_sweep(sr, plot_path) elif args.command == 'compare': analyzer = SweepAnalyzer() presets = [float(x) for x in args.presets.split(',')] scenarios = [{'name': f'{int(p)} barg', 'Pexit_barg': p, 'speed_f': 0.8 if p >= 500 else 1.0} for p in presets] print(f"\nComparing {len(scenarios)} scenarios...") comp = analyzer.compare_scenarios(scenarios) print(f"\n{'Scenario':<16} {'mdot':>10} {'eff':>10} {'Tc_peak':>10} {'pc_peak':>10} {'status':>10}") print("-" * 66) for sc in comp['scenarios']: s = sc['scalars'] print(f"{sc['name']:<16} {s['mdot_kgpm']:>10.4f} {s['mass_eff']:>10.4f} " f"{s['Tc_peak_K']:>10.1f} {s['pc_peak_barg']:>10.1f} " f"{sc['diagnostics']['overall_status']:>10}") elif args.command == 'baseline': baseline = IdealBaseline() lo, hi = args.pressure_range print(f"\nQuerying real pump data at {lo}-{hi} barg...") b = baseline.get_steady_state_benchmarks(lo, hi) print(f"\n Source: {b['source']}") print(f" Pressure band: {b['pressure_band']}") print(f" N samples: {b['n_samples']}") if b.get('mdot_kgpm_mean') is not None: print(f" Flow rate: {b['mdot_kgpm_mean']:.4f} +/- " f"{b.get('mdot_kgpm_std', 0) or 0:.4f} kg/min") if b.get('pt130_mean') is not None: print(f" Avg pressure: {b['pt130_mean']:.1f} bar") if b.get('tt130_mean') is not None: print(f" Avg temp: {b['tt130_mean']:.1f} K") if b.get('kWh_per_kg') is not None: print(f" Specific energy: {b['kWh_per_kg']:.4f} kWh/kg") if b.get('db_error'): print(f" DB error: {b['db_error']}") else: parser.print_help() if __name__ == '__main__': main()