import json import os import shutil import tempfile from collections.abc import Callable from dataclasses import dataclass from pathlib import Path from queue import Queue from threading import Lock, Thread, current_thread from typing import Any from bedit.core.application_log import get_logger log = get_logger(__name__) ResultCallback = Callable[[Any], None] ErrorCallback = Callable[[Exception], None] @dataclass(frozen=True) class _Request: expression: str parsed: bool callback: ResultCallback | None error_callback: ErrorCallback | None class OpenModelicaInterface: """Asynchronous, persistent interface to one OpenModelica session. The worker and OMC session are created lazily for the first request. Callback functions execute on the worker thread and must not manipulate Qt widgets. """ 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 load_model( self, model: str, callback: ResultCallback | None = None, error_callback: ErrorCallback | None = None, ) -> None: """Load a composed model into OpenModelica""" self.send_expression(f"loadString({json.dumps(model)})", callback, error_callback) def compile_model( self, model_name: str, callback: ResultCallback | None = None, error_callback: ErrorCallback | None = None, ) -> None: """Compile a loaded model in OpenModelica""" self.send_expression(f"buildModel({model_name})", callback, error_callback) def exec_system( self, cmd: str, callback: ResultCallback | None = None, error_callback: ErrorCallback | None = None, ) -> None: """Execute system command from OpenModelica""" self.send_expression(f"system(\"{cmd}\", \"system_out.txt\")", 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.""" request = _Request(expression, parsed, 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 = omc.sendExpression( request.expression, parsed=request.parsed ) except Exception as error: if request.error_callback is None: log.exception( "OpenModelica request failed: %s", request.expression ) 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 _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)