from __future__ import annotations import argparse from dataclasses import dataclass, field from datetime import UTC, datetime from math import nextafter, sqrt from pathlib import Path from PythonModels.core.solver import SolveIVPConfig, integrate_ode from PythonModels.reporting.amesim_results import AmesimResults, load_test_mql_amesim_results from PythonModels.reporting.test_mql_comparison import ( TestMqlComparisonResult, interpolate_series_value, write_test_mql_comparison_csv, ) from PythonModels.reporting.test_mql_output_schema import ( TestMqlOutputSchema, build_test_mql_output_schema, ) from PythonModels.reporting.test_mql_output_validation import ( TestMqlValidatedOutput, compare_validated_test_mql_output, validate_test_mql_output, ) from PythonModels.systems.test_mql import ( TestMqlPnl0001LineRhsDiagnostic, TestMqlSimulationResult, TestMqlSystem, ) from PythonModels.systems.test_mql_pneumatic import AMESIM_REFERENCE_PRESSURE_PA DEFAULT_FULL_STATE_COMPARISON_DATA_PATHS = ( "press@pn_c1_8", "vol@pn_c1_8", "vol1@pn_brp2_8", "vvol1@pn_brp2_8", "x1@mass_friction_endstops_10", "v1@mass_friction_endstops_10", "acc1@mass_friction_endstops_10", "x1@mass_friction_endstops_18", "v1@mass_friction_endstops_18", "acc1@mass_friction_endstops_18", "dm1@pneumatic_69", "xv@pn_morifice_1", "dm2@pn_morifice_1", ) @dataclass(frozen=True) class TestMqlFullStateSignalDiagnostic: data_path: str initial_time_s: float final_time_s: float initial_python_value: float initial_amesim_value: float initial_abs_error: float final_python_value: float final_amesim_value: float final_abs_error: float @dataclass(frozen=True) class TestMqlFullStateFlowDiagnostic: data_path: str initial_time_s: float final_time_s: float initial_python_dm1_g_s: float initial_amesim_dm1_g_s: float initial_python_canonical_kg_s: float initial_amesim_canonical_kg_s: float initial_canonical_abs_error_kg_s: float final_python_dm1_g_s: float final_amesim_dm1_g_s: float final_python_canonical_kg_s: float final_amesim_canonical_kg_s: float final_canonical_abs_error_kg_s: float @dataclass(frozen=True) class TestMqlFullStatePnvoDiagnostic: alias: str initial_time_s: float final_time_s: float initial_python_opening: float initial_amesim_opening: float final_python_opening: float final_amesim_opening: float initial_python_mass_flow_kg_s: float initial_amesim_mass_flow_kg_s: float final_python_mass_flow_kg_s: float final_amesim_mass_flow_kg_s: float final_mass_flow_abs_error_kg_s: float @dataclass(frozen=True) class TestMqlFullStateComparisonRun: system: TestMqlSystem closure: object amesim_results: AmesimResults output_schema: TestMqlOutputSchema result: TestMqlSimulationResult output: TestMqlValidatedOutput comparison: TestMqlComparisonResult @property def sample_count(self) -> int: return len(self.output.times) @property def signal_count(self) -> int: return len(self.output.data_paths) def metrics_by_max_abs_error(self): return tuple( sorted( self.comparison.metrics, key=lambda metric: metric.max_abs_error, reverse=True, ) ) @property def largest_abs_error_metric(self): metrics = self.metrics_by_max_abs_error() return metrics[0] if metrics else None def signal_diagnostic(self, data_path: str) -> TestMqlFullStateSignalDiagnostic: times = self.output.times python_values = self.output.series_by_data_path[data_path] amesim_values = self.amesim_results.series(data_path) initial_time = float(times[0]) final_time = float(times[-1]) initial_python = float(python_values[0]) final_python = float(python_values[-1]) initial_amesim = interpolate_series_value( self.amesim_results.times, amesim_values, initial_time, ) final_amesim = interpolate_series_value( self.amesim_results.times, amesim_values, final_time, ) return TestMqlFullStateSignalDiagnostic( data_path=data_path, initial_time_s=initial_time, final_time_s=final_time, initial_python_value=initial_python, initial_amesim_value=initial_amesim, initial_abs_error=abs(initial_python - initial_amesim), final_python_value=final_python, final_amesim_value=final_amesim, final_abs_error=abs(final_python - final_amesim), ) def pnl0001_mass_flow_diagnostic( self, data_path: str = "dm1@pneumatic_69", ) -> TestMqlFullStateFlowDiagnostic: diagnostic = self.signal_diagnostic(data_path) initial_python_canonical = -diagnostic.initial_python_value * 1.0e-3 initial_amesim_canonical = -diagnostic.initial_amesim_value * 1.0e-3 final_python_canonical = -diagnostic.final_python_value * 1.0e-3 final_amesim_canonical = -diagnostic.final_amesim_value * 1.0e-3 return TestMqlFullStateFlowDiagnostic( data_path=data_path, initial_time_s=diagnostic.initial_time_s, final_time_s=diagnostic.final_time_s, initial_python_dm1_g_s=diagnostic.initial_python_value, initial_amesim_dm1_g_s=diagnostic.initial_amesim_value, initial_python_canonical_kg_s=initial_python_canonical, initial_amesim_canonical_kg_s=initial_amesim_canonical, initial_canonical_abs_error_kg_s=abs( initial_python_canonical - initial_amesim_canonical ), final_python_dm1_g_s=diagnostic.final_python_value, final_amesim_dm1_g_s=diagnostic.final_amesim_value, final_python_canonical_kg_s=final_python_canonical, final_amesim_canonical_kg_s=final_amesim_canonical, final_canonical_abs_error_kg_s=abs( final_python_canonical - final_amesim_canonical ), ) def pnvo_diagnostic( self, alias: str = "pn_morifice_1", ) -> TestMqlFullStatePnvoDiagnostic: opening = self.signal_diagnostic(f"xv@{alias}") mass_flow = self.signal_diagnostic(f"dm2@{alias}") initial_python_mass_flow = mass_flow.initial_python_value * 1.0e-3 initial_amesim_mass_flow = mass_flow.initial_amesim_value * 1.0e-3 final_python_mass_flow = mass_flow.final_python_value * 1.0e-3 final_amesim_mass_flow = mass_flow.final_amesim_value * 1.0e-3 return TestMqlFullStatePnvoDiagnostic( alias=alias, initial_time_s=opening.initial_time_s, final_time_s=opening.final_time_s, initial_python_opening=opening.initial_python_value, initial_amesim_opening=opening.initial_amesim_value, final_python_opening=opening.final_python_value, final_amesim_opening=opening.final_amesim_value, initial_python_mass_flow_kg_s=initial_python_mass_flow, initial_amesim_mass_flow_kg_s=initial_amesim_mass_flow, final_python_mass_flow_kg_s=final_python_mass_flow, final_amesim_mass_flow_kg_s=final_amesim_mass_flow, final_mass_flow_abs_error_kg_s=abs( final_python_mass_flow - final_amesim_mass_flow ), ) def diagnostics_by_final_abs_error(self): return tuple( sorted( ( self.signal_diagnostic(data_path) for data_path in self.output.data_paths ), key=lambda diagnostic: diagnostic.final_abs_error, reverse=True, ) ) @property def largest_final_abs_error_diagnostic(self): diagnostics = self.diagnostics_by_final_abs_error() return diagnostics[0] if diagnostics else None def chamber_rhs_diagnostic(self, chamber_alias: str, sample_index: int = -1): state_vector = [row[sample_index] for row in self.result.y] return self.closure.variable_chamber_rhs_diagnostic( chamber_alias=chamber_alias, state_vector=state_vector, time_s=float(self.result.t[sample_index]), ) @dataclass(frozen=True) class TestMqlPnvoEventBoundaryDiagnostic: orifice_alias: str event_time_s: float integration_stop_time_s: float data_paths: tuple[str, ...] python_values_by_data_path: dict[str, float] amesim_values_by_data_path: dict[str, float] def abs_error(self, data_path: str) -> float: return abs( self.python_values_by_data_path[data_path] - self.amesim_values_by_data_path[data_path] ) @dataclass(frozen=True) class TestMqlPnvoEventWindowSegmentDiagnostic: t_start: float t_stop: float method: str rtol: float atol: float max_step: float rhs_evaluations: int success: bool message: str @dataclass(frozen=True) class TestMqlPnl0001PressureLossCalibrationDiagnostic: line_alias: str chamber_alias: str time_s: float amesim_cm: float amesim_dm1_g_s: float amesim_mass_flow_magnitude_kg_s: float amesim_line_gauge_pressure_pa: float amesim_chamber_gauge_pressure_pa: float amesim_line_temperature_k: float amesim_pressure_drop_pa: float current_darcy_pressure_drop_pa: float pressure_drop_multiplier: float candidate_pn2pipefr_dm1_g_s: float candidate_pn2pipefr_to_amesim_dm1_ratio: float amesim_linear_conductance_kg_s_sqrt_k_per_pa: float python_linear_conductance_kg_s_sqrt_k_per_pa: float python_to_amesim_linear_conductance_ratio: float @dataclass(frozen=True) class TestMqlPnvoFlowParameterDiagnostic: orifice_alias: str time_s: float amesim_cm: float amesim_dm2_g_s: float amesim_opening: float amesim_gas_velocity_m_s: float python_opening: float python_flow_coefficient: float python_effective_area_m2: float python_line_pressure_pa: float python_boundary_pressure_pa: float python_upstream_pressure_pa: float python_upstream_temperature_k: float python_mass_flow_kg_s: float python_cm: float python_to_amesim_cm_ratio: float @dataclass(frozen=True) class TestMqlPnvoEventWindowSampleDiagnostic: time_s: float data_paths: tuple[str, ...] python_values_by_data_path: dict[str, float] amesim_values_by_data_path: dict[str, float] pnl0001_rhs_diagnostics: tuple[TestMqlPnl0001LineRhsDiagnostic, ...] = field( default_factory=tuple ) pnl0001_pressure_loss_diagnostics: tuple[ TestMqlPnl0001PressureLossCalibrationDiagnostic, ... ] = field(default_factory=tuple) pnvo_flow_parameter_diagnostics: tuple[ TestMqlPnvoFlowParameterDiagnostic, ... ] = field(default_factory=tuple) def abs_error(self, data_path: str) -> float: return abs( self.python_values_by_data_path[data_path] - self.amesim_values_by_data_path[data_path] ) @dataclass(frozen=True) class TestMqlPnvoEventWindowDiagnostic: orifice_alias: str event_time_s: float final_time_s: float data_paths: tuple[str, ...] segment_diagnostics: tuple[TestMqlPnvoEventWindowSegmentDiagnostic, ...] sample_diagnostics: tuple[TestMqlPnvoEventWindowSampleDiagnostic, ...] python_values_by_data_path: dict[str, float] amesim_values_by_data_path: dict[str, float] def abs_error(self, data_path: str) -> float: return abs( self.python_values_by_data_path[data_path] - self.amesim_values_by_data_path[data_path] ) @dataclass(frozen=True) class TestMqlFullStateComparisonPathConfig: archive_path: Path = field( default_factory=lambda: Path(__file__).resolve().parents[2] / "AmesimModels" / "test_mql.ame" ) output_dir: Path | None = None @dataclass(frozen=True) class TestMqlFullStateComparisonExecutionConfig: write_summary: bool = True write_comparison_csv: bool = True data_paths: tuple[str, ...] | None = DEFAULT_FULL_STATE_COMPARISON_DATA_PATHS solver: SolveIVPConfig = field( default_factory=lambda: SolveIVPConfig(t_stop=1.0e-2, max_step=1.0e-3) ) t_eval: tuple[float, ...] | None = (0.0, 1.0e-2) inlet_node_pressure_pa: float = 15.31e6 resistance_boundary_pressure_pa: float = 15.29e6 inlet_node_temperature_k: float = 293.15 resistance_boundary_temperature_k: float = 293.15 @dataclass(frozen=True) class TestMqlFullStateComparisonScriptConfig: paths: TestMqlFullStateComparisonPathConfig = field( default_factory=TestMqlFullStateComparisonPathConfig ) execution: TestMqlFullStateComparisonExecutionConfig = field( default_factory=TestMqlFullStateComparisonExecutionConfig ) def _default_output_dir() -> Path: pythonmodels_root = Path(__file__).resolve().parents[1] timestamp = datetime.now(UTC).strftime("test_mql_full_state_%Y%m%d_%H%M%S_%f") return pythonmodels_root / "runs" / timestamp def run_test_mql_full_state_comparison( config: TestMqlFullStateComparisonScriptConfig | None = None, ) -> tuple[TestMqlFullStateComparisonRun, Path]: config = config or TestMqlFullStateComparisonScriptConfig() system = TestMqlSystem(archive_path=config.paths.archive_path) amesim_results = load_test_mql_amesim_results(config.paths.archive_path) output_schema = build_test_mql_output_schema(amesim_results) selected_paths = config.execution.data_paths spec = system.discover_pneumatic_branch_topology().chamber_segment_specs[0] result = system.simulate_full_state_series_from_spec( spec, inlet_node_pressure_pa=config.execution.inlet_node_pressure_pa, resistance_boundary_pressure_pa=( config.execution.resistance_boundary_pressure_pa ), inlet_node_temperature_k=config.execution.inlet_node_temperature_k, resistance_boundary_temperature_k=( config.execution.resistance_boundary_temperature_k ), config=config.execution.solver, t_eval=list(config.execution.t_eval) if config.execution.t_eval is not None else None, data_paths=selected_paths, ) closure = system.full_state_closure_from_spec( spec, inlet_node_pressure_pa=config.execution.inlet_node_pressure_pa, resistance_boundary_pressure_pa=( config.execution.resistance_boundary_pressure_pa ), inlet_node_temperature_k=config.execution.inlet_node_temperature_k, resistance_boundary_temperature_k=( config.execution.resistance_boundary_temperature_k ), ) series_by_data_path = { data_path: result.series[data_path] for data_path in result.series if data_path != "time" } output = validate_test_mql_output( times=result.t, series_by_data_path=series_by_data_path, schema=output_schema, data_paths=selected_paths, ) comparison = compare_validated_test_mql_output( times=output.times, series_by_data_path=output.series_by_data_path, schema=output_schema, amesim_results=amesim_results, data_paths=output.data_paths, ) run = TestMqlFullStateComparisonRun( system=system, closure=closure, amesim_results=amesim_results, output_schema=output_schema, result=result, output=output, comparison=comparison, ) output_dir = config.paths.output_dir or _default_output_dir() if config.execution.write_summary or config.execution.write_comparison_csv: output_dir.mkdir(parents=True, exist_ok=True) if config.execution.write_summary: (output_dir / "test_mql_full_state_comparison_summary.txt").write_text( format_test_mql_full_state_comparison_summary(run), encoding="utf-8", ) if config.execution.write_comparison_csv: write_test_mql_comparison_csv( output_dir=output_dir, python_times=output.times, python_series_by_data_path=output.series_by_data_path, amesim_results=amesim_results, data_paths=output.data_paths, ) return run, output_dir def run_test_mql_pnvo_event_boundary_diagnostic( config: TestMqlFullStateComparisonScriptConfig | None = None, *, orifice_alias: str = "pn_morifice_1", ) -> TestMqlPnvoEventBoundaryDiagnostic: config = config or TestMqlFullStateComparisonScriptConfig() system = TestMqlSystem(archive_path=config.paths.archive_path) control = system.pneumatic_assembly.variable_orifice_controls[orifice_alias] event_time_s = control.step.step_time_s integration_stop_time_s = nextafter(event_time_s, 0.0) solver_template = config.execution.solver solver = SolveIVPConfig( t_start=solver_template.t_start, t_stop=integration_stop_time_s, method=solver_template.method, rtol=solver_template.rtol, atol=solver_template.atol, max_step=solver_template.max_step, ) spec = system.discover_pneumatic_branch_topology().chamber_segment_specs[0] closure_kwargs = { "inlet_node_pressure_pa": config.execution.inlet_node_pressure_pa, "resistance_boundary_pressure_pa": ( config.execution.resistance_boundary_pressure_pa ), "inlet_node_temperature_k": config.execution.inlet_node_temperature_k, "resistance_boundary_temperature_k": ( config.execution.resistance_boundary_temperature_k ), } solution = system.simulate_full_state_from_spec( spec, config=solver, t_eval=[solver.t_start, integration_stop_time_s], **closure_kwargs, ) state_vector = [row[-1] for row in solution.y] closure = system.full_state_closure_from_spec(spec, **closure_kwargs) data_paths = ( "press@pn_c1_8", "dm1@pneumatic_69", f"xv@{orifice_alias}", f"dm2@{orifice_alias}", ) python_values = closure.data_path_values( time_s=event_time_s, state_vector=state_vector, data_paths=data_paths, ) amesim_results = load_test_mql_amesim_results(config.paths.archive_path) amesim_values = { data_path: interpolate_series_value( amesim_results.times, amesim_results.series(data_path), event_time_s, ) for data_path in data_paths } return TestMqlPnvoEventBoundaryDiagnostic( orifice_alias=orifice_alias, event_time_s=event_time_s, integration_stop_time_s=integration_stop_time_s, data_paths=data_paths, python_values_by_data_path=python_values, amesim_values_by_data_path=amesim_values, ) def format_test_mql_pnvo_event_boundary_summary( diagnostic: TestMqlPnvoEventBoundaryDiagnostic, ) -> str: lines = [ "Model: test_mql", f"Mode: PNVO event boundary diagnostic ({diagnostic.orifice_alias})", f"Event time: {diagnostic.event_time_s}", f"Integrated left limit: {diagnostic.integration_stop_time_s}", "Observation side: right-continuous STEP0 opening", ] for data_path in diagnostic.data_paths: lines.append( f" - {data_path}: " f"python={diagnostic.python_values_by_data_path[data_path]}, " f"amesim={diagnostic.amesim_values_by_data_path[data_path]}, " f"abs_error={diagnostic.abs_error(data_path)}" ) return "\n".join(lines) + "\n" def _pnl0001_pressure_loss_calibration_diagnostic( *, closure: object, amesim_results: AmesimResults, python_values_by_data_path: dict[str, float], line_alias: str, chamber_alias: str, time_s: float, ) -> TestMqlPnl0001PressureLossCalibrationDiagnostic: if line_alias != "pneumatic_69" or chamber_alias != "pn_c1_8": raise KeyError(f"Unsupported PNL0001 pressure-loss diagnostic: {line_alias}") line = closure.pneumatic_closure.components.p4_port3_remote_primary_line def amesim_value(data_path: str) -> float: return interpolate_series_value( amesim_results.times, amesim_results.series(data_path), time_s, ) amesim_cm = amesim_value(f"cm@{line_alias}") amesim_dm1_g_s = amesim_value(f"dm1@{line_alias}") amesim_line_gauge_pressure_pa = amesim_value(f"p2@{line_alias}") amesim_chamber_gauge_pressure_pa = amesim_value(f"press@{chamber_alias}") amesim_line_temperature_k = amesim_value(f"t2@{line_alias}") amesim_chamber_temperature_k = amesim_value(f"temp@{chamber_alias}") mass_flow_magnitude_kg_s = abs(amesim_dm1_g_s) * 1.0e-3 amesim_pressure_drop_pa = abs( amesim_line_gauge_pressure_pa - amesim_chamber_gauge_pressure_pa ) current_darcy_pressure_drop_pa = abs( line.darcy_pressure_drop_for_state( mass_flow_kg_s=mass_flow_magnitude_kg_s, pressure_pa=( amesim_line_gauge_pressure_pa + AMESIM_REFERENCE_PRESSURE_PA ), temperature_k=amesim_line_temperature_k, ) ) if current_darcy_pressure_drop_pa > 0.0: pressure_drop_multiplier = ( amesim_pressure_drop_pa / current_darcy_pressure_drop_pa ) else: pressure_drop_multiplier = ( float("inf") if amesim_pressure_drop_pa > 0.0 else 1.0 ) candidate_port1_to_port2_kg_s = line.pn2pipefr_mass_flow( port_1_pressure_pa=( amesim_chamber_gauge_pressure_pa + AMESIM_REFERENCE_PRESSURE_PA ), port_1_temperature_k=amesim_chamber_temperature_k, port_2_pressure_pa=( amesim_line_gauge_pressure_pa + AMESIM_REFERENCE_PRESSURE_PA ), port_2_temperature_k=amesim_line_temperature_k, ) candidate_pn2pipefr_dm1_g_s = -candidate_port1_to_port2_kg_s * 1.0e3 if amesim_dm1_g_s != 0.0: candidate_pn2pipefr_to_amesim_dm1_ratio = ( candidate_pn2pipefr_dm1_g_s / amesim_dm1_g_s ) else: candidate_pn2pipefr_to_amesim_dm1_ratio = ( float("inf") if candidate_pn2pipefr_dm1_g_s != 0.0 else 1.0 ) amesim_linear_conductance = _pnl0001_linear_conductance( dm1_g_s=amesim_dm1_g_s, temperature_k=amesim_line_temperature_k, pressure_drop_pa=amesim_pressure_drop_pa, ) python_pressure_drop_pa = abs( python_values_by_data_path[f"p2@{line_alias}"] - python_values_by_data_path[f"press@{chamber_alias}"] ) python_linear_conductance = _pnl0001_linear_conductance( dm1_g_s=python_values_by_data_path[f"dm1@{line_alias}"], temperature_k=python_values_by_data_path[f"t2@{line_alias}"], pressure_drop_pa=python_pressure_drop_pa, ) if amesim_linear_conductance > 0.0: conductance_ratio = python_linear_conductance / amesim_linear_conductance else: conductance_ratio = ( float("inf") if python_linear_conductance > 0.0 else 1.0 ) return TestMqlPnl0001PressureLossCalibrationDiagnostic( line_alias=line_alias, chamber_alias=chamber_alias, time_s=time_s, amesim_cm=amesim_cm, amesim_dm1_g_s=amesim_dm1_g_s, amesim_mass_flow_magnitude_kg_s=mass_flow_magnitude_kg_s, amesim_line_gauge_pressure_pa=amesim_line_gauge_pressure_pa, amesim_chamber_gauge_pressure_pa=amesim_chamber_gauge_pressure_pa, amesim_line_temperature_k=amesim_line_temperature_k, amesim_pressure_drop_pa=amesim_pressure_drop_pa, current_darcy_pressure_drop_pa=current_darcy_pressure_drop_pa, pressure_drop_multiplier=pressure_drop_multiplier, candidate_pn2pipefr_dm1_g_s=candidate_pn2pipefr_dm1_g_s, candidate_pn2pipefr_to_amesim_dm1_ratio=( candidate_pn2pipefr_to_amesim_dm1_ratio ), amesim_linear_conductance_kg_s_sqrt_k_per_pa=( amesim_linear_conductance ), python_linear_conductance_kg_s_sqrt_k_per_pa=( python_linear_conductance ), python_to_amesim_linear_conductance_ratio=conductance_ratio, ) def _pnl0001_linear_conductance( *, dm1_g_s: float, temperature_k: float, pressure_drop_pa: float, ) -> float: if temperature_k <= 0.0: raise ValueError("temperature_k must be positive") if pressure_drop_pa <= 0.0: return 0.0 return abs(dm1_g_s) * 1.0e-3 * sqrt(temperature_k) / pressure_drop_pa def _pnvo_flow_parameter_diagnostic( *, closure: object, amesim_results: AmesimResults, state_vector: list[float], orifice_alias: str, time_s: float, ) -> TestMqlPnvoFlowParameterDiagnostic: if orifice_alias != "pn_morifice_1": raise KeyError(f"Unsupported PNVO flow parameter diagnostic: {orifice_alias}") snapshot = closure.snapshot_at(time_s, state_vector).pneumatic orifice = closure.pneumatic_closure.components.p4_port3_remote_orifice if orifice.name != orifice_alias: raise KeyError(f"Unexpected PNVO diagnostic orifice: {orifice.name}") line_properties = snapshot.p4_port3_remote_orifice_line_port_2 boundary_properties = snapshot.p4_port3_remote_primary_line upstream_properties = ( line_properties if line_properties.p >= boundary_properties.p else boundary_properties ) python_mass_flow_kg_s = snapshot.p4_port3_remote_orifice_to_node_flow denominator = ( orifice.flow_coefficient * orifice.effective_area * upstream_properties.p ) python_cm = ( abs(python_mass_flow_kg_s) * sqrt(upstream_properties.T) / denominator if denominator > 0.0 and upstream_properties.T > 0.0 else 0.0 ) def amesim_value(data_path: str) -> float: return interpolate_series_value( amesim_results.times, amesim_results.series(data_path), time_s, ) amesim_cm = amesim_value(f"cm@{orifice_alias}") cm_ratio = python_cm / amesim_cm if amesim_cm != 0.0 else 0.0 return TestMqlPnvoFlowParameterDiagnostic( orifice_alias=orifice_alias, time_s=time_s, amesim_cm=amesim_cm, amesim_dm2_g_s=amesim_value(f"dm2@{orifice_alias}"), amesim_opening=amesim_value(f"xv@{orifice_alias}"), amesim_gas_velocity_m_s=amesim_value(f"gasvel@{orifice_alias}"), python_opening=orifice.opening, python_flow_coefficient=orifice.flow_coefficient, python_effective_area_m2=orifice.effective_area, python_line_pressure_pa=line_properties.p, python_boundary_pressure_pa=boundary_properties.p, python_upstream_pressure_pa=upstream_properties.p, python_upstream_temperature_k=upstream_properties.T, python_mass_flow_kg_s=python_mass_flow_kg_s, python_cm=python_cm, python_to_amesim_cm_ratio=cm_ratio, ) def run_test_mql_pnvo_event_window_diagnostic( config: TestMqlFullStateComparisonScriptConfig | None = None, *, orifice_alias: str = "pn_morifice_1", ) -> TestMqlPnvoEventWindowDiagnostic: config = config or TestMqlFullStateComparisonScriptConfig() system = TestMqlSystem(archive_path=config.paths.archive_path) control = system.pneumatic_assembly.variable_orifice_controls[orifice_alias] event_time_s = control.step.step_time_s final_time_s = 0.05 spec = system.discover_pneumatic_branch_topology().chamber_segment_specs[0] closure_kwargs = { "inlet_node_pressure_pa": config.execution.inlet_node_pressure_pa, "resistance_boundary_pressure_pa": ( config.execution.resistance_boundary_pressure_pa ), "inlet_node_temperature_k": config.execution.inlet_node_temperature_k, "resistance_boundary_temperature_k": ( config.execution.resistance_boundary_temperature_k ), } closure = system.full_state_closure_from_spec(spec, **closure_kwargs) state_vector = closure.initial_state_vector() solver_template = config.execution.solver segments = ( SolveIVPConfig( t_start=solver_template.t_start, t_stop=nextafter(event_time_s, 0.0), method=solver_template.method, rtol=solver_template.rtol, atol=solver_template.atol, max_step=solver_template.max_step, ), SolveIVPConfig( t_start=event_time_s, t_stop=0.040001, method="Radau", rtol=solver_template.rtol, atol=solver_template.atol, max_step=1.0e-7, first_step=1.0e-10, ), SolveIVPConfig( t_start=0.040001, t_stop=0.04001, method="Radau", rtol=solver_template.rtol, atol=solver_template.atol, max_step=1.0e-6, ), SolveIVPConfig( t_start=0.04001, t_stop=0.0401, method="Radau", rtol=solver_template.rtol, atol=solver_template.atol, max_step=1.0e-5, ), SolveIVPConfig( t_start=0.0401, t_stop=0.041, method="Radau", rtol=solver_template.rtol, atol=solver_template.atol, max_step=1.0e-5, ), SolveIVPConfig( t_start=0.041, t_stop=0.048, method="BDF", rtol=solver_template.rtol, atol=solver_template.atol, max_step=1.0e-5, ), SolveIVPConfig( t_start=0.048, t_stop=final_time_s, method="BDF", rtol=1.0e-5, atol=1.0e-8, max_step=1.0e-5, ), ) data_paths = ( "press@pn_c1_8", "temp@pn_c1_8", "vol@pn_c1_8", "mgas1@pn_c1_8", "p2@pneumatic_69", "t2@pneumatic_69", "mgas@pneumatic_69", "re@pneumatic_69", "v@pneumatic_69", "ff@pneumatic_69", "dm1@pneumatic_69", f"xv@{orifice_alias}", f"dm2@{orifice_alias}", ) sample_times = (event_time_s, 0.041, 0.042, 0.045, 0.048, final_time_s) state_vector_by_sample_time: dict[float, list[float]] = {} segment_diagnostics: list[TestMqlPnvoEventWindowSegmentDiagnostic] = [] for segment in segments: rhs_evaluations = 0 segment_sample_times = tuple( sample_time for sample_time in sample_times if segment.t_start <= sample_time <= segment.t_stop ) t_eval = tuple( sorted({segment.t_start, segment.t_stop, *segment_sample_times}) ) def rhs(time_s, values): nonlocal rhs_evaluations rhs_evaluations += 1 return closure.rhs_at(time_s, values) solution = integrate_ode( rhs=rhs, initial_state=state_vector, config=segment, t_eval=list(t_eval), ) segment_diagnostics.append( TestMqlPnvoEventWindowSegmentDiagnostic( t_start=segment.t_start, t_stop=segment.t_stop, method=segment.method, rtol=segment.rtol, atol=segment.atol, max_step=segment.max_step, rhs_evaluations=rhs_evaluations, success=bool(solution.success), message=str(solution.message), ) ) for sample_time in segment_sample_times: if sample_time in solution.t: sample_index = list(solution.t).index(sample_time) state_vector_by_sample_time[sample_time] = [ row[sample_index] for row in solution.y ] state_vector = [row[-1] for row in solution.y] if not solution.success: break amesim_results = load_test_mql_amesim_results(config.paths.archive_path) sample_diagnostics = [] for sample_time in sample_times: if sample_time not in state_vector_by_sample_time: continue python_sample_values = closure.data_path_values( time_s=sample_time, state_vector=state_vector_by_sample_time[sample_time], data_paths=data_paths, ) amesim_sample_values = { data_path: interpolate_series_value( amesim_results.times, amesim_results.series(data_path), sample_time, ) for data_path in data_paths } pnl0001_rhs_diagnostics = ( closure.pnl0001_line_rhs_diagnostic( line_alias="pneumatic_69", state_vector=state_vector_by_sample_time[sample_time], time_s=sample_time, ), ) pnl0001_pressure_loss_diagnostics = ( _pnl0001_pressure_loss_calibration_diagnostic( closure=closure, amesim_results=amesim_results, python_values_by_data_path=python_sample_values, line_alias="pneumatic_69", chamber_alias="pn_c1_8", time_s=sample_time, ), ) pnvo_flow_parameter_diagnostics = ( _pnvo_flow_parameter_diagnostic( closure=closure, amesim_results=amesim_results, state_vector=state_vector_by_sample_time[sample_time], orifice_alias=orifice_alias, time_s=sample_time, ), ) sample_diagnostics.append( TestMqlPnvoEventWindowSampleDiagnostic( time_s=sample_time, data_paths=data_paths, python_values_by_data_path=python_sample_values, amesim_values_by_data_path=amesim_sample_values, pnl0001_rhs_diagnostics=pnl0001_rhs_diagnostics, pnl0001_pressure_loss_diagnostics=pnl0001_pressure_loss_diagnostics, pnvo_flow_parameter_diagnostics=pnvo_flow_parameter_diagnostics, ) ) python_values = closure.data_path_values( time_s=final_time_s, state_vector=state_vector, data_paths=data_paths, ) amesim_values = { data_path: interpolate_series_value( amesim_results.times, amesim_results.series(data_path), final_time_s, ) for data_path in data_paths } return TestMqlPnvoEventWindowDiagnostic( orifice_alias=orifice_alias, event_time_s=event_time_s, final_time_s=final_time_s, data_paths=data_paths, segment_diagnostics=tuple(segment_diagnostics), sample_diagnostics=tuple(sample_diagnostics), python_values_by_data_path=python_values, amesim_values_by_data_path=amesim_values, ) def format_test_mql_pnvo_event_window_summary( diagnostic: TestMqlPnvoEventWindowDiagnostic, ) -> str: lines = [ "Model: test_mql", f"Mode: PNVO event window diagnostic ({diagnostic.orifice_alias})", f"Event time: {diagnostic.event_time_s}", f"Final time: {diagnostic.final_time_s}", "Segments:", ] for segment in diagnostic.segment_diagnostics: lines.append( f" - {segment.t_start} -> {segment.t_stop}: " f"method={segment.method}, rtol={segment.rtol}, " f"atol={segment.atol}, max_step={segment.max_step}, " f"rhs={segment.rhs_evaluations}, success={segment.success}" ) if diagnostic.sample_diagnostics: lines.append("Sample comparisons:") for sample in diagnostic.sample_diagnostics: lines.append(f" t={sample.time_s}") for data_path in sample.data_paths: lines.append( f" - {data_path}: " f"python={sample.python_values_by_data_path[data_path]}, " f"amesim={sample.amesim_values_by_data_path[data_path]}, " f"abs_error={sample.abs_error(data_path)}" ) for rhs in sample.pnl0001_rhs_diagnostics: lines.append( f" - rhs@{rhs.line_alias}: " f"chamber_to_line={rhs.chamber_to_line_flow_kg_s}, " f"node_to_line={rhs.node_to_line_flow_kg_s}, " f"dm_dt={rhs.mass_derivative_kg_s}, " f"dU_dt={rhs.energy_derivative_w}" ) for pressure_loss in sample.pnl0001_pressure_loss_diagnostics: lines.append( f" - pressure_loss@{pressure_loss.line_alias}: " f"cm={pressure_loss.amesim_cm}, " f"amesim_dm1={pressure_loss.amesim_dm1_g_s}, " f"amesim_dp={pressure_loss.amesim_pressure_drop_pa}, " f"darcy_dp={pressure_loss.current_darcy_pressure_drop_pa}, " f"dp_multiplier={pressure_loss.pressure_drop_multiplier}, " f"candidate_pn2pipefr_dm1=" f"{pressure_loss.candidate_pn2pipefr_dm1_g_s}, " f"candidate_pn2pipefr_ratio=" f"{pressure_loss.candidate_pn2pipefr_to_amesim_dm1_ratio}, " f"amesim_linear_k=" f"{pressure_loss.amesim_linear_conductance_kg_s_sqrt_k_per_pa}, " f"python_linear_k=" f"{pressure_loss.python_linear_conductance_kg_s_sqrt_k_per_pa}, " f"linear_k_ratio=" f"{pressure_loss.python_to_amesim_linear_conductance_ratio}" ) for flow_parameter in sample.pnvo_flow_parameter_diagnostics: lines.append( f" - flow_parameter@{flow_parameter.orifice_alias}: " f"amesim_cm={flow_parameter.amesim_cm}, " f"python_cm={flow_parameter.python_cm}, " f"cm_ratio={flow_parameter.python_to_amesim_cm_ratio}, " f"amesim_dm2={flow_parameter.amesim_dm2_g_s}, " f"python_m={flow_parameter.python_mass_flow_kg_s}, " f"opening={flow_parameter.python_opening}, " f"effective_area={flow_parameter.python_effective_area_m2}, " f"upstream_p={flow_parameter.python_upstream_pressure_pa}, " f"upstream_t={flow_parameter.python_upstream_temperature_k}, " f"amesim_gasvel={flow_parameter.amesim_gas_velocity_m_s}" ) lines.append("Final comparison:") for data_path in diagnostic.data_paths: lines.append( f" - {data_path}: " f"python={diagnostic.python_values_by_data_path[data_path]}, " f"amesim={diagnostic.amesim_values_by_data_path[data_path]}, " f"abs_error={diagnostic.abs_error(data_path)}" ) return "\n".join(lines) + "\n" def format_test_mql_full_state_comparison_summary( run: TestMqlFullStateComparisonRun, ) -> str: lines = [ "Model: test_mql", "Mode: Python 132 full-state closure comparison", f"Samples: {run.sample_count}", f"Compared signals: {run.signal_count}", f"Output schema signals: {run.output_schema.signal_count}", f"AMESim first saved interval: " f"{run.amesim_results.times[1] - run.amesim_results.times[0]}", f"Final time aligns with AMESim sample: " f"{_matches_amesim_sample(run, run.output.times[-1])}", f"Max absolute error: {run.comparison.max_abs_error}", f"Max relative error: {run.comparison.max_rel_error}", ] largest_metric = run.largest_abs_error_metric if largest_metric is not None: lines.append( "Largest absolute error: " f"{largest_metric.data_path}={largest_metric.max_abs_error}" ) largest_diagnostic = run.largest_final_abs_error_diagnostic if largest_diagnostic is not None: lines.append( "Largest final endpoint error: " f"{largest_diagnostic.data_path}={largest_diagnostic.final_abs_error}" ) lines.append("Metrics by max absolute error:") for metric in run.metrics_by_max_abs_error(): lines.append( f" - {metric.data_path}: max_abs_error={metric.max_abs_error}, " f"final_abs_error={metric.final_abs_error}" ) lines.append("Endpoint diagnostics by final absolute error:") for diagnostic in run.diagnostics_by_final_abs_error(): lines.append( f" - {diagnostic.data_path}: " f"initial_python={diagnostic.initial_python_value}, " f"initial_amesim={diagnostic.initial_amesim_value}, " f"final_python={diagnostic.final_python_value}, " f"final_amesim={diagnostic.final_amesim_value}, " f"final_abs_error={diagnostic.final_abs_error}" ) flow_diagnostic = _pneumatic_69_flow_diagnostic(run) if flow_diagnostic is not None: lines.append( "PNL0001 canonical mass-flow diagnostic: " f"{flow_diagnostic.data_path}" ) lines.extend( [ " - convention=Python chamber-to-line flow " "equals -AMESim dm1 * 1e-3", f" - initial_python_kg_s=" f"{flow_diagnostic.initial_python_canonical_kg_s}", f" - initial_amesim_kg_s=" f"{flow_diagnostic.initial_amesim_canonical_kg_s}", f" - final_python_kg_s=" f"{flow_diagnostic.final_python_canonical_kg_s}", f" - final_amesim_kg_s=" f"{flow_diagnostic.final_amesim_canonical_kg_s}", f" - final_abs_error_kg_s=" f"{flow_diagnostic.final_canonical_abs_error_kg_s}", ] ) pnvo_diagnostic = _pn_morifice_1_diagnostic(run) if pnvo_diagnostic is not None: lines.append(f"PNVO diagnostic: {pnvo_diagnostic.alias}") lines.extend( [ f" - initial_python_opening=" f"{pnvo_diagnostic.initial_python_opening}", f" - initial_amesim_opening=" f"{pnvo_diagnostic.initial_amesim_opening}", f" - final_python_opening=" f"{pnvo_diagnostic.final_python_opening}", f" - final_amesim_opening=" f"{pnvo_diagnostic.final_amesim_opening}", f" - final_python_mass_flow_kg_s=" f"{pnvo_diagnostic.final_python_mass_flow_kg_s}", f" - final_amesim_mass_flow_kg_s=" f"{pnvo_diagnostic.final_amesim_mass_flow_kg_s}", f" - final_mass_flow_abs_error_kg_s=" f"{pnvo_diagnostic.final_mass_flow_abs_error_kg_s}", ] ) chamber_diagnostic = _largest_chamber_rhs_diagnostic(run) if chamber_diagnostic is not None: lines.append( "Largest endpoint chamber RHS breakdown: " f"{chamber_diagnostic.chamber_alias}" ) lines.extend( [ f" - piston_alias={chamber_diagnostic.piston_alias}", f" - pressure_pa={chamber_diagnostic.chamber_pressure_pa}", f" - volume_m3={chamber_diagnostic.chamber_volume_m3}", f" - volume_rate_m3_s={chamber_diagnostic.chamber_volume_rate_m3_s}", f" - mass_derivative_kg_s={chamber_diagnostic.mass_derivative_kg_s}", f" - port_a_energy_flow_w={chamber_diagnostic.port_a_energy_flow_w}", f" - boundary_work_w={chamber_diagnostic.boundary_work_w}", f" - thermal_energy_flow_w=" f"{chamber_diagnostic.thermal_energy_flow_w}", f" - energy_derivative_w={chamber_diagnostic.energy_derivative_w}", ] ) return "\n".join(lines) + "\n" def _matches_amesim_sample( run: TestMqlFullStateComparisonRun, time_s: float, ) -> bool: return any( abs(float(sample_time) - float(time_s)) <= 1.0e-12 for sample_time in run.amesim_results.times ) def _pneumatic_69_flow_diagnostic(run: TestMqlFullStateComparisonRun): try: return run.pnl0001_mass_flow_diagnostic() except KeyError: return None def _pn_morifice_1_diagnostic(run: TestMqlFullStateComparisonRun): try: return run.pnvo_diagnostic() except KeyError: return None def _largest_chamber_rhs_diagnostic(run: TestMqlFullStateComparisonRun): diagnostic = run.largest_final_abs_error_diagnostic if diagnostic is None or "@" not in diagnostic.data_path: return None _signal, alias = diagnostic.data_path.split("@", 1) try: return run.chamber_rhs_diagnostic(alias) except KeyError: return None def main() -> None: parser = argparse.ArgumentParser() parser.add_argument( "--pnvo-event-boundary", action="store_true", help="compare the STEP0 left-state/right-opening boundary at t=0.04 s", ) parser.add_argument( "--pnvo-event-window", action="store_true", help="run the segmented PNVO opening window through the t=0.05 s save point", ) args = parser.parse_args() if args.pnvo_event_boundary: diagnostic = run_test_mql_pnvo_event_boundary_diagnostic() print(format_test_mql_pnvo_event_boundary_summary(diagnostic), end="") return if args.pnvo_event_window: diagnostic = run_test_mql_pnvo_event_window_diagnostic() print(format_test_mql_pnvo_event_window_summary(diagnostic), end="") return run, output_dir = run_test_mql_full_state_comparison() print(format_test_mql_full_state_comparison_summary(run), end="") print(f"Output directory: {output_dir}") if __name__ == "__main__": main()