'''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


_lowerCAmelCase				,    _lowerCAmelCase				,    _lowerCAmelCase  :Optional[Any]  				=    False, False, False






@dataclass
class 				UpperCAmelCase :




    '''simple docstring'''







    snake_case__							:   Optional[int]							=						None
    snake_case__							:   bool							=						True
    snake_case__							:   bool							=						True
    snake_case__							:   Optional[str]							=						None
    # Automatically constructed
    snake_case__							:   ClassVar[str]							=						"dict"
    snake_case__							:   ClassVar[Any]							=						pa.struct({"bytes": pa.binary(), "path": pa.string()} )
    snake_case__							:   str							=						field(default="Audio"   ,    init=a__   ,    repr=a__ )






    def __call__(       self )							-> List[str]:
         return self.pa_type






    def       _UpperCamelCase    (       self    ,							lowercase__ )							-> dict:
         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(lowerCamelCase_    ,							lowerCamelCase_ ):
              return {"bytes": None, "path": value}
         elif isinstance(lowerCamelCase_    ,							lowerCamelCase_ ):
              return {"bytes": value, "path": None}
         elif "array" in value:
              # convert the audio array to wav bytes
              SCREAMING_SNAKE_CASE					:			List[str]    	=       BytesIO()
              sf.write(lowerCamelCase_    ,							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					:			Any    	=       np.frombuffer(value['bytes']    ,							dtype=np.intaa ).astype(np.floataa ) / 32_767
                   else:
                        SCREAMING_SNAKE_CASE					:			Union[str, Any]    	=       np.memmap(value['path']    ,							dtype='h'    ,							mode='r' ).astype(np.floataa ) / 32_767

                   SCREAMING_SNAKE_CASE					:			Union[str, Any]    	=       BytesIO(bytes() )
                   sf.write(lowerCamelCase_    ,							lowerCamelCase_    ,							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    ,							lowercase__    ,							lowercase__ = None )							-> dict:
         if not self.decode:
              raise RuntimeError('Decoding is disabled for this feature. Please use Audio(decode=True) instead.' )

         SCREAMING_SNAKE_CASE	,   SCREAMING_SNAKE_CASE					:			Dict    	=       (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					:			Tuple    	=       xsplitext(lowerCamelCase_ )[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					:			int    	=       token_per_repo_id or {}
              SCREAMING_SNAKE_CASE					:			Optional[int]    	=       path.split('::' )[-1]
              try:
                   SCREAMING_SNAKE_CASE					:			int    	=       string_to_dict(lowerCamelCase_    ,							config.HUB_DATASETS_URL )['repo_id']
                   SCREAMING_SNAKE_CASE					:			Any    	=       token_per_repo_id[repo_id]
              except (ValueError, KeyError):
                   SCREAMING_SNAKE_CASE					:			int    	=       None

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

         else:
              SCREAMING_SNAKE_CASE	,   SCREAMING_SNAKE_CASE					:			List[str]    	=       sf.read(lowerCamelCase_ )

         SCREAMING_SNAKE_CASE					:			str    	=       array.T
         if self.mono:
              SCREAMING_SNAKE_CASE					:			Optional[int]    	=       librosa.to_mono(lowerCamelCase_ )
         if self.sampling_rate and self.sampling_rate != sampling_rate:
              SCREAMING_SNAKE_CASE					:			Dict    	=       librosa.resample(lowerCamelCase_    ,							orig_sr=lowerCamelCase_    ,							target_sr=self.sampling_rate )
              SCREAMING_SNAKE_CASE					:			Dict    	=       self.sampling_rate

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






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






    def       _UpperCamelCase    (       self    ,							lowercase__ )							-> pa.StructArray:

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

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




'''simple docstring'''





import copy
import json
import os
import tempfile

from transformers import is_torch_available

from .test_configuration_utils import config_common_kwargs






class 				UpperCAmelCase (       _SCREAMING_SNAKE_CASE ):




    '''simple docstring'''







    def __init__(       self    ,							lowercase__    ,							lowercase__=None    ,							lowercase__=True    ,							lowercase__=None    ,							**lowercase__ )							-> Optional[Any]:
         SCREAMING_SNAKE_CASE					:			int    	=       parent
         SCREAMING_SNAKE_CASE					:			str    	=       config_class
         SCREAMING_SNAKE_CASE					:			Union[str, Any]    	=       has_text_modality
         SCREAMING_SNAKE_CASE					:			Any    	=       kwargs
         SCREAMING_SNAKE_CASE					:			Union[str, Any]    	=       common_properties






    def       _UpperCamelCase    (       self )							-> Union[str, Any]:
         SCREAMING_SNAKE_CASE					:			Optional[int]    	=       self.config_class(**self.inputs_dict )
         SCREAMING_SNAKE_CASE					:			int    	=       (
             ['hidden_size', 'num_attention_heads', 'num_hidden_layers']
             if self.common_properties is None
             else self.common_properties
         )

         # Add common fields for text models
         if self.has_text_modality:
              common_properties.extend(['vocab_size'] )

         # Test that config has the common properties as getters
         for prop in common_properties:
              self.parent.assertTrue(hasattr(lowercase__    ,							lowercase__ )    ,							msg=F"""`{prop}` does not exist""" )

         # Test that config has the common properties as setter
         for idx, name in enumerate(lowercase__ ):
              try:
                   setattr(lowercase__    ,							lowercase__    ,							lowercase__ )
                   self.parent.assertEqual(
                       getattr(lowercase__    ,							lowercase__ )    ,							lowercase__    ,							msg=F"""`{name} value {idx} expected, but was {getattr(lowercase__    ,							lowercase__ )}""" )
              except NotImplementedError:
                   # Some models might not be able to implement setters for common_properties
                   # In that case, a NotImplementedError is raised
                   pass

        # Test if config class can be called with Config(prop_name=..)
         for idx, name in enumerate(lowercase__ ):
              try:
                   SCREAMING_SNAKE_CASE					:			Optional[int]    	=       self.config_class(**{name: idx} )
                   self.parent.assertEqual(
                       getattr(lowercase__    ,							lowercase__ )    ,							lowercase__    ,							msg=F"""`{name} value {idx} expected, but was {getattr(lowercase__    ,							lowercase__ )}""" )
              except NotImplementedError:
                   # Some models might not be able to implement setters for common_properties
                   # In that case, a NotImplementedError is raised
                   pass






    def       _UpperCamelCase    (       self )							-> List[str]:
         SCREAMING_SNAKE_CASE					:			Optional[Any]    	=       self.config_class(**self.inputs_dict )
         SCREAMING_SNAKE_CASE					:			Any    	=       json.loads(config.to_json_string() )
         for key, value in self.inputs_dict.items():
              self.parent.assertEqual(obj[key]    ,							lowercase__ )






    def       _UpperCamelCase    (       self )							-> Optional[int]:
         SCREAMING_SNAKE_CASE					:			Tuple    	=       self.config_class(**self.inputs_dict )

         with tempfile.TemporaryDirectory() as tmpdirname:
              SCREAMING_SNAKE_CASE					:			List[str]    	=       os.path.join(lowercase__    ,							'config.json' )
              config_first.to_json_file(lowercase__ )
              SCREAMING_SNAKE_CASE					:			Union[str, Any]    	=       self.config_class.from_json_file(lowercase__ )

         self.parent.assertEqual(config_second.to_dict()    ,							config_first.to_dict() )






    def       _UpperCamelCase    (       self )							-> Dict:
         SCREAMING_SNAKE_CASE					:			Optional[Any]    	=       self.config_class(**self.inputs_dict )

         with tempfile.TemporaryDirectory() as tmpdirname:
              config_first.save_pretrained(lowercase__ )
              SCREAMING_SNAKE_CASE					:			Optional[int]    	=       self.config_class.from_pretrained(lowercase__ )

         self.parent.assertEqual(config_second.to_dict()    ,							config_first.to_dict() )






    def       _UpperCamelCase    (       self )							-> int:
         SCREAMING_SNAKE_CASE					:			List[Any]    	=       self.config_class(**self.inputs_dict )

         SCREAMING_SNAKE_CASE					:			Optional[int]    	=       'test'
         with tempfile.TemporaryDirectory() as tmpdirname:
              SCREAMING_SNAKE_CASE					:			Tuple    	=       os.path.join(lowercase__    ,							lowercase__ )
              config_first.save_pretrained(lowercase__ )
              SCREAMING_SNAKE_CASE					:			Dict    	=       self.config_class.from_pretrained(lowercase__    ,							subfolder=lowercase__ )

         self.parent.assertEqual(config_second.to_dict()    ,							config_first.to_dict() )






    def       _UpperCamelCase    (       self )							-> List[Any]:
         SCREAMING_SNAKE_CASE					:			Dict    	=       self.config_class(**self.inputs_dict    ,							num_labels=5 )
         self.parent.assertEqual(len(config.idalabel )    ,							5 )
         self.parent.assertEqual(len(config.labelaid )    ,							5 )

         SCREAMING_SNAKE_CASE					:			Union[str, Any]    	=       3
         self.parent.assertEqual(len(config.idalabel )    ,							3 )
         self.parent.assertEqual(len(config.labelaid )    ,							3 )






    def       _UpperCamelCase    (       self )							-> Dict:
         if self.config_class.is_composition:
              return
         SCREAMING_SNAKE_CASE					:			int    	=       self.config_class()
         self.parent.assertIsNotNone(lowercase__ )






    def       _UpperCamelCase    (       self )							-> Tuple:
         SCREAMING_SNAKE_CASE					:			Union[str, Any]    	=       copy.deepcopy(lowercase__ )
         SCREAMING_SNAKE_CASE					:			Any    	=       self.config_class(**lowercase__ )
         SCREAMING_SNAKE_CASE					:			str    	=       []
         for key, value in config_common_kwargs.items():
              if key == "torch_dtype":
                   if not is_torch_available():
                        continue
                   else:
                        import torch

                        if config.torch_dtype != torch.floataa:
                             wrong_values.append(('torch_dtype', config.torch_dtype, torch.floataa) )
              elif getattr(lowercase__    ,							lowercase__ ) != value:
                   wrong_values.append((key, getattr(lowercase__    ,							lowercase__ ), value) )

         if len(lowercase__ ) > 0:
              SCREAMING_SNAKE_CASE					:			List[Any]    	=       '\n'.join([F"""- {v[0]}: got {v[1]} instead of {v[2]}""" for v in wrong_values] )
              raise ValueError(F"""The following keys were not properly set in the config:\n{errors}""" )






    def       _UpperCamelCase    (       self )							-> Optional[int]:
         self.create_and_test_config_common_properties()
         self.create_and_test_config_to_json_string()
         self.create_and_test_config_to_json_file()
         self.create_and_test_config_from_and_save_pretrained()
         self.create_and_test_config_from_and_save_pretrained_subfolder()
         self.create_and_test_config_with_num_labels()
         self.check_config_can_be_init_without_params()
         self.check_config_arguments_init()



