96 lines
4.7 KiB
Python
96 lines
4.7 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from bedit_core.models import Component, ComponentID, Document, GraphImplementation
|
|
from bedit_gui.models import Simulation, SimulationMethod
|
|
from bedit_gui.services import document_files
|
|
from bedit_gui.simulation_models import CompiledModel, SimulationRoot
|
|
from bedit_simulation import compile_component_sync
|
|
|
|
|
|
def component_choices(document: Document) -> list[tuple[ComponentID, Component, str]]:
|
|
choices: list[tuple[ComponentID, Component, str]] = []
|
|
|
|
def collect(items: dict[ComponentID, Component], path: tuple[str, ...] = ()) -> None:
|
|
for component_id, component in items.items():
|
|
component_path = (*path, component.name)
|
|
choices.append((component_id, component, ".".join(component_path)))
|
|
if isinstance(component.implementation, GraphImplementation):
|
|
collect(component.implementation.graph.components, component_path)
|
|
|
|
collect(document.root)
|
|
return choices
|
|
|
|
|
|
def load_and_compile_bedit(path: str | Path, *, component_selector: str | None = None, simulation_selector: str | None = None, working_directory: str | Path | None = None, omc_command: str = "omc") -> tuple[SimulationRoot, CompiledModel]:
|
|
"""Open a BEdit document, resolve one launch target, and compile it."""
|
|
if bool(component_selector) == bool(simulation_selector):
|
|
raise ValueError("specify exactly one component or simulation settings block")
|
|
source_path = Path(path).resolve()
|
|
document = document_files.load(source_path)
|
|
choices = component_choices(document)
|
|
settings_name: str | None = None
|
|
|
|
if component_selector is not None:
|
|
component_id, component, component_path = _find_component(choices, component_selector)
|
|
settings = _default_settings(component_id)
|
|
else:
|
|
database = document.metadata.get("simulation_database") if document.metadata is not None else None
|
|
if database is None or not hasattr(database, "simulations"):
|
|
raise ValueError("the BEdit document does not contain simulation settings")
|
|
matches = [(simulation_id, simulation) for simulation_id, simulation in database.simulations.items() if simulation.name == simulation_selector or str(simulation_id) == simulation_selector]
|
|
if len(matches) != 1:
|
|
raise ValueError(f"simulation settings {simulation_selector!r} were not found or are ambiguous")
|
|
_simulation_id, settings = matches[0]
|
|
component_id, component, component_path = _find_component(choices, str(settings.component))
|
|
settings_name = settings.name
|
|
|
|
build_directory = Path(working_directory) if working_directory is not None else Path(tempfile.mkdtemp(prefix="bedit-compiled-"))
|
|
root = SimulationRoot(
|
|
format_version=1,
|
|
source_document=str(source_path),
|
|
source_document_id=str(document.id),
|
|
component=component_id,
|
|
component_path=component_path,
|
|
settings_name=settings_name,
|
|
settings=settings,
|
|
)
|
|
return root, _compile(component, build_directory, omc_command)
|
|
|
|
|
|
def compile_simulation_root(root: SimulationRoot, *, working_directory: str | Path | None = None, omc_command: str = "omc") -> CompiledModel:
|
|
if root.source_document is None:
|
|
raise ValueError("the simulation does not reference a BEdit source document and cannot be recompiled")
|
|
document = document_files.load(root.source_document)
|
|
_component_id, component, _component_path = _find_component(component_choices(document), str(root.component))
|
|
build_directory = Path(working_directory) if working_directory is not None else Path(tempfile.mkdtemp(prefix="bedit-compiled-"))
|
|
return _compile(component, build_directory, omc_command)
|
|
|
|
|
|
def _find_component(choices: list[tuple[ComponentID, Component, str]], selector: str) -> tuple[ComponentID, Component, str]:
|
|
matches = [choice for choice in choices if str(choice[0]) == selector or choice[2] == selector]
|
|
if len(matches) != 1:
|
|
raise ValueError(f"component {selector!r} was not found or is ambiguous")
|
|
return matches[0]
|
|
|
|
|
|
def _default_settings(component_id: ComponentID) -> Simulation:
|
|
return Simulation(
|
|
component=component_id,
|
|
name="Default",
|
|
start_time=0.0,
|
|
duration=1.0,
|
|
use_timed_steps=False,
|
|
number_of_steps=500,
|
|
step_size=0.002,
|
|
method=SimulationMethod.DASSL,
|
|
dassl_tolerance=1e-6,
|
|
)
|
|
|
|
|
|
def _compile(component: Component, working_directory: Path, omc_command: str) -> CompiledModel:
|
|
build = compile_component_sync(component, working_directory, omc_command=omc_command)
|
|
return CompiledModel(build.model_name, str(build.executable.resolve()), str(build.executable.parent.resolve()), build.output, build.errors)
|