





"""simple docstring"""



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

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen
from ..table import array_cast
from ..utils.file_utils import is_local_path
from ..utils.py_utils import first_non_null_value, no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
       import PIL.Image

       from .features import FeatureType


a_ =			None
a_ =			"""<""" if sys.byteorder == """little""" else """>"""
# Origin: https://github.com/python-pillow/Pillow/blob/698951e19e19972aeed56df686868f1329981c12/src/PIL/Image.py#L3126 minus "|i1" which values are not preserved correctly when saving and loading an image
a_ =			[
    np.dtype("""|b1"""),
    np.dtype("""|u1"""),
    np.dtype("""<u2"""),
    np.dtype(""">u2"""),
    np.dtype("""<i2"""),
    np.dtype(""">i2"""),
    np.dtype("""<u4"""),
    np.dtype(""">u4"""),
    np.dtype("""<i4"""),
    np.dtype(""">i4"""),
    np.dtype("""<f4"""),
    np.dtype(""">f4"""),
    np.dtype("""<f8"""),
    np.dtype(""">f8"""),
]






@dataclass
class 		__snake_case					:





 """simple docstring"""


 _lowerCamelCase        =     True
 _lowerCamelCase        =     None
 # Automatically constructed
 _lowerCamelCase        =     "PIL.Image.Image"
 _lowerCamelCase        =     pa.struct({"""bytes""": pa.binary(), """path""": pa.string()}     )
 _lowerCamelCase        =     field(default="""Image"""							,							init=SCREAMING_SNAKE_CASE__							,							repr=SCREAMING_SNAKE_CASE__     )

 def __call__(    self  ):



     '''simple docstring'''




     return self.pa_type

 def 					UpperCamelCase__(    self					,						__lowerCamelCase  ):



     '''simple docstring'''




     if config.PIL_AVAILABLE:
         import PIL.Image
     else:
         raise ImportError('''To support encoding images, please install \'Pillow\'.'''  )

     if isinstance(__lowerCamelCase					,						__lowerCamelCase  ):
         __A    :						List[str]						 =     np.array(__lowerCamelCase  )

     if isinstance(__lowerCamelCase					,						__lowerCamelCase  ):
         return {"path": value, "bytes": None}
     elif isinstance(__lowerCamelCase					,						__lowerCamelCase  ):
         return {"path": None, "bytes": value}
     elif isinstance(__lowerCamelCase					,						np.ndarray  ):
         # convert the image array to PNG/TIFF bytes
         return encode_np_array(__lowerCamelCase  )
     elif isinstance(__lowerCamelCase					,						PIL.Image.Image  ):
         # convert the PIL image to bytes (default format is PNG/TIFF)
         return encode_pil_image(__lowerCamelCase  )
     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
         return {"bytes": None, "path": value.get('''path'''  )}
     elif value.get('''bytes'''  ) is not None or value.get('''path'''  ) is not None:
         # store the image bytes, and path is used to infer the image format using the file extension
         return {"bytes": value.get('''bytes'''  ), "path": value.get('''path'''  )}
     else:
         raise ValueError(
             F"""An image sample should have one of 'path' or 'bytes' but they are missing or None in {value}."""  )

 def 					UpperCamelCase__(    self					,						__lowerCamelCase					,						__lowerCamelCase=None  ):



     '''simple docstring'''




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

     if config.PIL_AVAILABLE:
         import PIL.Image
     else:
         raise ImportError('''To support decoding images, please install \'Pillow\'.'''  )

     if token_per_repo_id is None:
         __A    :						Dict						 =     {}

     __A ,       __A    :						int						 =     value['''path'''], value['''bytes''']
     if bytes_ is None:
         if path is None:
             raise ValueError(F"""An image should have one of 'path' or 'bytes' but both are None in {value}."""  )
         else:
             if is_local_path(__lowerCamelCase  ):
                 __A    :						Optional[int]						 =     PIL.Image.open(__lowerCamelCase  )
             else:
                 __A    :						Any						 =     path.split('''::'''  )[-1]
                 try:
                     __A    :						int						 =     string_to_dict(__lowerCamelCase					,						config.HUB_DATASETS_URL  )['''repo_id''']
                     __A    :						List[Any]						 =     token_per_repo_id.get(__lowerCamelCase  )
                 except ValueError:
                     __A    :						str						 =     None
                 with xopen(__lowerCamelCase					,						'''rb'''					,						use_auth_token=__lowerCamelCase  ) as f:
                     __A    :						Dict						 =     BytesIO(f.read()  )
                 __A    :						Optional[Any]						 =     PIL.Image.open(bytes_  )
     else:
         __A    :						Union[str, Any]						 =     PIL.Image.open(BytesIO(bytes_  )  )
     image.load()  # to avoid "Too many open files" errors
     return image

 def 					UpperCamelCase__(    self  ):



     '''simple docstring'''




     from .features import Value

     return (
         self
         if self.decode
         else {
             "bytes": Value('''binary'''  ),
             "path": Value('''string'''  ),
         }
     )

 def 					UpperCamelCase__(    self					,						__lowerCamelCase  ):



     '''simple docstring'''




     if pa.types.is_string(storage.type  ):
         __A    :						str						 =     pa.array([None] * len(__lowerCamelCase  )					,						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(__lowerCamelCase  )					,						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  ):
         if storage.type.get_field_index('''bytes'''  ) >= 0:
             __A    :						Any						 =     storage.field('''bytes'''  )
         else:
             __A    :						Optional[int]						 =     pa.array([None] * len(__lowerCamelCase  )					,						type=pa.binary()  )
         if storage.type.get_field_index('''path'''  ) >= 0:
             __A    :						str						 =     storage.field('''path'''  )
         else:
             __A    :						Optional[int]						 =     pa.array([None] * len(__lowerCamelCase  )					,						type=pa.string()  )
         __A    :						Tuple						 =     pa.StructArray.from_arrays([bytes_array, path_array]					,						['''bytes''', '''path''']					,						mask=storage.is_null()  )
     elif pa.types.is_list(storage.type  ):
         __A    :						str						 =     pa.array(
             [encode_np_array(np.array(__lowerCamelCase  )  )['''bytes'''] if arr is not None else None for arr in storage.to_pylist()]					,						type=pa.binary()					,						)
         __A    :						int						 =     pa.array([None] * len(__lowerCamelCase  )					,						type=pa.string()  )
         __A    :						Dict						 =     pa.StructArray.from_arrays(
             [bytes_array, path_array]					,						['''bytes''', '''path''']					,						mask=bytes_array.is_null()  )
     return array_cast(__lowerCamelCase					,						self.pa_type  )







 def 					UpperCamelCase__(    self					,						__lowerCamelCase  ):



     '''simple docstring'''




     @no_op_if_value_is_null
     def path_to_bytes(__lowerCamelCase  ):
         with xopen(__lowerCamelCase					,						'''rb'''  ) as f:
             __A    :						Optional[int]						 =     f.read()
         return bytes_

     __A    :						Any						 =     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    :						List[Any]						 =     pa.array(
         [os.path.basename(__lowerCamelCase  ) if path is not None else None for path in storage.field('''path'''  ).to_pylist()]					,						type=pa.string()					,						)
     __A    :						Optional[Any]						 =     pa.StructArray.from_arrays([bytes_array, path_array]					,						['''bytes''', '''path''']					,						mask=bytes_array.is_null()  )
     return array_cast(__lowerCamelCase					,						self.pa_type  )







