428 lines
14 KiB
Python
428 lines
14 KiB
Python
import json
|
|
import os
|
|
import shlex
|
|
import shutil
|
|
import socket
|
|
import tempfile
|
|
import time
|
|
import xml.etree.ElementTree as ET
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from queue import Queue
|
|
from threading import Event, Lock, Thread, current_thread
|
|
from typing import Any
|
|
|
|
from bedit.core.application_log import get_logger
|
|
from bedit.core.simulation.results import (
|
|
SimulationExecutionResult,
|
|
load_openmodelica_csv,
|
|
)
|
|
|
|
|
|
log = get_logger(__name__)
|
|
ResultCallback = Callable[[Any], None]
|
|
ErrorCallback = Callable[[Exception], None]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _Request:
|
|
description: str
|
|
operation: Callable[[Any, Path], Any]
|
|
callback: ResultCallback | None
|
|
error_callback: ErrorCallback | None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SimulationProgress:
|
|
phase: str
|
|
current_step_size: float
|
|
time: float
|
|
progress: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SimulationMessage:
|
|
stream: str
|
|
type: str
|
|
text: str
|
|
|
|
|
|
class OpenModelicaInterface:
|
|
"""Asynchronous, persistent interface to one OpenModelica session.
|
|
|
|
The worker and OMC session are created lazily for the first request. Callbacks
|
|
execute on background threads and must not manipulate Qt widgets directly.
|
|
"""
|
|
|
|
def __init__(self, executable_path: str = "") -> None:
|
|
self._executable_path = executable_path
|
|
self._lifecycle_lock = Lock()
|
|
self._queue: Queue[_Request | None] | None = None
|
|
self._worker: Thread | None = None
|
|
self._temp_dir: Path | None = None
|
|
|
|
@property
|
|
def executable_path(self) -> str:
|
|
return self._executable_path
|
|
|
|
def configure(self, executable_path: str) -> None:
|
|
"""Use a new executable path for subsequent requests."""
|
|
|
|
if executable_path == self._executable_path:
|
|
return
|
|
self.shutdown(wait=True)
|
|
self._executable_path = executable_path
|
|
|
|
def get_version(
|
|
self,
|
|
callback: ResultCallback | None = None,
|
|
error_callback: ErrorCallback | None = None,
|
|
) -> None:
|
|
"""Request the OpenModelica version without blocking the caller."""
|
|
self.send_expression("getVersion()", callback, error_callback)
|
|
|
|
def build_model(
|
|
self,
|
|
model: str,
|
|
model_name: str,
|
|
callback: ResultCallback | None = None,
|
|
error_callback: ErrorCallback | None = None,
|
|
) -> None:
|
|
"""Load and build one composed model as an ordered worker operation."""
|
|
|
|
def operation(omc, _temp_dir: Path):
|
|
loaded = omc.sendExpression(f"loadString({json.dumps(model)})")
|
|
if loaded is not True:
|
|
raise RuntimeError("OpenModelica could not load the composed model")
|
|
return omc.sendExpression(f"buildModel({model_name})")
|
|
|
|
self._submit("build model", operation, callback, error_callback)
|
|
|
|
def run_model(
|
|
self,
|
|
executable: str,
|
|
arguments: list[str],
|
|
progress_callback: Callable[[SimulationProgress], None] | None = None,
|
|
message_callback: Callable[[SimulationMessage], None] | None = None,
|
|
callback: ResultCallback | None = None,
|
|
error_callback: ErrorCallback | None = None,
|
|
) -> None:
|
|
"""Run a built model and consume its XML/TCP status stream."""
|
|
|
|
def operation(omc, temp_dir: Path):
|
|
return _run_model_with_tcp(
|
|
omc,
|
|
executable,
|
|
arguments,
|
|
temp_dir,
|
|
progress_callback,
|
|
message_callback,
|
|
)
|
|
|
|
self._submit("run simulation", operation, callback, error_callback)
|
|
|
|
def send_expression(
|
|
self,
|
|
expression: str,
|
|
callback: ResultCallback | None = None,
|
|
error_callback: ErrorCallback | None = None,
|
|
*,
|
|
parsed: bool = True,
|
|
) -> None:
|
|
"""Queue an OMC expression for ordered execution on the worker thread."""
|
|
|
|
def operation(omc, _temp_dir: Path):
|
|
return omc.sendExpression(expression, parsed=parsed)
|
|
|
|
self._submit(expression, operation, callback, error_callback)
|
|
|
|
def _submit(
|
|
self,
|
|
description: str,
|
|
operation: Callable[[Any, Path], Any],
|
|
callback: ResultCallback | None,
|
|
error_callback: ErrorCallback | None,
|
|
) -> None:
|
|
request = _Request(description, operation, callback, error_callback)
|
|
self._ensure_worker().put(request)
|
|
|
|
def shutdown(self, *, wait: bool = True) -> None:
|
|
"""Ask the worker to close its OMC session after queued requests."""
|
|
|
|
with self._lifecycle_lock:
|
|
worker = self._worker
|
|
queue = self._queue
|
|
self._worker = None
|
|
self._queue = None
|
|
if queue is not None:
|
|
queue.put(None)
|
|
if wait and worker is not None and worker is not current_thread():
|
|
worker.join()
|
|
if worker is None:
|
|
self._cleanup_temp_dir()
|
|
|
|
def __del__(self) -> None:
|
|
"""Best-effort fallback; normal application shutdown is explicit."""
|
|
|
|
try:
|
|
self.shutdown(wait=False)
|
|
except Exception:
|
|
pass
|
|
|
|
def _ensure_worker(self) -> Queue[_Request | None]:
|
|
with self._lifecycle_lock:
|
|
if self._worker is not None and self._worker.is_alive():
|
|
return self._queue
|
|
temp_dir = self._ensure_temp_dir_locked()
|
|
queue: Queue[_Request | None] = Queue()
|
|
worker = Thread(
|
|
target=self._worker_main,
|
|
args=(queue, self._executable_path, temp_dir),
|
|
name="bedit-openmodelica",
|
|
daemon=True,
|
|
)
|
|
self._queue = queue
|
|
self._worker = worker
|
|
worker.start()
|
|
return queue
|
|
|
|
def _worker_main(
|
|
self,
|
|
queue: Queue[_Request | None],
|
|
executable_path: str,
|
|
temp_dir: Path,
|
|
) -> None:
|
|
omc = None
|
|
try:
|
|
while (request := queue.get()) is not None:
|
|
try:
|
|
if omc is None:
|
|
omc = _create_session(executable_path)
|
|
changed_directory = omc.sendExpression(
|
|
f"cd({json.dumps(str(temp_dir))})"
|
|
)
|
|
if not changed_directory:
|
|
raise RuntimeError(
|
|
f"OpenModelica could not use {str(temp_dir)!r}"
|
|
)
|
|
result = request.operation(omc, temp_dir)
|
|
except Exception as error:
|
|
if request.error_callback is None:
|
|
log.exception(
|
|
"OpenModelica request failed: %s", request.description
|
|
)
|
|
else:
|
|
_deliver_callback(request.error_callback, error)
|
|
else:
|
|
if request.callback is not None:
|
|
_deliver_callback(request.callback, result)
|
|
finally:
|
|
transport_files = _ompython_transport_files(omc)
|
|
if omc is not None:
|
|
try:
|
|
omc.sendExpression("quit()")
|
|
except Exception:
|
|
log.debug("Could not close OpenModelica session", exc_info=True)
|
|
for path in transport_files:
|
|
try:
|
|
path.unlink(missing_ok=True)
|
|
except OSError:
|
|
log.debug("Could not remove OMPython file %s", path, exc_info=True)
|
|
self._cleanup_temp_dir(temp_dir)
|
|
|
|
def _ensure_temp_dir_locked(self) -> Path:
|
|
if self._temp_dir is None:
|
|
self._temp_dir = Path(tempfile.mkdtemp(prefix="bedit-openmodelica-"))
|
|
return self._temp_dir
|
|
|
|
def _cleanup_temp_dir(self, expected: Path | None = None) -> None:
|
|
with self._lifecycle_lock:
|
|
if expected is not None and self._temp_dir != expected:
|
|
temp_dir = expected
|
|
else:
|
|
temp_dir = self._temp_dir
|
|
self._temp_dir = None
|
|
if temp_dir is not None:
|
|
shutil.rmtree(temp_dir, ignore_errors=True)
|
|
|
|
|
|
def _deliver_callback(callback: Callable[[Any], None], value: Any) -> None:
|
|
try:
|
|
callback(value)
|
|
except Exception:
|
|
log.exception("OpenModelica callback failed")
|
|
|
|
|
|
def _create_session(executable_path: str):
|
|
from OMPython import OMCSessionZMQ
|
|
|
|
return OMCSessionZMQ(omhome=_openmodelica_home(executable_path))
|
|
|
|
|
|
def _run_model_with_tcp(
|
|
omc,
|
|
executable: str,
|
|
arguments: list[str],
|
|
temp_dir: Path,
|
|
progress_callback: Callable[[SimulationProgress], None] | None,
|
|
message_callback: Callable[[SimulationMessage], None] | None,
|
|
) -> SimulationExecutionResult:
|
|
executable_path = Path(executable)
|
|
executable_command = executable
|
|
if not executable_path.is_absolute() and executable_path.parent == Path("."):
|
|
executable_command = f"./{executable}"
|
|
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
|
|
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
server.bind(("127.0.0.1", 0))
|
|
server.listen(1)
|
|
server.settimeout(0.25)
|
|
port = server.getsockname()[1]
|
|
command = shlex.join([
|
|
executable_command,
|
|
*arguments,
|
|
f"-port={port}",
|
|
"-logFormat=xmltcp",
|
|
])
|
|
output_path = temp_dir / "simulation-output.txt"
|
|
command_finished = Event()
|
|
reader_finished = Event()
|
|
reader_errors: list[Exception] = []
|
|
|
|
def read_progress() -> None:
|
|
try:
|
|
connection = _accept_simulation_connection(server, command_finished)
|
|
with connection, connection.makefile(
|
|
"r", encoding="utf-8"
|
|
) as stream:
|
|
for line in stream:
|
|
_handle_simulation_xml(
|
|
line, progress_callback, message_callback
|
|
)
|
|
except Exception as error:
|
|
reader_errors.append(error)
|
|
finally:
|
|
reader_finished.set()
|
|
|
|
reader = Thread(
|
|
target=read_progress,
|
|
name="bedit-simulation-progress",
|
|
daemon=True,
|
|
)
|
|
reader.start()
|
|
log.info("Starting simulation through OpenModelica: %s", command)
|
|
try:
|
|
return_code = omc.sendExpression(
|
|
f"system({json.dumps(command)}, {json.dumps(str(output_path))})"
|
|
)
|
|
finally:
|
|
command_finished.set()
|
|
if not reader_finished.wait(20.0):
|
|
raise TimeoutError("Simulation progress connection did not close")
|
|
if reader_errors:
|
|
raise reader_errors[0]
|
|
if return_code != 0:
|
|
output = output_path.read_text(errors="replace") if output_path.exists() else ""
|
|
detail = f": {output.strip()}" if output.strip() else ""
|
|
raise RuntimeError(
|
|
f"Simulation process exited with status {return_code}{detail}"
|
|
)
|
|
result_path = temp_dir / f"{Path(executable).stem}_res.csv"
|
|
if not result_path.is_file():
|
|
raise RuntimeError(
|
|
f"OpenModelica did not create the expected result file {result_path.name!r}"
|
|
)
|
|
data = load_openmodelica_csv(result_path)
|
|
log.info(
|
|
"Loaded %d result columns from %s", len(data), result_path.name
|
|
)
|
|
return SimulationExecutionResult(
|
|
return_code=int(return_code),
|
|
result_file=str(result_path),
|
|
data=data,
|
|
)
|
|
|
|
|
|
def _accept_simulation_connection(
|
|
server: socket.socket, command_finished: Event
|
|
) -> socket.socket:
|
|
deadline = time.monotonic() + 15.0
|
|
while time.monotonic() < deadline:
|
|
try:
|
|
connection, _address = server.accept()
|
|
return connection
|
|
except socket.timeout:
|
|
if command_finished.is_set():
|
|
raise RuntimeError(
|
|
"Simulation command finished before opening its progress connection"
|
|
)
|
|
raise TimeoutError("Simulation did not connect to the progress server")
|
|
|
|
|
|
def _handle_simulation_xml(
|
|
line: str,
|
|
progress_callback: Callable[[SimulationProgress], None] | None,
|
|
message_callback: Callable[[SimulationMessage], None] | None,
|
|
) -> None:
|
|
text = line.strip()
|
|
if not text:
|
|
return
|
|
try:
|
|
element = ET.fromstring(text)
|
|
except ET.ParseError:
|
|
log.warning("Invalid simulation status XML: %s", text)
|
|
return
|
|
if element.tag == "status" and progress_callback is not None:
|
|
_deliver_callback(
|
|
progress_callback,
|
|
SimulationProgress(
|
|
phase=element.get("phase", ""),
|
|
current_step_size=float(element.get("currentStepSize", 0)),
|
|
time=float(element.get("time", 0)),
|
|
progress=int(float(element.get("progress", 0))),
|
|
),
|
|
)
|
|
elif element.tag == "message":
|
|
message = SimulationMessage(
|
|
stream=element.get("stream", ""),
|
|
type=element.get("type", ""),
|
|
text=element.get("text", ""),
|
|
)
|
|
log.info("OpenModelica %s: %s", message.stream, message.text)
|
|
if message_callback is not None:
|
|
_deliver_callback(message_callback, message)
|
|
|
|
|
|
def _ompython_transport_files(omc) -> set[Path]:
|
|
"""Return only the log and port files owned by this OMPython session."""
|
|
|
|
if omc is None:
|
|
return set()
|
|
process = getattr(omc, "omc_process", None)
|
|
if process is None:
|
|
return set()
|
|
files: set[Path] = set()
|
|
temp_dir = getattr(process, "_temp_dir", None)
|
|
file_base = getattr(process, "_omc_filebase", None)
|
|
if temp_dir is not None and file_base:
|
|
files.add(Path(temp_dir) / f"{file_base}.log")
|
|
try:
|
|
port_file = process._get_portfile_path()
|
|
except Exception:
|
|
port_file = None
|
|
if port_file is not None:
|
|
files.add(Path(port_file))
|
|
return files
|
|
|
|
|
|
def _openmodelica_home(executable_path: str) -> str | None:
|
|
"""Convert an optional omc executable path to the home expected by OMPython."""
|
|
|
|
if not executable_path.strip():
|
|
return None
|
|
path = Path(os.path.expandvars(executable_path)).expanduser()
|
|
if path.name.lower() in {"omc", "omc.exe"}:
|
|
return str(path.parent.parent)
|
|
return str(path)
|