lujingze
e18399c022
同步远端 PNL0003 诊断和大采样网格能力，语义合并活动感知的 60 秒真停滞判定与旧后端 15 分钟兼容兜底。 纳管热路径优化、15 单元运行证据、浏览器与 API 报告，并补充北京时间更新日志和遗留问题。
891 lines
32 KiB
Python
891 lines
32 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from app.main import compile_reactflow_network
|
|
from app.simulation.components.amesim.flow.orifices import (
|
|
AmesimPnvo001SignalOpening,
|
|
)
|
|
from app.simulation.components.amesim.flow.pipes import AmesimPnl0001
|
|
from app.simulation.registry import ComponentModelSpec, get_component_model_spec
|
|
from app.simulation.solvers.algebraic import (
|
|
CAUSAL_COORDINATE_KERNEL_ENVIRONMENT_VARIABLE,
|
|
CAUSAL_DIRECT_EQUATION_READERS_ENVIRONMENT_VARIABLE,
|
|
CAUSAL_DIRECT_SUM_ASSIGNMENTS_ENVIRONMENT_VARIABLE,
|
|
CAUSAL_EXECUTOR_V2_ENVIRONMENT_VARIABLE,
|
|
CAUSAL_FAST_PATH_ENVIRONMENT_VARIABLE,
|
|
PressureFlowSolver,
|
|
)
|
|
from app.simulation.systems.generic import GenericFluidSystem
|
|
from tests.test_amesim_mechanical_xml import zero_force_mass_project
|
|
from tests.test_amesim_pnl0001_xml import amesim_pnl0001_project
|
|
from tests.test_amesim_pnvo001_signal_xml import (
|
|
high_pressure_helium_step_project,
|
|
)
|
|
from tests.test_generic_system_xml_simulation import chain_project
|
|
|
|
|
|
class _UntrustedPnvo001(AmesimPnvo001SignalOpening):
|
|
# A module-name check alone would incorrectly trust this changed subclass.
|
|
__module__ = AmesimPnvo001SignalOpening.__module__
|
|
pressure_flow_callback_count = 0
|
|
|
|
def pressure_flow_equation_values(self) -> tuple[float, ...]:
|
|
self.pressure_flow_callback_count += 1
|
|
return super().pressure_flow_equation_values()
|
|
|
|
|
|
class _InheritedDirectReaderPnl0001(AmesimPnl0001):
|
|
"""A changed concrete class that does not repeat the scalar promise."""
|
|
|
|
|
|
class _NonfiniteDirectReaderPnl0001(AmesimPnl0001):
|
|
__module__ = AmesimPnl0001.__module__
|
|
|
|
def __init__(self, *args, **kwargs) -> None:
|
|
super().__init__(*args, **kwargs)
|
|
self.full_equation_evaluation_count = 0
|
|
self.direct_equation_evaluation_count = 0
|
|
|
|
def pressure_flow_equation_values(self) -> tuple[float, ...]:
|
|
self.full_equation_evaluation_count += 1
|
|
return super().pressure_flow_equation_values()
|
|
|
|
def pressure_flow_equation_value_readers(self):
|
|
def nonfinite_pressure_state_residual() -> float:
|
|
self.direct_equation_evaluation_count += 1
|
|
return float("nan")
|
|
|
|
return {
|
|
f"{self.name}:port_2_pressure_state": (
|
|
nonfinite_pressure_state_residual
|
|
),
|
|
}
|
|
|
|
|
|
def _system(
|
|
project,
|
|
*,
|
|
executor_v2: bool | None = False,
|
|
coordinate_kernel: bool | None = None,
|
|
direct_sum_assignments: bool | None = None,
|
|
direct_equation_readers: bool | None = None,
|
|
) -> GenericFluidSystem:
|
|
environment = {}
|
|
if executor_v2 is not None:
|
|
environment[CAUSAL_EXECUTOR_V2_ENVIRONMENT_VARIABLE] = (
|
|
"1" if executor_v2 else "0"
|
|
)
|
|
if coordinate_kernel is not None:
|
|
environment[CAUSAL_COORDINATE_KERNEL_ENVIRONMENT_VARIABLE] = (
|
|
"1" if coordinate_kernel else "0"
|
|
)
|
|
if direct_sum_assignments is not None:
|
|
environment[CAUSAL_DIRECT_SUM_ASSIGNMENTS_ENVIRONMENT_VARIABLE] = (
|
|
"1" if direct_sum_assignments else "0"
|
|
)
|
|
if direct_equation_readers is not None:
|
|
environment[CAUSAL_DIRECT_EQUATION_READERS_ENVIRONMENT_VARIABLE] = (
|
|
"1" if direct_equation_readers else "0"
|
|
)
|
|
with patch.dict(
|
|
os.environ,
|
|
environment,
|
|
):
|
|
if executor_v2 is None:
|
|
os.environ.pop(CAUSAL_EXECUTOR_V2_ENVIRONMENT_VARIABLE, None)
|
|
if coordinate_kernel is None:
|
|
os.environ.pop(
|
|
CAUSAL_COORDINATE_KERNEL_ENVIRONMENT_VARIABLE,
|
|
None,
|
|
)
|
|
if direct_sum_assignments is None:
|
|
os.environ.pop(
|
|
CAUSAL_DIRECT_SUM_ASSIGNMENTS_ENVIRONMENT_VARIABLE,
|
|
None,
|
|
)
|
|
if direct_equation_readers is None:
|
|
os.environ.pop(
|
|
CAUSAL_DIRECT_EQUATION_READERS_ENVIRONMENT_VARIABLE,
|
|
None,
|
|
)
|
|
return GenericFluidSystem(compile_reactflow_network(project))
|
|
|
|
|
|
def _system_with_pnl0001_class(component_class, **options) -> GenericFluidSystem:
|
|
original_spec = get_component_model_spec("amesim_pnl0001")
|
|
custom_spec = ComponentModelSpec(
|
|
component_class=component_class,
|
|
library=original_spec.library,
|
|
)
|
|
|
|
def custom_spec_lookup(model_type: str):
|
|
if model_type == "amesim_pnl0001":
|
|
return custom_spec
|
|
return get_component_model_spec(model_type)
|
|
|
|
with patch(
|
|
"app.simulation.registry.get_component_model_spec",
|
|
side_effect=custom_spec_lookup,
|
|
):
|
|
return _system(amesim_pnl0001_project(), **options)
|
|
|
|
|
|
class PressureFlowCausalExecutionTests(unittest.TestCase):
|
|
def test_direct_equation_reader_is_default_on_and_bitwise_optional(
|
|
self,
|
|
) -> None:
|
|
direct = _system(
|
|
amesim_pnl0001_project(),
|
|
executor_v2=True,
|
|
direct_equation_readers=True,
|
|
)
|
|
callback = _system(
|
|
amesim_pnl0001_project(),
|
|
executor_v2=True,
|
|
direct_equation_readers=False,
|
|
)
|
|
direct_state = direct.initial_state_vector()
|
|
callback_state = callback.initial_state_vector()
|
|
|
|
for time in (0.0, 0.001):
|
|
self.assertEqual(
|
|
direct.rhs(time, direct_state),
|
|
callback.rhs(time, callback_state),
|
|
)
|
|
self.assertEqual(
|
|
tuple(
|
|
unknown.read()
|
|
for unknown in direct.pressure_flow_solver.unknowns
|
|
),
|
|
tuple(
|
|
unknown.read()
|
|
for unknown in callback.pressure_flow_solver.unknowns
|
|
),
|
|
)
|
|
|
|
direct_diagnostics = (
|
|
direct.pressure_flow_solver.causal_execution_diagnostics()
|
|
)
|
|
callback_diagnostics = (
|
|
callback.pressure_flow_solver.causal_execution_diagnostics()
|
|
)
|
|
self.assertTrue(direct_diagnostics["directEquationReadersConfigured"])
|
|
self.assertEqual(direct_diagnostics["directEffortAnchorCount"], 1)
|
|
self.assertFalse(
|
|
callback_diagnostics["directEquationReadersConfigured"]
|
|
)
|
|
self.assertEqual(callback_diagnostics["directEffortAnchorCount"], 0)
|
|
|
|
def test_inherited_direct_equation_reader_does_not_opt_in_subclass(
|
|
self,
|
|
) -> None:
|
|
system = _system_with_pnl0001_class(
|
|
_InheritedDirectReaderPnl0001,
|
|
direct_equation_readers=True,
|
|
)
|
|
pipe = system.network.components["pnl_1"]
|
|
solver = system.pressure_flow_solver
|
|
|
|
self.assertIsInstance(pipe, _InheritedDirectReaderPnl0001)
|
|
self.assertEqual(
|
|
solver.causal_execution_diagnostics()["directEffortAnchorCount"],
|
|
0,
|
|
)
|
|
anchor = next(
|
|
anchor
|
|
for group in solver._effort_groups["p"]
|
|
for anchor in group.anchors
|
|
if anchor.equation_id == "pnl_1:port_2_pressure_state"
|
|
)
|
|
self.assertIsNone(anchor.causal_evaluate)
|
|
|
|
def test_nonfinite_direct_effort_reader_uses_authoritative_fallback(
|
|
self,
|
|
) -> None:
|
|
system = _system_with_pnl0001_class(
|
|
_NonfiniteDirectReaderPnl0001,
|
|
executor_v2=True,
|
|
direct_equation_readers=True,
|
|
)
|
|
pipe = system.network.components["pnl_1"]
|
|
solver = system.pressure_flow_solver
|
|
self.assertIsInstance(pipe, _NonfiniteDirectReaderPnl0001)
|
|
self.assertEqual(
|
|
solver.causal_execution_diagnostics()["directEffortAnchorCount"],
|
|
1,
|
|
)
|
|
|
|
derivative = system.rhs(0.0, system.initial_state_vector())
|
|
|
|
self.assertTrue(all(value == value for value in derivative))
|
|
diagnostics = solver.causal_execution_diagnostics()
|
|
self.assertFalse(diagnostics["enabled"])
|
|
self.assertEqual(
|
|
diagnostics["disabledReason"],
|
|
"nonFiniteCausalEffortAnchor",
|
|
)
|
|
self.assertGreaterEqual(diagnostics["legacyFallbackCount"], 1)
|
|
self.assertGreater(pipe.direct_equation_evaluation_count, 0)
|
|
self.assertGreater(pipe.full_equation_evaluation_count, 0)
|
|
|
|
pipe.direct_equation_evaluation_count = 0
|
|
pipe.full_equation_evaluation_count = 0
|
|
solver.propagate_equal_efforts(("p",))
|
|
|
|
self.assertEqual(pipe.direct_equation_evaluation_count, 0)
|
|
self.assertEqual(pipe.full_equation_evaluation_count, 1)
|
|
|
|
def test_direct_sum_keeps_same_module_subclass_on_callback_plan(self) -> None:
|
|
original_spec = get_component_model_spec("amesim_pnvo001")
|
|
custom_spec = ComponentModelSpec(
|
|
component_class=_UntrustedPnvo001,
|
|
library=original_spec.library,
|
|
)
|
|
|
|
def custom_spec_lookup(model_type: str):
|
|
if model_type == "amesim_pnvo001":
|
|
return custom_spec
|
|
return get_component_model_spec(model_type)
|
|
|
|
with patch(
|
|
"app.simulation.registry.get_component_model_spec",
|
|
side_effect=custom_spec_lookup,
|
|
):
|
|
system = _system(
|
|
high_pressure_helium_step_project(),
|
|
direct_sum_assignments=True,
|
|
)
|
|
|
|
solver = system.pressure_flow_solver
|
|
valve = system.network.components["valve_1"]
|
|
self.assertIsInstance(valve, _UntrustedPnvo001)
|
|
assignment = next(
|
|
assignment
|
|
for stage in solver._explicit_flow_plan
|
|
for assignment in stage.assignments
|
|
if assignment.equation_id == "valve_1:mass_flow_balance"
|
|
)
|
|
self.assertIsNone(assignment.evaluate)
|
|
self.assertIs(assignment.component, valve)
|
|
self.assertEqual(
|
|
solver.causal_execution_diagnostics()[
|
|
"directSumFlowAssignmentCount"
|
|
],
|
|
0,
|
|
)
|
|
|
|
valve.pressure_flow_callback_count = 0
|
|
system.rhs(0.041, system.initial_state_vector())
|
|
self.assertGreater(valve.pressure_flow_callback_count, 0)
|
|
|
|
def test_direct_sum_assignments_match_disabled_path_bitwise(self) -> None:
|
|
direct = _system(
|
|
high_pressure_helium_step_project(),
|
|
direct_sum_assignments=True,
|
|
)
|
|
callback = _system(
|
|
high_pressure_helium_step_project(),
|
|
direct_sum_assignments=False,
|
|
)
|
|
direct_state = direct.initial_state_vector()
|
|
callback_state = callback.initial_state_vector()
|
|
|
|
for time in (0.0, 0.041, 0.8):
|
|
self.assertEqual(
|
|
direct.rhs(time, direct_state),
|
|
callback.rhs(time, callback_state),
|
|
)
|
|
self.assertEqual(
|
|
tuple(
|
|
unknown.read()
|
|
for unknown in direct.pressure_flow_solver.unknowns
|
|
),
|
|
tuple(
|
|
unknown.read()
|
|
for unknown in callback.pressure_flow_solver.unknowns
|
|
),
|
|
)
|
|
|
|
direct_diagnostics = (
|
|
direct.pressure_flow_solver.causal_execution_diagnostics()
|
|
)
|
|
callback_diagnostics = (
|
|
callback.pressure_flow_solver.causal_execution_diagnostics()
|
|
)
|
|
self.assertTrue(
|
|
direct_diagnostics["directSumAssignmentsConfigured"]
|
|
)
|
|
self.assertFalse(
|
|
callback_diagnostics["directSumAssignmentsConfigured"]
|
|
)
|
|
self.assertGreater(
|
|
direct_diagnostics["directSumFlowAssignmentCount"],
|
|
0,
|
|
)
|
|
self.assertEqual(
|
|
callback_diagnostics["directSumFlowAssignmentCount"],
|
|
0,
|
|
)
|
|
self.assertEqual(
|
|
direct_diagnostics["compiledFlowAssignmentCount"],
|
|
callback_diagnostics["compiledFlowAssignmentCount"],
|
|
)
|
|
self.assertLess(
|
|
direct_diagnostics["directSumFlowAssignmentCount"],
|
|
direct_diagnostics["compiledFlowAssignmentCount"],
|
|
)
|
|
for key in (
|
|
"fastSolveCount",
|
|
"fullResidualAuditCount",
|
|
"auditFailureCount",
|
|
"legacyFallbackCount",
|
|
):
|
|
self.assertEqual(direct_diagnostics[key], callback_diagnostics[key])
|
|
self.assertGreater(direct_diagnostics["fastSolveCount"], 0)
|
|
self.assertEqual(direct_diagnostics["auditFailureCount"], 0)
|
|
self.assertEqual(direct_diagnostics["legacyFallbackCount"], 0)
|
|
|
|
def test_direct_sum_reader_rejection_uses_safe_component_fallback(
|
|
self,
|
|
) -> None:
|
|
callback = _system(
|
|
high_pressure_helium_step_project(),
|
|
direct_sum_assignments=False,
|
|
)
|
|
with patch.object(
|
|
PressureFlowSolver,
|
|
"_sum_to_zero_flow_target_reader",
|
|
side_effect=ValueError("synthetic unsupported sum"),
|
|
):
|
|
fallback = _system(
|
|
high_pressure_helium_step_project(),
|
|
direct_sum_assignments=True,
|
|
)
|
|
|
|
callback_state = callback.initial_state_vector()
|
|
fallback_state = fallback.initial_state_vector()
|
|
for time in (0.0, 0.041, 0.8):
|
|
self.assertEqual(
|
|
fallback.rhs(time, fallback_state),
|
|
callback.rhs(time, callback_state),
|
|
)
|
|
self.assertEqual(
|
|
tuple(
|
|
unknown.read()
|
|
for unknown in fallback.pressure_flow_solver.unknowns
|
|
),
|
|
tuple(
|
|
unknown.read()
|
|
for unknown in callback.pressure_flow_solver.unknowns
|
|
),
|
|
)
|
|
|
|
diagnostics = (
|
|
fallback.pressure_flow_solver.causal_execution_diagnostics()
|
|
)
|
|
self.assertTrue(diagnostics["directSumAssignmentsConfigured"])
|
|
self.assertEqual(diagnostics["directSumFlowAssignmentCount"], 0)
|
|
self.assertTrue(diagnostics["eligible"])
|
|
self.assertTrue(diagnostics["enabled"])
|
|
self.assertGreater(diagnostics["fastSolveCount"], 0)
|
|
self.assertEqual(diagnostics["auditFailureCount"], 0)
|
|
self.assertEqual(diagnostics["legacyFallbackCount"], 0)
|
|
|
|
def test_compiled_v2_is_enabled_by_default_and_can_be_disabled(self) -> None:
|
|
default = _system(zero_force_mass_project(), executor_v2=None)
|
|
disabled = _system(zero_force_mass_project(), executor_v2=False)
|
|
enabled = _system(zero_force_mass_project(), executor_v2=True)
|
|
|
|
self.assertTrue(
|
|
default.pressure_flow_solver.causal_execution_diagnostics()[
|
|
"executorV2Configured"
|
|
]
|
|
)
|
|
self.assertFalse(
|
|
disabled.pressure_flow_solver.causal_execution_diagnostics()[
|
|
"executorV2Configured"
|
|
]
|
|
)
|
|
self.assertTrue(
|
|
enabled.pressure_flow_solver.causal_execution_diagnostics()[
|
|
"executorV2Configured"
|
|
]
|
|
)
|
|
|
|
def test_coordinate_kernel_is_default_on_and_independently_disabled(
|
|
self,
|
|
) -> None:
|
|
default = _system(
|
|
zero_force_mass_project(),
|
|
executor_v2=True,
|
|
)
|
|
disabled = _system(
|
|
zero_force_mass_project(),
|
|
executor_v2=True,
|
|
coordinate_kernel=False,
|
|
)
|
|
|
|
default_diagnostics = (
|
|
default.pressure_flow_solver.causal_execution_diagnostics()
|
|
)
|
|
disabled_diagnostics = (
|
|
disabled.pressure_flow_solver.causal_execution_diagnostics()
|
|
)
|
|
self.assertTrue(default_diagnostics["coordinateKernelConfigured"])
|
|
self.assertTrue(default_diagnostics["coordinateKernelEnabled"])
|
|
self.assertFalse(disabled_diagnostics["coordinateKernelConfigured"])
|
|
self.assertFalse(disabled_diagnostics["coordinateKernelEnabled"])
|
|
|
|
def test_compiled_v2_matches_v1_bitwise_without_target_names(self) -> None:
|
|
compiled = _system(
|
|
high_pressure_helium_step_project(),
|
|
executor_v2=True,
|
|
)
|
|
v1 = _system(high_pressure_helium_step_project())
|
|
compiled_state = compiled.initial_state_vector()
|
|
v1_state = v1.initial_state_vector()
|
|
|
|
for time in (0.0, 0.041, 0.8):
|
|
self.assertEqual(
|
|
compiled.rhs(time, compiled_state),
|
|
v1.rhs(time, v1_state),
|
|
)
|
|
self.assertEqual(
|
|
tuple(
|
|
unknown.read()
|
|
for unknown in compiled.pressure_flow_solver.unknowns
|
|
),
|
|
tuple(
|
|
unknown.read()
|
|
for unknown in v1.pressure_flow_solver.unknowns
|
|
),
|
|
)
|
|
|
|
diagnostics = (
|
|
compiled.pressure_flow_solver.causal_execution_diagnostics()
|
|
)
|
|
self.assertGreater(diagnostics["executorV2FastSolveCount"], 0)
|
|
self.assertGreater(diagnostics["coordinateKernelFastSolveCount"], 0)
|
|
self.assertEqual(
|
|
diagnostics["compiledAssignmentCount"],
|
|
len(compiled.pressure_flow_solver.unknowns),
|
|
)
|
|
self.assertEqual(
|
|
diagnostics["executorV2RuntimeValidationFailureCount"],
|
|
0,
|
|
)
|
|
self.assertEqual(
|
|
diagnostics["canonicalCoordinateCount"],
|
|
diagnostics["logicalEffortCoordinateCount"]
|
|
+ diagnostics["compiledFlowAssignmentCount"],
|
|
)
|
|
self.assertEqual(
|
|
diagnostics["eliminatedEffortAliasCount"],
|
|
diagnostics["compiledEffortUnknownCount"]
|
|
- diagnostics["logicalEffortCoordinateCount"],
|
|
)
|
|
|
|
def test_compiled_v2_fast_solve_skips_legacy_seed_scan_and_scales(
|
|
self,
|
|
) -> None:
|
|
system = _system(zero_force_mass_project(), executor_v2=True)
|
|
state = system.initial_state_vector()
|
|
system.rhs(0.0, state)
|
|
solver = system.pressure_flow_solver
|
|
|
|
with patch.object(
|
|
solver,
|
|
"_solve_explicit_flow_unknowns",
|
|
wraps=solver._solve_explicit_flow_unknowns,
|
|
) as legacy_seed, patch.object(
|
|
solver,
|
|
"_pressure_flow_equation_values",
|
|
wraps=solver._pressure_flow_equation_values,
|
|
) as residuals, patch.object(
|
|
solver,
|
|
"_scales",
|
|
wraps=solver._scales,
|
|
) as scales, patch.object(
|
|
solver,
|
|
"_evaluate_explicit_flow_stage",
|
|
wraps=solver._evaluate_explicit_flow_stage,
|
|
) as allocating_stage:
|
|
diagnostics = solver.solve(effort_variables=("p",))
|
|
|
|
legacy_seed.assert_not_called()
|
|
residuals.assert_not_called()
|
|
scales.assert_not_called()
|
|
allocating_stage.assert_not_called()
|
|
self.assertIs(
|
|
diagnostics,
|
|
solver._causal_cached_fast_diagnostics,
|
|
)
|
|
self.assertTrue(diagnostics.causal_fast_path_used)
|
|
|
|
def test_compiled_v2_nonfinite_assignment_fuses_and_audits_same_solve(
|
|
self,
|
|
) -> None:
|
|
system = _system(zero_force_mass_project(), executor_v2=True)
|
|
state = system.initial_state_vector()
|
|
system.rhs(0.0, state)
|
|
solver = system.pressure_flow_solver
|
|
with patch.object(
|
|
solver,
|
|
"_execute_causal_coordinate_flow_plan",
|
|
return_value="nonFiniteCausalFlowAssignment",
|
|
):
|
|
diagnostics = solver.solve(effort_variables=("p",))
|
|
|
|
self.assertTrue(diagnostics.success)
|
|
self.assertTrue(diagnostics.residual_verified_this_solve)
|
|
execution = solver.causal_execution_diagnostics()
|
|
self.assertFalse(execution["enabled"])
|
|
self.assertEqual(
|
|
execution["disabledReason"],
|
|
"nonFiniteCausalFlowAssignment",
|
|
)
|
|
self.assertEqual(execution["executorV2RuntimeValidationFailureCount"], 1)
|
|
self.assertEqual(execution["legacyFallbackCount"], 1)
|
|
|
|
def test_compiled_v2_rejects_nonfinite_external_mechanical_effort(
|
|
self,
|
|
) -> None:
|
|
system = _system(zero_force_mass_project(), executor_v2=True)
|
|
state = system.initial_state_vector()
|
|
system.rhs(0.0, state)
|
|
solver = system.pressure_flow_solver
|
|
velocity = next(
|
|
unknown for unknown in solver.unknowns if unknown.variable == "v"
|
|
)
|
|
velocity.write(float("nan"))
|
|
|
|
with self.assertRaises(ValueError):
|
|
solver.solve(effort_variables=("p",))
|
|
|
|
execution = solver.causal_execution_diagnostics()
|
|
self.assertFalse(execution["enabled"])
|
|
self.assertEqual(
|
|
execution["disabledReason"],
|
|
"nonFiniteCausalExternalEffort",
|
|
)
|
|
self.assertEqual(execution["executorV2RuntimeValidationFailureCount"], 1)
|
|
self.assertEqual(execution["legacyFallbackCount"], 1)
|
|
|
|
def test_compiled_v2_stage_error_fuses_to_existing_fallback(self) -> None:
|
|
system = _system(zero_force_mass_project(), executor_v2=True)
|
|
state = system.initial_state_vector()
|
|
system.rhs(0.0, state)
|
|
solver = system.pressure_flow_solver
|
|
with patch.object(
|
|
solver,
|
|
"_execute_causal_coordinate_flow_plan",
|
|
return_value="causalFlowEvaluationFailed:ValueError",
|
|
):
|
|
diagnostics = solver.solve(effort_variables=("p",))
|
|
|
|
self.assertTrue(diagnostics.success)
|
|
self.assertTrue(diagnostics.residual_verified_this_solve)
|
|
execution = solver.causal_execution_diagnostics()
|
|
self.assertEqual(
|
|
execution["disabledReason"],
|
|
"causalFlowEvaluationFailed:ValueError",
|
|
)
|
|
self.assertEqual(execution["executorV2RuntimeValidationFailureCount"], 1)
|
|
|
|
def test_compiled_v2_assignment_count_drift_fuses_to_fallback(self) -> None:
|
|
system = _system(zero_force_mass_project(), executor_v2=True)
|
|
state = system.initial_state_vector()
|
|
system.rhs(0.0, state)
|
|
solver = system.pressure_flow_solver
|
|
with patch.object(
|
|
solver,
|
|
"_execute_causal_coordinate_flow_plan",
|
|
return_value="causalFlowAssignmentCountMismatch",
|
|
):
|
|
diagnostics = solver.solve(effort_variables=("p",))
|
|
|
|
self.assertTrue(diagnostics.success)
|
|
self.assertTrue(diagnostics.residual_verified_this_solve)
|
|
execution = solver.causal_execution_diagnostics()
|
|
self.assertEqual(
|
|
execution["disabledReason"],
|
|
"causalFlowAssignmentCountMismatch",
|
|
)
|
|
self.assertEqual(execution["executorV2RuntimeValidationFailureCount"], 1)
|
|
|
|
def test_compiled_v2_keeps_the_sixty_four_solve_audit_boundary(self) -> None:
|
|
system = _system(zero_force_mass_project(), executor_v2=True)
|
|
state = system.initial_state_vector()
|
|
solver = system.pressure_flow_solver
|
|
system.rhs(0.0, state)
|
|
|
|
for _iteration in range(64):
|
|
system.rhs(0.0, state)
|
|
before = solver.causal_execution_diagnostics()
|
|
self.assertEqual(before["executorV2FastSolveCount"], 64)
|
|
self.assertEqual(before["fullResidualAuditCount"], 1)
|
|
|
|
system.rhs(0.0, state)
|
|
after = solver.causal_execution_diagnostics()
|
|
self.assertEqual(after["executorV2FastSolveCount"], 64)
|
|
self.assertEqual(after["fullResidualAuditCount"], 2)
|
|
|
|
def test_strict_causal_rhs_matches_environment_disabled_legacy_bitwise(
|
|
self,
|
|
) -> None:
|
|
optimized = _system(high_pressure_helium_step_project())
|
|
with patch.dict(
|
|
os.environ,
|
|
{CAUSAL_FAST_PATH_ENVIRONMENT_VARIABLE: "0"},
|
|
):
|
|
legacy = _system(high_pressure_helium_step_project())
|
|
|
|
optimized_state = optimized.initial_state_vector()
|
|
legacy_state = legacy.initial_state_vector()
|
|
for time in (0.0, 0.041, 0.8):
|
|
optimized_derivative = optimized.rhs(time, optimized_state)
|
|
legacy_derivative = legacy.rhs(time, legacy_state)
|
|
self.assertEqual(optimized_derivative, legacy_derivative)
|
|
self.assertEqual(
|
|
tuple(
|
|
unknown.read()
|
|
for unknown in optimized.pressure_flow_solver.unknowns
|
|
),
|
|
tuple(
|
|
unknown.read()
|
|
for unknown in legacy.pressure_flow_solver.unknowns
|
|
),
|
|
)
|
|
|
|
global_diagnostics = (
|
|
optimized.pressure_flow_solver.causal_execution_diagnostics()
|
|
)
|
|
self.assertTrue(global_diagnostics["eligible"])
|
|
self.assertGreater(global_diagnostics["fastSolveCount"], 0)
|
|
self.assertGreaterEqual(
|
|
global_diagnostics["fullResidualAuditCount"],
|
|
1,
|
|
)
|
|
secondary = (
|
|
optimized._thermofluid_closure_plan.secondary_block_solvers[0]
|
|
)
|
|
secondary_diagnostics = secondary.causal_execution_diagnostics()
|
|
self.assertTrue(secondary_diagnostics["eligible"])
|
|
self.assertGreater(secondary_diagnostics["fastSolveCount"], 0)
|
|
|
|
legacy_diagnostics = (
|
|
legacy.pressure_flow_solver.causal_execution_diagnostics()
|
|
)
|
|
self.assertFalse(legacy_diagnostics["enabled"])
|
|
self.assertEqual(
|
|
legacy_diagnostics["disabledReason"],
|
|
"disabledByEnvironment",
|
|
)
|
|
self.assertEqual(legacy_diagnostics["fastSolveCount"], 0)
|
|
|
|
def test_multiple_effort_anchors_conservatively_keep_legacy_path(self) -> None:
|
|
system = _system(chain_project())
|
|
diagnostics = (
|
|
system.pressure_flow_solver.causal_execution_diagnostics()
|
|
)
|
|
|
|
self.assertFalse(diagnostics["eligible"])
|
|
self.assertFalse(diagnostics["enabled"])
|
|
self.assertEqual(
|
|
diagnostics["fallbackReason"],
|
|
"effortGroupDoesNotHaveOneAnchor",
|
|
)
|
|
|
|
def test_runtime_flow_coverage_failure_fuses_to_verified_legacy_path(
|
|
self,
|
|
) -> None:
|
|
system = _system(zero_force_mass_project())
|
|
state = system.initial_state_vector()
|
|
system.rhs(0.0, state)
|
|
solver = system.pressure_flow_solver
|
|
original = solver._solve_explicit_flow_unknowns
|
|
|
|
def hide_coverage(*args, **kwargs):
|
|
original(*args, **kwargs)
|
|
return set()
|
|
|
|
with patch.object(
|
|
solver,
|
|
"_solve_explicit_flow_unknowns",
|
|
side_effect=hide_coverage,
|
|
):
|
|
diagnostics = solver.solve(effort_variables=("p",))
|
|
|
|
self.assertTrue(diagnostics.success)
|
|
self.assertTrue(diagnostics.residual_verified_this_solve)
|
|
execution = solver.causal_execution_diagnostics()
|
|
self.assertFalse(execution["enabled"])
|
|
self.assertEqual(
|
|
execution["disabledReason"],
|
|
"causalRuntimeGateFailed",
|
|
)
|
|
self.assertEqual(execution["legacyFallbackCount"], 1)
|
|
|
|
def test_periodic_audit_failure_disables_fast_path_before_fallback(self) -> None:
|
|
system = _system(zero_force_mass_project())
|
|
state = system.initial_state_vector()
|
|
system.rhs(0.0, state)
|
|
solver = system.pressure_flow_solver
|
|
solver._causal_audit_interval = 0
|
|
original_values = solver._pressure_flow_equation_values
|
|
call_count = 0
|
|
|
|
def one_bad_audit_value():
|
|
nonlocal call_count
|
|
call_count += 1
|
|
values = original_values()
|
|
if call_count != 1:
|
|
return values
|
|
return (values[0] + 1.0, *values[1:])
|
|
|
|
with patch.object(
|
|
solver,
|
|
"_pressure_flow_equation_values",
|
|
side_effect=one_bad_audit_value,
|
|
):
|
|
diagnostics = solver.solve(effort_variables=("p",))
|
|
|
|
self.assertTrue(diagnostics.success)
|
|
execution = solver.causal_execution_diagnostics()
|
|
self.assertFalse(execution["enabled"])
|
|
self.assertEqual(
|
|
execution["disabledReason"],
|
|
"causalResidualAuditFailed",
|
|
)
|
|
self.assertEqual(execution["auditFailureCount"], 1)
|
|
self.assertEqual(execution["legacyFallbackCount"], 1)
|
|
|
|
def test_requested_audit_interrupts_periodic_fast_sequence(self) -> None:
|
|
system = _system(zero_force_mass_project())
|
|
state = system.initial_state_vector()
|
|
solver = system.pressure_flow_solver
|
|
|
|
system.rhs(0.0, state)
|
|
system.rhs(0.0, state)
|
|
before = solver.causal_execution_diagnostics()
|
|
self.assertEqual(before["fullResidualAuditCount"], 1)
|
|
self.assertEqual(before["fastSolveCount"], 1)
|
|
|
|
solver.request_causal_audit()
|
|
system.rhs(0.0, state)
|
|
after = solver.causal_execution_diagnostics()
|
|
self.assertEqual(after["fullResidualAuditCount"], 2)
|
|
self.assertEqual(after["fastSolveCount"], 1)
|
|
|
|
def test_fast_solve_skips_the_full_residual_evaluator(self) -> None:
|
|
system = _system(zero_force_mass_project())
|
|
state = system.initial_state_vector()
|
|
solver = system.pressure_flow_solver
|
|
system.rhs(0.0, state)
|
|
|
|
with patch.object(
|
|
solver,
|
|
"_pressure_flow_equation_values",
|
|
wraps=solver._pressure_flow_equation_values,
|
|
) as evaluate_all:
|
|
system.rhs(0.0, state)
|
|
|
|
evaluate_all.assert_not_called()
|
|
self.assertTrue(solver.last_diagnostics.causal_fast_path_used)
|
|
self.assertFalse(
|
|
solver.last_diagnostics.residual_verified_this_solve
|
|
)
|
|
|
|
def test_nonfinite_explicit_assignment_fuses_and_verifies_same_solve(
|
|
self,
|
|
) -> None:
|
|
system = _system(zero_force_mass_project())
|
|
state = system.initial_state_vector()
|
|
system.rhs(0.0, state)
|
|
solver = system.pressure_flow_solver
|
|
original = solver._evaluate_explicit_flow_stage
|
|
call_count = 0
|
|
|
|
def one_nonfinite_assignment(stage):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
values = original(stage)
|
|
if call_count != 1:
|
|
return values
|
|
return (float("nan"), *values[1:])
|
|
|
|
with patch.object(
|
|
solver,
|
|
"_evaluate_explicit_flow_stage",
|
|
side_effect=one_nonfinite_assignment,
|
|
):
|
|
diagnostics = solver.solve(effort_variables=("p",))
|
|
|
|
self.assertTrue(diagnostics.success)
|
|
self.assertTrue(diagnostics.residual_verified_this_solve)
|
|
execution = solver.causal_execution_diagnostics()
|
|
self.assertFalse(execution["enabled"])
|
|
self.assertEqual(
|
|
execution["disabledReason"],
|
|
"causalRuntimeGateFailed",
|
|
)
|
|
self.assertEqual(execution["legacyFallbackCount"], 1)
|
|
|
|
def test_nonpositive_pressure_fuses_and_verifies_same_solve(self) -> None:
|
|
system = _system(high_pressure_helium_step_project())
|
|
state = system.initial_state_vector()
|
|
system.rhs(0.0, state)
|
|
solver = system.pressure_flow_solver
|
|
original = solver._solve_explicit_flow_unknowns
|
|
pressure = next(
|
|
unknown for unknown in solver.unknowns if unknown.variable == "p"
|
|
)
|
|
|
|
def make_pressure_invalid(*args, **kwargs):
|
|
seeded = original(*args, **kwargs)
|
|
pressure.write(-1.0)
|
|
return seeded
|
|
|
|
with patch.object(
|
|
solver,
|
|
"_solve_explicit_flow_unknowns",
|
|
side_effect=make_pressure_invalid,
|
|
):
|
|
diagnostics = solver.solve(effort_variables=("p",))
|
|
|
|
self.assertTrue(diagnostics.success)
|
|
self.assertTrue(diagnostics.residual_verified_this_solve)
|
|
self.assertGreater(pressure.read(), 0.0)
|
|
execution = solver.causal_execution_diagnostics()
|
|
self.assertFalse(execution["enabled"])
|
|
self.assertEqual(
|
|
execution["disabledReason"],
|
|
"causalRuntimeGateFailed",
|
|
)
|
|
self.assertEqual(execution["legacyFallbackCount"], 1)
|
|
|
|
def test_default_periodic_audit_runs_after_sixty_four_fast_solves(self) -> None:
|
|
system = _system(zero_force_mass_project())
|
|
state = system.initial_state_vector()
|
|
solver = system.pressure_flow_solver
|
|
system.rhs(0.0, state)
|
|
|
|
for _iteration in range(64):
|
|
system.rhs(0.0, state)
|
|
before_boundary = solver.causal_execution_diagnostics()
|
|
self.assertEqual(before_boundary["auditInterval"], 64)
|
|
self.assertEqual(before_boundary["fastSolveCount"], 64)
|
|
self.assertEqual(before_boundary["fullResidualAuditCount"], 1)
|
|
|
|
system.rhs(0.0, state)
|
|
after_boundary = solver.causal_execution_diagnostics()
|
|
self.assertEqual(after_boundary["fastSolveCount"], 64)
|
|
self.assertEqual(after_boundary["fullResidualAuditCount"], 2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|