chore: Restructure some more grpc services.

doc/api_rstgen.py

@@ -154,7 +154,7 @@ def _get_file_path(folder_name: str, file_name: str):
         "api_upgrade",
         "batch_ops",
         "datamodel_se",
-        "datamodel_tui",
+        "text_interface",
         "events",
         "field_data",
         "health_check",

doc/changelog.d/5260.maintenance.md

@@ -0,0 +1 @@
+Restructure some more grpc services.

doc/changelog.d/5262.maintenance.md

@@ -0,0 +1 @@
+Refactor Batch Ops service.

doc/changelog.d/5264.maintenance.md

@@ -0,0 +1 @@
+Update monitors service.

doc/changelog.d/5265.maintenance.md

@@ -0,0 +1 @@
+Update solution variable

doc/source/user_guide/legacy/tui.rst

@@ -48,14 +48,14 @@ To see the documentation for the viscous model menu options, you can run:
    >>> help(solver_session.tui.define.models.viscous)
    Help on viscous in module ansys.fluent.core.generated.solver.tui_241 object:
 
-   class viscous(ansys.fluent.core.services.datamodel_tui.TUIMenu)
+   class viscous(ansys.fluent.core.services.text_interface.TUIMenu)
     |  viscous(service, version, mode, path)
     |
     |  Enters the viscous model menu.
     |
     |  Method resolution order:
     |      viscous
-    |      ansys.fluent.core.services.datamodel_tui.TUIMenu
+    |      ansys.fluent.core.services.text_interface.TUIMenu
     |      builtins.object
     |
     |  Methods defined here:

src/ansys/fluent/core/_grpc_services/__init__.py

@@ -36,6 +36,13 @@
 from ansys.fluent.core._grpc_services.application_runtime_service_v0 import (
     ApplicationRuntimeService as ApplicationRuntimeServiceV0,
 )
