

import inspect
import unittest

from datasets import load_dataset
from packaging import version

from transformers import BeitConfig
from transformers.models.auto import get_values
from transformers.testing_utils import require_torch, require_torch_multi_gpu, require_vision, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, _config_zero_init, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
     import torch
     from torch import nn

     from transformers import (
         MODEL_MAPPING,
         BeitForImageClassification,
         BeitForMaskedImageModeling,
         BeitForSemanticSegmentation,
         BeitModel,
     )
     from transformers.models.beit.modeling_beit import BEIT_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
     import PIL
     from PIL import Image

     from transformers import BeitImageProcessor





class __SCREAMING_SNAKE_CASE     :






    def __init__(       self,		_a,		_a=1_00,		_a=13,		_a=30,		_a=2,		_a=3,		_a=True,		_a=True,		_a=32,		_a=4,		_a=4,		_a=37,		_a="gelu",		_a=0.1,		_a=0.1,		_a=10,		_a=0.02,		_a=3,		_a=None,		_a=[0, 1, 2, 3],		)   ->					Union[str, Any]:
       __SCREAMING_SNAKE_CASE       =				parent
       __SCREAMING_SNAKE_CASE       =				1_00
       __SCREAMING_SNAKE_CASE       =				batch_size
       __SCREAMING_SNAKE_CASE       =				image_size
       __SCREAMING_SNAKE_CASE       =				patch_size
       __SCREAMING_SNAKE_CASE       =				num_channels
       __SCREAMING_SNAKE_CASE       =				is_training
       __SCREAMING_SNAKE_CASE       =				use_labels
       __SCREAMING_SNAKE_CASE       =				hidden_size
       __SCREAMING_SNAKE_CASE       =				num_hidden_layers
       __SCREAMING_SNAKE_CASE       =				num_attention_heads
       __SCREAMING_SNAKE_CASE       =				intermediate_size
       __SCREAMING_SNAKE_CASE       =				hidden_act
       __SCREAMING_SNAKE_CASE       =				hidden_dropout_prob
       __SCREAMING_SNAKE_CASE       =				attention_probs_dropout_prob
       __SCREAMING_SNAKE_CASE       =				type_sequence_label_size
       __SCREAMING_SNAKE_CASE       =				initializer_range
       __SCREAMING_SNAKE_CASE       =				scope
       __SCREAMING_SNAKE_CASE       =				out_indices
       __SCREAMING_SNAKE_CASE       =				num_labels

       # in BeiT, the seq length equals the number of patches + 1 (we add 1 for the [CLS] token)
       __SCREAMING_SNAKE_CASE       =				(image_size // patch_size) ** 2
       __SCREAMING_SNAKE_CASE       =				num_patches + 1






    def 						__lowerCAmelCase    (       self )   ->					Optional[Any]:
       __SCREAMING_SNAKE_CASE       =				floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size] )

       __SCREAMING_SNAKE_CASE       =				None
       __SCREAMING_SNAKE_CASE       =				None
       if self.use_labels:
          __SCREAMING_SNAKE_CASE       =				ids_tensor([self.batch_size],		self.type_sequence_label_size )
          __SCREAMING_SNAKE_CASE       =				ids_tensor([self.batch_size, self.image_size, self.image_size],		self.num_labels )

       __SCREAMING_SNAKE_CASE       =				self.get_config()

       return config, pixel_values, labels, pixel_labels






    def 						__lowerCAmelCase    (       self )   ->					str:
       return BeitConfig(
           vocab_size=self.vocab_size,		image_size=self.image_size,		patch_size=self.patch_size,		num_channels=self.num_channels,		hidden_size=self.hidden_size,		num_hidden_layers=self.num_hidden_layers,		num_attention_heads=self.num_attention_heads,		intermediate_size=self.intermediate_size,		hidden_act=self.hidden_act,		hidden_dropout_prob=self.hidden_dropout_prob,		attention_probs_dropout_prob=self.attention_probs_dropout_prob,		is_decoder=__lowerCamelCase,		initializer_range=self.initializer_range,		out_indices=self.out_indices,		)






    def 						__lowerCAmelCase    (       self,		_a,		_a,		_a,		_a )   ->					List[str]:
       __SCREAMING_SNAKE_CASE       =				BeitModel(config=__lowerCamelCase )
       model.to(__lowerCamelCase )
       model.eval()
       __SCREAMING_SNAKE_CASE       =				model(__lowerCamelCase )
       self.parent.assertEqual(result.last_hidden_state.shape,		(self.batch_size, self.seq_length, self.hidden_size) )






    def 						__lowerCAmelCase    (       self,		_a,		_a,		_a,		_a )   ->					Optional[int]:
       __SCREAMING_SNAKE_CASE       =				BeitForMaskedImageModeling(config=__lowerCamelCase )
       model.to(__lowerCamelCase )
       model.eval()
       __SCREAMING_SNAKE_CASE       =				model(__lowerCamelCase )
       self.parent.assertEqual(result.logits.shape,		(self.batch_size, self.seq_length - 1, self.vocab_size) )






    def 						__lowerCAmelCase    (       self,		_a,		_a,		_a,		_a )   ->					List[Any]:
       __SCREAMING_SNAKE_CASE       =				self.type_sequence_label_size
       __SCREAMING_SNAKE_CASE       =				BeitForImageClassification(__lowerCamelCase )
       model.to(__lowerCamelCase )
       model.eval()
       __SCREAMING_SNAKE_CASE       =				model(__lowerCamelCase,		labels=__lowerCamelCase )
       self.parent.assertEqual(result.logits.shape,		(self.batch_size, self.type_sequence_label_size) )

       # test greyscale images
       __SCREAMING_SNAKE_CASE       =				1
       __SCREAMING_SNAKE_CASE       =				BeitForImageClassification(__lowerCamelCase )
       model.to(__lowerCamelCase )
       model.eval()

       __SCREAMING_SNAKE_CASE       =				floats_tensor([self.batch_size, 1, self.image_size, self.image_size] )
       __SCREAMING_SNAKE_CASE       =				model(__lowerCamelCase,		labels=__lowerCamelCase )
       self.parent.assertEqual(result.logits.shape,		(self.batch_size, self.type_sequence_label_size) )






    def 						__lowerCAmelCase    (       self,		_a,		_a,		_a,		_a )   ->					Optional[int]:
       __SCREAMING_SNAKE_CASE       =				self.num_labels
       __SCREAMING_SNAKE_CASE       =				BeitForSemanticSegmentation(__lowerCamelCase )
       model.to(__lowerCamelCase )
       model.eval()
       __SCREAMING_SNAKE_CASE       =				model(__lowerCamelCase )
       self.parent.assertEqual(
           result.logits.shape,		(self.batch_size, self.num_labels, self.image_size * 2, self.image_size * 2) )
       __SCREAMING_SNAKE_CASE       =				model(__lowerCamelCase,		labels=__lowerCamelCase )
       self.parent.assertEqual(
           result.logits.shape,		(self.batch_size, self.num_labels, self.image_size * 2, self.image_size * 2) )






    def 						__lowerCAmelCase    (       self )   ->					Tuple:
       __SCREAMING_SNAKE_CASE       =				self.prepare_config_and_inputs()
       __SCREAMING_SNAKE_CASE       =				config_and_inputs
       __SCREAMING_SNAKE_CASE       =				{'''pixel_values''': pixel_values}
       return config, inputs_dict






