diff --git a/README.md b/README.md index 10d8b87..a67439c 100644 --- a/README.md +++ b/README.md @@ -138,6 +138,32 @@ The following example prompts show how you can interact with the VEOS MCP server +
+OSA Creation + +- **veos_create_osa** + - Title: Create a new VEOS OSA file + - Description: Creates and saves a new, empty VEOS offline simulation application using the headless VEOS Model Console. + - Parameters: + - `osa_path` (string): Destination path of the OSA file. The parent directory must already exist. + - Read-only: **false** + +
+ +
+File Import + +- **veos_import_file** + - Title: Import a file into a VEOS simulation system + - Description: Imports a VEOS-supported file into an existing OSA and saves it using the headless VEOS Model or Build Console. Supported types are OSA, JSON, FMU, SIC, BSC, SMC, classic VECU, and adaptive VECU. + - Parameters: + - `osa_path` (string): Path to the OSA simulation system to modify. + - `import_file_path` (string): Path to the file to import using the default VEOS import and build settings. + - `file_type` (string, optional): Explicit import type. Required for VECU (`classic-vecu` or `adaptive-vecu`) and optional for other supported types. + - Read-only: **false** + +
+
Simulator Control diff --git a/src/veos_mcp/runtime.py b/src/veos_mcp/runtime.py index 98a92d7..01e9fd5 100644 --- a/src/veos_mcp/runtime.py +++ b/src/veos_mcp/runtime.py @@ -9,10 +9,9 @@ from veos_mcp.models.errors import ErrorType, VeosError from veos_mcp.veos_cli import VeosCli from veos_mcp.veos_path_resolver import ( - check_veos_installation_exists, get_linux_installations, get_windows_installations, - resolve_veos_path_for_version, + resolve_veos_installation, ) mcp = FastMCP( @@ -25,17 +24,17 @@ def configure_cli(*, veos_version: str | None, veos_bin_path: str | None) -> None: - """Configure the server with the VEOS CLI installation directory.""" + """Configure the server with the selected VEOS installation.""" veos_installations = get_windows_installations() if sys.platform.startswith("win32") else get_linux_installations() - - if veos_bin_path is not None: - veos_path = check_veos_installation_exists(veos_installations, veos_bin_path) - else: - veos_path = resolve_veos_path_for_version(veos_installations, veos_version) + installation = resolve_veos_installation( + veos_installations, + veos_version=veos_version, + veos_bin_path=veos_bin_path, + ) global _veos_cli - _veos_cli = VeosCli(veos_path=veos_path) + _veos_cli = VeosCli(veos_path=installation.executable_path) def get_cli() -> VeosCli: diff --git a/src/veos_mcp/tools/__init__.py b/src/veos_mcp/tools/__init__.py index 57ef11e..788f252 100644 --- a/src/veos_mcp/tools/__init__.py +++ b/src/veos_mcp/tools/__init__.py @@ -1,12 +1,16 @@ """MCP tools exposed by the VEOS MCP server.""" +from veos_mcp.tools import file_import as file_import from veos_mcp.tools import log_file_access as log_file_access +from veos_mcp.tools import osa_creation as osa_creation from veos_mcp.tools import sim_control as sim_control from veos_mcp.tools import system_extraction as system_extraction from veos_mcp.tools import system_modification as system_modification __all__ = [ + "file_import", "log_file_access", + "osa_creation", "sim_control", "system_extraction", "system_modification", diff --git a/src/veos_mcp/tools/file_import.py b/src/veos_mcp/tools/file_import.py new file mode 100644 index 0000000..3b84600 --- /dev/null +++ b/src/veos_mcp/tools/file_import.py @@ -0,0 +1,97 @@ +"""MCP tool for importing files with the VEOS Model and Build Consoles.""" + +from pathlib import Path +from typing import Literal + +from mcp.types import CallToolResult, ToolAnnotations + +from veos_mcp.runtime import ( + create_command_result_response_error, + create_command_result_response_success, + create_error_response, + get_cli, + mcp, +) + +ImportFileType = Literal[ + "osa", + "json", + "fmu", + "sic", + "bsc", + "smc", + "classic-vecu", + "adaptive-vecu", +] + +_MODEL_IMPORT_TYPES = {"osa", "json"} +_BUILD_TYPES_BY_EXTENSION = { + ".fmu": "fmu", + ".sic": "sic", + ".bsc": "bsc", + ".smc": "smc", +} + + +def _resolve_file_type(import_file_path: str, file_type: ImportFileType | None) -> str | None: + if file_type is not None: + return file_type + + extension = Path(import_file_path).suffix.lower() + if extension in {".osa", ".json"}: + return extension.removeprefix(".") + return _BUILD_TYPES_BY_EXTENSION.get(extension) + + +@mcp.tool( + name="veos_import_file", + title="Import a file into a VEOS simulation system", + description=( + "Imports a VEOS-supported file into an existing OSA and saves the updated OSA using the headless VEOS " + "Model or Build Console. File types OSA, JSON, FMU, SIC, BSC, and SMC are inferred from their extension. " + "For VECU files, file_type must be classic-vecu or adaptive-vecu. An explicit file_type can also be used " + "for a supported container with a nonstandard extension. VEOS uses the default build settings." + ), + annotations=ToolAnnotations( + readOnlyHint=False, + destructiveHint=False, + idempotentHint=False, + openWorldHint=False, + ), +) +def veos_import_file( + osa_path: str, + import_file_path: str, + file_type: ImportFileType | None = None, +) -> CallToolResult: + """Import a supported file into an OSA simulation system.""" + resolved_file_type = _resolve_file_type(import_file_path, file_type) + if resolved_file_type is None: + return create_error_response( + f"Could not determine the VEOS import type for {import_file_path}. " + "Specify file_type; VECU files require classic-vecu or adaptive-vecu." + ) + + if resolved_file_type in _MODEL_IMPORT_TYPES: + command_result = get_cli().run_model( + "import", + osa_path, + "--path", + import_file_path, + "--save-only-on-success", + ) + else: + command_result = get_cli().run_build( + resolved_file_type, + import_file_path, + "--output-file", + osa_path, + ) + + if not command_result.success: + return create_command_result_response_error( + command_result, + f"Failed to import file {import_file_path} into VEOS OSA {osa_path}.", + ) + + return create_command_result_response_success(command_result) diff --git a/src/veos_mcp/tools/osa_creation.py b/src/veos_mcp/tools/osa_creation.py new file mode 100644 index 0000000..0fc8fd4 --- /dev/null +++ b/src/veos_mcp/tools/osa_creation.py @@ -0,0 +1,42 @@ +"""MCP tool for creating VEOS simulation systems with the Model Console.""" + +from mcp.types import CallToolResult, ToolAnnotations + +from veos_mcp.runtime import ( + create_command_result_response_error, + create_command_result_response_success, + get_cli, + mcp, +) + + +@mcp.tool( + name="veos_create_osa", + title="Create a new VEOS OSA file", + description=( + "Creates and saves a new, empty VEOS offline simulation application (OSA) at the specified path using the " + "headless VEOS Model Console. The parent directory must already exist." + ), + annotations=ToolAnnotations( + readOnlyHint=False, + destructiveHint=False, + idempotentHint=False, + openWorldHint=False, + ), +) +def veos_create_osa(osa_path: str) -> CallToolResult: + """Create a new OSA simulation system.""" + command_result = get_cli().run_model( + "modify", + osa_path, + "--modifications", + "[]", + "--create-new", + "--save-only-on-success", + "--verbosity", + "Error", + ) + if not command_result.success: + return create_command_result_response_error(command_result, f"Failed to create VEOS OSA file {osa_path}.") + + return create_command_result_response_success(command_result) diff --git a/src/veos_mcp/veos_cli.py b/src/veos_mcp/veos_cli.py index 4979059..74e9217 100644 --- a/src/veos_mcp/veos_cli.py +++ b/src/veos_mcp/veos_cli.py @@ -8,6 +8,7 @@ from veos_mcp.models.cli_command_result import CliCommandResult, CommandResultCode DEFAULT_COMMAND_TIMEOUT_SECONDS = 40.0 +DEFAULT_BUILD_COMMAND_TIMEOUT_SECONDS = 600.0 _COMMAND_GATE = Lock() @@ -130,9 +131,11 @@ def __init__( *, veos_path: Path, command_timeout_seconds: float = DEFAULT_COMMAND_TIMEOUT_SECONDS, + build_command_timeout_seconds: float = DEFAULT_BUILD_COMMAND_TIMEOUT_SECONDS, ) -> None: self.veos_path = veos_path self.command_timeout_seconds = command_timeout_seconds + self.build_command_timeout_seconds = build_command_timeout_seconds def run_sim(self, *arguments: str) -> CliCommandResult: """Run the VEOS simulator CLI with serialized access.""" @@ -142,14 +145,24 @@ def run_model(self, *arguments: str) -> CliCommandResult: """Run the VEOS model CLI with serialized access.""" return self._run_locked(self.veos_path, ("model", *arguments)) + def run_build(self, *arguments: str) -> CliCommandResult: + """Run the VEOS build CLI with serialized access.""" + return self._run_locked( + self.veos_path, + ("build", *arguments), + command_timeout_seconds=self.build_command_timeout_seconds, + ) + def _run_locked( self, executable_path: Path, arguments: tuple[str, ...], + *, + command_timeout_seconds: float | None = None, ) -> CliCommandResult: with _COMMAND_GATE: return run_process_command( executable_path, arguments, - command_timeout_seconds=self.command_timeout_seconds, + command_timeout_seconds=(self.command_timeout_seconds if command_timeout_seconds is None else command_timeout_seconds), ) diff --git a/src/veos_mcp/veos_path_resolver.py b/src/veos_mcp/veos_path_resolver.py index 7a968be..d1614ab 100644 --- a/src/veos_mcp/veos_path_resolver.py +++ b/src/veos_mcp/veos_path_resolver.py @@ -25,6 +25,11 @@ class _VeosInstallation: bin_path: Path source_name: str + @property + def executable_path(self) -> Path: + executable_name = "veos.exe" if sys.platform.startswith("win32") else "veos" + return self.bin_path / executable_name + def _normalize_release(release: str | None) -> str | None: if release is None: @@ -163,16 +168,33 @@ def get_linux_installations( def resolve_veos_path_for_version(installations: list[_VeosInstallation], veos_version: str | None) -> Path: - candidate_installation = _select_installation(installations, veos_version) - return candidate_installation.bin_path / "veos.exe" if sys.platform.startswith("win32") else candidate_installation.bin_path / "veos" + return _select_installation(installations, veos_version).executable_path def check_veos_installation_exists(installations: list[_VeosInstallation], veos_bin_path: str) -> Path: for installation in installations: if installation.bin_path == Path(veos_bin_path): - return installation.bin_path / "veos.exe" if sys.platform.startswith("win32") else installation.bin_path / "veos" + return installation.executable_path raise ValueError( f"Provided VEOS path is invalid, make sure that the provided path '{veos_bin_path}' " f"points to the bin directory of an existing VEOS installation. " f"Following installations were found: {', '.join(str(installation.bin_path) for installation in installations)}." ) + + +def resolve_veos_installation( + installations: list[_VeosInstallation], + *, + veos_version: str | None, + veos_bin_path: str | None, +) -> _VeosInstallation: + if veos_bin_path is None: + return _select_installation(installations, veos_version) + + requested_bin_path = Path(veos_bin_path) + for installation in installations: + if installation.bin_path == requested_bin_path: + return installation + + check_veos_installation_exists(installations, veos_bin_path) + raise AssertionError("Unreachable") diff --git a/tests/test_mcp_surface_smoketest.py b/tests/test_mcp_surface_smoketest.py index 8c61684..2b21431 100644 --- a/tests/test_mcp_surface_smoketest.py +++ b/tests/test_mcp_surface_smoketest.py @@ -3,6 +3,7 @@ import asyncio import os import sys +from typing import Any from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client @@ -22,8 +23,10 @@ def test_list_all_tools() -> None: expected_tools = [ "veos_add_signal_connections", "veos_apply_config", + "veos_create_osa", "veos_get_all_signals_and_ports", "veos_get_log_file", + "veos_import_file", "veos_list_all_available_log_files", "veos_load", "veos_remove_signal_connections", @@ -46,6 +49,33 @@ async def list_tool_names() -> list[str]: assert actual_tools == expected_tools +def test_import_tool_exposes_all_cli_file_types() -> None: + """Test that clients can select every supported CLI import route.""" + + async def get_file_type_schema() -> dict[str, Any]: + server_params = create_server_params() + + async with stdio_client(server_params) as (read, write): + async with ClientSession(read, write) as session: + await session.initialize() + tools = await session.list_tools() + import_tool = next(tool for tool in tools.tools if tool.name == "veos_import_file") + return import_tool.inputSchema["properties"]["file_type"] + + file_type_schema = asyncio.run(get_file_type_schema()) + + assert file_type_schema["anyOf"][0]["enum"] == [ + "osa", + "json", + "fmu", + "sic", + "bsc", + "smc", + "classic-vecu", + "adaptive-vecu", + ] + + def test_list_all_resource_templates() -> None: """Test that all expected resource templates can be listed from the stdio server.""" diff --git a/tests/test_veos_cli.py b/tests/test_veos_cli.py new file mode 100644 index 0000000..5237eea --- /dev/null +++ b/tests/test_veos_cli.py @@ -0,0 +1,32 @@ +"""Tests for VEOS CLI subprocess routing.""" + +from pathlib import Path +from unittest.mock import patch + +from veos_mcp.models.cli_command_result import CliCommandResult, CommandResultCode +from veos_mcp.veos_cli import VeosCli + + +def test_run_build_uses_build_command_and_build_timeout() -> None: + expected_result = CliCommandResult( + success=True, + exit_code=0, + code=CommandResultCode.OK, + stdout="built", + stderr="", + ) + cli = VeosCli( + veos_path=Path("C:/VEOS/veos.exe"), + command_timeout_seconds=40, + build_command_timeout_seconds=600, + ) + + with patch("veos_mcp.veos_cli.run_process_command", return_value=expected_result) as run_command: + result = cli.run_build("fmu", "model.fmu", "--output-file", "target.osa") + + assert result is expected_result + run_command.assert_called_once_with( + Path("C:/VEOS/veos.exe"), + ("build", "fmu", "model.fmu", "--output-file", "target.osa"), + command_timeout_seconds=600, + ) diff --git a/tests/tool_test_helpers.py b/tests/tool_test_helpers.py index df09afd..b1b35ab 100644 --- a/tests/tool_test_helpers.py +++ b/tests/tool_test_helpers.py @@ -18,6 +18,16 @@ def run_model(self, *arguments: str) -> CliCommandResult: return self._result +class RecordingBuildCliMock: + def __init__(self, result: CliCommandResult) -> None: + self._result = result + self.build_calls: list[tuple[str, ...]] = [] + + def run_build(self, *arguments: str) -> CliCommandResult: + self.build_calls.append(arguments) + return self._result + + class RecordingSimCliMock: def __init__(self, sim_result: CliCommandResult) -> None: self._sim_result = sim_result diff --git a/tests/tools/test_file_import.py b/tests/tools/test_file_import.py new file mode 100644 index 0000000..4f96181 --- /dev/null +++ b/tests/tools/test_file_import.py @@ -0,0 +1,82 @@ +"""Tests for the VEOS file import tool.""" + +from typing import cast + +import pytest +from mcp.types import CallToolResult + +from tests.tool_test_helpers import RecordingBuildCliMock, RecordingModelCliMock, assert_error_text_content +from veos_mcp import runtime +from veos_mcp.models.cli_command_result import CliCommandResult, CommandResultCode +from veos_mcp.tools.file_import import veos_import_file + + +def create_command_result(*, success: bool = True) -> CliCommandResult: + return CliCommandResult( + success=success, + exit_code=0 if success else 1, + code=CommandResultCode.OK if success else CommandResultCode.PROCESS_FAILED, + stdout="imported" if success else "", + stderr="" if success else "failed", + ) + + +@pytest.mark.parametrize( + ("import_file_path", "expected_build_type"), + [ + ("model.fmu", "fmu"), + ("model.sic", "sic"), + ("model.bsc", "bsc"), + ("model.smc", "smc"), + ], +) +def test_import_file_infers_and_builds_container_type(monkeypatch, import_file_path: str, expected_build_type: str) -> None: + cli = RecordingBuildCliMock(create_command_result()) + monkeypatch.setattr(runtime, "_veos_cli", cli) + + result = cast(CallToolResult, veos_import_file("target.osa", import_file_path)) + + assert cli.build_calls == [(expected_build_type, import_file_path, "--output-file", "target.osa")] + assert result.isError is False + + +@pytest.mark.parametrize("file_type", ["classic-vecu", "adaptive-vecu"]) +def test_import_file_builds_explicit_vecu_type(monkeypatch, file_type: str) -> None: + cli = RecordingBuildCliMock(create_command_result()) + monkeypatch.setattr(runtime, "_veos_cli", cli) + + result = cast(CallToolResult, veos_import_file("target.osa", "model.vecu", file_type)) + + assert cli.build_calls == [(file_type, "model.vecu", "--output-file", "target.osa")] + assert result.isError is False + + +@pytest.mark.parametrize("import_file_path", ["source.osa", "participant.json"]) +def test_import_file_uses_model_console_for_osa_and_json(monkeypatch, import_file_path: str) -> None: + cli = RecordingModelCliMock(create_command_result()) + monkeypatch.setattr(runtime, "_veos_cli", cli) + + result = cast(CallToolResult, veos_import_file("target.osa", import_file_path)) + + assert cli.model_calls == [("import", "target.osa", "--path", import_file_path, "--save-only-on-success")] + assert result.isError is False + + +def test_import_file_requires_explicit_type_for_vecu(monkeypatch) -> None: + result = cast(CallToolResult, veos_import_file("target.osa", "model.vecu")) + + assert result.isError is True + assert_error_text_content( + result, + "Could not determine the VEOS import type for model.vecu. Specify file_type; VECU files require classic-vecu or adaptive-vecu.", + ) + + +def test_import_file_returns_command_error(monkeypatch) -> None: + cli = RecordingBuildCliMock(create_command_result(success=False)) + monkeypatch.setattr(runtime, "_veos_cli", cli) + + result = cast(CallToolResult, veos_import_file("target.osa", "model.fmu")) + + assert result.isError is True + assert_error_text_content(result, "Failed to import file model.fmu into VEOS OSA target.osa.") diff --git a/tests/tools/test_osa_creation.py b/tests/tools/test_osa_creation.py new file mode 100644 index 0000000..1da2879 --- /dev/null +++ b/tests/tools/test_osa_creation.py @@ -0,0 +1,57 @@ +"""Tests for the VEOS OSA creation tool.""" + +from typing import cast + +from mcp.types import CallToolResult + +from tests.tool_test_helpers import ( + RecordingModelCliMock, + assert_command_result_structured_content, + assert_error_text_content, +) +from veos_mcp import runtime +from veos_mcp.models.cli_command_result import CliCommandResult, CommandResultCode +from veos_mcp.tools.osa_creation import veos_create_osa + + +def create_command_result(*, success: bool) -> CliCommandResult: + return CliCommandResult( + success=success, + exit_code=0 if success else 1, + code=CommandResultCode.OK if success else CommandResultCode.PROCESS_FAILED, + stdout="created" if success else "", + stderr="" if success else "failed", + ) + + +def test_create_osa_invokes_headless_model_console(monkeypatch) -> None: + cli = RecordingModelCliMock(create_command_result(success=True)) + monkeypatch.setattr(runtime, "_veos_cli", cli) + + result = cast(CallToolResult, veos_create_osa("new-system.osa")) + + assert cli.model_calls == [ + ( + "modify", + "new-system.osa", + "--modifications", + "[]", + "--create-new", + "--save-only-on-success", + "--verbosity", + "Error", + ) + ] + assert result.isError is False + assert result.structuredContent is not None + assert_command_result_structured_content("veos_create_osa", result.structuredContent) + + +def test_create_osa_returns_error_when_model_console_fails(monkeypatch) -> None: + cli = RecordingModelCliMock(create_command_result(success=False)) + monkeypatch.setattr(runtime, "_veos_cli", cli) + + result = cast(CallToolResult, veos_create_osa("existing.osa")) + + assert result.isError is True + assert_error_text_content(result, "Failed to create VEOS OSA file existing.osa.")