228 lines
7.1 KiB
Python
228 lines
7.1 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from app.simulation.components.amesim.mechanical.translational import (
|
|
AmesimF000,
|
|
AmesimForc,
|
|
AmesimMecmas21,
|
|
)
|
|
from app.simulation.components.amesim.signals.sources import AmesimStep0
|
|
from app.simulation.core.medium import IdealGasMedium
|
|
from app.simulation.solvers.solver import SolveIVPConfig
|
|
from app.simulation.systems.generic import (
|
|
SimulationSampleTimeError,
|
|
simulation_preparation_issues,
|
|
simulation_sample_times,
|
|
)
|
|
from app.simulation.systems.network import SimulationNetwork
|
|
from app.system_xml import validate_system_xml_document
|
|
|
|
|
|
def _system_xml(
|
|
*,
|
|
t_start: str,
|
|
t_stop: str,
|
|
sample_step: str,
|
|
) -> str:
|
|
return f"""<System schemaVersion="3" unitSystem="SI">
|
|
<Simulation
|
|
tStart="{t_start}"
|
|
tStop="{t_stop}"
|
|
sampleStep="{sample_step}"
|
|
maxStep="0.01"
|
|
method="BDF"
|
|
/>
|
|
<Components>
|
|
<Component id="cylinder_1" type="cylinder" modelVersion="1.0.0">
|
|
<Parameter name="volume" value="0.01"/>
|
|
<Parameter name="p0" value="35000000"/>
|
|
<Parameter name="T0" value="300"/>
|
|
</Component>
|
|
</Components>
|
|
<Connections/>
|
|
</System>"""
|
|
|
|
|
|
def _fanout_mechanical_network(
|
|
*,
|
|
second_island_has_storage: bool,
|
|
) -> SimulationNetwork:
|
|
medium = IdealGasMedium()
|
|
network = SimulationNetwork("signal-fanout-physical-islands")
|
|
|
|
command = AmesimStep0(
|
|
"command",
|
|
medium,
|
|
initial=0.0,
|
|
final=1.0,
|
|
time=0.0,
|
|
)
|
|
force_a = AmesimForc("force_a")
|
|
mass_a = AmesimMecmas21("mass_a", medium)
|
|
zero_a = AmesimF000("zero_a")
|
|
force_b = AmesimForc("force_b")
|
|
zero_b = AmesimF000("zero_b")
|
|
components = [command, force_a, mass_a, zero_a, force_b, zero_b]
|
|
mass_b = None
|
|
if second_island_has_storage:
|
|
mass_b = AmesimMecmas21("mass_b", medium)
|
|
components.append(mass_b)
|
|
|
|
for component in components:
|
|
network.add_component(component)
|
|
|
|
network.connect("command", "out", "force_a", "res")
|
|
network.connect("command", "out", "force_b", "res")
|
|
network.connect("force_a", "port_2", "mass_a", "port_1")
|
|
network.connect("mass_a", "port_2", "zero_a", "port_1")
|
|
if mass_b is None:
|
|
network.connect("force_b", "port_2", "zero_b", "port_1")
|
|
else:
|
|
network.connect("force_b", "port_2", "mass_b", "port_1")
|
|
network.connect("mass_b", "port_2", "zero_b", "port_1")
|
|
return network
|
|
|
|
|
|
class SimulationSampleTimeSafetyTests(unittest.TestCase):
|
|
def test_grid_contains_both_endpoints_and_is_strictly_increasing(self) -> None:
|
|
times = simulation_sample_times(
|
|
SolveIVPConfig(t_start=1.0, t_stop=2.0),
|
|
0.3,
|
|
)
|
|
|
|
self.assertEqual(times[0], 1.0)
|
|
self.assertEqual(times[-1], 2.0)
|
|
self.assertGreaterEqual(len(times), 2)
|
|
self.assertTrue(
|
|
all(first < second for first, second in zip(times, times[1:]))
|
|
)
|
|
|
|
def test_step_larger_than_duration_still_returns_two_endpoints(self) -> None:
|
|
self.assertEqual(
|
|
simulation_sample_times(
|
|
SolveIVPConfig(t_start=1.0, t_stop=1.25),
|
|
10.0,
|
|
),
|
|
[1.0, 1.25],
|
|
)
|
|
|
|
def test_grid_can_exceed_the_legacy_point_limit(self) -> None:
|
|
times = simulation_sample_times(
|
|
SolveIVPConfig(t_start=0.0, t_stop=5.0),
|
|
0.0001,
|
|
)
|
|
|
|
self.assertEqual(len(times), 50001)
|
|
self.assertEqual(times[0], 0.0)
|
|
self.assertEqual(times[-1], 5.0)
|
|
|
|
def test_system_xml_accepts_more_than_the_legacy_point_limit(self) -> None:
|
|
report = validate_system_xml_document(
|
|
_system_xml(t_start="0", t_stop="5", sample_step="0.0001")
|
|
)
|
|
|
|
self.assertTrue(report.valid, report.issues)
|
|
|
|
def test_tiny_step_is_rejected_before_an_oversized_grid_is_allocated(self) -> None:
|
|
with self.assertRaises(SimulationSampleTimeError) as caught:
|
|
simulation_sample_times(
|
|
SolveIVPConfig(t_start=0.0, t_stop=1.0),
|
|
1.0e-300,
|
|
)
|
|
|
|
self.assertEqual(
|
|
caught.exception.code,
|
|
"SIMULATION_SAMPLE_COUNT_UNREPRESENTABLE",
|
|
)
|
|
|
|
def test_non_finite_derived_duration_has_a_stable_error(self) -> None:
|
|
with self.assertRaises(SimulationSampleTimeError) as caught:
|
|
simulation_sample_times(
|
|
SolveIVPConfig(t_start=-1.0e308, t_stop=1.0e308),
|
|
1.0e308,
|
|
)
|
|
|
|
self.assertEqual(caught.exception.code, "SIMULATION_TIME_SPAN_NOT_FINITE")
|
|
|
|
def test_step_that_cannot_advance_absolute_time_is_rejected(self) -> None:
|
|
with self.assertRaises(SimulationSampleTimeError) as caught:
|
|
simulation_sample_times(
|
|
SolveIVPConfig(t_start=1.0e16, t_stop=1.0e16 + 4.0),
|
|
1.0,
|
|
)
|
|
|
|
self.assertEqual(
|
|
caught.exception.code,
|
|
"SIMULATION_SAMPLE_TIME_UNREPRESENTABLE",
|
|
)
|
|
|
|
def test_system_xml_reports_sample_grid_failures_during_semantic_validation(
|
|
self,
|
|
) -> None:
|
|
cases = (
|
|
(
|
|
_system_xml(t_start="0", t_stop="1", sample_step="1e-300"),
|
|
"SIMULATION_SAMPLE_COUNT_UNREPRESENTABLE",
|
|
),
|
|
(
|
|
_system_xml(
|
|
t_start="-1e308",
|
|
t_stop="1e308",
|
|
sample_step="1e308",
|
|
),
|
|
"SIMULATION_TIME_SPAN_NOT_FINITE",
|
|
),
|
|
(
|
|
_system_xml(
|
|
t_start="10000000000000000",
|
|
t_stop="10000000000000004",
|
|
sample_step="1",
|
|
),
|
|
"SIMULATION_SAMPLE_TIME_UNREPRESENTABLE",
|
|
),
|
|
)
|
|
|
|
for xml, expected_code in cases:
|
|
with self.subTest(expected_code=expected_code):
|
|
report = validate_system_xml_document(xml)
|
|
self.assertFalse(report.valid)
|
|
self.assertIn(
|
|
expected_code,
|
|
{issue.code for issue in report.issues},
|
|
)
|
|
|
|
|
|
class PhysicalIslandSafetyTests(unittest.TestCase):
|
|
def test_signal_fanout_does_not_hide_an_unanchored_physical_island(self) -> None:
|
|
issues = simulation_preparation_issues(
|
|
_fanout_mechanical_network(second_island_has_storage=False)
|
|
)
|
|
island_issues = [
|
|
issue
|
|
for issue in issues
|
|
if issue.code == "ALGEBRAIC_ISLAND_HAS_NO_STORAGE"
|
|
]
|
|
|
|
self.assertEqual(len(island_issues), 1)
|
|
self.assertIn("force_b", island_issues[0].message)
|
|
self.assertIn("zero_b", island_issues[0].message)
|
|
self.assertNotIn("command", island_issues[0].message)
|
|
self.assertNotIn("mass_a", island_issues[0].message)
|
|
|
|
def test_signal_fanout_between_two_anchored_physical_islands_is_allowed(
|
|
self,
|
|
) -> None:
|
|
issues = simulation_preparation_issues(
|
|
_fanout_mechanical_network(second_island_has_storage=True)
|
|
)
|
|
|
|
self.assertNotIn(
|
|
"ALGEBRAIC_ISLAND_HAS_NO_STORAGE",
|
|
{issue.code for issue in issues},
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|