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




@dataclass
class 		SCREAMING_SNAKE_CASE     :







				'''simple docstring'''
				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=SCREAMING_SNAKE_CASE   , repr=SCREAMING_SNAKE_CASE	)





				def __call__(						self					:						Tuple					)							->List[Any]:


										'''simple docstring'''




										return self.pa_type





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

																						_UpperCamelCase			:			Tuple   			=						BytesIO(bytes()					)
																						sf.write(lowercase__       ,		lowercase__       ,		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					:						Dict       ,		lowercase__					:						dict       ,		lowercase__					:						Optional[Dict[str, Union[str, bool, None]]] = None					)							->dict:


										'''simple docstring'''




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

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

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

										else:
																_UpperCamelCase			,       _UpperCamelCase			:			Optional[int]   			=						sf.read(lowercase__					)

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

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





				def  snake_case__				(						self					:						int					)							->Union["FeatureType", Dict[str, "FeatureType"]]:


										'''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					:						Dict       ,		lowercase__					:						Union[pa.StringArray, pa.StructArray]					)							->pa.StructArray:


										'''simple docstring'''




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







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


										'''simple docstring'''




										@no_op_if_value_is_null
										def path_to_bytes(lowercase__					:						Any					):
																with xopen(lowercase__       ,		"rb"					) as f:
																						_UpperCamelCase			:			Optional[Any]   			=						f.read()
																return bytes_

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







				'''simple docstring'''





				def __init__(						self					:						Any       ,		lowercase__					:						Dict       ,		lowercase__					:						int=3       ,		lowercase__					:						Optional[Any]=32       ,		lowercase__					:						Optional[int]=3       ,		lowercase__					:						Any=10       ,		lowercase__					:						Tuple=[10, 20, 30, 40]       ,		lowercase__					:						List[str]=[1, 1, 2, 1]       ,		lowercase__					:						int=True       ,		lowercase__					:						int=True       ,		lowercase__					:						int="relu"       ,		lowercase__					:						List[str]=3       ,		lowercase__					:						Dict=None       ,		)							->List[Any]:


										'''simple docstring'''




										_UpperCamelCase			:			Tuple   			=						parent
										_UpperCamelCase			:			Tuple   			=						batch_size
										_UpperCamelCase			:			List[str]   			=						image_size
										_UpperCamelCase			:			Union[str, Any]   			=						num_channels
										_UpperCamelCase			:			List[str]   			=						embeddings_size
										_UpperCamelCase			:			Union[str, Any]   			=						hidden_sizes
										_UpperCamelCase			:			List[Any]   			=						depths
										_UpperCamelCase			:			Any   			=						is_training
										_UpperCamelCase			:			Union[str, Any]   			=						use_labels
										_UpperCamelCase			:			Tuple   			=						hidden_act
										_UpperCamelCase			:			int   			=						num_labels
										_UpperCamelCase			:			Union[str, Any]   			=						scope
										_UpperCamelCase			:			Union[str, Any]   			=						len(lowercase__					)





				def  snake_case__				(						self					:						int					)							->List[Any]:


										'''simple docstring'''




										_UpperCamelCase			:			str   			=						floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]					)

										_UpperCamelCase			:			int   			=						None
										if self.use_labels:
																_UpperCamelCase			:			Optional[int]   			=						ids_tensor([self.batch_size]       ,		self.num_labels					)

										_UpperCamelCase			:			List[str]   			=						self.get_config()

										return config, pixel_values, labels





				def  snake_case__				(						self					:						Any					)							->Dict:


										'''simple docstring'''




										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					:						Tuple       ,		lowercase__					:						Tuple       ,		lowercase__					:						Any       ,		lowercase__					:						Any					)							->Any:


										'''simple docstring'''




										_UpperCamelCase			:			Any   			=						TFResNetModel(config=lowercase__					)
										_UpperCamelCase			:			Dict   			=						model(lowercase__					)
										# 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					:						int       ,		lowercase__					:						str       ,		lowercase__					:						int       ,		lowercase__					:						Any					)							->Optional[int]:


										'''simple docstring'''




										_UpperCamelCase			:			str   			=						self.num_labels
										_UpperCamelCase			:			List[str]   			=						TFResNetForImageClassification(lowercase__					)
										_UpperCamelCase			:			Tuple   			=						model(lowercase__       ,		labels=lowercase__					)
										self.parent.assertEqual(result.logits.shape       ,		(self.batch_size, self.num_labels)					)







				def  snake_case__				(						self					:						int					)							->List[Any]:


										'''simple docstring'''




										_UpperCamelCase			:			List[str]   			=						self.prepare_config_and_inputs()
										_UpperCamelCase			,       _UpperCamelCase			,       _UpperCamelCase			:			List[str]   			=						config_and_inputs
										_UpperCamelCase			:			List[str]   			=						{"pixel_values": pixel_values}
										return config, inputs_dict





@require_tf
class 		SCREAMING_SNAKE_CASE     (					SCREAMING_SNAKE_CASE   , SCREAMING_SNAKE_CASE   , unittest.TestCase	):







				'''simple docstring'''
				UpperCAmelCase__							=							(TFResNetModel, TFResNetForImageClassification) if is_tf_available() else ()
				UpperCAmelCase__							=							(
				    {'''feature-extraction''': TFResNetModel, '''image-classification''': TFResNetForImageClassification}
				    if is_tf_available()
				    else {}
				)

				UpperCAmelCase__							=							False
				UpperCAmelCase__							=							False
				UpperCAmelCase__							=							False
				UpperCAmelCase__							=							False
				UpperCAmelCase__							=							False





				def  snake_case__				(						self					:						int					)							->Any:


										'''simple docstring'''




										_UpperCamelCase			:			int   			=						TFResNetModelTester(self					)
										_UpperCamelCase			:			Union[str, Any]   			=						ConfigTester(self       ,		config_class=lowercase__       ,		has_text_modality=lowercase__					)





				def  snake_case__				(						self					:						Union[str, Any]					)							->List[Any]:


										'''simple docstring'''




										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					:						Dict					)							->List[str]:


										'''simple docstring'''




										return





				@unittest.skip(reason="ResNet does not use inputs_embeds"					)
				def  snake_case__				(						self					:						Union[str, Any]					)							->str:


										'''simple docstring'''




										pass





				@unittest.skip(reason="ResNet does not support input and output embeddings"					)
				def  snake_case__				(						self					:						Optional[Any]					)							->Union[str, Any]:


										'''simple docstring'''




										pass





				def  snake_case__				(						self					:						Tuple					)							->Optional[int]:


										'''simple docstring'''




										_UpperCamelCase			,       _UpperCamelCase			:			Any   			=						self.model_tester.prepare_config_and_inputs_for_common()

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

																_UpperCamelCase			:			Union[str, Any]   			=						["pixel_values"]
																self.assertListEqual(arg_names[:1]       ,		lowercase__					)





				def  snake_case__				(						self					:						Optional[int]					)							->Optional[Any]:


										'''simple docstring'''




										_UpperCamelCase			:			Dict   			=						self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_model(*lowercase__					)





				def  snake_case__				(						self					:						List[str]					)							->str:


										'''simple docstring'''




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

																_UpperCamelCase			:			List[Any]   			=						outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

																_UpperCamelCase			:			Union[str, Any]   			=						self.model_tester.num_stages
																self.assertEqual(len(lowercase__					)       ,		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]       ,		)

										_UpperCamelCase			,       _UpperCamelCase			:			Dict   			=						self.model_tester.prepare_config_and_inputs_for_common()
										_UpperCamelCase			:			Any   			=						["basic", "bottleneck"]
										for model_class in self.all_model_classes:
																for layer_type in layers_type:
																						_UpperCamelCase			:			Optional[Any]   			=						layer_type
																						_UpperCamelCase			:			str   			=						True
																						check_hidden_states_output(lowercase__       ,		lowercase__       ,		lowercase__					)

																						# check that output_hidden_states also work using config
																						del inputs_dict["output_hidden_states"]
																						_UpperCamelCase			:			int   			=						True

																						check_hidden_states_output(lowercase__       ,		lowercase__       ,		lowercase__					)





				def  snake_case__				(						self					:						Optional[int]					)							->str:


										'''simple docstring'''




										_UpperCamelCase			:			List[Any]   			=						self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_for_image_classification(*lowercase__					)







				@slow
				def  snake_case__				(						self					:						List[Any]					)							->Dict:


										'''simple docstring'''




										for model_name in TF_RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																_UpperCamelCase			:			Optional[Any]   			=						TFResNetModel.from_pretrained(lowercase__					)
																self.assertIsNotNone(lowercase__					)





def 						__A							(     )     ->			Dict:




						'''simple docstring'''
						_UpperCamelCase			:			Optional[int]   			=						Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"					)
						return image






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







				'''simple docstring'''





				@cached_property
				def  snake_case__				(						self					:						List[Any]					)							->Union[str, Any]:


										'''simple docstring'''




										return (
										    AutoImageProcessor.from_pretrained(TF_RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[0]					)
										    if is_vision_available()
										    else None
										)







				@slow
				def  snake_case__				(						self					:						Optional[int]					)							->List[str]:


										'''simple docstring'''




										_UpperCamelCase			:			Dict   			=						TFResNetForImageClassification.from_pretrained(TF_RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[0]					)

										_UpperCamelCase			:			Optional[int]   			=						self.default_image_processor
										_UpperCamelCase			:			Optional[Any]   			=						prepare_img()
										_UpperCamelCase			:			str   			=						image_processor(images=lowercase__       ,		return_tensors="tf"					)

										# forward pass
										_UpperCamelCase			:			List[str]   			=						model(**lowercase__					)

										# verify the logits
										_UpperCamelCase			:			Optional[Any]   			=						tf.TensorShape((1, 1_000)					)
										self.assertEqual(outputs.logits.shape       ,		lowercase__					)

										_UpperCamelCase			:			Optional[int]   			=						tf.constant([-1_1.1_0_6_9, -9.7_8_7_7, -8.3_7_7_7]					)

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


