





'''simple docstring'''




import os
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen, xsplitext
from ..table import array_cast
from ..utils.py_utils import no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
  from .features import FeatureType


lowercase__      ,				lowercase__      ,				lowercase__       : Optional[int]         =       False, False, False


@dataclass
class 					SCREAMING_SNAKE_CASE       :
     lowerCAmelCase						     =			None
     lowerCAmelCase						     =			True
     lowerCAmelCase						     =			True
     lowerCAmelCase						     =			None
     # Automatically constructed
     lowerCAmelCase						     =			"dict"
     lowerCAmelCase						     =			pa.struct({'''bytes''': pa.binary(), '''path''': pa.string()}				)
     lowerCAmelCase						     =			field(default='''Audio'''   ,						init=a__   ,						repr=a__				)


     def __call__(		self):

          '''simple docstring'''






          return self.pa_type


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






          try:
               import soundfile as sf  # soundfile is a dependency of librosa, needed to decode audio files.
          except ImportError as err:
               raise ImportError('To support encoding audio data, please install \'soundfile\'.') from err
          if isinstance(_UpperCAmelCase		,      _UpperCAmelCase):
               return {"bytes": None, "path": value}
          elif isinstance(_UpperCAmelCase		,      _UpperCAmelCase):
               return {"bytes": value, "path": None}
          elif "array" in value:
               # convert the audio array to wav bytes
               __A		:					Optional[int]						=			BytesIO()
               sf.write(_UpperCAmelCase		,      value['array']		,      value['sampling_rate']		,      format='wav')
               return {"bytes": buffer.getvalue(), "path": None}
          elif value.get('path') is not None and os.path.isfile(value['path']):
               # we set "bytes": None to not duplicate the data if they're already available locally
               if value["path"].endswith('pcm'):
                    # "PCM" only has raw audio bytes
                    if value.get('sampling_rate') is None:
                         # At least, If you want to convert "PCM-byte" to "WAV-byte", you have to know sampling rate
                         raise KeyError('To use PCM files, please specify a \'sampling_rate\' in Audio object')
                    if value.get('bytes'):
                         # If we already had PCM-byte, we don`t have to make "read file, make bytes" (just use it!)
                         __A		:					Tuple						=			np.frombuffer(value['bytes']		,      dtype=np.intaa).astype(np.floataa) / 3_2767
                    else:
                         __A		:					List[Any]						=			np.memmap(value['path']		,      dtype='h'		,      mode='r').astype(np.floataa) / 3_2767

                    __A		:					Optional[Any]						=			BytesIO(bytes())
                    sf.write(_UpperCAmelCase		,      _UpperCAmelCase		,      value['sampling_rate']		,      format='wav')
                    return {"bytes": buffer.getvalue(), "path": None}
               else:
                    return {"bytes": None, "path": value.get('path')}
          elif value.get('bytes') is not None or value.get('path') is not None:
               # store the audio bytes, and path is used to infer the audio format using the file extension
               return {"bytes": value.get('bytes'), "path": value.get('path')}
          else:
               raise ValueError(
                   F'An audio sample should have one of \'path\' or \'bytes\' but they are missing or None in {value}.')


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase = None):

          '''simple docstring'''






          if not self.decode:
               raise RuntimeError('Decoding is disabled for this feature. Please use Audio(decode=True) instead.')

          __A					,__A		:					Any						=			(value['path'], BytesIO(value['bytes'])) if value['bytes'] is not None else (value['path'], None)
          if path is None and file is None:
               raise ValueError(F'An audio sample should have one of \'path\' or \'bytes\' but both are None in {value}.')

          try:
               import librosa
               import soundfile as sf
          except ImportError as err:
               raise ImportError('To support decoding audio files, please install \'librosa\' and \'soundfile\'.') from err

          __A		:					Tuple						=			xsplitext(_UpperCAmelCase)[1][1:].lower() if path is not None else None
          if not config.IS_OPUS_SUPPORTED and audio_format == "opus":
               raise RuntimeError(
                   'Decoding \'opus\' files requires system library \'libsndfile\'>=1.0.31, '
                   'You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. ')
          elif not config.IS_MP3_SUPPORTED and audio_format == "mp3":
               raise RuntimeError(
                   'Decoding \'mp3\' files requires system library \'libsndfile\'>=1.1.0, '
                   'You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. ')

          if file is None:
               __A		:					List[str]						=			token_per_repo_id or {}
               __A		:					Optional[int]						=			path.split('::')[-1]
               try:
                    __A		:					str						=			string_to_dict(_UpperCAmelCase		,      config.HUB_DATASETS_URL)['repo_id']
                    __A		:					Union[str, Any]						=			token_per_repo_id[repo_id]
               except (ValueError, KeyError):
                    __A		:					List[Any]						=			None

               with xopen(_UpperCAmelCase		,      'rb'		,      use_auth_token=_UpperCAmelCase) as f:
                    __A					,__A		:					Optional[Any]						=			sf.read(_UpperCAmelCase)

          else:
               __A					,__A		:					Tuple						=			sf.read(_UpperCAmelCase)

          __A		:					Union[str, Any]						=			array.T
          if self.mono:
               __A		:					Optional[Any]						=			librosa.to_mono(_UpperCAmelCase)
          if self.sampling_rate and self.sampling_rate != sampling_rate:
               __A		:					List[Any]						=			librosa.resample(_UpperCAmelCase		,      orig_sr=_UpperCAmelCase		,      target_sr=self.sampling_rate)
               __A		:					Optional[Any]						=			self.sampling_rate

          return {"path": path, "array": array, "sampling_rate": sampling_rate}


     def 				SCREAMING_SNAKE_CASE    (		self):

          '''simple docstring'''






          from .features import Value

          if self.decode:
               raise ValueError('Cannot flatten a decoded Audio feature.')
          return {
              "bytes": Value('binary'),
              "path": Value('string'),
          }


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






          if pa.types.is_string(storage.type):
               __A		:					str						=			pa.array([None] * len(_UpperCAmelCase)		,      type=pa.binary())
               __A		:					Optional[int]						=			pa.StructArray.from_arrays([bytes_array, storage]		,      ['bytes', 'path']		,      mask=storage.is_null())
          elif pa.types.is_binary(storage.type):
               __A		:					Optional[Any]						=			pa.array([None] * len(_UpperCAmelCase)		,      type=pa.string())
               __A		:					str						=			pa.StructArray.from_arrays([storage, path_array]		,      ['bytes', 'path']		,      mask=storage.is_null())
          elif pa.types.is_struct(storage.type) and storage.type.get_all_field_indices('array'):
               __A		:					Dict						=			pa.array([Audio().encode_example(_UpperCAmelCase) if x is not None else None for x in storage.to_pylist()])
          elif pa.types.is_struct(storage.type):
               if storage.type.get_field_index('bytes') >= 0:
                    __A		:					Tuple						=			storage.field('bytes')
               else:
                    __A		:					Any						=			pa.array([None] * len(_UpperCAmelCase)		,      type=pa.binary())
               if storage.type.get_field_index('path') >= 0:
                    __A		:					List[Any]						=			storage.field('path')
               else:
                    __A		:					Dict						=			pa.array([None] * len(_UpperCAmelCase)		,      type=pa.string())
               __A		:					Optional[Any]						=			pa.StructArray.from_arrays([bytes_array, path_array]		,      ['bytes', 'path']		,      mask=storage.is_null())
          return array_cast(_UpperCAmelCase		,      self.pa_type)


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






          @no_op_if_value_is_null
          def path_to_bytes(_UpperCAmelCase):
               with xopen(_UpperCAmelCase		,      'rb') as f:
                    __A		:					Union[str, Any]						=			f.read()
               return bytes_

          __A		:					int						=			pa.array(
              [
                  (path_to_bytes(x['path']) if x['bytes'] is None else x['bytes']) if x is not None else None
                  for x in storage.to_pylist()
              ]		,      type=pa.binary()		,      )
          __A		:					Optional[int]						=			pa.array(
              [os.path.basename(_UpperCAmelCase) if path is not None else None for path in storage.field('path').to_pylist()]		,      type=pa.string()		,      )
          __A		:					Any						=			pa.StructArray.from_arrays([bytes_array, path_array]		,      ['bytes', 'path']		,      mask=bytes_array.is_null())
          return array_cast(_UpperCAmelCase		,      self.pa_type)





