from __future__ import annotations from dataclasses import dataclass, field from datetime import UTC, datetime from pathlib import Path from PythonModels.core.solver import SolveIVPConfig 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 TestMqlSimulationResult, TestMqlSystem 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", ) @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 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 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 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-5, max_step=1.0e-6) ) t_eval: tuple[float, ...] | None = (0.0, 1.0e-5) 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 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"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}", ] ) 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" - energy_derivative_w={chamber_diagnostic.energy_derivative_w}", ] ) return "\n".join(lines) + "\n" def _pneumatic_69_flow_diagnostic(run: TestMqlFullStateComparisonRun): try: return run.pnl0001_mass_flow_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: 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()