



"""simple docstring"""






import inspect
import unittest

from transformers import ResNetConfig
from transformers.testing_utils import require_torch, require_vision, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_vision_available

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


if is_torch_available():
		import torch
		from torch import nn

		from transformers import ResNetBackbone, ResNetForImageClassification, ResNetModel
		from transformers.models.resnet.modeling_resnet import 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				:    Optional[Any],    lowerCamelCase				:    Dict,    lowerCamelCase				:    Any=3,    lowerCamelCase				:    List[Any]=32,    lowerCamelCase				:    Dict=3,    lowerCamelCase				:    Optional[Any]=10,    lowerCamelCase				:    List[Any]=[10, 20, 30, 40],    lowerCamelCase				:    Any=[1, 1, 2, 1],    lowerCamelCase				:    Optional[Any]=True,    lowerCamelCase				:    str=True,    lowerCamelCase				:    Tuple="relu",    lowerCamelCase				:    Dict=3,    lowerCamelCase				:    Union[str, Any]=None,    lowerCamelCase				:    List[str]=["stage2", "stage3", "stage4"],    lowerCamelCase				:    Any=[2, 3, 4],    )->   List[Any]:
								lowerCamelCase__						:						Union[str, Any]						      =parent
								lowerCamelCase__						:						str						      =batch_size
								lowerCamelCase__						:						Tuple						      =image_size
								lowerCamelCase__						:						List[str]						      =num_channels
								lowerCamelCase__						:						Dict						      =embeddings_size
								lowerCamelCase__						:						Optional[int]						      =hidden_sizes
								lowerCamelCase__						:						Dict						      =depths
								lowerCamelCase__						:						Optional[int]						      =is_training
								lowerCamelCase__						:						Any						      =use_labels
								lowerCamelCase__						:						List[Any]						      =hidden_act
								lowerCamelCase__						:						Union[str, Any]						      =num_labels
								lowerCamelCase__						:						Tuple						      =scope
								lowerCamelCase__						:						Optional[int]						      =len(lowerCamelCase	)
								lowerCamelCase__						:						Any						      =out_features
								lowerCamelCase__						:						Optional[Any]						      =out_indices



					def 						snake_case			(			self				:    List[str]	)->   Union[str, Any]:
								lowerCamelCase__						:						Tuple						      =floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]	)

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

								lowerCamelCase__						:						str						      =self.get_config()

								return config, pixel_values, labels



					def 						snake_case			(			self				:    Any	)->   str:
								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,    out_features=self.out_features,    out_indices=self.out_indices,    )



					def 						snake_case			(			self				:    str,    lowerCamelCase				:    Optional[Any],    lowerCamelCase				:    Any,    lowerCamelCase				:    str	)->   Dict:
								lowerCamelCase__						:						Optional[int]						      =ResNetModel(config=lowerCamelCase	)
								model.to(lowerCamelCase	)
								model.eval()
								lowerCamelCase__						:						List[str]						      =model(lowerCamelCase	)
								# 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				:    Dict,    lowerCamelCase				:    Optional[int],    lowerCamelCase				:    Optional[int],    lowerCamelCase				:    List[str]	)->   Optional[Any]:
								lowerCamelCase__						:						Any						      =self.num_labels
								lowerCamelCase__						:						Optional[int]						      =ResNetForImageClassification(lowerCamelCase	)
								model.to(lowerCamelCase	)
								model.eval()
								lowerCamelCase__						:						int						      =model(lowerCamelCase,    labels=lowerCamelCase	)
								self.parent.assertEqual(result.logits.shape,    (self.batch_size, self.num_labels)	)



					def 						snake_case			(			self				:    List[str],    lowerCamelCase				:    List[Any],    lowerCamelCase				:    str,    lowerCamelCase				:    Any	)->   Tuple:
								lowerCamelCase__						:						str						      =ResNetBackbone(config=lowerCamelCase	)
								model.to(lowerCamelCase	)
								model.eval()
								lowerCamelCase__						:						List[str]						      =model(lowerCamelCase	)

								# verify feature maps
								self.parent.assertEqual(len(result.feature_maps	),    len(config.out_features	)	)
								self.parent.assertListEqual(list(result.feature_maps[0].shape	),    [self.batch_size, self.hidden_sizes[1], 4, 4]	)

								# verify channels
								self.parent.assertEqual(len(model.channels	),    len(config.out_features	)	)
								self.parent.assertListEqual(model.channels,    config.hidden_sizes[1:]	)

								# verify backbone works with out_features=None
								lowerCamelCase__						:						Optional[int]						      =None
								lowerCamelCase__						:						Tuple						      =ResNetBackbone(config=lowerCamelCase	)
								model.to(lowerCamelCase	)
								model.eval()
								lowerCamelCase__						:						int						      =model(lowerCamelCase	)

								# verify feature maps
								self.parent.assertEqual(len(result.feature_maps	),    1	)
								self.parent.assertListEqual(list(result.feature_maps[0].shape	),    [self.batch_size, self.hidden_sizes[-1], 1, 1]	)

								# verify channels
								self.parent.assertEqual(len(model.channels	),    1	)
								self.parent.assertListEqual(model.channels,    [config.hidden_sizes[-1]]	)




					def 						snake_case			(			self				:    List[Any]	)->   str:
								lowerCamelCase__						:						Tuple						      =self.prepare_config_and_inputs()
								lowerCamelCase__      ,  lowerCamelCase__      ,  lowerCamelCase__						:						Tuple						      =config_and_inputs
								lowerCamelCase__						:						List[Any]						      ={'''pixel_values''': pixel_values}
								return config, inputs_dict





