
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_ 				=			False, False, False





@dataclass
class     __UpperCamelCase :






     """simple docstring"""







     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=lowerCAmelCase__ ,						repr=lowerCAmelCase__       )



     def __call__(				self							:							int ):







       """simple docstring"""




       return self.pa_type



     def  UpperCAmelCase__							(				self							:							str     ,						_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
         __SCREAMING_SNAKE_CASE      :	List[Any]					=		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!)
             __SCREAMING_SNAKE_CASE      :	Optional[Any]					=		np.frombuffer(value['''bytes''']     ,						dtype=np.intaa ).astype(np.floataa ) / 3_2767
           else:
             __SCREAMING_SNAKE_CASE      :	Optional[int]					=		np.memmap(value['''path''']     ,						dtype='''h'''     ,						mode='''r''' ).astype(np.floataa ) / 3_2767

           __SCREAMING_SNAKE_CASE      :	str					=		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  UpperCAmelCase__							(				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.''' )

       __SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE      :	Optional[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

       __SCREAMING_SNAKE_CASE      :	List[Any]					=		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:
         __SCREAMING_SNAKE_CASE      :	str					=		token_per_repo_id or {}
         __SCREAMING_SNAKE_CASE      :	int					=		path.split('''::''' )[-1]
         try:
           __SCREAMING_SNAKE_CASE      :	int					=		string_to_dict(_A     ,						config.HUB_DATASETS_URL )['''repo_id''']
           __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		token_per_repo_id[repo_id]
         except (ValueError, KeyError):
           __SCREAMING_SNAKE_CASE      :	str					=		None

         with xopen(_A     ,						'''rb'''     ,						use_auth_token=_A ) as f:
           __SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		sf.read(_A )

       else:
         __SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE      :	int					=		sf.read(_A )

       __SCREAMING_SNAKE_CASE      :	Tuple					=		array.T
       if self.mono:
         __SCREAMING_SNAKE_CASE      :	List[Any]					=		librosa.to_mono(_A )
       if self.sampling_rate and self.sampling_rate != sampling_rate:
         __SCREAMING_SNAKE_CASE      :	Any					=		librosa.resample(_A     ,						orig_sr=_A     ,						target_sr=self.sampling_rate )
         __SCREAMING_SNAKE_CASE      :	List[Any]					=		self.sampling_rate

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



     def  UpperCAmelCase__							(				self							:							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  UpperCAmelCase__							(				self							:							Any     ,						_A							:							Union[pa.StringArray, pa.StructArray] ):







       """simple docstring"""




       if pa.types.is_string(storage.type ):
         __SCREAMING_SNAKE_CASE      :	str					=		pa.array([None] * len(_A )     ,						type=pa.binary() )
         __SCREAMING_SNAKE_CASE      :	int					=		pa.StructArray.from_arrays([bytes_array, storage]     ,						['''bytes''', '''path''']     ,						mask=storage.is_null() )
       elif pa.types.is_binary(storage.type ):
         __SCREAMING_SNAKE_CASE      :	Dict					=		pa.array([None] * len(_A )     ,						type=pa.string() )
         __SCREAMING_SNAKE_CASE      :	List[Any]					=		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''' ):
         __SCREAMING_SNAKE_CASE      :	Optional[Any]					=		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:
           __SCREAMING_SNAKE_CASE      :	List[str]					=		storage.field('''bytes''' )
         else:
           __SCREAMING_SNAKE_CASE      :	Optional[int]					=		pa.array([None] * len(_A )     ,						type=pa.binary() )
         if storage.type.get_field_index('''path''' ) >= 0:
           __SCREAMING_SNAKE_CASE      :	Tuple					=		storage.field('''path''' )
         else:
           __SCREAMING_SNAKE_CASE      :	int					=		pa.array([None] * len(_A )     ,						type=pa.string() )
         __SCREAMING_SNAKE_CASE      :	List[str]					=		pa.StructArray.from_arrays([bytes_array, path_array]     ,						['''bytes''', '''path''']     ,						mask=storage.is_null() )
       return array_cast(_A     ,						self.pa_type )




     def  UpperCAmelCase__							(				self							:							Union[str, Any]     ,						_A							:							pa.StructArray ):







       """simple docstring"""




       @no_op_if_value_is_null
       def path_to_bytes(_A							:							Tuple ):
         with xopen(_A     ,						'''rb''' ) as f:
           __SCREAMING_SNAKE_CASE      :	List[Any]					=		f.read()
         return bytes_

       __SCREAMING_SNAKE_CASE      :	Optional[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()     ,						)
       __SCREAMING_SNAKE_CASE      :	Tuple					=		pa.array(
           [os.path.basename(_A ) if path is not None else None for path in storage.field('''path''' ).to_pylist()]     ,						type=pa.string()     ,						)
       __SCREAMING_SNAKE_CASE      :	Optional[Any]					=		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

import packaging.version


