

"""simple docstring"""






import logging
import os
from dataclasses import dataclass
from typing import List, Optional, Union

import tqdm
from filelock import FileLock

from transformers import (
    BartTokenizer,
    BartTokenizerFast,
    DataProcessor,
    PreTrainedTokenizer,
    RobertaTokenizer,
    RobertaTokenizerFast,
    XLMRobertaTokenizer,
    is_tf_available,
    is_torch_available,
)


_lowerCAmelCase						=		logging.getLogger(__name__)







@dataclass(frozen=UpperCAmelCase__						)
class 	__UpperCamelCase    :
  _UpperCAmelCase          =		42
  _UpperCAmelCase          =		42
  _UpperCAmelCase          =		None
  _UpperCAmelCase          =		None
  _UpperCAmelCase          =		None







@dataclass(frozen=UpperCAmelCase__						)
class 	__UpperCamelCase    :
  _UpperCAmelCase          =		42
  _UpperCAmelCase          =		None
  _UpperCAmelCase          =		None
  _UpperCAmelCase          =		None
  _UpperCAmelCase          =		None


if is_torch_available():
 import torch
 from torch.utils.data import Dataset







 class 	__UpperCamelCase    (    UpperCAmelCase__						):
   _UpperCAmelCase          =		42

   def __init__(		self    ,_A    ,_A    ,_A    ,_A = None    ,_A=False    ,_A = False    ,):
       '''simple docstring'''

       _lowerCAmelCase       :   str  =    hans_processors[task]()

       _lowerCAmelCase       :   List[Any]  =    os.path.join(
           _A    ,'cached_{}_{}_{}_{}'.format(
               'dev' if evaluate else 'train'    ,tokenizer.__class__.__name__    ,str(_A      )    ,_A    ,)    ,)
       _lowerCAmelCase       :   List[str]  =    processor.get_labels()
       if tokenizer.__class__ in (
           RobertaTokenizer,
           RobertaTokenizerFast,
           XLMRobertaTokenizer,
           BartTokenizer,
           BartTokenizerFast,
       ):
           # HACK(label indices are swapped in RoBERTa pretrained model)
           _lowerCAmelCase       :   Optional[Any]  =    label_list[2], label_list[1]
       _lowerCAmelCase       :   List[str]  =    label_list

       # Make sure only the first process in distributed training processes the dataset,
       # and the others will use the cache.
       _lowerCAmelCase       :   List[Any]  =    cached_features_file + """.lock"""
       with FileLock(_A      ):
           if os.path.exists(_A      ) and not overwrite_cache:
               logger.info(F"""Loading features from cached file {cached_features_file}"""      )
               _lowerCAmelCase       :   Any  =    torch.load(_A      )
           else:
               logger.info(F"""Creating features from dataset file at {data_dir}"""      )

               _lowerCAmelCase       :   Optional[Any]  =    (
                   processor.get_dev_examples(_A      ) if evaluate else processor.get_train_examples(_A      )
               )

               logger.info('Training examples: %s'    ,len(_A      )      )
               _lowerCAmelCase       :   List[Any]  =    hans_convert_examples_to_features(_A    ,_A    ,_A    ,_A      )
               logger.info('Saving features into cached file %s'    ,_A      )
               torch.save(self.features    ,_A      )

   def __len__(		self      ):
       '''simple docstring'''

       return len(self.features      )

   def __getitem__(		self    ,_A      ):
       '''simple docstring'''

       return self.features[i]







   def __lowerCamelCase		(		self      ):
       '''simple docstring'''

       return self.label_list


