o }oi3@sddlZddlZddlmZmZmZddlmZmZm Z m Z ddl m Z ddlmZmZddlmZddlmZddlmZejdkrFeZneZe d Ze d ZGd d d eeeefZGd ddeeeefZdS)N)Path PosixPath WindowsPath)GenericOptionalTupleTypeVar)FileLockTimeout) TrainerFn)ckpt_to_context_subdir)loggingntSourceTTargetTcseZdZdZdZdZdefddZdedefdd Z fd d Z dde ed e defddZ dde edefddZdde ede fddZZS) Connectora A generic connector class that provides a framework for transforming a source type (SourceT) to a target type (TargetT) while handling file paths based on the operating system. Attributes ---------- default_path (Optional[Path]): A default path used when no path is explicitly provided. Methods ------- init() -> TargetT: Should be implemented to initialize the target type from the source type. apply(output_path: Path) -> Path: Should be implemented to apply the transformation and save the result at the output path. __new__(cls, *args, **kwargs) -> 'Connector': Creates a new instance of the connector, using default_path if no path is provided. __call__(output_path: Optional[Path] = None, overwrite: bool = False) -> Path: Processes the transformation and handles file operations like overwriting. local_path(base_path: Optional[Path] = None) -> Path: Computes the local path for storage based on a base path or a default cache home. is_in_cache(base_path: Optional[Path] = None) -> bool: Checks if the transformed data is already cached at the specified base path. NireturncCt)zIShould be implemented to initialize the target type from the source type.NotImplementedError)selfrO/home/ubuntu/.local/lib/python3.10/site-packages/nemo/lightning/io/connector.pyinitGzConnector.init output_pathcKr)zYShould be implemented to apply the transformation and save the result at the output path.r)rrkwargsrrrapplyKrzConnector.applycs@|jdur|sd|vrt||jStj|g|Ri|S)Npath) default_pathsuper__new__)clsargsr __class__rrr!OszConnector.__new__F overwritec Ks|p|}||jd}t|}|rd}zz5|j|jd$|r-|r-t||s>|j |fi|}|p=|}Wdn1sHwYWn$t y^t d|t yr}z t d|d}~wwW|rzt|W|Sty}zt d|d|WYd}~|Sd}~ww|S|rzt|Wwty}zt d|d|WYd}~wd}~www)Nz.lockF)timeoutz6Timeout occurred while trying to acquire the lock for zAn error occurred: zFailed to remove lock file z: ) local_path with_suffixsuffixr existsacquire LOCK_TIMEOUTshutilrmtreerr r error ExceptionosremoveOSErrorwarning) rrr&r _output_path lock_pathlock to_returnerrr__call__VsT      "zConnector.__call__ base_pathcCs2|r|}n ddlm}t|}|t|ddS)zQComputes the local path for storage based on a base path or a default cache home.r)NEMO_CACHE_HOME:///)nemo.lightning.baser=rstrreplace)rr<_baser=rrrr(zs  zConnector.local_pathcCs|j|dS)zLChecks if the transformed data is already cached at the specified base path.)r<)r(r+)rr<rrr is_in_cacheszConnector.is_in_cache)NFN)__name__ __module__ __qualname____doc__rr-rrrrr!rboolr;r(rD __classcell__rrr$rr&s $ rc @seZdZdZ ddejdeejdejfddZdd e dejd e ddfd d Z dd e deejde de ejejffddZ ddee de fddZdejfddZdddZdS)ModelConnectora, A specialized connector that extends the generic Connector to handle model-specific operations such as setup, save, and load using the Lightning framework. Methods ------- nemo_setup(model: pl.LightningModule, trainer: Optional[pl.Trainer] = None) -> pl.Trainer: Sets up the model and trainer using a specified strategy, preparing it for training or inference. nemo_save(output_path: Path, trainer: pl.Trainer): Saves the model's state to the specified path using the trainer's current strategy. nemo_load(path: Path, trainer: Optional[pl.Trainer] = None, cpu: bool = True) -> Tuple[Any, pl.Trainer]: Loads a model from the specified path, optionally using a CPU-focused strategy, and returns the model and trainer. Nmodeltrainerrc Osddlm}m}m}|p|dd||ddd|d}tj|j_|j ||j | sjd|j_ | +||j |Wd n1sMwYWd |SWd |S1sewY|S) a Sets up the model and trainer using a specified strategy, preparing it for training or inference. Args: model (pl.LightningModule): The model to be set up. trainer (Optional[pl.Trainer]): The trainer to be used, if not provided a new one will be created. Returns ------- pl.Trainer: The trainer configured with the model and strategy. rMegatronStrategyTrainer _strategy_libcpuFT)ckpt_save_optimizeralways_save_context)devices acceleratorstrategyN)nemo.lightningrPrQrRr FITTINGstatefnrYconnectsetup_environment state_dict lazy_init init_modulemegatron_lazy_init_contextconfigconfigure_model) rrMrNr#rrPrQrR_trainerrrr nemo_setups&    ( zModelConnector.nemo_setupTrdump_iocCsd|j_d|j_|j|t|}|jddd||t|jddr-|jjj ddddl m }ddl m }|rM|rO||jt|d gd d Sd Sd S) av Saves the model's state to the specified path using the trainer's current strategy. Args: output_path (Path): The path where the model checkpoint will be saved. trainer (pl.Trainer): The trainer with the strategy to save the model. dump_io (bool): If True, the IO configuration will be saved to the output path. FT)parentsexist_ok async_save)blockingr)TrainerContext)is_global_rank_zerorM) yaml_attrsN)rY_setup_optimizers_init_model_parallelsetuprmkdirsave_checkpointgetattr checkpoint_iomaybe_finalize_save_checkpointnemo.lightning.io.plrmnemo.utils.get_rankrn from_trainerio_dumpr )rrrNrhrmrnrrr nemo_saves      zModelConnector.nemo_saverrTcCsddlm}ddlm}m}m}ddlm}||dd} d| j_ d| j_ d| j_ || || j du} g} | r<| | j |pM|d |rDd nd |d dd | d} | j| | j| s~|rz|| j | Wdn1stwYn| | j| | rddlm} | | _| | } dd| jjD}| jjj| |dd|d}| jj|dd| | fS| j|| | fS)a Loads a model from the specified path. Args: path (Path): The path from which the model will be loaded. trainer (Optional[pl.Trainer]): The trainer to be used, if not provided a new one will be created. cpu (bool): If True, the model will be loaded with a CPU-focused strategy. Returns ------- Tuple[pl.LightningModule, pl.Trainer]: The loaded model and the trainer configured with the model. r)#set_modelopt_spec_if_exists_in_ckptrO) load_contextrM)subpathNFrSrTgpupytorch)ddpsetup_optimizers)rWrXrY callbacks)ckpt_to_weights_subdircSsi|] \}}d|vr||qS)z .adapter.r).0kvrrr sz,ModelConnector.nemo_load..) is_saving)sharded_state_dict)strict)nemo.collections.llm.modeloptr}rZrPrQrRnemo.lightning.io.apir~rdfp8 fp8_paramperform_initializationmodel_transformappendrYr^r_r`megatron_cpu_init_contextrerrrxrrNmegatron_parallelritemsrvload_checkpointload_model_state_dict)rrrNrTr}rPrQrRr~rM is_peft_ckptrrfradapter_sharded_state_dict adapter_staterrr nemo_loadsT                zModelConnector.nemo_loadr<cCsd|rt|}n ddlm}t|}t|dr(|jdvr#||jjS||jS|t|ddS)Nr)NEMO_MODELS_CACHEr?)z.ptz.pthr>) rr@rrA startswithr*parentnamerB)rr<rCrrrrr("s     zModelConnector.local_pathcCsDt|dr|j|_t|drt|jdr |jj|j_dSdSdSdS)z#Called after checkpoint is imported tokenizer__io__N)hasattrrr)rrMrrron_import_ckpt2s zModelConnector.on_import_ckpt/tmp/nemo_tokenizercCs(ddlm}|j|dd}|||S)z,Save HF tokenizer to the imported NeMo modelr) AutoTokenizerT)trust_remote_code) transformersrfrom_pretrainedsave_pretrained)rtokenizer_name_or_path save_pathrtokrrrsave_hf_tokenizer_assets9s  z'ModelConnector.save_hf_tokenizer_assetsrE)T)NT)r)rFrGrHrIplLightningModulerrQrgrrJr|rrr(rrrrrrrLs0 # IrL)r2r.pathlibrrrtypingrrrrlightning.pytorchrrfilelockr r lightning.pytorch.trainer.statesr nemo.lightning.ckpt_utilsr nemo.utilsr rBasePathrrrrLrrrrs      d