diff --git a/doc/changelog.d/2417.maintenance.md b/doc/changelog.d/2417.maintenance.md new file mode 100644 index 0000000000..a9be41aff4 --- /dev/null +++ b/doc/changelog.d/2417.maintenance.md @@ -0,0 +1 @@ +Prepare tools to v1 diff --git a/src/ansys/geometry/core/_grpc/_services/v1/conversions.py b/src/ansys/geometry/core/_grpc/_services/v1/conversions.py index fa81a6c545..74ce3d903a 100644 --- a/src/ansys/geometry/core/_grpc/_services/v1/conversions.py +++ b/src/ansys/geometry/core/_grpc/_services/v1/conversions.py @@ -1401,3 +1401,42 @@ def serialize_entity_identifier(entity): for entity in getattr(response, "deleted_bodies", []) ], } + + +def get_standard_tracker_response(response) -> dict: + """Get a standard dictionary response from a TrackerCommandResponse gRPC object. + + Parameters + ---------- + response : TrackerCommandResponse + The gRPC TrackerCommandResponse object. + + Returns + ------- + dict + A dictionary representing the standard tracker response + """ + return { + "success": response.command_response.success, + "tracker_response": serialize_tracker_command_response(response.tracked_changes), + } + + +def get_tracker_response_with_created_bodies(response) -> dict: + """Get a dictionary response from a TrackerCommandResponse gRPC object including created bodies. + + Parameters + ---------- + response : TrackerCommandResponse + The gRPC TrackerCommandResponse object. + + Returns + ------- + dict + A dictionary representing the tracker response with created bodies. + """ + serialized_response = get_standard_tracker_response(response) + serialized_response["created_bodies"] = serialized_response["tracker_response"].get( + "created_bodies", [] + ) + return serialized_response diff --git a/src/ansys/geometry/core/_grpc/_services/v1/prepare_tools.py b/src/ansys/geometry/core/_grpc/_services/v1/prepare_tools.py index 3ad0170b45..5d22caf230 100644 --- a/src/ansys/geometry/core/_grpc/_services/v1/prepare_tools.py +++ b/src/ansys/geometry/core/_grpc/_services/v1/prepare_tools.py @@ -25,7 +25,15 @@ from ansys.geometry.core.errors import protect_grpc +from ..base.conversions import from_measurement_to_server_length from ..base.prepare_tools import GRPCPrepareToolsService +from .conversions import ( + build_grpc_id, + from_enclosure_options_to_grpc_enclosure_options, + get_standard_tracker_response, + get_tracker_response_with_created_bodies, + serialize_tracker_command_response, +) class GRPCPrepareToolsServiceV1(GRPCPrepareToolsService): # pragma: no cover @@ -43,54 +51,315 @@ class GRPCPrepareToolsServiceV1(GRPCPrepareToolsService): # pragma: no cover @protect_grpc def __init__(self, channel: grpc.Channel): # noqa: D102 - from ansys.api.geometry.v1.preparetools_pb2_grpc import PrepareToolsStub + from ansys.api.discovery.v1.operations.prepare_pb2_grpc import PrepareStub - self.stub = PrepareToolsStub(channel) + self.stub = PrepareStub(channel) @protect_grpc def extract_volume_from_faces(self, **kwargs) -> dict: # noqa: D102 - raise NotImplementedError + from ansys.api.discovery.v1.operations.prepare_pb2 import ExtractVolumeFromFacesRequest + + # Create the request - assumes all inputs are valid and of the proper type + request = ExtractVolumeFromFacesRequest( + sealing_face_ids=[build_grpc_id(face) for face in kwargs["sealing_faces"]], + inside_face_ids=[build_grpc_id(face) for face in kwargs["inside_faces"]], + ) + + # Call the gRPC service + response = self.stub.ExtractVolumeFromFaces(request) + + # Return the response - formatted as a dictionary + return get_tracker_response_with_created_bodies(response) @protect_grpc def extract_volume_from_edge_loops(self, **kwargs) -> dict: # noqa: D102 - raise NotImplementedError + from ansys.api.discovery.v1.operations.prepare_pb2 import ExtractVolumeFromEdgeLoopsRequest + + # Create the request - assumes all inputs are valid and of the proper type + request = ExtractVolumeFromEdgeLoopsRequest( + sealing_edge_ids=[build_grpc_id(edge) for edge in kwargs["sealing_edges"]], + inside_face_ids=[build_grpc_id(face) for face in kwargs["inside_faces"]], + ) + + # Call the gRPC service + response = self.stub.ExtractVolumeFromEdgeLoops(request) + + # Return the response - formatted as a dictionary + return get_tracker_response_with_created_bodies(response) @protect_grpc def remove_rounds(self, **kwargs) -> dict: # noqa: D102 - raise NotImplementedError + from ansys.api.discovery.v1.operations.prepare_pb2 import RemoveRoundsRequest + from google.protobuf.wrappers_pb2 import BoolValue + + # Create the request - assumes all inputs are valid and of the proper type + request = RemoveRoundsRequest( + selection_ids=[build_grpc_id(round) for round in kwargs["rounds"]], + auto_shrink=BoolValue(value=kwargs["auto_shrink"]), + ) + + # Call the gRPC service + response = self.stub.RemoveRounds(request) + + # Return the response - formatted as a dictionary + return get_standard_tracker_response(response) @protect_grpc def share_topology(self, **kwargs) -> dict: # noqa: D102 - raise NotImplementedError + from ansys.api.discovery.v1.operations.prepare_pb2 import ShareTopologyRequest + from google.protobuf.wrappers_pb2 import BoolValue, DoubleValue + + # Create the request - assumes all inputs are valid and of the proper type + request = ShareTopologyRequest( + selection_ids=[build_grpc_id(body) for body in kwargs["bodies"]], + tolerance=DoubleValue(value=from_measurement_to_server_length(kwargs["tolerance"])), + preserve_instances=BoolValue(value=kwargs["preserve_instances"]), + ) + + # Call the gRPC service + response = self.stub.ShareTopology(request) + + # Return the response - formatted as a dictionary + return get_standard_tracker_response(response) @protect_grpc def enhanced_share_topology(self, **kwargs) -> dict: # noqa: D102 - raise NotImplementedError + from ansys.api.discovery.v1.operations.prepare_pb2 import ShareTopologyRequest + from google.protobuf.wrappers_pb2 import BoolValue, DoubleValue + + # Create the request - assumes all inputs are valid and of the proper type + request = ShareTopologyRequest( + selection_ids=[build_grpc_id(body) for body in kwargs["bodies"]], + tolerance=DoubleValue(value=from_measurement_to_server_length(kwargs["tolerance"])), + preserve_instances=BoolValue(value=kwargs["preserve_instances"]), + ) + + # Call the gRPC service + response = self.stub.EnhancedShareTopology(request) + + # Return the response - formatted as a dictionary + return { + "success": response.success, + "found": response.found, + "repaired": response.repaired, + "tracker_response": serialize_tracker_command_response(response.tracked_changes), + "created_bodies_monikers": [ + created_body.id + for created_body in response.tracked_changes.get("created_bodies", []) + ], + "modified_bodies_monikers": [ + modified_body.id + for modified_body in response.tracked_changes.get("modified_bodies", []) + ], + } @protect_grpc def find_logos(self, **kwargs) -> dict: # noqa: D102 - raise NotImplementedError + from ansys.api.discovery.v1.operations.prepare_pb2 import FindLogoOptions, FindLogosRequest + + # Check height objects + min_height = ( + from_measurement_to_server_length(kwargs["min_height"]) + if kwargs["min_height"] is not None + else None + ) + max_height = ( + from_measurement_to_server_length(kwargs["max_height"]) + if kwargs["max_height"] is not None + else None + ) + + # Create the request - assumes all inputs are valid and of the proper type + request = FindLogosRequest( + body_ids=[build_grpc_id(body) for body in kwargs["bodies"]], + options=FindLogoOptions( + min_height=min_height, + max_height=max_height, + ), + ) + + # Call the gRPC service + response = self.stub.FindLogos(request) + + # Return the response - formatted as a dictionary + return { + "id": response.id, + "face_ids": [face.id for face in response.logo_faces], + } @protect_grpc def find_and_remove_logos(self, **kwargs) -> dict: # noqa: D102 - raise NotImplementedError + from ansys.api.discovery.v1.operations.prepare_pb2 import FindLogoOptions, FindLogosRequest + + # Check height objects + min_height = ( + from_measurement_to_server_length(kwargs["min_height"]) + if kwargs["min_height"] is not None + else None + ) + max_height = ( + from_measurement_to_server_length(kwargs["max_height"]) + if kwargs["max_height"] is not None + else None + ) + + # Create the request - assumes all inputs are valid and of the proper type + request = FindLogosRequest( + body_ids=[build_grpc_id(body) for body in kwargs["bodies"]], + options=FindLogoOptions( + min_height=min_height, + max_height=max_height, + ), + ) + + # Call the gRPC service + response = self.stub.FindAndRemoveLogos(request) + + # Return the response - formatted as a dictionary + return get_standard_tracker_response(response) @protect_grpc - def remove_logo(self, **kwargs) -> dict: # noqa: D102 - raise NotImplementedError + def remove_logo(self, **kwargs): # noqa: D102 + from ansys.api.discovery.v1.operations.prepare_pb2 import RemoveLogoRequest + + # Create the request - assumes all inputs are valid and of the proper type + request = RemoveLogoRequest( + face_ids=[build_grpc_id(id) for id in kwargs["face_ids"]], + ) + + # Call the gRPC service + response = self.stub.RemoveLogo(request) + + # Return the response - formatted as a dictionary + return get_standard_tracker_response(response) @protect_grpc def detect_helixes(self, **kwargs) -> dict: # noqa: D102 - raise NotImplementedError + from ansys.api.discovery.v1.operations.prepare_pb2 import ( + FindHelixesOptions, + ) + + from ansys.geometry.core.shapes.parameterization import Interval + + from ..base.conversions import ( + from_measurement_to_server_length, + to_distance, + ) + from .conversions import ( + from_grpc_curve_to_curve, + from_grpc_point_to_point3d, + ) + + # Create the request - assumes all inputs are valid and of the proper type + request = FindHelixesOptions( + body_ids=[build_grpc_id(body) for body in kwargs["bodies"]], + options=FindHelixesOptions( + min_radius=from_measurement_to_server_length(kwargs["min_radius"]), + max_radius=from_measurement_to_server_length(kwargs["max_radius"]), + fit_radius_error=from_measurement_to_server_length(kwargs["fit_radius_error"]), + ), + ) + + # Call the gRPC service + response = self.stub.FindHelixes(request) + + # If no helixes, return empty dictionary + if len(response.helixes) == 0: + return {"helixes": []} + + # Return the response - formatted as a dictionary + return { + "helixes": [ + { + "trimmed_curve": { + "geometry": from_grpc_curve_to_curve(helix.trimmed_curve.curve), + "start": from_grpc_point_to_point3d(helix.trimmed_curve.start), + "end": from_grpc_point_to_point3d(helix.trimmed_curve.end), + "interval": Interval( + helix.trimmed_curve.interval_start, helix.trimmed_curve.interval_end + ), + "length": to_distance(helix.trimmed_curve.length).value, + }, + "edges": [ + { + "id": edge.id, + "parent_id": edge.parent.id, + "curve_type": edge.curve_type, + "is_reversed": edge.is_reversed, + } + for edge in helix.edges + ], + } + for helix in response.helixes + ] + } @protect_grpc def create_box_enclosure(self, **kwargs) -> dict: # noqa: D102 - raise NotImplementedError + from ansys.api.discovery.v1.operations.prepare_pb2 import CreateEnclosureBoxRequest + + grpc_enclosure_options = from_enclosure_options_to_grpc_enclosure_options( + kwargs["enclosure_options"] + ) + + # Create the request - assumes all inputs are valid and of the proper type + request = CreateEnclosureBoxRequest( + body_ids=[build_grpc_id(body.id) for body in kwargs["bodies"]], + x_low=from_measurement_to_server_length(kwargs["x_low"]), + x_high=from_measurement_to_server_length(kwargs["x_high"]), + y_low=from_measurement_to_server_length(kwargs["y_low"]), + y_high=from_measurement_to_server_length(kwargs["y_high"]), + z_low=from_measurement_to_server_length(kwargs["z_low"]), + z_high=from_measurement_to_server_length(kwargs["z_high"]), + enclosure_options=grpc_enclosure_options, + ) + + # Call the gRPC service + response = self.stub.CreateEnclosureBox(request) + + # Return the response - formatted as a dictionary + return get_tracker_response_with_created_bodies(response) @protect_grpc def create_cylinder_enclosure(self, **kwargs) -> dict: # noqa: D102 - raise NotImplementedError + from ansys.api.discovery.v1.operations.prepare_pb2 import CreateEnclosureCylinderRequest + + grpc_enclosure_options = from_enclosure_options_to_grpc_enclosure_options( + kwargs["enclosure_options"] + ) + + # Create the request - assumes all inputs are valid and of the proper type + request = CreateEnclosureCylinderRequest( + body_ids=[build_grpc_id(body.id) for body in kwargs["bodies"]], + axial_distance_low=from_measurement_to_server_length(kwargs["axial_distance_low"]), + axial_distance_high=from_measurement_to_server_length(kwargs["axial_distance_high"]), + radial_distance=from_measurement_to_server_length(kwargs["radial_distance"]), + enclosure_options=grpc_enclosure_options, + ) + + # Call the gRPC service + response = self.stub.CreateEnclosureCylinder(request) + + # Return the response - formatted as a dictionary + return get_tracker_response_with_created_bodies(response) @protect_grpc def create_sphere_enclosure(self, **kwargs) -> dict: # noqa: D102 - raise NotImplementedError + from ansys.api.discovery.v1.operations.prepare_pb2 import CreateEnclosureSphereRequest + + grpc_enclosure_options = from_enclosure_options_to_grpc_enclosure_options( + kwargs["enclosure_options"] + ) + + # Create the request - assumes all inputs are valid and of the proper type + request = CreateEnclosureSphereRequest( + body_ids=[build_grpc_id(body.id) for body in kwargs["bodies"]], + radial_distance=from_measurement_to_server_length(kwargs["radial_distance"]), + enclosure_options=grpc_enclosure_options, + ) + + # Call the gRPC service + response = self.stub.CreateEnclosureSphere(request) + + # Return the response - formatted as a dictionary + return get_tracker_response_with_created_bodies(response) diff --git a/src/ansys/geometry/core/tools/prepare_tools.py b/src/ansys/geometry/core/tools/prepare_tools.py index 2b712205e1..9f5284e799 100644 --- a/src/ansys/geometry/core/tools/prepare_tools.py +++ b/src/ansys/geometry/core/tools/prepare_tools.py @@ -154,7 +154,10 @@ def extract_volume_from_faces( if response.get("success"): bodies_ids = response.get("created_bodies") if len(bodies_ids) > 0: - parent_design._update_design_inplace() + if not pyansys_geom.USE_TRACKER_TO_UPDATE_DESIGN: + parent_design._update_design_inplace() + else: + parent_design._update_from_tracker(response.get("tracker_response")) return get_bodies_from_ids(parent_design, bodies_ids) else: self._grpc_client.log.info("Failed to extract volume from faces...") @@ -209,7 +212,10 @@ def extract_volume_from_edge_loops( if response.get("success"): bodies_ids = response.get("created_bodies") if len(bodies_ids) > 0: - parent_design._update_design_inplace() + if not pyansys_geom.USE_TRACKER_TO_UPDATE_DESIGN: + parent_design._update_design_inplace() + else: + parent_design._update_from_tracker(response.get("tracker_response")) return get_bodies_from_ids(parent_design, bodies_ids) else: self._grpc_client.log.info("Failed to extract volume from edge loops...") @@ -249,7 +255,10 @@ def remove_rounds(self, faces: list["Face"], auto_shrink: bool = False) -> bool: ) if response.get("success"): - parent_design._update_design_inplace() + if not pyansys_geom.USE_TRACKER_TO_UPDATE_DESIGN: + parent_design._update_design_inplace() + else: + parent_design._update_from_tracker(response.get("tracker_response")) else: self._grpc_client.log.info("Failed to remove rounds...") @@ -297,6 +306,12 @@ def share_topology( preserve_instances=preserve_instances, ) + parent_design = get_design_from_body(bodies[0]) + if not pyansys_geom.USE_TRACKER_TO_UPDATE_DESIGN: + parent_design._update_design_inplace() + else: + parent_design._update_from_tracker(response.get("tracker_response")) + return response.get("success") @min_backend_version(25, 2, 0) @@ -343,12 +358,18 @@ def enhanced_share_topology( preserve_instances=preserve_instances, ) + parent_design = get_design_from_body(bodies[0]) + if not pyansys_geom.USE_TRACKER_TO_UPDATE_DESIGN: + parent_design._update_design_inplace() + else: + parent_design._update_from_tracker(response.get("tracker_response")) + message = RepairToolMessage( success=response.get("success"), - created_bodies=response.get("created_bodies_monikers"), - modified_bodies=response.get("modified_bodies_monikers"), found=response.get("found"), repaired=response.get("repaired"), + created_bodies=response.get("created_bodies_monikers"), + modified_bodies=response.get("modified_bodies_monikers"), ) return message