impl v1 methods and conversions for admin stub

Author: jacobrkerstetter

src/ansys/geometry/core/_grpc/_services/v1/admin.py

@@ -21,11 +21,15 @@
 # SOFTWARE.
 """Module containing the admin service implementation for v1."""
 
+import warnings
+
 import grpc
+import semver
 
 from ansys.geometry.core.errors import protect_grpc
 
 from ..base.admin import GRPCAdminService
+from .conversions import from_grpc_backend_type_to_backend_type
 
 
 class GRPCAdminServiceV1(GRPCAdminService):  # pragma: no cover
@@ -43,18 +47,83 @@ class GRPCAdminServiceV1(GRPCAdminService):  # pragma: no cover
 
     @protect_grpc
     def __init__(self, channel: grpc.Channel):  # noqa: D102
-        from ansys.api.dbu.v1.admin_pb2_grpc import AdminStub
+        from ansys.api.discovery.v1.commands.application_pb2_grpc import ApplicationStub
+        from ansys.api.discovery.v1.commands.communication_pb2_grpc import CommunicationStub
 
-        self.stub = AdminStub(channel)
+        self.admin_stub = ApplicationStub(channel)
+        self.communication_stub = CommunicationStub(channel)
 
     @protect_grpc
     def get_backend(self, **kwargs) -> dict:  # noqa: D102
-        raise NotImplementedError
+        # TODO: Remove this context and filter once the protobuf UserWarning is downgraded to INFO
+        # https://github.com/grpc/grpc/issues/37609
+        with warnings.catch_warnings():
+            warnings.filterwarnings(
+                "ignore", "Protobuf gencode version", UserWarning, "google.protobuf.runtime_version"
+            )
+            from ansys.api.discovery.v1.commands.application_pb2 import GetBackendRequest
+
+        # Create the request - assumes all inputs are valid and of the proper type
+        request = GetBackendRequest()
+
+        # Call the gRPC service
+        response = self.admin_stub.GetBackend(request=request)
+
+        # COMPATIBILITY HACK: retrieve the backend version -- for versions after 24R1
+        ver = response.version
+        backend_version = semver.Version(ver.major_release, ver.minor_release, ver.service_pack)
+        api_server_build_info = f"{ver.build_number}" if ver.build_number != 0 else "N/A"
+        product_build_info = (
+            response.backend_version_info.strip() if response.backend_version_info else "N/A"
+        )
+
+        # Convert the response to a dictionary
+        return {
+            "backend": from_grpc_backend_type_to_backend_type(response.type),
+            "version": backend_version,
+            "api_server_build_info": api_server_build_info,
+            "product_build_info": product_build_info,
+            "additional_info": {k: v for k, v in response.additional_build_info.items()},
+        }
 
     @protect_grpc
     def get_logs(self, **kwargs) -> dict:  # noqa: D102
-        raise NotImplementedError
+        from ansys.api.discovery.v1.commands.communication_pb2 import LogsRequest
+        from ansys.api.discovery.v1.commonenums_pb2 import LogsPeriodType, LogsTarget
+
+        # Create the request - assumes all inputs are valid and of the proper type
+        request = LogsRequest(
+            target=LogsTarget.LOGSTARGET_CLIENT,
+            period_type=(
+                LogsPeriodType.LOGSPERIODTIME_CURRENT
+                if not kwargs["all_logs"]
+                else LogsPeriodType.LOGSPERIODTIME_ALL
+            ),
+            null_path=None,
+            null_period=None,
+        )
+
+        # Call the gRPC service
+        logs_generator = self.communication_stub.GetLogs(request)
+        logs: dict[str, str] = {}
+
+        # Convert the response to a dictionary
+        for chunk in logs_generator:
+            if chunk.log_name not in logs:
+                logs[chunk.log_name] = ""
+            logs[chunk.log_name] += chunk.log_chunk.decode()
+
+        return {"logs": logs}
 
     @protect_grpc
     def get_service_status(self, **kwargs) -> dict:  # noqa: D102
-        raise NotImplementedError
+        from ansys.api.discovery.v1.commands.communication_pb2 import HealthRequest
+
+        # Create the request - assumes all inputs are valid and of the proper type
+        request = HealthRequest()
+
+        # Call the gRPC service
+        response = self.communication_stub.Health(request=request)
+
+        # Convert the response to a dictionary
+        return {"healthy": True if response.message == "I am healthy!" else False}

src/ansys/geometry/core/_grpc/_services/v1/conversions.py

@@ -0,0 +1,68 @@
+# Copyright (C) 2023 - 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.
+"""Module containing v1 related conversions from PyAnsys Geometry objects to gRPC messages."""
+
+from typing import TYPE_CHECKING
+
+from ansys.api.discovery.v1.commonenums_pb2 import BackendType as GRPCBackendType
+
+if TYPE_CHECKING:
+    from ansys.geometry.core.connection.backend import BackendType
+
+def from_grpc_backend_type_to_backend_type(
+    grpc_backend_type: GRPCBackendType,
+) -> "BackendType":
+    """Convert a gRPC backend type to a backend type.
+
+    Parameters
+    ----------
+    backend_type : GRPCBackendType
+        Source backend type.
+
+    Returns
+    -------
+    BackendType
+        Converted backend type.
+    """
+    from ansys.geometry.core.connection.backend import BackendType
+
+    # Map the gRPC backend type to the corresponding BackendType
+    backend_type = None
+
+    if grpc_backend_type == GRPCBackendType.BACKENDTYPE_DISCOVERY:
+        backend_type = BackendType.DISCOVERY
+    elif grpc_backend_type == GRPCBackendType.BACKENDTYPE_SPACECLAIM:
+        backend_type = BackendType.SPACECLAIM
+    elif grpc_backend_type == GRPCBackendType.BACKENDTYPE_WINDOWS_DMS:
+        backend_type = BackendType.WINDOWS_SERVICE
+    elif grpc_backend_type == GRPCBackendType.BACKENDTYPE_LINUX_DMS:
+        backend_type = BackendType.LINUX_SERVICE
+    elif grpc_backend_type == GRPCBackendType.BACKENDTYPE_CORE_SERVICE_LINUX:
+        backend_type = BackendType.CORE_LINUX
+    elif grpc_backend_type == GRPCBackendType.BACKENDTYPE_CORE_SERVICE_WINDOWS:
+        backend_type = BackendType.CORE_WINDOWS
+    elif grpc_backend_type == GRPCBackendType.BACKENDTYPE_DISCOVERY_HEADLESS:
+        backend_type = BackendType.DISCOVERY_HEADLESS
+    else:
+        raise ValueError(f"Invalid backend type: {grpc_backend_type}")
+
+    return backend_type