"""Case component"""from__future__importannotationsimportjsonimporttempfileimporttimefromtypingimportAny,Iterator,List,Unionimportpydanticaspdfrom..importerror_messagesfrom..cloud.requestsimportMoveCaseItem,MoveToFolderRequestfrom..cloud.rest_apiimportRestApifrom..exceptionsimportFlow360RuntimeError,Flow360ValidationError,Flow360ValueErrorfrom..logimportlogfrom.flow360_params.flow360_paramsimportFlow360Params,UnvalidatedFlow360Paramsfrom.folderimportFolderfrom.interfacesimportCaseInterface,FolderInterface,VolumeMeshInterfacefrom.resource_baseimport(Flow360Resource,Flow360ResourceBaseModel,Flow360ResourceListBase,Flow360Status,ResourceDraft,before_submit_only,is_object_cloud_resource,)from.results.case_resultsimport(ActuatorDiskResultCSVModel,AeroacousticsResultCSVModel,BETForcesResultCSVModel,CaseDownloadable,CFLResultCSVModel,ForceDistributionResultCSVModel,LinearResidualsResultCSVModel,MaxResidualLocationResultCSVModel,MinMaxStateResultCSVModel,MonitorsResultModel,NonlinearResidualsResultCSVModel,ResultBaseModel,ResultsDownloaderSettings,ResultTarGZModel,SurfaceForcesResultCSVModel,SurfaceHeatTrasferResultCSVModel,TotalForcesResultCSVModel,UserDefinedDynamicsResultModel,)from.utilsimportis_valid_uuid,shared_account_confirm_proceed,validate_typefrom.validatorimportValidatorclassCaseBase:""" Case Base component """defcopy(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: """returnself.retry(name,params,solver_version=solver_version,tags=tags)# pylint: disable=no-memberdefretry(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=nameorself.nameorself.info.nameparams=paramsorself.params.copy(deep=True)new_case=Case.create(name,params,other_case=self,solver_version=solver_version,tags=tags)returnnew_casedefcontinuation(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: """returnself.fork(name,params,tags)# pylint: disable=no-memberdeffork(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=nameorself.nameorself.info.nameparams=paramsorself.params.copy(deep=True)returnCase.create(name,params,parent_case=self,tags=tags)classCaseMeta(Flow360ResourceBaseModel):""" CaseMeta data component """id:str=pd.Field(alias="caseId")case_mesh_id:str=pd.Field(alias="caseMeshId")parent_id:Union[str,None]=pd.Field(alias="parentId")status:Flow360Status=pd.Field()# pylint: disable=no-self-argument@pd.validator("status")defset_status_type(cls,value:Flow360Status):"""set_status_type when case uploaded"""ifvalueisFlow360Status.UPLOADED:returnFlow360Status.CASE_UPLOADEDreturnvaluedefto_case(self)->Case:""" returns Case object from case meta info """returnCase(self.id)# pylint: disable=too-many-instance-attributesclassCaseDraft(CaseBase,ResourceDraft):""" Case Draft component (before submission) """# pylint: disable=too-many-argumentsdef__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=nameself.params=paramsself.volume_mesh_id=volume_mesh_idself.parent_case=parent_caseself.parent_id=parent_idself.other_case=other_caseself.tags=tagsself.solver_version=solver_versionself._id=Noneself._submitted_case=NoneResourceDraft.__init__(self)self.validate_case_inputs()def__str__(self):returnself.params.__str__()@propertydefparams(self)->Flow360Params:""" returns case params """returnself._params@params.setterdefparams(self,value:Flow360Params):""" sets case params (before submit only) """ifnotisinstance(value,Flow360Params)andnotisinstance(value,UnvalidatedFlow360Params):raiseFlow360ValueError("params are not of type Flow360Params.")self._params=value@propertydefname(self)->str:""" returns case name """returnself._name@name.setterdefname(self,value)->str:""" sets case name """self._name=value@propertydefvolume_mesh_id(self):""" returns volume mesh id """returnself._volume_mesh_id@volume_mesh_id.setterdefvolume_mesh_id(self,value):""" sets volume mesh id """self._volume_mesh_id=valuedefto_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 """ifnotself.is_cloud_resource():raiseFlow360RuntimeError(f"Case name={self.name} is in draft state. Run .submit() before calling this function.")returnCase(self.id)@before_submit_onlydefsubmit(self,force_submit:bool=False)->Case:""" submits case to cloud for running """assertself.nameassertself.volume_mesh_idorself.other_caseorself.parent_idorself.parent_caseassertself.paramsself.validate_case_inputs(pre_submit_checks=True)ifnotshared_account_confirm_proceed():raiseFlow360ValueError("User aborted resource submit.")volume_mesh_id=self.volume_mesh_idparent_id=self.parent_idifparent_idisnotNone:self.parent_case=Case(self.parent_id)ifisinstance(self.parent_case,CaseDraft):self.parent_case=self.parent_case.to_case()ifisinstance(self.other_case,CaseDraft):self.other_case=self.other_case.to_case()ifself.other_caseisnotNoneandself.other_case.has_parent():self.parent_case=self.other_case.parentifself.parent_caseisnotNone:parent_id=self.parent_case.idvolume_mesh_id=self.parent_case.volume_mesh_idif(self.solver_versionisnotNoneandself.parent_case.solver_version!=self.solver_version):raiseFlow360RuntimeError(error_messages.change_solver_version_error(self.parent_case.solver_version,self.solver_version))self.solver_version=self.parent_case.solver_versionvolume_mesh_id=volume_mesh_idorself.other_case.volume_mesh_idifself.solver_versionisNone:volume_mesh_info=Flow360ResourceBaseModel(**RestApi(VolumeMeshInterface.endpoint,id=volume_mesh_id).get())self.solver_version=volume_mesh_info.solver_versionis_valid_uuid(volume_mesh_id)self.validator_api(self.params,volume_mesh_id=volume_mesh_id,solver_version=self.solver_version,raise_on_error=(notforce_submit),)data={"name":self.name,"meshId":volume_mesh_id,"runtimeParams":self.params.flow360_json(),"tags":self.tags,"parentId":parent_id,}ifself.solver_versionisnotNone:data["solverVersion"]=self.solver_versionresp=RestApi(CaseInterface.endpoint).post(json=data,path=f"volumemeshes/{volume_mesh_id}/case",)info=CaseMeta(**resp)self._id=info.idself._submitted_case=Case(self.id)log.info(f"Case successfully submitted: {self._submitted_case.short_description()}")returnself._submitted_casedefvalidate_case_inputs(self,pre_submit_checks=False):""" validates case inputs (before submit only) """ifself.volume_mesh_idisnotNoneandself.other_caseisnotNone:raiseFlow360ValueError("You cannot specify both volume_mesh_id AND other_case.")ifself.parent_idisnotNoneandself.parent_caseisnotNone:raiseFlow360ValueError("You cannot specify both parent_id AND parent_case.")ifself.parent_idisnotNoneorself.parent_caseisnotNone:ifself.volume_mesh_idisnotNoneorself.other_caseisnotNone:raiseFlow360ValueError("You cannot specify volume_mesh_id OR other_case when parent case provided.")is_valid_uuid(self.volume_mesh_id,allow_none=True)ifpre_submit_checks:is_object_cloud_resource(self.other_case)is_object_cloud_resource(self.parent_case)@classmethoddefvalidator_api(cls,params:Flow360Params,volume_mesh_id,solver_version:str=None,raise_on_error:bool=True,):""" validation api: validates case parameters before submitting """returnValidator.CASE.validate(params,mesh_id=volume_mesh_id,solver_version=solver_version,raise_on_error=raise_on_error,)# pylint: disable=too-many-instance-attributes
[docs]classCase(CaseBase,Flow360Resource):""" Case component """# pylint: disable=redefined-builtindef__init__(self,id:str):super().__init__(interface=CaseInterface,info_type_class=CaseMeta,id=id,)self._params=Noneself._raw_params=Noneself._results=CaseResultsModel(case=self)@classmethoddef_from_meta(cls,meta:CaseMeta):validate_type(meta,"meta",CaseMeta)case=cls(id=meta.id)case._set_meta(meta)returncase@propertydefparams(self)->Flow360Params:""" returns case params """ifself._paramsisNone:self._raw_params=json.loads(self.get(method="runtimeParams")["content"])try:withtempfile.NamedTemporaryFile(mode="w",suffix=".json",delete=False)astemp_file:json.dump(self._raw_params,temp_file)self._params=Flow360Params(temp_file.name)exceptpd.ValidationErroraserr:raiseFlow360ValidationError(error_messages.params_fetching_error(err))fromerrreturnself._params@propertydefparams_as_dict(self)->dict:""" returns case params as dictionary """ifself._raw_paramsisNone:self._raw_params=json.loads(self.get(method="runtimeParams")["content"])returnself._raw_params
[docs]defhas_parent(self)->bool:"""Check if case has parent case Returns ------- bool True when case has parent, False otherwise """returnself.info.parent_idisnotNone
@propertydefparent(self)->Case:"""parent case Returns ------- Case parent case object Raises ------ RuntimeError When case does not have parent """ifself.has_parent():returnCase(self.info.parent_id)raiseFlow360RuntimeError("Case does not have parent case.")@propertydefinfo(self)->CaseMeta:""" returns metadata info for case """returnsuper().info@propertydefvolume_mesh_id(self):""" returns volume mesh id """returnself.info.case_mesh_id@propertydefresults(self)->CaseResultsModel:""" returns results object to managing case results """returnself._results
[docs]defis_steady(self):""" returns True when case is steady state """returnself.params.time_stepping.time_step_size=="inf"
[docs]defhas_actuator_disks(self):""" returns True when case has actuator disk """ifself.params.actuator_disksisnotNone:iflen(self.params.actuator_disks)>0:returnTruereturnFalse
[docs]defhas_bet_disks(self):""" returns True when case has BET disk """ifself.params.bet_disksisnotNone:iflen(self.params.bet_disks)>0:returnTruereturnFalse
[docs]defhas_isosurfaces(self):""" returns True when case has isosurfaces """returnself.params.iso_surface_outputisnotNone
[docs]defhas_monitors(self):""" returns True when case has monitors """returnself.params.monitor_outputisnotNone
[docs]defhas_aeroacoustics(self):""" returns True when case has aeroacoustics """returnself.params.aeroacoustic_outputisnotNone
[docs]defhas_user_defined_dynamics(self):""" returns True when case has user defined dynamics """returnself.params.user_defined_dynamicsisnotNone
[docs]defis_finished(self):""" returns False when case is in running or preprocessing state """returnself.status.is_final()
[docs]defmove_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).put(MoveToFolderRequest(dest_folder_id=folder.id,items=[MoveCaseItem(id=self.id)]).dict(),method="move",)returnself
@classmethoddef_interface(cls):returnCaseInterface@classmethoddef_meta_class(cls):""" returns case meta info class: CaseMeta """returnCaseMeta@classmethoddef_params_ancestor_id_name(cls):""" returns volumeMeshId name """return"meshId"
[docs]@classmethoddeffrom_cloud(cls,case_id:str):""" get case from cloud """returncls(case_id)
# pylint: disable=too-many-arguments
[docs]@classmethoddefcreate(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: """assertnameassertvolume_mesh_idorother_caseorparent_idorparent_caseassertparamsifnotisinstance(params,Flow360Params)andnotisinstance(params,UnvalidatedFlow360Params):raiseFlow360ValueError("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,)returnnew_case
[docs]defwait(self,timeout_minutes=60):"""Wait until the Case finishes processing, refresh periodically"""start_time=time.time()whileself.is_finished()isFalse:iftime.time()-start_time>timeout_minutes*60:raiseTimeoutError("Timeout: Process did not finish within the specified timeout period")time.sleep(2)
# pylint: disable=unnecessary-lambdaclassCaseResultsModel(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))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())# forcestotal_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())force_distribution:ForceDistributionResultCSVModel=pd.Field(default_factory=lambda:ForceDistributionResultCSVModel())# user defined:user_defined_dynamics:UserDefinedDynamicsResultModel=pd.Field(default_factory=lambda:UserDefinedDynamicsResultModel())# otherssurface_heat_transfer:SurfaceHeatTrasferResultCSVModel=pd.Field(default_factory=lambda:SurfaceHeatTrasferResultCSVModel())aeroacoustics:AeroacousticsResultCSVModel=pd.Field(default_factory=lambda:AeroacousticsResultCSVModel())_downloader_settings:ResultsDownloaderSettings=pd.PrivateAttr(ResultsDownloaderSettings())# pylint: disable=no-self-argument, protected-access@pd.root_validator(pre=False)defpass_download_function(cls,values):""" Pass download methods into fields of the case results """if"case"notinvalues:raiseValueError("case (type Case) is required")ifnotisinstance(values["case"],Case):raiseTypeError("case must be of type Case")forfieldincls.__fields__.values():iffield.nameinvalues.keys():value=values[field.name]ifisinstance(value,ResultBaseModel):value._download_method=values["case"]._download_filevalue._get_params_method=lambda:values["case"].paramsvalues[field.name]=valuereturnvalues# pylint: disable=no-self-argument, protected-access@pd.validator("monitors","user_defined_dynamics",always=True)defpass_get_files_function(cls,value,values):""" Pass file getters into fields of the case results """value.get_download_file_list_method=values["case"].get_download_file_listreturnvalue# pylint: disable=no-self-argument, protected-access@pd.validator("bet_forces",always=True)defpass_has_bet_forces_function(cls,value,values):""" Pass check to see if result is downloadable based on params """value._is_downloadable=values["case"].has_bet_disksreturnvalue# pylint: disable=no-self-argument, protected-access@pd.validator("actuator_disks",always=True)defpass_has_actuator_disks_function(cls,value,values):""" Pass check to see if result is downloadable based on params """value._is_downloadable=values["case"].has_actuator_disksreturnvalue# pylint: disable=no-self-argument, protected-access@pd.validator("isosurfaces",always=True)defpass_has_isosurfaces_function(cls,value,values):""" Pass check to see if result is downloadable based on params """value._is_downloadable=values["case"].has_isosurfacesreturnvalue# pylint: disable=no-self-argument, protected-access@pd.validator("monitors",always=True)defpass_has_monitors_function(cls,value,values):""" Pass check to see if result is downloadable based on params """value._is_downloadable=values["case"].has_monitorsreturnvalue# pylint: disable=no-self-argument, protected-access@pd.validator("aeroacoustics",always=True)defpass_has_aeroacoustics_function(cls,value,values):""" Pass check to see if result is downloadable based on params """value._is_downloadable=values["case"].has_aeroacousticsreturnvalue# pylint: disable=no-self-argument, protected-access@pd.validator("user_defined_dynamics",always=True)defpass_has_user_defined_dynamics_function(cls,value,values):""" Pass check to see if result is downloadable based on params """value._is_downloadable=values["case"].has_user_defined_dynamicsreturnvaluedef_execute_downloading(self):""" Download all specified and available results for the case """for_,valueinself.__dict__.items():ifisinstance(value,ResultBaseModel):# we download if explicitly set set_downloader(<result_name>=True),# or all=True but only when is not result=Falsetry_download=value.do_downloadisTrueifself._downloader_settings.allisTrueandvalue.do_downloadisnotFalse:try_download=value._is_downloadable()isTrueiftry_downloadisTrue:value.download(to_folder=self._downloader_settings.destination,overwrite=self._downloader_settings.overwrite,)defset_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 providedifsum(argisnotNoneforargin[folder_name,use_case_name,use_case_id])!=1:raiseValueError("Exactly one argument should be provided.")iffolder_nameisnotNone:self._downloader_settings.destination=folder_nameifuse_case_nameisTrue:self._downloader_settings.destination=self.case.nameifuse_case_idisTrue:self._downloader_settings.destination=self.case.id# pylint: disable=too-many-arguments, too-many-locals, redefined-builtindefdownload(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,actuator_disks:bool=None,force_distribution:bool=None,user_defined_dynamics: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. 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. """self.surfaces.do_download=surfaceself.volumes.do_download=volumeself.slices.do_download=slicesself.isosurfaces.do_download=isosurfacesself.monitors.do_download=monitorsself.nonlinear_residuals.do_download=nonlinear_residualsself.linear_residuals.do_download=linear_residualsself.cfl.do_download=cflself.minmax_state.do_download=minmax_stateself.max_residual_location.do_download=max_residual_locationself.surface_forces.do_download=surface_forcesself.total_forces.do_download=total_forcesself.bet_forces.do_download=bet_forcesself.actuator_disks.do_download=actuator_disksself.force_distribution.do_download=force_distributionself.user_defined_dynamics.do_download=user_defined_dynamicsself.aeroacoustics.do_download=aeroacousticsself.surface_heat_transfer.do_download=surface_heat_transferself._downloader_settings.all=allself._downloader_settings.overwrite=overwriteifdestinationisnotNone:self.set_destination(folder_name=destination)self._execute_downloading()defdownload_file_by_name(self,file_name,to_file=None,to_folder=".",overwrite:bool=True):""" Download file by name """returnself.case._download_file(file_name=file_name,to_file=to_file,to_folder=to_folder,overwrite=overwrite)classCaseList(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,)deffilter(self):""" flitering list, not implemented yet """raiseNotImplementedError("Filters are not implemented yet")# resp = list(filter(lambda i: i['caseStatus'] != 'deleted', resp))# pylint: disable=useless-parent-delegationdef__getitem__(self,index)->Case:""" returns CaseMeta info item of the list """returnsuper().__getitem__(index)def__iter__(self)->Iterator[Case]:returnsuper().__iter__()