o  iZ@sHddlmZddlZddlZddlZddlmZmZmZddl m Z ddl m Z m Z ddlZddlmZddlmZddlmZmZmZmZmZmZed ejed \ZZed d d \ZZeddd \e_Zeddd d\ZZeddd \Z Zedded\Z!Ze rddl"m#Z#n ed ejeddd\Z#ZdZ$ee%dZ&GdddZ'dS)) annotationsN)CallableMappingSequence)Path) TYPE_CHECKINGAny)Dataset) get_logger) CommonKeys IgniteInfo ensure_tuple flatten_dict min_versionoptional_importz ignite.engineEventsmlflowz1Please install mlflow before using MLFlowHandler.) descriptorzmlflow.entitiesz:Please install mlflow.entities before using MLFlowHandler.zmlflow.exceptionsMlflowException)namerpandasz0Please install pandas for recording the dataset.tqdmz4.47.0)Enginer decorator)as_typeLoss) module_namec@seZdZdZddgZdddddddejddddded dddd d fdZd,d-Zd[d0d1Z d\d4d5Z d\d6d7Z d8d9Z e d:d;Zd]d^d@dAZd_dCdDZd`dadHdIZdJdKZdbdLdMZdbdNdOZd\dPdQZd\dRdSZd\dTdUZd\dVdWZdcdXdYZdS)d MLFlowHandlera MLFlowHandler defines a set of Ignite Event-handlers for the MLFlow tracking logics. It can be used for any Ignite Engine(trainer, validator and evaluator). And it can track both epoch level and iteration level logging, then MLFlow can store the data and visualize. The expected data source is Ignite ``engine.state.output`` and ``engine.state.metrics``. Default behaviors: - When EPOCH_COMPLETED, track each dictionary item in ``engine.state.metrics`` in MLFlow. - When ITERATION_COMPLETED, track expected item in ``self.output_transform(engine.state.output)`` in MLFlow, default to `Loss`. Usage example is available in the tutorial: https://github.com/Project-MONAI/tutorials/blob/master/3d_segmentation/unet_segmentation_3d_ignite.ipynb. Args: tracking_uri: connects to a tracking URI. can also set the `MLFLOW_TRACKING_URI` environment variable to have MLflow find a URI from there. in both cases, the URI can either be an HTTP/HTTPS URI for a remote server, a database connection string, or a local path to log data to a directory. The URI defaults to path `mlruns`. for more details: https://mlflow.org/docs/latest/python_api/mlflow.html#mlflow.set_tracking_uri. iteration_log: whether to log data to MLFlow when iteration completed, default to `True`. ``iteration_log`` can be also a function and it will be interpreted as an event filter (see https://pytorch.org/ignite/generated/ignite.engine.events.Events.html for details). Event filter function accepts as input engine and event value (iteration) and should return True/False. epoch_log: whether to log data to MLFlow when epoch completed, default to `True`. ``epoch_log`` can be also a function and it will be interpreted as an event filter. See ``iteration_log`` argument for more details. epoch_logger: customized callable logger for epoch level logging with MLFlow. Must accept parameter "engine", use default logger if None. iteration_logger: customized callable logger for iteration level logging with MLFlow. Must accept parameter "engine", use default logger if None. dataset_logger: customized callable logger to log the dataset information with MLFlow. Must accept parameter "dataset_dict", use default logger if None. dataset_dict: a dictionary in which the key is the name of the dataset and the value is a PyTorch dataset, that needs to be recorded. This arg is only useful when MLFlow version >= 2.4.0. For more details about how to log data with MLFlow, please go to the website: https://mlflow.org/docs/latest/python_api/mlflow.data.html. dataset_keys: a key or a collection of keys to indicate contents in the dataset that need to be stored by MLFlow. output_transform: a callable that is used to transform the ``ignite.engine.state.output`` into a scalar to track, or a dictionary of {key: scalar}. By default this value logging happens when every iteration completed. The default behavior is to track loss from output[0] as output is a decollated list and we replicated loss value for every item of the decollated list. `engine.state` and `output_transform` inherit from the ignite concept: https://pytorch-ignite.ai/concepts/03-state/, explanation and usage example are in the tutorial: https://github.com/Project-MONAI/tutorials/blob/master/modules/batch_output_transform.ipynb. global_epoch_transform: a callable that is used to customize global epoch number. For example, in evaluation, the evaluator engine might want to track synced epoch number with the trainer engine. state_attributes: expected attributes from `engine.state`, if provided, will extract them when epoch completed. tag_name: when iteration output is a scalar, `tag_name` is used to track, defaults to `'Loss'`. experiment_name: the experiment name of MLflow, default to `'monai_experiment'`. An experiment can be used to record several runs. run_name: the run name in an experiment. A run can be used to record information about a workflow, like the loss, metrics and so on. experiment_param: a dict recording parameters which will not change through the whole workflow, like torch version, cuda version and so on. artifacts: paths to images that need to be recorded after running the workflow. optimizer_param_names: parameter names in the optimizer that need to be recorded during running the workflow, default to `'lr'`. close_on_complete: whether to close the mlflow run in `complete` phase in workflow, default to False. For more details of MLFlow usage, please refer to: https://mlflow.org/docs/latest/index.html. max_epochs epoch_lengthNTcCs|dS)Nrxr r _/home/dell461/cl/sdc2/last_ska_mid/HISourceFinder-master-l/src/monai/handlers/mlflow_handler.pyszMLFlowHandler.cCs|SNr r!r r r#r$smonai_experimentlrF tracking_uri str | None iteration_log$bool | Callable[[Engine, int], bool] epoch_log epoch_loggerCallable[[Engine], Any] | Noneiteration_loggerdataset_logger-Callable[[Mapping[str, Dataset]], Any] | None dataset_dictMapping[str, Dataset] | None dataset_keysstroutput_transformrglobal_epoch_transformstate_attributesSequence[str] | Nonetag_nameexperiment_namerun_nameexperiment_param dict | None artifactsstr | Sequence[Path] | Noneoptimizer_param_namesstr | Sequence[str]close_on_completeboolreturnNonecCs||_||_||_||_||_| |_| |_| |_| |_| |_ ||_ ||_ t ||_ t ||_tj|r4|ndd|_tjjtjjj|_||_d|_d|_||_t ||_dS)N)r()r*r,r-r/r0r6r7r8r:r;r<r=r r?rArZ MlflowClientcliententitiesZ RunStatus to_stringFINISHEDrun_finish_statusrC experimentcur_runr2r4)selfr(r*r,r-r/r0r2r4r6r7r8r:r;r<r=r?rArCr r r#__init__|s*  zMLFlowHandler.__init__ param_dictdictcCsP|jdurdSt|}|j|jjjj}|j}|D] }||vr%||=qdS)z Delete parameters in given dict, if they are already logged by current mlflow run. Args: param_dict: parameter dict to be logged to mlflow. N) rMlistkeysrGget_runinforun_iddataparams)rNrPkey_listlog_dataZlog_param_dictkeyr r r#_delete_exist_param_in_dicts  z)MLFlowHandler._delete_exist_param_in_dictenginercCs||jtjs|tj|j|jr0||jtjs0tj}t|jr)||jd}|||j|j rP||j tj sPtj }t|j rI||j d}|||j ||j tj s`|tj |j |jru||jtj sw|tj |jdSdSdS)z Register a set of Ignite Event-Handlers to a specified Ignite engine. Args: engine: Ignite Engine, it can be a trainer, validator or evaluator. ) event_filterN)has_event_handlerstartrSTARTEDadd_event_handlerr*iteration_completedITERATION_COMPLETEDcallabler,epoch_completedEPOCH_COMPLETEDcomplete COMPLETEDrCclose)rNr]eventr r r#attachs"    zMLFlowHandler.attachcsjstdjdjsWjdur dtdnjj jj }fdd|D}fdd|D}|rLj |d j j _n jjjj d _jr`jfd d jD}||jrjdSjdS) z? Check MLFlow status and start if not active. zFailed to set experiment '' as the active experimentNrun_z %Y%m%d_%H%M%Scs"g|] }|jjks js|qSr )rUr<.0r)r<rNr r# "z'MLFlowHandler.start..csg|] }|jjjkr|qSr )rUstatusrKrorNr r#rr) experiment_idr<ci|] }|tj|dqSr%getattrstaterpattrr]r r# z'MLFlowHandler.start..)_set_experimentrL ValueErrorr;rMr<timestrftimerGZ search_runsrxrTrUrVZ create_runr= _log_paramsdefault_tracking_paramsr\r0r2_default_dataset_log)rNr]runsattrsr )r]r<rNr#r`s&    zMLFlowHandler.startc Cs|j}|sPtdD]F}z|j|j}|s"|j|j}|j|}Wn+tyO}zdt|vrCt dt d|dkrB|n|WYd}~q d}~ww|j tjjjkratd|jd||_dS)NZRESOURCE_ALREADY_EXISTSz4Experiment already exists; delaying before retrying.z!Cannot set a deleted experiment 'rm)rLrangerGZget_experiment_by_namer;Zcreate_experimentZget_experimentrr5loggerwarningrsleepZlifecycle_stagerrHZLifecycleStageZACTIVEr)rNrLZ _retry_timerxer r r#rs.        zMLFlowHandler._set_experimentcCs&|j}|d|j|d|jdiS)NZ_digest_samplesnum_rows)rdigestprofile)Zpandas_dataset dataset_namer r r#_get_pandas_dataset_infos z&MLFlowHandler._get_pandas_dataset_infotrain sample_dictdict[str, Any]contextc s|jstd|j|jjj|_fdd|jjjD}tt |}d|}t |}t j j||dttfdd|jjj}t |sit jg}|jj|jjj|dt} || dSdS) Nz,Current Run is not Active to log the datasetcsg|] }|jjr|qSr )datasetr startswith)rpr")rr r#rrrvz.MLFlowHandler._log_dataset..Z _dataset_)rcs|jjjkSr%)rrr!)rr r#r$sz,MLFlowHandler._log_dataset..)rVdatasets)rMrrGrTrUrVinputsZdataset_inputsr5lenr DataFramerrW from_pandasrRfilterrHZ DatasetInputZ_to_mlflow_entityZ log_inputsrrr) rNrrZ logged_setZ dataset_countrZ sample_dfZexist_dataset_listrZ dataset_infor )rrr# _log_datasets"   zMLFlowHandler._log_datasetrXcCs>|jstddd|D}|jj|jjjg|gddS)Nz'Current Run is not Active to log paramscSs"g|] \}}tj|t|qSr )rrHParamr5rpr[valuer r r#rr)rsz-MLFlowHandler._log_params..rVmetricsrXtags)rMritemsrG log_batchrUrV)rNrXZ params_arrr r r#r&szMLFlowHandler._log_paramsrstep int | Nonecs\|jstd|jjj}ttdfddt|D}|jj ||ggddS)Nz(Current Run is not Active to log metricsics&g|]\}}tj||pdqS)r)rrHMetricrr timestampr r#rr2sz.MLFlowHandler._log_metrics..r) rMrrUrVintrrrrGr)rNrrrVZ metrics_arrr rr# _log_metrics,s   zMLFlowHandler._log_metricscCsjg}|jD]-}|s qtj|r||qt|D]\}}}|D]}tj||}||q"qq|S)z Log artifacts to mlflow. Given a path, all files in the path will be logged recursively. Given a file, it will be logged to mlflow. )r?ospathisfileappendwalkjoin)rN artifact_list path_nameroot_ filenamesfilename file_pathr r r#_parse_artifacts7s    zMLFlowHandler._parse_artifactscCs>|jr|jr|}|D]}|j|jjj|q dSdSdS)zM Handler for train or validation/evaluation completed Event. N)r?rMrrGZ log_artifactrUrV)rNrartifactr r r#rhJs zMLFlowHandler.completecCs*|jr|j|jjj|jd|_dSdS)z9 Stop current running logger of MLFlow. N)rMrGZset_terminatedrUrVrKrur r r#rjSs zMLFlowHandler.closecC&|jdur ||dS||dS)a Handler for train or validation/evaluation epoch completed Event. Track epoch level log, default values are from Ignite `engine.state.metrics` dict. Args: engine: Ignite Engine, it can be a trainer, validator or evaluator. N)r-_default_epoch_logrNr]r r r#rf\ zMLFlowHandler.epoch_completedcCr)z Handler for train or validation/evaluation iteration completed Event. Track iteration level log. Args: engine: Ignite Engine, it can be a trainer, validator or evaluator. N)r/_default_iteration_logrr r r#rcjrz!MLFlowHandler.iteration_completedcs`jj}|sdS|jj}|j||d|jdur.fdd|jD}|j||ddSdS)a( Execute epoch level log operation. Default to track the values from Ignite `engine.state.metrics` dict and track the values of specified attributes of `engine.state`. Args: engine: Ignite Engine, it can be a trainer, validator or evaluator. Nrcryr%rzr}rr r#rrz4MLFlowHandler._default_epoch_log..)r|rr7epochrr8)rNr]Zlog_dict current_epochrr rr#rxs  z MLFlowHandler._default_epoch_logcs||jj}|dur dSt|ts!|jt|tjr|n|i}|j ||jj dt |drM|j }|j D]fddt|jD}|j ||jj dq5dSdS)a Execute iteration log operation based on Ignite `engine.state.output` data. Log the values from `self.output_transform(engine.state.output)`. Since `engine.state.output` is a decollated list and we replicated the loss value for every item of the decollated list, the default behavior is to track the loss from `output[0]`. Args: engine: Ignite Engine, it can be a trainer, validator or evaluator. Nr optimizercs(i|]\}}d|t|qS)Z_group_)float)rpi param_group param_namer r#rsz8MLFlowHandler._default_iteration_log..)r6r|output isinstancerQr:torchTensoritemr iterationhasattrrrA enumerate param_groups)rNr]lossZ cur_optimizerrXr rr#rs     z$MLFlowHandler._default_iteration_logc Cs|durdSt|dkrtd|D]d\}}|dur%td|di}t|dg}t|d|dD]<}|jD]6}||vrEg||<||vrN||}ntd |d t |t sjtd t |d |d q;|| |q;q6| ||qdS)a Execute dataset log operation based on the input dataset_dict. The dataset_dict should have a format like: { "dataset_name0": dataset0, "dataset_name1": dataset1, ...... } The keys stand for names of datasets, which will be logged as prefixes of dataset names in MLFlow. The values are PyTorch datasets from which sample names are abstracted to build a Pandas DataFrame. If the input dataset_dict is None, this function will directly return and do nothing. To use this function, every sample in the input datasets must contain keys specified by the `dataset_keys` parameter. This function will log a PandasDataset to MLFlow inputs, generated from the Pandas DataFrame. For more details about PandasDataset, please refer to this link: https://mlflow.org/docs/latest/python_api/mlflow.data.html#mlflow.data.pandas_dataset.PandasDataset Please note that it may take a while to record the dataset if it has too many samples. Args: dataset_dict: a dictionary in which the key is the name of the dataset and the value is a PyTorch dataset, that needs to be recorded. NrzThere is no dataset to log!zThe z0 dataset of is None. Cannot record it by MLFlow.rWzRecording the z datasetzUnexpect key 'z' in the sample.zExpected type string, got type z of the z( name.May log an empty dataset in MLFlow)rwarningswarnrAttributeErrorr{rr4KeyErrorrr5typerr) rNr2Z dataset_typerrZdataset_samplessampler[Z value_to_logr r r#rs0      z"MLFlowHandler._default_dataset_log)&r(r)r*r+r,r+r-r.r/r.r0r1r2r3r4r5r6rr7rr8r9r:r5r;r5r<r)r=r>r?r@rArBrCrDrErF)rPrQrErF)r]rrErF)r)rrrr5rErF)rXrrErFr%)rrrrrErF)rErF)r2r3rErF)__name__ __module__ __qualname____doc__rr IMAGE DEFAULT_TAGrOr\rlr`r staticmethodrrrrrrhrjrfrcrrrr r r r#r2sNG +           r)( __future__rrrrcollections.abcrrrpathlibrtypingrrrtorch.utils.datar monai.apps.utilsr monai.utilsr r r rrrOPT_IMPORT_VERSIONrrrrHrrr ignite.enginerrrrrr r r r#s: