


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 			SCREAMING_SNAKE_CASE__  (    lowercase__						,   unittest.TestCase      ):
      snake_case__    :					int					       =						DanceDiffusionPipeline
      snake_case__    :					Optional[int]					       =						UNCONDITIONAL_AUDIO_GENERATION_PARAMS
      snake_case__    :					Tuple					       =						PipelineTesterMixin.required_optional_params - {
          '''callback''',
          '''latents''',
          '''callback_steps''',
          '''output_type''',
          '''num_images_per_prompt''',
      }
      snake_case__    :					Optional[int]					       =						UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
      snake_case__    :					List[Any]					       =						False
      snake_case__    :					Dict					       =						False
      def 	SCREAMING_SNAKE_CASE							(	self    :	int					)		->      int:
             torch.manual_seed(0					)
             a_ :		Dict							=  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=_a					,							use_timestep_embedding=_a					,							time_embedding_type='fourier'					,							mid_block_type='UNetMidBlock1D'					,							down_block_types=('DownBlock1DNoSkip', 'DownBlock1D', 'AttnDownBlock1D')					,							up_block_types=('AttnUpBlock1D', 'UpBlock1D', 'UpBlock1DNoSkip')					,							)
             a_ :		List[Any]							=  IPNDMScheduler()

             a_ :		str							=  {
                 'unet': unet,
                 'scheduler': scheduler,
             }
             return components
      def 	SCREAMING_SNAKE_CASE							(	self    :	Union[str, Any]					,							SCREAMING_SNAKE_CASE__    :	Dict					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]=0					)		->      Any:






             if str(_a					).startswith('mps'					):
                    a_ :		List[Any]							=  torch.manual_seed(_a					)
             else:
                    a_ :		Any							=  torch.Generator(device=_a					).manual_seed(_a					)
             a_ :		str							=  {
                 'batch_size': 1,
                 'generator': generator,
                 'num_inference_steps': 4,
             }
             return inputs
      def 	SCREAMING_SNAKE_CASE							(	self    :	Tuple					)		->      Optional[int]:
             a_ :		int							=  'cpu'  # ensure determinism for the device-dependent torch.Generator
             a_ :		Dict							=  self.get_dummy_components()
             a_ :		Any							=  DanceDiffusionPipeline(**_a					)
             a_ :		List[str]							=  pipe.to(_a					)
             pipe.set_progress_bar_config(disable=_a					)

             a_ :		Dict							=  self.get_dummy_inputs(_a					)
             a_ :		List[Any]							=  pipe(**_a					)
             a_ :		Tuple							=  output.audios

             a_ :		List[str]							=  audio[0, -3:, -3:]

             assert audio.shape == (1, 2, components["unet"].sample_size)
             a_ :		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    :	Union[str, Any]					)		->      Any:
             return super().test_save_load_local()
      @skip_mps
      def 	SCREAMING_SNAKE_CASE							(	self    :	List[str]					)		->      int:
             return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3E-3					)
      @skip_mps
      def 	SCREAMING_SNAKE_CASE							(	self    :	Optional[int]					)		->      Union[str, Any]:
             return super().test_save_load_optional_components()
      @skip_mps
      def 	SCREAMING_SNAKE_CASE							(	self    :	Union[str, Any]					)		->      Union[str, Any]:
             return super().test_attention_slicing_forward_pass()

      def 	SCREAMING_SNAKE_CASE							(	self    :	Tuple					)		->      Union[str, Any]:
             super().test_inference_batch_single_identical(expected_max_diff=3E-3					)