if is_tf_available():
 import tensorflow as tf







 class 	__UpperCamelCase    :
   _UpperCAmelCase          =		42

   def __init__(		self    ,_A    ,_A    ,_A    ,_A = 128    ,_A=False    ,_A = False    ,):
       '''simple docstring'''

       _lowerCAmelCase       :   Dict  =    hans_processors[task]()
       _lowerCAmelCase       :   int  =    processor.get_labels()
       if tokenizer.__class__ in (
           RobertaTokenizer,
           RobertaTokenizerFast,
           XLMRobertaTokenizer,
           BartTokenizer,
           BartTokenizerFast,
       ):
           # HACK(label indices are swapped in RoBERTa pretrained model)
           _lowerCAmelCase       :   List[str]  =    label_list[2], label_list[1]
       _lowerCAmelCase       :   Optional[Any]  =    label_list

       _lowerCAmelCase       :   Tuple  =    processor.get_dev_examples(_A      ) if evaluate else processor.get_train_examples(_A      )
       _lowerCAmelCase       :   Any  =    hans_convert_examples_to_features(_A    ,_A    ,_A    ,_A      )

       def gen():
           for ex_index, ex in tqdm.tqdm(enumerate(self.features      )    ,desc='convert examples to features'      ):
               if ex_index % 1_0000 == 0:
                   logger.info('Writing example %d of %d' % (ex_index, len(_A      ))      )

               yield (
                   {
                       "example_id": 0,
                       "input_ids": ex.input_ids,
                       "attention_mask": ex.attention_mask,
                       "token_type_ids": ex.token_type_ids,
                   },
                   ex.label,
               )

       _lowerCAmelCase       :   int  =    tf.data.Dataset.from_generator(
           _A    ,(
               {
                   'example_id': tf.intaa,
                   'input_ids': tf.intaa,
                   'attention_mask': tf.intaa,
                   'token_type_ids': tf.intaa,
               },
               tf.intaa,
           )    ,(
               {
                   'example_id': tf.TensorShape([]      ),
                   'input_ids': tf.TensorShape([None, None]      ),
                   'attention_mask': tf.TensorShape([None, None]      ),
                   'token_type_ids': tf.TensorShape([None, None]      ),
               },
               tf.TensorShape([]      ),
           )    ,)

   def __lowerCamelCase		(		self      ):
       '''simple docstring'''

       return self.dataset

   def __len__(		self      ):
       '''simple docstring'''

       return len(self.features      )

   def __getitem__(		self    ,_A      ):
       '''simple docstring'''

       return self.features[i]







   def __lowerCamelCase		(		self      ):
       '''simple docstring'''

       return self.label_list







class 	__UpperCamelCase    (    UpperCAmelCase__						):

  def __lowerCamelCase		(		self    ,_A      ):
      '''simple docstring'''

      return self._create_examples(self._read_tsv(os.path.join(_A    ,'heuristics_train_set.txt'      )      )    ,'train'      )

  def __lowerCamelCase		(		self    ,_A      ):
      '''simple docstring'''

      return self._create_examples(self._read_tsv(os.path.join(_A    ,'heuristics_evaluation_set.txt'      )      )    ,'dev'      )

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      return ["contradiction", "entailment", "neutral"]







  def __lowerCamelCase		(		self    ,_A    ,_A      ):
      '''simple docstring'''

      _lowerCAmelCase       :   Union[str, Any]  =    []
      for i, line in enumerate(_A      ):
          if i == 0:
              continue
          _lowerCAmelCase       :   Dict  =    """%s-%s""" % (set_type, line[0])
          _lowerCAmelCase       :   List[str]  =    line[5]
          _lowerCAmelCase       :   List[Any]  =    line[6]
          _lowerCAmelCase       :   Tuple  =    line[7][2:] if line[7].startswith('ex'      ) else line[7]
          _lowerCAmelCase       :   Union[str, Any]  =    line[0]
          examples.append(InputExample(guid=_A    ,text_a=_A    ,text_b=_A    ,label=_A    ,pairID=_A      )      )
      return examples





def 		lowerCamelCase__		(					_lowerCamelCase ,   _lowerCamelCase ,   _lowerCamelCase ,   _lowerCamelCase ,   ):

    '''simple docstring'''



    _lowerCAmelCase       :   str  =    {label: i for i, label in enumerate(_lowerCamelCase   )}

    _lowerCAmelCase       :   List[Any]  =    []
    for ex_index, example in tqdm.tqdm(enumerate(_lowerCamelCase   ) ,   desc='convert examples to features'   ):
        if ex_index % 10000 == 0:
            logger.info('Writing example %d' % (ex_index)   )

        _lowerCAmelCase       :   List[Any]  =    tokenizer(
            example.text_a ,   example.text_b ,   add_special_tokens=_lowerCamelCase ,   max_length=_lowerCamelCase ,   padding='max_length' ,   truncation=_lowerCamelCase ,   return_overflowing_tokens=_lowerCamelCase ,   )

        _lowerCAmelCase       :   Union[str, Any]  =    label_map[example.label] if example.label in label_map else 0

        _lowerCAmelCase       :   str  =    int(example.pairID   )

        features.append(InputFeatures(**_lowerCamelCase ,   label=_lowerCamelCase ,   pairID=_lowerCamelCase   )   )

    for i, example in enumerate(examples[:5]   ):
        logger.info('*** Example ***'   )
        logger.info(f"""guid: {example}"""   )
        logger.info(f"""features: {features[i]}"""   )

    return features


