243 lines
8.9 KiB
Python
243 lines
8.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import textwrap
|
|
import unittest
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
class SimulationPerformancePipelineTests(unittest.TestCase):
|
|
def test_off_mode_does_not_change_result_diagnostics(self) -> None:
|
|
script = textwrap.dedent(
|
|
"""
|
|
from app.main import (
|
|
build_reactflow_system_xml,
|
|
run_system_xml_simulation,
|
|
)
|
|
from tests.test_amesim_pnl00r_xml import amesim_pnl00r_project
|
|
|
|
result = run_system_xml_simulation(
|
|
build_reactflow_system_xml(amesim_pnl00r_project())
|
|
)
|
|
assert result["success"], result["message"]
|
|
print("performance" in result["diagnostics"])
|
|
"""
|
|
)
|
|
environment = os.environ.copy()
|
|
environment["SIMULATIONAPP_PROFILE"] = "off"
|
|
completed = subprocess.run(
|
|
[sys.executable, "-c", script],
|
|
cwd=PROJECT_ROOT,
|
|
env=environment,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
)
|
|
|
|
self.assertEqual(completed.stdout.strip(), "False")
|
|
|
|
def test_standard_mode_reports_coarse_pipeline_phases_per_run(self) -> None:
|
|
script = textwrap.dedent(
|
|
"""
|
|
import json
|
|
|
|
from app.main import (
|
|
build_reactflow_system_xml,
|
|
run_system_xml_simulation,
|
|
)
|
|
from tests.test_amesim_pnl00r_xml import amesim_pnl00r_project
|
|
|
|
xml = build_reactflow_system_xml(amesim_pnl00r_project())
|
|
snapshots = []
|
|
for _ in range(2):
|
|
result = run_system_xml_simulation(xml)
|
|
assert result["success"], result["message"]
|
|
performance = result["diagnostics"]["performance"]
|
|
snapshots.append(
|
|
{
|
|
"mode": performance["mode"],
|
|
"calls": {
|
|
name: metrics["calls"]
|
|
for name, metrics in performance["phases"].items()
|
|
},
|
|
"propertyCount": len(performance["properties"]),
|
|
}
|
|
)
|
|
print(json.dumps(snapshots))
|
|
"""
|
|
)
|
|
environment = os.environ.copy()
|
|
environment["SIMULATIONAPP_PROFILE"] = "standard"
|
|
completed = subprocess.run(
|
|
[sys.executable, "-c", script],
|
|
cwd=PROJECT_ROOT,
|
|
env=environment,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
)
|
|
snapshots = json.loads(completed.stdout)
|
|
required_phases = {
|
|
"simulation.total",
|
|
"simulation.xml_validation",
|
|
"simulation.network_compilation",
|
|
"simulation.system_construction",
|
|
"simulation.sample_initialization",
|
|
"simulation.integration",
|
|
"simulation.postprocessing",
|
|
"simulation.result_assembly",
|
|
"simulation.response_assembly",
|
|
}
|
|
audit_only_phases = {
|
|
"simulation.rhs",
|
|
"simulation.closure",
|
|
"simulation.refresh",
|
|
"simulation.pressure_flow",
|
|
"simulation.pneumatic_volume",
|
|
"simulation.stream",
|
|
"simulation.signal",
|
|
"simulation.derivatives",
|
|
}
|
|
|
|
self.assertEqual(len(snapshots), 2)
|
|
for snapshot in snapshots:
|
|
self.assertEqual(snapshot["mode"], "standard")
|
|
self.assertTrue(required_phases.issubset(snapshot["calls"]))
|
|
self.assertEqual(snapshot["calls"]["simulation.total"], 1)
|
|
self.assertEqual(snapshot["calls"]["simulation.xml_validation"], 1)
|
|
self.assertEqual(snapshot["calls"]["simulation.network_compilation"], 1)
|
|
self.assertEqual(snapshot["calls"]["simulation.integration"], 1)
|
|
self.assertEqual(snapshot["calls"]["simulation.postprocessing"], 1)
|
|
self.assertEqual(snapshot["calls"]["simulation.result_assembly"], 1)
|
|
self.assertEqual(snapshot["calls"]["simulation.response_assembly"], 1)
|
|
self.assertTrue(audit_only_phases.isdisjoint(snapshot["calls"]))
|
|
self.assertEqual(snapshot["propertyCount"], 0)
|
|
|
|
def test_audit_mode_reports_hot_phases_and_property_metrics(self) -> None:
|
|
script = textwrap.dedent(
|
|
"""
|
|
import json
|
|
|
|
from app.main import (
|
|
build_reactflow_system_xml,
|
|
run_system_xml_simulation,
|
|
)
|
|
from tests.test_amesim_pnl00r_xml import amesim_pnl00r_project
|
|
|
|
result = run_system_xml_simulation(
|
|
build_reactflow_system_xml(amesim_pnl00r_project())
|
|
)
|
|
assert result["success"], result["message"]
|
|
print(json.dumps(result["diagnostics"]["performance"]))
|
|
"""
|
|
)
|
|
environment = os.environ.copy()
|
|
environment["SIMULATIONAPP_PROFILE"] = "audit"
|
|
completed = subprocess.run(
|
|
[sys.executable, "-c", script],
|
|
cwd=PROJECT_ROOT,
|
|
env=environment,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
)
|
|
snapshot = json.loads(completed.stdout)
|
|
|
|
for phase in (
|
|
"simulation.rhs",
|
|
"simulation.closure",
|
|
"simulation.refresh",
|
|
"simulation.pressure_flow",
|
|
"simulation.pneumatic_volume",
|
|
"simulation.stream",
|
|
"simulation.signal",
|
|
"simulation.derivatives",
|
|
):
|
|
self.assertGreater(snapshot["phases"][phase]["calls"], 0)
|
|
self.assertGreater(len(snapshot["properties"]), 0)
|
|
self.assertGreater(snapshot["propertyOutermostNs"], 0)
|
|
|
|
def test_audit_property_totals_match_run_local_cache(self) -> None:
|
|
script = textwrap.dedent(
|
|
"""
|
|
import json
|
|
|
|
from app.simulation.components.amesim.media.mediums import (
|
|
AmesimHeliumPengRobinsonMedium,
|
|
)
|
|
from app.simulation.performance import profile_run
|
|
from app.simulation.property_cache import property_cache_run
|
|
|
|
medium = AmesimHeliumPengRobinsonMedium()
|
|
pressure = 15.3e6
|
|
temperature = 293.15
|
|
volume = 0.057
|
|
with property_cache_run() as cache:
|
|
assert cache is not None
|
|
with profile_run() as trace:
|
|
density = medium.density(pressure, temperature)
|
|
medium.density(pressure, temperature)
|
|
medium.isentropic_density_pressure_factor(
|
|
pressure, temperature, 1.0e6
|
|
)
|
|
medium.isentropic_density_pressure_factor(
|
|
pressure, temperature, 1.0e6
|
|
)
|
|
enthalpy = medium.specific_enthalpy_at_pressure(
|
|
pressure, temperature
|
|
)
|
|
medium.temperature_from_pressure_enthalpy(pressure, enthalpy)
|
|
medium.temperature_from_pressure_enthalpy(pressure, enthalpy)
|
|
mass = density * volume
|
|
energy = mass * medium.specific_internal_energy_at_pressure(
|
|
pressure, temperature
|
|
)
|
|
medium.properties_from_mU(mass, energy, volume)
|
|
medium.properties_from_mU(mass, energy, volume)
|
|
cache_info = cache.info()
|
|
snapshot = trace.snapshot()
|
|
cached_properties = [
|
|
metrics
|
|
for metrics in snapshot["properties"].values()
|
|
if metrics["cacheLookups"] > 0
|
|
]
|
|
print(json.dumps({
|
|
"cacheHits": cache_info.hits,
|
|
"cacheMisses": cache_info.misses,
|
|
"propertyHits": sum(item["cacheHits"] for item in cached_properties),
|
|
"propertyMisses": sum(item["cacheMisses"] for item in cached_properties),
|
|
}))
|
|
"""
|
|
)
|
|
environment = os.environ.copy()
|
|
environment["SIMULATIONAPP_PROFILE"] = "audit"
|
|
environment["SIMULATIONAPP_PROPERTY_CACHE"] = "on"
|
|
completed = subprocess.run(
|
|
[sys.executable, "-c", script],
|
|
cwd=PROJECT_ROOT,
|
|
env=environment,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
)
|
|
totals = json.loads(completed.stdout)
|
|
|
|
self.assertGreater(totals["cacheHits"], 0)
|
|
self.assertGreater(totals["cacheMisses"], 0)
|
|
self.assertEqual(totals["cacheHits"], totals["propertyHits"])
|
|
self.assertEqual(totals["cacheMisses"], totals["propertyMisses"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|