"""
Volume mesh component
"""
# pylint: disable=too-many-lines
from __future__ import annotations
import os.path
import threading
from enum import Enum
from functools import cached_property
from typing import Any, Iterator, List, Optional, Union
import numpy as np
import pydantic as pd_v2
# This will be split into two files in a future commit...
# For now we keep it to not mix new features with file
# structure refactors
import pydantic.v1 as pd
from flow360.cloud.compress_upload import compress_and_upload_chunks
from flow360.cloud.flow360_requests import (
CopyExampleVolumeMeshRequest,
LengthUnitType,
NewVolumeMeshRequestV2,
)
from flow360.cloud.heartbeat import post_upload_heartbeat
from flow360.cloud.rest_api import RestApi
from flow360.component.simulation.folder import Folder
from flow360.component.utils import VolumeMeshFile
from flow360.component.v1.cloud.flow360_requests import NewVolumeMeshRequest
from flow360.component.v1.meshing.params import VolumeMeshingParams
from flow360.exceptions import (
Flow360CloudFileError,
Flow360FileError,
Flow360NotImplementedError,
Flow360RuntimeError,
Flow360ValueError,
)
from flow360.flags import Flags
from flow360.log import log
from flow360.solver_version import Flow360Version
from .case import Case, CaseDraft
from .interfaces import VolumeMeshInterface, VolumeMeshInterfaceV2
from .resource_base import (
AssetMetaBaseModel,
AssetMetaBaseModelV2,
Flow360Resource,
Flow360ResourceListBase,
ResourceDraft,
)
from .results.base_results import PerEntityResultCSVModel
from .simulation.primitives import GenericVolume, Surface
from .simulation.web.asset_base import AssetBase
from .types import COMMENTS
from .utils import (
CompressionFormat,
MeshFileFormat,
MeshNameParser,
UGRIDEndianness,
shared_account_confirm_proceed,
validate_type,
zstd_compress,
)
from .v1.boundaries import NoSlipWall
from .v1.flow360_params import Flow360MeshParams, Flow360Params, _GenericBoundaryWrapper
from .v1.params_base import params_generic_validator
from .validator import Validator
def get_datatype(dataset):
"""
Get datatype of dataset
:param dataset:
:return:
"""
data_raw = np.empty(dataset.shape, dataset.dtype)
dataset.read_direct(data_raw)
data_str = "".join([chr(i) for i in dataset])
return data_str
def get_no_slip_walls(params: Union[Flow360Params, Flow360MeshParams]):
"""
Get wall boundary names
:param params:
:return:
"""
assert params
if (
isinstance(params, Flow360MeshParams)
and params.boundaries
and params.boundaries.no_slip_walls
):
return params.boundaries.no_slip_walls
if isinstance(params, Flow360Params) and params.boundaries:
return [
wall_name
for wall_name, wall in params.boundaries.dict().items()
# pylint: disable=no-member
if wall_name != COMMENTS and _GenericBoundaryWrapper(v=wall).v.type == NoSlipWall().type
]
return []
def get_boundaries_from_sliding_interfaces(params: Union[Flow360Params, Flow360MeshParams]):
"""
Get wall boundary names
:param params:
:return:
"""
assert params
res = []
# Sliding interfaces are deprecated - we need to handle this somehow
# if params.sliding_interfaces and params.sliding_interfaces.rotating_patches:
# res += params.sliding_interfaces.rotating_patches[:]
# if params.sliding_interfaces and params.sliding_interfaces.stationary_patches:
# res += params.sliding_interfaces.stationary_patches[:]
return res
# pylint: disable=too-many-branches
def get_boundaries_from_file(cgns_file: str, solver_version: str = None):
"""
Get boundary names from CGNS file
:param cgns_file:
:param solver_version:
:return:
"""
import h5py # pylint: disable=import-outside-toplevel
names = []
with h5py.File(cgns_file, "r") as h5_file:
base = h5_file["Base"]
for zone_name, zone in base.items():
if zone_name == " data":
continue
if zone.attrs["label"].decode() != "Zone_t":
continue
zone_type = get_datatype(base[f"{zone_name}/ZoneType/ data"])
if zone_type not in ["Structured", "Unstructured"]:
continue
for section_name, section in zone.items():
if section_name == " data":
continue
if "label" not in section.attrs:
continue
if solver_version and Flow360Version(solver_version) < Flow360Version(
"release-22.2.1.0"
):
if section.attrs["label"].decode() != "Elements_t":
continue
element_type_tag = int(zone[f"{section_name}/ data"][0])
if element_type_tag in [5, 7]:
names.append(f"{zone_name}/{section_name}")
if element_type_tag == 20:
first_element_type_tag = zone[f"{section_name}/ElementConnectivity/ data"][
0
]
if first_element_type_tag in [5, 7]:
names.append(f"{zone_name}/{section_name}")
else:
if section.attrs["label"].decode() != "ZoneBC_t":
continue
for bc_name, bc_zone in section.items():
if bc_zone.attrs["label"].decode() == "BC_t":
names.append(f"{zone_name}/{bc_name}")
return names
def validate_cgns(
cgns_file: str, params: Union[Flow360Params, Flow360MeshParams], solver_version=None
):
"""
Validate CGNS file
:param cgns_file:
:param params:
:param solver_version:
:return:
"""
assert cgns_file
assert params
boundaries_in_file = get_boundaries_from_file(cgns_file, solver_version)
boundaries_in_params = get_no_slip_walls(params) + get_boundaries_from_sliding_interfaces(
params
)
boundaries_in_file = set(boundaries_in_file)
boundaries_in_params = set(boundaries_in_params)
if not boundaries_in_file.issuperset(boundaries_in_params):
raise Flow360ValueError(
"The following input boundary names from mesh json are not found in mesh:"
+ f" {' '.join(boundaries_in_params - boundaries_in_file)}."
+ f" Boundary names in cgns: {' '.join(boundaries_in_file)}"
+ f" Boundary names in params: {' '.join(boundaries_in_file)}"
)
log.info(
f'Notice: {" ".join(boundaries_in_file - boundaries_in_params)} is '
+ "tagged as wall in cgns file, but not in input params"
)
class VolumeMeshLog(Enum):
"""
Volume mesh log
"""
USER_LOG = "user.log"
PY_LOG = "validateFlow360Mesh.py.log"
class VolumeMeshDownloadable(Enum):
"""
Volume mesh downloadable files
"""
CONFIG_JSON = "config.json"
BOUNDING_BOX = "meshBoundaryBoundingBox.json"
# pylint: disable=E0213
class VolumeMeshMeta(AssetMetaBaseModel, extra=pd.Extra.allow):
"""
VolumeMeshMeta component
"""
id: str = pd.Field(alias="meshId")
name: str = pd.Field(alias="meshName")
created_at: str = pd.Field(alias="meshAddTime")
surface_mesh_id: Optional[str] = pd.Field(alias="surfaceMeshId")
mesh_params: Union[Flow360MeshParams, None, dict] = pd.Field(alias="meshParams")
mesh_format: Union[MeshFileFormat, None] = pd.Field(alias="meshFormat")
file_name: Union[str, None] = pd.Field(alias="fileName")
endianness: UGRIDEndianness = pd.Field(alias="meshEndianness")
compression: CompressionFormat = pd.Field(alias="meshCompression")
boundaries: Union[List, None]
@pd.validator("mesh_params", pre=True)
def init_mesh_params(cls, value):
"""
validator for mesh_params
"""
return params_generic_validator(value, Flow360MeshParams)
@pd.validator("endianness", pre=True)
def init_endianness(cls, value):
"""
validator for endianess
"""
return UGRIDEndianness(value) or UGRIDEndianness.NONE
@pd.validator("compression", pre=True)
def init_compression(cls, value):
"""
validator for compression
"""
try:
return CompressionFormat(value)
except ValueError:
return CompressionFormat.NONE
def to_volume_mesh(self) -> VolumeMesh:
"""
returns VolumeMesh object from volume mesh meta info
"""
return VolumeMesh(self.id)
class VolumeMeshDraft(ResourceDraft):
"""
Volume mesh draft component (before submit)
"""
# pylint: disable=too-many-arguments, too-many-instance-attributes
def __init__(
self,
file_name: str = None,
params: Union[Flow360MeshParams, VolumeMeshingParams] = None,
name: str = None,
surface_mesh_id=None,
tags: List[str] = None,
solver_version=None,
endianess: UGRIDEndianness = None,
isascii: bool = False,
):
if file_name is not None and not os.path.exists(file_name):
raise Flow360FileError(f"File '{file_name}' not found.")
if endianess is not None:
raise Flow360NotImplementedError(
"endianess selections not supported, it is inferred from filename"
)
if isascii is True:
raise Flow360NotImplementedError("isascii not supported")
self.params = None
if params is not None:
if not isinstance(params, Flow360MeshParams) and not isinstance(
params, VolumeMeshingParams
):
raise ValueError(
f"params={params} are not of type Flow360MeshParams OR VolumeMeshingParams"
)
self.params = params.copy(deep=True)
if file_name is not None:
mesh_parser = MeshNameParser(file_name)
if not mesh_parser.is_valid_volume_mesh():
raise Flow360ValueError(
f"Unsupported volume mesh file extensions: {mesh_parser.format.ext()}. "
f"Supported: [{MeshFileFormat.UGRID.ext()},{MeshFileFormat.CGNS.ext()}]."
)
if name is None and file_name is not None:
name = os.path.splitext(os.path.basename(file_name))[0]
self.file_name = file_name
self.name = name
self.surface_mesh_id = surface_mesh_id
self.tags = tags
self.solver_version = solver_version
self._id = None
self.compress_method = CompressionFormat.ZST
ResourceDraft.__init__(self)
def _submit_from_surface(self, force_submit: bool = False):
self.validator_api(
self.params, solver_version=self.solver_version, raise_on_error=(not force_submit)
)
body = {
"name": self.name,
"tags": self.tags,
"surfaceMeshId": self.surface_mesh_id,
"config": self.params.flow360_json(),
"format": "cgns",
}
if Flags.beta_features() and self.params.version is not None:
body["version"] = self.params.version
if Flags.beta_features() and self.params.version is not None:
if self.params.version == "v2":
body["format"] = "aflr3"
if self.solver_version:
body["solverVersion"] = self.solver_version
resp = RestApi(VolumeMeshInterface.endpoint).post(body)
if not resp:
return None
info = VolumeMeshMeta(**resp)
# setting _id will disable "WARNING: You have not submitted..." warning message
self._id = info.id
mesh = VolumeMesh(self.id)
log.info(f"VolumeMesh successfully submitted: {mesh.short_description()}")
return mesh
# pylint: disable=protected-access, too-many-locals
def _submit_upload_mesh(self, progress_callback=None):
assert os.path.exists(self.file_name)
mesh_parser = MeshNameParser(self.file_name)
mesh_format = mesh_parser.format
endianness = mesh_parser.endianness
original_compression = mesh_parser.compression
compression = (
original_compression
if original_compression != CompressionFormat.NONE
else self.compress_method
)
if mesh_format is MeshFileFormat.CGNS:
remote_file_name = "volumeMesh"
else:
remote_file_name = "mesh"
remote_file_name = (
f"{remote_file_name}{endianness.ext()}{mesh_format.ext()}{compression.ext()}"
)
name = self.name
if name is None:
name = os.path.splitext(os.path.basename(self.file_name))[0]
if Flags.beta_features() and self.params is not None:
req = NewVolumeMeshRequest(
name=name,
file_name=remote_file_name,
tags=self.tags,
format=mesh_format.value,
endianness=endianness.value,
compression=compression.value,
params=self.params,
solver_version=self.solver_version,
version=self.params.version,
)
else:
req = NewVolumeMeshRequest(
name=name,
file_name=remote_file_name,
tags=self.tags,
format=mesh_format.value,
endianness=endianness.value,
compression=compression.value,
params=self.params,
solver_version=self.solver_version,
)
resp = RestApi(VolumeMeshInterface.endpoint).post(req.dict())
if not resp:
return None
info = VolumeMeshMeta(**resp)
# setting _id will disable "WARNING: You have not submitted..." warning message
self._id = info.id
mesh = VolumeMesh(self.id)
# parallel compress and upload
if (
original_compression == CompressionFormat.NONE
and self.compress_method == CompressionFormat.BZ2
):
upload_id = mesh.create_multipart_upload(remote_file_name)
compress_and_upload_chunks(self.file_name, upload_id, mesh, remote_file_name)
elif (
original_compression == CompressionFormat.NONE
and self.compress_method == CompressionFormat.ZST
):
compressed_file_name = zstd_compress(self.file_name)
mesh._upload_file(
remote_file_name, compressed_file_name, progress_callback=progress_callback
)
os.remove(compressed_file_name)
else:
mesh._upload_file(remote_file_name, self.file_name, progress_callback=progress_callback)
mesh._complete_upload(remote_file_name)
log.info(f"VolumeMesh successfully uploaded: {mesh.short_description()}")
return mesh
def submit(self, progress_callback=None, force_submit: bool = False) -> VolumeMesh:
"""submit mesh to cloud
Parameters
----------
progress_callback : callback, optional
Use for custom progress bar, by default None
Returns
-------
VolumeMesh
VolumeMesh object with id
"""
if not shared_account_confirm_proceed():
raise Flow360ValueError("User aborted resource submit.")
if self.file_name is not None:
return self._submit_upload_mesh(progress_callback)
if self.surface_mesh_id is not None and self.name is not None and self.params is not None:
return self._submit_from_surface(force_submit=force_submit)
raise Flow360ValueError(
"You must provide volume mesh file for upload or surface mesh Id with meshing parameters."
)
@classmethod
def validator_api(
cls, params: VolumeMeshingParams, solver_version=None, raise_on_error: bool = True
):
"""
validation api: validates surface meshing parameters before submitting
"""
return Validator.VOLUME_MESH.validate(
params, solver_version=solver_version, raise_on_error=raise_on_error
)
[docs]
class VolumeMesh(Flow360Resource):
"""
Volume mesh component
"""
# pylint: disable=redefined-builtin
def __init__(self, id: str):
super().__init__(
interface=VolumeMeshInterface,
meta_class=VolumeMeshMeta,
id=id,
)
self.__mesh_params = None
@classmethod
def _from_meta(cls, meta: VolumeMeshMeta):
validate_type(meta, "meta", VolumeMeshMeta)
volume_mesh = cls(id=meta.id)
volume_mesh._set_meta(meta)
return volume_mesh
@property
def info(self) -> VolumeMeshMeta:
return super().info
@property
def _mesh_params(self) -> Flow360MeshParams:
"""
returns mesh params
"""
if self.__mesh_params is None:
self.__mesh_params = self.info.mesh_params
return self.__mesh_params
@property
def no_slip_walls(self):
"""
returns mesh no_slip_walls
"""
if self._mesh_params is None:
return None
return self._mesh_params.boundaries.no_slip_walls
@property
def all_boundaries(self):
"""
returns mesh no_slip_walls
"""
return self.info.boundaries
# pylint: disable=too-many-arguments,R0801
def download_file(
self,
file_name: Union[str, VolumeMeshDownloadable],
to_file=None,
to_folder=".",
overwrite: bool = True,
progress_callback=None,
**kwargs,
):
"""
Download file from surface mesh
:param file_name:
:param to_file:
:return:
"""
if isinstance(file_name, VolumeMeshDownloadable):
file_name = file_name.value
return super()._download_file(
file_name,
to_file=to_file,
to_folder=to_folder,
overwrite=overwrite,
progress_callback=progress_callback,
**kwargs,
)
# pylint: disable=R0801
def download(self, to_file=None, to_folder=".", overwrite: bool = True):
"""
Download volume mesh file
:param to_file:
:return:
"""
status = self.status
if not status.is_final():
log.warning(f"Cannot download file because status={status}")
return None
remote_file_name = self.info.file_name
if remote_file_name is None:
remote_file_name = self._remote_file_name()
return super()._download_file(
remote_file_name,
to_file=to_file,
to_folder=to_folder,
overwrite=overwrite,
)
# pylint: disable=arguments-differ
def _complete_upload(self, remote_file_name):
"""
Complete volume mesh upload
:return:
"""
resp = self.post({}, method=f"completeUpload?fileName={remote_file_name}")
self._info = VolumeMeshMeta(**resp)
@classmethod
def _interface(cls):
return VolumeMeshInterface
@classmethod
def _meta_class(cls):
"""
returns volume mesh meta info class: VolumeMeshMeta
"""
return VolumeMeshMeta
@classmethod
def _params_ancestor_id_name(cls):
"""
returns surfaceMeshId name
"""
return "surfaceMeshId"
@classmethod
def from_cloud(cls, mesh_id: str):
"""
Get volume mesh info from cloud
:param mesh_id:
:return:
"""
return cls(mesh_id)
def _remote_file_name(self):
"""
mesh filename on cloud
"""
remote_file_name = None
for file in self.get_download_file_list():
try:
MeshNameParser(file["fileName"])
remote_file_name = file["fileName"]
except Flow360RuntimeError:
continue
if remote_file_name is None:
raise Flow360CloudFileError(f"No volume mesh file found for id={self.id}")
return remote_file_name
@classmethod
def from_file(
cls,
file_name: str,
params: Union[Flow360MeshParams, None] = None,
name: str = None,
tags: List[str] = None,
solver_version=None,
endianess: UGRIDEndianness = None,
isascii: bool = False,
) -> VolumeMeshDraft:
"""
Upload volume mesh from file
:param volume_mesh_name:
:param file_name:
:param params:
:param tags:
:param solver_version:
:return:
"""
return VolumeMeshDraft(
file_name=file_name,
name=name,
tags=tags,
solver_version=solver_version,
params=params,
endianess=endianess,
isascii=isascii,
)
@classmethod
def copy_from_example(
cls,
example_id: str,
name: str = None,
) -> VolumeMesh:
"""
Create a new volume mesh by copying from an example mesh identified by `example_id`.
Parameters
----------
example_id : str
The unique identifier of the example volume mesh to copy from.
name : str, optional
The name to assign to the new volume mesh. If not provided, the name
of the example volume mesh will be used.
Returns
-------
VolumeMesh
A new instance of VolumeMesh copied from the example mesh if successful.
Examples
--------
>>> new_mesh = VolumeMesh.copy_from_example('example_id_123', name='New Mesh')
"""
if name is None:
eg_vm = cls(example_id)
name = eg_vm.name
req = CopyExampleVolumeMeshRequest(example_id=example_id, name=name)
resp = RestApi(f"{VolumeMeshInterface.endpoint}/examples/copy").post(req.dict())
if not resp:
raise RuntimeError("Something went wrong when accessing example mesh.")
info = VolumeMeshMeta(**resp)
return cls(info.id)
@classmethod
def create(
cls,
name: str,
params: VolumeMeshingParams,
surface_mesh_id: str,
tags: List[str] = None,
solver_version=None,
) -> VolumeMeshDraft:
"""
Create volume mesh from surface mesh
"""
return VolumeMeshDraft(
name=name,
surface_mesh_id=surface_mesh_id,
solver_version=solver_version,
params=params,
tags=tags,
)
def create_case(
self,
name: str,
params: Flow360Params,
tags: List[str] = None,
solver_version: str = None,
) -> CaseDraft:
"""
Create new case
:param name:
:param params:
:param tags:
:return:
"""
return Case.create(
name, params, volume_mesh_id=self.id, tags=tags, solver_version=solver_version
)
class VolumeMeshList(Flow360ResourceListBase):
"""
VolumeMesh List component
"""
def __init__(
self,
surface_mesh_id: str = None,
from_cloud: bool = True,
include_deleted: bool = False,
limit=100,
):
super().__init__(
ancestor_id=surface_mesh_id,
from_cloud=from_cloud,
include_deleted=include_deleted,
limit=limit,
resourceClass=VolumeMesh,
)
def filter(self):
"""
flitering list, not implemented yet
"""
raise NotImplementedError("Filters are not implemented yet")
# resp = list(filter(lambda i: i['caseStatus'] != 'deleted', resp))
# pylint: disable=useless-parent-delegation
def __getitem__(self, index) -> VolumeMesh:
"""
returns VolumeMeshMeta item of the list
"""
return super().__getitem__(index)
# pylint: disable=useless-parent-delegation
def __iter__(self) -> Iterator[VolumeMesh]:
return super().__iter__()
###==== V2 API version ===###
class VolumeMeshStatusV2(Enum):
"""Status of volume mesh resource, the is_final method is overloaded"""
SUBMITTED = "submitted"
UPLOADING = "uploading"
UPLOADED = "uploaded"
COMPLETED = "completed"
PENDING = "pending"
GENERATING = "generating"
ERROR = "error"
def is_final(self):
"""
Checks if status is final for volume mesh resource
Returns
-------
bool
True if status is final, False otherwise.
"""
if self in [VolumeMeshStatusV2.COMPLETED, VolumeMeshStatusV2.ERROR]:
return True
return False
class VolumeMeshMetaV2(AssetMetaBaseModelV2):
"""
VolumeMeshMetaV2 component
"""
status: VolumeMeshStatusV2 = pd_v2.Field() # Overshadowing to ensure correct is_final() method
file_name: Optional[str] = pd_v2.Field(None, alias="fileName")
class VolumeMeshStats(pd_v2.BaseModel):
"""
Mesh stats
"""
n_nodes: int = pd_v2.Field(..., alias="nNodes")
n_triangles: int = pd_v2.Field(..., alias="nTriangles")
n_quadrilaterals: int = pd_v2.Field(..., alias="nQuadrilaterals")
n_tetrahedron: int = pd_v2.Field(..., alias="nTetrahedron")
n_prism: int = pd_v2.Field(..., alias="nPrism")
n_pyramid: int = pd_v2.Field(..., alias="nPyramid")
n_hexahedron: int = pd_v2.Field(..., alias="nHexahedron")
n_tet_wedge: int = pd_v2.Field(..., alias="nTetWedge")
class VolumeMeshBoundingBox(PerEntityResultCSVModel):
"""
VolumeMeshBoundingBox
"""
remote_file_name: str = pd.Field(VolumeMeshDownloadable.BOUNDING_BOX.value, frozen=True)
_variables: Optional[List[str]] = None
@property
def entities(self):
"""
Returns list of entities (boundary names) available for this result
"""
return self.values.keys()
def _filtered_sum(self):
pass
def _get_range(self, df, min_key: str, max_key: str) -> float:
if min_key not in df.index or max_key not in df.index:
return 0.0
min_val = df.loc[min_key].min()
max_val = df.loc[max_key].max()
return max_val - min_val
@property
def length(self) -> float:
"""
Compute and return the length of the bounding box.
The length is calculated as the difference between the maximum and minimum
x-coordinate values from the bounding box data.
Returns:
float: The computed length along the x-axis.
"""
df = self.as_dataframe()
return self._get_range(df, "xmin", "xmax")
@property
def width(self) -> float:
"""
Compute and return the width of the bounding box.
The width is calculated as the difference between the maximum and minimum
y-coordinate values from the bounding box data.
Returns:
float: The computed width along the y-axis.
"""
df = self.as_dataframe()
return self._get_range(df, "ymin", "ymax")
@property
def height(self) -> float:
"""
Compute and return the height of the bounding box.
The height is calculated as the difference between the maximum and minimum
z-coordinate values from the bounding box data.
Returns:
float: The computed height along the z-axis.
"""
df = self.as_dataframe()
return self._get_range(df, "zmin", "zmax")
class VolumeMeshDraftV2(ResourceDraft):
"""
Volume mesh draft component
"""
# pylint: disable=too-many-arguments
def __init__(
self,
file_names: str,
project_name: str = None,
solver_version: str = None,
length_unit: LengthUnitType = "m",
tags: List[str] = None,
folder: Optional[Folder] = None,
):
self.file_name = file_names
self.project_name = project_name
self.tags = tags if tags is not None else []
self.length_unit = length_unit
self.solver_version = solver_version
self.folder = folder
self._validate()
ResourceDraft.__init__(self)
def _validate(self):
self._validate_volume_mesh()
def _validate_volume_mesh(self):
if self.file_name is not None:
try:
VolumeMeshFile(file_names=self.file_name)
except pd.ValidationError as e:
raise Flow360FileError(str(e)) from e
if self.project_name is None:
self.project_name = os.path.splitext(os.path.basename(self.file_name))[0]
log.warning(
"`project_name` is not provided. "
f"Using the volume mesh file name {self.project_name} as project name."
)
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.solver_version is None:
raise Flow360ValueError("solver_version field is required.")
# pylint: disable=protected-access, too-many-locals
def submit(
self, description="", progress_callback=None, compress=True, run_async=False
) -> VolumeMeshV2:
"""
Submit volume mesh to cloud and create a new project
Parameters
----------
description : str, optional
description of the project, by default ""
progress_callback : callback, optional
Use for custom progress bar, by default None
compress : boolean, optional
Compress the volume mesh file when sending to S3, default is True
fetch_entities : boolean, optional
Whether to fetch and populate the entity info object after submitting, default is False
run_async : bool, optional
Whether to submit volume mesh asynchronously (default is False).
Returns
-------
VolumeMeshV2
Volume mesh object with id
"""
self._validate()
if not shared_account_confirm_proceed():
raise Flow360ValueError("User aborted resource submit.")
mesh_parser = MeshNameParser(self.file_name)
original_compression = mesh_parser.compression
mesh_format = mesh_parser.format
file_name_no_compression = mesh_parser.file_name_no_compression
compression = original_compression
do_compression = False
if compress and original_compression == CompressionFormat.NONE:
compression = CompressionFormat.ZST
do_compression = True
original_file_with_compression = f"{file_name_no_compression}{compression.ext()}"
req = NewVolumeMeshRequestV2(
name=self.project_name,
solver_version=self.solver_version,
tags=self.tags,
file_name=original_file_with_compression,
parent_folder_id=self.folder.id if self.folder else "ROOT.FLOW360",
length_unit=self.length_unit,
format=mesh_format.value,
description=description,
)
# Create new volume mesh resource and project
req_dict = req.dict()
resp = RestApi(VolumeMeshInterfaceV2.endpoint).post(req_dict)
info = VolumeMeshMetaV2(**resp)
# setting _id will disable "WARNING: You have not submitted..." warning message
self._id = info.id
renamed_file_on_remote = info.file_name
volume_mesh = VolumeMeshV2(info.id)
# Upload volume mesh file, keep posting the heartbeat to keep server patient about uploading.
heartbeat_info = {"resourceId": info.id, "resourceType": "VolumeMesh", "stop": False}
heartbeat_thread = threading.Thread(target=post_upload_heartbeat, args=(heartbeat_info,))
heartbeat_thread.start()
# Compress (if not already compressed) and upload
if do_compression:
zstd_compress(self.file_name, original_file_with_compression)
volume_mesh._webapi._upload_file(
renamed_file_on_remote,
original_file_with_compression,
progress_callback=progress_callback,
)
os.remove(original_file_with_compression)
else:
volume_mesh._webapi._upload_file(
renamed_file_on_remote, self.file_name, progress_callback=progress_callback
)
if mesh_parser.is_ugrid():
expected_local_mapbc_file = mesh_parser.get_associated_mapbc_filename()
if os.path.isfile(expected_local_mapbc_file):
remote_mesh_parser = MeshNameParser(renamed_file_on_remote)
volume_mesh._webapi._upload_file(
remote_mesh_parser.get_associated_mapbc_filename(),
mesh_parser.get_associated_mapbc_filename(),
progress_callback=progress_callback,
)
else:
log.warning(
f"The expected mapbc file {expected_local_mapbc_file} specifying "
"user-specified boundary names doesn't exist."
)
heartbeat_info["stop"] = True
heartbeat_thread.join()
# Start processing pipeline
volume_mesh._webapi._complete_upload()
log.info(f"VolumeMesh successfully submitted: {volume_mesh.short_description()}")
self._id = info.id
if run_async:
return volume_mesh
log.debug("Waiting for volume mesh to be processed.")
volume_mesh._webapi.get_info()
return VolumeMeshV2.from_cloud(volume_mesh.id)
[docs]
class VolumeMeshV2(AssetBase):
"""
Volume mesh component for workbench (simulation V2)
"""
_interface_class = VolumeMeshInterfaceV2
_meta_class = VolumeMeshMetaV2
_draft_class = VolumeMeshDraftV2
_web_api_class = Flow360Resource
_mesh_stats_file = "meshStats.json"
_cloud_resource_type_name = "VolumeMesh"
[docs]
@classmethod
# pylint: disable=redefined-builtin
def from_cloud(cls, id: str, **kwargs) -> VolumeMeshV2:
"""
Parameters
----------
id : str
ID of the volume mesh resource in the cloud
Returns
-------
VolumeMeshV2
Volume mesh object
"""
asset_obj = super().from_cloud(id, **kwargs)
return asset_obj
[docs]
@classmethod
def from_local_storage(
cls, mesh_id: str = None, local_storage_path="", meta_data: VolumeMeshMetaV2 = None
) -> VolumeMeshV2:
"""
Parameters
----------
mesh_id : str
ID of the volume mesh resource
local_storage_path:
The folder of the project, defaults to current working directory
Returns
-------
VolumeMeshV2
Volume mesh object
"""
return super()._from_local_storage(
asset_id=mesh_id, local_storage_path=local_storage_path, meta_data=meta_data
)
[docs]
@classmethod
# pylint: disable=too-many-arguments,arguments-renamed
def from_file(
cls,
file_name: str,
project_name: str = None,
solver_version: str = None,
length_unit: LengthUnitType = "m",
tags: List[str] = None,
folder: Optional[Folder] = None,
) -> VolumeMeshDraftV2:
"""
Parameters
----------
file_name : str
The name of the input volume mesh file (``*.cgns``, ``*.ugrid``)
project_name : str, optional
The name of the newly created project, defaults to file name if empty
solver_version: str
Solver version to use for the project
length_unit: LengthUnitType
Length unit to use for the project ("m", "mm", "cm", "inch", "ft")
tags: List[str]
List of string tags to be added to the project upon creation
folder : Optional[Folder], optional
Parent folder for the project. If None, creates in root.
Returns
-------
VolumeMeshDraftV2
Draft of the volume mesh to be submitted
"""
# For type hint only but proper fix is to fully abstract the Draft class too.
return super().from_file(
file_names=file_name,
project_name=project_name,
solver_version=solver_version,
length_unit=length_unit,
tags=tags,
folder=folder,
)
# pylint: disable=useless-parent-delegation
[docs]
def get_dynamic_default_settings(self, simulation_dict: dict):
"""Get the default volume mesh settings from the simulation dict"""
return super().get_dynamic_default_settings(simulation_dict)
@cached_property
def stats(self) -> VolumeMeshStats:
"""
Get mesh stats
Returns
-------
VolumeMeshStats
return VolumeMeshStats object
"""
# pylint: disable=protected-access
data = self._webapi._parse_json_from_cloud(self._mesh_stats_file)
return VolumeMeshStats(**data)
@cached_property
def bounding_box(self) -> VolumeMeshBoundingBox:
"""
Get mesh bounding box
Returns
-------
VolumeMeshBoundingBox
return VolumeMeshBoundingBox object
"""
# pylint: disable=protected-access
data = self._webapi._parse_json_from_cloud(VolumeMeshDownloadable.BOUNDING_BOX.value)
bbox = VolumeMeshBoundingBox.from_dict(data)
return bbox
@property
def boundary_names(self) -> List[str]:
"""
Retrieve all boundary names available in this volume mesh as a list
Returns
-------
List[str]
List of boundary names contained within the volume mesh
"""
self.internal_registry = self._entity_info.get_persistent_entity_registry(
internal_registry=self.internal_registry
)
# pylint: disable=protected-access
return [surface.name for surface in self.internal_registry.view(Surface)._entities]
@property
def zone_names(self) -> List[str]:
"""
Retrieve all volume zone names available in this volume mesh as a list
Returns
-------
List[str]
List of zone names contained within the volume mesh
"""
self.internal_registry = self._entity_info.get_persistent_entity_registry(
internal_registry=self.internal_registry
)
# pylint: disable=protected-access
return [volume.name for volume in self.internal_registry.view(GenericVolume)._entities]
def __getitem__(self, key: str):
"""
Parameters
----------
key : str
The name of the entity to be found
Returns
-------
Surface
The boundary object
"""
# 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.volumes[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.")
self.internal_registry = self._entity_info.get_persistent_entity_registry(
internal_registry=self.internal_registry
)
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.")