@require_torch
class 			__SCREAMING_SNAKE_CASE      ( lowerCAmelCase_						,  lowerCAmelCase_						,  unittest.TestCase  ):


					'''simple docstring'''





					_a       =				(
					    (
					        ResNetModel,
					        ResNetForImageClassification,
					        ResNetBackbone,
					    )
					    if is_torch_available()
					    else ()
					)
					_a       =				(
					    {'feature-extraction': ResNetModel, 'image-classification': ResNetForImageClassification}
					    if is_torch_available()
					    else {}
					)

					_a       =				True
					_a       =				False
					_a       =				False
					_a       =				False
					_a       =				False



					def 						snake_case			(			self				:    Tuple	)->   str:
								lowerCamelCase__						:						Dict						      =ResNetModelTester(self	)
								lowerCamelCase__						:						Tuple						      =ConfigTester(self,    config_class=lowerCamelCase,    has_text_modality=lowerCamelCase	)



					def 						snake_case			(			self				:    Dict	)->   Dict:
								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				:    Any	)->   Tuple:
								return



					@unittest.skip(reason='''ResNet does not use inputs_embeds'''	)
					def 						snake_case			(			self				:    Tuple	)->   Optional[int]:
								pass



					@unittest.skip(reason='''ResNet does not support input and output embeddings'''	)
					def 						snake_case			(			self				:    Any	)->   Dict:
								pass



					def 						snake_case			(			self				:    Tuple	)->   Dict:
								lowerCamelCase__      ,  lowerCamelCase__						:						Optional[int]						      =self.model_tester.prepare_config_and_inputs_for_common()

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

											lowerCamelCase__						:						Union[str, Any]						      =['''pixel_values''']
											self.assertListEqual(arg_names[:1],    lowerCamelCase	)



					def 						snake_case			(			self				:    str	)->   str:
								lowerCamelCase__						:						str						      =self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_model(*lowerCamelCase	)



					def 						snake_case			(			self				:    Dict	)->   List[Any]:
								lowerCamelCase__						:						Union[str, Any]						      =self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_backbone(*lowerCamelCase	)



					def 						snake_case			(			self				:    List[Any]	)->   Tuple:
								lowerCamelCase__      ,  lowerCamelCase__						:						Optional[int]						      =self.model_tester.prepare_config_and_inputs_for_common()

								for model_class in self.all_model_classes:
											lowerCamelCase__						:						str						      =model_class(config=lowerCamelCase	)
											for name, module in model.named_modules():
														if isinstance(lowerCamelCase,    (nn.BatchNormad, nn.GroupNorm)	):
																	self.assertTrue(
																	    torch.all(module.weight == 1	),    msg=F'''Parameter {name} of model {model_class} seems not properly initialized''',    )
																	self.assertTrue(
																	    torch.all(module.bias == 0	),    msg=F'''Parameter {name} of model {model_class} seems not properly initialized''',    )



					def 						snake_case			(			self				:    List[str]	)->   Tuple:
								def check_hidden_states_output(lowerCamelCase				:    Tuple,    lowerCamelCase				:    Any,    lowerCamelCase				:    int	):
											lowerCamelCase__						:						List[str]						      =model_class(lowerCamelCase	)
											model.to(lowerCamelCase	)
											model.eval()

											with torch.no_grad():
														lowerCamelCase__						:						Optional[int]						      =model(**self._prepare_for_class(lowerCamelCase,    lowerCamelCase	)	)

											lowerCamelCase__						:						int						      =outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

											lowerCamelCase__						:						Any						      =self.model_tester.num_stages
											self.assertEqual(len(lowerCamelCase	),    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],    )

								lowerCamelCase__      ,  lowerCamelCase__						:						Optional[Any]						      =self.model_tester.prepare_config_and_inputs_for_common()
								lowerCamelCase__						:						List[str]						      =['''basic''', '''bottleneck''']
								for model_class in self.all_model_classes:
											for layer_type in layers_type:
														lowerCamelCase__						:						List[Any]						      =layer_type
														lowerCamelCase__						:						List[Any]						      =True
														check_hidden_states_output(lowerCamelCase,    lowerCamelCase,    lowerCamelCase	)

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

														check_hidden_states_output(lowerCamelCase,    lowerCamelCase,    lowerCamelCase	)



					@unittest.skip(reason='''ResNet does not use feedforward chunking'''	)
					def 						snake_case			(			self				:    List[str]	)->   int:
								pass



					def 						snake_case			(			self				:    Optional[Any]	)->   str:
								lowerCamelCase__						:						Tuple						      =self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_image_classification(*lowerCamelCase	)




					@slow
					def 						snake_case			(			self				:    int	)->   Optional[int]:
								for model_name in RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
											lowerCamelCase__						:						Dict						      =ResNetModel.from_pretrained(lowerCamelCase	)
											self.assertIsNotNone(lowerCamelCase	)






