





'''simple docstring'''





import os
import socket
from contextlib import contextmanager

import torch

from ..commands.config.default import write_basic_config  # noqa: F401
from ..state import PartialState
from .dataclasses import DistributedType
from .imports import is_deepspeed_available, is_tpu_available
from .transformer_engine import convert_model
from .versions import is_torch_version


if is_deepspeed_available():
				from deepspeed import DeepSpeedEngine

if is_tpu_available(check_device=False):
				import torch_xla.core.xla_model as xm







def       UpperCAmelCase							(     _lowerCamelCase      :  List[str]     ):





						'''simple docstring'''


						if is_torch_version("<"			,   "2.0.0"     ) or not hasattr(_lowerCamelCase			,   "_dynamo"     ):
												return False
						return isinstance(_lowerCamelCase			,   torch._dynamo.eval_frame.OptimizedModule     )







def       UpperCAmelCase							(     _lowerCamelCase      :  Any			,   _lowerCamelCase      :  bool = True     ):





						'''simple docstring'''


						SCREAMING_SNAKE_CASE__     :				Any   =					(torch.nn.parallel.DistributedDataParallel, torch.nn.DataParallel)

						SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					is_compiled_module(_lowerCamelCase     )
						if is_compiled:
												SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					model
												SCREAMING_SNAKE_CASE__     :				List[str]   =					model._orig_mod

						if is_deepspeed_available():
												options += (DeepSpeedEngine,)

						while isinstance(_lowerCamelCase			,   _lowerCamelCase     ):
												SCREAMING_SNAKE_CASE__     :				List[Any]   =					model.module

						if not keep_fpaa_wrapper:
												SCREAMING_SNAKE_CASE__     :				Any   =					getattr(_lowerCamelCase			,   "forward"     )
												SCREAMING_SNAKE_CASE__     :				Any   =					model.__dict__.pop("_original_forward"			,   _lowerCamelCase     )
												if original_forward is not None:
																		while hasattr(_lowerCamelCase			,   "__wrapped__"     ):
																								SCREAMING_SNAKE_CASE__     :				List[Any]   =					forward.__wrapped__
																								if forward == original_forward:
																														break
																		SCREAMING_SNAKE_CASE__     :				Optional[int]   =					forward
												if getattr(_lowerCamelCase			,   "_converted_to_transformer_engine"			,   _lowerCamelCase     ):
																		convert_model(_lowerCamelCase			,   to_transformer_engine=_lowerCamelCase     )

						if is_compiled:
												SCREAMING_SNAKE_CASE__     :				Optional[int]   =					model
												SCREAMING_SNAKE_CASE__     :				List[str]   =					compiled_model

						return model







def       UpperCAmelCase							(     ):





						'''simple docstring'''


						PartialState().wait_for_everyone()







def       UpperCAmelCase							(     _lowerCamelCase      :  str			,   _lowerCamelCase      :  Optional[int]     ):





						'''simple docstring'''


						if PartialState().distributed_type == DistributedType.TPU:
												xm.save(_lowerCamelCase			,   _lowerCamelCase     )
						elif PartialState().local_process_index == 0:
												torch.save(_lowerCamelCase			,   _lowerCamelCase     )







@contextmanager
def       UpperCAmelCase							(     **_lowerCamelCase      :  Optional[Any]     ):





						'''simple docstring'''


						for key, value in kwargs.items():
												SCREAMING_SNAKE_CASE__     :				List[Any]   =					str(_lowerCamelCase     )

						yield

						for key in kwargs:
												if key.upper() in os.environ:
																		del os.environ[key.upper()]







def       UpperCAmelCase							(     _lowerCamelCase      :  Tuple     ):





						'''simple docstring'''


						if not hasattr(_lowerCamelCase			,   "__qualname__"     ) and not hasattr(_lowerCamelCase			,   "__name__"     ):
												SCREAMING_SNAKE_CASE__     :				Any   =					getattr(_lowerCamelCase			,   "__class__"			,   _lowerCamelCase     )
						if hasattr(_lowerCamelCase			,   "__qualname__"     ):
												return obj.__qualname__
						if hasattr(_lowerCamelCase			,   "__name__"     ):
												return obj.__name__
						return str(_lowerCamelCase     )