+from ansys.fluent.core._grpc_services.batch_ops_service_v0 import (
+    BatchOpsService as BatchOpsServiceV0,
+)
+from ansys.fluent.core._grpc_services.events_service import EventsService
+from ansys.fluent.core._grpc_services.events_service_v0 import (
+    EventsService as EventsServiceV0,
+)
 from ansys.fluent.core._grpc_services.field_data_service import FieldDataService
 from ansys.fluent.core._grpc_services.field_data_service_v0 import (
     FieldDataService as FieldDataServiceV0,
@@ -44,6 +51,10 @@
 from ansys.fluent.core._grpc_services.health_check_service_v0 import (
     HealthCheckService as HealthCheckServiceV0,
 )
+from ansys.fluent.core._grpc_services.monitor_service import MonitorService
+from ansys.fluent.core._grpc_services.monitor_service_v0 import (
+    MonitorService as MonitorServiceV0,
+)
 from ansys.fluent.core._grpc_services.object_model_service import ObjectModelService
 from ansys.fluent.core._grpc_services.object_model_service_v0 import (
     ObjectModelService as ObjectModelServiceV0,
@@ -62,6 +73,20 @@
 from ansys.fluent.core._grpc_services.settings_service_v0 import (
     SettingsService as SettingsServiceV0,
 )
+from ansys.fluent.core._grpc_services.solution_variable_service import (
+    SolutionVariableService,
+)
+from ansys.fluent.core._grpc_services.solution_variable_service_v0 import (
+    SolutionVariableService as SolutionVariableServiceV0,
+)
+from ansys.fluent.core._grpc_services.text_interface_service import TextInterfaceService
+from ansys.fluent.core._grpc_services.text_interface_service_v0 import (
+    TextInterfaceService as TextInterfaceServiceV0,
+)
+from ansys.fluent.core._grpc_services.transcript_service import TranscriptService
+from ansys.fluent.core._grpc_services.transcript_service_v0 import (
+    TranscriptService as TranscriptServiceV0,
+)
 
 
 def _server_supports_v1(channel) -> bool:
@@ -205,3 +230,53 @@ def object_model(self) -> ObjectModelService | ObjectModelServiceV0:
             if self._proto_version == "v1"
             else self._get_instantiated_grpc_service(ObjectModelServiceV0)
         )
+
+    @cached_property
+    def events(self) -> EventsService | EventsServiceV0:
+        """gRPC stub for events operations."""
+        return (
+            self._get_instantiated_grpc_service(EventsService)
+            if self._proto_version == "v1"
+            else self._get_instantiated_grpc_service(EventsServiceV0)
+        )
+
+    @cached_property
+    def batch_ops(self) -> BatchOpsServiceV0:
+        """gRPC stub for batch RPC operations (v0 only — no v1 implementation)."""
+        return self._get_instantiated_grpc_service(BatchOpsServiceV0)
+
+    @cached_property
+    def transcript(self) -> TranscriptService | TranscriptServiceV0:
+        """gRPC stub for transcript operations."""
+        return (
+            self._get_instantiated_grpc_service(TranscriptService)
+            if self._proto_version == "v1"
+            else self._get_instantiated_grpc_service(TranscriptServiceV0)
+        )
+
+    @cached_property
+    def text_interface(self) -> TextInterfaceService | TextInterfaceServiceV0:
+        """gRPC stub for text interface operations."""
+        return (
+            self._get_instantiated_grpc_service(TextInterfaceService)
+            if self._proto_version == "v1"
+            else self._get_instantiated_grpc_service(TextInterfaceServiceV0)
+        )
+
+    @cached_property
+    def monitor(self) -> MonitorService | MonitorServiceV0:
+        """gRPC stub for monitor operations."""
+        return (
+            self._get_instantiated_grpc_service(MonitorService)
+            if self._proto_version == "v1"
+            else self._get_instantiated_grpc_service(MonitorServiceV0)
+        )
+
+    @cached_property
+    def solution_variable(self) -> SolutionVariableService | SolutionVariableServiceV0:
+        """gRPC stub for solution variable operations."""
+        return (
+            self._get_instantiated_grpc_service(SolutionVariableService)
+            if self._proto_version == "v1"
+            else self._get_instantiated_grpc_service(SolutionVariableServiceV0)
+        )

src/ansys/fluent/core/_grpc_services/application_runtime_service.py

@@ -233,7 +233,7 @@ def get_gpu_config(self) -> bool | list[int]:
             case "specific_gpu_ids":
                 return list(response.specific_gpu_ids.gpu_ids)
 
-    def start_python_journal(self, journal_name: str | None = None) -> int:
+    def start_python_journal(self, journal_name: str | None = None) -> str:
         """StartPythonJournal RPC."""
         request = application_runtime_pb2.StartPythonJournalRequest()
         if journal_name:

src/ansys/fluent/core/_grpc_services/application_runtime_service_v0.py

@@ -114,7 +114,7 @@ def get_app_mode(self) -> str:
             case app_utilities_pb2.APP_MODE_SOLVER_AERO:
                 return "solver_aero"
 
-    def start_python_journal(self, journal_name: str | None = None) -> int:
+    def start_python_journal(self, journal_name: str | None = None) -> str:
         """StartPythonJournal RPC."""
         request = app_utilities_pb2.StartPythonJournalRequest()
         if journal_name:

src/ansys/fluent/core/_grpc_services/batch_ops_service_v0.py

@@ -0,0 +1,151 @@
+# Copyright (C) 2021 - 2026 Synopsys, Inc. and ANSYS, Inc. All rights reserved.
+# 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.
+
+"""Batch RPC service (v0 proto API)."""
+
+import inspect
+import logging
+import sys
+from types import ModuleType
+
+import grpc
+
+import ansys.api.fluent.v0 as api
+from ansys.api.fluent.v0 import batch_ops_pb2, batch_ops_pb2_grpc
+from ansys.fluent.core.services._protocols import ServiceProtocol
+from ansys.fluent.core.services.interceptors import GrpcErrorInterceptor
+
+network_logger: logging.Logger = logging.getLogger("pyfluent.networking")
+
+
+class BatchOpsService(ServiceProtocol):
+    """Class wrapping the batch RPC service of Fluent (v0 proto API)."""
+
+    _api_module = api
+    _proto_module = batch_ops_pb2
+
+    def __init__(
+        self,
+        channel: grpc.Channel,
+        metadata: list[tuple[str, str]],
+        fluent_error_state=None,
+    ) -> None:
+        """__init__ method of BatchOpsService class."""
+        del fluent_error_state  # unused
+        intercept_channel = grpc.intercept_channel(
+            channel,
+            GrpcErrorInterceptor(),
+        )
+        self._stub = batch_ops_pb2_grpc.BatchOpsStub(intercept_channel)
+        self._metadata = metadata
+        self._proto_files: list[ModuleType] | None = None
+        self._response_cls_cache: dict[tuple[str, str, str], type | None] = {}
+
+    def _ensure_proto_files(self) -> None:
+        """Lazily populate the list of all known proto file descriptors."""
+        if self._proto_files is not None:
+            return
+        owner_proto_files = [
+            mod
+            for _, mod in inspect.getmembers(self._api_module, inspect.ismodule)
+            if hasattr(mod, "DESCRIPTOR")
+        ]
+        loaded_proto_files = [
+            mod
+            for name, mod in sys.modules.items()
+            if name.endswith("_pb2") and hasattr(mod, "DESCRIPTOR")
+        ]
+        self._proto_files = owner_proto_files + [
+            mod for mod in loaded_proto_files if mod not in owner_proto_files
+        ]
+
+    def _get_response_cls(self, package: str, service: str, method: str) -> type | None:
+        """Return the proto response class for the given RPC, or None if not found."""
+        key = (package, service, method)
+        if key in self._response_cls_cache:
+            return self._response_cls_cache[key]
+        self._ensure_proto_files()
+        result = None
+        for file in self._proto_files:
+            file_desc = file.DESCRIPTOR
+            if file_desc.package == package:
+                service_desc = file_desc.services_by_name.get(service)
+                if service_desc:
+                    method_desc = service_desc.methods_by_name.get(method)
+                    if (
+                        method_desc
+                        and not method_desc.client_streaming
+                        and not method_desc.server_streaming
+                    ):
+                        response_cls_name = method_desc.output_type.name
+                        try:
+                            result = getattr(file, response_cls_name)
+                            break
+                        except AttributeError:
+                            pass
+        self._response_cls_cache[key] = result
+        return result
+
+    def get_op_metadata(
+        self, package: str, service: str, method: str
+    ) -> tuple[bool, type | None]:
+        """Return ``(is_supported, response_cls)`` for the given RPC.
+
+        An operation is considered supported when it is not a getter and the
+        proto descriptor exposes a non-streaming unary response type.
+        """
+        if method.startswith("Get") or method.startswith("get"):
+            return False, None
+        response_cls = self._get_response_cls(package, service, method)
+        return response_cls is not None, response_cls
+
+    def execute(self, ops: list) -> list[tuple]:
+        """Execute a batch of queued operations and return ``(status, result)`` pairs.
+
+        Execute is a bidirectional-streaming RPC: the client sends a stream of
+        ``ExecuteRequest`` messages and the server returns a corresponding
+        stream of ``ExecuteResponse`` messages.  Each response is deserialized
+        into the appropriate proto message type before being returned.
+        """
+        requests = (
+            self._proto_module.ExecuteRequest(
+                package=op._package,
+                service=op._service_name,
+                method=op._method,
+                request_body=op._request_body,
+            )
+            for op in ops
+        )
+        results = []
+        for op, response in zip(
+            ops, self._stub.Execute(requests, metadata=self._metadata)
+        ):
+            result = None
+            if op.response_cls is not None:
+                result = op.response_cls()
+                try:
+                    result.ParseFromString(response.response_body)
+                except Exception as ex:
+                    network_logger.warning(ex)
+            results.append((response.status, result))
+        return results

src/ansys/fluent/core/services/events_v1.py → src/ansys/fluent/core/_grpc_services/events_service.py

@@ -21,66 +21,49 @@
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 # SOFTWARE.
 
-"""Wrapper over the events gRPC service of Fluent (v1 proto API).
-
-All shared logic lives in events.py (v0). This module adds v1-specific
-pause/resume-on-solution-event RPC methods that migrated from the
-ApplicationRuntime service in the v1 proto API.
-"""
+"""Wrapper over the events gRPC service of Fluent (v1 proto API)."""
 
 import grpc
 
-from ansys.api.fluent.v1 import events_pb2 as EventsProtoModule
-from ansys.api.fluent.v1 import events_pb2_grpc as EventsGrpcModule
-from ansys.fluent.core.services.events import EventsService as _EventsServiceV0
+from ansys.api.fluent.v1 import events_pb2, events_pb2_grpc
+from ansys.fluent.core._grpc_services.streaming_service import StreamingService
+from ansys.fluent.core.services._protocols import ServiceProtocol
 from ansys.fluent.core.streaming_services.events_streaming_v1 import SolverEvent
 
 
-class EventsService(_EventsServiceV0):
+class EventsService(StreamingService, ServiceProtocol):
     """Class wrapping the events gRPC service of Fluent (v1 proto API)."""
 
-    def _create_stub(self, channel: grpc.Channel):
-        """Create the v1 gRPC stub."""
-        return EventsGrpcModule.EventsStub(channel)
-
-    def pause_solve_for(
-        self, request: EventsProtoModule.PauseSolveForRequest
-    ) -> EventsProtoModule.PauseSolveForResponse:
-        """PauseSolveFor RPC of Events service (v1)."""
-        return self._stub.PauseSolveFor(request, metadata=self._metadata)
-
-    def resume_solve(
-        self, request: EventsProtoModule.ResumeSolveRequest
-    ) -> EventsProtoModule.ResumeSolveResponse:
-        """ResumeSolve RPC of Events service (v1)."""
-        return self._stub.ResumeSolve(request, metadata=self._metadata)
-
-    def cancel_pause_solve(
-        self, request: EventsProtoModule.CancelPauseSolveRequest
-    ) -> EventsProtoModule.CancelPauseSolveResponse:
-        """CancelPauseSolve RPC of Events service (v1)."""
-        return self._stub.CancelPauseSolve(request, metadata=self._metadata)
+    def __init__(
+        self, channel: grpc.Channel, metadata: list[tuple[str, str]], fluent_error_state
+    ):
+        """__init__ method of EventsService class."""
+        super().__init__(
+            stub=events_pb2_grpc.EventsStub(channel),
+            metadata=metadata,
+        )
+        del fluent_error_state  # unused in v1
 
     def register_pause_on_solution_events(self, solution_event: SolverEvent) -> int:
         """Register pause on solution events."""
-        request = EventsProtoModule.PauseSolveForRequest()
-        request.solution_event = EventsProtoModule.SOLUTION_EVENT_UNSPECIFIED
+        request = events_pb2.PauseSolveForRequest()
+        request.solution_event = events_pb2.SOLUTION_EVENT_UNSPECIFIED
         match solution_event:
             case SolverEvent.ITERATION_ENDED:
-                request.solution_event = EventsProtoModule.SOLUTION_EVENT_ITERATION
+                request.solution_event = events_pb2.SOLUTION_EVENT_ITERATION
             case SolverEvent.TIMESTEP_ENDED:
-                request.solution_event = EventsProtoModule.SOLUTION_EVENT_TIME_STEP
-        response = self.pause_solve_for(request)
+                request.solution_event = events_pb2.SOLUTION_EVENT_TIME_STEP
+        response = self._stub.PauseSolveFor(request, metadata=self._metadata)
         return response.registration_id
 
     def resume_on_solution_event(self, registration_id: int) -> None:
         """Resume on solution event."""
-        request = EventsProtoModule.ResumeSolveRequest()
+        request = events_pb2.ResumeSolveRequest()
         request.registration_id = registration_id
-        self.resume_solve(request)
+        self._stub.ResumeSolve(request, metadata=self._metadata)
 
     def unregister_pause_on_solution_events(self, registration_id: int) -> None:
         """Unregister pause on solution events."""
-        request = EventsProtoModule.CancelPauseSolveRequest()
+        request = events_pb2.CancelPauseSolveRequest()
         request.registration_id = registration_id
-        self.cancel_pause_solve(request)
+        self._stub.CancelPauseSolve(request, metadata=self._metadata)