"""
Geometry component
"""
from __future__ import annotations
import json
import os
import threading
from enum import Enum
from typing import Any, List, Literal, Optional, Union
import pydantic as pd
from flow360_schema.framework.physical_dimensions import Length
from flow360_schema.framework.validation.context import DeserializationContext
from flow360_schema.geometry.selection_evaluation import (
dedup,
modeling_kernel,
render_topology_tree,
)
from flow360_schema.models.entities.geometry_entities import Edge, GeometryBodyGroup
from flow360_schema.models.entities.surface_entities import Surface
from flow360_schema.models.simulation.framework.updater_utils import (
deprecation_reminder,
)
from pydantic import TypeAdapter
from flow360.cloud.flow360_requests import (
GeometryFileMeta,
LengthUnitType,
NewGeometryDependencyRequest,
NewGeometryRequest,
)
from flow360.cloud.heartbeat import post_upload_heartbeat
from flow360.cloud.rest_api import RestApi
from flow360.component.interfaces import GeometryInterface
from flow360.component.resource_base import (
AssetMetaBaseModelV2,
Flow360Resource,
ResourceDraft,
SubmissionMode,
)
from flow360.component.resource_status import is_final_resource_status
from flow360.component.simulation.folder import Folder
from flow360.component.simulation.web.asset_base import AssetBase
from flow360.component.utils import (
SUPPORTED_GEOMETRY_FILE_PATTERNS,
GeometryFiles,
MeshNameParser,
resolve_length_unit,
shared_account_confirm_proceed,
)
from flow360.environment import current_environment
from flow360.exceptions import Flow360FileError, Flow360ValueError
from flow360.log import log
from flow360.solver_version import warn_if_minor_release
GeometryWorkflow = Literal["standard", "catalyst"]
class GeometryStatus(Enum):
"""Status of geometry resource, the is_final method is overloaded"""
ERROR = "error"
UPLOADED = "uploaded"
UPLOADING = "uploading"
RUNNING = "running"
GENERATING = "generating"
PROCESSED = "processed"
DELETED = "deleted"
PENDING = "pending"
UNKNOWN = "unknown"
def is_final(self):
"""
Checks if status is final for geometry resource
Returns
-------
bool
True if status is final, False otherwise.
"""
return is_final_resource_status("Geometry", self)
# pylint: disable=R0801
class GeometryMeta(AssetMetaBaseModelV2):
"""
GeometryMeta component
"""
status: GeometryStatus = pd.Field() # Overshadowing to ensure correct is_final() method
dependency: bool = pd.Field(False)
class GeometryDraft(ResourceDraft):
"""
Unified Geometry Draft component for uploading geometry files.
This class handles both:
- Creating a new project with geometry as the root asset
- Adding geometry as a dependency to an existing project
The submission mode is determined by how the draft is created (via factory methods
on the Geometry class) and affects the behavior of the submit() method.
All geometries are conceptually equivalent - they are components that can be used
to create the final geometry for simulation. The distinction between "root" and
"dependency" is only about where the geometry is uploaded (new project vs existing
project), not about any fundamental difference in the geometry itself.
"""
# pylint: disable=too-many-arguments, too-many-instance-attributes
def __init__(
self,
file_names: Union[List[str], str],
cad_importer_version: Literal["v1", "v2"],
project_name: str = None,
solver_version: str = None,
length_unit: Optional[LengthUnitType] = None,
tags: List[str] = None,
folder: Optional[Folder] = None,
workflow: GeometryWorkflow = "standard",
):
"""
Initialize a GeometryDraft with common attributes.
For creating a new project (root geometry):
Use Geometry.from_file() which sets project_name, solver_version, folder
For adding to existing project (dependency geometry):
Use Geometry.import_to_project() which sets the dependency context
Parameters
----------
file_names : Union[List[str], str]
Path(s) to the geometry file(s)
length_unit : LengthUnitType, optional
Unit of length (default is "m")
tags : List[str], optional
Tags to assign to the geometry (default is None)
project_name : str, optional
Name of the project (for project root mode)
solver_version : str, optional
Solver version (for project root mode)
folder : Optional[Folder], optional
Parent folder (for project root mode)
"""
self._file_names = file_names
self.project_name = project_name
self.tags = tags if tags is not None else []
self.length_unit = resolve_length_unit(length_unit)
self.solver_version = solver_version
self.folder = folder
self.workflow = workflow
# FXC-3289: CAD Importer version frozen at upload (None == backend default v1).
self.cad_importer_version = cad_importer_version
# pylint: disable=fixme
# TODO: create a DependableResourceDraft for GeometryDraft and SurfaceMeshDraft
self.dependency_name = None
self.dependency_project_id = None
self._submission_mode: SubmissionMode = SubmissionMode.PROJECT_ROOT
self._validate_geometry()
ResourceDraft.__init__(self)
def _validate_geometry(self):
"""Validate geometry files and length unit."""
if not isinstance(self.file_names, list) or len(self.file_names) == 0:
raise Flow360FileError("file_names field has to be a non-empty list.")
try:
GeometryFiles(file_names=self.file_names)
except pd.ValidationError as e:
raise Flow360FileError(str(e)) from e
for geometry_file in self.file_names:
if not os.path.exists(geometry_file):
raise Flow360FileError(f"{geometry_file} not found.")
if self.length_unit not in LengthUnitType.__args__:
raise Flow360ValueError(
f"specified length_unit : {self.length_unit} is invalid. "
f"Valid options are: {list(LengthUnitType.__args__)}"
)
if self.workflow not in ("standard", "catalyst"):
raise Flow360ValueError(
f"specified workflow : {self.workflow} is invalid. "
"Valid options are: ['standard', 'catalyst']"
)
def _set_default_project_name(self):
"""Set default project name if not provided for project creation."""
if self.project_name is None:
self.project_name = os.path.splitext(os.path.basename(self.file_names[0]))[0]
log.warning(
"`project_name` is not provided. "
f"Using the first geometry file name {self.project_name} as project name."
)
def _validate_submission_context(self):
"""Validate context for submission based on mode."""
if self._submission_mode is None:
raise ValueError("[Internal] Geometry submission context not set.")
if self._submission_mode == SubmissionMode.PROJECT_ROOT and self.solver_version is None:
raise Flow360ValueError("solver_version field is required.")
if self._submission_mode == SubmissionMode.PROJECT_DEPENDENCY:
if self.dependency_name is None or self.dependency_project_id is None:
raise ValueError(
"[Internal] Dependency name and project ID must be set for geometry dependency submission."
)
@property
def file_names(self) -> List[str]:
"""Geometry file paths as a list."""
if isinstance(self._file_names, str):
return [self._file_names]
return self._file_names
def set_dependency_context(
self,
name: str,
project_id: str,
) -> None:
"""
Configure this draft to add geometry to an existing project.
Called internally by Geometry.import_to_project().
"""
self._submission_mode = SubmissionMode.PROJECT_DEPENDENCY
self.dependency_name = name
self.dependency_project_id = project_id
def _preprocess_mapbc_files(self) -> List[str]:
"""Find and return associated mapbc files for UGRID geometry files."""
mapbc_files = []
for file_path in self.file_names:
mesh_parser = MeshNameParser(file_path)
if mesh_parser.is_ugrid() and os.path.isfile(
mesh_parser.get_associated_mapbc_filename()
):
file_name_mapbc = mesh_parser.get_associated_mapbc_filename()
mapbc_files.append(file_name_mapbc)
return mapbc_files
def _create_project_root_resource(
self, mapbc_files: List[str], description: str = ""
) -> GeometryMeta:
"""Create a new geometry resource that will be the root of a new project."""
self._set_default_project_name()
req = NewGeometryRequest(
name=self.project_name,
solver_version=self.solver_version,
tags=self.tags,
files=[
GeometryFileMeta(
name=os.path.basename(file_path),
type="main",
)
for file_path in self.file_names + mapbc_files
],
parent_folder_id=self.folder.id if self.folder else "ROOT.FLOW360",
length_unit=self.length_unit,
description=description,
use_catalyst=self.workflow == "catalyst",
cad_importer_version=self.cad_importer_version,
)
resp = RestApi(GeometryInterface.endpoint, environment_provider=current_environment).post(
req.dict()
)
return GeometryMeta(**resp)
def _create_dependency_resource(
self, mapbc_files: List[str], description: str = "", draft_id: str = "", icon: str = ""
) -> GeometryMeta:
"""Create a geometry resource as a dependency in an existing project."""
req = NewGeometryDependencyRequest(
name=self.dependency_name,
project_id=self.dependency_project_id,
draft_id=draft_id,
files=[
GeometryFileMeta(
name=os.path.basename(file_path),
type="main",
)
for file_path in self.file_names + mapbc_files
],
length_unit=self.length_unit,
tags=self.tags,
description=description,
icon=icon,
cad_importer_version=self.cad_importer_version,
)
resp = RestApi(GeometryInterface.endpoint, environment_provider=current_environment).post(
req.dict(), method="dependency"
)
return GeometryMeta(**resp)
def _upload_files(
self,
info: GeometryMeta,
mapbc_files: List[str],
progress_callback=None,
) -> Geometry:
"""Upload geometry files to the cloud."""
# pylint: disable=protected-access
geometry = Geometry(info.id)
heartbeat_info = {"resourceId": info.id, "resourceType": "Geometry", "stop": False}
# Keep posting the heartbeat to keep server patient about uploading.
heartbeat_thread = threading.Thread(target=post_upload_heartbeat, args=(heartbeat_info,))
heartbeat_thread.start()
try:
for file_path in self.file_names + mapbc_files:
geometry._webapi._upload_file(
remote_file_name=os.path.basename(file_path),
file_name=file_path,
progress_callback=progress_callback,
)
finally:
heartbeat_info["stop"] = True
heartbeat_thread.join()
# Kick off pipeline
geometry._webapi._complete_upload()
# Setting _id will disable "WARNING: You have not submitted..." warning message
self._id = info.id
return geometry
# pylint: disable=duplicate-code
def submit(
self,
description: str = "",
progress_callback=None,
run_async: bool = False,
draft_id: str = "",
icon: str = "",
) -> Geometry:
"""
Submit geometry to cloud.
The behavior depends on how this draft was created:
- If created via Geometry.from_file(): Creates a new project with this geometry as root
- If created via Geometry.import_to_project(): Adds geometry to an existing project
Parameters
----------
description : str, optional
Description of the geometry/project (default is "")
progress_callback : callback, optional
Use for custom progress bar (default is None)
run_async : bool, optional
Whether to return immediately after upload without waiting for processing
(default is False)
draft_id : str, optional
ID of the draft to add geometry to (only used for dependency mode, default is "")
icon : str, optional
Icon for the geometry (only used for dependency mode, default is "")
Returns
-------
Geometry
Geometry object with id
Raises
------
Flow360ValueError
If submission context is not set or user aborts
"""
self._validate_geometry()
self._validate_submission_context()
if not shared_account_confirm_proceed():
raise Flow360ValueError("User aborted resource submit.")
mapbc_files = self._preprocess_mapbc_files()
# Create the geometry resource based on submission mode
if self._submission_mode == SubmissionMode.PROJECT_ROOT:
info = self._create_project_root_resource(mapbc_files, description)
log_message = "Geometry successfully submitted"
else:
info = self._create_dependency_resource(mapbc_files, description, draft_id, icon)
log_message = "New geometry successfully submitted to the project"
# Upload files
geometry = self._upload_files(info, mapbc_files, progress_callback)
log.info(f"{log_message}: {geometry.short_description()}")
if run_async:
return geometry
log.info("Waiting for geometry to be processed.")
return Geometry.from_cloud(info.id)
[docs]
class Geometry(AssetBase): # pylint: disable=too-many-public-methods
"""
Geometry component for workbench (simulation V2)
"""
_interface_class = GeometryInterface
_meta_class = GeometryMeta
_draft_class = GeometryDraft
_web_api_class = Flow360Resource
_cloud_resource_type_name = "Geometry"
# Input files accepted by from_file(): CAD source files plus surface-mesh files.
_default_download_patterns = (
[f"*{ext}" for ext in SUPPORTED_GEOMETRY_FILE_PATTERNS]
+ [f"*{pattern}" for pattern in MeshNameParser.all_patterns("surface")]
+ ["*.mapbc"]
)
# pylint: disable=redefined-builtin
def __init__(self, id: Union[str, None]):
super().__init__(id)
self._project_length_unit = None
self._topology_tree = None
@property
@deprecation_reminder(
"`Geometry.face_group_tag` is deprecated and will be replaced by a new API "
"in a future release.",
)
def face_group_tag(self):
"getter for face_group_tag"
return self._entity_info.face_group_tag
@face_group_tag.setter
@deprecation_reminder("")
def face_group_tag(self, new_value: str):
raise SyntaxError("Cannot set face_group_tag.")
@property
@deprecation_reminder(
"`Geometry.edge_group_tag` is deprecated and will be replaced by a new API "
"in a future release.",
)
def edge_group_tag(self):
"getter for edge_group_tag"
return self._entity_info.edge_group_tag
@edge_group_tag.setter
@deprecation_reminder("")
def edge_group_tag(self, new_value: str):
raise SyntaxError("Cannot set edge_group_tag.")
@property
@deprecation_reminder(
"`Geometry.body_group_tag` is deprecated and will be replaced by a new API "
"in a future release.",
)
def body_group_tag(self):
"getter for body_group_tag"
return self._entity_info.body_group_tag
@body_group_tag.setter
@deprecation_reminder("")
def body_group_tag(self, new_value: str):
raise SyntaxError("Cannot set body_group_tag.")
def _apply_dynamic_default_settings(self, simulation_dict: dict):
"""Populate this geometry's dynamic defaults from the asset's simulation dict.
Part of asset initialization, not a user-facing accessor. Mutates
``default_settings["geometry_accuracy"]`` and ``_project_length_unit``.
"""
def _get_default_geometry_accuracy(simulation_dict: dict):
"""Get the default geometry accuracy from the simulation json"""
if simulation_dict.get("meshing") is None:
return None
if simulation_dict["meshing"].get("defaults") is None:
return None
if simulation_dict["meshing"]["defaults"].get("geometry_accuracy") is None:
return None
return simulation_dict["meshing"]["defaults"]["geometry_accuracy"]
self.default_settings["geometry_accuracy"] = (
self._entity_info.default_geometry_accuracy
if self._entity_info.default_geometry_accuracy
else _get_default_geometry_accuracy(simulation_dict=simulation_dict)
)
# Cache project length unit for OBB (avoids extra API call in create_draft)
asset_cache = simulation_dict.get("private_attribute_asset_cache", {})
length_unit_raw = asset_cache.get("project_length_unit")
if length_unit_raw is not None:
adapter = TypeAdapter(Length.PositiveFloat64)
with DeserializationContext():
self._project_length_unit = adapter.validate_python(length_unit_raw)
else:
self._project_length_unit = None
[docs]
@classmethod
# pylint: disable=redefined-builtin
def from_cloud(cls, id: str, **kwargs) -> Geometry:
"""Create asset with the given ID"""
asset_obj = super().from_cloud(id, **kwargs)
return asset_obj
[docs]
@classmethod
# pylint: disable=too-many-arguments
def from_file(
cls,
file_names: Union[List[str], str],
project_name: str = None,
solver_version: str = None,
length_unit: Optional[LengthUnitType] = None,
tags: List[str] = None,
folder: Optional[Folder] = None,
workflow: GeometryWorkflow = "standard",
cad_importer_version: Literal["v1", "v2"] = "v1",
) -> GeometryDraft:
warn_if_minor_release(solver_version)
return GeometryDraft(
file_names=file_names,
project_name=project_name,
solver_version=solver_version,
length_unit=length_unit,
tags=tags,
folder=folder,
workflow=workflow,
cad_importer_version=cad_importer_version,
)
[docs]
@classmethod
# pylint: disable=too-many-arguments
def import_to_project(
cls,
name: str,
file_names: Union[List[str], str],
project_id: str,
length_unit: Optional[LengthUnitType] = None,
tags: List[str] = None,
cad_importer_version: Literal["v1", "v2"] = "v1",
) -> GeometryDraft:
"""
Create a geometry draft for adding to an existing project.
This creates a geometry that will be added as a supplementary component
(dependency) to an existing project, rather than creating a new project.
Parameters
----------
name : str
Name for the geometry component
file_names : Union[List[str], str]
Path(s) to the geometry file(s)
project_id : str
ID of the existing project to add this geometry to
length_unit : LengthUnitType, optional
Unit of length (default is "m")
tags : List[str], optional
Tags to assign to the geometry (default is None)
cad_importer_version : Literal["v1", "v2"], optional
CAD Importer version for the dependency geometry. Normally inherited
from the project root so the meshes stay compatible (default "v1").
Returns
-------
GeometryDraft
A draft configured for submission to an existing project
"""
draft = GeometryDraft(
file_names=file_names,
length_unit=length_unit,
tags=tags,
cad_importer_version=cad_importer_version,
)
draft.set_dependency_context(name=name, project_id=project_id)
return draft
[docs]
def show_available_groupings(self, verbose_mode: bool = False):
"""Display all the possible groupings for faces and edges"""
self._show_available_entity_groups(
"faces",
ignored_attribute_tags=["__all__", "faceId"],
show_ids_in_each_group=verbose_mode,
)
self._show_available_entity_groups(
"edges",
ignored_attribute_tags=["__all__", "edgeId"],
show_ids_in_each_group=verbose_mode,
)
self._show_available_entity_groups(
"bodies",
ignored_attribute_tags=["__all__", "bodyId"],
show_ids_in_each_group=verbose_mode,
)
[docs]
@classmethod
def from_local_storage(
cls, geometry_id: str = None, local_storage_path="", meta_data: GeometryMeta = None
) -> Geometry:
"""
Parameters
----------
geometry_id : str
ID of the geometry resource
local_storage_path:
The folder of the project, defaults to current working directory
Returns
-------
Geometry
Geometry object
"""
return super()._from_local_storage(
asset_id=geometry_id, local_storage_path=local_storage_path, meta_data=meta_data
)
def _show_available_entity_groups(
self,
entity_type_name: Literal["faces", "edges", "bodies"],
ignored_attribute_tags: list = None,
show_ids_in_each_group: bool = False,
) -> None:
"""
Display all the grouping info for the given entity type
"""
if entity_type_name not in ["faces", "edges", "bodies"]:
raise Flow360ValueError(
f"entity_type_name: {entity_type_name} is invalid. Valid options are: ['faces', 'edges', 'bodies']"
)
# pylint: disable=no-member
if entity_type_name == "faces":
attribute_names = self._entity_info.face_attribute_names
grouped_items = self._entity_info.grouped_faces
elif entity_type_name == "edges":
attribute_names = self._entity_info.edge_attribute_names
grouped_items = self._entity_info.grouped_edges
else:
attribute_names = self._entity_info.body_attribute_names
grouped_items = self._entity_info.grouped_bodies
log.info(f" >> Available attribute tags for grouping **{entity_type_name}**:")
for tag_index, attribute_tag in enumerate(attribute_names):
if ignored_attribute_tags is not None and attribute_tag in ignored_attribute_tags:
continue
log.info(
f" >> Tag '{tag_index}': {attribute_tag}. Grouping with this tag results in:"
)
for index, entity in enumerate(grouped_items[tag_index]):
log.info(f" >> [{index}]: {entity.name}")
if show_ids_in_each_group is True:
log.info(f" IDs: {entity.private_attribute_sub_components}")
[docs]
@deprecation_reminder(
"`Geometry.group_faces_by_tag()` is deprecated and will be replaced by a new API "
"in a future release.",
)
def group_faces_by_tag(self, tag_name: str) -> None:
"""
Group faces by tag name
.. deprecated::
This function will be replaced by a new API in a future release.
"""
# pylint: disable=protected-access,no-member
self.internal_registry = self._entity_info._group_entity_by_tag(
"face", tag_name, self.internal_registry
)
[docs]
@deprecation_reminder(
"`Geometry.group_edges_by_tag()` is deprecated and will be replaced by a new API "
"in a future release.",
)
def group_edges_by_tag(self, tag_name: str) -> None:
"""
Group edges by tag name
.. deprecated::
This function will be replaced by a new API in a future release.
"""
# pylint: disable=protected-access,no-member
self.internal_registry = self._entity_info._group_entity_by_tag(
"edge", tag_name, self.internal_registry
)
[docs]
@deprecation_reminder(
"`Geometry.group_bodies_by_tag()` is deprecated and will not be supported.",
)
def group_bodies_by_tag(self, tag_name: str) -> None:
"""
Group bodies by tag name
.. deprecated::
This function will not be supported.
"""
# pylint: disable=protected-access,no-member
self.internal_registry = self._entity_info._group_entity_by_tag(
"body", tag_name, self.internal_registry
)
[docs]
@deprecation_reminder(
"`Geometry.reset_face_grouping()` is deprecated and will be replaced by a new API "
"in a future release.",
)
def reset_face_grouping(self) -> None:
"""Reset the face grouping
.. deprecated::
This function will be replaced by a new API in a future release.
"""
# pylint: disable=protected-access,no-member
self.internal_registry = self._entity_info._reset_grouping("face", self.internal_registry)
[docs]
@deprecation_reminder(
"`Geometry.reset_edge_grouping()` is deprecated and will be replaced by a new API "
"in a future release.",
)
def reset_edge_grouping(self) -> None:
"""Reset the edge grouping
.. deprecated::
This function will be replaced by a new API in a future release.
"""
# pylint: disable=protected-access,no-member
self.internal_registry = self._entity_info._reset_grouping("edge", self.internal_registry)
[docs]
@deprecation_reminder(
"`Geometry.reset_body_grouping()` is deprecated and will not be supported.",
)
def reset_body_grouping(self) -> None:
"""Reset the body grouping
.. deprecated::
This function will not be supported.
"""
# pylint: disable=protected-access,no-member
self.internal_registry = self._entity_info._reset_grouping("body", self.internal_registry)
def _rename_entity(
self,
entity_type_name: Literal["face", "edge", "body"],
current_name_pattern: str,
new_name_prefix: str,
):
"""
Rename the entity
Parameters
----------
entity_type_name : Literal["face", "edge", "body"]
The type of entity that needs renaming
current_name_pattern:
The current name of a single entity or the name pattern of the entities
new_name_prefix:
The new name of a single entity or the new name prefix of the entities
"""
# pylint: disable=too-many-boolean-expressions
# Read the tags from entity_info directly: the public properties are
# deprecated and renaming should not emit their warnings.
if (
(entity_type_name == "face" and not self._entity_info.face_group_tag)
or (entity_type_name == "edge" and not self._entity_info.edge_group_tag)
or (entity_type_name == "body" and not self._entity_info.body_group_tag)
):
raise Flow360ValueError(
f"Renaming failed: Could not find {entity_type_name} grouping info in the draft's simulation settings."
"Please group them first before renaming the entities."
)
matched_entities = self.internal_registry.find_by_naming_pattern(
pattern=current_name_pattern
)
if entity_type_name == "body":
matched_entities = [
entity for entity in matched_entities if isinstance(entity, GeometryBodyGroup)
]
if entity_type_name == "face":
matched_entities = [
entity for entity in matched_entities if isinstance(entity, Surface)
]
if entity_type_name == "edge":
matched_entities = [entity for entity in matched_entities if isinstance(entity, Edge)]
matched_entities = sorted(
matched_entities,
key=lambda x: x.name,
)
if len(matched_entities) == 0:
raise Flow360ValueError(
f"Renaming failed: No entity is found to match the input name pattern: {current_name_pattern}."
)
for idx, entity in enumerate(matched_entities):
new_name = (
f"{new_name_prefix}_{(idx+1):04d}" if len(matched_entities) > 1 else new_name_prefix
)
if self.internal_registry.find_by_naming_pattern(new_name):
raise Flow360ValueError(
f"Renaming failed: An entity with the new name: {new_name} already exists."
)
entity._force_set_attr("name", new_name) # pylint:disable=protected-access
[docs]
def rename_edges(self, current_name_pattern: str, new_name_prefix: str):
"""
Rename the edge in the current edge group
Parameters
----------
current_name_pattern:
The current name of a single edge or the name pattern of the edges
new_name_prefix:
The new name of a single edge or the new name prefix of the edges
"""
self._rename_entity(
entity_type_name="edge",
current_name_pattern=current_name_pattern,
new_name_prefix=new_name_prefix,
)
[docs]
def rename_surfaces(self, current_name_pattern: str, new_name_prefix: str):
"""
Rename the face in the current face group
Parameters
----------
current_name_pattern:
The current name of a single face or the name pattern of the faces
new_name_prefix:
The new name of a single face or the new name prefix of the faces
"""
self._rename_entity(
entity_type_name="face",
current_name_pattern=current_name_pattern,
new_name_prefix=new_name_prefix,
)
[docs]
def rename_body_groups(self, current_name_pattern: str, new_name_prefix: str):
"""
Rename the body in the current body group
Parameters
----------
current_name_pattern:
The current name of a single body or the name pattern of the bodies
new_name_prefix:
The new name of a single body or the new name prefix of the bodies
"""
self._rename_entity(
entity_type_name="body",
current_name_pattern=current_name_pattern,
new_name_prefix=new_name_prefix,
)
@deprecation_reminder(
"accessing entities via `geometry[...]` is deprecated and will be replaced by a new API "
"in a future release.",
)
def __getitem__(self, key: str):
"""
Get the entity by name.
`key` is the name of the entity or the naming pattern if wildcard is used.
.. deprecated::
Accessing entities via `geometry[...]` will be replaced by a new API in a future
release.
"""
# pylint: disable=import-outside-toplevel
from flow360.component.simulation.draft_context import get_active_draft
if get_active_draft() is not None:
log.warning(
"Accessing entities via asset[key] while a DraftContext is active. "
"Use draft.surfaces[key] or draft.body_groups[key] instead to ensure "
"modifications are tracked in the draft's entity_info."
)
if isinstance(key, str) is False:
raise Flow360ValueError(f"Entity naming pattern: {key} is not a string.")
if hasattr(self, "internal_registry") is False or self.internal_registry is None:
raise Flow360ValueError("The faces/edges/bodies in geometry are not grouped yet.")
return self.internal_registry.find_by_naming_pattern(
key, enforce_output_as_list=False, error_when_no_match=True
)
def __setitem__(self, key: str, value: Any):
raise NotImplementedError("Assigning/setting entities is not supported.")
# ------------------------------------------------------------------
# Rule-based face/entity selection (schema-side; grouping itself is draft state)
# ------------------------------------------------------------------
def _get_topology_tree(self) -> dict:
"""Fetch (once, cached) the geometry's topology tree via the same endpoint
the Web UI uses: ``GET /v2/geometries/<id>/topology-tree``."""
if self._topology_tree is not None:
return self._topology_tree
try:
resp = self._webapi.get(method="topology-tree")
except Exception as exc: # pylint: disable=broad-except
try:
status = self._webapi.status.value
except Exception: # pylint: disable=broad-except
# Diagnosing the failure must never replace it.
status = "unavailable"
raise Flow360ValueError(
"Could not fetch the geometry's topology tree "
f"(GET v2/geometries/{self.id}/topology-tree): {exc}. "
f"Geometry status: '{status}' — the tree exists only once processing "
"has completed."
) from exc
tree_file = resp["topologyTreeData"]
if tree_file is None:
raise Flow360ValueError(
f"Geometry {self.id} has no topology tree: it is written when processing "
"completes, by solver versions that support rule-based face grouping."
)
self._topology_tree = json.loads(tree_file)["topologyTree"]
return self._topology_tree
@property
def topology_tree(self) -> dict:
"""The geometry's topology tree (fetched once, cached). Read-only info:
rule AUTHORING is instance-agnostic (the module-level ``root`` token from
flow360_schema.geometry.selection); to EVALUATE a rule against this
geometry, pass rules to create_draft(face_grouping=...) or hand this tree
to ``resolve_rule`` / ``resolve_selection``."""
return self._get_topology_tree()
@property
def modeling_kernel(self):
"""The geometry's ModelFile ``modelingKernel`` (CADGeometry, EGADSGeometry,
BinarySTL, ASCIISTL, UGRID, or CGNS); a set if a multi-file geometry mixes
kernels, or None for a pre-kernel tree."""
return modeling_kernel(self._get_topology_tree())
@property
def face_ids(self) -> list:
"""Every face id of the geometry (all leaf faces parsed into entity_info), as
an ordered, de-duplicated list."""
return dedup(self._entity_info.all_face_ids)
[docs]
def show_topology_tree(self, show_attributes: bool = True, max_depth=None) -> str:
"""Return a printable ASCII rendering of the geometry's topology tree (the same
tree the rule engine walks). Each node shows every field it carries as
key=value: name, color, material, attributes -- the fields rules are written
against. A trailing summary lists the distinct face colors.
show_attributes: include each node's attributes map (default True; pass False
to hide it -- real CAD attribute maps can be large). max_depth: limit depth
below the root (default None = full tree)."""
return render_topology_tree(self._get_topology_tree(), show_attributes, max_depth)