def 		__lowercase     (   )   ->List[str]:



    '''simple docstring'''







    if config.PIL_AVAILABLE:
        import PIL.Image
    else:
        raise ImportError('''To support encoding images, please install \'Pillow\'.'''						)

    global _IMAGE_COMPRESSION_FORMATS
    if _IMAGE_COMPRESSION_FORMATS is None:
        PIL.Image.init()
        __A    :						Tuple						 =     list(set(PIL.Image.OPEN.keys()						) & set(PIL.Image.SAVE.keys()						)						)
    return _IMAGE_COMPRESSION_FORMATS







def 		__lowercase     (   snake_case_  :  "PIL.Image.Image"						)   ->bytes:



    '''simple docstring'''







    __A    :						Tuple						 =     BytesIO()
    if image.format in list_image_compression_formats():
        __A    :						str						 =     image.format
    else:
        __A    :						Optional[int]						 =     '''PNG''' if image.mode in ['''1''', '''L''', '''LA''', '''RGB''', '''RGBA'''] else '''TIFF'''
    image.save(snake_case_				,format=snake_case_						)
    return buffer.getvalue()







def 		__lowercase     (   snake_case_  :  "PIL.Image.Image"						)   ->dict:



    '''simple docstring'''







    if hasattr(snake_case_				,'''filename'''						) and image.filename != "":
        return {"path": image.filename, "bytes": None}
    else:
        return {"path": None, "bytes": image_to_bytes(snake_case_						)}







