


'''simple docstring'''

import gc
import unittest

import numpy as np
import torch
import torch.nn.functional as F
from transformers import (
    ClapTextConfig,
    ClapTextModelWithProjection,
    RobertaTokenizer,
    SpeechTaHifiGan,
    SpeechTaHifiGanConfig,
)

from diffusers import (
    AudioLDMPipeline,
    AutoencoderKL,
    DDIMScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.utils import is_xformers_available, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from ..pipeline_params import TEXT_TO_AUDIO_BATCH_PARAMS, TEXT_TO_AUDIO_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()





class      snake_case    (    a__			,						unittest.TestCase			):





  """simple docstring"""




  _lowerCamelCase		    =  AudioLDMPipeline
  _lowerCamelCase		    =  TEXT_TO_AUDIO_PARAMS
  _lowerCamelCase		    =  TEXT_TO_AUDIO_BATCH_PARAMS
  _lowerCamelCase		    =  frozenset(
      [
          "num_inference_steps",
          "num_waveforms_per_prompt",
          "generator",
          "latents",
          "output_type",
          "return_dict",
          "callback",
          "callback_steps",
      ]			)






  def   snake_case		(    self     ):







   """simple docstring"""
   torch.manual_seed(0     )
   lowerCamelCase_      =  UNetaDConditionModel(
       block_out_channels=(32, 64)			,  layers_per_block=2			,  sample_size=32			,  in_channels=4			,  out_channels=4			,  down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")			,  up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")			,  cross_attention_dim=(32, 64)			,  class_embed_type="simple_projection"			,  projection_class_embeddings_input_dim=32			,  class_embeddings_concat=_UpperCAmelCase			,  )
   lowerCamelCase_      =  DDIMScheduler(
       beta_start=0.00_085			,  beta_end=0.012			,  beta_schedule="scaled_linear"			,  clip_sample=_UpperCAmelCase			,  set_alpha_to_one=_UpperCAmelCase			,  )
   torch.manual_seed(0     )
   lowerCamelCase_      =  AutoencoderKL(
       block_out_channels=[32, 64]			,  in_channels=1			,  out_channels=1			,  down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]			,  up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]			,  latent_channels=4			,  )
   torch.manual_seed(0     )
   lowerCamelCase_      =  ClapTextConfig(
       bos_token_id=0			,  eos_token_id=2			,  hidden_size=32			,  intermediate_size=37			,  layer_norm_eps=1e-05			,  num_attention_heads=4			,  num_hidden_layers=5			,  pad_token_id=1			,  vocab_size=1000			,  projection_dim=32			,  )
   lowerCamelCase_      =  ClapTextModelWithProjection(_UpperCAmelCase     )
   lowerCamelCase_      =  RobertaTokenizer.from_pretrained("hf-internal-testing/tiny-random-roberta"			,  model_max_length=77     )

   lowerCamelCase_      =  SpeechTaHifiGanConfig(
       model_in_dim=8			,  sampling_rate=1_6000			,  upsample_initial_channel=16			,  upsample_rates=[2, 2]			,  upsample_kernel_sizes=[4, 4]			,  resblock_kernel_sizes=[3, 7]			,  resblock_dilation_sizes=[[1, 3, 5], [1, 3, 5]]			,  normalize_before=_UpperCAmelCase			,  )

   lowerCamelCase_      =  SpeechTaHifiGan(_UpperCAmelCase     )

   lowerCamelCase_      =  {
       'unet': unet,
       'scheduler': scheduler,
       'vae': vae,
       'text_encoder': text_encoder,
       'tokenizer': tokenizer,
       'vocoder': vocoder,
   }
   return components






  def   snake_case		(    self			,  UpperCamelCase			,  UpperCamelCase=0     ):







   """simple docstring"""






   if str(_UpperCAmelCase     ).startswith("mps"     ):
    lowerCamelCase_      =  torch.manual_seed(_UpperCAmelCase     )
   else:
    lowerCamelCase_      =  torch.Generator(device=_UpperCAmelCase     ).manual_seed(_UpperCAmelCase     )
   lowerCamelCase_      =  {
       'prompt': 'A hammer hitting a wooden surface',
       'generator': generator,
       'num_inference_steps': 2,
       'guidance_scale': 6.0,
   }
   return inputs






  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  'cpu'  # ensure determinism for the device-dependent torch.Generator

   lowerCamelCase_      =  self.get_dummy_components()
   lowerCamelCase_      =  AudioLDMPipeline(**_UpperCAmelCase     )
   lowerCamelCase_      =  audioldm_pipe.to(_UpperCAmelCase     )
   audioldm_pipe.set_progress_bar_config(disable=_UpperCAmelCase     )

   lowerCamelCase_      =  self.get_dummy_inputs(_UpperCAmelCase     )
   lowerCamelCase_      =  audioldm_pipe(**_UpperCAmelCase     )
   lowerCamelCase_      =  output.audios[0]

   assert audio.ndim == 1
   assert len(_UpperCAmelCase     ) == 256

   lowerCamelCase_      =  audio[:10]
   lowerCamelCase_      =  np.array(
       [-0.0_050, 0.0_050, -0.0_060, 0.0_033, -0.0_026, 0.0_033, -0.0_027, 0.0_033, -0.0_028, 0.0_033]     )

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






  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  self.get_dummy_components()
   lowerCamelCase_      =  AudioLDMPipeline(**_UpperCAmelCase     )
   lowerCamelCase_      =  audioldm_pipe.to(_UpperCAmelCase     )
   lowerCamelCase_      =  audioldm_pipe.to(_UpperCAmelCase     )
   audioldm_pipe.set_progress_bar_config(disable=_UpperCAmelCase     )

   lowerCamelCase_      =  self.get_dummy_inputs(_UpperCAmelCase     )
   lowerCamelCase_      =  3 * [inputs['prompt']]

   # forward
   lowerCamelCase_      =  audioldm_pipe(**_UpperCAmelCase     )
   lowerCamelCase_      =  output.audios[0]

   lowerCamelCase_      =  self.get_dummy_inputs(_UpperCAmelCase     )
   lowerCamelCase_      =  3 * [inputs.pop("prompt"     )]

   lowerCamelCase_      =  audioldm_pipe.tokenizer(
       _UpperCAmelCase			,  padding="max_length"			,  max_length=audioldm_pipe.tokenizer.model_max_length			,  truncation=_UpperCAmelCase			,  return_tensors="pt"			,  )
   lowerCamelCase_      =  text_inputs['input_ids'].to(_UpperCAmelCase     )

   lowerCamelCase_      =  audioldm_pipe.text_encoder(
       _UpperCAmelCase			,  )
   lowerCamelCase_      =  prompt_embeds.text_embeds
   # additional L_2 normalization over each hidden-state
   lowerCamelCase_      =  F.normalize(_UpperCAmelCase			,  dim=-1     )

   lowerCamelCase_      =  prompt_embeds

   # forward
   lowerCamelCase_      =  audioldm_pipe(**_UpperCAmelCase     )
   lowerCamelCase_      =  output.audios[0]

   assert np.abs(audio_a - audio_a     ).max() < 1e-2






  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  self.get_dummy_components()
   lowerCamelCase_      =  AudioLDMPipeline(**_UpperCAmelCase     )
   lowerCamelCase_      =  audioldm_pipe.to(_UpperCAmelCase     )
   lowerCamelCase_      =  audioldm_pipe.to(_UpperCAmelCase     )
   audioldm_pipe.set_progress_bar_config(disable=_UpperCAmelCase     )

   lowerCamelCase_      =  self.get_dummy_inputs(_UpperCAmelCase     )
   lowerCamelCase_      =  3 * ['this is a negative prompt']
   lowerCamelCase_      =  negative_prompt
   lowerCamelCase_      =  3 * [inputs['prompt']]

   # forward
   lowerCamelCase_      =  audioldm_pipe(**_UpperCAmelCase     )
   lowerCamelCase_      =  output.audios[0]

   lowerCamelCase_      =  self.get_dummy_inputs(_UpperCAmelCase     )
   lowerCamelCase_      =  3 * [inputs.pop("prompt"     )]

   lowerCamelCase_      =  []
   for p in [prompt, negative_prompt]:
    lowerCamelCase_      =  audioldm_pipe.tokenizer(
        _UpperCAmelCase			,  padding="max_length"			,  max_length=audioldm_pipe.tokenizer.model_max_length			,  truncation=_UpperCAmelCase			,  return_tensors="pt"			,  )
    lowerCamelCase_      =  text_inputs['input_ids'].to(_UpperCAmelCase     )

    lowerCamelCase_      =  audioldm_pipe.text_encoder(
        _UpperCAmelCase			,  )
    lowerCamelCase_      =  text_embeds.text_embeds
    # additional L_2 normalization over each hidden-state
    lowerCamelCase_      =  F.normalize(_UpperCAmelCase			,  dim=-1     )

    embeds.append(_UpperCAmelCase     )

   lowerCamelCase_      =  embeds

   # forward
   lowerCamelCase_      =  audioldm_pipe(**_UpperCAmelCase     )
   lowerCamelCase_      =  output.audios[0]

   assert np.abs(audio_a - audio_a     ).max() < 1e-2






  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  'cpu'  # ensure determinism for the device-dependent torch.Generator
   lowerCamelCase_      =  self.get_dummy_components()
   lowerCamelCase_      =  PNDMScheduler(skip_prk_steps=_UpperCAmelCase     )
   lowerCamelCase_      =  AudioLDMPipeline(**_UpperCAmelCase     )
   lowerCamelCase_      =  audioldm_pipe.to(_UpperCAmelCase     )
   audioldm_pipe.set_progress_bar_config(disable=_UpperCAmelCase     )

   lowerCamelCase_      =  self.get_dummy_inputs(_UpperCAmelCase     )
   lowerCamelCase_      =  'egg cracking'
   lowerCamelCase_      =  audioldm_pipe(**_UpperCAmelCase			,  negative_prompt=_UpperCAmelCase     )
   lowerCamelCase_      =  output.audios[0]

   assert audio.ndim == 1
   assert len(_UpperCAmelCase     ) == 256

   lowerCamelCase_      =  audio[:10]
   lowerCamelCase_      =  np.array(
       [-0.0_051, 0.0_050, -0.0_060, 0.0_034, -0.0_026, 0.0_033, -0.0_027, 0.0_033, -0.0_028, 0.0_032]     )

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






  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  'cpu'  # ensure determinism for the device-dependent torch.Generator
   lowerCamelCase_      =  self.get_dummy_components()
   lowerCamelCase_      =  PNDMScheduler(skip_prk_steps=_UpperCAmelCase     )
   lowerCamelCase_      =  AudioLDMPipeline(**_UpperCAmelCase     )
   lowerCamelCase_      =  audioldm_pipe.to(_UpperCAmelCase     )
   audioldm_pipe.set_progress_bar_config(disable=_UpperCAmelCase     )

   lowerCamelCase_      =  'A hammer hitting a wooden surface'

   # test num_waveforms_per_prompt=1 (default)
   lowerCamelCase_      =  audioldm_pipe(_UpperCAmelCase			,  num_inference_steps=2     ).audios

   assert audios.shape == (1, 256)

   # test num_waveforms_per_prompt=1 (default) for batch of prompts
   lowerCamelCase_      =  2
   lowerCamelCase_      =  audioldm_pipe([prompt] * batch_size			,  num_inference_steps=2     ).audios

   assert audios.shape == (batch_size, 256)

   # test num_waveforms_per_prompt for single prompt
   lowerCamelCase_      =  2
   lowerCamelCase_      =  audioldm_pipe(_UpperCAmelCase			,  num_inference_steps=2			,  num_waveforms_per_prompt=_UpperCAmelCase     ).audios

   assert audios.shape == (num_waveforms_per_prompt, 256)

   # test num_waveforms_per_prompt for batch of prompts
   lowerCamelCase_      =  2
   lowerCamelCase_      =  audioldm_pipe(
       [prompt] * batch_size			,  num_inference_steps=2			,  num_waveforms_per_prompt=_UpperCAmelCase     ).audios

   assert audios.shape == (batch_size * num_waveforms_per_prompt, 256)






  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  'cpu'  # ensure determinism for the device-dependent torch.Generator
   lowerCamelCase_      =  self.get_dummy_components()
   lowerCamelCase_      =  AudioLDMPipeline(**_UpperCAmelCase     )
   lowerCamelCase_      =  audioldm_pipe.to(_UpperCAmelCase     )
   audioldm_pipe.set_progress_bar_config(disable=_UpperCAmelCase     )
   lowerCamelCase_      =  audioldm_pipe.vocoder.config.sampling_rate

   lowerCamelCase_      =  self.get_dummy_inputs(_UpperCAmelCase     )
   lowerCamelCase_      =  audioldm_pipe(audio_length_in_s=0.016			,  **_UpperCAmelCase     )
   lowerCamelCase_      =  output.audios[0]

   assert audio.ndim == 1
   assert len(_UpperCAmelCase     ) / vocoder_sampling_rate == 0.016

   lowerCamelCase_      =  audioldm_pipe(audio_length_in_s=0.032			,  **_UpperCAmelCase     )
   lowerCamelCase_      =  output.audios[0]

   assert audio.ndim == 1
   assert len(_UpperCAmelCase     ) / vocoder_sampling_rate == 0.032






  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  self.get_dummy_components()
   lowerCamelCase_      =  AudioLDMPipeline(**_UpperCAmelCase     )
   lowerCamelCase_      =  audioldm_pipe.to(_UpperCAmelCase     )
   audioldm_pipe.set_progress_bar_config(disable=_UpperCAmelCase     )

   lowerCamelCase_      =  ['hey']

   lowerCamelCase_      =  audioldm_pipe(_UpperCAmelCase			,  num_inference_steps=1     )
   lowerCamelCase_      =  output.audios.shape
   assert audio_shape == (1, 256)

   lowerCamelCase_      =  audioldm_pipe.vocoder.config
   config.model_in_dim *= 2
   lowerCamelCase_      =  SpeechTaHifiGan(_UpperCAmelCase     ).to(_UpperCAmelCase     )
   lowerCamelCase_      =  audioldm_pipe(_UpperCAmelCase			,  num_inference_steps=1     )
   lowerCamelCase_      =  output.audios.shape
   # waveform shape is unchanged, we just have 2x the number of mel channels in the spectrogram
   assert audio_shape == (1, 256)






  def   snake_case		(    self     ):







   """simple docstring"""
   self._test_attention_slicing_forward_pass(test_mean_pixel_difference=_UpperCAmelCase     )






  def   snake_case		(    self     ):







   """simple docstring"""
   self._test_inference_batch_single_identical(test_mean_pixel_difference=_UpperCAmelCase     )







  @unittest.skipIf(
      torch_device != "cuda" or not is_xformers_available()			,  reason="XFormers attention is only available with CUDA and `xformers` installed"			,  )
  def   snake_case		(    self     ):







   """simple docstring"""
   self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=_UpperCAmelCase     )





