Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 25 additions & 8 deletions flow360/component/simulation/framework/param_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -152,40 +152,57 @@ def register_entity_list(model: Flow360BaseModel, registry: EntityRegistry) -> N
register_entity_list(field, registry)


def _update_entity_full_name(
def _update_entity_full_name( # pylint:disable=too-many-branches
model: Flow360BaseModel,
target_entity_type: Union[type[_SurfaceEntityBase], type[_VolumeEntityBase]],
volume_mesh_meta_data: dict,
):
) -> List[str]:
"""
Update Surface/Boundary with zone name from volume mesh metadata.

Returns a list of warning messages for entities that could not be found.
"""
warnings = []
for field in model.__dict__.values():
# Skip the AssetCache since updating there makes no difference
if isinstance(field, AssetCache):
continue

if isinstance(field, target_entity_type):
# pylint: disable=protected-access
field._update_entity_info_with_metadata(volume_mesh_meta_data)
warning = field._update_entity_info_with_metadata(volume_mesh_meta_data)
if warning is not None:
warnings.append(warning)

if isinstance(field, EntityList):
for entity in field.stored_entities:
if isinstance(entity, target_entity_type):
# pylint: disable=protected-access
entity._update_entity_info_with_metadata(volume_mesh_meta_data)
warning = entity._update_entity_info_with_metadata(volume_mesh_meta_data)
if warning is not None:
warnings.append(warning)

elif isinstance(field, (list, tuple)):
for item in field:
if isinstance(item, target_entity_type):
item._update_entity_info_with_metadata( # pylint: disable=protected-access
volume_mesh_meta_data
warning = (
item._update_entity_info_with_metadata( # pylint: disable=protected-access
volume_mesh_meta_data
)
)
if warning is not None:
warnings.append(warning)
elif isinstance(item, Flow360BaseModel):
_update_entity_full_name(item, target_entity_type, volume_mesh_meta_data)
warnings.extend(
_update_entity_full_name(item, target_entity_type, volume_mesh_meta_data)
)

elif isinstance(field, Flow360BaseModel):
_update_entity_full_name(field, target_entity_type, volume_mesh_meta_data)
warnings.extend(
_update_entity_full_name(field, target_entity_type, volume_mesh_meta_data)
)

return warnings


def _update_zone_boundaries_with_metadata(
Expand Down
33 changes: 21 additions & 12 deletions flow360/component/simulation/primitives.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,11 +23,14 @@
from flow360.component.simulation.user_code.core.types import ValueOrExpression
from flow360.component.simulation.utils import BoundingBoxType, model_attribute_unlock
from flow360.component.types import Axis
from flow360.exceptions import Flow360BoundaryMissingError


def _get_boundary_full_name(surface_name: str, volume_mesh_meta: dict[str, dict]) -> str:
"""Ideally volume_mesh_meta should be a pydantic model.
def _get_boundary_full_name(
surface_name: str, volume_mesh_meta: dict[str, dict]
) -> tuple[Union[str, None], Union[str, None]]:
"""Return (full_name, warning_message).

Ideally volume_mesh_meta should be a pydantic model.

TODO: Note that the same surface_name may appear in different blocks. E.g.
`farFieldBlock/slipWall`, and `plateBlock/slipWall`. Currently the mesher does not support splitting boundary into
Expand All @@ -40,16 +43,18 @@ def _get_boundary_full_name(surface_name: str, volume_mesh_meta: dict[str, dict]
if (
match is not None and match.group(1) == surface_name
) or existing_boundary_name == surface_name:
return existing_boundary_name
return (existing_boundary_name, None)
if surface_name == "symmetric":
# Provides more info when the symmetric boundary is not auto generated.
raise Flow360BoundaryMissingError(
return (
None,
f"Parent zone not found for boundary: {surface_name}. "
"It is likely that it was never auto generated because the condition is not met."
"It is likely that it was never auto generated because the condition is not met.",
)
raise Flow360BoundaryMissingError(
return (
None,
f"Parent zone not found for surface {surface_name}. "
"It may have been deleted due to overlapping with generated symmetry plane."
"It may have been deleted due to overlapping with generated symmetry plane.",
)


Expand Down Expand Up @@ -217,14 +222,18 @@ class _SurfaceEntityBase(EntityBase, metaclass=ABCMeta):
private_attribute_registry_bucket_name: Literal["SurfaceEntityType"] = "SurfaceEntityType"
private_attribute_full_name: Optional[str] = pd.Field(None, frozen=True)

def _update_entity_info_with_metadata(self, volume_mesh_meta_data: dict) -> None:
def _update_entity_info_with_metadata(self, volume_mesh_meta_data: dict) -> Optional[str]:
"""
Update parent zone name once the volume mesh is done.

Returns warning message if boundary not found, None otherwise.
"""
full_name, warning = _get_boundary_full_name(self.name, volume_mesh_meta_data)
if warning is not None:
return warning
with model_attribute_unlock(self, "private_attribute_full_name"):
self.private_attribute_full_name = _get_boundary_full_name(
self.name, volume_mesh_meta_data
)
self.private_attribute_full_name = full_name
return None

@property
def full_name(self):
Expand Down
11 changes: 7 additions & 4 deletions flow360/component/simulation/simulation_params.py
Original file line number Diff line number Diff line change
Expand Up @@ -633,20 +633,23 @@ def used_entity_registry(self) -> EntityRegistry:
registry = self._update_entity_private_attrs(registry)
return registry

def _update_param_with_actual_volume_mesh_meta(self, volume_mesh_meta_data: dict):
def _update_param_with_actual_volume_mesh_meta(self, volume_mesh_meta_data: dict) -> list[str]:
"""
Update the zone info from the actual volume mesh before solver execution.
Will be executed in the casePipeline as part of preprocessing.
Some thoughts:
Do we also need to update the params when the **surface meshing** is done?

Returns a list of warning messages for entities that could not be found in metadata.
"""
# pylint:disable=no-member
used_entity_registry = self.used_entity_registry
# Below includes the Ghost entities.
_update_entity_full_name(self, _SurfaceEntityBase, volume_mesh_meta_data)
_update_entity_full_name(self, _VolumeEntityBase, volume_mesh_meta_data)
warnings = []
warnings.extend(_update_entity_full_name(self, _SurfaceEntityBase, volume_mesh_meta_data))
warnings.extend(_update_entity_full_name(self, _VolumeEntityBase, volume_mesh_meta_data))
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Volume entities silently fail without generating warnings

Medium Severity

The PR converts surface entity "zone not found" errors to warnings, but _VolumeEntityBase._update_entity_info_with_metadata still returns None and was not updated to return warnings. When _update_entity_full_name is called with _VolumeEntityBase, it attempts to collect warnings from _update_entity_info_with_metadata, but the method always returns None. This means volume zones that can't be found in metadata will silently fail to update without generating any warning, defeating the PR's purpose of surfacing these issues as warnings.

Additional Locations (1)

Fix in Cursor Fix in Web

_update_zone_boundaries_with_metadata(used_entity_registry, volume_mesh_meta_data)
return self
return warnings

def is_steady(self):
"""
Expand Down