def       UpperCAmelCase							(     _lowerCamelCase      :  str			,   _lowerCamelCase      :  List[str]     ):





						'''simple docstring'''


						for key, value in source.items():
												if isinstance(_lowerCamelCase			,   _lowerCamelCase     ):
																		SCREAMING_SNAKE_CASE__     :				Tuple   =					destination.setdefault(_lowerCamelCase			,   {}     )
																		merge_dicts(_lowerCamelCase			,   _lowerCamelCase     )
												else:
																		SCREAMING_SNAKE_CASE__     :				int   =					value

						return destination







def       UpperCAmelCase							(     _lowerCamelCase      :  int = None     ):





						'''simple docstring'''


						if port is None:
												SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					29_500
						with socket.socket(socket.AF_INET			,   socket.SOCK_STREAM     ) as s:
												return s.connect_ex(("localhost", port)     ) == 0






import unittest

from datasets import load_dataset

from transformers.pipelines import pipeline
from transformers.testing_utils import is_pipeline_test, nested_simplify, require_torch, slow



@is_pipeline_test
@require_torch
class    _a     (				unittest.TestCase							):






							"""simple docstring"""

							@require_torch
							def      A_  (      self				:     Dict    )	->str:
													SCREAMING_SNAKE_CASE__     :				Any   =					pipeline(
													    task="zero-shot-audio-classification"							,	model="hf-internal-testing/tiny-clap-htsat-unfused"    )
													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					load_dataset("ashraq/esc50"    )
													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					dataset["train"]["audio"][-1]["array"]
													SCREAMING_SNAKE_CASE__     :				int   =					audio_classifier(a							,	candidate_labels=["Sound of a dog", "Sound of vaccum cleaner"]    )
													self.assertEqual(
													    nested_simplify(a    )							,	[{"score": 0.501, "label": "Sound of a dog"}, {"score": 0.499, "label": "Sound of vaccum cleaner"}]							,	)

							@unittest.skip("No models are available in TF"    )
							def      A_  (      self				:     int    )	->Union[str, Any]:
													pass

							@slow
							@require_torch
							def      A_  (      self				:     int    )	->str:
													SCREAMING_SNAKE_CASE__     :				List[str]   =					pipeline(
													    task="zero-shot-audio-classification"							,	model="laion/clap-htsat-unfused"							,	)
													# This is an audio of a dog
													SCREAMING_SNAKE_CASE__     :				int   =					load_dataset("ashraq/esc50"    )
													SCREAMING_SNAKE_CASE__     :				str   =					dataset["train"]["audio"][-1]["array"]
													SCREAMING_SNAKE_CASE__     :				List[Any]   =					audio_classifier(a							,	candidate_labels=["Sound of a dog", "Sound of vaccum cleaner"]    )

													self.assertEqual(
													    nested_simplify(a    )							,	[
													        {"score": 0.999, "label": "Sound of a dog"},
													        {"score": 0.001, "label": "Sound of vaccum cleaner"},
													    ]							,	)

													SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					audio_classifier([audio] * 5							,	candidate_labels=["Sound of a dog", "Sound of vaccum cleaner"]    )
													self.assertEqual(
													    nested_simplify(a    )							,	[
													        [
													            {"score": 0.999, "label": "Sound of a dog"},
													            {"score": 0.001, "label": "Sound of vaccum cleaner"},
													        ],
													    ]
													    * 5							,	)
													SCREAMING_SNAKE_CASE__     :				int   =					audio_classifier(
													    [audio] * 5							,	candidate_labels=["Sound of a dog", "Sound of vaccum cleaner"]							,	batch_size=5    )
													self.assertEqual(
													    nested_simplify(a    )							,	[
													        [
													            {"score": 0.999, "label": "Sound of a dog"},
													            {"score": 0.001, "label": "Sound of vaccum cleaner"},
													        ],
													    ]
													    * 5							,	)






							@unittest.skip("No models are available in TF"    )
							def      A_  (      self				:     Optional[int]    )	->Union[str, Any]:
													pass