@require_torch
class __SCREAMING_SNAKE_CASE     (	_snake_case			,						_snake_case			,						unittest.TestCase		):
    SCREAMING_SNAKE_CASE__      =(
        (BeitModel, BeitForImageClassification, BeitForMaskedImageModeling, BeitForSemanticSegmentation)
        if is_torch_available()
        else ()
    )
    SCREAMING_SNAKE_CASE__      =(
        {
            """feature-extraction""": BeitModel,
            """image-classification""": BeitForImageClassification,
            """image-segmentation""": BeitForSemanticSegmentation,
        }
        if is_torch_available()
        else {}
    )

    SCREAMING_SNAKE_CASE__      =False
    SCREAMING_SNAKE_CASE__      =False
    SCREAMING_SNAKE_CASE__      =False






    def 						__lowerCAmelCase    (       self )   ->					str:
       __SCREAMING_SNAKE_CASE       =				BeitModelTester(self )
       __SCREAMING_SNAKE_CASE       =				ConfigTester(self,		config_class=__lowerCamelCase,		has_text_modality=__lowerCamelCase,		hidden_size=37 )






    def 						__lowerCAmelCase    (       self )   ->					List[Any]:
       self.config_tester.run_common_tests()






    @unittest.skip(reason="BEiT does not use inputs_embeds" )
    def 						__lowerCAmelCase    (       self )   ->					Optional[Any]:
       pass






    @require_torch_multi_gpu
    @unittest.skip(reason="BEiT has some layers using `add_module` which doesn\'t work well with `nn.DataParallel`" )
    def 						__lowerCAmelCase    (       self )   ->					Optional[int]:
       pass






    def 						__lowerCAmelCase    (       self )   ->					Optional[int]:
       __SCREAMING_SNAKE_CASE       =				self.model_tester.prepare_config_and_inputs_for_common()

       for model_class in self.all_model_classes:
          __SCREAMING_SNAKE_CASE       =				model_class(__lowerCamelCase )
          self.assertIsInstance(model.get_input_embeddings(),		(nn.Module) )
          __SCREAMING_SNAKE_CASE       =				model.get_output_embeddings()
          self.assertTrue(x is None or isinstance(__lowerCamelCase,		nn.Linear ) )






    def 						__lowerCAmelCase    (       self )   ->					List[Any]:
       __SCREAMING_SNAKE_CASE       =				self.model_tester.prepare_config_and_inputs_for_common()

       for model_class in self.all_model_classes:
          __SCREAMING_SNAKE_CASE       =				model_class(__lowerCamelCase )
          __SCREAMING_SNAKE_CASE       =				inspect.signature(model.forward )
          # signature.parameters is an OrderedDict => so arg_names order is deterministic
          __SCREAMING_SNAKE_CASE       =				[*signature.parameters.keys()]

          __SCREAMING_SNAKE_CASE       =				['''pixel_values''']
          self.assertListEqual(arg_names[:1],		__lowerCamelCase )






    def 						__lowerCAmelCase    (       self )   ->					Any:
       __SCREAMING_SNAKE_CASE       =				self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_model(*__lowerCamelCase )






    def 						__lowerCAmelCase    (       self )   ->					Tuple:
       __SCREAMING_SNAKE_CASE       =				self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_for_masked_lm(*__lowerCamelCase )






    def 						__lowerCAmelCase    (       self )   ->					str:
       __SCREAMING_SNAKE_CASE       =				self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_for_image_classification(*__lowerCamelCase )






    def 						__lowerCAmelCase    (       self )   ->					Dict:
       __SCREAMING_SNAKE_CASE       =				self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_for_semantic_segmentation(*__lowerCamelCase )






    def 						__lowerCAmelCase    (       self )   ->					str:
       if not self.model_tester.is_training:
          return

       __SCREAMING_SNAKE_CASE       =				self.model_tester.prepare_config_and_inputs_for_common()
       __SCREAMING_SNAKE_CASE       =				True

       for model_class in self.all_model_classes:
          # we don't test BeitForMaskedImageModeling
          if model_class in [*get_values(__lowerCamelCase ), BeitForMaskedImageModeling]:
             continue

          __SCREAMING_SNAKE_CASE       =				model_class(__lowerCamelCase )
          model.to(__lowerCamelCase )
          model.train()
          __SCREAMING_SNAKE_CASE       =				self._prepare_for_class(__lowerCamelCase,		__lowerCamelCase,		return_labels=__lowerCamelCase )
          __SCREAMING_SNAKE_CASE       =				model(**__lowerCamelCase ).loss
          loss.backward()






    def 						__lowerCAmelCase    (       self )   ->					Any:
       __SCREAMING_SNAKE_CASE       =				self.model_tester.prepare_config_and_inputs_for_common()
       if not self.model_tester.is_training:
          return

       __SCREAMING_SNAKE_CASE       =				False
       __SCREAMING_SNAKE_CASE       =				True

       for model_class in self.all_model_classes:
          # we don't test BeitForMaskedImageModeling
          if (
              model_class in [*get_values(__lowerCamelCase ), BeitForMaskedImageModeling]
              or not model_class.supports_gradient_checkpointing
          ):
             continue

          __SCREAMING_SNAKE_CASE       =				model_class(__lowerCamelCase )
          model.gradient_checkpointing_enable()
          model.to(__lowerCamelCase )
          model.train()
          __SCREAMING_SNAKE_CASE       =				self._prepare_for_class(__lowerCamelCase,		__lowerCamelCase,		return_labels=__lowerCamelCase )
          __SCREAMING_SNAKE_CASE       =				model(**__lowerCamelCase ).loss
          loss.backward()






    def 						__lowerCAmelCase    (       self )   ->					Tuple:
       __SCREAMING_SNAKE_CASE       =				self.model_tester.prepare_config_and_inputs_for_common()

       __SCREAMING_SNAKE_CASE       =				_config_zero_init(__lowerCamelCase )
       for model_class in self.all_model_classes:
          __SCREAMING_SNAKE_CASE       =				model_class(config=__lowerCamelCase )
          for name, param in model.named_parameters():
             # we skip lambda parameters as these require special initial values
             # determined by config.layer_scale_init_value
             if "lambda" in name:
                continue
             if param.requires_grad:
                self.assertIn(
                    ((param.data.mean() * 1E9).round() / 1E9).item(),		[0.0, 1.0],		msg=f'''Parameter {name} of model {model_class} seems not properly initialized''',		)






    @slow
    def 						__lowerCAmelCase    (       self )   ->					List[Any]:
       for model_name in BEIT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
          __SCREAMING_SNAKE_CASE       =				BeitModel.from_pretrained(__lowerCamelCase )
          self.assertIsNotNone(__lowerCamelCase )






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



   """simple docstring"""


   __SCREAMING_SNAKE_CASE       =				Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"  )
   return image