def 	snake_case__  (						):
			"""simple docstring"""
			lowerCamelCase__						:						str						      =Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''      )
			return image

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


					'''simple docstring'''





					@cached_property
					def 						snake_case			(			self				:    Tuple	)->   Dict:
								return (
								    AutoImageProcessor.from_pretrained(RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[0]	)
								    if is_vision_available()
								    else None
								)




					@slow
					def 						snake_case			(			self				:    List[Any]	)->   Optional[int]:
								lowerCamelCase__						:						List[Any]						      =ResNetForImageClassification.from_pretrained(RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[0]	).to(lowerCamelCase	)

								lowerCamelCase__						:						List[str]						      =self.default_image_processor
								lowerCamelCase__						:						Dict						      =prepare_img()
								lowerCamelCase__						:						Any						      =image_processor(images=lowerCamelCase,    return_tensors='''pt'''	).to(lowerCamelCase	)

								# forward pass
								with torch.no_grad():
											lowerCamelCase__						:						str						      =model(**lowerCamelCase	)

								# verify the logits
								lowerCamelCase__						:						List[Any]						      =torch.Size((1, 1000)	)
								self.assertEqual(outputs.logits.shape,    lowerCamelCase	)

								lowerCamelCase__						:						str						      =torch.tensor([-11.1_069, -9.7_877, -8.3_777]	).to(lowerCamelCase	)

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





@require_torch
class 			__SCREAMING_SNAKE_CASE      ( lowerCAmelCase_						,  unittest.TestCase  ):


					'''simple docstring'''





					_a       =				(ResNetBackbone,) if is_torch_available() else ()
					_a       =				False
					_a       =				ResNetConfig



					def 						snake_case			(			self				:    Optional[Any]	)->   Union[str, Any]:
								lowerCamelCase__						:						Union[str, Any]						      =ResNetModelTester(self	)










import subprocess
from typing import Union

import numpy as np
import requests

from ..utils import add_end_docstrings, is_torch_available, is_torchaudio_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_torch_available():
  from ..models.auto.modeling_auto import MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING

_lowerCAmelCase    : List[str]		= logging.get_logger(__name__)



def     lowerCAmelCase		(		_lowerCAmelCase				: bytes    ,				_lowerCAmelCase				: int						):




 """simple docstring"""







 UpperCAmelCase__							=    F'''{sampling_rate}'''
 UpperCAmelCase__							=    "1"
 UpperCAmelCase__							=    "f32le"
 UpperCAmelCase__							=    [
     "ffmpeg",
     "-i",
     "pipe:0",
     "-ac",
     ac,
     "-ar",
     ar,
     "-f",
     format_for_conversion,
     "-hide_banner",
     "-loglevel",
     "quiet",
     "pipe:1",
 ]

 try:
  UpperCAmelCase__							=    subprocess.Popen(_lowerCAmelCase    ,				stdin=subprocess.PIPE    ,				stdout=subprocess.PIPE						)
 except FileNotFoundError:
  raise ValueError("ffmpeg was not found but is required to load audio files from filename"						)
 UpperCAmelCase__							=    ffmpeg_process.communicate(_lowerCAmelCase						)
 UpperCAmelCase__							=    output_stream[0]

 UpperCAmelCase__							=    np.frombuffer(_lowerCAmelCase    ,				np.floataa						)
 if audio.shape[0] == 0:
  raise ValueError("Malformed soundfile"						)
 return audio


