diff --git a/doc/changelog.d/3627.miscellaneous.md b/doc/changelog.d/3627.miscellaneous.md new file mode 100644 index 00000000000..0d10efb2132 --- /dev/null +++ b/doc/changelog.d/3627.miscellaneous.md @@ -0,0 +1 @@ +Feat: first tentative for plugin mapdl mechanism python api \ No newline at end of file diff --git a/src/ansys/mapdl/core/errors.py b/src/ansys/mapdl/core/errors.py index 60ca407ef1e..7b8ce53b385 100644 --- a/src/ansys/mapdl/core/errors.py +++ b/src/ansys/mapdl/core/errors.py @@ -366,6 +366,27 @@ def __init__(self, msg=""): super().__init__(msg) +class PluginError(MapdlRuntimeError): + """Raised when a plugin fails""" + + def __init__(self, msg=""): + super().__init__(msg) + + +class PluginLoadError(PluginError): + """Raised when a plugin fails to load""" + + def __init__(self, msg=""): + super().__init__(msg) + + +class PluginUnloadError(PluginError): + """Raised when a plugin fails to unload""" + + def __init__(self, msg=""): + super().__init__(msg) + + # handler for protect_grpc def handler(sig, frame): # pragma: no cover """Pass signal to custom interrupt handler.""" diff --git a/src/ansys/mapdl/core/mapdl_core.py b/src/ansys/mapdl/core/mapdl_core.py index c2642e3f59d..f3785bf6bfe 100644 --- a/src/ansys/mapdl/core/mapdl_core.py +++ b/src/ansys/mapdl/core/mapdl_core.py @@ -85,6 +85,7 @@ from ansys.mapdl.core.mapdl import MapdlBase from ansys.mapdl.core.mapdl_geometry import Geometry, LegacyGeometry from ansys.mapdl.core.parameters import Parameters + from ansys.mapdl.core.plugin import ansPlugin from ansys.mapdl.core.solution import Solution from ansys.mapdl.core.xpl import ansXpl @@ -350,6 +351,8 @@ def __init__( self._xpl: Optional[ansXpl] = None # Initialized in mapdl_grpc + self._plugin: Optional[ansPlugin] = None # Initialized in mapdl_grpc + from ansys.mapdl.core.component import ComponentManager self._componentmanager: ComponentManager = ComponentManager(self) @@ -1081,6 +1084,26 @@ def graphics_backend(self, value: GraphicsBackend): """Set the graphics backend to be used.""" self._graphics_backend = value + @property + def plugins(self) -> "ansPlugin": + """MAPDL plugin handler + + Plugin Manager for MAPDL + + Examples + -------- + + >>> from ansys import Mapdl + >>> mapdl = Mapdl() + >>> plugin = mapdl.plugin + >>> plugin.load('PluginDPF') + """ + if self._plugin is None: + from ansys.mapdl.core.plugin import ansPlugin + + self._plugin = ansPlugin(self) + return self._plugin + @property @requires_package("ansys.mapdl.reader", softerror=True) def result(self): diff --git a/src/ansys/mapdl/core/plugin.py b/src/ansys/mapdl/core/plugin.py new file mode 100644 index 00000000000..373f83aed1d --- /dev/null +++ b/src/ansys/mapdl/core/plugin.py @@ -0,0 +1,244 @@ +# Copyright (C) 2016 - 2025 ANSYS, Inc. and/or its affiliates. +# SPDX-License-Identifier: MIT +# +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all +# copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. + +"""Contains the ansPlugin class.""" +import re +from warnings import warn +import weakref + +from ansys.mapdl.core import Mapdl +from ansys.mapdl.core.errors import PluginError, PluginLoadError, PluginUnloadError +from ansys.mapdl.core.logging import Logger + + +class ansPlugin: + """ + ANSYS MAPDL Plugin Manager. + + Examples + -------- + >>> from ansys.mapdl.core import launch_mapdl + >>> mapdl = launch_mapdl() + >>> plugin = mapdl.plugin + + Load a plugin in the MAPDL Session + """ + + def __init__(self, mapdl: Mapdl): + """Initialize the class.""" + from ansys.mapdl.core.mapdl_grpc import MapdlGrpc + + if not isinstance(mapdl, MapdlGrpc): # pragma: no cover + raise TypeError("Must be initialized using an 'MapdlGrpc' object") + + self._mapdl_weakref = weakref.ref(mapdl) + self._filename = None + self._open = False + + @property + def _mapdl(self) -> Mapdl: + """Return the weakly referenced instance of mapdl.""" + return self._mapdl_weakref() + + @property + def _log(self) -> Logger: + """Return the logger from the MAPDL instance.""" + return self._mapdl._log + + def _parse_commands(self, response: str) -> list[str]: + """ + Parse the response string to extract commands. + + Parameters + ---------- + response : str + The response string containing commands. + + Returns + ------- + list[str] + A list of commands extracted from the response. + """ + if not response: + return [] + + # Assuming commands are separated by newlines + return re.findall(r"New command \[(.*)\] registered", response) + + def _set_commands(self, commands: list[str], plugin_name: str = "NOT_SET") -> None: + """ + Set commands to be executed. + + Parameters + ---------- + commands : list[str] + List of commands to be set. + """ + if not commands: + return + + mapdl = self._mapdl + + for each_command in commands: + each_command = each_command.replace("*", "star").replace("/", "slash") + + if hasattr(mapdl, each_command): + # We are allowing to overwrite existing commands + warn(f"Command '{each_command}' already exists in the MAPDL instance.") + + def passer(self, *args, **kwargs): + return self.run(*args, **kwargs) + + # Inject docstring + passer.__doc__ = f"""Command from plugin {plugin_name}: {each_command}. + Use this plugin documentation to understand the command and its parameters. + + Automatically generated docstring by ansPlugin. + """ + setattr(mapdl, each_command, passer) + self._log.info( + f"Command '{each_command}' from plugin '{plugin_name}' set successfully." + ) + + def _deleter_commands( + self, commands: list[str], plugin_name: str = "NOT_SET" + ) -> None: + """ + Delete commands from the MAPDL instance. + + Parameters + ---------- + commands : list[str] + List of commands to be deleted. + """ + if not commands: + return + + mapdl = self._mapdl + + for each_command in commands: + if hasattr(mapdl, each_command): + delattr(mapdl, each_command) + self._log.info( + f"Command '{each_command}' from '{plugin_name}' deleted successfully." + ) + + def _load_commands(self, response: str, plugin_name: str) -> None: + """ + Load commands from the response string. + + Parameters + ---------- + response : str + The response string containing commands to be loaded. + """ + if not response: + return + + commands = self._parse_commands(response) + if not commands: + self._log.warning("No commands found in the response.") + return + self._set_commands(commands, plugin_name=plugin_name) + + def load(self, plugin_name: str, feature: str = "") -> str: + """ + Loads a plugin into MAPDL. + + Parameters + ---------- + plugin_name : str + Name of the plugin to load. + feature : str + Feature or module to activate in the plugin. + + Raises + ------ + PluginLoadError + If the plugin fails to load. + """ + + command = f"*PLUG,LOAD,{plugin_name},{feature}" + response = self._mapdl.run(command) + if "error" in response.lower(): + raise PluginLoadError( + f"Failed to load plugin '{plugin_name}' with feature '{feature}'." + ) + self._log.info( + f"Plugin '{plugin_name}' with feature '{feature}' loaded successfully." + ) + self._load_commands(response, plugin_name=plugin_name) + return response + + def unload(self, plugin_name: str) -> str: + """ + Unloads a plugin from MAPDL. + + Parameters + ---------- + plugin_name : str + Name of the plugin to unload. + + Raises + ------ + PluginUnloadError + If the plugin fails to unload. + """ + + command = f"*PLUG,UNLOAD,{plugin_name}" + response = self._mapdl.run(command) + + if not response: + return "" + + if "error" in response.lower(): + raise PluginUnloadError(f"Failed to unload plugin '{plugin_name}'.") + + self._log.info(f"Plugin '{plugin_name}' unloaded successfully.") + + commands = self._parse_commands(response) + self._deleter_commands(commands, plugin_name=plugin_name) + + return response + + def list(self) -> list[str]: + """ + Lists all currently loaded plugins in MAPDL. + + Returns + ------- + list + A list of loaded plugin names. + + Raises + ------ + RuntimeError + If the plugin list cannot be retrieved. + """ + + command = "*PLUG,LIST" + response = self._mapdl.run(command) or "" + if response and "error" in response.lower(): + raise PluginError("Failed to retrieve the list of loaded plugins.") + + # Parse response and extract plugin names (assuming response is newline-separated text) + return [] diff --git a/tests/test_plugin.py b/tests/test_plugin.py new file mode 100644 index 00000000000..2c973f29c10 --- /dev/null +++ b/tests/test_plugin.py @@ -0,0 +1,103 @@ +# Copyright (C) 2016 - 2025 ANSYS, Inc. and/or its affiliates. +# SPDX-License-Identifier: MIT +# +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all +# copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. + +"""Test the plugin implementation""" + +import pytest + +pytestmark = pytest.mark.random_order(disabled=True) + +from ansys.mapdl.core import Mapdl +from ansys.mapdl.core.plugin import ansPlugin + +pytestmark = pytest.mark.random_order(disabled=True) + +TEST_PLUGIN = "PluginDPF" + + +@pytest.fixture() +def plugins(mapdl: Mapdl) -> ansPlugin: + if mapdl.version < 25.2: + pytest.skip( + "Plugin support is only for versions 25.2 and above", + allow_module_level=True, + ) + + return mapdl.plugins + + +@pytest.fixture() +def dpf_load_response(plugins: ansPlugin) -> ansPlugin: + yield plugins.load(TEST_PLUGIN) + plugins.unload(TEST_PLUGIN) + + +def test_plugin_load(plugins): + assert plugins.load(TEST_PLUGIN) is not None + + +@pytest.mark.xfail(reason="Plugin unload not implemented in MAPDL yet") +def test_plugin_list(plugins, dpf_load_response): + assert TEST_PLUGIN in plugins.list(), "Plugin should be loaded" + + +def test_plugin_unload(plugins): + plugins.unload(TEST_PLUGIN) + assert TEST_PLUGIN not in plugins.list(), "Plugin should be unloaded" + + +def test_parse_commands(plugins, dpf_load_response): + commands = plugins._parse_commands(dpf_load_response) + + assert isinstance(commands, list), "Commands should be a list" + assert len(commands) > 0, "Commands list should not be empty" + assert "*DPF" in commands, "Expected command '*DPF' should be in the list" + + +def test_load_commands(plugins, dpf_load_response): + commands = plugins._parse_commands(dpf_load_response) + assert isinstance(commands, list), "Commands should be a list" + assert len(commands) > 0, "Commands list should not be empty" + + for command in commands: + assert hasattr(plugins._mapdl, command) + + +def test_deleter_commands(plugins, dpf_load_response): + commands = plugins._parse_commands(dpf_load_response) + assert isinstance(commands, list), "Commands should be a list" + assert len(commands) > 0, "Commands list should not be empty" + + plugins._deleter_commands(commands, TEST_PLUGIN) + + for command in commands: + assert not hasattr( + plugins._mapdl, command + ), f"Command {command} should be deleted" + + +def test_unload_plugin_twice(plugins): + plugins.load(TEST_PLUGIN) + assert f"Close of the {TEST_PLUGIN} Plugin" in plugins.unload(TEST_PLUGIN) + assert ( + plugins.unload(TEST_PLUGIN) == "" + ), "Unloading a plugin twice should return an empty string"