@slow
class      snake_case    (    unittest.TestCase			):





  """simple docstring"""




  def   snake_case		(    self     ):







   """simple docstring"""
   super().tearDown()
   gc.collect()
   torch.cuda.empty_cache()






  def   snake_case		(    self			,  UpperCamelCase			,  UpperCamelCase="cpu"			,  UpperCamelCase=torch.floataa			,  UpperCamelCase=0     ):







   """simple docstring"""
   lowerCamelCase_      =  torch.Generator(device=_UpperCAmelCase     ).manual_seed(_UpperCAmelCase     )
   lowerCamelCase_      =  np.random.RandomState(_UpperCAmelCase     ).standard_normal((1, 8, 128, 16)     )
   lowerCamelCase_      =  torch.from_numpy(_UpperCAmelCase     ).to(device=_UpperCAmelCase			,  dtype=_UpperCAmelCase     )
   lowerCamelCase_      =  {
       'prompt': 'A hammer hitting a wooden surface',
       'latents': latents,
       'generator': generator,
       'num_inference_steps': 3,
       'guidance_scale': 2.5,
   }
   return inputs






  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  AudioLDMPipeline.from_pretrained("cvssp/audioldm"     )
   lowerCamelCase_      =  audioldm_pipe.to(_UpperCAmelCase     )
   audioldm_pipe.set_progress_bar_config(disable=_UpperCAmelCase     )

   lowerCamelCase_      =  self.get_inputs(_UpperCAmelCase     )
   lowerCamelCase_      =  25
   lowerCamelCase_      =  audioldm_pipe(**_UpperCAmelCase     ).audios[0]

   assert audio.ndim == 1
   assert len(_UpperCAmelCase     ) == 8_1920

   lowerCamelCase_      =  audio[7_7230:7_7240]
   lowerCamelCase_      =  np.array(
       [-0.4_884, -0.4_607, 0.0_023, 0.5_007, 0.5_896, 0.5_151, 0.3_813, -0.0_208, -0.3_687, -0.4_315]     )
   lowerCamelCase_      =  np.abs(expected_slice - audio_slice     ).max()
   assert max_diff < 1e-2







  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  AudioLDMPipeline.from_pretrained("cvssp/audioldm"     )
   lowerCamelCase_      =  LMSDiscreteScheduler.from_config(audioldm_pipe.scheduler.config     )
   lowerCamelCase_      =  audioldm_pipe.to(_UpperCAmelCase     )
   audioldm_pipe.set_progress_bar_config(disable=_UpperCAmelCase     )

   lowerCamelCase_      =  self.get_inputs(_UpperCAmelCase     )
   lowerCamelCase_      =  audioldm_pipe(**_UpperCAmelCase     ).audios[0]

   assert audio.ndim == 1
   assert len(_UpperCAmelCase     ) == 8_1920

   lowerCamelCase_      =  audio[2_7780:2_7790]
   lowerCamelCase_      =  np.array([-0.2_131, -0.0_873, -0.0_124, -0.0_189, 0.0_569, 0.1_373, 0.1_883, 0.2_886, 0.3_297, 0.2_212]     )
   lowerCamelCase_      =  np.abs(expected_slice - audio_slice     ).max()
   assert max_diff < 3e-2










