chore: v1 implementation of admin and assembly condition stubs

doc/changelog.d/2391.maintenance.md

@@ -0,0 +1 @@
+V1 implementation of admin and assembly condition stubs

src/ansys/geometry/core/_grpc/_services/_service.py

@@ -24,7 +24,7 @@
 
 from .._version import GeometryApiProtos, set_proto_version
 from .base.admin import GRPCAdminService
-from .base.assembly_controls import GRPCAssemblyControlsService
+from .base.assembly_condition import GRPCAssemblyConditionService
 from .base.beams import GRPCBeamsService
 from .base.bodies import GRPCBodyService
 from .base.commands import GRPCCommandsService
@@ -87,7 +87,7 @@ def __init__(self, channel: grpc.Channel, version: GeometryApiProtos | str | Non
 
         # Lazy load all the services
         self._admin = None
-        self._assembly_controls = None
+        self._assembly_condition = None
         self._beams = None
         self._bodies = None
         self._commands = None
@@ -137,30 +137,30 @@ def admin(self) -> GRPCAdminService:
         return self._admin
 
     @property
-    def assembly_controls(self) -> GRPCAssemblyControlsService:
+    def assembly_condition(self) -> GRPCAssemblyConditionService:
         """
-        Get the assembly controls service for the specified version.
+        Get the assembly condition service for the specified version.
 
         Returns
         -------
-        GRPCAssemblyControlsService
-            The assembly controls service for the specified version.
+        GRPCAssemblyConditionService
+            The assembly condition service for the specified version.
         """
-        if not self._assembly_controls:
-            # Import the appropriate assembly controls service based on the version
-            from .v0.assembly_controls import GRPCAssemblyControlsServiceV0
-            from .v1.assembly_controls import GRPCAssemblyControlsServiceV1
+        if not self._assembly_condition:
+            # Import the appropriate assembly condition service based on the version
+            from .v0.assembly_condition import GRPCAssemblyConditionServiceV0
+            from .v1.assembly_condition import GRPCAssemblyConditionServiceV1
 
             if self.version == GeometryApiProtos.V0:
-                self._assembly_controls = GRPCAssemblyControlsServiceV0(self.channel)
+                self._assembly_condition = GRPCAssemblyConditionServiceV0(self.channel)
             elif self.version == GeometryApiProtos.V1:  # pragma: no cover
                 # V1 is not implemented yet
-                self._assembly_controls = GRPCAssemblyControlsServiceV1(self.channel)
+                self._assembly_condition = GRPCAssemblyConditionServiceV1(self.channel)
             else:  # pragma: no cover
                 # This should never happen as the version is set in the constructor
                 raise ValueError(f"Unsupported version: {self.version}")
 
-        return self._assembly_controls
+        return self._assembly_condition
 
     @property
     def beams(self) -> GRPCBeamsService:

src/ansys/geometry/core/_grpc/_services/base/assembly_controls.py → src/ansys/geometry/core/_grpc/_services/base/assembly_condition.py

@@ -19,15 +19,15 @@
 # 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 the assembly controls service implementation (abstraction layer)."""
+"""Module containing the assembly condition service implementation (abstraction layer)."""
 
 from abc import ABC, abstractmethod
 
 import grpc
 
 
-class GRPCAssemblyControlsService(ABC):  # pragma: no cover
-    """Assembly controls service for gRPC communication with the Geometry server.
+class GRPCAssemblyConditionService(ABC):  # pragma: no cover
+    """Assembly condition service for gRPC communication with the Geometry server.
 
     Parameters
     ----------
@@ -36,7 +36,7 @@ class GRPCAssemblyControlsService(ABC):  # pragma: no cover
     """
 
     def __init__(self, channel: grpc.Channel):
-        """Initialize the GRPCAssemblyControls class."""
+        """Initialize the GRPCAssemblyCondition class."""
         pass
 
     @abstractmethod

src/ansys/geometry/core/_grpc/_services/v0/assembly_controls.py → src/ansys/geometry/core/_grpc/_services/v0/assembly_condition.py

@@ -19,21 +19,21 @@
 # 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 the assembly controls service implementation for v0."""
+"""Module containing the assembly condition service implementation for v0."""
 
 import grpc
 
 from ansys.geometry.core.errors import protect_grpc
 
-from ..base.assembly_controls import GRPCAssemblyControlsService
+from ..base.assembly_condition import GRPCAssemblyConditionService
 from .conversions import build_grpc_id
 
 
-class GRPCAssemblyControlsServiceV0(GRPCAssemblyControlsService):
-    """Assembly controls service for gRPC communication with the Geometry server.
+class GRPCAssemblyConditionServiceV0(GRPCAssemblyConditionService):
+    """Assembly condition service for gRPC communication with the Geometry server.
 
     This class provides methods to interact with the Geometry server's
-    assembly controls service. It is specifically designed for the v0 version of the
+    assembly condition service. It is specifically designed for the v0 version of the
     Geometry API.
 
     Parameters
@@ -66,7 +66,7 @@ def create_align_condition(self, **kwargs) -> dict:  # noqa: D102
 
         # Return the response - formatted as a dictionary
         return {
-            "moniker": resp.condition.moniker,
+            "id": resp.condition.moniker,
             "is_deleted": resp.condition.is_deleted,
             "is_enabled": resp.condition.is_enabled,
             "is_satisfied": resp.condition.is_satisfied,
@@ -93,7 +93,7 @@ def create_tangent_condition(self, **kwargs) -> dict:  # noqa: D102
 
         # Return the response - formatted as a dictionary
         return {
-            "moniker": resp.condition.moniker,
+            "id": resp.condition.moniker,
             "is_deleted": resp.condition.is_deleted,
             "is_enabled": resp.condition.is_enabled,
             "is_satisfied": resp.condition.is_satisfied,
@@ -120,7 +120,7 @@ def create_orient_condition(self, **kwargs) -> dict:  # noqa: D102
 
         # Return the response - formatted as a dictionary
         return {
-            "moniker": resp.condition.moniker,
+            "id": resp.condition.moniker,
             "is_deleted": resp.condition.is_deleted,
             "is_enabled": resp.condition.is_enabled,
             "is_satisfied": resp.condition.is_satisfied,

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,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
+        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)
+
+        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/assembly_condition.py

@@ -0,0 +1,147 @@
+# 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 the assembly condition service implementation for v1."""
+
+import grpc
+
+from ansys.geometry.core.errors import protect_grpc
+
+from ..base.assembly_condition import GRPCAssemblyConditionService
+from .conversions import build_grpc_id
+
+
+class GRPCAssemblyConditionServiceV1(GRPCAssemblyConditionService):
+    """Assembly condition service for gRPC communication with the Geometry server.
+
+    This class provides methods to interact with the Geometry server's
+    assembly condition service. It is specifically designed for the v1 version of the
+    Geometry API.
+
+    Parameters
+    ----------
+    channel : grpc.Channel
+        The gRPC channel to the server.
+    """
+
+    @protect_grpc
+    def __init__(self, channel: grpc.Channel):  # noqa: D102
+        from ansys.api.discovery.v1.design.relationships.assemblycondition_pb2_grpc import (
+            AssemblyConditionStub,
+        )
+
+        self.stub = AssemblyConditionStub(channel)
+
+    @protect_grpc
+    def create_align_condition(self, **kwargs) -> dict:  # noqa: D102
+        from ansys.api.discovery.v1.design.relationships.assemblycondition_pb2 import (
+            CreateAlignRequest,
+            CreateAlignRequestData,
+        )
+
+        # Create the request - assumes all inputs are valid and of the proper type
+        request = CreateAlignRequest(
+            request_data=[
+                CreateAlignRequestData(
+                    parent=build_grpc_id(kwargs["parent_id"]),
+                    geometric_a=build_grpc_id(kwargs["geometric_a_id"]),
+                    geometric_b=build_grpc_id(kwargs["geometric_b_id"]),
+                )
+            ]
+        )
+
+        # Call the gRPC service
+        resp = self.stub.CreateAlign(request).response_data[0]
+
+        # Return the response - formatted as a dictionary
+        return {
+            "id": resp.align_condition.condition.id,
+            "is_deleted": resp.align_condition.condition.is_deleted,
+            "is_enabled": resp.align_condition.condition.is_enabled,
+            "is_satisfied": resp.align_condition.condition.is_satisfied,
+            "offset": resp.align_condition.offset,
+            "is_reversed": resp.align_condition.is_reversed,
+            "is_valid": resp.align_condition.is_valid,
+        }
+
+    @protect_grpc
+    def create_tangent_condition(self, **kwargs) -> dict:  # noqa: D102
+        from ansys.api.discovery.v1.design.relationships.assemblycondition_pb2 import (
+            CreateTangentRequest,
+            CreateTangentRequestData,
+        )
+
+        # Create the request - assumes all inputs are valid and of the proper type
+        request = CreateTangentRequest(
+            request_data=[
+                CreateTangentRequestData(
+                    parent=build_grpc_id(kwargs["parent_id"]),
+                    geometric_a=build_grpc_id(kwargs["geometric_a_id"]),
+                    geometric_b=build_grpc_id(kwargs["geometric_b_id"]),
+                )
+            ]
+        )
+
+        # Call the gRPC service
+        resp = self.stub.CreateTangent(request).response_data[0]
+
+        # Return the response - formatted as a dictionary
+        return {
+            "id": resp.tangent_condition.condition.id,
+            "is_deleted": resp.tangent_condition.condition.is_deleted,
+            "is_enabled": resp.tangent_condition.condition.is_enabled,
+            "is_satisfied": resp.tangent_condition.condition.is_satisfied,
+            "offset": resp.tangent_condition.offset,
+            "is_reversed": resp.tangent_condition.is_reversed,
+            "is_valid": resp.tangent_condition.is_valid,
+        }
+
+    @protect_grpc
+    def create_orient_condition(self, **kwargs) -> dict:  # noqa: D102
+        from ansys.api.discovery.v1.design.relationships.assemblycondition_pb2 import (
+            CreateOrientRequest,
+            CreateOrientRequestData,
+        )
+
+        # Create the request - assumes all inputs are valid and of the proper type
+        request = CreateOrientRequest(
+            request_data=[
+                CreateOrientRequestData(
+                    parent=build_grpc_id(kwargs["parent_id"]),
+                    geometric_a=build_grpc_id(kwargs["geometric_a_id"]),
+                    geometric_b=build_grpc_id(kwargs["geometric_b_id"]),
+                )
+            ]
+        )
+
+        # Call the gRPC service
+        resp = self.stub.CreateOrient(request).response_data[0]
+
+        # Return the response - formatted as a dictionary
+        return {
+            "id": resp.orient_condition.condition.id,
+            "is_deleted": resp.orient_condition.condition.is_deleted,
+            "is_enabled": resp.orient_condition.condition.is_enabled,
+            "is_satisfied": resp.orient_condition.condition.is_satisfied,
+            "offset": resp.orient_condition.offset,
+            "is_reversed": resp.orient_condition.is_reversed,
+            "is_valid": resp.orient_condition.is_valid,
+        }