优化压力流量求解并达到四路性能门槛

This commit is contained in:
huojiarong committed 2026-08-15 11:45:22 +00:00
1 parent 6572defaa4
commit 6a064892e2
20 files changed
+711 -175

No files matched your search

@@ -41,6 +41,10 @@ class AmesimMechanicalPublicComponentTests(unittest.TestCase):
residuals = converter.pressure_flow_equation_residuals()
self.assertAlmostEqual(converter.output_force, 20.0)
self.assertEqual(
converter.pressure_flow_equation_values(),
tuple(residual.value for residual in residuals),
)
self.assertAlmostEqual(residuals[0].value, 0.0)
self.assertEqual(converter.component_result_values(), {"force": 20.0})
self.assertEqual(converter.parameter_values, {"direction": 1.0})
@@ -54,10 +58,13 @@ class AmesimMechanicalPublicComponentTests(unittest.TestCase):
converter.res.signal = 20.0
converter.port_2.f = 20.0
self.assertAlmostEqual(
converter.pressure_flow_equation_residuals()[0].value,
0.0,
residuals = converter.pressure_flow_equation_residuals()
self.assertEqual(
converter.pressure_flow_equation_values(),
tuple(residual.value for residual in residuals),
)
self.assertAlmostEqual(residuals[0].value, 0.0)
self.assertEqual(converter.parameter_values, {"direction": -1.0})
self.assertEqual(converter.component_result_values(), {"force": -20.0})
@@ -231,9 +238,17 @@ class AmesimMechanicalPublicComponentTests(unittest.TestCase):
residuals = node.pressure_flow_equation_residuals()
self.assertEqual(node.active_ports, ("port_1", "port_2", "port_3"))
self.assertEqual(
tuple(definition.name for definition in node.active_port_definitions),
node.active_ports,
)
self.assertAlmostEqual(node.total_force, 10.0)
self.assertAlmostEqual(node.force_balance, 0.0)
self.assertEqual(len(residuals), 5 + 18 * 3)
self.assertEqual(len(residuals), 5)
self.assertEqual(
node.pressure_flow_equation_values(),
tuple(residual.value for residual in residuals),
)
self.assertTrue(all(abs(residual.value) <= 1.0e-12 for residual in residuals))
self.assertEqual(node.component_result_values(), {"tforce": 10.0})
+2
View File
@@ -423,6 +423,8 @@ class AmesimMechanicalXmlTests(unittest.TestCase):
result = run_system_xml_simulation(xml)
self.assertTrue(result["success"], result["message"])
self.assertNotIn("node_1.port_4.x", result["series"])
self.assertNotIn("node_1.port_21.f", result["series"])
self.assertAlmostEqual(result["series"]["mass_1.a"][0], 5.0)
def test_elastic_contact_project_compiles_and_simulates(self) -> None:
+71
View File
@@ -241,6 +241,77 @@ class IntegrateOdeTests(unittest.TestCase):
)
)
def test_segmented_solver_reports_implicit_work_by_event_segment(self) -> None:
import numpy as np
import scipy.integrate
class CountingBDF:
def __init__(self, _fun, t0, y0, t_bound, **_kwargs):
self.t = float(t0)
self.y = np.asarray(y0, dtype=float)
self.t_bound = float(t_bound)
self.status = "running"
self.nfev = 2
self.njev = 1
self.nlu = 0
def step(self):
self.t = self.t_bound
self.nfev += 3
self.nlu += 2
self.status = "finished"
return None
def dense_output(self):
state = self.y.copy()
return lambda _time: state.copy()
with patch.object(scipy.integrate, "BDF", CountingBDF):
result = integrate_ode(
rhs=lambda _time, _state: [0.0],
initial_state=[1.0],
config=SolveIVPConfig(
t_start=0.0,
t_stop=1.0,
method="BDF",
max_step=1.0,
),
t_eval=[0.0, 0.4, 1.0],
breakpoints=[0.4],
)
self.assertTrue(result.success, result.message)
self.assertEqual(len(result.solver_segments), 2)
self.assertEqual(
[segment.as_dict() for segment in result.solver_segments],
[
{
"startTime": 0.0,
"requestedStopTime": 0.4,
"simulatedUntil": 0.4,
"nfev": 5,
"njev": 1,
"nlu": 2,
"acceptedStepCount": 1,
"solverStartCount": 1,
"stateTransitionCount": 0,
"recoverableRetryCount": 0,
},
{
"startTime": 0.4,
"requestedStopTime": 1.0,
"simulatedUntil": 1.0,
"nfev": 5,
"njev": 1,
"nlu": 2,
"acceptedStepCount": 1,
"solverStartCount": 1,
"stateTransitionCount": 0,
"recoverableRetryCount": 0,
},
],
)
def test_segmented_solver_can_cancel_after_crossing_a_breakpoint(self) -> None:
callback_times: list[float] = []
cancellation_requested = False
@@ -322,6 +322,23 @@ class GenericSystemXmlSimulationTests(unittest.TestCase):
self.assertFalse(result.success)
self.assertEqual(result.status, "cancelled")
self.assertGreaterEqual(result.diagnostics["sampleCount"], 2)
integration = result.diagnostics["integration"]
self.assertEqual(integration["method"], "BDF")
self.assertEqual(integration["segmentCount"], len(integration["segments"]))
self.assertEqual(
integration["totals"]["nfev"],
sum(segment["nfev"] for segment in integration["segments"]),
)
sparsity = integration["jacobianSparsity"]
self.assertGreater(sparsity["nonzeroCount"], 0)
self.assertGreater(sparsity["colorGroupCount"], 0)
self.assertEqual(
integration["totals"]["finiteDifferenceRhsEstimate"],
sum(
segment["finiteDifferenceRhsEstimate"]
for segment in integration["segments"]
),
)
self.assertGreater(result.simulated_until, 0.0)
self.assertLess(result.simulated_until, 0.05)
self.assertEqual(
@@ -219,6 +219,25 @@ class PressureFlowSolverInitializationTests(unittest.TestCase):
self.assertAlmostEqual(high.port_b.m_flow, -valve.port_2.m_flow, places=10)
self.assertAlmostEqual(low.port_a.m_flow, -valve.port_3.m_flow, places=10)
def test_compiled_plan_does_not_rebuild_equation_metadata_during_solve(self) -> None:
boundary = AmesimPnpl01("compiled_closed")
boundary.port_1.p = 100_000.0
boundary.port_1.m_flow = 1.0
network = SimulationNetwork("compiled-closed-boundary")
network.add_component(boundary)
solver = PressureFlowSolver(network)
with patch.object(
boundary,
"pressure_flow_equation_residuals",
side_effect=AssertionError("equation metadata rebuilt at runtime"),
):
diagnostics = solver.solve()
self.assertTrue(diagnostics.success)
self.assertEqual(diagnostics.evaluations, 0)
self.assertEqual(boundary.port_1.m_flow, 0.0)
@staticmethod
def _closed_boundary_solver() -> PressureFlowSolver:
boundary = AmesimPnpl01("closed")