


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


__snake_case							,	__snake_case							,	__snake_case   							=						False, False, False



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







       def __call__(			self: Optional[int]					):






            '''simple docstring'''




            return self.pa_type







       def 		snake_case_	(			self: Tuple,A_: Union[str, bytes, dict]					):






            '''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(A_,A_					):
                 return {"bytes": None, "path": value}
            elif isinstance(A_,A_					):
                 return {"bytes": value, "path": None}
            elif "array" in value:
                 # convert the audio array to wav bytes
                 __UpperCamelCase =			BytesIO()
                 sf.write(A_,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!)
                           __UpperCamelCase =			np.frombuffer(value['bytes'],dtype=np.intaa					).astype(np.floataa					) / 3_2767
                      else:
                           __UpperCamelCase =			np.memmap(value['path'],dtype='h',mode='r'					).astype(np.floataa					) / 3_2767

                      __UpperCamelCase =			BytesIO(bytes()					)
                      sf.write(A_,A_,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 		snake_case_	(			self: List[Any],A_: dict,A_: Optional[Dict[str, Union[str, bool, None]]] = None					):






            '''simple docstring'''




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

            __UpperCamelCase,    __UpperCamelCase =			(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

            __UpperCamelCase =			xsplitext(A_					)[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:
                 __UpperCamelCase =			token_per_repo_id or {}
                 __UpperCamelCase =			path.split('::'					)[-1]
                 try:
                      __UpperCamelCase =			string_to_dict(A_,config.HUB_DATASETS_URL					)['repo_id']
                      __UpperCamelCase =			token_per_repo_id[repo_id]
                 except (ValueError, KeyError):
                      __UpperCamelCase =			None

                 with xopen(A_,'rb',use_auth_token=A_					) as f:
                      __UpperCamelCase,    __UpperCamelCase =			sf.read(A_					)

            else:
                 __UpperCamelCase,    __UpperCamelCase =			sf.read(A_					)

            __UpperCamelCase =			array.T
            if self.mono:
                 __UpperCamelCase =			librosa.to_mono(A_					)
            if self.sampling_rate and self.sampling_rate != sampling_rate:
                 __UpperCamelCase =			librosa.resample(A_,orig_sr=A_,target_sr=self.sampling_rate					)
                 __UpperCamelCase =			self.sampling_rate

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







       def 		snake_case_	(			self: List[Any]					):






            '''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 		snake_case_	(			self: str,A_: Union[pa.StringArray, pa.StructArray]					):






            '''simple docstring'''




            if pa.types.is_string(storage.type					):
                 __UpperCamelCase =			pa.array([None] * len(A_					),type=pa.binary()					)
                 __UpperCamelCase =			pa.StructArray.from_arrays([bytes_array, storage],['bytes', 'path'],mask=storage.is_null()					)
            elif pa.types.is_binary(storage.type					):
                 __UpperCamelCase =			pa.array([None] * len(A_					),type=pa.string()					)
                 __UpperCamelCase =			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'					):
                 __UpperCamelCase =			pa.array([Audio().encode_example(A_					) 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:
                      __UpperCamelCase =			storage.field('bytes'					)
                 else:
                      __UpperCamelCase =			pa.array([None] * len(A_					),type=pa.binary()					)
                 if storage.type.get_field_index('path'					) >= 0:
                      __UpperCamelCase =			storage.field('path'					)
                 else:
                      __UpperCamelCase =			pa.array([None] * len(A_					),type=pa.string()					)
                 __UpperCamelCase =			pa.StructArray.from_arrays([bytes_array, path_array],['bytes', 'path'],mask=storage.is_null()					)
            return array_cast(A_,self.pa_type					)







       def 		snake_case_	(			self: Optional[Any],A_: pa.StructArray					):






            '''simple docstring'''




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

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






import argparse
import os
import re


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_dummies.py
__snake_case   							=						'''src/diffusers'''

# Matches is_xxx_available()
__snake_case   							=						re.compile(r'''is\_([a-z_]*)_available\(\)''')
# Matches from xxx import bla
__snake_case   							=						re.compile(r'''\s+from\s+\S*\s+import\s+([^\(\s].*)\n''')


__snake_case   							=						'''
{0} = None
'''

__snake_case   							=						'''
class {0}(metaclass=DummyObject):
    _backends = {1}

    def __init__(self, *args, **kwargs):
        requires_backends(self, {1})

    @classmethod
    def from_config(cls, *args, **kwargs):
        requires_backends(cls, {1})

    @classmethod
    def from_pretrained(cls, *args, **kwargs):
        requires_backends(cls, {1})
'''


__snake_case   							=						'''
def {0}(*args, **kwargs):
    requires_backends({0}, {1})
'''






def 		_A		(		_lowercase )					-> int:


     """simple docstring"""


     __UpperCamelCase =			_re_backend.findall(_lowercase )
     if len(_lowercase ) == 0:
          return None

     return "_and_".join(_lowercase )






def 		_A		(		)					-> Tuple:


     """simple docstring"""


     with open(os.path.join(_lowercase					,  '__init__.py' )					,  'r'					,  encoding='utf-8'					,  newline='\n' ) as f:
          __UpperCamelCase =			f.readlines()

     # Get to the point we do the actual imports for type checking
     __UpperCamelCase =			0
     __UpperCamelCase =			{}
     # Go through the end of the file
     while line_index < len(_lowercase ):
          # If the line contains is_backend_available, we grab all objects associated with the `else` block
          __UpperCamelCase =			find_backend(lines[line_index] )
          if backend is not None:
               while not lines[line_index].startswith('else:' ):
                    line_index += 1
               line_index += 1
               __UpperCamelCase =			[]
               # Until we unindent, add backend objects to the list
               while line_index < len(_lowercase ) and len(lines[line_index] ) > 1:
                    __UpperCamelCase =			lines[line_index]
                    __UpperCamelCase =			_re_single_line_import.search(_lowercase )
                    if single_line_import_search is not None:
                         objects.extend(single_line_import_search.groups()[0].split(', ' ) )
                    elif line.startswith(' ' * 8 ):
                         objects.append(line[8:-2] )
                    line_index += 1

               if len(_lowercase ) > 0:
                    __UpperCamelCase =			objects
          else:
               line_index += 1

     return backend_specific_objects






def 		_A		(		_lowercase					,  _lowercase )					-> Union[str, Any]:


     """simple docstring"""


     if name.isupper():
          return DUMMY_CONSTANT.format(_lowercase )
     elif name.islower():
          return DUMMY_FUNCTION.format(_lowercase					,  _lowercase )
     else:
          return DUMMY_CLASS.format(_lowercase					,  _lowercase )






def 		_A		(		_lowercase=None )					-> Optional[Any]:


     """simple docstring"""


     if backend_specific_objects is None:
          __UpperCamelCase =			read_init()
     # For special correspondence backend to module name as used in the function requires_modulename
     __UpperCamelCase =			{}

     for backend, objects in backend_specific_objects.items():
          __UpperCamelCase =			'[' + ', '.join(f'''"{b}"''' for b in backend.split('_and_' ) ) + ']'
          __UpperCamelCase =			'# This file is autogenerated by the command `make fix-copies`, do not edit.\n'
          dummy_file += "from ..utils import DummyObject, requires_backends\n\n"
          dummy_file += "\n".join([create_dummy_object(_lowercase					,  _lowercase ) for o in objects] )
          __UpperCamelCase =			dummy_file

     return dummy_files






def 		_A		(		_lowercase=False )					-> List[str]:


     """simple docstring"""


     __UpperCamelCase =			create_dummy_files()
     # For special correspondence backend to shortcut as used in utils/dummy_xxx_objects.py
     __UpperCamelCase =			{'torch': 'pt'}

     # Locate actual dummy modules and read their content.
     __UpperCamelCase =			os.path.join(_lowercase					,  'utils' )
     __UpperCamelCase =			{
         backend: os.path.join(_lowercase					,  f'''dummy_{short_names.get(_lowercase					,  _lowercase )}_objects.py''' )
         for backend in dummy_files.keys()
     }

     __UpperCamelCase =			{}
     for backend, file_path in dummy_file_paths.items():
          if os.path.isfile(_lowercase ):
               with open(_lowercase					,  'r'					,  encoding='utf-8'					,  newline='\n' ) as f:
                    __UpperCamelCase =			f.read()
          else:
               __UpperCamelCase =			''

     for backend in dummy_files.keys():
          if dummy_files[backend] != actual_dummies[backend]:
               if overwrite:
                    print(
                        f'''Updating diffusers.utils.dummy_{short_names.get(_lowercase					,  _lowercase )}_objects.py as the main '''
                        '__init__ has new objects.' )
                    with open(dummy_file_paths[backend]					,  'w'					,  encoding='utf-8'					,  newline='\n' ) as f:
                         f.write(dummy_files[backend] )
               else:
                    raise ValueError(
                        'The main __init__ has objects that are not present in '
                        f'''diffusers.utils.dummy_{short_names.get(_lowercase					,  _lowercase )}_objects.py. Run `make fix-copies` '''
                        'to fix this.' )


if __name__ == "__main__":
    __snake_case   							=						argparse.ArgumentParser()
    parser.add_argument('''--fix_and_overwrite''', action='''store_true''', help='''Whether to fix inconsistencies.''')
    __snake_case   							=						parser.parse_args()

    check_dummies(args.fix_and_overwrite)