'''simple docstring'''







import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, DDIMScheduler, DDPMScheduler, StableDiffusionUpscalePipeline, UNetaDConditionModel
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu


enable_full_determinism()

class 	_snake_case(			unittest.TestCase      ):






 def 						_UpperCamelCase (self       :  List[str]    )       ->   str:






    """simple docstring"""


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






 @property
 def 						_UpperCamelCase (self       :  int    )       ->   Any:






    """simple docstring"""


    A__       =     1
    A__       =     3
    A__       =     (32, 32)

    A__       =     floats_tensor((batch_size, num_channels) + sizes    ,		rng=random.Random(0    )    ).to(a    )
    return image






 @property
 def 						_UpperCamelCase (self       :  Union[str, Any]    )       ->   Tuple:






    """simple docstring"""


    torch.manual_seed(0    )
    A__       =     UNetaDConditionModel(
        block_out_channels=(32, 32, 64)    ,		layers_per_block=2    ,		sample_size=32    ,		in_channels=7    ,		out_channels=4    ,		down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D', 'CrossAttnDownBlock2D')    ,		up_block_types=('CrossAttnUpBlock2D', 'CrossAttnUpBlock2D', 'UpBlock2D')    ,		cross_attention_dim=32    ,		attention_head_dim=8    ,		use_linear_projection=a    ,		only_cross_attention=(True, True, False)    ,		num_class_embeds=1_00    ,		)
    return model






 @property
 def 						_UpperCamelCase (self       :  Tuple    )       ->   Optional[Any]:






    """simple docstring"""


    torch.manual_seed(0    )
    A__       =     AutoencoderKL(
        block_out_channels=[32, 32, 64]    ,		in_channels=3    ,		out_channels=3    ,		down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D', 'DownEncoderBlock2D']    ,		up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D', 'UpDecoderBlock2D']    ,		latent_channels=4    ,		)
    return model






 @property
 def 						_UpperCamelCase (self       :  str    )       ->   List[str]:






    """simple docstring"""


    torch.manual_seed(0    )
    A__       =     CLIPTextConfig(
        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=10_00    ,		hidden_act='gelu'    ,		projection_dim=5_12    ,		)
    return CLIPTextModel(a    )






 def 						_UpperCamelCase (self       :  str    )       ->   List[str]:






    """simple docstring"""


    A__       =     'cpu'  # ensure determinism for the device-dependent torch.Generator
    A__       =     self.dummy_cond_unet_upscale
    A__       =     DDPMScheduler()
    A__       =     DDIMScheduler(prediction_type='v_prediction'    )
    A__       =     self.dummy_vae
    A__       =     self.dummy_text_encoder
    A__       =     CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'    )

    A__       =     self.dummy_image.cpu().permute(0    ,		2    ,		3    ,		1    )[0]
    A__       =     Image.fromarray(np.uinta(a    )    ).convert('RGB'    ).resize((64, 64)    )

    # make sure here that pndm scheduler skips prk
    A__       =     StableDiffusionUpscalePipeline(
        unet=a    ,		low_res_scheduler=a    ,		scheduler=a    ,		vae=a    ,		text_encoder=a    ,		tokenizer=a    ,		max_noise_level=3_50    ,		)
    A__       =     sd_pipe.to(a    )
    sd_pipe.set_progress_bar_config(disable=a    )

    A__       =     'A painting of a squirrel eating a burger'
    A__       =     torch.Generator(device=a    ).manual_seed(0    )
    A__       =     sd_pipe(
        [prompt]    ,		image=a    ,		generator=a    ,		guidance_scale=6.0    ,		noise_level=20    ,		num_inference_steps=2    ,		output_type='np'    ,		)

    A__       =     output.images

    A__       =     torch.Generator(device=a    ).manual_seed(0    )
    A__       =     sd_pipe(
        [prompt]    ,		image=a    ,		generator=a    ,		guidance_scale=6.0    ,		noise_level=20    ,		num_inference_steps=2    ,		output_type='np'    ,		return_dict=a    ,		)[0]

    A__       =     image[0, -3:, -3:, -1]
    A__       =     image_from_tuple[0, -3:, -3:, -1]

    A__       =     low_res_image.size[0] * 4
    assert image.shape == (1, expected_height_width, expected_height_width, 3)
    A__       =     np.array([0.3113, 0.3910, 0.4272, 0.4859, 0.5061, 0.4652, 0.5362, 0.5715, 0.5661]    )

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






 def 						_UpperCamelCase (self       :  Optional[int]    )       ->   Optional[int]:






    """simple docstring"""


    A__       =     'cpu'  # ensure determinism for the device-dependent torch.Generator
    A__       =     self.dummy_cond_unet_upscale
    A__       =     DDPMScheduler()
    A__       =     DDIMScheduler(prediction_type='v_prediction'    )
    A__       =     self.dummy_vae
    A__       =     self.dummy_text_encoder
    A__       =     CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'    )

    A__       =     self.dummy_image.cpu().permute(0    ,		2    ,		3    ,		1    )[0]
    A__       =     Image.fromarray(np.uinta(a    )    ).convert('RGB'    ).resize((64, 64)    )

    # make sure here that pndm scheduler skips prk
    A__       =     StableDiffusionUpscalePipeline(
        unet=a    ,		low_res_scheduler=a    ,		scheduler=a    ,		vae=a    ,		text_encoder=a    ,		tokenizer=a    ,		max_noise_level=3_50    ,		)
    A__       =     sd_pipe.to(a    )
    sd_pipe.set_progress_bar_config(disable=a    )

    A__       =     'A painting of a squirrel eating a burger'
    A__       =     sd_pipe(
        2 * [prompt]    ,		image=2 * [low_res_image]    ,		guidance_scale=6.0    ,		noise_level=20    ,		num_inference_steps=2    ,		output_type='np'    ,		)
    A__       =     output.images
    assert image.shape[0] == 2

    A__       =     torch.Generator(device=a    ).manual_seed(0    )
    A__       =     sd_pipe(
        [prompt]    ,		image=a    ,		generator=a    ,		num_images_per_prompt=2    ,		guidance_scale=6.0    ,		noise_level=20    ,		num_inference_steps=2    ,		output_type='np'    ,		)
    A__       =     output.images
    assert image.shape[0] == 2






 @unittest.skipIf(torch_device != 'cuda'    ,		'This test requires a GPU'    )
 def 						_UpperCamelCase (self       :  List[Any]    )       ->   Union[str, Any]:






    """simple docstring"""


    A__       =     self.dummy_cond_unet_upscale
    A__       =     DDPMScheduler()
    A__       =     DDIMScheduler(prediction_type='v_prediction'    )
    A__       =     self.dummy_vae
    A__       =     self.dummy_text_encoder
    A__       =     CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'    )

    A__       =     self.dummy_image.cpu().permute(0    ,		2    ,		3    ,		1    )[0]
    A__       =     Image.fromarray(np.uinta(a    )    ).convert('RGB'    ).resize((64, 64)    )

    # put models in fp16, except vae as it overflows in fp16
    A__       =     unet.half()
    A__       =     text_encoder.half()

    # make sure here that pndm scheduler skips prk
    A__       =     StableDiffusionUpscalePipeline(
        unet=a    ,		low_res_scheduler=a    ,		scheduler=a    ,		vae=a    ,		text_encoder=a    ,		tokenizer=a    ,		max_noise_level=3_50    ,		)
    A__       =     sd_pipe.to(a    )
    sd_pipe.set_progress_bar_config(disable=a    )

    A__       =     'A painting of a squirrel eating a burger'
    A__       =     torch.manual_seed(0    )
    A__       =     sd_pipe(
        [prompt]    ,		image=a    ,		generator=a    ,		num_inference_steps=2    ,		output_type='np'    ,		).images

    A__       =     low_res_image.size[0] * 4
    assert image.shape == (1, expected_height_width, expected_height_width, 3)




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






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






    """simple docstring"""


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






 def 						_UpperCamelCase (self       :  Optional[Any]    )       ->   Optional[int]:






    """simple docstring"""


    A__       =     load_image(
        'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
        '/sd2-upscale/low_res_cat.png'    )
    A__       =     load_numpy(
        'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-upscale'
        '/upsampled_cat.npy'    )

    A__       =     'stabilityai/stable-diffusion-x4-upscaler'
    A__       =     StableDiffusionUpscalePipeline.from_pretrained(a    )
    pipe.to(a    )
    pipe.set_progress_bar_config(disable=a    )
    pipe.enable_attention_slicing()

    A__       =     'a cat sitting on a park bench'

    A__       =     torch.manual_seed(0    )
    A__       =     pipe(
        prompt=a    ,		image=a    ,		generator=a    ,		output_type='np'    ,		)
    A__       =     output.images[0]

    assert image.shape == (5_12, 5_12, 3)
    assert np.abs(expected_image - image    ).max() < 1e-3






 def 						_UpperCamelCase (self       :  Any    )       ->   str:






    """simple docstring"""


    A__       =     load_image(
        'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
        '/sd2-upscale/low_res_cat.png'    )
    A__       =     load_numpy(
        'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-upscale'
        '/upsampled_cat_fp16.npy'    )

    A__       =     'stabilityai/stable-diffusion-x4-upscaler'
    A__       =     StableDiffusionUpscalePipeline.from_pretrained(
        a    ,		torch_dtype=torch.floataa    ,		)
    pipe.to(a    )
    pipe.set_progress_bar_config(disable=a    )
    pipe.enable_attention_slicing()

    A__       =     'a cat sitting on a park bench'

    A__       =     torch.manual_seed(0    )
    A__       =     pipe(
        prompt=a    ,		image=a    ,		generator=a    ,		output_type='np'    ,		)
    A__       =     output.images[0]

    assert image.shape == (5_12, 5_12, 3)
    assert np.abs(expected_image - image    ).max() < 5e-1






 def 						_UpperCamelCase (self       :  Dict    )       ->   List[str]:






    """simple docstring"""


    torch.cuda.empty_cache()
    torch.cuda.reset_max_memory_allocated()
    torch.cuda.reset_peak_memory_stats()

    A__       =     load_image(
        'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
        '/sd2-upscale/low_res_cat.png'    )

    A__       =     'stabilityai/stable-diffusion-x4-upscaler'
    A__       =     StableDiffusionUpscalePipeline.from_pretrained(
        a    ,		torch_dtype=torch.floataa    ,		)
    pipe.to(a    )
    pipe.set_progress_bar_config(disable=a    )
    pipe.enable_attention_slicing(1    )
    pipe.enable_sequential_cpu_offload()

    A__       =     'a cat sitting on a park bench'

    A__       =     torch.manual_seed(0    )
    A__       =     pipe(
        prompt=a    ,		image=a    ,		generator=a    ,		num_inference_steps=5    ,		output_type='np'    ,		)

    A__       =     torch.cuda.max_memory_allocated()
    # make sure that less than 2.9 GB is allocated
    assert mem_bytes < 2.9 * 10**9


