"""
Case component
"""
# pylint: disable=too-many-lines
from __future__ import annotations
import json
import os
import tempfile
from typing import TYPE_CHECKING, Any, Iterator, List, Optional, Union
import pydantic as pd
import pydantic.v1 as pd_v1
from flow360_schema.models.simulation.simulation_params import SimulationParams
from .. import error_messages
from ..cloud.flow360_requests import (
MoveCaseItem,
MoveToFolderRequest,
RenameAssetRequestV2,
)
from ..cloud.rest_api import RestApi
from ..cloud.utils import is_valid_uuid
from ..environment import current_environment
from ..exceptions import Flow360RuntimeError, Flow360ValidationError, Flow360ValueError
from ..log import log
from .folder import Folder
from .interfaces import (
CaseInterface,
CaseInterfaceV2,
FolderInterface,
VolumeMeshInterface,
)
from .resource_base import (
AssetMetaBaseModel,
AssetMetaBaseModelV2,
Flow360Resource,
Flow360ResourceListBase,
Flow360Status,
ResourceDraft,
before_submit_only,
is_object_cloud_resource,
)
from .results.base_results import ResultTarGZModel
from .results.case_results import (
ActuatorDiskResultCSVModel,
AeroacousticsResultCSVModel,
BETForcesRadialDistributionResultCSVModel,
BETForcesResultCSVModel,
CaseDownloadable,
CFLResultCSVModel,
CustomForceResultModel,
ForceDistributionsResultModel,
LegacyForceDistributionResultCSVModel,
LinearResidualsResultCSVModel,
MaxResidualLocationResultCSVModel,
MinMaxStateResultCSVModel,
MonitorsResultModel,
NonlinearResidualsResultCSVModel,
PorousMediumResultCSVModel,
RendersResultModel,
ResultBaseModel,
ResultsDownloaderSettings,
SurfaceForcesResultCSVModel,
SurfaceHeatTransferResultCSVModel,
TotalForcesResultCSVModel,
UserDefinedDynamicsResultModel,
XSlicingForceDistributionResultCSVModel,
YSlicingForceDistributionResultCSVModel,
)
from .simulation import services
from .simulation.web.asset_webapi import CaseWebApi
from .utils import (
_local_download_file_list_overwrite,
_local_download_overwrite,
shared_account_confirm_proceed,
validate_type,
)
from .v1.flow360_params import Flow360Params, UnvalidatedFlow360Params
from .validator import Validator
if TYPE_CHECKING:
from flow360.component.volume_mesh import VolumeMeshV2
class CaseBase:
"""
Case Base component
"""
def copy(
self,
name: str = None,
params: Flow360Params = None,
solver_version: str = None,
tags: List[str] = None,
) -> CaseDraft:
"""
Alias for retry case
:param name:
:param params:
:param tags:
:return:
"""
return self.retry(name, params, solver_version=solver_version, tags=tags)
# pylint: disable=no-member
def retry(
self,
name: str = None,
params: Flow360Params = None,
solver_version: str = None,
tags: List[str] = None,
) -> CaseDraft:
"""
Retry case
:param name:
:param params:
:param tags:
:return:
"""
name = name or self.name or self.info.name
params = params or self.params.copy(deep=True)
new_case = Case.create(
name, params, other_case=self, solver_version=solver_version, tags=tags
)
return new_case
def continuation(
self, name: str = None, params: Flow360Params = None, tags: List[str] = None
) -> CaseDraft:
"""
Alias for fork a case to continue simulation
:param name:
:param params:
:param tags:
:return:
"""
return self.fork(name, params, tags)
# pylint: disable=no-member
def fork(
self, name: str = None, params: Flow360Params = None, tags: List[str] = None
) -> CaseDraft:
"""
Fork a case to continue simulation
:param name:
:param params:
:param tags:
:return:
"""
name = name or self.name or self.info.name
params = params or self.params.copy(deep=True)
return Case.create(name, params, parent_case=self, tags=tags)
class CaseMeta(AssetMetaBaseModel):
"""
CaseMeta data component
"""
id: str = pd_v1.Field(alias="caseId")
case_mesh_id: str = pd_v1.Field(alias="caseMeshId")
parent_id: Union[str, None] = pd_v1.Field(alias="parentId")
status: Flow360Status = pd_v1.Field()
# Resource status change, revisit when updating the case class
@pd_v1.validator("status")
@classmethod
def set_status_type(cls, value: Flow360Status):
"""set_status_type when case uploaded"""
if value is Flow360Status.UPLOADED:
return Flow360Status.CASE_UPLOADED
return value
def to_case(self) -> Case:
"""
returns Case object from case meta info
"""
return Case(self.id)
class CaseMetaV2(AssetMetaBaseModelV2):
"""
CaseMetaV2 component
"""
id: str = pd.Field(alias="caseId")
status: Flow360Status = pd.Field()
def to_case(self) -> Case:
"""
returns Case object from case meta info
"""
return Case(self.id)
class ComputeUsage(pd.BaseModel):
"""
Compute recorded for a case.
``type`` is ``"FLEX_CREDIT"`` or ``"VGPU_ALLOCATION"``. ``value`` is in ``unit``:
``"FLEX_CREDIT"`` for FlexCredit cases, and the contracted GPU equivalent
(``"H200_EQUIVALENT"`` or ``"A100_EQUIVALENT"``) for Virtual GPU cases, which is not
the physical GPU model. Fractions are kept, and a recorded zero is a real zero.
"""
model_config = pd.ConfigDict(frozen=True)
type: str
value: float
unit: str
class ResourceUsage(pd.BaseModel):
"""
Usage recorded for a resource.
``compute`` is None when the cloud has no compute usage on record, for example for a case
that has not finished, or for a Virtual GPU case submitted before usage reporting existed.
"""
model_config = pd.ConfigDict(frozen=True)
compute: Optional[ComputeUsage]
# pylint: disable=too-many-instance-attributes
class CaseDraft(CaseBase, ResourceDraft):
"""
Case Draft component (before submission)
"""
# pylint: disable=too-many-arguments
def __init__(
self,
name: str,
params: Flow360Params,
volume_mesh_id: str = None,
tags: List[str] = None,
parent_id: str = None,
other_case: Case = None,
parent_case: Case = None,
solver_version: str = None,
):
self.name = name
self.params = params
self.volume_mesh_id = volume_mesh_id
self.parent_case = parent_case
self.parent_id = parent_id
self.other_case = other_case
self.tags = tags
self.solver_version = solver_version
self._id = None
self._submitted_case = None
ResourceDraft.__init__(self)
self.validate_case_inputs()
def __str__(self):
return self.params.__str__()
@property
def params(self) -> Flow360Params:
"""
returns case params
"""
return self._params
@params.setter
def params(self, value: Flow360Params):
"""
sets case params (before submit only)
"""
if not isinstance(value, Flow360Params) and not isinstance(value, UnvalidatedFlow360Params):
raise Flow360ValueError("params are not of type Flow360Params.")
self._params = value
@property
def name(self) -> str:
"""
returns case name
"""
return self._name
@name.setter
def name(self, value) -> str:
"""
sets case name
"""
self._name = value
@property
def volume_mesh_id(self):
"""
returns volume mesh id
"""
return self._volume_mesh_id
@volume_mesh_id.setter
def volume_mesh_id(self, value):
"""
sets volume mesh id
"""
self._volume_mesh_id = value
def to_case(self) -> Case:
"""Return Case from CaseDraft (must be after .submit())
Returns
-------
Case
Case representation
Raises
------
RuntimeError
Raises error when case is before submission, i.e., is in draft state
"""
if not self.is_cloud_resource():
raise Flow360RuntimeError(
f"Case name={self.name} is in draft state. Run .submit() before calling this function."
)
return Case(self.id)
@before_submit_only
def submit(self, force_submit: bool = False) -> Case:
"""
Submits case to cloud for running.
"""
assert self.name
assert self.volume_mesh_id or self.other_case or self.parent_id or self.parent_case
assert self.params
self.validate_case_inputs(pre_submit_checks=True)
if not shared_account_confirm_proceed():
raise Flow360ValueError("User aborted resource submit.")
volume_mesh_id = self.volume_mesh_id
parent_id = self.parent_id
if parent_id is not None:
self.parent_case = Case(self.parent_id)
if isinstance(self.parent_case, CaseDraft):
self.parent_case = self.parent_case.to_case()
if isinstance(self.other_case, CaseDraft):
self.other_case = self.other_case.to_case()
if self.other_case is not None and self.other_case.has_parent():
self.parent_case = self.other_case.parent
if self.parent_case is not None:
parent_id = self.parent_case.id
volume_mesh_id = self.parent_case.volume_mesh_id
if (
self.solver_version is not None
and self.parent_case.solver_version != self.solver_version
):
raise Flow360RuntimeError(
error_messages.change_solver_version_error(
self.parent_case.solver_version, self.solver_version
)
)
self.solver_version = self.parent_case.solver_version
volume_mesh_id = volume_mesh_id or self.other_case.volume_mesh_id
if self.solver_version is None:
volume_mesh_info = AssetMetaBaseModel(
**RestApi(
VolumeMeshInterface.endpoint,
id=volume_mesh_id,
environment_provider=current_environment,
).get()
)
self.solver_version = volume_mesh_info.solver_version
is_valid_uuid(volume_mesh_id)
self.validator_api(
self.params,
volume_mesh_id=volume_mesh_id,
solver_version=self.solver_version,
raise_on_error=(not force_submit),
)
data = {
"name": self.name,
"meshId": volume_mesh_id,
"runtimeParams": self.params.json(),
"tags": self.tags,
"parentId": parent_id,
}
if self.solver_version is not None:
data["solverVersion"] = self.solver_version
resp = RestApi(CaseInterface.endpoint, environment_provider=current_environment).post(
json=data,
path=f"volumemeshes/{volume_mesh_id}/case",
)
info = CaseMeta(**resp)
# setting _id will disable "WARNING: You have not submitted..." warning message
self._id = info.id
self._submitted_case = Case(self.id)
kwargs = {}
if self._submitted_case.project_id is not None:
kwargs = {"project_id": self._submitted_case.project_id}
log.info(f"Case successfully submitted: {self._submitted_case.short_description(**kwargs)}")
return self._submitted_case
def validate_case_inputs(self, pre_submit_checks=False):
"""
validates case inputs (before submit only)
"""
if self.volume_mesh_id is not None and self.other_case is not None:
raise Flow360ValueError("You cannot specify both volume_mesh_id AND other_case.")
if self.parent_id is not None and self.parent_case is not None:
raise Flow360ValueError("You cannot specify both parent_id AND parent_case.")
if self.parent_id is not None or self.parent_case is not None:
if self.volume_mesh_id is not None or self.other_case is not None:
raise Flow360ValueError(
"You cannot specify volume_mesh_id OR other_case when parent case provided."
)
is_valid_uuid(self.volume_mesh_id, allow_none=True)
if pre_submit_checks:
is_object_cloud_resource(self.other_case)
is_object_cloud_resource(self.parent_case)
@classmethod
def validator_api(
cls,
params: Flow360Params,
volume_mesh_id,
solver_version: str = None,
raise_on_error: bool = True,
):
"""
validation api: validates case parameters before submitting
"""
return Validator.CASE.validate(
params,
mesh_id=volume_mesh_id,
solver_version=solver_version,
raise_on_error=raise_on_error,
)
# pylint: disable=too-many-instance-attributes,too-many-public-methods
[docs]
class Case(CaseBase, Flow360Resource):
"""
Case component
"""
_manifest_path = "visualize/manifest/manifest.json"
_cloud_resource_type_name = "Case"
_web_api_v2_class = Flow360Resource
# pylint: disable=redefined-builtin
def __init__(self, id: str):
super().__init__(
interface=CaseInterface,
meta_class=CaseMeta,
id=id,
)
self._params = None
self._raw_params = None
self._results = CaseResultsModel(case=self)
self._manifest = None
# _web_api_v2 handles all V2 communications for Case
self._web_api_v2 = self._web_api_v2_class(
interface=CaseInterfaceV2,
meta_class=CaseMetaV2,
id=id,
)
@classmethod
def _from_meta(cls, meta: CaseMeta):
validate_type(meta, "meta", CaseMeta)
case = cls(id=meta.id)
case._set_meta(meta)
return case
[docs]
def get_simulation_params(self):
"""
returns simulation params
"""
# pylint: disable=import-outside-toplevel
from botocore.exceptions import ClientError as CloudFileNotFoundError
try:
params_as_dict = self._parse_json_from_cloud("simulation.json")
except CloudFileNotFoundError as err:
raise Flow360ValueError(
"Simulation params not found for this case. It is likely it was created with old interface."
f" Original error: {str(err)}"
) from err
# if the params come from GUI, it can contain data that is not conformal with SimulationParams thus cleaning
param, errors, _ = services.validate_model(
params_as_dict=params_as_dict,
validated_by=services.ValidationCalledBy.LOCAL,
root_item_type=None,
validation_level=None,
)
if errors is not None:
raise Flow360ValidationError(
f"Error found in simulation params. The param may be created by an incompatible version. {errors}",
)
return param
@property
def params(self) -> Union[Flow360Params, SimulationParams]:
"""
returns case params
"""
if self._params is None:
try:
self._params = self.get_simulation_params()
return self._params
except Flow360ValueError:
pass
self._raw_params = json.loads(self.get(method="runtimeParams")["content"])
try:
with tempfile.NamedTemporaryFile(
mode="w", suffix=".json", delete=False
) as temp_file:
json.dump(self._raw_params, temp_file)
self._params = Flow360Params(temp_file.name)
except pd_v1.ValidationError as err:
raise Flow360ValidationError(error_messages.params_fetching_error(err)) from err
return self._params
@property
def params_as_dict(self) -> dict:
"""
returns case params as dictionary
"""
if self._raw_params is None:
self._raw_params = json.loads(self.get(method="runtimeParams")["content"])
return self._raw_params
[docs]
def has_parent(self) -> bool:
"""Check if case has parent case
Returns
-------
bool
True when case has parent, False otherwise
"""
return self.info.parent_id is not None
@property
def parent(self) -> Case:
"""parent case
Returns
-------
Case
parent case object
Raises
------
RuntimeError
When case does not have parent
"""
if self.has_parent():
return Case(self.info.parent_id)
raise Flow360RuntimeError("Case does not have parent case.")
@property
def info(self) -> CaseMetaV2:
"""
returns metadata info for case
"""
return super().info
@property
def info_v2(self) -> CaseMetaV2:
"""
returns metadata v2 info for case
"""
return self._web_api_v2.info
@property
def project_id(self) -> Optional[str]:
"""Returns the project id of the case if case was run with V2 interface."""
if isinstance(self.info, CaseMeta):
try:
return self.info.projectId
except AttributeError:
# No project id at all in V1 resources
return None
if isinstance(self.info, CaseMetaV2):
return self.info.project_id
raise ValueError("Case info is not of type CaseMeta or CaseMetaV2")
@property
def tags(self) -> List[str]:
"""
get case tags
"""
return self._web_api_v2.info.tags
@property
def volume_mesh(self) -> VolumeMeshV2:
"""
returns volume mesh
"""
from_cache = self.local_resource_cache[self.volume_mesh_id]
if from_cache is not None:
return from_cache
# pylint: disable=import-outside-toplevel,cyclic-import
from .volume_mesh import VolumeMeshV2
return VolumeMeshV2(self.volume_mesh_id)
@property
def volume_mesh_id(self):
"""
returns volume mesh id
"""
return self.info.case_mesh_id
@property
def results(self) -> CaseResultsModel:
"""
returns results object to managing case results
"""
return self._results
@property
def usage(self) -> ResourceUsage:
"""
Usage recorded for the case, fetched from the cloud on every access.
``usage.compute`` stays None until the case finishes, so call :meth:`wait` first.
Read it once (``usage = case.usage``) rather than once per field.
"""
return ResourceUsage.model_validate(CaseWebApi(self.id).get_usage()["usage"])
# pylint: disable=no-member
[docs]
def is_steady(self):
"""
returns True when case is steady state
"""
if isinstance(self.params, Flow360Params):
return self.params.time_stepping.time_step_size == "inf"
return self.params.is_steady()
[docs]
def has_actuator_disks(self):
"""
returns True when case has actuator disk
"""
if isinstance(self.params, Flow360Params):
return self.params.actuator_disks is not None and len(self.params.actuator_disks) > 0
return self.params.has_actuator_disks()
[docs]
def has_bet_disks(self):
"""
returns True when case has BET disk
"""
if isinstance(self.params, Flow360Params):
return self.params.bet_disks is not None and len(self.params.bet_disks) > 0
return self.params.has_bet_disks()
[docs]
def has_isosurfaces(self):
"""
returns True when case has isosurfaces
"""
if isinstance(self.params, Flow360Params):
return self.params.iso_surface_output is not None
return self.params.has_isosurfaces()
[docs]
def has_monitors(self):
"""
returns True when case has monitors
"""
if isinstance(self.params, Flow360Params):
return self.params.monitor_output is not None
return self.params.has_monitors()
[docs]
def has_volume_output(self):
"""
returns True when case has volume output
"""
if isinstance(self.params, Flow360Params):
return self.params.volume_output is not None
return self.params.has_volume_output()
[docs]
def has_aeroacoustics(self):
"""
returns True when case has aeroacoustics
"""
if isinstance(self.params, Flow360Params):
return self.params.aeroacoustic_output is not None
return self.params.has_aeroacoustics()
[docs]
def has_user_defined_dynamics(self):
"""
returns True when case has user defined dynamics
"""
if isinstance(self.params, Flow360Params):
return self.params.user_defined_dynamics is not None
return self.params.has_user_defined_dynamics()
[docs]
def has_custom_forces(self):
"""
returns True when case has custom forces
"""
if isinstance(self.params, Flow360Params):
return False
return self.params.has_custom_forces()
[docs]
def has_force_distributions(self):
"""
returns True when case has force distributions
"""
if isinstance(self.params, Flow360Params):
return False
return self.params.has_force_distributions()
[docs]
def has_renders(self):
"""
returns True when case has render outputs
"""
if isinstance(self.params, Flow360Params):
return False
return any(
getattr(output, "output_type", None) == "RenderOutput"
for output in (getattr(self.params, "outputs", None) or [])
)
[docs]
def move_to_folder(self, folder: Folder):
"""
Move the current case to the specified folder.
Parameters
----------
folder : Folder
The destination folder where the item will be moved.
Returns
-------
self
Returns the modified item after it has been moved to the new folder.
Notes
-----
This method sends a REST API request to move the current item to the specified folder.
The `folder` parameter should be an instance of the `Folder` class with a valid ID.
"""
RestApi(FolderInterface.endpoint, environment_provider=current_environment).put(
MoveToFolderRequest(dest_folder_id=folder.id, items=[MoveCaseItem(id=self.id)]).dict(),
method="move",
)
return self
[docs]
def rename(self, new_name: str):
"""
Rename the current case.
Parameters
----------
new_name : str
The new name for the case.
"""
RestApi(CaseInterfaceV2.endpoint, environment_provider=current_environment).patch(
RenameAssetRequestV2(name=new_name).dict(), method=self.id
)
[docs]
def delete(self): # pylint: disable=arguments-differ
"""
Delete the current case.
Goes through the v2 endpoint. The v1 one runs the S3 cleanup inline before it
stops the pipeline job, so a failed cleanup leaves a deleted case still running.
"""
self._web_api_v2.delete()
[docs]
def stop(self):
"""
Stop the running case, the same way Stop does in the web UI.
The solver stops at an upcoming step, its results are uploaded as usual, and the
case finishes as ``completed``. This returns once the request is sent; call
:meth:`wait` to block until the case finishes. As in the web UI, stopping a case
re-creates its draft in the project.
Raises
------
Flow360RuntimeError
If the case is not running: it has not started yet, or it has already finished.
"""
CaseWebApi(self.id).stop()
@classmethod
def _interface(cls):
return CaseInterface
@classmethod
def _meta_class(cls):
"""
returns case meta info class: CaseMeta
"""
return CaseMeta
@classmethod
def _params_ancestor_id_name(cls):
"""
returns volumeMeshId name
"""
return "meshId"
[docs]
@classmethod
def from_cloud(cls, case_id: str):
"""
get case from cloud
"""
return cls(case_id)
[docs]
@classmethod
def from_local_storage(cls, local_storage_path, meta_data: CaseMeta) -> Case:
"""
Create a `Case` instance from local storage.
Parameters
----------
local_storage_path : str
The path to the local storage directory.
meta_data : CaseMeta
Case metadata containing:
- id (str): The unique identifier for the case.
- name (str): The name of the case.
- user_id (str): The user ID associated with the case, can be "local".
Returns
-------
Case
An instance of `Case` with data loaded from local storage.
"""
_local_download_file = _local_download_overwrite(local_storage_path, cls.__name__)
_local_download_file_list = _local_download_file_list_overwrite(local_storage_path)
case = cls._from_meta(meta_data)
case._download_file = _local_download_file
case.get_download_file_list = _local_download_file_list
case._results = CaseResultsModel(case=case)
case.results.set_local_storage(local_storage_path, keep_remote_structure=True)
return case
# pylint: disable=too-many-arguments
[docs]
@classmethod
def create(
cls,
name: str,
params: Flow360Params,
volume_mesh_id: str = None,
tags: List[str] = None,
parent_id: str = None,
other_case: Case = None,
parent_case: Case = None,
solver_version: str = None,
) -> CaseDraft:
"""
Create new case
:param name:
:param params:
:param volume_mesh_id:
:param other_case:
:param tags:
:param parent_id:
:param parent_case:
:return:
"""
assert name
assert volume_mesh_id or other_case or parent_id or parent_case
assert params
if not isinstance(params, Flow360Params) and not isinstance(
params, UnvalidatedFlow360Params
):
raise Flow360ValueError("params are not of type Flow360Params.")
new_case = CaseDraft(
name=name,
volume_mesh_id=volume_mesh_id,
params=params.copy(),
parent_id=parent_id,
other_case=other_case,
parent_case=parent_case,
tags=tags,
solver_version=solver_version,
)
return new_case
# pylint: disable=unnecessary-lambda
[docs]
class CaseResultsModel(pd.BaseModel):
"""
Pydantic models for case results
"""
case: Any = pd.Field()
# tar.gz results:
surfaces: ResultTarGZModel = pd.Field(
default_factory=lambda: ResultTarGZModel(remote_file_name=CaseDownloadable.SURFACES.value)
)
volumes: ResultTarGZModel = pd.Field(
default_factory=lambda: ResultTarGZModel(remote_file_name=CaseDownloadable.VOLUMES.value)
)
slices: ResultTarGZModel = pd.Field(
default_factory=lambda: ResultTarGZModel(remote_file_name=CaseDownloadable.SLICES.value)
)
isosurfaces: ResultTarGZModel = pd.Field(
default_factory=lambda: ResultTarGZModel(
remote_file_name=CaseDownloadable.ISOSURFACES.value
)
)
# Rendered outputs (visualize/renders/): MP4 per render in mode="video",
# or per-frame PNGs in mode="frames". Downloaded by render-output name.
renders: RendersResultModel = pd.Field(default_factory=lambda: RendersResultModel())
monitors: MonitorsResultModel = pd.Field(MonitorsResultModel())
# convergence:
nonlinear_residuals: NonlinearResidualsResultCSVModel = pd.Field(
default_factory=lambda: NonlinearResidualsResultCSVModel()
)
linear_residuals: LinearResidualsResultCSVModel = pd.Field(
default_factory=lambda: LinearResidualsResultCSVModel()
)
cfl: CFLResultCSVModel = pd.Field(default_factory=lambda: CFLResultCSVModel())
minmax_state: MinMaxStateResultCSVModel = pd.Field(
default_factory=lambda: MinMaxStateResultCSVModel()
)
max_residual_location: MaxResidualLocationResultCSVModel = pd.Field(
default_factory=lambda: MaxResidualLocationResultCSVModel()
)
# forces
total_forces: TotalForcesResultCSVModel = pd.Field(
default_factory=lambda: TotalForcesResultCSVModel()
)
surface_forces: SurfaceForcesResultCSVModel = pd.Field(
default_factory=lambda: SurfaceForcesResultCSVModel()
)
actuator_disks: ActuatorDiskResultCSVModel = pd.Field(
default_factory=lambda: ActuatorDiskResultCSVModel()
)
bet_forces: BETForcesResultCSVModel = pd.Field(
default_factory=lambda: BETForcesResultCSVModel()
)
porous_media: PorousMediumResultCSVModel = pd.Field(
default_factory=lambda: PorousMediumResultCSVModel()
)
bet_forces_radial_distribution: BETForcesRadialDistributionResultCSVModel = pd.Field(
default_factory=lambda: BETForcesRadialDistributionResultCSVModel()
)
legacy_force_distribution: LegacyForceDistributionResultCSVModel = pd.Field(
default_factory=lambda: LegacyForceDistributionResultCSVModel()
)
x_slicing_force_distribution: XSlicingForceDistributionResultCSVModel = pd.Field(
default_factory=lambda: XSlicingForceDistributionResultCSVModel()
)
y_slicing_force_distribution: YSlicingForceDistributionResultCSVModel = pd.Field(
default_factory=lambda: YSlicingForceDistributionResultCSVModel()
)
# user defined:
user_defined_dynamics: UserDefinedDynamicsResultModel = pd.Field(
default_factory=lambda: UserDefinedDynamicsResultModel()
)
custom_forces: CustomForceResultModel = pd.Field(
default_factory=lambda: CustomForceResultModel()
)
force_distributions: ForceDistributionsResultModel = pd.Field(
default_factory=lambda: ForceDistributionsResultModel()
)
# others
surface_heat_transfer: SurfaceHeatTransferResultCSVModel = pd.Field(
default_factory=lambda: SurfaceHeatTransferResultCSVModel()
)
aeroacoustics: AeroacousticsResultCSVModel = pd.Field(
default_factory=lambda: AeroacousticsResultCSVModel()
)
local_storage: Optional[str] = None
_downloader_settings: ResultsDownloaderSettings = pd.PrivateAttr(ResultsDownloaderSettings())
[docs]
@pd.model_validator(mode="after")
def pass_download_function(self):
"""
Pass download methods into fields of the case results.
"""
if self.case is None:
raise ValueError("case (type Case) is required")
if not isinstance(self.case, Case):
raise TypeError("case must be of type Case")
for field_name in self.__class__.model_fields: # pylint:disable = not-an-iterable
value = getattr(self, field_name)
if isinstance(value, ResultBaseModel):
# pylint: disable=protected-access,no-member
value._download_method = self.case._download_file
# pylint: disable=protected-access,no-member
value._get_params_method = lambda: self.case.params
value.local_storage = self.local_storage
if hasattr(value, "get_download_file_list_method"):
value.get_download_file_list_method = self.case.get_download_file_list
return self
# pylint: disable=protected-access,no-member
[docs]
@pd.model_validator(mode="after")
def pass_has_functions(self):
"""
Pass check to see if result is downloadable based on params
"""
has_function_map = {
"actuator_disks": self.case.has_actuator_disks,
"bet_forces": self.case.has_bet_disks,
"bet_forces_radial_distribution": self.case.has_bet_disks,
"isosurfaces": self.case.has_isosurfaces,
"monitors": self.case.has_monitors,
"volumes": self.case.has_volume_output,
"aeroacoustics": self.case.has_aeroacoustics,
"user_defined_dynamics": self.case.has_user_defined_dynamics,
"custom_forces": self.case.has_custom_forces,
"force_distributions": self.case.has_force_distributions,
"renders": self.case.has_renders,
}
for field_name in self.__class__.model_fields: # pylint:disable = not-an-iterable
value = getattr(self, field_name)
if isinstance(value, ResultBaseModel):
function = has_function_map.get(field_name, lambda: True)
value._is_downloadable = function # pylint: disable=protected-access
return self
def _execute_downloading(self):
"""
Download all specified and available results for the case
"""
for _, value in self.__dict__.items():
if isinstance(value, ResultBaseModel):
# we download if explicitly set set_downloader(<result_name>=True),
# or all=True but only when is not result=False
# pylint: disable=no-member
try_download = value.do_download is True
if self._downloader_settings.all is True and value.do_download is not False:
try_download = value._is_downloadable() is True
if try_download is True:
value.download(
to_folder=self._downloader_settings.destination,
overwrite=self._downloader_settings.overwrite,
)
[docs]
def set_destination(
self, folder_name: str = None, use_case_name: bool = None, use_case_id: bool = None
):
"""
Set the destination for downloading files.
Parameters
----------
folder_name : str, optional
The name of the folder where files will be downloaded.
use_case_name : bool, optional
Whether to use the use case name for the destination.
use_case_id : bool, optional
Whether to use the use case ID for the destination.
Raises
------
ValueError
If more than one argument is provided or if no arguments are provided.
"""
# Check if only one argument is provided
if sum(arg is not None for arg in [folder_name, use_case_name, use_case_id]) != 1:
raise ValueError("Exactly one argument should be provided.")
# pylint: disable=assigning-non-slot
if folder_name is not None:
self._downloader_settings.destination = folder_name
# pylint: disable=no-member
if use_case_name is True:
self._downloader_settings.destination = self.case.name
if use_case_id is True:
self._downloader_settings.destination = self.case.id
# pylint: disable=too-many-arguments, too-many-locals, redefined-builtin
[docs]
def download(
self,
surface: bool = None,
volume: bool = None,
slices: bool = None,
isosurfaces: bool = None,
monitors: bool = None,
nonlinear_residuals: bool = None,
linear_residuals: bool = None,
cfl: bool = None,
minmax_state: bool = None,
max_residual_location: bool = None,
surface_forces: bool = None,
total_forces: bool = None,
bet_forces: bool = None,
bet_forces_radial_distribution: bool = None,
actuator_disks: bool = None,
legacy_force_distribution: bool = None,
x_slicing_force_distribution: bool = None,
y_slicing_force_distribution: bool = None,
user_defined_dynamics: bool = None,
custom_forces: bool = None,
force_distributions: bool = None,
aeroacoustics: bool = None,
surface_heat_transfer: bool = None,
all: bool = None,
overwrite: bool = False,
destination: str = None,
):
"""
Download result files associated with the case.
Parameters
----------
surface : bool, optional
Download surface result file if True.
volume : bool, optional
Download volume result file if True.
nonlinear_residuals : bool, optional
Download nonlinear residuals file if True.
linear_residuals : bool, optional
Download linear residuals file if True.
cfl : bool, optional
Download CFL file if True.
minmax_state : bool, optional
Download minmax state file if True.
surface_forces : bool, optional
Download surface forces file if True.
total_forces : bool, optional
Download total forces file if True.
bet_forces : bool, optional
Download BET (Blade Element Theory) forces file if True.
bet_forces_radial_distribution : bool, optional
Download BET (Blade Element Theory) forces radial distribution file if True.
actuator_disk_output : bool, optional
Download actuator disk output file if True.
all : bool, optional
Download all result files if True. Ignore file if explicitly set: <result_name>=False
overwrite : bool, optional
If True, overwrite existing files with the same name in the destination.
destination : str, optional
Location to save downloaded files. If None, files will be saved in the current directory under ID folder.
"""
# pylint: disable=assigning-non-slot
self.surfaces.do_download = surface
self.volumes.do_download = volume
self.slices.do_download = slices
self.isosurfaces.do_download = isosurfaces
self.monitors.do_download = monitors
self.nonlinear_residuals.do_download = nonlinear_residuals
self.linear_residuals.do_download = linear_residuals
self.cfl.do_download = cfl
self.minmax_state.do_download = minmax_state
self.max_residual_location.do_download = max_residual_location
self.surface_forces.do_download = surface_forces
self.total_forces.do_download = total_forces
self.bet_forces.do_download = bet_forces
self.bet_forces_radial_distribution.do_download = bet_forces_radial_distribution
self.actuator_disks.do_download = actuator_disks
self.legacy_force_distribution.do_download = legacy_force_distribution
self.x_slicing_force_distribution.do_download = x_slicing_force_distribution
self.y_slicing_force_distribution.do_download = y_slicing_force_distribution
self.user_defined_dynamics.do_download = user_defined_dynamics
self.custom_forces.do_download = custom_forces
self.force_distributions.do_download = force_distributions
self.aeroacoustics.do_download = aeroacoustics
self.surface_heat_transfer.do_download = surface_heat_transfer
self._downloader_settings.all = all
self._downloader_settings.overwrite = overwrite
if destination is not None:
self.set_destination(folder_name=destination)
self._execute_downloading()
[docs]
def download_file_by_name(self, file_name, to_file=None, to_folder=".", overwrite: bool = True):
"""
Download file by name
"""
# pylint: disable=no-member
return self.case._download_file(
file_name=file_name, to_file=to_file, to_folder=to_folder, overwrite=overwrite
)
[docs]
def set_local_storage(self, local_storage: str, keep_remote_structure: bool = False):
"""
Set local storage for fetching data from. Used with Case.from_local_storage(...)
Parameters
----------
local_storage : str
Path to local folder
keep_remote_structure : bool, optional
When true, remote folder structure is assumed to be preserved, otherwise flat structure, by default False
"""
for field_name in self.__class__.model_fields: # pylint:disable = not-an-iterable
value = getattr(self, field_name)
if isinstance(value, ResultBaseModel):
if keep_remote_structure is True:
# pylint: disable=protected-access,no-member
value.local_storage = os.path.dirname(
os.path.join(local_storage, value._remote_path())
)
else:
value.local_storage = local_storage
class CaseList(Flow360ResourceListBase):
"""
Case List component
"""
def __init__(
self, mesh_id: str = None, from_cloud: bool = True, include_deleted: bool = False, limit=100
):
super().__init__(
ancestor_id=mesh_id,
from_cloud=from_cloud,
include_deleted=include_deleted,
limit=limit,
resourceClass=Case,
)
def filter(self):
"""
flitering list, not implemented yet
"""
raise NotImplementedError("Filters are not implemented yet")
# resp = list(filter(lambda i: i['caseStatus'] != 'deleted', resp))
# pylint: disable=useless-parent-delegation
def __getitem__(self, index) -> Case:
"""
returns CaseMeta info item of the list
"""
return super().__getitem__(index)
def __iter__(self) -> Iterator[Case]:
return super().__iter__()