@slow
@require_torch_gpu
class 			SCREAMING_SNAKE_CASE__  (    unittest.TestCase      ):
      def 	SCREAMING_SNAKE_CASE							(	self    :	Tuple					)		->      Union[str, Any]:
             super().tearDown()
             gc.collect()
             torch.cuda.empty_cache()
      def 	SCREAMING_SNAKE_CASE							(	self    :	List[str]					)		->      Optional[Any]:
             a_ :		Tuple							=  torch_device

             a_ :		str							=  DanceDiffusionPipeline.from_pretrained('harmonai/maestro-150k'					)
             a_ :		Optional[Any]							=  pipe.to(_a					)
             pipe.set_progress_bar_config(disable=_a					)

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

             a_ :		Optional[Any]							=  audio[0, -3:, -3:]

             assert audio.shape == (1, 2, pipe.unet.sample_size)
             a_ :		Any							=  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    :	List[str]					)		->      List[Any]:
             a_ :		List[str]							=  torch_device

             a_ :		Union[str, Any]							=  DanceDiffusionPipeline.from_pretrained('harmonai/maestro-150k'					,							torch_dtype=torch.floataa					)
             a_ :		Optional[int]							=  pipe.to(_a					)
             pipe.set_progress_bar_config(disable=_a					)

             a_ :		List[Any]							=  torch.manual_seed(0					)
             a_ :		Optional[int]							=  pipe(generator=_a					,							num_inference_steps=1_0_0					,							audio_length_in_s=4.096					)
             a_ :		Tuple							=  output.audios

             a_ :		Union[str, Any]							=  audio[0, -3:, -3:]

             assert audio.shape == (1, 2, pipe.unet.sample_size)
             a_ :		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






import logging
from pathlib import Path

import numpy as np
import pytorch_lightning as pl
import torch
from pytorch_lightning.callbacks import EarlyStopping, ModelCheckpoint
from pytorch_lightning.utilities import rank_zero_only
from utils_rag import save_json







def 	SCREAMING_SNAKE_CASE_						(      __A			:      Any    )     ->		Optional[int]:

       """simple docstring"""

       a_ :		str							=  filter(lambda __A    : p.requires_grad  ,   model.parameters()    )
       a_ :		List[Any]							=  sum([np.prod(p.size()    ) for p in model_parameters]    )
       return params


UpperCAmelCase_				:			Dict   =	logging.getLogger(__name__)







def 	SCREAMING_SNAKE_CASE_						(      __A			:      int  ,   __A			:      int    )     ->		List[str]:

       """simple docstring"""

       if metric == "rouge2":
              a_ :		Dict							=  '{val_avg_rouge2:.4f}-{step_count}'
       elif metric == "bleu":
              a_ :		Dict							=  '{val_avg_bleu:.4f}-{step_count}'
       elif metric == "em":
              a_ :		Union[str, Any]							=  '{val_avg_em:.4f}-{step_count}'
       elif metric == "loss":
              a_ :		str							=  '{val_avg_loss:.4f}-{step_count}'
       else:
              raise NotImplementedError(
                  F"""seq2seq callbacks only support rouge2 and bleu, got {metric}, You can make your own by adding to this"""
                  ' function.'    )

       a_ :		Dict							=  ModelCheckpoint(
           dirpath=__A  ,   filename=__A  ,   monitor=F"""val_{metric}"""  ,   mode='max'  ,   save_top_k=1  ,   every_n_epochs=1  ,   )

       return checkpoint_callback







def 	SCREAMING_SNAKE_CASE_						(      __A			:      int  ,   __A			:      List[Any]    )     ->		int:

       """simple docstring"""

       return EarlyStopping(
           monitor=F"""val_{metric}"""  ,   mode='min' if 'loss' in metric else 'max'  ,   patience=__A  ,   verbose=__A  ,   )
