o }oi@slddlZddlmZmZmZddlmZddlmZddl m Z m Z m Z m Z ddlmZGdddeZdS) N)AnyDictLiteral)EVAL_DATALOADERS)parallel_state)DefaultTaskEncoder WorkerConfigget_savable_loaderget_train_dataset)EnergonMultiModalDataModulecseZdZdZ          d!d ed ed ed ed ededededededBdedBddffdd Zd"de dfddZ de fddZ de eefddffdd ZZS)#DiffusionDataModulea A PyTorch Lightning DataModule for handling multimodal datasets with images and text. This data module is designed to work with multimodal datasets that involve both images and text. It provides a seamless interface to load training and validation data, manage batching, and handle the state of the data pipeline across training epochs. The module integrates with the Megatron-Energon framework for efficient data handling in large-scale distributed training. Attributes: path (str): Path to the energon dataset. tokenizer (Tokenizer): The tokenizer used for processing text. image_processor (ImageProcessor): The image processor used for preprocessing images. seq_length (int): The maximum sequence length for tokenized text. micro_batch_size (int): The batch size for training and validation. num_workers (int): Number of workers for data loading. pin_memory (bool): Whether to pin memory in the DataLoader. multimodal_sample_config (MultiModalSampleConfig): Configuration object for multimodal samples. task_encoder (MultiModalTaskEncoder): Encoder responsible for encoding and batching samples. init_global_step (int): The initial global step for the trainer, used for resuming training. data_sampler (SequentialMegatronSampler): Sampler responsible for generating sequential samples. train_dataloader_object (Optional): The DataLoader object for training data. val_dataloader_object (Optional): The DataLoader object for validation data. TNFʚ;path seq_lengthmicro_batch_sizeglobal_batch_size num_workers pin_memory task_encoderuse_train_split_for_valvirtual_epoch_lengthpacking_buffer_sizemax_samples_per_sequencereturnc s@tj|dd||||||d ||_| |_d|_| |_| |_dS)a Initialize the EnergonMultiModalDataModule. Parameters: path (str): Path to the dataset. tokenizer (Tokenizer): The tokenizer used for processing text. image_processor (ImageProcessor): The image processor used for preprocessing images. seq_length (int, optional): The maximum sequence length for tokenized text. Defaults to 2048. micro_batch_size (int, optional): The batch size for training and validation. Defaults to 1. num_workers (int, optional): Number of workers for data loading. Defaults to 1. pin_memory (bool, optional): Whether to pin memory in the DataLoader. Defaults to True. N) r tokenizerimage_processorrrrrrrr)super__init__rrnum_workers_valrr) selfrrrrrrrrrrr __class__p/home/ubuntu/.local/lib/python3.10/site-packages/nemo/collections/diffusion/data/diffusion_energon_datamodule.pyr 2s   zDiffusionDataModule.__init__valsplit)trainr'c CsD|dvrtd|jr d}t|j|j|j||jd||j|jd }|S)a Provide the dataset for training or validation. This method retrieves the dataset for the specified split (either 'train' or 'val') and configures it according to the worker configuration. Parameters: worker_config: Configuration for the data loader workers. split (Literal['train', 'val'], optional): The data split to retrieve ('train' or 'val'). Defaults to 'val'. Returns: Dataset: The dataset configured for the specified split. >r'r)z=Invalid value for split. Allowed values are 'train' or 'val'.r)N) batch_sizer worker_configrshuffle_buffer_size split_partrr) ValueErrorrr rrrrrr)r"r+r(_datasetr%r%r&datasets_provider^s  z%DiffusionDataModule.datasets_providercCs|jr|S|jr |jSts#d|j}t|t |j }n%t }t }t }td|d|d|t|||j |ddd}|j|dd }t||d }||_|jS) an Initialize and return the validation DataLoader. This method initializes the DataLoader for the validation dataset. It ensures that the parallel state is initialized correctly for distributed training and returns a configured DataLoader object. Returns: EVAL_DATALOADERS: The DataLoader for the validation dataset. zjMuiltimodal val data loader parallel state is not initialized using default worker config with no_workers zrank z world_size z data_parallel_group Nr)rank world_sizerdata_parallel_groupworker_debug_pathworker_log_levelr')r()r+)rtrain_dataloaderval_dataloader_objectris_initializedrlogginginfordefault_worker_configr!get_data_parallel_rankget_data_parallel_world_sizeget_data_parallel_groupr0r )r"messager+r1r2r3 val_datasetenergon_loaderr%r%r&val_dataloader}s6   z"DiffusionDataModule.val_dataloader state_dictc sJz t|WdSty$}ztd|WYd}~dSd}~ww)az Load the state of the data module from a checkpoint. This method is called when loading a checkpoint. It restores the state of the data module, including the state of the dataloader and the number of consumed samples. Parameters: state_dict (Dict[str, Any]): The state dictionary containing the saved state of the data module. z#datamodule.load_state_dict failed N)rload_state_dict Exceptionr9warning)r"rCer#r%r&rDs  z#DiffusionDataModule.load_state_dict) r rrrTNFrNN)r')__name__ __module__ __qualname____doc__strintboolrr rr0rrBrrrD __classcell__r%r%r#r&r sP     ,&*r )r9typingrrr!lightning.pytorch.utilities.typesr megatron.corermegatron.energonrrr r -nemo.collections.multimodal.data.energon.baser r r%r%r%r&s