"""Project interface for setting up and running simulations"""
# pylint: disable=no-member, too-many-lines
# To be honest I do not know why pylint is insistent on treating
# ProjectMeta instances as FieldInfo, I'd rather not have this line
from __future__ import annotations
from enum import Enum
from typing import Dict, Iterable, List, Literal, Optional, Union
import pydantic as pd
import typing_extensions
from flow360_schema.framework.physical_dimensions import Length
from flow360_schema.geometry.selection_rules import FaceGroupingRules
from flow360_schema.models.asset_cache import CoordinateSystemStatus, MirrorStatus
from flow360_schema.models.entities.surface_entities import ImportedSurface
from flow360_schema.models.entity_info import (
GeometryEntityInfo,
merge_geometry_entity_info,
prefix_model_ids_with_geometry_id,
)
from flow360_schema.models.simulation.meshing_param.params import ModularMeshingWorkflow
from flow360_schema.models.simulation.simulation_params import SimulationParams
from flow360_schema.models.simulation.units import validate_length
from pydantic import PositiveInt
from flow360 import error_messages
from flow360.cloud.flow360_requests import CloneVolumeMeshRequest, LengthUnitType
from flow360.cloud.rest_api import RestApi
from flow360.component.case import Case
from flow360.component.cloud_examples import (
copy_example,
fetch_examples,
find_example_by_name,
)
from flow360.component.geometry import Geometry, GeometryWorkflow
from flow360.component.geometry_face_grouping import apply_face_grouping_rules
from flow360.component.interfaces import (
GeometryInterface,
ProjectInterface,
SurfaceMeshInterfaceV2,
VolumeMeshInterfaceV2,
)
from flow360.component.project_utils import (
apply_and_inform_grouping_selections,
enforce_length_scale_sanity,
ensure_cad_importer_compatible_with_mesher,
load_status_from_asset,
read_root_cad_importer_version,
set_up_params_for_uploading,
)
from flow360.component.resource_base import Flow360Resource
from flow360.component.simulation.draft_context.context import (
DraftContext,
get_active_draft,
)
from flow360.component.simulation.folder import Folder, resolve_parent_folder
from flow360.component.simulation.web.asset_base import AssetBase
from flow360.component.simulation.web.draft import Draft
from flow360.component.simulation.web.project_records import (
get_project_records,
show_projects_with_keyword_filter,
)
from flow360.component.simulation.web.project_tree import (
build_project_tree,
get_project_tree_parent_id,
)
from flow360.component.simulation.web.utils import (
get_project_dependency_resource_metadata,
)
from flow360.component.surface_mesh_v2 import SurfaceMeshV2
from flow360.component.utils import (
AssetShortID,
GeometryFiles,
SurfaceMeshFile,
VolumeMeshFile,
formatting_validation_errors,
formatting_validation_warnings,
get_short_asset_id,
parse_datetime,
wrapstring,
)
from flow360.component.volume_mesh import VolumeMeshMetaV2, VolumeMeshV2
from flow360.component.workspace import Workspace
from flow360.environment import current_environment
from flow360.exceptions import (
Flow360ConfigError,
Flow360Error,
Flow360FileError,
Flow360RuntimeError,
Flow360ValueError,
Flow360WebError,
Flow360WebTimeoutError,
)
from flow360.log import log
from flow360.solver_version import Flow360Version, warn_if_minor_release
from flow360.version import __solver_version__
AssetOrResource = Union[type[AssetBase], type[Flow360Resource]]
def _ensure_geometry_compatibility_with_mesher(
project: Project,
use_geometry_AI: bool, # pylint: disable=invalid-name
params: SimulationParams,
):
# Discrete geometry (STL/UGRID/CGNS) or multiple B-Rep files cannot be
# surface-meshed without Geometry AI. The ModularMeshingWorkflow (snappy)
# path consumes discrete geometry natively and is exempt. When a draft is
# active, its entity_info already merges in any import_geometry() files.
active_draft = get_active_draft()
# pylint: disable=protected-access
entity_info = (
active_draft._entity_info if active_draft is not None else project._root_asset._entity_info
)
# An active draft started from a SurfaceMesh exposes a SurfaceMeshEntityInfo
# (no body_attribute_names) — there's no geometry surface-meshing step to
# gate in that case, so the check is a no-op.
if (
use_geometry_AI is False
and project.metadata.root_item_type is RootType.GEOMETRY
and not isinstance(params.meshing, ModularMeshingWorkflow)
and isinstance(entity_info, GeometryEntityInfo)
and "groupByFile" in entity_info.body_attribute_names
):
geometry_files, discrete_files = (
entity_info._get_processed_file_list() # pylint: disable=protected-access
)
if discrete_files:
raise Flow360ValueError(
f"Discrete geometry models detected ({', '.join(discrete_files)}). "
"Geometry AI must be activated to generate the surface mesh "
"from discrete geometry models."
)
if len(geometry_files) > 1:
names = [n.removesuffix(".egads") for n in geometry_files]
raise Flow360ValueError(
f"Multiple geometry models detected ({len(names)} files: {', '.join(names)}). "
"Geometry AI must be activated to generate the surface mesh "
"from multiple geometry models."
)
class RootType(Enum):
"""
Enum for root object types in the project.
Attributes
----------
GEOMETRY : str
Represents a geometry root object.
SURFACE_MESH : str
Represents a surface mesh root object.
VOLUME_MESH : str
Represents a volume mesh root object.
"""
GEOMETRY = "Geometry"
SURFACE_MESH = "SurfaceMesh"
VOLUME_MESH = "VolumeMesh"
class ProjectDependencyType(Enum):
"""
Enum for dependency resource types in the project.
Attributes
----------
GEOMETRY : str
Represents a geometry dependency resource.
SURFACE_MESH : str
Represents a surface mesh dependency resource.
"""
GEOMETRY = "Geometry"
SURFACE_MESH = "SurfaceMesh"
def _reject_rule_grouping_on_multi_geometry(face_grouping, active_geometry_dependencies) -> None:
"""Rules are resolved against the primary geometry's topology tree alone, while a
multi-geometry draft's merged entity_info also carries the other components' faces.
Those faces can never match a rule, so they would stay out of every group while the
rule-based tag is active — refuse rather than group part of the draft."""
if not isinstance(face_grouping, FaceGroupingRules) or not active_geometry_dependencies:
return
raise Flow360ValueError(
"Rule-based face grouping does not support multi-geometry drafts yet "
f"({len(active_geometry_dependencies)} additional geometry component(s) active): "
"rules resolve against the primary geometry's topology tree, so the other "
"components' faces would be left ungrouped. Use a face-grouping tag instead."
)
def _load_face_grouping_rules(new_run_from) -> Optional[FaceGroupingRules]:
"""The rules stored in the source asset's simulation.json asset cache, if any."""
return load_status_from_asset(
asset=new_run_from,
status_class=FaceGroupingRules,
cache_key="face_grouping_rules",
)
def _select_face_grouping(face_grouping, inherited, new_run_from):
"""Decide which face grouping the draft gets, given what the caller asked for and
what the source asset already carries. Returns ``(face_grouping, carried_rules)``:
``face_grouping`` goes through the normal resolve/materialize + tag-selection path,
while ``carried_rules`` is a payload to keep as metadata only — on mesh assets the
resolved grouping is already baked into the entity_info, so re-resolving it would
be both pointless and impossible (no topology tree)."""
if inherited is None:
return face_grouping, None
if isinstance(face_grouping, str):
raise Flow360ValueError(
"The source asset carries rule-based face-grouping rules; activating the "
f"'{face_grouping}' grouping tag would silently discard them. Pass "
"face_grouping=FaceGroupingRules(...) to replace the rules, or omit "
"face_grouping to keep them."
)
if face_grouping is not None:
log.info(
"Replacing the %d inherited face-grouping rule group(s) from %s with the "
"%d group(s) passed to create_draft.",
len(inherited.groups),
new_run_from.__class__.__name__,
len(face_grouping.groups),
)
return face_grouping, None
inherited_names = ", ".join(group.name for group in inherited.groups)
if isinstance(new_run_from, Geometry):
log.info(
"Inheriting %d face-grouping rule group(s) from the source geometry "
"(%s); they are re-resolved against its topology tree.",
len(inherited.groups),
inherited_names,
)
return inherited, None
log.info(
"Carrying over %d face-grouping rule group(s) from the source %s (%s); the "
"resolved grouping already lives in its entity_info.",
len(inherited.groups),
new_run_from.__class__.__name__,
inherited_names,
)
return None, inherited
# pylint: disable-next=too-many-locals,too-many-arguments
[docs]
def create_draft(
*,
new_run_from: Union[Geometry, SurfaceMeshV2, VolumeMeshV2],
face_grouping: Optional[Union[str, FaceGroupingRules]] = None,
edge_grouping: Optional[str] = None,
include_geometries: Optional[List[Geometry]] = None,
exclude_geometries: Optional[List[Geometry]] = None,
imported_surfaces: Optional[List[ImportedSurface]] = None,
) -> DraftContext:
"""
Create a local draft context from a cloud asset for your run.
A draft is an isolated, in-memory snapshot of an asset's entity information. It lets you
inspect and modify entities (surfaces, edges, volumes, body groups, etc.) locally without
mutating the cloud asset.
Draft allows you to:
- Override grouping tags for faces and edges (geometry assets only).
- Include or exclude geometry components (projects with a geometry root asset only).
- Register additional imported surfaces (surface mesh dependencies in the project).
- Access entities in the draft via `DraftContext` properties.
- Manage coordinate systems and mirror actions through `draft.coordinate_systems` and `draft.mirror`.
Parameters
----------
new_run_from : Union[Geometry, SurfaceMeshV2, VolumeMeshV2]
The cloud asset to create the draft from.
face_grouping : Optional[Union[str, FaceGroupingRules]]
Face grouping to activate for geometry assets. A string selects an existing
grouping tag. A :class:`FaceGroupingRules` payload defines RULE-BASED grouping
as draft state: the rules are resolved against the geometry's topology tree
(first match wins by group order; faces no rule matches go to the ``is_remaining``
catch-all, which is required once there are any; overlaps warn loudly; a rule
group left with no faces or a group spanning several model files is refused),
the resulting groups are built in the draft's isolated entity
info (the geometry asset itself stays immutable), and the rules + resolved map
are persisted automatically when the draft is submitted (including
``draft_only=True``). When None, the draft uses the default grouping stored on
the geometry asset.
edge_grouping : Optional[str]
Edge grouping tag to activate for geometry assets. When None, the draft uses the
default grouping stored on the geometry asset. If the activated geometry does not
have edge grouping attributes, this option is ignored.
include_geometries : Optional[List[Geometry]]
Additional geometry components to activate in the draft (projects with a geometry
root asset only). The selected geometries are merged into the draft entity info.
exclude_geometries : Optional[List[Geometry]]
Geometry components to deactivate from the draft (projects with a geometry root asset
only). If a geometry is not currently active, its exclusion is ignored with a
warning.
imported_surfaces : Optional[List[ImportedSurface]]
Imported surface meshes to register in the draft. This is commonly obtained from
`Project.imported_surfaces` or created via `Project.import_surface_mesh(...)`.
Returns
-------
DraftContext
A draft context manager. Use it with `with` to set it as the active draft context.
Raises
------
Flow360RuntimeError
If `new_run_from` is not a cloud asset instance.
Flow360ValueError
If draft creation is attempted from a `Case` or an imported `Geometry`, if geometry
components are requested for a non-geometry-root project, or if an invalid grouping
tag is specified.
Example
-------
>>> import flow360 as fl
>>> geometry = fl.Geometry.from_cloud(id="...")
>>> with fl.create_draft(new_run_from=geometry, face_grouping="groupByName") as draft:
... print(draft.surfaces["wing*"])
"""
# region -----------------------------Private implementations Below-----------------------------
def _resolve_active_geometry_dependencies(
current_geometry_dependencies: List,
include_geometries: List[Geometry],
exclude_geometries: List[Geometry],
) -> Dict[str, Geometry]:
active_geometry_dependencies = {
geometry_dependency["id"]: Geometry.from_cloud(geometry_dependency["id"])
for geometry_dependency in current_geometry_dependencies
}
for geometry in include_geometries:
if geometry.id not in active_geometry_dependencies:
active_geometry_dependencies[geometry.id] = geometry
for geometry in exclude_geometries:
excluded_geometry = active_geometry_dependencies.pop(geometry.id, None)
if excluded_geometry is None:
log.warning(
f"Geometry {geometry.name} not found among current dependencies. Ignoring its exclusion."
)
return active_geometry_dependencies
def _merge_geometry_entity_info(
new_run_from, entity_info, active_geometry_dependencies: Dict[str, Geometry]
):
"""Merge the geometry entity info based on the root and imported geometries."""
if not active_geometry_dependencies:
return entity_info
# Add root geometry to components to be merged
project = Project.from_cloud(new_run_from.info.project_id)
root_geometry = project.geometry
active_geometry_dependencies.update({root_geometry.id: root_geometry})
merged_entity_info = merge_geometry_entity_info(
current_entity_info=entity_info,
entity_info_components=[
prefix_model_ids_with_geometry_id(geometry.entity_info, geometry_id=geometry.id)
for geometry in active_geometry_dependencies.values()
],
)
return merged_entity_info
# endregion ------------------------------------------------------------------------------------
if not isinstance(new_run_from, AssetBase):
raise Flow360RuntimeError("create_draft expects a cloud asset instance as `new_run_from`.")
if isinstance(new_run_from, Geometry) and new_run_from.info.dependency:
raise Flow360ValueError("Draft creation from an imported Geometry is not supported.")
if isinstance(new_run_from, Case):
raise Flow360ValueError("Draft creation from a Case is not supported.")
# The entity_info property returns a fresh deserialization per access; that copy
# is exclusively ours, so ownership transfers to the draft without another deep copy.
entity_info_copy = new_run_from.entity_info
if not isinstance(entity_info_copy, GeometryEntityInfo) and (
include_geometries or exclude_geometries
):
raise Flow360ValueError(
"Only project with a geometry root asset supports editing geometry components."
"Please use create_draft without `include_geometries` and `exclude_geometries`."
)
face_grouping, carried_rules = _select_face_grouping(
face_grouping, _load_face_grouping_rules(new_run_from), new_run_from
)
# Resolve geometry components and merge entity info if applicable
active_geometry_dependencies = {}
if isinstance(entity_info_copy, GeometryEntityInfo):
active_geometry_dependencies = _resolve_active_geometry_dependencies(
current_geometry_dependencies=new_run_from.info.geometry_dependencies or [],
include_geometries=include_geometries or [],
exclude_geometries=exclude_geometries or [],
)
# before the merge, which adds the root geometry to the components
_reject_rule_grouping_on_multi_geometry(face_grouping, active_geometry_dependencies)
entity_info_copy = _merge_geometry_entity_info(
new_run_from=new_run_from,
entity_info=entity_info_copy,
active_geometry_dependencies=active_geometry_dependencies,
)
face_grouping_rules = carried_rules
if isinstance(face_grouping, FaceGroupingRules):
if not isinstance(new_run_from, Geometry):
raise Flow360ValueError(
"Rule-based face grouping (FaceGroupingRules) requires a geometry `new_run_from`."
)
face_grouping_rules = face_grouping
face_grouping = apply_face_grouping_rules(
face_grouping, new_run_from.topology_tree, entity_info_copy
)
if face_grouping_rules is not None:
log.warning(
"!!! Rule-based face grouping is a beta feature, available from the Python "
"client only in this release (the web UI cannot show or edit the rules); "
"it may change in future releases. !!!"
)
apply_and_inform_grouping_selections(
entity_info=entity_info_copy,
face_grouping=face_grouping,
edge_grouping=edge_grouping,
new_run_from_geometry=isinstance(new_run_from, Geometry),
)
mirror_status = load_status_from_asset(
asset=new_run_from,
status_class=MirrorStatus,
cache_key="mirror_status",
)
coordinate_system_status = load_status_from_asset(
asset=new_run_from,
status_class=CoordinateSystemStatus,
cache_key="coordinate_system_status",
)
return DraftContext(
entity_info=entity_info_copy,
mirror_status=mirror_status,
coordinate_system_status=coordinate_system_status,
imported_surfaces=imported_surfaces,
imported_geometries=list(active_geometry_dependencies.values()),
geometry_root=new_run_from if isinstance(new_run_from, Geometry) else None,
face_grouping_rules=face_grouping_rules,
topology_tree_provider=(
(lambda: new_run_from.topology_tree) if isinstance(new_run_from, Geometry) else None
),
)
class ProjectMeta(pd.BaseModel, extra="allow"):
"""
Metadata class for a project.
Attributes
----------
user_id : str
The user ID associated with the project.
id : str
The project ID.
name : str
The name of the project.
description : str
The project description.
tags : List[str]
List of tags associated with the project.
root_item_id : str
ID of the root item in the project.
root_item_type : RootType
Type of the root item (Geometry or SurfaceMesh or VolumeMesh).
"""
user_id: str = pd.Field(alias="userId")
id: str = pd.Field()
name: str = pd.Field()
description: str | None = pd.Field(default=None)
tags: List[str] = pd.Field(default_factory=list)
root_item_id: str = pd.Field(alias="rootItemId")
root_item_type: RootType = pd.Field(alias="rootItemType")
class ProjectTreeNode(pd.BaseModel):
"""
ProjectTreeNode class containing the info of an asset item in a project tree.
Attributes
----------
asset_id : str
ID of the asset.
asset_name : str
Name of the asset.
asset_type : str
Type of the asset.
parent_id : Optional[str]
ID of the parent asset.
case_mesh_id : Optional[str]
ID of the case's mesh.
case_mesh_label : Optional[str]
Label the mesh of a forked case using a different mesh.
children : List
List of the child assets of the current asset.
min_length_short_id : int
The minimum length of the short asset id, excluding
hyphen and asset prefix.
"""
asset_id: str = pd.Field()
asset_name: str = pd.Field()
asset_type: str = pd.Field()
parent_id: Optional[str] = pd.Field(None)
case_mesh_id: Optional[str] = pd.Field(None)
case_mesh_label: Optional[str] = pd.Field(None)
children: List = pd.Field([])
min_length_short_id: PositiveInt = pd.Field(7)
def construct_string(self, line_width):
"""Define the output info within when printing a project tree in the terminal"""
title_line = "<<" + self.asset_type + ">>"
name_line = f"name: {self.asset_name}"
id_line = f"id: {self.short_id}"
# Dynamically compute the line_width for each asset block to ensure
# 1. The asset type title always occupies a single line
# 2. The id and name line width is no more than the input line_width but is as small as possible.
max_line_width = min(line_width, max(len(name_line), len(id_line)))
block_line_width = max(len(title_line), max_line_width)
name_line = wrapstring(long_str=f"name: {self.asset_name}", str_length=block_line_width)
id_line = wrapstring(long_str=f"id: {self.short_id}", str_length=block_line_width)
return f"{title_line.center(block_line_width)}\n{name_line}\n{id_line}"
def add_child(self, child: ProjectTreeNode):
"""Add a child asset of the current asset"""
self.children.append(child)
def remove_child(self, child_to_remove: ProjectTreeNode):
"""Remove a child asset of the current asset"""
self.children = [child for child in self.children if child is not child_to_remove]
@property
def short_id(self) -> str:
"""Compute short asset id"""
return get_short_asset_id(
full_asset_id=self.asset_id, num_character=self.min_length_short_id
)
@property
def edge_label(self) -> str:
"""
Add edge label in the printed project tree to
display the different volume mesh used in a forked case.
"""
if self.case_mesh_label:
prefix = "Using VolumeMesh:\n"
mesh_short_id = get_short_asset_id(
full_asset_id=self.case_mesh_label,
num_character=self.min_length_short_id,
)
return prefix + mesh_short_id.center(len(prefix))
return None
class ProjectTree(pd.BaseModel):
"""
ProjectTree class containing the project tree.
Attributes
----------
root : ProjectTreeNode
Root item of the project.
nodes : dict[str, ProjectTreeNode]
Dict of all nodes in the project tree.
short_id_map: dict[str, List[str]]
Dict of short_id to full_id mapping, used to ensure every short_id is unique in the project.
"""
root: ProjectTreeNode = pd.Field(None)
nodes: dict[str, ProjectTreeNode] = pd.Field({})
short_id_map: dict[str, List[str]] = pd.Field({})
def _update_case_mesh_label(self):
"""Check and remove unnecessary case mesh label"""
for node_id in self._get_asset_ids_by_type(asset_type="Case"):
node = self.nodes.get(node_id)
parent_node = self._get_parent_node(node=node)
if not parent_node:
continue
if parent_node.asset_type != "Case" or node.case_mesh_id == parent_node.case_mesh_id:
node.case_mesh_label = None
def _update_node_short_id(self):
"""Update the minimum length of short ID to ensure each node has a unique short ID"""
if len(self.nodes) == len(self.short_id_map):
pass
full_id_to_update = []
short_id_duplicate = []
for short_id, full_ids in self.short_id_map.items():
if len(full_ids) > 1:
short_id_duplicate.append(short_id)
common_prefix = full_ids[0]
for full_id in full_ids[1:]:
while not full_id.startswith(common_prefix):
common_prefix = common_prefix[:-1]
common_prefix_processed = "".join(common_prefix.split("-")[1:])
for full_id in full_ids:
# pylint: disable=unsubscriptable-object
self.nodes[full_id].min_length_short_id = len(common_prefix_processed) + 1
full_id_to_update.append(full_id)
for full_id in full_id_to_update:
# pylint: disable=unsubscriptable-object
self.short_id_map.update({self.nodes[full_id].short_id: [full_id]})
for short_id in short_id_duplicate:
self.short_id_map.pop(short_id, None)
def _get_parent_node(self, node: ProjectTreeNode):
"""Get the parent node of the input node"""
if not node.parent_id:
return None
return self.nodes.get(node.parent_id, None)
def _has_node(self, asset_id: str) -> bool:
"""Use asset_id to check if the asset already exists in the project tree"""
return asset_id in self.nodes
def _get_asset_ids_by_type(
self, asset_type: str = Literal["Geometry", "SurfaceMesh", "VolumeMesh", "Case"]
):
"""Get the list of asset_ids in the project tree by asset_type."""
return [node.asset_id for node in self.nodes.values() if node.asset_type == asset_type]
@classmethod
def _create_new_node(cls, asset_record: dict):
"""Create a new node based on the asset record from API call"""
parent_id = get_project_tree_parent_id(asset_record)
case_mesh_id = asset_record["parentId"] if asset_record["type"] == "Case" else None
new_node = ProjectTreeNode(
asset_id=asset_record["id"],
asset_name=asset_record["name"],
asset_type=asset_record["type"],
parent_id=parent_id,
case_mesh_id=case_mesh_id,
case_mesh_label=case_mesh_id,
)
return new_node
def _update_short_id_map(self, new_node: ProjectTreeNode):
# pylint: disable=unsupported-assignment-operation,unsubscriptable-object
if new_node.short_id not in self.short_id_map.keys():
self.short_id_map[new_node.short_id] = []
self.short_id_map[new_node.short_id].append(new_node.asset_id)
def add(self, asset_record: dict):
"""Add new node to the existing project tree"""
if self._has_node(asset_id=asset_record["id"]):
return False
new_node = ProjectTree._create_new_node(asset_record)
self._update_short_id_map(new_node)
if new_node.parent_id is None:
self.root = new_node
for node in self.nodes.values():
if node.parent_id == new_node.asset_id:
new_node.add_child(child=node)
if node.asset_id == new_node.parent_id:
node.add_child(child=new_node)
self.nodes.update({new_node.asset_id: new_node})
self._update_node_short_id()
self._update_case_mesh_label()
return True
def remove_node(self, node_id: str):
"""Remove node from the tree"""
node = self.nodes.get(node_id)
if not node:
return
if node.parent_id and self._has_node(node.parent_id):
# pylint: disable=unsubscriptable-object
self.nodes[node.parent_id].remove_child(node)
self.nodes.pop(node.asset_id)
def construct_tree(self, asset_records: List[dict]):
"""Construct the entire project tree"""
def create_node(asset_record):
new_node = ProjectTree._create_new_node(asset_record)
self._update_short_id_map(new_node)
return new_node
def add_child(parent, child):
parent.add_child(child)
self.root, self.nodes = build_project_tree(
asset_records,
create_node=create_node,
add_child=add_child,
)
self._update_node_short_id()
self._update_case_mesh_label()
@pd.validate_call
def get_full_asset_id(self, query_asset: AssetShortID) -> str:
"""
Returns full asset id of a certain asset type given the query_id.
Raises
------
Flow360ValueError
1. If derived asset type from query_id does not match the asset type.
2. If query_id is too short.
3. If query_id is does not exist in the project tree.
Returns
-------
The full asset id.
"""
asset_type_ids = self._get_asset_ids_by_type(asset_type=query_asset.asset_type)
if len(asset_type_ids) == 0:
raise Flow360ValueError(f"No {query_asset.asset_type} is available in this project.")
if query_asset.asset_id is None:
# The latest asset of this asset_type will be returned.
return asset_type_ids[-1]
for asset_id in asset_type_ids:
if asset_id.startswith(query_asset.asset_id):
return asset_id
raise Flow360ValueError(
f"This asset does not exist in this project. Please check the input asset ID ({query_asset.asset_id})."
)
# pylint: disable=too-many-public-methods
class _Unset: # pylint: disable=too-few-public-methods
"""Sentinel for ``update()`` arguments that were not provided (distinct from an explicit ``None``)."""
_UNSET = _Unset()
def _read_only_metadata_error(field: str) -> AttributeError:
"""Build the error raised when assigning to a read-only project metadata property."""
return AttributeError(
f"Project '{field}' is read-only; use project.update({field}=...) to change it."
)
[docs]
class Project(pd.BaseModel):
"""
Project class containing the interface for creating and running simulations.
"""
metadata: ProjectMeta = pd.Field(description="Metadata of the project.")
project_tree: ProjectTree = pd.Field()
solver_version: str = pd.Field(frozen=True, description="Version of the solver being used.")
_root_asset: Union[Geometry, SurfaceMeshV2, VolumeMeshV2] = pd.PrivateAttr(None)
_root_webapi: Optional[RestApi] = pd.PrivateAttr(None)
_project_webapi: Optional[RestApi] = pd.PrivateAttr(None)
_length_unit: Optional[Length.PositiveFloat64] = pd.PrivateAttr(None)
[docs]
@classmethod
def show_remote(cls, search_keyword: Union[None, str] = None):
"""
Shows all projects on the cloud.
Parameters
----------
search_keyword : str, optional
"""
show_projects_with_keyword_filter(search_keyword)
@property
def name(self) -> str:
"""
Returns the name of the project.
Returns
-------
str
The project name.
"""
return self.metadata.name
@name.setter
def name(self, value):
raise _read_only_metadata_error("name")
@property
def id(self) -> str:
"""
Returns the ID of the project.
Returns
-------
str
The project ID.
"""
return self.metadata.id
@property
def tags(self) -> List[str]:
"""
Returns the tags of the project.
Returns
-------
List[str]
List of the project's tags.
"""
return self.metadata.tags
@tags.setter
def tags(self, value):
raise _read_only_metadata_error("tags")
@property
def description(self) -> str | None:
"""
Returns the description of the project.
Returns
-------
str | None
The project description.
"""
return self.metadata.description
@description.setter
def description(self, value):
raise _read_only_metadata_error("description")
@property
def length_unit(self) -> Length.PositiveFloat64:
"""
Returns the length unit of the project.
Returns
-------
Length.PositiveFloat64
The length unit.
"""
if self._length_unit is None:
raise Flow360ValueError("[Internal] Simulation params do not contain length unit info.")
return self._length_unit
@property
def geometry(self) -> Geometry:
"""
Returns the geometry asset of the project. There is always only one geometry asset per project.
Raises
------
Flow360ValueError
If the geometry asset is not available for the project.
Returns
-------
Geometry
The geometry asset.
"""
self._check_initialized()
if self.metadata.root_item_type is not RootType.GEOMETRY:
raise Flow360ValueError(
"Geometry asset is only present in projects initialized from geometry."
)
return self._root_asset
[docs]
def get_surface_mesh(self, asset_id: str = None) -> SurfaceMeshV2:
"""
Returns the surface mesh asset of the project.
Parameters
----------
asset_id : str, optional
The ID of the asset from among the generated assets in this project instance. If not provided,
the property contains the most recently run asset.
Raises
------
Flow360ValueError
If the surface mesh asset is not available for the project.
Returns
-------
SurfaceMeshV2
The surface mesh asset.
"""
self._check_initialized()
asset_id = self.project_tree.get_full_asset_id(
query_asset=AssetShortID(asset_id=asset_id, asset_type="SurfaceMesh")
)
return SurfaceMeshV2.from_cloud(id=asset_id)
@property
def surface_mesh(self) -> SurfaceMeshV2:
"""
Returns the last used surface mesh asset of the project.
If the project is initialized from surface mesh, the surface mesh asset is the root asset.
Raises
------
Flow360ValueError
If the surface mesh asset is not available for the project.
Returns
-------
SurfaceMeshV2
The surface mesh asset.
"""
if self.metadata.root_item_type is RootType.SURFACE_MESH:
return self._root_asset
log.warning(
f"Accessing surface mesh from a project initialized from {self.metadata.root_item_type.name}. "
"Please use the root asset for assigning entities to SimulationParams."
)
return self.get_surface_mesh()
[docs]
def get_volume_mesh(self, asset_id: str = None) -> VolumeMeshV2:
"""
Returns the volume mesh asset of the project.
Parameters
----------
asset_id : str, optional
The ID of the asset from among the generated assets in this project instance. If not provided,
the property contains the most recently run asset.
Raises
------
Flow360ValueError
If the volume mesh asset is not available for the project.
Returns
-------
VolumeMeshV2
The volume mesh asset.
"""
self._check_initialized()
asset_id = self.project_tree.get_full_asset_id(
query_asset=AssetShortID(asset_id=asset_id, asset_type="VolumeMesh")
)
return VolumeMeshV2.from_cloud(id=asset_id)
@property
def volume_mesh(self) -> VolumeMeshV2:
"""
Returns the last used volume mesh asset of the project.
Raises
------
Flow360ValueError
If the volume mesh asset is not available for the project.
Returns
-------
VolumeMeshV2
The volume mesh asset.
"""
if self.metadata.root_item_type is RootType.VOLUME_MESH:
return self._root_asset
log.warning(
f"Accessing volume mesh from a project initialized from {self.metadata.root_item_type.name}. "
"Please use the root asset for assigning entities to SimulationParams."
)
return self.get_volume_mesh()
[docs]
def get_case(self, asset_id: str = None) -> Case:
"""
Returns the last used case asset of the project.
Parameters
----------
asset_id : str, optional
The ID of the asset from among the generated assets in this project instance. If not provided,
the property contains the most recently run asset.
Raises
------
Flow360ValueError
If the case asset is not available for the project.
Returns
-------
Case
The case asset.
"""
self._check_initialized()
asset_id = self.project_tree.get_full_asset_id(
query_asset=AssetShortID(asset_id=asset_id, asset_type="Case")
)
return Case.from_cloud(case_id=asset_id)
@property
def case(self):
"""
Returns the case asset of the project.
Raises
------
Flow360ValueError
If the case asset is not available for the project.
Returns
-------
Case
The case asset.
"""
return self.get_case()
[docs]
def get_surface_mesh_ids(self) -> Iterable[str]:
"""
Returns the available IDs of surface meshes in the project
Returns
-------
Iterable[str]
An iterable of asset IDs.
"""
# pylint: disable=protected-access
return self.project_tree._get_asset_ids_by_type(asset_type="SurfaceMesh")
[docs]
def get_volume_mesh_ids(self):
"""
Returns the available IDs of volume meshes in the project
Returns
-------
Iterable[str]
An iterable of asset IDs.
"""
# pylint: disable=protected-access
return self.project_tree._get_asset_ids_by_type(asset_type="VolumeMesh")
[docs]
def get_case_ids(self, tags: Optional[List[str]] = None) -> List[str]:
"""
Returns the available IDs of cases in the project, optionally filtered by tags.
Parameters
----------
tags : List[str], optional
List of tags to filter cases by. If None or empty tags list, returns all case IDs.
Returns
-------
Iterable[str]
An iterable of case IDs. If tags are provided, filters to return only
case IDs that have at least one matching tag.
"""
# pylint: disable=protected-access
all_case_ids = self.project_tree._get_asset_ids_by_type(asset_type="Case")
if not tags:
return all_case_ids
# Filter cases by tags
filtered_case_ids = []
for case_id in all_case_ids:
case = self.get_case(asset_id=case_id)
if set(tags) & set(case.info_v2.tags):
filtered_case_ids.append(case_id)
return filtered_case_ids
[docs]
@classmethod
def get_project_ids(cls, tags: Optional[List[str]] = None) -> List[str]:
"""
Returns the available IDs of projects, optionally filtered by tags.
Parameters
----------
tags : List[str], optional
List of tags to filter projects by. If None, returns all project IDs.
Returns
-------
List[str]
A list of project IDs. If tags are provided, filters to return only
project IDs that have at least one matching tag.
"""
project_records, _ = get_project_records("", tags=tags)
return [record.project_id for record in project_records.records]
# pylint: disable=too-many-arguments, too-many-locals
@classmethod
def _create_project_from_files(
cls,
*,
files: Union[GeometryFiles, SurfaceMeshFile, VolumeMeshFile],
cad_importer_version: Literal["v1", "v2"],
name: str = None,
solver_version: str = __solver_version__,
length_unit: Optional[LengthUnitType] = None,
tags: List[str] = None,
run_async: bool = False,
folder: Optional[Folder] = None,
workflow: GeometryWorkflow = "standard",
description: str = "",
):
"""
Initializes a project from a file.
Parameters
----------
files : Union[GeometryFiles, SurfaceMeshFile, VolumeMeshFile]
Path to the files.
name : str, optional
Name of the project (default is None).
solver_version : str, optional
Version of the solver (default is None).
length_unit : LengthUnitType, optional
Unit of length (default is "m").
tags : list of str, optional
Tags to assign to the project (default is None).
run_async : bool, optional
Whether to create the project asynchronously (default is False).
folder : Optional[Folder], optional
Parent folder for the project. If None, creates in root.
Returns
-------
Project
An instance of the project. Or Project ID when run_async is True.
Raises
------
Flow360ValueError
If the project cannot be initialized from the file.
"""
root_asset = None
# pylint:disable = protected-access
files._check_files_existence()
if isinstance(files, GeometryFiles):
draft = Geometry.from_file(
files.file_names,
name,
solver_version,
length_unit,
tags,
folder=folder,
workflow=workflow,
cad_importer_version=cad_importer_version,
)
elif isinstance(files, SurfaceMeshFile):
draft = SurfaceMeshV2.from_file(
files.file_names, name, solver_version, length_unit, tags, folder=folder
)
elif isinstance(files, VolumeMeshFile):
draft = VolumeMeshV2.from_file(
files.file_names, name, solver_version, length_unit, tags, folder=folder
)
else:
raise Flow360FileError(
"Cannot detect the intended project root with the given file(s)."
)
root_asset = draft.submit(run_async=run_async, description=description)
if run_async:
log.info(
f"The input file(s) has been successfully uploaded to project: {root_asset.project_id} "
"and is being processed on cloud. Only the project ID string is returned. "
"To retrieve this project later, use 'Project.from_cloud(project_id)'. "
)
return root_asset.project_id
if not root_asset:
raise Flow360ValueError(f"Couldn't initialize asset from {files.file_names}")
project_id = root_asset.project_id
project_api = RestApi(
ProjectInterface.endpoint, id=project_id, environment_provider=current_environment
)
info = project_api.get()
if not info:
raise Flow360ValueError(f"Couldn't retrieve project info for {project_id}")
project = Project(
metadata=ProjectMeta(**info),
project_tree=ProjectTree(),
solver_version=root_asset.solver_version,
)
project._project_webapi = project_api
if isinstance(files, GeometryFiles):
project._root_webapi = RestApi(
GeometryInterface.endpoint,
id=root_asset.id,
environment_provider=current_environment,
)
elif isinstance(files, SurfaceMeshFile):
project._root_webapi = RestApi(
SurfaceMeshInterfaceV2.endpoint,
id=root_asset.id,
environment_provider=current_environment,
)
elif isinstance(files, VolumeMeshFile):
project._root_webapi = RestApi(
VolumeMeshInterfaceV2.endpoint,
id=root_asset.id,
environment_provider=current_environment,
)
project._root_asset = root_asset
project._load_root_length_unit()
project._get_tree_from_cloud()
return project
@classmethod
# pylint: disable=too-many-locals
def _create_project_from_volume_mesh_clone(
cls,
*,
volume_mesh: VolumeMeshV2,
name: str,
solver_version: str,
length_unit: LengthUnitType,
tags: Optional[List[str]],
run_async: bool,
folder: Optional[Folder],
description: str = "",
):
"""
Creates a project by cloning an existing cloud volume mesh.
Parameters
----------
volume_mesh : VolumeMeshV2
The cloud volume mesh to clone.
name : str, optional
Name of the project.
solver_version : str, optional
Version of the solver.
length_unit : LengthUnitType, optional
Unit of length.
tags : list of str, optional
Tags to assign to the project.
run_async : bool
Whether to create the project asynchronously.
folder : Optional[Folder], optional
Parent folder for the project. If None, creates in root.
Returns
-------
Project
An instance of the project. Or Project ID when run_async is True.
"""
req = CloneVolumeMeshRequest(
name=name,
solver_version=solver_version,
tags=tags,
parent_folder_id=folder.id if folder else "ROOT.FLOW360",
length_unit=length_unit,
description=description,
original_volume_mesh_id=volume_mesh.id,
)
resp = RestApi(
VolumeMeshInterfaceV2.endpoint, environment_provider=current_environment
).post(req.dict(), method="clone")
vm_info = VolumeMeshMetaV2(**resp)
if run_async:
log.info(
f"Volume mesh clone submitted to project: {vm_info.project_id}. "
"Only the project ID string is returned. "
"To retrieve this project later, use 'Project.from_cloud(project_id)'. "
)
return vm_info.project_id
return cls.from_cloud(vm_info.project_id)
@classmethod
def _resolve_from_volume_mesh_defaults(
cls,
*,
file: Union[str, VolumeMeshV2],
name: Optional[str],
solver_version: Optional[str],
length_unit: Optional[LengthUnitType],
tags: Optional[List[str]],
):
"""Resolve branch-specific defaults for from_volume_mesh()."""
resolved_name = name
resolved_solver_version = solver_version
resolved_length_unit = length_unit
resolved_tags = tags
default_values = {}
if isinstance(file, VolumeMeshV2):
volume_mesh = file
if resolved_solver_version is None:
resolved_solver_version = volume_mesh.solver_version
default_values["solver_version"] = resolved_solver_version
if resolved_length_unit is None:
source_project = Project.from_cloud(project_id=volume_mesh.info.project_id)
resolved_length_unit = str(source_project.length_unit.units)
default_values["length_unit"] = resolved_length_unit
if resolved_name is None:
resolved_name = volume_mesh.info.name
default_values["name"] = resolved_name
if resolved_tags is None:
resolved_tags = volume_mesh.tags
default_values["tags"] = resolved_tags
if default_values:
defaults_summary = "\n\t".join(
f"{key}={value!r}" for key, value in default_values.items()
)
log.info(
f"The following default values are applied to create the project "
f"from cloning a cloud volume mesh:\n\t{defaults_summary}"
)
else:
if resolved_solver_version is None:
resolved_solver_version = __solver_version__
if resolved_length_unit is None:
resolved_length_unit = "m"
if resolved_tags is None:
resolved_tags = []
return (
resolved_name,
resolved_solver_version,
resolved_length_unit,
resolved_tags,
)
[docs]
@classmethod
@pd.validate_call(
config={
"arbitrary_types_allowed": True
} # Folder (v2: component/simulation/folder.py) does not have validate() defined
)
def from_geometry(
cls,
files: Union[str, list[str]],
/,
name: str = None,
solver_version: str = __solver_version__,
length_unit: Optional[LengthUnitType] = None,
tags: List[str] = None,
run_async: bool = False,
folder_or_workspace: Optional[Union[Folder, Workspace]] = None,
workflow: GeometryWorkflow = "standard",
cad_importer_version: Literal["v1", "v2"] = "v1",
description: str = "",
**kwargs,
):
"""
Initializes a project from local geometry files.
Parameters
----------
files : Union[str, list[str]] (positional argument only)
Geometry file paths.
name : str, optional
Name of the project (default is None).
solver_version : str, optional
Version of the solver (default is None).
length_unit : LengthUnitType, optional
Unit of length (default is "m").
tags : list of str, optional
Tags to assign to the project (default is None).
run_async : bool, optional
Whether to create project asynchronously (default is False).
folder_or_workspace : Optional[Union[Folder, Workspace]], optional
Target location for the project: a parent folder, or a workspace to create
the project in its root folder, e.g. ``fl.Workspace.get_shared()`` for the
shared company workspace. If None, creates in the private workspace root.
The legacy ``folder=`` keyword is still accepted as an alias.
workflow : {"standard", "catalyst"}, optional
Workflow used for project geometry preparation. Use `"catalyst"`
for geometry preparation recommended for GAI and snappy workflows
(default is `"standard"`).
cad_importer_version: Literal["v1", "v2"]
CAD Importer version frozen at geometry upload.
'v1' (default) is compatible with all surface meshers;
'v2' is an alternative BRep importer compatible with the default
surface mesher and GeometryAI, but not the standalone beta
in-house mesher (beta without GeometryAI).
description : str, optional
Project description (default is empty string).
Returns
-------
Project
An instance of the project. Or Project ID when run_async is True.
Raises
------
Flow360FileError
If the project cannot be initialized from the file.
Example
-------
>>> my_project = fl.Project.from_geometry(
... "/path/to/my/geometry/my_geometry.csm",
... name="My_Project_name",
... solver_version="release-Major.Minor"
... length_unit="cm"
... tags=["Quarter 1", "Revision 2"]
... )
"""
folder = resolve_parent_folder(folder_or_workspace, kwargs, "folder")
try:
validated_files = GeometryFiles(file_names=files)
except pd.ValidationError as err:
# pylint:disable = raise-missing-from
raise Flow360FileError(f"Geometry file error: {str(err)}")
return cls._create_project_from_files(
files=validated_files,
name=name,
solver_version=solver_version,
length_unit=length_unit,
tags=tags,
run_async=run_async,
folder=folder,
workflow=workflow,
cad_importer_version=cad_importer_version,
description=description,
)
[docs]
@classmethod
@pd.validate_call(config={"arbitrary_types_allowed": True})
def from_surface_mesh(
cls,
file: str,
/,
name: str = None,
solver_version: str = __solver_version__,
length_unit: Optional[LengthUnitType] = None,
tags: List[str] = None,
run_async: bool = False,
folder_or_workspace: Optional[Union[Folder, Workspace]] = None,
description: str = "",
**kwargs,
):
"""
Initializes a project from a local surface mesh file.
Parameters
----------
file : str (positional argument only)
Surface mesh file path. For UGRID file the mapbc
file needs to be renamed with the same prefix under same folder.
name : str, optional
Name of the project (default is None).
solver_version : str, optional
Version of the solver (default is None).
length_unit : LengthUnitType, optional
Unit of length (default is "m").
tags : list of str, optional
Tags to assign to the project (default is None).
run_async : bool, optional
Whether to create project asynchronously (default is False).
folder_or_workspace : Optional[Union[Folder, Workspace]], optional
Target location for the project: a parent folder, or a workspace to create
the project in its root folder, e.g. ``fl.Workspace.get_shared()`` for the
shared company workspace. If None, creates in the private workspace root.
The legacy ``folder=`` keyword is still accepted as an alias.
description : str, optional
Project description (default is empty string).
Returns
-------
Project
An instance of the project. Or Project ID when run_async is True.
Raises
------
Flow360FileError
If the project cannot be initialized from the file.
Example
-------
>>> my_project = fl.Project.from_surface_mesh(
... "/path/to/my/mesh/my_mesh.ugrid",
... name="My_Project_name",
... solver_version="release-Major.Minor"
... length_unit="inch"
... tags=["Quarter 1", "Revision 2"]
... )
"""
folder = resolve_parent_folder(folder_or_workspace, kwargs, "folder")
try:
validated_files = SurfaceMeshFile(file_names=file)
except pd.ValidationError as err:
# pylint:disable = raise-missing-from
raise Flow360FileError(f"Surface mesh file error: {str(err)}")
return cls._create_project_from_files(
files=validated_files,
name=name,
solver_version=solver_version,
length_unit=length_unit,
tags=tags,
run_async=run_async,
folder=folder,
cad_importer_version=None,
description=description,
)
[docs]
@classmethod
@pd.validate_call(config={"arbitrary_types_allowed": True})
# pylint: disable=too-many-locals
def from_volume_mesh(
cls,
file: Union[str, VolumeMeshV2],
/,
name: str = None,
solver_version: Optional[str] = None,
length_unit: Optional[LengthUnitType] = None,
tags: Optional[List[str]] = None,
run_async: bool = False,
folder_or_workspace: Optional[Union[Folder, Workspace]] = None,
description: str = "",
**kwargs,
):
"""
Initializes a project from a local volume mesh file or an existing cloud volume mesh.
Parameters
----------
file : Union[str, VolumeMeshV2] (positional argument only)
Volume mesh file path or a VolumeMeshV2 cloud object.
For file path: UGRID files need the mapbc file renamed with the same prefix
under the same folder.
For VolumeMeshV2: creates a new project by cloning the existing volume mesh.
The solver_version and length_unit default to the values from the original
volume mesh's project.
name : str, optional
Name of the project (default is None for file input,
or the original project's solver version for VolumeMeshV2 input).
solver_version : str, optional
Version of the solver (default is the current solver version for file input,
or the original volume mesh's solver version for VolumeMeshV2 input).
length_unit : LengthUnitType, optional
Unit of length (default is "m" for file input, or the original project's
length unit for VolumeMeshV2 input).
tags : list of str, optional
Tags to assign to the project (default is None for file input,
or the original volume mesh's tags for VolumeMeshV2 input).
run_async : bool
Whether to create project asynchronously (default is False).
folder_or_workspace : Optional[Union[Folder, Workspace]], optional
Target location for the project: a parent folder, or a workspace to create
the project in its root folder, e.g. ``fl.Workspace.get_shared()`` for the
shared company workspace. If None, creates in the private workspace root.
The legacy ``folder=`` keyword is still accepted as an alias.
description : str, optional
Project description (default is empty string).
Returns
-------
Project
An instance of the project. Or Project ID when run_async is True.
Raises
------
Flow360FileError
If the project cannot be initialized from the file.
Example
-------
>>> my_project = fl.Project.from_volume_mesh(
... "/path/to/my/mesh/my_mesh.cgns",
... name="My_Project_name",
... solver_version="release-Major.Minor"
... length_unit="inch"
... tags=["Quarter 1", "Revision 2"]
... )
>>> volume_mesh = fl.VolumeMeshV2.from_cloud("vm-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx")
>>> my_project = fl.Project.from_volume_mesh(volume_mesh, name="Cloned_Project")
"""
folder = resolve_parent_folder(folder_or_workspace, kwargs, "folder")
(
resolved_name,
resolved_solver_version,
resolved_length_unit,
resolved_tags,
) = cls._resolve_from_volume_mesh_defaults(
file=file,
name=name,
solver_version=solver_version,
length_unit=length_unit,
tags=tags,
)
if isinstance(file, VolumeMeshV2):
# the file branch is covered when the asset draft is created
warn_if_minor_release(solver_version)
return cls._create_project_from_volume_mesh_clone(
volume_mesh=file,
name=resolved_name,
solver_version=resolved_solver_version,
length_unit=resolved_length_unit,
tags=resolved_tags,
run_async=run_async,
folder=folder,
description=description,
)
try:
validated_files = VolumeMeshFile(file_names=file)
except pd.ValidationError as err:
# pylint:disable = raise-missing-from
raise Flow360FileError(f"Volume mesh file error: {str(err)}")
return cls._create_project_from_files(
files=validated_files,
name=resolved_name,
solver_version=resolved_solver_version,
length_unit=resolved_length_unit,
tags=resolved_tags,
run_async=run_async,
folder=folder,
cad_importer_version=None,
description=description,
)
[docs]
@classmethod
@typing_extensions.deprecated(
"Creating a project with `from_file` is deprecated and will be replaced by a new API "
"in a future release.",
category=None,
)
@pd.validate_call
def from_file(
cls,
file: Union[str, list[str]],
name: str = None,
solver_version: str = __solver_version__,
length_unit: LengthUnitType = "m",
tags: List[str] = None,
run_async: bool = False,
):
"""
[Deprecated function]
Initializes a project from a file.
Parameters
----------
file : str
Path to the file.
name : str, optional
Name of the project (default is None).
solver_version : str, optional
Version of the solver (default is None).
length_unit : LengthUnitType, optional
Unit of length (default is "m").
tags : list of str, optional
Tags to assign to the project (default is None).
run_async : bool, optional
Whether to create project asynchronously (default is False).
Returns
-------
Project
An instance of the project. Or Project ID when run_async is True.
Raises
------
Flow360ValueError
If the project cannot be initialized from the file.
"""
log.warning(
"DeprecationWarning: Creating a project with `from_file` is deprecated "
"and will be replaced by a new API in a future release."
)
def _detect_input_file_type(file: Union[str, list[str]]):
errors = []
for model in [GeometryFiles, VolumeMeshFile]:
try:
return model(file_names=file)
except pd.ValidationError as e:
errors.append(e)
raise Flow360FileError(f"Input file {file} cannot be recognized.\nErrors: {errors}")
return cls._create_project_from_files(
files=_detect_input_file_type(file=file),
name=name,
solver_version=solver_version,
length_unit=length_unit,
tags=tags,
run_async=run_async,
cad_importer_version="v1",
)
def _check_conflicts_with_existing_dependency_resources(
self, name: str, resource_type: ProjectDependencyType
):
resp = self._project_webapi.post(method="dependency/namecheck", json={"name": name})
if resp.get("status") == "success":
return
if (
resource_type == ProjectDependencyType.GEOMETRY
and resp["conflictResourceId"].startswith("geo")
) or (
resource_type == ProjectDependencyType.SURFACE_MESH
and resp["conflictResourceId"].startswith("sm")
):
raise Flow360ValueError(
f"A {resource_type.value} with the name '{name}' already exists in the project. "
"Please use a different name."
)
def _import_dependency_resource_from_file(
self,
*,
files: Union[GeometryFiles, SurfaceMeshFile],
name: str,
length_unit: LengthUnitType = "m",
tags: List[str] = None,
run_async: bool = False,
):
# pylint:disable = protected-access
files._check_files_existence()
if isinstance(files, GeometryFiles):
self._check_conflicts_with_existing_dependency_resources(
name=name, resource_type=ProjectDependencyType.GEOMETRY
)
draft = Geometry.import_to_project(
name=name,
file_names=files.file_names,
project_id=self.id,
length_unit=length_unit,
tags=tags,
# Keep the dependency geometry on the same CAD Importer as the
# project root; a mismatch would make the meshes incompatible.
cad_importer_version=read_root_cad_importer_version(self._root_asset),
)
elif isinstance(files, SurfaceMeshFile):
self._check_conflicts_with_existing_dependency_resources(
name=name, resource_type=ProjectDependencyType.SURFACE_MESH
)
draft = SurfaceMeshV2.import_to_project(
name=name,
file_name=files.file_names,
project_id=self.id,
length_unit=length_unit,
tags=tags,
)
else:
raise Flow360ValueError(f"Unsupported file type: {type(files)}")
dependency_resource = draft.submit(run_async=run_async)
return dependency_resource
[docs]
def import_geometry(
self,
file: Union[str, list[str]],
/,
name: str,
length_unit: LengthUnitType = "m",
tags: List[str] = None,
run_async: bool = False,
):
"""
Imports a geometry dependency resource from local geometry files into the project.
Parameters
----------
file : Union[str, list[str]] (positional argument only)
Geometry file paths.
name : str
Name of the geometry dependency resource.
length_unit : LengthUnitType, optional
Unit of length (default is "m").
tags : list of str, optional
Tags to assign to the geometry dependency resource (default is None).
run_async : bool, optional
Whether to create the geometry dependency resource asynchronously (default is False).
Returns
-------
Geometry
An instance of the geometry resource added to the project.
Raises
------
Flow360FileError
If the geometry dependency resource cannot be initialized from the file.
"""
try:
validated_files = GeometryFiles(file_names=file)
except pd.ValidationError as err:
# pylint:disable = raise-missing-from
raise Flow360FileError(f"Geometry file error: {str(err)}")
return self._import_dependency_resource_from_file(
files=validated_files,
name=name,
length_unit=length_unit,
tags=tags,
run_async=run_async,
)
[docs]
def import_surface_mesh(
self,
file: str,
/,
name: str,
length_unit: LengthUnitType = "m",
tags: List[str] = None,
run_async: bool = False,
) -> ImportedSurface:
"""
Imports a surface mesh dependency resource from a local surface mesh file into the project.
The imported surface is a post-processing entity, also called a sample surface. It is
not meshed and carries no boundary condition; the solution is interpolated onto it and
it can only be assigned to `SurfaceOutput`, `TimeAverageSurfaceOutput` and
`SurfaceIntegralOutput`. Register the returned handle on a draft through the
``imported_surfaces`` argument of `create_draft` before referencing it.
Accepted formats are STL, CGNS and UGRID. A UGRID file requires its companion
``.mapbc`` file next to it, and the file must be uncompressed. Volume mesh files in
UGRID or CGNS format are also accepted, in which case their boundary faces are used.
Every patch in the file is merged into a single surface. The coordinates are consumed
as stored, in the coordinate frame and length unit of the volume mesh, so a file
authored in other units must be converted before import.
Parameters
----------
file : str (positional argument only)
Surface mesh file path.
name : str
Name of the surface mesh dependency resource. This is also the key used to look the
surface up on the draft, and the suffix used in the resulting output filenames.
length_unit : LengthUnitType, optional
Unit of length recorded on the resource (default is "m"). Note that this is not
applied to the surface coordinates when the case runs.
tags : list of str, optional
Tags to assign to the surface mesh dependency resource (default is None).
run_async : bool, optional
Whether to create the surface mesh dependency resource asynchronously (default is False).
Returns
-------
ImportedSurface
An ImportedSurface object with the file name and surface mesh ID.
Raises
------
Flow360FileError
If the surface mesh dependency resource cannot be initialized from the file.
Notes
-----
Nodes of an imported surface that do not fall inside the fluid domain, because they lie
inside a solid body or outside the mesh, are written as ``NaN`` in a surface output and
are excluded from a surface integral. The surface is never checked against the geometry
or trimmed to fit, so an integral over a surface that clips the geometry silently covers
only the part inside the fluid domain.
"""
try:
validated_files = SurfaceMeshFile(file_names=file)
except pd.ValidationError as err:
# pylint:disable = raise-missing-from
raise Flow360FileError(f"Surface mesh file error: {str(err)}")
surface_mesh = self._import_dependency_resource_from_file(
files=validated_files,
name=name,
length_unit=length_unit,
tags=tags,
run_async=run_async,
)
# Deterministic id: ImportedSurface handles are re-created from dependency
# metadata each session, and saved assignments are keyed by this id.
return ImportedSurface(
name=name,
private_attribute_id=f"{name}_defaultBody",
surface_mesh_id=surface_mesh.id,
)
def _get_dependency_resources_from_cloud(self, resource_type: ProjectDependencyType):
"""
Get all imported dependency resources of a given type in the project.
Parameters
----------
resource_type : ProjectDependencyType
The type of dependency resource to retrieve.
"""
dependency_metadata = get_project_dependency_resource_metadata(
project_id=self.id, resource_type=resource_type.value
)
if resource_type == ProjectDependencyType.GEOMETRY:
imported_resources = [
Geometry.from_cloud(item.resource_id) for item in dependency_metadata
]
elif resource_type == ProjectDependencyType.SURFACE_MESH:
imported_resources = [
ImportedSurface(
name=item.name,
private_attribute_id=f"{item.name}_defaultBody",
surface_mesh_id=item.resource_id,
)
for item in dependency_metadata
]
else:
raise Flow360ValueError(f"Unsupported imported resource type: {resource_type}")
return imported_resources
@property
def imported_geometries(self) -> List[Geometry]:
"""
Get all imported geometry components in the project.
Returns
-------
List[Geometry]
A list of Geometry objects representing the imported geometry components.
"""
return self._get_dependency_resources_from_cloud(
resource_type=ProjectDependencyType.GEOMETRY
)
@property
def imported_surfaces(self) -> List[ImportedSurface]:
"""
Get all imported surface components in the project.
Returns
-------
List[ImportedSurface]
A list of ImportedSurface objects representing the imported surface components.
"""
return self._get_dependency_resources_from_cloud(
resource_type=ProjectDependencyType.SURFACE_MESH
)
@classmethod
def _get_user_requested_entity_info(
cls,
*,
current_project_id: str,
new_run_from: Optional[Union[Geometry, SurfaceMeshV2, VolumeMeshV2, Case]] = None,
):
"""
Get the entity info requested by the users when they specify `new_run_from` when calling
Project.from_cloud()
"""
user_requested_entity_info = None
if new_run_from is None:
return user_requested_entity_info
if new_run_from.project_id is None:
# Can only happen to case created using V1 interface.
raise ValueError(
"The supplied case resource for `new_run_from` was created using old interface and "
"cannot be used as the starting point of a new run."
)
if current_project_id != new_run_from.project_id:
raise ValueError(
"The supplied cloud resource for `new_run_from` does not belong to the project."
)
if isinstance(new_run_from, Case):
user_requested_entity_info = new_run_from.get_simulation_params()
if isinstance(new_run_from, (Geometry, SurfaceMeshV2, VolumeMeshV2)):
user_requested_entity_info = new_run_from.params
return user_requested_entity_info
[docs]
@classmethod
@pd.validate_call(
config={"arbitrary_types_allowed": True} # Geometry etc do not have validate() defined
)
def from_cloud(
cls,
project_id: str,
*,
new_run_from: Optional[Union[Geometry, SurfaceMeshV2, VolumeMeshV2, Case]] = None,
):
"""
Loads a project from the cloud.
Parameters
----------
project_id : str
ID of the project.
new_run_from: Optional[Union[Geometry, SurfaceMeshV2, VolumeMeshV2, Case]]
The cloud resource that the current run should be based on.
The root asset will use entity settings (grouping, transformation etc) from this resource.
This results in the same behavior when user clicks New run on webUI.
By default this will be the root asset (what user uploaded) of the project.
TODO: We can add 'last' as one option to automatically start
from the latest created asset within the project.
Returns
-------
Project
An instance of the project.
Raises
------
Flow360WebError
If the project cannot be loaded from the cloud.
Flow360ValueError
If the root asset cannot be retrieved for the project.
"""
project_info = AssetShortID(asset_id=project_id, asset_type="Project")
project_api = RestApi(
ProjectInterface.endpoint,
id=project_info.asset_id,
environment_provider=current_environment,
)
info = project_api.get()
if not isinstance(info, dict):
raise Flow360WebError(
f"Cannot load project {project_info.asset_id}, missing project data."
)
if not info:
raise Flow360WebError(f"Couldn't retrieve project info for {project_info.asset_id}")
meta = ProjectMeta(**info)
root_asset = None
root_type = meta.root_item_type
if (
new_run_from is not None
and isinstance(new_run_from, (Geometry, SurfaceMeshV2, VolumeMeshV2, Case)) is False
):
# Should have been caught by the validate_call?
raise ValueError(
"The supplied `new_run_from` is not valid. Please check the function description for more details."
)
entity_info_param = cls._get_user_requested_entity_info(
current_project_id=project_info.asset_id, new_run_from=new_run_from
)
if root_type == RootType.GEOMETRY:
root_asset = Geometry.from_cloud(meta.root_item_id, entity_info_param=entity_info_param)
elif root_type == RootType.SURFACE_MESH:
root_asset = SurfaceMeshV2.from_cloud(
meta.root_item_id, entity_info_param=entity_info_param
)
elif root_type == RootType.VOLUME_MESH:
root_asset = VolumeMeshV2.from_cloud(
meta.root_item_id, entity_info_param=entity_info_param
)
if not root_asset:
raise Flow360ValueError(f"Couldn't retrieve root asset for {project_info.asset_id}")
project = Project(
metadata=meta, project_tree=ProjectTree(), solver_version=root_asset.solver_version
)
project._project_webapi = project_api
if root_type == RootType.GEOMETRY:
project._root_asset = root_asset
project._root_webapi = RestApi(
GeometryInterface.endpoint,
id=root_asset.id,
environment_provider=current_environment,
)
elif root_type == RootType.SURFACE_MESH:
project._root_asset = root_asset
project._root_webapi = RestApi(
SurfaceMeshInterfaceV2.endpoint,
id=root_asset.id,
environment_provider=current_environment,
)
elif root_type == RootType.VOLUME_MESH:
project._root_asset = root_asset
project._root_webapi = RestApi(
VolumeMeshInterfaceV2.endpoint,
id=root_asset.id,
environment_provider=current_environment,
)
project._load_root_length_unit()
project._get_tree_from_cloud()
return project
[docs]
@classmethod
@pd.validate_call
def from_example(cls, example_id: Optional[str] = None, by_name: Optional[str] = None):
"""
Creates a project from an existing example in the cloud.
Parameters
----------
example_id : str, optional
ID of the example to copy. Mutually exclusive with `by_name`.
by_name : str, optional
Name of the example to copy. Uses fuzzy matching to find the best match.
Mutually exclusive with `example_id`.
Returns
-------
Project
An instance of the project created from the example.
Raises
------
Flow360ValueError
If neither or both `example_id` and `by_name` are provided, or if no matching
example is found when using `by_name`.
Flow360WebError
If the example cannot be copied or the project cannot be loaded.
"""
if example_id is None and by_name is None:
raise Flow360ValueError("Either 'example_id' or 'by_name' must be provided.")
if example_id is not None and by_name is not None:
raise Flow360ValueError("'example_id' and 'by_name' are mutually exclusive.")
if by_name is not None:
examples = fetch_examples()
matched_example, score = find_example_by_name(by_name, examples)
if score < 1.0:
similarity_pct = score * 100
log.info(
f"Found closest match for '{by_name}': '{matched_example.title}' "
f"(similarity: {similarity_pct:.2f} %)"
)
example_id = matched_example.id
project_id = copy_example(example_id)
return cls.from_cloud(project_id)
def _check_initialized(self):
"""
Checks if the project instance has been initialized correctly.
Raises
------
Flow360ValueError
If the project is not initialized.
"""
if not self.metadata or not self.solver_version or not self._root_asset:
raise Flow360ValueError(
"Project not initialized - use Project.from_file or Project.from_cloud"
)
# pylint: disable=protected-access
def _get_tree_from_cloud(self, destination_obj: AssetOrResource = None):
"""
Get the project tree from cloud.
Parameters
----------
destination_obj : AssetOrResource
The destination asset after submitting a job. If provided, only assets along
the path to this asset will be fetched to update the local project tree. Otherwise,
the entire project tree will be refreshed using the latest tree from cloud.
"""
self._check_initialized()
asset_records = []
if destination_obj:
method = "path"
resp = self._project_webapi.get(
method=method,
params={
"itemId": destination_obj.id,
"itemType": destination_obj._cloud_resource_type_name,
},
)
for key, val in resp.items():
if not val:
continue
if key == "cases":
asset_records += resp["cases"]
continue
asset_records.append(val)
else:
method = "tree"
resp = self._project_webapi.get(method=method)
asset_records = resp["records"]
self.project_tree = ProjectTree()
asset_records = sorted(
asset_records,
key=lambda d: parse_datetime(d["updatedAt"]),
)
if method == "tree":
self.project_tree.construct_tree(asset_records=asset_records)
return False
is_duplicate = True
for record in asset_records:
success = self.project_tree.add(asset_record=record)
if success:
is_duplicate = False
return is_duplicate
[docs]
def refresh_project_tree(self):
"""Refresh the local project tree by fetching the latest project tree from cloud."""
return self._get_tree_from_cloud()
def _identify_unacknowledged_run(
self, *, draft: Draft, target: AssetOrResource
) -> Optional[str]:
"""Which resource a run produced, when its acknowledgement never reached the client.
The draft is the evidence that it ran at all: the cloud deletes it only once the
run's resource exists, so a draft that goes away means the run went through. Which
resource is then the newest of the target type — as close as a client can get,
since the cloud identifies a run by a sequence number it never hands out. Returns
None when the run cannot be pinned to a resource, having reported why — the two
ways that happens mean different things to the user, while the submission ends the
same way either way.
"""
if not draft.confirm_run_started():
log.error(
"The cloud never confirmed the submission. It may still be processing, or it "
"may have failed."
)
return None
# pylint: disable=protected-access
asset_type = target._cloud_resource_type_name
try:
records = self._project_webapi.get(method="tree")["records"]
except Flow360Error as error:
# An unreadable project is one more way of not finding the resource, not a
# second kind of failure — and never a raw error thrown at a caller that asked
# for a return value instead.
records = []
log.debug(f"Could not read the project while looking for the {asset_type}: {error}")
newest = max(
(record for record in records if record["type"] == asset_type),
key=lambda record: parse_datetime(record["createdAt"]),
default={},
)
if "id" not in newest:
log.error(
f"The submission went through, but the {asset_type} it created could not be "
"found in this project."
)
return None
log.warning(
f"Returning the newest {asset_type} in this project. The submission did go through, "
"but the cloud never said which resource it created, so check the name and creation "
f"time to confirm this {asset_type} is the one submitted here."
)
return newest["id"]
[docs]
def update(
self,
*,
name: str | _Unset = _UNSET,
description: str | None | _Unset = _UNSET,
tags: list[str] | _Unset = _UNSET,
) -> None:
"""
Update project metadata on the cloud.
Only the arguments you pass are sent; omitted fields are left unchanged.
Provided values (including ``None`` and an empty string) are forwarded
verbatim and validated by the backend. The local metadata is refreshed
afterwards.
Parameters
----------
name : str, optional
New project name.
description : str, optional
New project description.
tags : list of str, optional
New project tags.
"""
provided = {"name": name, "description": description, "tags": tags}
payload = {key: value for key, value in provided.items() if value is not _UNSET}
self._project_webapi.patch(payload)
self.metadata = ProjectMeta(**self._project_webapi.get())
[docs]
def rename(self, new_name: str):
"""
Rename the current project.
Parameters
----------
new_name : str
The new name for the project.
"""
self.update(name=new_name)
[docs]
def delete(self):
"""Delete the current project"""
return self._project_webapi.delete()
[docs]
def print_project_tree(self, line_width: int = 30, is_horizontal: bool = True):
"""Print the project tree to the terminal.
Parameters
----------
line_width : str
The maximum number of characters in each line.
is_horizontal : bool
Choose if the project tree is printed in horizontal (default) or vertical direction.
"""
# pylint: disable=import-outside-toplevel
# Defer importing since this package introduces 2 empty lines of output in the Jupyter Notebook when imported..
from PrettyPrint import PrettyPrintTree
PrettyPrintTree(
get_children=lambda x: x.children,
get_val=lambda x: x.construct_string(line_width=line_width),
get_label=lambda x: x.edge_label if x.edge_label else None,
color="",
border=True,
orientation=PrettyPrintTree.Horizontal if is_horizontal else PrettyPrintTree.Vertical,
)(
self.project_tree.root,
)
def _load_root_length_unit(self):
"""
Reads the root asset's simulation JSON (via the asset's cached fetch) and
extracts the project length unit through the schema migration pipeline.
Raises
------
Flow360ValueError
If the root item type or ID is missing from project metadata, or
the length unit is absent from the migrated simulation params.
Flow360WebError
If the simulation JSON cannot be retrieved.
"""
self._check_initialized()
root_type = self.metadata.root_item_type
root_id = self.metadata.root_item_id
if not root_type or not root_id:
raise Flow360ValueError("Root item type or ID is missing from project metadata")
# pylint: disable-next=protected-access
simulation_json = AssetBase._get_simulation_json(
asset=self._root_asset, clean_front_end_keys=True
)
length_value = simulation_json.get("private_attribute_asset_cache", {}).get(
"project_length_unit"
)
if length_value is None:
raise Flow360ValueError("[Internal] Simulation params do not contain length unit info.")
self._length_unit = validate_length(length_value)
# pylint: disable=too-many-arguments, too-many-locals
def _run(
self,
*,
params: SimulationParams,
target: AssetOrResource,
draft_name: str,
fork_from: Case,
interpolate_to_mesh: VolumeMeshV2,
run_async: bool,
solver_version: str,
use_beta_mesher: bool,
use_geometry_AI: bool,
raise_on_error: bool,
tags: List[str],
draft_only: bool,
job_type: Optional[Literal["TIME_SHARED_VGPU", "FLEX_CREDIT"]] = None,
priority: Optional[int] = None,
**kwargs,
):
"""
Runs a simulation for the project.
Parameters
----------
params : SimulationParams
The simulation parameters to use for the run
target : AssetOrResource
The target asset or resource to run the simulation against.
draft_name : str, optional
The name of the draft to create for the simulation run (default is None).
fork_from : Case, optional
The parent case to fork from if fork (default is None).
interpolate_to_mesh : VolumeMeshV2, optional
If specified, forked case will interpolate parent case's results to this mesh before running solver.
run_async : bool, optional
Specifies whether the simulation should run asynchronously (default is True).
use_beta_mesher : bool, optional
Whether to use the beta mesher (default is None). Must be True when using GeometryAI.
use_geometry_AI : bool, optional
Whether to use the GeometryAI (default is False).
raise_on_error: bool, optional
Option to raise if submission error occurs
tags: List[str], optional
A list of tags to add to the target asset.
draft_only: bool, optional
Whether to only create and submit a draft and not run the simulation.
Validation errors are logged, and the draft is still kept and returned.
job_type: Optional[Literal["TIME_SHARED_VGPU", "FLEX_CREDIT"]]
The billing job type to use for the run request.
Returns
-------
AssetOrResource
The destination asset or the draft if `draft_only` is True.
Raises
------
Flow360ValueError
If the simulation parameters lack required length unit information, or if the
root asset (Geometry or VolumeMesh) is not initialized.
"""
# pylint: disable=too-many-branches,too-many-statements,too-many-return-statements
warn_if_minor_release(solver_version)
if fork_from is not None:
# A fork restarts from the parent case's solution, so it has to stay within the
# parent's release series. Only the patch level may differ.
parent_version = fork_from.solver_version
if not parent_version:
raise Flow360ValueError(f"Solver version of {fork_from.id} is unknown.")
if not solver_version:
solver_version = parent_version
elif Flow360Version(solver_version).series != Flow360Version(parent_version).series:
raise Flow360ValueError(
error_messages.change_solver_version_error(parent_version, solver_version)
)
if use_beta_mesher is None:
if use_geometry_AI is True:
log.info("Beta mesher is enabled to use Geometry AI.")
use_beta_mesher = True
elif self.metadata.root_item_type is RootType.SURFACE_MESH:
log.info("Beta mesher is enabled because project started with surface mesh.")
use_beta_mesher = True
else:
use_beta_mesher = False
if use_geometry_AI is True and use_beta_mesher is False:
raise Flow360ValueError("Enabling Geometry AI requires also enabling beta mesher.")
_ensure_geometry_compatibility_with_mesher(
project=self, use_geometry_AI=use_geometry_AI, params=params
)
root_asset = self._root_asset
if interpolate_to_mesh is not None:
project_vm = Project.from_cloud(project_id=interpolate_to_mesh.project_id)
root_asset = project_vm._root_asset
params = set_up_params_for_uploading(
params=params,
root_asset=root_asset,
length_unit=self.length_unit,
use_beta_mesher=use_beta_mesher,
use_geometry_AI=use_geometry_AI,
fork_from=fork_from,
)
ensure_cad_importer_compatible_with_mesher(params)
if not enforce_length_scale_sanity(params):
if raise_on_error:
raise Flow360ValueError(
"Submission terminated due to a potential length scale mismatch. Acknowledge via "
'`with warning_bypass("potential_length_scale_mismatch"):` to proceed.'
)
return None
active_draft = get_active_draft()
if active_draft is None and params.private_attribute_asset_cache.imported_surfaces:
raise Flow360ValueError(
"ImportedSurface feature requires an active DraftContext. "
"Please use `with project.create_draft(imported_surfaces=[...]):` "
"before calling run(). Or ensure that the submission call is inside the `with` block, not after it."
)
source_item_type = self.metadata.root_item_type.value if fork_from is None else "Case"
start_from = kwargs.get("start_from", None)
job_tags = kwargs.get("job_tags", None)
all_tags = []
if tags is not None:
all_tags += tags
if job_tags is not None:
all_tags += job_tags
draft = Draft.create(
name=draft_name,
project_id=self.metadata.id,
source_item_id=self.metadata.root_item_id if fork_from is None else fork_from.id,
source_item_type=source_item_type,
solver_version=solver_version if solver_version else self.solver_version,
fork_case=fork_from is not None,
interpolation_volume_mesh_id=interpolate_to_mesh.id if interpolate_to_mesh else None,
tags=all_tags,
).submit()
params.pre_submit_summary()
# The uploaded document is the raw params dump, exactly as the WebUI submits;
# validation runs remotely against the draft's own solver version — the
# authoritative oracle — instead of this client's schema. Validator products
# (defaulting/normalizing) are derived by the pipeline's validation at
# translate time and never travel in the document.
#
# A draft that cannot proceed is deleted before /run: no run has been requested
# yet, so it is pure litter, and a user iterating on their params would collect
# one per attempt. A draft_only draft is kept even when validation rejects it,
# because the draft is what the user asked for and its params can be inspected in
# the WebUI. The /run failure path further down keeps its draft too, because a
# timeout there can mean the server accepted the job anyway.
try:
draft.activate_dependencies(active_draft)
draft.update_simulation_params(params)
errors, warnings = draft.validate_simulation_json(
root_item_type=self.metadata.root_item_type.value,
up_to=target._cloud_resource_type_name,
)
except (RuntimeError, Flow360WebError) as error:
draft.delete_draft()
message = f"Submission terminated due to error: {error}"
if raise_on_error:
raise ValueError(message) from None
log.error(message)
return None
if warnings:
log.warning(
f"Validation warnings found: {formatting_validation_warnings(warnings=warnings)}"
)
if draft_only:
if errors is not None:
log.error(f"Validation error found: {formatting_validation_errors(errors=errors)}")
# pylint: disable=import-outside-toplevel
import click
log.info("Draft submitted, copy the link to browser to view the draft:")
# Not using log.info to avoid the link being wrapped and thus not clickable.
click.secho(draft.web_url, fg="blue", underline=True)
return draft
if errors is not None:
draft.delete_draft()
message = (
"Submission terminated due to validation error:"
f"{formatting_validation_errors(errors=errors)}"
)
if raise_on_error:
raise ValueError(message)
log.error(message)
return None
try:
destination_id = draft.run_up_to_target_asset(
target,
source_item_type=source_item_type,
use_beta_mesher=params.private_attribute_asset_cache.use_inhouse_mesher,
use_geometry_AI=use_geometry_AI,
start_from=start_from,
job_type=job_type,
priority=priority,
)
except Flow360WebTimeoutError:
destination_id = self._identify_unacknowledged_run(draft=draft, target=target)
if destination_id is None:
if raise_on_error:
raise ValueError(
"The cloud did not confirm which resource this submission created. "
"Check the project before submitting again."
) from None
return None
except (RuntimeError, Flow360WebError):
if raise_on_error:
raise ValueError("Submission terminated due to error.") from None
return None
self._project_webapi.patch(
# pylint: disable=protected-access
json={
"lastOpenItemId": destination_id,
"lastOpenItemType": target._cloud_resource_type_name,
}
)
destination_obj = target.from_cloud(destination_id)
# Remove when converting Case to V2
kwargs = {}
if isinstance(destination_obj, Case):
kwargs = {"project_id": destination_obj.project_id}
log.info(f"Successfully submitted: {destination_obj.short_description(**kwargs)}")
if not run_async:
destination_obj.wait()
is_duplicate = self._get_tree_from_cloud(destination_obj=destination_obj)
if is_duplicate:
target_asset_type = target._cloud_resource_type_name
log.warning(
f"The {target_asset_type} that matches the input already exists in project. "
f"No new {target_asset_type} will be generated."
)
return destination_obj
[docs]
@pd.validate_call
def generate_surface_mesh(
self,
params: SimulationParams,
name: str = "SurfaceMesh",
run_async: bool = True,
solver_version: str = None,
use_beta_mesher: bool = None,
use_geometry_AI: bool = False, # pylint: disable=invalid-name
raise_on_error: bool = True,
tags: List[str] = None,
draft_only: bool = False,
**kwargs,
):
"""
Runs the surface mesher for the project.
Parameters
----------
params : SimulationParams
Simulation parameters for running the mesher.
name : str, optional
Name of the surface mesh (default is "SurfaceMesh").
run_async : bool, optional
Whether to run the mesher asynchronously (default is True).
solver_version : str, optional
Optional solver version to use during this run (defaults to the project solver version)
use_beta_mesher : bool, optional
Whether to use the beta mesher (default is None). Must be True when using GeometryAI.
use_geometry_AI : bool, optional
Whether to use the GeometryAI (default is False).
raise_on_error: bool, optional
Option to raise if submission error occurs (default is True)
tags: List[str], optional
A list of tags to add to the generated surface mesh.
draft_only: bool, optional
Whether to only create and submit a draft and not generate the surface mesh.
Validation errors are logged, and the draft is still kept and returned.
Raises
------
Flow360ValueError
If the root item type is not Geometry.
Returns
-------
SurfaceMeshV2 | Draft
The surface mesh asset or the draft if `draft_only` is True.
"""
self._check_initialized()
if self.metadata.root_item_type is not RootType.GEOMETRY:
raise Flow360ValueError(
"Surface mesher can only be run by projects with a geometry root asset"
)
surface_mesh = self._run(
params=params,
target=SurfaceMeshV2,
draft_name=name,
run_async=run_async,
fork_from=None,
interpolate_to_mesh=None,
solver_version=solver_version,
use_beta_mesher=use_beta_mesher,
use_geometry_AI=use_geometry_AI,
raise_on_error=raise_on_error,
tags=tags,
draft_only=draft_only,
**kwargs,
)
return surface_mesh
[docs]
@pd.validate_call
def generate_volume_mesh(
self,
params: SimulationParams,
name: str = "VolumeMesh",
run_async: bool = True,
solver_version: str = None,
use_beta_mesher: bool = None,
use_geometry_AI: bool = False, # pylint: disable=invalid-name
raise_on_error: bool = True,
tags: List[str] = None,
draft_only: bool = False,
**kwargs,
):
"""
Runs the volume mesher for the project.
Parameters
----------
params : SimulationParams
Simulation parameters for running the mesher.
name : str, optional
Name of the volume mesh (default is "VolumeMesh").
run_async : bool, optional
Whether to run the mesher asynchronously (default is True).
solver_version : str, optional
Optional solver version to use during this run (defaults to the project solver version)
use_beta_mesher : bool, optional
Whether to use the beta mesher (default is None). Must be True when using GeometryAI.
use_geometry_AI : bool, optional
Whether to use the GeometryAI (default is False).
raise_on_error: bool, optional
Option to raise if submission error occurs (default is True)
tags: List[str], optional
A list of tags to add to the generated volume mesh.
draft_only: bool, optional
Whether to only create and submit a draft and not generate the volume mesh.
Validation errors are logged, and the draft is still kept and returned.
Raises
------
Flow360ValueError
If the root item type is not Geometry.
Returns
-------
VolumeMeshV2 | Draft
The volume mesh asset or the draft if `draft_only` is True.
"""
self._check_initialized()
if (
self.metadata.root_item_type is not RootType.GEOMETRY
and self.metadata.root_item_type is not RootType.SURFACE_MESH
):
raise Flow360ValueError(
"Volume mesher can only be run by projects with a geometry or surface mesh root asset"
)
volume_mesh_or_draft = self._run(
params=params,
target=VolumeMeshV2,
draft_name=name,
run_async=run_async,
fork_from=None,
interpolate_to_mesh=None,
solver_version=solver_version,
use_beta_mesher=use_beta_mesher,
use_geometry_AI=use_geometry_AI,
raise_on_error=raise_on_error,
tags=tags,
draft_only=draft_only,
**kwargs,
)
if draft_only:
draft = volume_mesh_or_draft
return draft
volume_mesh = volume_mesh_or_draft
return volume_mesh
[docs]
@pd.validate_call(config={"arbitrary_types_allowed": True})
def run_case(
self,
params: SimulationParams,
name: str = "Case",
run_async: bool = True,
fork_from: Optional[Case] = None,
interpolate_to_mesh: Optional[VolumeMeshV2] = None,
solver_version: str = None,
use_beta_mesher: bool = None,
use_geometry_AI: bool = False, # pylint: disable=invalid-name
raise_on_error: bool = True,
tags: List[str] = None,
draft_only: bool = False,
billing_method: Optional[Literal["VirtualGPU", "FlexCredit"]] = None,
priority: Optional[int] = None,
**kwargs,
):
"""
Runs a case for the project.
Parameters
----------
params : SimulationParams
Simulation parameters for running the case.
name : str, optional
Name of the case (default is "Case").
run_async : bool, optional
Whether to run the case asynchronously (default is True).
fork_from : Case, optional
Which Case we should fork from (if fork).
interpolate_to_mesh : VolumeMeshV2, optional
If specified, forked case will interpolate parent case results to this mesh before running solver.
solver_version : str, optional
Optional solver version to use during this run (defaults to the project solver version).
When `fork_from` is used the parent case's solver version is inherited, and requesting
a version outside the parent's release series raises a `Flow360ValueError`.
use_beta_mesher : bool, optional
Whether to use the beta mesher (default is None). Must be True when using GeometryAI.
use_geometry_AI : bool, optional
Whether to use the GeometryAI (default is False).
raise_on_error: bool, optional
Option to raise if submission error occurs (default is True)
tags: List[str], optional
A list of tags to add to the case.
draft_only: bool, optional
Whether to only create and submit a draft and not run the case.
Validation errors are logged, and the draft is still kept and returned.
billing_method: Optional[Literal["VirtualGPU", "FlexCredit"]]
Override to default billing method.
priority: Optional[int]
Queue priority for Virtual GPU jobs, from 1 (lowest) to 10 (highest).
Only applicable when ``billing_method="VirtualGPU"``.
Returns
-------
Case | Draft
The case asset or the draft if `draft_only` is True.
"""
if interpolate_to_mesh is not None and fork_from is None:
raise Flow360ConfigError(
"Interpolation to mesh is only supported when forking from a case."
)
# Map user-facing billing_method to API-level job_type
job_type = None
if billing_method is not None and not draft_only:
if billing_method == "FlexCredit":
log.info("The case will be submitted to regular queue and billed with FlexCredits.")
job_type = "FLEX_CREDIT"
elif billing_method == "VirtualGPU":
account_info = RestApi(
"flow360/account", environment_provider=current_environment
).get()
if not account_info.get("timeSharedVGpuEnabled", False):
raise Flow360ValueError(
"Virtual GPU billing is not enabled for this account. "
"Please contact support to enable Virtual GPU access."
)
log.info(
"The case will be submitted to Virtual GPU "
"and billed with Virtual GPU allocation."
)
job_type = "TIME_SHARED_VGPU"
elif draft_only and billing_method is not None:
log.info(
"`billing_method` input to `run_case()` ignored since"
" no billing is necessary when submitting just the draft."
)
# Validate priority: only applicable for VirtualGPU billing
effective_priority = None
if priority is not None:
if billing_method != "VirtualGPU":
log.warning(
"`priority` is only applicable when `billing_method='VirtualGPU'`. Ignoring."
)
elif draft_only:
log.info("`priority` ignored when `draft_only=True`.")
else:
effective_priority = priority
self._check_initialized()
case_or_draft = self._run(
params=params,
target=Case,
draft_name=name,
run_async=run_async,
fork_from=fork_from,
interpolate_to_mesh=interpolate_to_mesh,
solver_version=solver_version,
use_beta_mesher=use_beta_mesher,
use_geometry_AI=use_geometry_AI,
raise_on_error=raise_on_error,
tags=tags,
draft_only=draft_only,
job_type=job_type,
priority=effective_priority,
**kwargs,
)
if draft_only:
draft = case_or_draft
return draft
case = case_or_draft
return case