@add_end_docstrings(lowerCAmelCase		)
class       _UpperCamelCase   (					lowerCAmelCase		):
      def __init__(      self :Any			,   *lowerCamelCase :Any			,   **lowerCamelCase :Optional[Any]   )  ->      Optional[int]:
       # Default, might be overriden by the model.config.
       UpperCAmelCase__							=    5
       super().__init__(*lowerCamelCase			,   **lowerCamelCase   )

       if self.framework != "pt":
        raise ValueError(f'''The {self.__class__} is only available in PyTorch.'''   )

       self.check_model_type(lowerCamelCase   )
      def __call__(      self :Optional[Any]			,   lowerCamelCase :Union[np.ndarray, bytes, str]			,   **lowerCamelCase :List[Any]			,   )  ->      Tuple:
       return super().__call__(lowerCamelCase			,   **lowerCamelCase   )
      def 						UpperCAmelCase_    (      self :str			,   lowerCamelCase :Dict=None			,   **lowerCamelCase :str   )  ->      int:
       # No parameters on this pipeline right now
       UpperCAmelCase__							=    {}
       if top_k is not None:
        if top_k > self.model.config.num_labels:
         UpperCAmelCase__							=    self.model.config.num_labels
        UpperCAmelCase__							=    top_k
       return {}, {}, postprocess_params
      def 						UpperCAmelCase_    (      self :Optional[Any]			,   lowerCamelCase :Dict   )  ->      Optional[int]:
       if isinstance(lowerCamelCase			,   lowerCamelCase   ):
        if inputs.startswith("http://"   ) or inputs.startswith("https://"   ):
         # We need to actually check for a real protocol, otherwise it's impossible to use a local file
         # like http_huggingface_co.png
         UpperCAmelCase__							=    requests.get(lowerCamelCase   ).content
        else:
         with open(lowerCamelCase			,   "rb"   ) as f:
          UpperCAmelCase__							=    f.read()

       if isinstance(lowerCamelCase			,   lowerCamelCase   ):
        UpperCAmelCase__							=    ffmpeg_read(lowerCamelCase			,   self.feature_extractor.sampling_rate   )

       if isinstance(lowerCamelCase			,   lowerCamelCase   ):
        # Accepting `"array"` which is the key defined in `datasets` for
        # better integration
        if not ("sampling_rate" in inputs and ("raw" in inputs or "array" in inputs)):
         raise ValueError(
             "When passing a dictionary to AudioClassificationPipeline, the dict needs to contain a "
             "\"raw\" key containing the numpy array representing the audio and a \"sampling_rate\" key, "
             "containing the sampling_rate associated with that array"   )

        UpperCAmelCase__							=    inputs.pop("raw"			,   lowerCamelCase   )
        if _inputs is None:
         # Remove path which will not be used from `datasets`.
         inputs.pop("path"			,   lowerCamelCase   )
         UpperCAmelCase__							=    inputs.pop("array"			,   lowerCamelCase   )
        UpperCAmelCase__							=    inputs.pop("sampling_rate"   )
        UpperCAmelCase__							=    _inputs
        if in_sampling_rate != self.feature_extractor.sampling_rate:
         import torch

         if is_torchaudio_available():
          from torchaudio import functional as F
         else:
          raise ImportError(
              "torchaudio is required to resample audio samples in AudioClassificationPipeline. "
              "The torchaudio package can be installed through: `pip install torchaudio`."   )

         UpperCAmelCase__							=    F.resample(
             torch.from_numpy(lowerCamelCase   )			,   lowerCamelCase			,   self.feature_extractor.sampling_rate   ).numpy()

       if not isinstance(lowerCamelCase			,   np.ndarray   ):
        raise ValueError("We expect a numpy ndarray as input"   )
       if len(inputs.shape   ) != 1:
        raise ValueError("We expect a single channel audio input for AudioClassificationPipeline"   )

       UpperCAmelCase__							=    self.feature_extractor(
           lowerCamelCase			,   sampling_rate=self.feature_extractor.sampling_rate			,   return_tensors="pt"   )
       return processed
      def 						UpperCAmelCase_    (      self :Tuple			,   lowerCamelCase :Union[str, Any]   )  ->      List[Any]:
       UpperCAmelCase__							=    self.model(**lowerCamelCase   )
       return model_outputs







      def 						UpperCAmelCase_    (      self :List[str]			,   lowerCamelCase :Optional[Any]			,   lowerCamelCase :List[Any]=5   )  ->      Tuple:
       UpperCAmelCase__							=    model_outputs.logits[0].softmax(-1   )
       UpperCAmelCase__			,			UpperCAmelCase__							=    probs.topk(lowerCamelCase   )

       UpperCAmelCase__							=    scores.tolist()
       UpperCAmelCase__							=    ids.tolist()

       UpperCAmelCase__							=    [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(lowerCamelCase			,   lowerCamelCase   )]

       return labels





