Files
SystemSimulationApp/PythonModels/scripts/run_test_mql_full_state_comparison.py
T

1193 lines
44 KiB
Python

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
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}")
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
)
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,
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"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()