





from math import acos, sin
from typing import List, Tuple, Union

import numpy as np
import torch
from PIL import Image

from ...models import AutoencoderKL, UNetaDConditionModel
from ...schedulers import DDIMScheduler, DDPMScheduler
from ...utils import randn_tensor
from ..pipeline_utils import AudioPipelineOutput, BaseOutput, DiffusionPipeline, ImagePipelineOutput
from .mel import Mel





class      lowercase__    (							__A  ):
  __UpperCamelCase	     =  ["""vqvae"""]







  def __init__(	self    ,						_lowercase    ,						_lowercase    ,						_lowercase    ,						_lowercase    ,						):
       super().__init__()
       self.register_modules(unet=_lowercase    ,						scheduler=_lowercase    ,						mel=_lowercase    ,						vqvae=_lowercase						)







  def 	UpperCAmelCase__		(	self						):
       return 50 if isinstance(self.scheduler    ,						_lowercase						) else 1_000







  @torch.no_grad()
  def __call__(	self    ,						_lowercase = 1    ,						_lowercase = None    ,						_lowercase = None    ,						_lowercase = 0    ,						_lowercase = 0    ,						_lowercase = None    ,						_lowercase = None    ,						_lowercase = 0    ,						_lowercase = 0    ,						_lowercase = None    ,						_lowercase = 0    ,						_lowercase = None    ,						_lowercase = None    ,						_lowercase=True    ,						):
       lowerCAmelCase_					:       Tuple    =		steps or self.get_default_steps()
       self.scheduler.set_timesteps(_lowercase						)
       lowerCAmelCase_					:       Union[str, Any]    =		step_generator or generator
       # For backwards compatibility
       if type(self.unet.config.sample_size						) == int:
            lowerCAmelCase_					:       Optional[Any]    =		(self.unet.config.sample_size, self.unet.config.sample_size)
       if noise is None:
            lowerCAmelCase_					:       Tuple    =		randn_tensor(
                (
                    batch_size,
                    self.unet.config.in_channels,
                    self.unet.config.sample_size[0],
                    self.unet.config.sample_size[1],
                )    ,						generator=_lowercase    ,						device=self.device    ,						)
       lowerCAmelCase_					:       List[str]    =		noise
       lowerCAmelCase_					:       List[str]    =		None

       if audio_file is not None or raw_audio is not None:
            self.mel.load_audio(_lowercase    ,						_lowercase						)
            lowerCAmelCase_					:       Tuple    =		self.mel.audio_slice_to_image(_lowercase						)
            lowerCAmelCase_					:       List[Any]    =		np.frombuffer(input_image.tobytes()    ,						dtype="""uint8"""						).reshape(
                (input_image.height, input_image.width)						)
            lowerCAmelCase_					:       Optional[Any]    =		(input_image / 255) * 2 - 1
            lowerCAmelCase_					:       Optional[Any]    =		torch.tensor(input_image[np.newaxis, :, :]    ,						dtype=torch.float						).to(self.device						)

            if self.vqvae is not None:
                 lowerCAmelCase_					:       List[Any]    =		self.vqvae.encode(torch.unsqueeze(_lowercase    ,						0						)						).latent_dist.sample(
                     generator=_lowercase						)[0]
                 lowerCAmelCase_					:       str    =		self.vqvae.config.scaling_factor * input_images

            if start_step > 0:
                 lowerCAmelCase_					:       str    =		self.scheduler.add_noise(_lowercase    ,						_lowercase    ,						self.scheduler.timesteps[start_step - 1]						)

            lowerCAmelCase_					:       Dict    =		(
                self.unet.config.sample_size[1] * self.mel.get_sample_rate() / self.mel.x_res / self.mel.hop_length
            )
            lowerCAmelCase_					:       Union[str, Any]    =		int(mask_start_secs * pixels_per_second						)
            lowerCAmelCase_					:       str    =		int(mask_end_secs * pixels_per_second						)
            lowerCAmelCase_					:       List[Any]    =		self.scheduler.add_noise(_lowercase    ,						_lowercase    ,						torch.tensor(self.scheduler.timesteps[start_step:]						)						)

       for step, t in enumerate(self.progress_bar(self.scheduler.timesteps[start_step:]						)						):
            if isinstance(self.unet    ,						_lowercase						):
                 lowerCAmelCase_					:       List[Any]    =		self.unet(_lowercase    ,						_lowercase    ,						_lowercase						)["""sample"""]
            else:
                 lowerCAmelCase_					:       Any    =		self.unet(_lowercase    ,						_lowercase						)["""sample"""]

            if isinstance(self.scheduler    ,						_lowercase						):
                 lowerCAmelCase_					:       str    =		self.scheduler.step(
                     model_output=_lowercase    ,						timestep=_lowercase    ,						sample=_lowercase    ,						eta=_lowercase    ,						generator=_lowercase    ,						)["""prev_sample"""]
            else:
                 lowerCAmelCase_					:       List[str]    =		self.scheduler.step(
                     model_output=_lowercase    ,						timestep=_lowercase    ,						sample=_lowercase    ,						generator=_lowercase    ,						)["""prev_sample"""]

            if mask is not None:
                 if mask_start > 0:
                      lowerCAmelCase_					:       Optional[Any]    =		mask[:, step, :, :mask_start]
                 if mask_end > 0:
                      lowerCAmelCase_					:       Optional[Any]    =		mask[:, step, :, -mask_end:]

       if self.vqvae is not None:
            # 0.18215 was scaling factor used in training to ensure unit variance
            lowerCAmelCase_					:       Dict    =		1 / self.vqvae.config.scaling_factor * images
            lowerCAmelCase_					:       Dict    =		self.vqvae.decode(_lowercase						)["""sample"""]

       lowerCAmelCase_					:       List[Any]    =		(images / 2 + 0.5).clamp(0    ,						1						)
       lowerCAmelCase_					:       List[Any]    =		images.cpu().permute(0    ,						2    ,						3    ,						1						).numpy()
       lowerCAmelCase_					:       Optional[Any]    =		(images * 255).round().astype("""uint8"""						)
       lowerCAmelCase_					:       str    =		list(
           (Image.fromarray(_[:, :, 0]						) for _ in images)
           if images.shape[3] == 1
           else (Image.fromarray(_lowercase    ,						mode="""RGB"""						).convert("""L"""						) for _ in images)						)

       lowerCAmelCase_					:       str    =		[self.mel.image_to_audio(_lowercase						) for _ in images]
       if not return_dict:
            return images, (self.mel.get_sample_rate(), audios)

       return BaseOutput(**AudioPipelineOutput(np.array(_lowercase						)[:, np.newaxis, :]						)    ,						**ImagePipelineOutput(_lowercase						)						)







  @torch.no_grad()
  def 	UpperCAmelCase__		(	self    ,						_lowercase    ,						_lowercase = 50						):
       assert isinstance(self.scheduler    ,						_lowercase						)
       self.scheduler.set_timesteps(_lowercase						)
       lowerCAmelCase_					:       List[Any]    =		np.array(
           [np.frombuffer(image.tobytes()    ,						dtype="""uint8"""						).reshape((1, image.height, image.width)						) for image in images]						)
       lowerCAmelCase_					:       List[str]    =		(sample / 255) * 2 - 1
       lowerCAmelCase_					:       Optional[Any]    =		torch.Tensor(_lowercase						).to(self.device						)

       for t in self.progress_bar(torch.flip(self.scheduler.timesteps    ,						(0,)						)						):
            lowerCAmelCase_					:       int    =		t - self.scheduler.config.num_train_timesteps // self.scheduler.num_inference_steps
            lowerCAmelCase_					:       Optional[int]    =		self.scheduler.alphas_cumprod[t]
            lowerCAmelCase_					:       Any    =		(
                self.scheduler.alphas_cumprod[prev_timestep]
                if prev_timestep >= 0
                else self.scheduler.final_alpha_cumprod
            )
            lowerCAmelCase_					:       Optional[int]    =		1 - alpha_prod_t
            lowerCAmelCase_					:       Union[str, Any]    =		self.unet(_lowercase    ,						_lowercase						)["""sample"""]
            lowerCAmelCase_					:       int    =		(1 - alpha_prod_t_prev) ** 0.5 * model_output
            lowerCAmelCase_					:       Dict    =		(sample - pred_sample_direction) * alpha_prod_t_prev ** (-0.5)
            lowerCAmelCase_					:       Optional[Any]    =		sample * alpha_prod_t ** 0.5 + beta_prod_t ** 0.5 * model_output

       return sample







  @staticmethod
  def 	UpperCAmelCase__		(	_lowercase    ,						_lowercase    ,						_lowercase						):
       lowerCAmelCase_					:       Optional[int]    =		acos(torch.dot(torch.flatten(_lowercase						)    ,						torch.flatten(_lowercase						)						) / torch.norm(_lowercase						) / torch.norm(_lowercase						)						)
       return sin((1 - alpha) * theta						) * xa / sin(_lowercase						) + sin(alpha * theta						) * xa / sin(_lowercase						)