def 		__lowercase     (   snake_case_  :  np.ndarray						)   ->dict:



    '''simple docstring'''







    if config.PIL_AVAILABLE:
        import PIL.Image
    else:
        raise ImportError('''To support encoding images, please install \'Pillow\'.'''						)

    __A    :						Optional[Any]						 =     array.dtype
    __A    :						List[str]						 =     dtype.byteorder if dtype.byteorder != '''=''' else _NATIVE_BYTEORDER
    __A    :						int						 =     dtype.kind
    __A    :						Any						 =     dtype.itemsize

    __A    :						List[Any]						 =     None

    # Multi-channel array case (only np.dtype("|u1") is allowed)
    if array.shape[2:]:
        __A    :						List[Any]						 =     np.dtype('''|u1'''						)
        if dtype_kind not in ["u", "i"]:
            raise TypeError(
                F"""Unsupported array dtype {dtype} for image encoding. Only {dest_dtype} is supported for multi-channel arrays."""						)
        if dtype is not dest_dtype:
            warnings.warn(F"""Downcasting array dtype {dtype} to {dest_dtype} to be compatible with 'Pillow'"""						)
    # Exact match
    elif dtype in _VALID_IMAGE_ARRAY_DTPYES:
        __A    :						str						 =     dtype
    else:  # Downcast the type within the kind (np.can_cast(from_type, to_type, casting="same_kind") doesn't behave as expected, so do it manually)
        while dtype_itemsize >= 1:
            __A    :						Union[str, Any]						 =     dtype_byteorder + dtype_kind + str(snake_case_						)
            __A    :						Tuple						 =     np.dtype(snake_case_						)
            if dest_dtype in _VALID_IMAGE_ARRAY_DTPYES:
                warnings.warn(F"""Downcasting array dtype {dtype} to {dest_dtype} to be compatible with 'Pillow'"""						)
                break
            else:
                dtype_itemsize //= 2

    if dest_dtype is None:
        raise TypeError(
            F"""Cannot convert dtype {dtype} to a valid image dtype. Valid image dtypes: {_VALID_IMAGE_ARRAY_DTPYES}"""						)
    __A    :						Optional[Any]						 =     PIL.Image.fromarray(array.astype(snake_case_						)						)
    return {"path": None, "bytes": image_to_bytes(snake_case_						)}







def 		__lowercase     (   snake_case_  :  Union[List[str], List[dict], List[np.ndarray], List["PIL.Image.Image"]]						)   ->List[dict]:



    '''simple docstring'''







    if config.PIL_AVAILABLE:
        import PIL.Image
    else:
        raise ImportError('''To support encoding images, please install \'Pillow\'.'''						)

    if objs:
        __A ,       __A    :						Optional[int]						 =     first_non_null_value(snake_case_						)
        if isinstance(snake_case_				,snake_case_						):
            return [{"path": obj, "bytes": None} if obj is not None else None for obj in objs]
        if isinstance(snake_case_				,np.ndarray						):
            __A    :						str						 =     no_op_if_value_is_null(snake_case_						)
            return [obj_to_image_dict_func(snake_case_						) for obj in objs]
        elif isinstance(snake_case_				,PIL.Image.Image						):
            __A    :						Any						 =     no_op_if_value_is_null(snake_case_						)
            return [obj_to_image_dict_func(snake_case_						) for obj in objs]
        else:
            return objs
    else:
        return objs











