admin stub

Author: umutsoysalansys

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

@@ -22,11 +22,18 @@
 """Module containing the admin service implementation for v1."""
 
 import grpc
+import warnings
+
+import semver
 
 from ansys.geometry.core.errors import protect_grpc
 
 from ..base.admin import GRPCAdminService
 
+# Define BackendType if not already imported
+class BackendType:
+    DISCOVERY = "DISCOVERY"
+
 
 class GRPCAdminServiceV1(GRPCAdminService):  # pragma: no cover
     """Admin service for gRPC communication with the Geometry server.
@@ -43,18 +50,82 @@ 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
 
-        self.stub = AdminStub(channel)
+        self.stub = ApplicationStub(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 google.protobuf.empty_pb2 import Empty
+
+        # Create the request - assumes all inputs are valid and of the proper type
+        request = Empty()
+
+        # Call the gRPC service
+        response = self.stub.GetBackend(request=request)
+
+        # COMPATIBILITY HACK: retrieve the backend version -- for versions after 24R1
+        if hasattr(response, "version"):
+            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"
+            )
+        else:  # pragma: no cover
+            # If the version is not available, set a default version
+            backend_version = semver.Version(24, 1, 0)
+            api_server_build_info = "N/A"
+            product_build_info = "N/A"
+
+        # Convert the response to a dictionary
+        return {
+            "backend": BackendType.DISCOVERY,
+            "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.dbu.v1.admin_pb2 import LogsRequest, LogsTarget, PeriodType
+
+        # Create the request - assumes all inputs are valid and of the proper type
+        request = LogsRequest(
+            target=LogsTarget.CLIENT,
+            period_type=PeriodType.CURRENT if not kwargs["all_logs"] else PeriodType.ALL,
+            null_path=None,
+            null_period=None,
+        )
+
+        # Call the gRPC service
+        logs_generator = self.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 google.protobuf.empty_pb2 import Empty
+
+        # Create the request - assumes all inputs are valid and of the proper type
+        request = Empty()
+
+        # Call the gRPC service
+        response = self.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/designs.py

@@ -43,9 +43,9 @@ class GRPCDesignsServiceV1(GRPCDesignsService):  # pragma: no cover
 
     @protect_grpc
     def __init__(self, channel: grpc.Channel):  # noqa: D102
-        from ansys.api.dbu.v1.designs_pb2_grpc import DesignsStub
+        from ansys.api.discovery.v1.design.designdoc_pb2_grpc import DesignDocStub
 
-        self.stub = DesignsStub(channel)
+        self.stub = DesignDocStub(channel)
 
     @protect_grpc
     def open(self, **kwargs) -> dict:  # noqa: D102
@@ -93,7 +93,21 @@ def upload_file(self, **kwargs) -> dict:  # noqa: D102
 
     @protect_grpc
     def upload_file_stream(self, **kwargs) -> dict:  # noqa: D102
-        raise NotImplementedError
+        from ansys.api.discovery.v1.commands_pb2 import UploadFileRequest
+
+        # Create the request - assumes all inputs are valid and of the proper type
+        request = UploadFileRequest(
+            data=kwargs["data"],
+            file_name=kwargs["file_name"],
+            open=kwargs["open_file"],
+            import_options=kwargs["import_options"].to_dict(),
+        )
+
+        # Call the gRPC service
+        response = self.commands_stub.UploadFile(request)
+
+        # Return the response - formatted as a dictionary
+        return {"file_path": response.file_path}
 
     @protect_grpc
     def stream_design_tessellation(self, **kwargs) -> dict:  # noqa: D102