
import argparse
import glob
import logging
import os
import time
from argparse import Namespace

import numpy as np
import torch
from lightning_base import BaseTransformer, add_generic_args, generic_train
from torch.utils.data import DataLoader, TensorDataset

from transformers import glue_compute_metrics as compute_metrics
from transformers import glue_convert_examples_to_features as convert_examples_to_features
from transformers import glue_output_modes, glue_tasks_num_labels
from transformers import glue_processors as processors


lowerCamelCase  =logging.getLogger(__name__)




class 					_lowerCamelCase			(   UpperCamelCase_			):





     """simple docstring"""







     SCREAMING_SNAKE_CASE_		 =	'''sequence-classification'''






     def __init__(				self    ,    __SCREAMING_SNAKE_CASE	)	-> Optional[Any]:




           """simple docstring"""







           if type(__SCREAMING_SNAKE_CASE	) == dict:
                 UpperCamelCase__     : List[str]											=					Namespace(**__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : List[Any]											=					glue_output_modes[hparams.task]
           UpperCamelCase__     : str											=					glue_tasks_num_labels[hparams.task]

           super().__init__(__SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    self.mode	)






     def  __SCREAMING_SNAKE_CASE      (				self    ,    **__SCREAMING_SNAKE_CASE	)	-> Optional[int]:




           """simple docstring"""







           return self.model(**__SCREAMING_SNAKE_CASE	)






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE	)	-> Tuple:




           """simple docstring"""







           UpperCamelCase__     : Dict											=					{'''input_ids''': batch[0], '''attention_mask''': batch[1], '''labels''': batch[3]}

           if self.config.model_type not in ["distilbert", "bart"]:
                 UpperCamelCase__     : str											=					batch[2] if self.config.model_type in ['''bert''', '''xlnet''', '''albert'''] else None

           UpperCamelCase__     : int											=					self(**__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : Optional[int]											=					outputs[0]

           UpperCamelCase__     : Dict											=					self.trainer.lr_schedulers[0]['''scheduler''']
           UpperCamelCase__     : Union[str, Any]											=					{'''loss''': loss, '''rate''': lr_scheduler.get_last_lr()[-1]}
           return {"loss": loss, "log": tensorboard_logs}






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> int:




           """simple docstring"""







           UpperCamelCase__     : List[str]											=					self.hparams
           UpperCamelCase__     : int											=					processors[args.task]()
           UpperCamelCase__     : Optional[int]											=					processor.get_labels()

           for mode in ["train", "dev"]:
                 UpperCamelCase__     : str											=					self._feature_file(__SCREAMING_SNAKE_CASE	)
                 if os.path.exists(__SCREAMING_SNAKE_CASE	) and not args.overwrite_cache:
                       logger.info('''Loading features from cached file %s'''    ,    __SCREAMING_SNAKE_CASE	)
                 else:
                       logger.info('''Creating features from dataset file at %s'''    ,    args.data_dir	)
                       UpperCamelCase__     : Union[str, Any]											=					(
                           processor.get_dev_examples(args.data_dir	)
                           if mode == '''dev'''
                           else processor.get_train_examples(args.data_dir	)
                       )
                       UpperCamelCase__     : Any											=					convert_examples_to_features(
                           __SCREAMING_SNAKE_CASE    ,    self.tokenizer    ,    max_length=args.max_seq_length    ,    label_list=self.labels    ,    output_mode=args.glue_output_mode    ,    )
                       logger.info('''Saving features into cached file %s'''    ,    __SCREAMING_SNAKE_CASE	)
                       torch.save(__SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE	)






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE = False	)	-> DataLoader:




           """simple docstring"""







           UpperCamelCase__     : Union[str, Any]											=					'''dev''' if mode == '''test''' else mode

           UpperCamelCase__     : Optional[int]											=					self._feature_file(__SCREAMING_SNAKE_CASE	)
           logger.info('''Loading features from cached file %s'''    ,    __SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : Union[str, Any]											=					torch.load(__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : Union[str, Any]											=					torch.tensor([f.input_ids for f in features]    ,    dtype=torch.long	)
           UpperCamelCase__     : List[str]											=					torch.tensor([f.attention_mask for f in features]    ,    dtype=torch.long	)
           UpperCamelCase__     : Optional[int]											=					torch.tensor([f.token_type_ids for f in features]    ,    dtype=torch.long	)
           if self.hparams.glue_output_mode == "classification":
                 UpperCamelCase__     : List[Any]											=					torch.tensor([f.label for f in features]    ,    dtype=torch.long	)
           elif self.hparams.glue_output_mode == "regression":
                 UpperCamelCase__     : Any											=					torch.tensor([f.label for f in features]    ,    dtype=torch.float	)

           return DataLoader(
               TensorDataset(__SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE	)    ,    batch_size=__SCREAMING_SNAKE_CASE    ,    shuffle=__SCREAMING_SNAKE_CASE    ,    )






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE	)	-> List[str]:




           """simple docstring"""







           UpperCamelCase__     : Union[str, Any]											=					{'''input_ids''': batch[0], '''attention_mask''': batch[1], '''labels''': batch[3]}

           if self.config.model_type not in ["distilbert", "bart"]:
                 UpperCamelCase__     : Union[str, Any]											=					batch[2] if self.config.model_type in ['''bert''', '''xlnet''', '''albert'''] else None

           UpperCamelCase__     : List[str]											=					self(**__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__					,UpperCamelCase__     : List[str]											=					outputs[:2]
           UpperCamelCase__     : Optional[int]											=					logits.detach().cpu().numpy()
           UpperCamelCase__     : int											=					inputs['''labels'''].detach().cpu().numpy()

           return {"val_loss": tmp_eval_loss.detach().cpu(), "pred": preds, "target": out_label_ids}






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE	)	-> tuple:




           """simple docstring"""







           UpperCamelCase__     : List[str]											=					torch.stack([x['''val_loss'''] for x in outputs]	).mean().detach().cpu().item()
           UpperCamelCase__     : int											=					np.concatenate([x['''pred'''] for x in outputs]    ,    axis=0	)

           if self.hparams.glue_output_mode == "classification":
                 UpperCamelCase__     : Tuple											=					np.argmax(__SCREAMING_SNAKE_CASE    ,    axis=1	)
           elif self.hparams.glue_output_mode == "regression":
                 UpperCamelCase__     : Dict											=					np.squeeze(__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : int											=					np.concatenate([x['''target'''] for x in outputs]    ,    axis=0	)
           UpperCamelCase__     : Dict											=					[[] for _ in range(out_label_ids.shape[0]	)]
           UpperCamelCase__     : Any											=					[[] for _ in range(out_label_ids.shape[0]	)]

           UpperCamelCase__     : List[Any]											=					{**{'''val_loss''': val_loss_mean}, **compute_metrics(self.hparams.task    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE	)}

           UpperCamelCase__     : Optional[int]											=					dict(results.items()	)
           UpperCamelCase__     : Dict											=					results
           return ret, preds_list, out_label_list






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE	)	-> dict:




           """simple docstring"""







           UpperCamelCase__					,UpperCamelCase__					,UpperCamelCase__     : str											=					self._eval_end(__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : Tuple											=					ret['''log''']
           return {"val_loss": logs["val_loss"], "log": logs, "progress_bar": logs}






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE	)	-> dict:




           """simple docstring"""







           UpperCamelCase__					,UpperCamelCase__					,UpperCamelCase__     : Tuple											=					self._eval_end(__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : Union[str, Any]											=					ret['''log''']
           # `val_loss` is the key returned by `self._eval_end()` but actually refers to `test_loss`
           return {"avg_test_loss": logs["val_loss"], "log": logs, "progress_bar": logs}






     @staticmethod
     def  __SCREAMING_SNAKE_CASE      (				__SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE	)	-> Optional[Any]:




           """simple docstring"""







           BaseTransformer.add_model_specific_args(__SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE	)
           parser.add_argument(
               '''--max_seq_length'''    ,    default=1_2_8    ,    type=__SCREAMING_SNAKE_CASE    ,    help=(
                   '''The maximum total input sequence length after tokenization. Sequences longer '''
                   '''than this will be truncated, sequences shorter will be padded.'''
               )    ,    )

           parser.add_argument(
               '''--task'''    ,    default=''''''    ,    type=__SCREAMING_SNAKE_CASE    ,    required=__SCREAMING_SNAKE_CASE    ,    help='''The GLUE task to run'''    ,    )
           parser.add_argument(
               '''--gpus'''    ,    default=0    ,    type=__SCREAMING_SNAKE_CASE    ,    help='''The number of GPUs allocated for this, it is by default 0 meaning none'''    ,    )

           parser.add_argument(
               '''--overwrite_cache'''    ,    action='''store_true'''    ,    help='''Overwrite the cached training and evaluation sets'''	)

           return parser


def 				SCREAMING_SNAKE_CASE_   (      ):
      UpperCamelCase__     : int											=					argparse.ArgumentParser()
      add_generic_args(UpperCamelCase__			,		os.getcwd()		)
      UpperCamelCase__     : int											=					GLUETransformer.add_model_specific_args(UpperCamelCase__			,		os.getcwd()		)
      UpperCamelCase__     : Any											=					parser.parse_args()

      # If output_dir not provided, a folder will be generated in pwd
      if args.output_dir is None:
            UpperCamelCase__     : Union[str, Any]											=					os.path.join(
                '''./results'''			,		f'''{args.task}_{time.strftime("%Y%m%d_%H%M%S"		)}'''			,		)
            os.makedirs(args.output_dir		)

      UpperCamelCase__     : Tuple											=					GLUETransformer(UpperCamelCase__		)
      UpperCamelCase__     : Dict											=					generic_train(UpperCamelCase__			,		UpperCamelCase__		)

      # Optionally, predict on dev set and write to output_dir
      if args.do_predict:
            UpperCamelCase__     : int											=					sorted(glob.glob(os.path.join(args.output_dir			,		'''checkpoint-epoch=*.ckpt'''		)			,		recursive=UpperCamelCase__		)		)
            UpperCamelCase__     : Optional[Any]											=					model.load_from_checkpoint(checkpoints[-1]		)
            return trainer.test(UpperCamelCase__		)


if __name__ == "__main__":
    main()




import gc
import unittest

import numpy as np
import torch

from diffusers import DanceDiffusionPipeline, IPNDMScheduler, UNetaDModel
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS, UNCONDITIONAL_AUDIO_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()




class 					_lowerCamelCase			(   UpperCamelCase_						,				unittest.TestCase			):





     """simple docstring"""







     SCREAMING_SNAKE_CASE_		 =	DanceDiffusionPipeline
     SCREAMING_SNAKE_CASE_		 =	UNCONDITIONAL_AUDIO_GENERATION_PARAMS
     SCREAMING_SNAKE_CASE_		 =	PipelineTesterMixin.required_optional_params - {
         '''callback''',
         '''latents''',
         '''callback_steps''',
         '''output_type''',
         '''num_images_per_prompt''',
     }
     SCREAMING_SNAKE_CASE_		 =	UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
     SCREAMING_SNAKE_CASE_		 =	False
     SCREAMING_SNAKE_CASE_		 =	False






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Any:




           """simple docstring"""







           torch.manual_seed(0	)
           UpperCamelCase__     : List[str]											=					UNetaDModel(
               block_out_channels=(3_2, 3_2, 6_4)    ,    extra_in_channels=1_6    ,    sample_size=5_1_2    ,    sample_rate=1_6_0_0_0    ,    in_channels=2    ,    out_channels=2    ,    flip_sin_to_cos=__SCREAMING_SNAKE_CASE    ,    use_timestep_embedding=__SCREAMING_SNAKE_CASE    ,    time_embedding_type='''fourier'''    ,    mid_block_type='''UNetMidBlock1D'''    ,    down_block_types=('''DownBlock1DNoSkip''', '''DownBlock1D''', '''AttnDownBlock1D''')    ,    up_block_types=('''AttnUpBlock1D''', '''UpBlock1D''', '''UpBlock1DNoSkip''')    ,    )
           UpperCamelCase__     : Union[str, Any]											=					IPNDMScheduler()

           UpperCamelCase__     : List[str]											=					{
               '''unet''': unet,
               '''scheduler''': scheduler,
           }
           return components






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE=0	)	-> Any:




           """simple docstring"""







           if str(__SCREAMING_SNAKE_CASE	).startswith('''mps'''	):
                 UpperCamelCase__     : Optional[Any]											=					torch.manual_seed(__SCREAMING_SNAKE_CASE	)
           else:
                 UpperCamelCase__     : Optional[Any]											=					torch.Generator(device=__SCREAMING_SNAKE_CASE	).manual_seed(__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : int											=					{
               '''batch_size''': 1,
               '''generator''': generator,
               '''num_inference_steps''': 4,
           }
           return inputs






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> List[Any]:




           """simple docstring"""







           UpperCamelCase__     : str											=					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
           UpperCamelCase__     : Dict											=					self.get_dummy_components()
           UpperCamelCase__     : str											=					DanceDiffusionPipeline(**__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : List[Any]											=					pipe.to(__SCREAMING_SNAKE_CASE	)
           pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : int											=					self.get_dummy_inputs(__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : Dict											=					pipe(**__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : List[str]											=					output.audios

           UpperCamelCase__     : Dict											=					audio[0, -3:, -3:]

           assert audio.shape == (1, 2, components["unet"].sample_size)
           UpperCamelCase__     : List[Any]											=					np.array([-0.7265, 1.0000, -0.8388, 0.1175, 0.9498, -1.0000]	)
           assert np.abs(audio_slice.flatten() - expected_slice	).max() < 1e-2






     @skip_mps
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> List[Any]:




           """simple docstring"""







           return super().test_save_load_local()






     @skip_mps
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Tuple:




           """simple docstring"""







           return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3	)






     @skip_mps
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> str:




           """simple docstring"""







           return super().test_save_load_optional_components()






     @skip_mps
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Optional[int]:




           """simple docstring"""







           return super().test_attention_slicing_forward_pass()






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Tuple:




           """simple docstring"""







           super().test_inference_batch_single_identical(expected_max_diff=3e-3	)




@slow
@require_torch_gpu
class 					_lowerCamelCase			(   unittest.TestCase			):





     """simple docstring"""







     def  __SCREAMING_SNAKE_CASE      (				self	)	-> int:




           """simple docstring"""







           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Union[str, Any]:




           """simple docstring"""







           UpperCamelCase__     : Any											=					torch_device

           UpperCamelCase__     : Any											=					DanceDiffusionPipeline.from_pretrained('''harmonai/maestro-150k'''	)
           UpperCamelCase__     : int											=					pipe.to(__SCREAMING_SNAKE_CASE	)
           pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : int											=					torch.manual_seed(0	)
           UpperCamelCase__     : Optional[Any]											=					pipe(generator=__SCREAMING_SNAKE_CASE    ,    num_inference_steps=1_0_0    ,    audio_length_in_s=4.096	)
           UpperCamelCase__     : str											=					output.audios

           UpperCamelCase__     : List[Any]											=					audio[0, -3:, -3:]

           assert audio.shape == (1, 2, pipe.unet.sample_size)
           UpperCamelCase__     : Tuple											=					np.array([-0.0192, -0.0231, -0.0318, -0.0059, 0.0002, -0.0020]	)

           assert np.abs(audio_slice.flatten() - expected_slice	).max() < 1e-2






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Union[str, Any]:




           """simple docstring"""







           UpperCamelCase__     : Any											=					torch_device

           UpperCamelCase__     : Union[str, Any]											=					DanceDiffusionPipeline.from_pretrained('''harmonai/maestro-150k'''    ,    torch_dtype=torch.floataa	)
           UpperCamelCase__     : Tuple											=					pipe.to(__SCREAMING_SNAKE_CASE	)
           pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : int											=					torch.manual_seed(0	)
           UpperCamelCase__     : Union[str, Any]											=					pipe(generator=__SCREAMING_SNAKE_CASE    ,    num_inference_steps=1_0_0    ,    audio_length_in_s=4.096	)
           UpperCamelCase__     : List[Any]											=					output.audios

           UpperCamelCase__     : List[Any]											=					audio[0, -3:, -3:]

           assert audio.shape == (1, 2, pipe.unet.sample_size)
           UpperCamelCase__     : Optional[Any]											=					np.array([-0.0367, -0.0488, -0.0771, -0.0525, -0.0444, -0.0341]	)

           assert np.abs(audio_slice.flatten() - expected_slice	).max() < 1e-2


