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


__A   ,__A   ,__A			   = False, False, False


@dataclass
class     UpperCAmelCase						:



    """simple docstring"""

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






    def __call__(   self							):
       return self.pa_type






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

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

       lowercase__, lowercase__:   Union[str, 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

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

          with xopen(_UpperCAmelCase    ,			'''rb'''    ,			use_auth_token=_UpperCAmelCase							) as f:
             lowercase__, lowercase__:   List[Any]					=      sf.read(_UpperCAmelCase							)

       else:
          lowercase__, lowercase__:   Tuple					=      sf.read(_UpperCAmelCase							)

       lowercase__:   Tuple					=      array.T
       if self.mono:
          lowercase__:   Dict					=      librosa.to_mono(_UpperCAmelCase							)
       if self.sampling_rate and self.sampling_rate != sampling_rate:
          lowercase__:   Optional[Any]					=      librosa.resample(_UpperCAmelCase    ,			orig_sr=_UpperCAmelCase    ,			target_sr=self.sampling_rate							)
          lowercase__:   Tuple					=      self.sampling_rate

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






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






    def _snake_case			(   self    ,			_UpperCAmelCase							):

       @no_op_if_value_is_null
       def path_to_bytes(_UpperCAmelCase							):
          with xopen(_UpperCAmelCase    ,			'''rb'''							) as f:
             lowercase__:   Tuple					=      f.read()
          return bytes_

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

"""simple docstring"""






from __future__ import annotations

import inspect
import unittest

import numpy as np

from transformers import ResNetConfig
from transformers.testing_utils import require_tf, require_vision, slow
from transformers.utils import cached_property, is_tf_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
       import tensorflow as tf

       from transformers import TFResNetForImageClassification, TFResNetModel
       from transformers.models.resnet.modeling_tf_resnet import TF_RESNET_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
       from PIL import Image

       from transformers import AutoImageProcessor


class     UpperCAmelCase						:



    """simple docstring"""

    def __init__(   self    ,			_UpperCAmelCase    ,			_UpperCAmelCase=3    ,			_UpperCAmelCase=32    ,			_UpperCAmelCase=3    ,			_UpperCAmelCase=10    ,			_UpperCAmelCase=[10, 20, 30, 40]    ,			_UpperCAmelCase=[1, 1, 2, 1]    ,			_UpperCAmelCase=True    ,			_UpperCAmelCase=True    ,			_UpperCAmelCase="relu"    ,			_UpperCAmelCase=3    ,			_UpperCAmelCase=None    ,			):
       lowercase__:   Optional[Any]					=      parent
       lowercase__:   Union[str, Any]					=      batch_size
       lowercase__:   int					=      image_size
       lowercase__:   Optional[Any]					=      num_channels
       lowercase__:   Optional[int]					=      embeddings_size
       lowercase__:   Dict					=      hidden_sizes
       lowercase__:   Union[str, Any]					=      depths
       lowercase__:   str					=      is_training
       lowercase__:   Optional[int]					=      use_labels
       lowercase__:   List[str]					=      hidden_act
       lowercase__:   Dict					=      num_labels
       lowercase__:   Any					=      scope
       lowercase__:   Optional[Any]					=      len(_UpperCAmelCase							)






    def _snake_case			(   self							):
       lowercase__:   Union[str, Any]					=      floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]							)

       lowercase__:   List[Any]					=      None
       if self.use_labels:
          lowercase__:   Any					=      ids_tensor([self.batch_size]    ,			self.num_labels							)

       lowercase__:   Optional[int]					=      self.get_config()

       return config, pixel_values, labels






    def _snake_case			(   self							):
       return ResNetConfig(
           num_channels=self.num_channels    ,			embeddings_size=self.embeddings_size    ,			hidden_sizes=self.hidden_sizes    ,			depths=self.depths    ,			hidden_act=self.hidden_act    ,			num_labels=self.num_labels    ,			image_size=self.image_size    ,			)






    def _snake_case			(   self    ,			_UpperCAmelCase    ,			_UpperCAmelCase    ,			_UpperCAmelCase							):
       lowercase__:   Optional[Any]					=      TFResNetModel(config=_UpperCAmelCase							)
       lowercase__:   Dict					=      model(_UpperCAmelCase							)
       # expected last hidden states: B, C, H // 32, W // 32
       self.parent.assertEqual(
           result.last_hidden_state.shape    ,			(self.batch_size, self.hidden_sizes[-1], self.image_size // 32, self.image_size // 32)    ,			)






    def _snake_case			(   self    ,			_UpperCAmelCase    ,			_UpperCAmelCase    ,			_UpperCAmelCase							):
       lowercase__:   str					=      self.num_labels
       lowercase__:   int					=      TFResNetForImageClassification(_UpperCAmelCase							)
       lowercase__:   Optional[Any]					=      model(_UpperCAmelCase    ,			labels=_UpperCAmelCase							)
       self.parent.assertEqual(result.logits.shape    ,			(self.batch_size, self.num_labels)							)






    def _snake_case			(   self							):
       lowercase__:   int					=      self.prepare_config_and_inputs()
       lowercase__, lowercase__, lowercase__:   Optional[Any]					=      config_and_inputs
       lowercase__:   Union[str, Any]					=      {'''pixel_values''': pixel_values}
       return config, inputs_dict



@require_tf
class     UpperCAmelCase						(_UpperCAmelCase    ,_UpperCAmelCase    ,unittest.TestCase  ):



    """simple docstring"""

    _UpperCAmelCase	:Any						    = (TFResNetModel, TFResNetForImageClassification) if is_tf_available() else ()
    _UpperCAmelCase	:List[str]						    = (
        {"feature-extraction": TFResNetModel, "image-classification": TFResNetForImageClassification}
        if is_tf_available()
        else {}
    )

    _UpperCAmelCase	:Any						    = False
    _UpperCAmelCase	:List[str]						    = False
    _UpperCAmelCase	:Optional[Any]						    = False
    _UpperCAmelCase	:Tuple						    = False
    _UpperCAmelCase	:List[Any]						    = False






    def _snake_case			(   self							):
       lowercase__:   Union[str, Any]					=      TFResNetModelTester(self							)
       lowercase__:   Optional[int]					=      ConfigTester(self    ,			config_class=_UpperCAmelCase    ,			has_text_modality=_UpperCAmelCase							)






    def _snake_case			(   self							):
       self.create_and_test_config_common_properties()
       self.config_tester.create_and_test_config_to_json_string()
       self.config_tester.create_and_test_config_to_json_file()
       self.config_tester.create_and_test_config_from_and_save_pretrained()
       self.config_tester.create_and_test_config_with_num_labels()
       self.config_tester.check_config_can_be_init_without_params()
       self.config_tester.check_config_arguments_init()






    def _snake_case			(   self							):
       return






    @unittest.skip(reason='''ResNet does not use inputs_embeds'''							)
    def _snake_case			(   self							):
       pass






    @unittest.skip(reason='''ResNet does not support input and output embeddings'''							)
    def _snake_case			(   self							):
       pass






    def _snake_case			(   self							):
       lowercase__, lowercase__:   int					=      self.model_tester.prepare_config_and_inputs_for_common()

       for model_class in self.all_model_classes:
          lowercase__:   Optional[Any]					=      model_class(_UpperCAmelCase							)
          lowercase__:   str					=      inspect.signature(model.call							)
          # signature.parameters is an OrderedDict => so arg_names order is deterministic
          lowercase__:   int					=      [*signature.parameters.keys()]

          lowercase__:   Optional[Any]					=      ['''pixel_values''']
          self.assertListEqual(arg_names[:1]    ,			_UpperCAmelCase							)






    def _snake_case			(   self							):
       lowercase__:   Any					=      self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_model(*_UpperCAmelCase							)






    def _snake_case			(   self							):
       def check_hidden_states_output(_UpperCAmelCase    ,			_UpperCAmelCase    ,			_UpperCAmelCase							):
          lowercase__:   Union[str, Any]					=      model_class(_UpperCAmelCase							)
          lowercase__:   List[Any]					=      model(**self._prepare_for_class(_UpperCAmelCase    ,			_UpperCAmelCase							)							)

          lowercase__:   Union[str, Any]					=      outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

          lowercase__:   Tuple					=      self.model_tester.num_stages
          self.assertEqual(len(_UpperCAmelCase							)    ,			expected_num_stages + 1							)

          # ResNet's feature maps are of shape (batch_size, num_channels, height, width)
          self.assertListEqual(
              list(hidden_states[0].shape[-2:]							)    ,			[self.model_tester.image_size // 4, self.model_tester.image_size // 4]    ,			)

       lowercase__, lowercase__:   Optional[Any]					=      self.model_tester.prepare_config_and_inputs_for_common()
       lowercase__:   Union[str, Any]					=      ['''basic''', '''bottleneck''']
       for model_class in self.all_model_classes:
          for layer_type in layers_type:
             lowercase__:   Tuple					=      layer_type
             lowercase__:   Any					=      True
             check_hidden_states_output(_UpperCAmelCase    ,			_UpperCAmelCase    ,			_UpperCAmelCase							)

             # check that output_hidden_states also work using config
             del inputs_dict["output_hidden_states"]
             lowercase__:   List[str]					=      True

             check_hidden_states_output(_UpperCAmelCase    ,			_UpperCAmelCase    ,			_UpperCAmelCase							)






    def _snake_case			(   self							):
       lowercase__:   Optional[int]					=      self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_for_image_classification(*_UpperCAmelCase							)






    @slow
    def _snake_case			(   self							):
       for model_name in TF_RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
          lowercase__:   Dict					=      TFResNetModel.from_pretrained(_UpperCAmelCase							)
          self.assertIsNotNone(_UpperCAmelCase							)



def SCREAMING_SNAKE_CASE__			(   )  ->  List[Any]:
   lowercase__:   List[Any]					=      Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png''' )
   return image




@require_tf
@require_vision
class     UpperCAmelCase						(unittest.TestCase  ):



    """simple docstring"""

    @cached_property
    def _snake_case			(   self							):
       return (
           AutoImageProcessor.from_pretrained(TF_RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[0]							)
           if is_vision_available()
           else None
       )






    @slow
    def _snake_case			(   self							):
       lowercase__:   Tuple					=      TFResNetForImageClassification.from_pretrained(TF_RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[0]							)

       lowercase__:   Any					=      self.default_image_processor
       lowercase__:   List[Any]					=      prepare_img()
       lowercase__:   List[Any]					=      image_processor(images=_UpperCAmelCase    ,			return_tensors='''tf'''							)

       # forward pass
       lowercase__:   Dict					=      model(**_UpperCAmelCase							)

       # verify the logits
       lowercase__:   int					=      tf.TensorShape((1, 1000)							)
       self.assertEqual(outputs.logits.shape    ,			_UpperCAmelCase							)

       lowercase__:   Dict					=      tf.constant([-11.1_069, -9.7_877, -8.3_777]							)

       self.assertTrue(np.allclose(outputs.logits[0, :3].numpy()    ,			_UpperCAmelCase    ,			atol=1e-4							)							)