__a	 =      """examples/"""
__a	 =      {
    """examples""": (re.compile(R"""^check_min_version\(\"[^\"]+\"\)\s*$""", re.MULTILINE), """check_min_version(\"VERSION\")\n"""),
    """init""": (re.compile(R"""^__version__\s+=\s+\"([^\"]+)\"\s*$""", re.MULTILINE), """__version__ = \"VERSION\"\n"""),
    """setup""": (re.compile(R"""^(\s*)version\s*=\s*\"[^\"]+\",""", re.MULTILINE), R"""\1version=\"VERSION\","""),
    """doc""": (re.compile(R"""^(\s*)release\s*=\s*\"[^\"]+\"$""", re.MULTILINE), """release = \"VERSION\"\n"""),
}
__a	 =      {
    """init""": """src/diffusers/__init__.py""",
    """setup""": """setup.py""",
}
__a	 =      """README.md"""
def 	_UpperCamelCase					(    lowerCAmelCase_	,    lowerCAmelCase_	,    lowerCAmelCase_ )    ->Optional[Any]:
   with open(lowerCAmelCase_	,    """r"""	,    encoding="""utf-8"""	,    newline="""\n""" ) as f:
      UpperCAmelCase      					=  f.read()
   UpperCAmelCase					,	UpperCAmelCase      					=  REPLACE_PATTERNS[pattern]
   UpperCAmelCase      					=  replace.replace("""VERSION"""	,    lowerCAmelCase_ )
   UpperCAmelCase      					=  re_pattern.sub(lowerCAmelCase_	,    lowerCAmelCase_ )
   with open(lowerCAmelCase_	,    """w"""	,    encoding="""utf-8"""	,    newline="""\n""" ) as f:
      f.write(lowerCAmelCase_ )
def 	_UpperCamelCase					(    lowerCAmelCase_ )    ->int:
   for folder, directories, fnames in os.walk(lowerCAmelCase_ ):
      # Removing some of the folders with non-actively maintained examples from the walk
      if "research_projects" in directories:
         directories.remove("""research_projects""" )
      if "legacy" in directories:
         directories.remove("""legacy""" )
      for fname in fnames:
         if fname.endswith(""".py""" ):
            update_version_in_file(os.path.join(lowerCAmelCase_	,    lowerCAmelCase_ )	,    lowerCAmelCase_	,    pattern="""examples""" )
def 	_UpperCamelCase					(    lowerCAmelCase_	,    lowerCAmelCase_=False )    ->Optional[int]:
   for pattern, fname in REPLACE_FILES.items():
      update_version_in_file(lowerCAmelCase_	,    lowerCAmelCase_	,    lowerCAmelCase_ )
   if not patch:
      update_version_in_examples(lowerCAmelCase_ )
def 	_UpperCamelCase					(    )    ->List[str]:
   UpperCAmelCase      					=  """🤗 Transformers currently provides the following architectures"""
   UpperCAmelCase      					=  """1. Want to contribute a new model?"""
   with open(lowerCAmelCase_	,    """r"""	,    encoding="""utf-8"""	,    newline="""\n""" ) as f:
      UpperCAmelCase      					=  f.readlines()

   # Find the start of the list.
   UpperCAmelCase      					=  0
   while not lines[start_index].startswith(_start_prompt ):
      start_index += 1
   start_index += 1

   UpperCAmelCase      					=  start_index
   # Update the lines in the model list.
   while not lines[index].startswith(_end_prompt ):
      if lines[index].startswith("""1.""" ):
         UpperCAmelCase      					=  lines[index].replace(
             """https://huggingface.co/docs/diffusers/main/model_doc"""	,    """https://huggingface.co/docs/diffusers/model_doc"""	,    )
      index += 1

   with open(lowerCAmelCase_	,    """w"""	,    encoding="""utf-8"""	,    newline="""\n""" ) as f:
      f.writelines(lowerCAmelCase_ )
def 	_UpperCamelCase					(    )    ->List[str]:
   with open(REPLACE_FILES["""init"""]	,    """r""" ) as f:
      UpperCAmelCase      					=  f.read()
   UpperCAmelCase      					=  REPLACE_PATTERNS["""init"""][0].search(lowerCAmelCase_ ).groups()[0]
   return packaging.version.parse(lowerCAmelCase_ )
def 	_UpperCamelCase					(    lowerCAmelCase_=False )    ->int:
   UpperCAmelCase      					=  get_version()
   if patch and default_version.is_devrelease:
      raise ValueError("""Can't create a patch version from the dev branch, checkout a released version!""" )
   if default_version.is_devrelease:
      UpperCAmelCase      					=  default_version.base_version
   elif patch:
      UpperCAmelCase      					=  F"""{default_version.major}.{default_version.minor}.{default_version.micro + 1}"""
   else:
      UpperCAmelCase      					=  F"""{default_version.major}.{default_version.minor + 1}.0"""

   # Now let's ask nicely if that's the right one.
   UpperCAmelCase      					=  input(F"""Which version are you releasing? [{default_version}]""" )
   if len(lowerCAmelCase_ ) == 0:
      UpperCAmelCase      					=  default_version

   print(F"""Updating version to {version}.""" )
   global_version_update(lowerCAmelCase_	,    patch=lowerCAmelCase_ )
def 	_UpperCamelCase					(    )    ->int:
   UpperCAmelCase      					=  get_version()
   UpperCAmelCase      					=  F"""{current_version.major}.{current_version.minor + 1}.0.dev0"""
   UpperCAmelCase      					=  current_version.base_version

   # Check with the user we got that right.
   UpperCAmelCase      					=  input(F"""Which version are we developing now? [{dev_version}]""" )
   if len(lowerCAmelCase_ ) == 0:
      UpperCAmelCase      					=  dev_version

   print(F"""Updating version to {version}.""" )
   global_version_update(lowerCAmelCase_ )


#    print("Cleaning main README, don't forget to run `make fix-copies`.")
#    clean_main_ref_in_model_list()


if __name__ == "__main__":
 __a	 =      argparse.ArgumentParser()
 parser.add_argument("""--post_release""", action="""store_true""", help="""Whether this is pre or post release.""")
 parser.add_argument("""--patch""", action="""store_true""", help="""Whether or not this is a patch release.""")
 __a	 =      parser.parse_args()
 if not args.post_release:
  pre_release_work(patch=args.patch)
 elif args.patch:
  print("""Nothing to do after a patch :-)""")
 else:
  post_release_work()