_lowerCAmelCase						=		{
    """hans""": 3,
}

_lowerCAmelCase						=		{
    """hans""": HansProcessor,
}




"""simple docstring"""






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 	__UpperCamelCase    (    a__							,							unittest.TestCase						):
  _UpperCAmelCase          =		DanceDiffusionPipeline
  _UpperCAmelCase          =		UNCONDITIONAL_AUDIO_GENERATION_PARAMS
  _UpperCAmelCase          =		PipelineTesterMixin.required_optional_params - {
      "callback",
      "latents",
      "callback_steps",
      "output_type",
      "num_images_per_prompt",
  }
  _UpperCAmelCase          =		UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
  _UpperCAmelCase          =		False
  _UpperCAmelCase          =		False

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      torch.manual_seed(0      )
      _lowerCAmelCase       :   List[Any]  =    UNetaDModel(
          block_out_channels=(32, 32, 64)    ,extra_in_channels=16    ,sample_size=512    ,sample_rate=1_6000    ,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')    ,)
      _lowerCAmelCase       :   int  =    IPNDMScheduler()

      _lowerCAmelCase       :   Union[str, Any]  =    {
          'unet': unet,
          'scheduler': scheduler,
      }
      return components

  def __lowerCamelCase		(		self    ,_A    ,_A=0      ):
      '''simple docstring'''

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

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

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

      _lowerCAmelCase       :   Union[str, Any]  =    self.get_dummy_inputs(_A      )
      _lowerCAmelCase       :   List[str]  =    pipe(**_A      )
      _lowerCAmelCase       :   List[Any]  =    output.audios

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

      assert audio.shape == (1, 2, components["unet"].sample_size)
      _lowerCAmelCase       :   Optional[Any]  =    np.array([-0.7_2_6_5, 1.0_0_0_0, -0.8_3_8_8, 0.1_1_7_5, 0.9_4_9_8, -1.0_0_0_0]      )
      assert np.abs(audio_slice.flatten() - expected_slice      ).max() < 1E-2

  @skip_mps
  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      return super().test_save_load_local()

  @skip_mps
  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

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

  @skip_mps
  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      return super().test_save_load_optional_components()

  @skip_mps
  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      return super().test_attention_slicing_forward_pass()







  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

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







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

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

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

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      _lowerCAmelCase       :   List[Any]  =    torch_device

      _lowerCAmelCase       :   int  =    DanceDiffusionPipeline.from_pretrained('harmonai/maestro-150k'      )
      _lowerCAmelCase       :   int  =    pipe.to(_A      )
      pipe.set_progress_bar_config(disable=_A      )

      _lowerCAmelCase       :   Optional[int]  =    torch.manual_seed(0      )
      _lowerCAmelCase       :   str  =    pipe(generator=_A    ,num_inference_steps=100    ,audio_length_in_s=4.0_9_6      )
      _lowerCAmelCase       :   str  =    output.audios

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

      assert audio.shape == (1, 2, pipe.unet.sample_size)
      _lowerCAmelCase       :   Union[str, Any]  =    np.array([-0.0_1_9_2, -0.0_2_3_1, -0.0_3_1_8, -0.0_0_5_9, 0.0_0_0_2, -0.0_0_2_0]      )

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







  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      _lowerCAmelCase       :   str  =    torch_device

      _lowerCAmelCase       :   Tuple  =    DanceDiffusionPipeline.from_pretrained('harmonai/maestro-150k'    ,torch_dtype=torch.floataa      )
      _lowerCAmelCase       :   Optional[int]  =    pipe.to(_A      )
      pipe.set_progress_bar_config(disable=_A      )

      _lowerCAmelCase       :   Union[str, Any]  =    torch.manual_seed(0      )
      _lowerCAmelCase       :   Optional[int]  =    pipe(generator=_A    ,num_inference_steps=100    ,audio_length_in_s=4.0_9_6      )
      _lowerCAmelCase       :   Union[str, Any]  =    output.audios

      _lowerCAmelCase       :   int  =    audio[0, -3:, -3:]

      assert audio.shape == (1, 2, pipe.unet.sample_size)
      _lowerCAmelCase       :   List[str]  =    np.array([-0.0_3_6_7, -0.0_4_8_8, -0.0_7_7_1, -0.0_5_2_5, -0.0_4_4_4, -0.0_3_4_1]      )

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





