"""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
import json
from enum import Enum
from typing import Dict, Iterable, List, Literal, Optional, Union
import pydantic as pd
import typing_extensions
from pydantic import PositiveInt
from flow360.cloud.flow360_requests import (
CloneVolumeMeshRequest,
LengthUnitType,
RenameAssetRequestV2,
)
from flow360.cloud.http_util import http
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
from flow360.component.interfaces import (
GeometryInterface,
ProjectInterface,
SurfaceMeshInterfaceV2,
VolumeMeshInterfaceV2,
)
from flow360.component.project_utils import (
apply_and_inform_grouping_selections,
deep_copy_entity_info,
load_status_from_asset,
set_up_params_for_uploading,
validate_params_with_context,
)
from flow360.component.resource_base import Flow360Resource
from flow360.component.simulation.draft_context.context import (
DraftContext,
get_active_draft,
)
from flow360.component.simulation.draft_context.coordinate_system_manager import (
CoordinateSystemStatus,
)
from flow360.component.simulation.draft_context.mirror import MirrorStatus
from flow360.component.simulation.entity_info import (
GeometryEntityInfo,
merge_geometry_entity_info,
)
from flow360.component.simulation.folder import Folder
from flow360.component.simulation.primitives import ImportedSurface
from flow360.component.simulation.simulation_params import SimulationParams
from flow360.component.simulation.unit_system import LengthType
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.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.exceptions import (
Flow360ConfigError,
Flow360FileError,
Flow360RuntimeError,
Flow360ValueError,
Flow360WebError,
)
from flow360.log import log
from flow360.version import __solver_version__
AssetOrResource = Union[type[AssetBase], type[Flow360Resource]]
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"
# pylint: disable=too-many-arguments
[docs]
def create_draft(
*,
new_run_from: Union[Geometry, SurfaceMeshV2, VolumeMeshV2],
face_grouping: Optional[str] = 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[str]
Face grouping tag to activate for geometry assets. When None, the draft uses the
default grouping stored on the geometry asset. The tag must be one of the available
face grouping attributes (and, when multiple geometry components are active, must be
available across the activated components).
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=[
geometry.entity_info 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.")
if not isinstance(new_run_from.entity_info, 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`."
)
# Deep copy entity_info for draft isolation
entity_info_copy = deep_copy_entity_info(new_run_from.entity_info)
# Resolve geometry components and merge entity info if applicable
active_geometry_dependencies = {}
if isinstance(new_run_from.entity_info, 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 [],
)
entity_info_copy = _merge_geometry_entity_info(
new_run_from=new_run_from,
entity_info=entity_info_copy,
active_geometry_dependencies=active_geometry_dependencies,
)
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()),
)
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.
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()
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"""
if asset_id in self.nodes.keys():
return True
return False
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 = (
asset_record["parentCaseId"]
if asset_record["parentCaseId"]
else asset_record["parentId"]
)
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"""
for asset_record in asset_records:
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
self.nodes.update({new_node.asset_id: new_node})
for node in self.nodes.values():
if node.parent_id and self._has_node(node.parent_id):
# pylint: disable=unsubscriptable-object
self.nodes[node.parent_id].add_child(node)
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
[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)
_root_simulation_json: Optional[dict] = 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 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
@property
def length_unit(self) -> LengthType.Positive:
"""
Returns the length unit of the project.
Returns
-------
LengthType.Positive
The length unit.
"""
defaults = self._root_simulation_json
cache_key = "private_attribute_asset_cache"
length_key = "project_length_unit"
if cache_key not in defaults or length_key not in defaults[cache_key]:
raise Flow360ValueError("[Internal] Simulation params do not contain length unit info.")
return LengthType.validate(defaults[cache_key][length_key])
@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
@classmethod
def _create_project_from_files(
cls,
*,
files: Union[GeometryFiles, SurfaceMeshFile, VolumeMeshFile],
name: str = None,
solver_version: str = __solver_version__,
length_unit: LengthUnitType = "m",
tags: List[str] = None,
run_async: bool = False,
folder: Optional[Folder] = None,
):
"""
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
)
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)
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)
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)
elif isinstance(files, SurfaceMeshFile):
project._root_webapi = RestApi(SurfaceMeshInterfaceV2.endpoint, id=root_asset.id)
elif isinstance(files, VolumeMeshFile):
project._root_webapi = RestApi(VolumeMeshInterfaceV2.endpoint, id=root_asset.id)
project._root_asset = root_asset
project._get_root_simulation_json()
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],
):
"""
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,
original_volume_mesh_id=volume_mesh.id,
)
resp = RestApi(VolumeMeshInterfaceV2.endpoint).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: LengthUnitType = "m",
tags: List[str] = None,
run_async: bool = False,
folder: Optional[Folder] = None,
):
"""
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 : 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
------
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"]
... )
"""
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,
)
[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: LengthUnitType = "m",
tags: List[str] = None,
run_async: bool = False,
folder: Optional[Folder] = None,
):
"""
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 : 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
------
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"]
... )
"""
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,
)
[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: Optional[Folder] = None,
):
"""
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 : 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
------
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")
"""
(
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):
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,
)
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,
)
[docs]
@classmethod
@typing_extensions.deprecated(
"Creating project with `from_file` is deprecated. Please use `from_geometry()`, "
"`from_surface_mesh()` or `from_volume_mesh()` instead.",
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 project with `from_file` is deprecated. "
+ "Please use `from_geometry()`, `from_surface_mesh()` "
+ "or `from_volume_mesh()` instead."
)
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,
)
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,
)
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.
Parameters
----------
file : str (positional argument only)
Surface mesh file path.
name : str
Name of the surface mesh dependency resource.
length_unit : LengthUnitType, optional
Unit of length (default is "m").
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.
"""
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,
)
return ImportedSurface(
name=name,
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,
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)
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)
elif root_type == RootType.SURFACE_MESH:
project._root_asset = root_asset
project._root_webapi = RestApi(SurfaceMeshInterfaceV2.endpoint, id=root_asset.id)
elif root_type == RootType.VOLUME_MESH:
project._root_asset = root_asset
project._root_webapi = RestApi(VolumeMeshInterfaceV2.endpoint, id=root_asset.id)
project._get_root_simulation_json()
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()
[docs]
def rename(self, new_name: str):
"""
Rename the current project.
Parameters
----------
new_name : str
The new name for the project.
"""
RestApi(ProjectInterface.endpoint).patch(
RenameAssetRequestV2(name=new_name).dict(), method=self.id
)
[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 _get_root_simulation_json(self):
"""
Loads the default simulation JSON for the project based on the root asset type.
Raises
------
Flow360ValueError
If the root item type or ID is missing from project metadata.
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")
resp = self._root_webapi.get(method="simulation/file", params={"type": "simulation"})
if not isinstance(resp, dict) or "simulationJson" not in resp:
raise Flow360WebError("Couldn't retrieve default simulation JSON for the project")
simulation_json = json.loads(resp["simulationJson"])
self._root_simulation_json = simulation_json
# 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 Geometry AI.
use_geometry_AI : bool, optional
Whether to use the Geometry AI (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.
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
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
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.")
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,
)
params, errors, warnings = validate_params_with_context(
params=params,
root_item_type=self.metadata.root_item_type.value,
up_to=target._cloud_resource_type_name,
)
if warnings:
log.warning(
f"Validation warnings found during local validation: "
f"{formatting_validation_warnings(warnings=warnings)}"
)
if errors is not None:
log.error(
f"Validation error found during local validation: {formatting_validation_errors(errors=errors)}"
)
if raise_on_error:
raise ValueError("Submission terminated due to local validation error.")
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()
draft.activate_dependencies(active_draft)
draft.update_simulation_params(params)
if draft_only:
# 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
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 RuntimeError:
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 Geometry AI.
use_geometry_AI : bool, optional
Whether to use the Geometry AI (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.
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 Geometry AI.
use_geometry_AI : bool, optional
Whether to use the Geometry AI (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.
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)
use_beta_mesher : bool, optional
Whether to use the beta mesher (default is None). Must be True when using Geometry AI.
use_geometry_AI : bool, optional
Whether to use the Geometry AI (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.
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 = http.get("flow360/account")
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
