Implemented solver_workflow in solver mode

codegen/datamodelgen.py

@@ -107,7 +107,14 @@ class DataModelGenerator:
     def __init__(self):
         self.version = get_version_for_filepath()
         self._static_info: Dict[str, DataModelStaticInfo] = {
-            "workflow": DataModelStaticInfo("workflow", ("meshing",), self.version),
+            "workflow": DataModelStaticInfo(
+                "workflow",
+                (
+                    "meshing",
+                    "solver",
+                ),
+                self.version,
+            ),
             "meshing": DataModelStaticInfo("meshing", ("meshing",), self.version),
             "PartManagement": DataModelStaticInfo(
                 "PartManagement", ("meshing",), self.version
@@ -121,7 +128,14 @@ def __init__(self):
             "preferences": DataModelStaticInfo(
                 "preferences", ("meshing", "solver", "flicing,"), self.version
             ),
+            "solverworkflow": DataModelStaticInfo(
+                "solverworkflow", ("solver",), self.version
+            )
+            if int(self.version) >= 231
+            else None,
         }
+        if not self._static_info["solverworkflow"]:
+            del self._static_info["solverworkflow"]
         self._delete_generated_files()
         self._populate_static_info()
 

src/ansys/fluent/core/session.py

@@ -32,16 +32,24 @@ def parse_server_info_file(filename: str):
     return ip, port, password
 
 
-def _get_preferences(session):
+def _get_datamodel_attributes(session, attribute: str):
     try:
         preferences_module = importlib.import_module(
-            f"ansys.fluent.core.datamodel_{session.version}.preferences"
+            f"ansys.fluent.core.datamodel_{session.version}." + attribute
         )
-        return preferences_module.Root(session._se_service, "preferences", [])
+        return preferences_module.Root(session._se_service, attribute, [])
     except (ImportError, ModuleNotFoundError):
         LOG.warning(_CODEGEN_MSG_DATAMODEL)
 
 
+def _get_preferences(session):
+    return _get_datamodel_attributes(session, "preferences")
+
+
+def _get_solverworkflow(session):
+    return _get_datamodel_attributes(session, "solverworkflow")
+
+
 class _BaseSession:
     """Instantiates a Fluent connection.
 
@@ -76,6 +84,7 @@ def __init__(self, fluent_connection: _FluentConnection):
         self.scheme_eval = self.fluent_connection.scheme_eval
         self._uploader = None
         self._preferences = None
+        self._solverworkflow = None
 
     @classmethod
     def create_from_server_info_file(
@@ -260,6 +269,7 @@ def __init__(
 
         self._uploader = None
         self._preferences = None
+        self._solverworkflow = None
 
     @classmethod
     def create_from_server_info_file(
@@ -360,6 +370,13 @@ def preferences(self):
             self._preferences = _get_preferences(self)
         return self._preferences
 
+    @property
+    def solverworkflow(self):
+        """solverworkflow datamodel root."""
+        if self._solverworkflow is None:
+            self._solverworkflow = _get_solverworkflow(self)
+        return self._solverworkflow
+
     class Solver:
         def __init__(self, fluent_connection: _FluentConnection):
             self._fluent_connection = fluent_connection

src/ansys/fluent/core/session_base_meshing.py

@@ -2,12 +2,12 @@
 
 from ansys.fluent.core.fluent_connection import _FluentConnection
 from ansys.fluent.core.meshing.meshing import Meshing
-from ansys.fluent.core.meshing.workflow import MeshingWorkflow
 from ansys.fluent.core.services.datamodel_se import PyMenuGeneric
 from ansys.fluent.core.services.datamodel_tui import TUIMenuGeneric
 from ansys.fluent.core.session_shared import _CODEGEN_MSG_DATAMODEL, _CODEGEN_MSG_TUI
 from ansys.fluent.core.utils.fluent_version import get_version_for_filepath
 from ansys.fluent.core.utils.logging import LOG
+from ansys.fluent.core.workflow import WorkflowWrapper
 
 
 class _BaseMeshing:
@@ -89,7 +89,7 @@ def _workflow_se(self):
     @property
     def workflow(self):
         if not self._workflow:
-            self._workflow = MeshingWorkflow(self._workflow_se, self.meshing)
+            self._workflow = WorkflowWrapper(self._workflow_se, self.meshing)
         return self._workflow
 
     @property

src/ansys/fluent/core/session_solver.py

@@ -2,11 +2,19 @@
 
 import importlib
 
+from ansys.fluent.core.services.datamodel_se import PyMenuGeneric
 from ansys.fluent.core.services.datamodel_tui import TUIMenuGeneric
-from ansys.fluent.core.session import _CODEGEN_MSG_TUI, _BaseSession, _get_preferences
+from ansys.fluent.core.session import (
+    _CODEGEN_MSG_TUI,
+    _BaseSession,
+    _get_preferences,
+    _get_solverworkflow,
+)
+from ansys.fluent.core.session_shared import _CODEGEN_MSG_DATAMODEL
 from ansys.fluent.core.solver.flobject import get_root as settings_get_root
 from ansys.fluent.core.utils.fluent_version import get_version_for_filepath
 from ansys.fluent.core.utils.logging import LOG
+from ansys.fluent.core.workflow import WorkflowWrapper
 
 
 class Solver(_BaseSession):
@@ -23,8 +31,10 @@ def __init__(
         self._se_service = fluent_connection.datamodel_service_se
         self._settings_service = self.fluent_connection.settings_service
         self._tui = None
+        self._workflow = None
         self._settings_root = None
         self._version = None
+        self._solverworkflow = None
 
     @property
     def version(self):
@@ -47,6 +57,25 @@ def tui(self):
                 self._tui = TUIMenuGeneric([], self._tui_service)
         return self._tui
 
+    @property
+    def _workflow_se(self):
+        """workflow datamodel root."""
+        try:
+            workflow_module = importlib.import_module(
+                f"ansys.fluent.core.datamodel_{self.version}.workflow"
+            )
+            workflow_se = workflow_module.Root(self._se_service, "workflow", [])
+        except (ImportError, ModuleNotFoundError):
+            LOG.warning(_CODEGEN_MSG_DATAMODEL)
+            workflow_se = PyMenuGeneric(self._se_service, "workflow")
+        return workflow_se
+
+    @property
+    def workflow(self):
+        if not self._workflow:
+            self._workflow = WorkflowWrapper(self._workflow_se, Solver)
+        return self._workflow
+
     @property
     def _root(self):
         """root settings object."""
@@ -107,3 +136,10 @@ def preferences(self):
         if self._preferences is None:
             self._preferences = _get_preferences(self)
         return self._preferences
+
+    @property
+    def solverworkflow(self):
+        """solverworkflow datamodel root."""
+        if self._solverworkflow is None:
+            self._solverworkflow = _get_solverworkflow(self)
+        return self._solverworkflow

src/ansys/fluent/core/utils/setup_for_fluent.py

@@ -20,12 +20,14 @@ def setup_for_fluent(*args, **kwargs):
     globals = {}
     if "mode" in kwargs.keys() and kwargs["mode"] == "meshing":
         globals["meshing"] = session.meshing
-        globals["workflow"] = session.workflow
         globals["PartManagement"] = session.PartManagement
         globals["PMFileManagement"] = session.PMFileManagement
-        globals["preferences"] = session.preferences
         globals["solver"] = Solver(fluent_connection=session.fluent_connection)
     elif "mode" in kwargs.keys() and kwargs["mode"] == "solver":
         globals["solver"] = session
-        globals["preferences"] = session.preferences
+
+    globals["preferences"] = session.preferences
+    globals["solverworkflow"] = session.solverworkflow
+    globals["workflow"] = session.workflow
+
     return globals

src/ansys/fluent/core/meshing/workflow.py → src/ansys/fluent/core/workflow.py

@@ -1,28 +1,28 @@
 from ansys.fluent.core.services.datamodel_se import MakeReadOnly, PyCallableStateObject
 
 
-def _new_command_for_task(task, meshing):
+def _new_command_for_task(task, session):
     class NewCommandError(Exception):
         def __init__(self, task_name):
             super().__init__(f"Could not create command for meshing task {task_name}")
 
     task_cmd_name = task.CommandName()
-    cmd_creator = getattr(meshing, task_cmd_name)
+    cmd_creator = getattr(session, task_cmd_name)
     if cmd_creator:
         new_cmd = cmd_creator.new()
         if new_cmd:
             return new_cmd
     raise NewCommandError(task._name_())
 
 
-class MeshingWorkflow:
+class WorkflowWrapper:
     class TaskContainer(PyCallableStateObject):
-        def __init__(self, meshing):
-            self._meshing_container = meshing
-            self._task_container = meshing._workflow.TaskObject
+        def __init__(self, command_source):
+            self._container = command_source
+            self._task_container = command_source._workflow.TaskObject
 
         def __getitem__(self, name):
-            return MeshingWorkflow.Task(self._meshing_container, name)
+            return WorkflowWrapper.Task(self._container, name)
 
         def __getattr__(self, attr):
             return getattr(self._task_container, attr)
@@ -37,12 +37,12 @@ def __dir__(self):
             )
 
     class Task(PyCallableStateObject):
-        def __init__(self, meshing, name):
+        def __init__(self, command_source, name):
             self.__dict__.update(
                 dict(
-                    _workflow=meshing._workflow,
-                    _meshing=meshing._meshing,
-                    _task=meshing._workflow.TaskObject[name],
+                    _workflow=command_source._workflow,
+                    _source=command_source._command_source,
+                    _task=command_source._workflow.TaskObject[name],
                     _cmd=None,
                 )
             )
@@ -60,7 +60,7 @@ def _refreshed_command(self):
 
         def _command(self):
             if not self._cmd:
-                self._cmd = _new_command_for_task(self._task, self._meshing)
+                self._cmd = _new_command_for_task(self._task, self._source)
             return self._cmd
 
         def __getattr__(self, attr):
@@ -77,16 +77,16 @@ def __dir__(self):
                 set(list(self.__dict__.keys()) + dir(type(self)) + dir(self._task))
             )
 
-    def __init__(self, workflow, meshing):
+    def __init__(self, workflow, command_source):
         self._workflow = workflow
-        self._meshing = meshing
+        self._command_source = command_source
 
     def task(self, name):
-        return MeshingWorkflow.Task(self, name)
+        return WorkflowWrapper.Task(self, name)
 
     @property
     def TaskObject(self):
-        return MeshingWorkflow.TaskContainer(self)
+        return WorkflowWrapper.TaskContainer(self)
 
     def __getattr__(self, attr):
         return getattr(self._workflow, attr)

tests/test_session.py

@@ -5,6 +5,7 @@
 import grpc
 import pytest
 from util.meshing_workflow import new_mesh_session  # noqa: F401
+from util.solver_workflow import new_solver_session  # noqa: F401
 
 from ansys.api.fluent.v0 import health_pb2, health_pb2_grpc
 import ansys.fluent.core as pyfluent
@@ -202,3 +203,11 @@ def test_get_fluent_mode(new_mesh_session):
     assert session.fluent_connection.get_current_fluent_mode() == "meshing"
     session = session.switch_to_solver()
     assert session.fluent_connection.get_current_fluent_mode() == "solver"
+
+
+@pytest.mark.fluent_231
+def test_solverworkflow_in_solver_session(new_solver_session):
+    solver = new_solver_session
+    solver_dir = dir(solver)
+    for attr in ("preferences", "solverworkflow", "tui", "workflow"):
+        assert attr in solver_dir