@require_torch
@require_vision
class __SCREAMING_SNAKE_CASE     (	unittest.TestCase		):






    @cached_property
    def 						__lowerCAmelCase    (       self )   ->					Any:
       return BeitImageProcessor.from_pretrained("microsoft/beit-base-patch16-224" ) if is_vision_available() else None






    @slow
    def 						__lowerCAmelCase    (       self )   ->					int:
       __SCREAMING_SNAKE_CASE       =				BeitForMaskedImageModeling.from_pretrained("microsoft/beit-base-patch16-224-pt22k" ).to(__lowerCamelCase )

       __SCREAMING_SNAKE_CASE       =				self.default_image_processor
       __SCREAMING_SNAKE_CASE       =				prepare_img()
       __SCREAMING_SNAKE_CASE       =				image_processor(images=__lowerCamelCase,		return_tensors="pt" ).pixel_values.to(__lowerCamelCase )

       # prepare bool_masked_pos
       __SCREAMING_SNAKE_CASE       =				torch.ones((1, 1_96),		dtype=torch.bool ).to(__lowerCamelCase )

       # forward pass
       with torch.no_grad():
          __SCREAMING_SNAKE_CASE       =				model(pixel_values=__lowerCamelCase,		bool_masked_pos=__lowerCamelCase )
       __SCREAMING_SNAKE_CASE       =				outputs.logits

       # verify the logits
       __SCREAMING_SNAKE_CASE       =				torch.Size((1, 1_96, 81_92) )
       self.assertEqual(logits.shape,		__lowerCamelCase )

       __SCREAMING_SNAKE_CASE       =				torch.tensor(
           [[-3.2437, 0.5072, -13.9174], [-3.2456, 0.4948, -13.9401], [-3.2033, 0.5121, -13.8550]] ).to(__lowerCamelCase )

       self.assertTrue(torch.allclose(logits[bool_masked_pos][:3, :3],		__lowerCamelCase,		atol=1E-2 ) )






    @slow
    def 						__lowerCAmelCase    (       self )   ->					Optional[Any]:
       __SCREAMING_SNAKE_CASE       =				BeitForImageClassification.from_pretrained("microsoft/beit-base-patch16-224" ).to(__lowerCamelCase )

       __SCREAMING_SNAKE_CASE       =				self.default_image_processor
       __SCREAMING_SNAKE_CASE       =				prepare_img()
       __SCREAMING_SNAKE_CASE       =				image_processor(images=__lowerCamelCase,		return_tensors="pt" ).to(__lowerCamelCase )

       # forward pass
       with torch.no_grad():
          __SCREAMING_SNAKE_CASE       =				model(**__lowerCamelCase )
       __SCREAMING_SNAKE_CASE       =				outputs.logits

       # verify the logits
       __SCREAMING_SNAKE_CASE       =				torch.Size((1, 10_00) )
       self.assertEqual(logits.shape,		__lowerCamelCase )

       __SCREAMING_SNAKE_CASE       =				torch.tensor([-1.2385, -1.0987, -1.0108] ).to(__lowerCamelCase )

       self.assertTrue(torch.allclose(logits[0, :3],		__lowerCamelCase,		atol=1E-4 ) )

       __SCREAMING_SNAKE_CASE       =				2_81
       self.assertEqual(logits.argmax(-1 ).item(),		__lowerCamelCase )






    @slow
    def 						__lowerCAmelCase    (       self )   ->					List[Any]:
       __SCREAMING_SNAKE_CASE       =				BeitForImageClassification.from_pretrained("microsoft/beit-large-patch16-224-pt22k-ft22k" ).to(
           __lowerCamelCase )

       __SCREAMING_SNAKE_CASE       =				self.default_image_processor
       __SCREAMING_SNAKE_CASE       =				prepare_img()
       __SCREAMING_SNAKE_CASE       =				image_processor(images=__lowerCamelCase,		return_tensors="pt" ).to(__lowerCamelCase )

       # forward pass
       with torch.no_grad():
          __SCREAMING_SNAKE_CASE       =				model(**__lowerCamelCase )
       __SCREAMING_SNAKE_CASE       =				outputs.logits

       # verify the logits
       __SCREAMING_SNAKE_CASE       =				torch.Size((1, 2_18_41) )
       self.assertEqual(logits.shape,		__lowerCamelCase )

       __SCREAMING_SNAKE_CASE       =				torch.tensor([1.6881, -0.2787, 0.5901] ).to(__lowerCamelCase )

       self.assertTrue(torch.allclose(logits[0, :3],		__lowerCamelCase,		atol=1E-4 ) )

       __SCREAMING_SNAKE_CASE       =				23_96
       self.assertEqual(logits.argmax(-1 ).item(),		__lowerCamelCase )






    @slow
    def 						__lowerCAmelCase    (       self )   ->					Any:
       __SCREAMING_SNAKE_CASE       =				BeitForSemanticSegmentation.from_pretrained("microsoft/beit-base-finetuned-ade-640-640" )
       __SCREAMING_SNAKE_CASE       =				model.to(__lowerCamelCase )

       __SCREAMING_SNAKE_CASE       =				BeitImageProcessor(do_resize=__lowerCamelCase,		size=6_40,		do_center_crop=__lowerCamelCase )

       __SCREAMING_SNAKE_CASE       =				load_dataset("hf-internal-testing/fixtures_ade20k",		split="test" )
       __SCREAMING_SNAKE_CASE       =				Image.open(ds[0]["file"] )
       __SCREAMING_SNAKE_CASE       =				image_processor(images=__lowerCamelCase,		return_tensors="pt" ).to(__lowerCamelCase )

       # forward pass
       with torch.no_grad():
          __SCREAMING_SNAKE_CASE       =				model(**__lowerCamelCase )
       __SCREAMING_SNAKE_CASE       =				outputs.logits

       # verify the logits
       __SCREAMING_SNAKE_CASE       =				torch.Size((1, 1_50, 1_60, 1_60) )
       self.assertEqual(logits.shape,		__lowerCamelCase )

       __SCREAMING_SNAKE_CASE       =				version.parse(PIL.__version__ ) < version.parse("9.0.0" )

       if is_pillow_less_than_a:
          __SCREAMING_SNAKE_CASE       =				torch.tensor(
              [
                  [[-4.9225, -2.3954, -3.0522], [-2.8822, -1.0046, -1.7561], [-2.9549, -1.3228, -2.1347]],
                  [[-5.8168, -3.4129, -4.0778], [-3.8651, -2.2214, -3.0277], [-3.8356, -2.4643, -3.3535]],
                  [[-0.0078, 3.9952, 4.0754], [2.9856, 4.6944, 5.0035], [3.2413, 4.7813, 4.9969]],
              ],		device=__lowerCamelCase,		)
       else:
          __SCREAMING_SNAKE_CASE       =				torch.tensor(
              [
                  [[-4.8960, -2.3688, -3.0355], [-2.8478, -0.9836, -1.7418], [-2.9449, -1.3332, -2.1456]],
                  [[-5.8081, -3.4124, -4.1006], [-3.8561, -2.2081, -3.0323], [-3.8365, -2.4601, -3.3669]],
                  [[-0.0309, 3.9868, 4.0540], [2.9640, 4.6877, 4.9976], [3.2081, 4.7690, 4.9942]],
              ],		device=__lowerCamelCase,		)

       self.assertTrue(torch.allclose(logits[0, :3, :3, :3],		__lowerCamelCase,		atol=1E-4 ) )






    @slow
    def 						__lowerCAmelCase    (       self )   ->					Union[str, Any]:
       __SCREAMING_SNAKE_CASE       =				BeitForSemanticSegmentation.from_pretrained("microsoft/beit-base-finetuned-ade-640-640" )
       __SCREAMING_SNAKE_CASE       =				model.to(__lowerCamelCase )

       __SCREAMING_SNAKE_CASE       =				BeitImageProcessor(do_resize=__lowerCamelCase,		size=6_40,		do_center_crop=__lowerCamelCase )

       __SCREAMING_SNAKE_CASE       =				load_dataset("hf-internal-testing/fixtures_ade20k",		split="test" )
       __SCREAMING_SNAKE_CASE       =				Image.open(ds[0]["file"] )
       __SCREAMING_SNAKE_CASE       =				image_processor(images=__lowerCamelCase,		return_tensors="pt" ).to(__lowerCamelCase )

       # forward pass
       with torch.no_grad():
          __SCREAMING_SNAKE_CASE       =				model(**__lowerCamelCase )

       __SCREAMING_SNAKE_CASE       =				outputs.logits.detach().cpu()

       __SCREAMING_SNAKE_CASE       =				image_processor.post_process_semantic_segmentation(outputs=__lowerCamelCase,		target_sizes=[(5_00, 3_00)] )
       __SCREAMING_SNAKE_CASE       =				torch.Size((5_00, 3_00) )
       self.assertEqual(segmentation[0].shape,		__lowerCamelCase )

       __SCREAMING_SNAKE_CASE       =				image_processor.post_process_semantic_segmentation(outputs=__lowerCamelCase )
       __SCREAMING_SNAKE_CASE       =				torch.Size((1_60, 1_60) )
       self.assertEqual(segmentation[0].shape,		__lowerCamelCase )





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




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


 def __call__(      self							:	Any ):
    return self.pa_type


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

          UpperCAmelCase__	:Optional[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        __SCREAMING_SNAKE_CASE					(      self							:	List[Any]			,       __lowerCamelCase							:	dict			,       __lowerCamelCase							:	Optional[Dict[str, Union[str, bool, None]]] = None ):
    if not self.decode:
       raise RuntimeError('''Decoding is disabled for this feature. Please use Audio(decode=True) instead.''' )

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

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

    else:
       UpperCAmelCase__   ,						UpperCAmelCase__	:List[Any]      =      sf.read(__lowerCamelCase )

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

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


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





 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[Any]			,       __lowerCamelCase							:	pa.StructArray ):

    @no_op_if_value_is_null
    def path_to_bytes(__lowerCamelCase							:	Dict ):
       with xopen(__lowerCamelCase			,       '''rb''' ) as f:
          UpperCAmelCase__	:Any      =      f.read()
       return bytes_

    UpperCAmelCase__	:Union[str, 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()			,       )
    UpperCAmelCase__	:Optional[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()			,       )
    UpperCAmelCase__	:Optional[int]      =      pa.StructArray.from_arrays([bytes_array, path_array]			,       ['''bytes''', '''path''']			,       mask=bytes_array.is_null() )
    return array_cast(__lowerCamelCase			,       self.pa_type )