"""simple docstring"""



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 		__snake_case					(		SCREAMING_SNAKE_CASE__     ):





 """simple docstring"""


 _lowerCamelCase        =     ["""vqvae"""]

 def __init__(    self					,						__lowerCamelCase					,						__lowerCamelCase					,						__lowerCamelCase					,						__lowerCamelCase					,						):



     '''simple docstring'''




     super().__init__()
     self.register_modules(unet=__lowerCamelCase					,						scheduler=__lowerCamelCase					,						mel=__lowerCamelCase					,						vqvae=__lowerCamelCase  )

 def 					UpperCamelCase__(    self  ):



     '''simple docstring'''




     return 50 if isinstance(self.scheduler					,						__lowerCamelCase  ) else 1000

 @torch.no_grad()
 def __call__(    self					,						__lowerCamelCase = 1					,						__lowerCamelCase = None					,						__lowerCamelCase = None					,						__lowerCamelCase = 0					,						__lowerCamelCase = 0					,						__lowerCamelCase = None					,						__lowerCamelCase = None					,						__lowerCamelCase = 0					,						__lowerCamelCase = 0					,						__lowerCamelCase = None					,						__lowerCamelCase = 0					,						__lowerCamelCase = None					,						__lowerCamelCase = None					,						__lowerCamelCase=True					,						):



     '''simple docstring'''




     __A    :						Union[str, Any]						 =     steps or self.get_default_steps()
     self.scheduler.set_timesteps(__lowerCamelCase  )
     __A    :						Any						 =     step_generator or generator
     # For backwards compatibility
     if type(self.unet.config.sample_size  ) == int:
         __A    :						Optional[Any]						 =     (self.unet.config.sample_size, self.unet.config.sample_size)
     if noise is None:
         __A    :						Dict						 =     randn_tensor(
             (
                 batch_size,
                 self.unet.config.in_channels,
                 self.unet.config.sample_size[0],
                 self.unet.config.sample_size[1],
             )					,						generator=__lowerCamelCase					,						device=self.device					,						)
     __A    :						Optional[int]						 =     noise
     __A    :						int						 =     None

     if audio_file is not None or raw_audio is not None:
         self.mel.load_audio(__lowerCamelCase					,						__lowerCamelCase  )
         __A    :						int						 =     self.mel.audio_slice_to_image(__lowerCamelCase  )
         __A    :						Optional[Any]						 =     np.frombuffer(input_image.tobytes()					,						dtype='''uint8'''  ).reshape(
             (input_image.height, input_image.width)  )
         __A    :						Union[str, Any]						 =     (input_image / 255) * 2 - 1
         __A    :						Any						 =     torch.tensor(input_image[np.newaxis, :, :]					,						dtype=torch.float  ).to(self.device  )

         if self.vqvae is not None:
             __A    :						Any						 =     self.vqvae.encode(torch.unsqueeze(__lowerCamelCase					,						0  )  ).latent_dist.sample(
                 generator=__lowerCamelCase  )[0]
             __A    :						Optional[Any]						 =     self.vqvae.config.scaling_factor * input_images

         if start_step > 0:
             __A    :						Union[str, Any]						 =     self.scheduler.add_noise(__lowerCamelCase					,						__lowerCamelCase					,						self.scheduler.timesteps[start_step - 1]  )

         __A    :						Optional[int]						 =     (
             self.unet.config.sample_size[1] * self.mel.get_sample_rate() / self.mel.x_res / self.mel.hop_length
         )
         __A    :						int						 =     int(mask_start_secs * pixels_per_second  )
         __A    :						Dict						 =     int(mask_end_secs * pixels_per_second  )
         __A    :						int						 =     self.scheduler.add_noise(__lowerCamelCase					,						__lowerCamelCase					,						torch.tensor(self.scheduler.timesteps[start_step:]  )  )

     for step, t in enumerate(self.progress_bar(self.scheduler.timesteps[start_step:]  )  ):
         if isinstance(self.unet					,						__lowerCamelCase  ):
             __A    :						List[str]						 =     self.unet(__lowerCamelCase					,						__lowerCamelCase					,						__lowerCamelCase  )['''sample''']
         else:
             __A    :						List[str]						 =     self.unet(__lowerCamelCase					,						__lowerCamelCase  )['''sample''']

         if isinstance(self.scheduler					,						__lowerCamelCase  ):
             __A    :						Optional[Any]						 =     self.scheduler.step(
                 model_output=__lowerCamelCase					,						timestep=__lowerCamelCase					,						sample=__lowerCamelCase					,						eta=__lowerCamelCase					,						generator=__lowerCamelCase					,						)['''prev_sample''']
         else:
             __A    :						Optional[Any]						 =     self.scheduler.step(
                 model_output=__lowerCamelCase					,						timestep=__lowerCamelCase					,						sample=__lowerCamelCase					,						generator=__lowerCamelCase					,						)['''prev_sample''']

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

     if self.vqvae is not None:
         # 0.18215 was scaling factor used in training to ensure unit variance
         __A    :						Union[str, Any]						 =     1 / self.vqvae.config.scaling_factor * images
         __A    :						List[str]						 =     self.vqvae.decode(__lowerCamelCase  )['''sample''']

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

     __A    :						Dict						 =     [self.mel.image_to_audio(__lowerCamelCase  ) for _ in images]
     if not return_dict:
         return images, (self.mel.get_sample_rate(), audios)

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

 @torch.no_grad()
 def 					UpperCamelCase__(    self					,						__lowerCamelCase					,						__lowerCamelCase = 50  ):



     '''simple docstring'''




     assert isinstance(self.scheduler					,						__lowerCamelCase  )
     self.scheduler.set_timesteps(__lowerCamelCase  )
     __A    :						str						 =     np.array(
         [np.frombuffer(image.tobytes()					,						dtype='''uint8'''  ).reshape((1, image.height, image.width)  ) for image in images]  )
     __A    :						Any						 =     (sample / 255) * 2 - 1
     __A    :						Dict						 =     torch.Tensor(__lowerCamelCase  ).to(self.device  )

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

     return sample







 @staticmethod
 def 					UpperCamelCase__(    __lowerCamelCase					,						__lowerCamelCase					,						__lowerCamelCase  ):



     '''simple docstring'''




     __A    :						Tuple						 =     acos(torch.dot(torch.flatten(__lowerCamelCase  )					,						torch.flatten(__lowerCamelCase  )  ) / torch.norm(__lowerCamelCase  ) / torch.norm(__lowerCamelCase  )  )
     return sin((1 - alpha) * theta  ) * xa / sin(__lowerCamelCase  ) + sin(alpha * theta  ) * xa / sin(__lowerCamelCase  )




