|
23 | 23 |
|
24 | 24 | from enum import Enum, unique |
25 | 25 |
|
26 | | -from google.protobuf.empty_pb2 import Empty |
27 | 26 | import grpc |
28 | 27 |
|
29 | 28 | # ATTEMPT v0 IMPORT |
30 | 29 | try: |
31 | | - import ansys.api.dbu.v0.admin_pb2_grpc as dbu_v0_admin_pb2_grpc |
| 30 | + from ansys.api.dbu.v0.admin_pb2_grpc import AdminStub as V0HealthStub |
| 31 | + from google.protobuf.empty_pb2 import Empty as V0HealthRequest |
32 | 32 | except ImportError: |
33 | | - dbu_v0_admin_pb2_grpc = None |
| 33 | + V0HealthStub = None |
| 34 | + V0HealthRequest = None |
| 35 | + |
34 | 36 |
|
35 | 37 | # ATTEMPT v1 IMPORT |
36 | 38 | try: |
37 | | - import ansys.api.dbu.v1.admin_pb2_grpc as dbu_v1_admin_pb2_grpc |
| 39 | + from ansys.api.discovery.v1.commands.communication_pb2 import HealthRequest as V1HealthRequest |
| 40 | + from ansys.api.discovery.v1.commands.communication_pb2_grpc import ( |
| 41 | + CommunicationStub as V1HealthStub, |
| 42 | + ) |
38 | 43 | except ImportError: |
39 | | - dbu_v1_admin_pb2_grpc = None |
| 44 | + V1HealthStub = None |
| 45 | + V1HealthRequest = None |
40 | 46 |
|
41 | 47 |
|
42 | 48 | @unique |
43 | 49 | class GeometryApiProtos(Enum): |
44 | 50 | """Enumeration of the supported versions of the gRPC API protocol.""" |
45 | 51 |
|
46 | | - V0 = 0, dbu_v0_admin_pb2_grpc |
47 | | - V1 = 1, dbu_v1_admin_pb2_grpc |
| 52 | + V0 = 0, V0HealthStub, V0HealthRequest |
| 53 | + V1 = 1, V1HealthStub, V1HealthRequest |
48 | 54 |
|
49 | 55 | @staticmethod |
50 | 56 | def get_latest_version() -> "GeometryApiProtos": |
@@ -85,13 +91,14 @@ def verify_supported(self, channel: grpc.Channel) -> bool: |
85 | 91 | ----- |
86 | 92 | This method checks if the server supports the gRPC API protocol version. |
87 | 93 | """ |
88 | | - pb2_grpc = self.value[1] |
89 | | - if pb2_grpc is None: |
| 94 | + StubClass = self.value[1] # noqa: N806 |
| 95 | + RequestClass = self.value[2] # noqa: N806 |
| 96 | + if StubClass is None: |
90 | 97 | return False |
91 | 98 |
|
92 | 99 | try: |
93 | | - admin_stub = pb2_grpc.AdminStub(channel) |
94 | | - admin_stub.Health(Empty()) |
| 100 | + admin_stub = StubClass(channel) |
| 101 | + admin_stub.Health(RequestClass()) |
95 | 102 | return True |
96 | 103 | except grpc.RpcError: |
97 | 104 | return False |
@@ -126,9 +133,17 @@ def set_proto_version( |
126 | 133 | # If no version specified... Attempt to use all of them, starting |
127 | 134 | # with the latest version |
128 | 135 | if version is None: |
129 | | - version = GeometryApiProtos.get_latest_version() |
130 | | - while not version.verify_supported(channel): |
131 | | - version = GeometryApiProtos.from_int_value(version.value[0] - 1) |
| 136 | + # TODO: Once blitz is over, we will enable this... |
| 137 | + # https://github.com/ansys/pyansys-geometry/issues/2366 |
| 138 | + # |
| 139 | + # ---> REPLACE THIS BLOCK <--- |
| 140 | + # version = GeometryApiProtos.get_latest_version() |
| 141 | + # while not version.verify_supported(channel): |
| 142 | + # version = GeometryApiProtos.from_int_value(version.value[0] - 1) |
| 143 | + # ---> REPLACE THIS BLOCK <--- |
| 144 | + # ---> DELETE THIS BLOCK <--- |
| 145 | + version = GeometryApiProtos.V0 |
| 146 | + # ---> DELETE THIS BLOCK <--- |
132 | 147 |
|
133 | 148 | # Return the version |
134 | 149 | return version |
0 commit comments