101 lines
2.9 KiB
Python
101 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
from app.simulation.paths import PROJECT_ROOT, SIMULATION_RUNS_DIR
|
|
from app.simulation.examples.test_mql.system import (
|
|
TestMqlRunConfig,
|
|
TestMqlSimulationResult,
|
|
TestMqlSystem,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PreparedTestMqlRun:
|
|
run_config: TestMqlRunConfig
|
|
repo_root: Path
|
|
output_dir: Path
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TestMqlRunResult:
|
|
run_config: TestMqlRunConfig
|
|
prepared_run: PreparedTestMqlRun
|
|
system: TestMqlSystem
|
|
result: TestMqlSimulationResult
|
|
summary_path: Path
|
|
|
|
|
|
def format_test_mql_summary(system: TestMqlSystem) -> str:
|
|
snapshot = system.snapshot()
|
|
lines = [
|
|
"Model: test_mql",
|
|
f"Source archive: {system.archive_path}",
|
|
f"Components: {snapshot.component_count}",
|
|
f"Connections: {snapshot.connection_count}",
|
|
f"Continuous states in AMESim modelinfo: {snapshot.continuous_state_count}",
|
|
f"Discrete states in AMESim modelinfo: {snapshot.discrete_state_count}",
|
|
"Global parameters:",
|
|
]
|
|
for name, value in sorted(snapshot.global_parameters.items()):
|
|
lines.append(f" - {name}: {value}")
|
|
lines.append("Component submodels:")
|
|
for name, count in sorted(snapshot.submodel_counts.items()):
|
|
lines.append(f" - {name}: {count}")
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def _default_run_output_dir() -> Path:
|
|
timestamp = datetime.now(UTC).strftime("test_mql_%Y%m%d_%H%M%S_%f")
|
|
return SIMULATION_RUNS_DIR / timestamp
|
|
|
|
|
|
def prepare_test_mql_run(
|
|
*,
|
|
run_config: TestMqlRunConfig | None = None,
|
|
output_dir: Path | None = None,
|
|
) -> PreparedTestMqlRun:
|
|
return PreparedTestMqlRun(
|
|
run_config=run_config or TestMqlRunConfig(),
|
|
repo_root=PROJECT_ROOT,
|
|
output_dir=output_dir or _default_run_output_dir(),
|
|
)
|
|
|
|
|
|
def run_prepared_test_mql(prepared_run: PreparedTestMqlRun) -> TestMqlRunResult:
|
|
system = TestMqlSystem()
|
|
result = system.simulate(prepared_run.run_config)
|
|
prepared_run.output_dir.mkdir(parents=True, exist_ok=True)
|
|
summary_path = prepared_run.output_dir / "test_mql_model_summary.txt"
|
|
summary_path.write_text(format_test_mql_summary(system), encoding="utf-8")
|
|
return TestMqlRunResult(
|
|
run_config=prepared_run.run_config,
|
|
prepared_run=prepared_run,
|
|
system=system,
|
|
result=result,
|
|
summary_path=summary_path,
|
|
)
|
|
|
|
|
|
def run_test_mql(
|
|
*,
|
|
run_config: TestMqlRunConfig | None = None,
|
|
output_dir: Path | None = None,
|
|
) -> TestMqlRunResult:
|
|
return run_prepared_test_mql(
|
|
prepare_test_mql_run(run_config=run_config, output_dir=output_dir)
|
|
)
|
|
|
|
|
|
def main() -> None:
|
|
run = run_test_mql()
|
|
print(format_test_mql_summary(run.system), end="")
|
|
print(f"Samples: {len(run.result.t)}")
|
|
print(f"Output directory: {run.prepared_run.output_dir}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|