import json
import os
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from requests.exceptions import HTTPError

from transformers.utils import (
    CONFIG_NAME,
    FLAX_WEIGHTS_NAME,
    TF2_WEIGHTS_NAME,
    TRANSFORMERS_CACHE,
    WEIGHTS_NAME,
    cached_file,
    get_file_from_repo,
    has_file,
)


UpperCAmelCase_					:   List[str] 							=						"""hf-internal-testing/tiny-random-bert"""
UpperCAmelCase_					:   List[Any] 							=						os.path.join(TRANSFORMERS_CACHE, """models--hf-internal-testing--tiny-random-bert""")
UpperCAmelCase_					:   Tuple 							=						"""9b8c223d42b2188cb49d29af482996f9d0f3e5a6"""





class      lowercase__    (							unittest.TestCase  ):







  def 	UpperCAmelCase__		(	self						):
       lowerCAmelCase_					:       Union[str, Any]    =		cached_file(_lowercase    ,						_lowercase						)
       # Should have downloaded the file in here
       self.assertTrue(os.path.isdir(_lowercase						)						)
       # Cache should contain at least those three subfolders:
       for subfolder in ["blobs", "refs", "snapshots"]:
            self.assertTrue(os.path.isdir(os.path.join(_lowercase    ,						_lowercase						)						)						)
       with open(os.path.join(_lowercase    ,						"""refs"""    ,						"""main"""						)						) as f:
            lowerCAmelCase_					:       Any    =		f.read()
       self.assertEqual(_lowercase    ,						os.path.join(_lowercase    ,						"""snapshots"""    ,						_lowercase    ,						_lowercase						)						)
       self.assertTrue(os.path.isfile(_lowercase						)						)

       # File is cached at the same place the second time.
       lowerCAmelCase_					:       Dict    =		cached_file(_lowercase    ,						_lowercase						)
       self.assertEqual(_lowercase    ,						_lowercase						)

       # Using a specific revision to test the full commit hash.
       lowerCAmelCase_					:       Optional[int]    =		cached_file(_lowercase    ,						_lowercase    ,						revision="""9b8c223"""						)
       self.assertEqual(_lowercase    ,						os.path.join(_lowercase    ,						"""snapshots"""    ,						_lowercase    ,						_lowercase						)						)







  def 	UpperCAmelCase__		(	self						):
       with self.assertRaisesRegex(_lowercase    ,						"""is not a valid model identifier"""						):
            lowerCAmelCase_					:       Dict    =		cached_file("""tiny-random-bert"""    ,						_lowercase						)

       with self.assertRaisesRegex(_lowercase    ,						"""is not a valid git identifier"""						):
            lowerCAmelCase_					:       Tuple    =		cached_file(_lowercase    ,						_lowercase    ,						revision="""aaaa"""						)

       with self.assertRaisesRegex(_lowercase    ,						"""does not appear to have a file named"""						):
            lowerCAmelCase_					:       List[Any]    =		cached_file(_lowercase    ,						"""conf"""						)







  def 	UpperCAmelCase__		(	self						):
       with self.assertRaisesRegex(_lowercase    ,						"""does not appear to have a file named"""						):
            lowerCAmelCase_					:       Any    =		cached_file(_lowercase    ,						"""conf"""						)

       with open(os.path.join(_lowercase    ,						"""refs"""    ,						"""main"""						)						) as f:
            lowerCAmelCase_					:       str    =		f.read()
       self.assertTrue(os.path.isfile(os.path.join(_lowercase    ,						""".no_exist"""    ,						_lowercase    ,						"""conf"""						)						)						)

       lowerCAmelCase_					:       Union[str, Any]    =		cached_file(_lowercase    ,						"""conf"""    ,						_raise_exceptions_for_missing_entries=_lowercase						)
       self.assertIsNone(_lowercase						)

       lowerCAmelCase_					:       Any    =		cached_file(_lowercase    ,						"""conf"""    ,						local_files_only=_lowercase    ,						_raise_exceptions_for_missing_entries=_lowercase						)
       self.assertIsNone(_lowercase						)

       lowerCAmelCase_					:       Any    =		mock.Mock()
       lowerCAmelCase_					:       Optional[Any]    =		500
       lowerCAmelCase_					:       Any    =		{}
       lowerCAmelCase_					:       List[str]    =		HTTPError
       lowerCAmelCase_					:       List[str]    =		{}

       # Under the mock environment we get a 500 error when trying to reach the tokenizer.
       with mock.patch("""requests.Session.request"""    ,						return_value=_lowercase						) as mock_head:
            lowerCAmelCase_					:       Union[str, Any]    =		cached_file(_lowercase    ,						"""conf"""    ,						_raise_exceptions_for_connection_errors=_lowercase						)
            self.assertIsNone(_lowercase						)
            # This check we did call the fake head request
            mock_head.assert_called()







  def 	UpperCAmelCase__		(	self						):
       self.assertTrue(has_file("""hf-internal-testing/tiny-bert-pt-only"""    ,						_lowercase						)						)
       self.assertFalse(has_file("""hf-internal-testing/tiny-bert-pt-only"""    ,						_lowercase						)						)
       self.assertFalse(has_file("""hf-internal-testing/tiny-bert-pt-only"""    ,						_lowercase						)						)







  def 	UpperCAmelCase__		(	self						):
       # `get_file_from_repo` returns None if the file does not exist
       self.assertIsNone(get_file_from_repo("""bert-base-cased"""    ,						"""ahah.txt"""						)						)

       # The function raises if the repository does not exist.
       with self.assertRaisesRegex(_lowercase    ,						"""is not a valid model identifier"""						):
            get_file_from_repo("""bert-base-case"""    ,						_lowercase						)

       # The function raises if the revision does not exist.
       with self.assertRaisesRegex(_lowercase    ,						"""is not a valid git identifier"""						):
            get_file_from_repo("""bert-base-cased"""    ,						_lowercase    ,						revision="""ahaha"""						)

       lowerCAmelCase_					:       Union[str, Any]    =		get_file_from_repo("""bert-base-cased"""    ,						_lowercase						)
       # The name is the cached name which is not very easy to test, so instead we load the content.
       lowerCAmelCase_					:       Any    =		json.loads(open(_lowercase    ,						"""r"""						).read()						)
       self.assertEqual(config["""hidden_size"""]    ,						768						)







  def 	UpperCAmelCase__		(	self						):
       with tempfile.TemporaryDirectory() as tmp_dir:
            lowerCAmelCase_					:       Optional[Any]    =		Path(_lowercase						) / """a.txt"""
            filename.touch()
            self.assertEqual(get_file_from_repo(_lowercase    ,						"""a.txt"""						)    ,						str(_lowercase						)						)

            self.assertIsNone(get_file_from_repo(_lowercase    ,						"""b.txt"""						)						)