class 			SCREAMING_SNAKE_CASE__  (    pl.Callback      ):
      def 	SCREAMING_SNAKE_CASE							(	self    :	Dict					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]					,							SCREAMING_SNAKE_CASE__    :	List[str]					)		->      Any:
             a_ :		int							=  {F"""lr_group_{i}""": param['lr'] for i, param in enumerate(pl_module.trainer.optimizers[0].param_groups					)}
             pl_module.logger.log_metrics(SCREAMING_SNAKE_CASE__					)
      @rank_zero_only
      def 	SCREAMING_SNAKE_CASE							(	self    :	List[str]					,							SCREAMING_SNAKE_CASE__    :	pl.Trainer					,							SCREAMING_SNAKE_CASE__    :	pl.LightningModule					,							SCREAMING_SNAKE_CASE__    :	str					,							SCREAMING_SNAKE_CASE__    :	List[str]=True					)		->      None:
             logger.info(F"""***** {type_path} results at step {trainer.global_step:05d} *****"""					)
             a_ :		List[str]							=  trainer.callback_metrics
             trainer.logger.log_metrics({k: v for k, v in metrics.items() if k not in ['log', 'progress_bar', 'preds']}					)
             # Log results
             a_ :		Optional[int]							=  Path(pl_module.hparams.output_dir					)
             if type_path == "test":
                    a_ :		Tuple							=  od / 'test_results.txt'
                    a_ :		Optional[int]							=  od / 'test_generations.txt'
             else:
                    # this never gets hit. I prefer not to save intermediate generations, and results are in metrics.json
                    # If people want this it will be easy enough to add back.
                    a_ :		str							=  od / F"""{type_path}_results/{trainer.global_step:05d}.txt"""
                    a_ :		Optional[Any]							=  od / F"""{type_path}_generations/{trainer.global_step:05d}.txt"""
                    results_file.parent.mkdir(exist_ok=SCREAMING_SNAKE_CASE__					)
                    generations_file.parent.mkdir(exist_ok=SCREAMING_SNAKE_CASE__					)
             with open(SCREAMING_SNAKE_CASE__					,							'a+'					) as writer:
                    for key in sorted(SCREAMING_SNAKE_CASE__					):
                           if key in ["log", "progress_bar", "preds"]:
                                  continue
                           a_ :		int							=  metrics[key]
                           if isinstance(SCREAMING_SNAKE_CASE__					,							torch.Tensor					):
                                  a_ :		List[str]							=  val.item()
                           a_ :		int							=  F"""{key}: {val:.6f}\n"""
                           writer.write(SCREAMING_SNAKE_CASE__					)

             if not save_generations:
                    return

             if "preds" in metrics:
                    a_ :		Optional[Any]							=  '\n'.join(metrics['preds']					)
                    generations_file.open('w+'					).write(SCREAMING_SNAKE_CASE__					)
      @rank_zero_only
      def 	SCREAMING_SNAKE_CASE							(	self    :	Any					,							SCREAMING_SNAKE_CASE__    :	Dict					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]					)		->      Dict:
             try:
                    a_ :		Any							=  pl_module.model.model.num_parameters()
             except AttributeError:
                    a_ :		List[str]							=  pl_module.model.num_parameters()

             a_ :		Any							=  count_trainable_parameters(SCREAMING_SNAKE_CASE__					)
             # mp stands for million parameters
             trainer.logger.log_metrics({'n_params': npars, 'mp': npars / 1E6, 'grad_mp': n_trainable_pars / 1E6}					)
      @rank_zero_only
      def 	SCREAMING_SNAKE_CASE							(	self    :	Union[str, Any]					,							SCREAMING_SNAKE_CASE__    :	pl.Trainer					,							SCREAMING_SNAKE_CASE__    :	pl.LightningModule					)		->      Optional[int]:
             save_json(pl_module.metrics					,							pl_module.metrics_save_path					)
             return self._write_logs(SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					,							'test'					)

      @rank_zero_only
      def 	SCREAMING_SNAKE_CASE							(	self    :	Any					,							SCREAMING_SNAKE_CASE__    :	pl.Trainer					,							SCREAMING_SNAKE_CASE__    :	Optional[Any]					)		->      Optional[Any]:
             save_json(pl_module.metrics					,							pl_module.metrics_save_path					)
             # Uncommenting this will save val generations
             # return self._write_logs(trainer, pl_module, "valid")