'''simple docstring'''




import numpy as np
import pandas as pd
from sklearn.preprocessing import MinMaxScaler
from tensorflow.keras.layers import LSTM, Dense
from tensorflow.keras.models import Sequential

if __name__ == "__main__":
  lowercase__    :      List[Any]						=    pd.read_csv('''sample_data.csv''', header=None)
  lowercase__    :      Union[str, Any]						=    df.shape[:1][0]
  # If you're using some other dataset input the target column
  lowercase__    :      Any						=    df.iloc[:, 1:2]
  lowercase__    :      int						=    actual_data.values.reshape(len_data, 1)
  lowercase__    :      int						=    MinMaxScaler().fit_transform(actual_data)
  lowercase__    :      Dict						=    10
  lowercase__    :      List[str]						=    5
  lowercase__    :      Dict						=    20
  lowercase__    :      Dict						=    len_data - periods * look_back
  lowercase__    :      Any						=    actual_data[:division]
  lowercase__    :      Optional[int]						=    actual_data[division - look_back :]
  lowercase__	,							lowercase__    :      Optional[Any]						=    [], []
  lowercase__	,							lowercase__    :      Tuple						=    [], []

  for i in range(0, len(train_data) - forward_days - look_back + 1):
    train_x.append(train_data[i : i + look_back])
    train_y.append(train_data[i + look_back : i + look_back + forward_days])
  for i in range(0, len(test_data) - forward_days - look_back + 1):
    test_x.append(test_data[i : i + look_back])
    test_y.append(test_data[i + look_back : i + look_back + forward_days])
  lowercase__    :      List[Any]						=    np.array(train_x)
  lowercase__    :      List[Any]						=    np.array(test_x)
  lowercase__    :      str						=    np.array([list(i.ravel()) for i in train_y])
  lowercase__    :      Any						=    np.array([list(i.ravel()) for i in test_y])

  lowercase__    :      Optional[Any]						=    Sequential()
  model.add(LSTM(1_28, input_shape=(look_back, 1), return_sequences=True))
  model.add(LSTM(64, input_shape=(1_28, 1)))
  model.add(Dense(forward_days))
  model.compile(loss='''mean_squared_error''', optimizer='''adam''')
  lowercase__    :      Tuple						=    model.fit(
      x_train, y_train, epochs=1_50, verbose=1, shuffle=True, batch_size=4
  )
  lowercase__    :      int						=    model.predict(x_test)