import logging
import os
import sys
import threading
from logging import (
    CRITICAL,  # NOQA
    DEBUG,  # NOQA
    ERROR,  # NOQA
    FATAL,  # NOQA
    INFO,  # NOQA
    NOTSET,  # NOQA
    WARN,  # NOQA
    WARNING,  # NOQA
)
from typing import Optional

from tqdm import auto as tqdm_lib


UpperCamelCase__       							=      threading.Lock()
UpperCamelCase__       							=      None

UpperCamelCase__       							=      {
    'debug': logging.DEBUG,
    'info': logging.INFO,
    'warning': logging.WARNING,
    'error': logging.ERROR,
    'critical': logging.CRITICAL,
}

UpperCamelCase__       							=      logging.WARNING

UpperCamelCase__       							=      True




def      lowerCAmelCase_				(  )  ->       int:







      '''simple docstring'''


      UpperCAmelCase__			=						os.getenv("DIFFUSERS_VERBOSITY",						__A							)
      if env_level_str:
            if env_level_str in log_levels:
                  return log_levels[env_level_str]
            else:
                  logging.getLogger().warning(
                      f"""Unknown option DIFFUSERS_VERBOSITY={env_level_str}, """
                      f"""has to be one of: { ", ".join(log_levels.keys()							) }"""							)
      return _default_log_level




def      lowerCAmelCase_				(  )  ->       str:







      '''simple docstring'''


      return __name__.split("."							)[0]




def      lowerCAmelCase_				(  )  ->       logging.Logger:







      '''simple docstring'''


      return logging.getLogger(_get_library_name()							)




def      lowerCAmelCase_				(  )  ->       None:







      '''simple docstring'''


      global _default_handler

      with _lock:
            if _default_handler:
                  # This library has already configured the library root logger.
                  return
            UpperCAmelCase__			=						logging.StreamHandler()  # Set sys.stderr as stream.
            UpperCAmelCase__			=						sys.stderr.flush

            # Apply our default configuration to the library root logger.
            UpperCAmelCase__			=						_get_library_root_logger()
            library_root_logger.addHandler(_default_handler							)
            library_root_logger.setLevel(_get_default_logging_level()							)
            UpperCAmelCase__			=						False




def      lowerCAmelCase_				(  )  ->       None:







      '''simple docstring'''


      global _default_handler

      with _lock:
            if not _default_handler:
                  return

            UpperCAmelCase__			=						_get_library_root_logger()
            library_root_logger.removeHandler(_default_handler							)
            library_root_logger.setLevel(logging.NOTSET							)
            UpperCAmelCase__			=						None




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







      '''simple docstring'''


      return log_levels




def      lowerCAmelCase_				(  __A = None							)  ->       logging.Logger:







      '''simple docstring'''


      if name is None:
            UpperCAmelCase__			=						_get_library_name()

      _configure_library_root_logger()
      return logging.getLogger(__A							)




def      lowerCAmelCase_				(  )  ->       int:







      '''simple docstring'''


      _configure_library_root_logger()
      return _get_library_root_logger().getEffectiveLevel()




def      lowerCAmelCase_				(  __A							)  ->       None:







      '''simple docstring'''


      _configure_library_root_logger()
      _get_library_root_logger().setLevel(__A							)




def      lowerCAmelCase_				(  )  ->       Any:







      '''simple docstring'''


      return set_verbosity(__A							)




def      lowerCAmelCase_				(  )  ->       Optional[Any]:







      '''simple docstring'''


      return set_verbosity(__A							)




def      lowerCAmelCase_				(  )  ->       Any:







      '''simple docstring'''


      return set_verbosity(__A							)




def      lowerCAmelCase_				(  )  ->       List[Any]:







      '''simple docstring'''


      return set_verbosity(__A							)




def      lowerCAmelCase_				(  )  ->       None:







      '''simple docstring'''


      _configure_library_root_logger()

      assert _default_handler is not None
      _get_library_root_logger().removeHandler(_default_handler							)




def      lowerCAmelCase_				(  )  ->       None:







      '''simple docstring'''


      _configure_library_root_logger()

      assert _default_handler is not None
      _get_library_root_logger().addHandler(_default_handler							)




def      lowerCAmelCase_				(  __A							)  ->       None:







      '''simple docstring'''


      _configure_library_root_logger()

      assert handler is not None
      _get_library_root_logger().addHandler(__A							)




def      lowerCAmelCase_				(  __A							)  ->       None:







      '''simple docstring'''


      _configure_library_root_logger()

      assert handler is not None and handler not in _get_library_root_logger().handlers
      _get_library_root_logger().removeHandler(__A							)




def      lowerCAmelCase_				(  )  ->       None:







      '''simple docstring'''


      _configure_library_root_logger()
      UpperCAmelCase__			=						False




def      lowerCAmelCase_				(  )  ->       None:







      '''simple docstring'''


      _configure_library_root_logger()
      UpperCAmelCase__			=						True




def      lowerCAmelCase_				(  )  ->       None:







      '''simple docstring'''


      UpperCAmelCase__			=						_get_library_root_logger().handlers

      for handler in handlers:
            UpperCAmelCase__			=						logging.Formatter("[%(levelname)s|%(filename)s:%(lineno)s] %(asctime)s >> %(message)s"							)
            handler.setFormatter(__A							)




def      lowerCAmelCase_				(  )  ->       None:







      '''simple docstring'''


      UpperCAmelCase__			=						_get_library_root_logger().handlers

      for handler in handlers:
            handler.setFormatter(__A							)




def      lowerCAmelCase_				(  self,						*__A,						**__A							)  ->       List[Any]:







      '''simple docstring'''


      UpperCAmelCase__			=						os.getenv("DIFFUSERS_NO_ADVISORY_WARNINGS",						__A							)
      if no_advisory_warnings:
            return
      self.warning(*__A,						**__A							)


UpperCamelCase__       							=      warning_advice





class   A							:


   def __init__(self  :		str ,						*__UpperCAmelCase  :		Dict ,						**__UpperCAmelCase  :		str	)		->       Optional[Any]:  # pylint: disable=unused-argument
         """simple docstring"""



         UpperCAmelCase__			=						args[0] if args else None


   def __iter__(self  :		Tuple	)		->       Union[str, Any]:
         """simple docstring"""



         return iter(self._iterator	)


   def __getattr__(self  :		Any ,						__UpperCAmelCase  :		Union[str, Any]	)		->       Optional[int]:
         """simple docstring"""



         def empty_fn(*__UpperCAmelCase  :		Dict ,						**__UpperCAmelCase  :		int	):  # pylint: disable=unused-argument
               return

         return empty_fn


   def __enter__(self  :		Dict	)		->       Tuple:
         """simple docstring"""



         return self





   def __exit__(self  :		Optional[int] ,						__UpperCAmelCase  :		str ,						__UpperCAmelCase  :		Optional[Any] ,						__UpperCAmelCase  :		Dict	)		->       Optional[int]:
         """simple docstring"""



         return





class   A							:


   def __call__(self  :		Optional[int] ,						*__UpperCAmelCase  :		str ,						**__UpperCAmelCase  :		Tuple	)		->       Any:
         """simple docstring"""



         if _tqdm_active:
               return tqdm_lib.tqdm(*__UpperCAmelCase ,						**__UpperCAmelCase	)
         else:
               return EmptyTqdm(*__UpperCAmelCase ,						**__UpperCAmelCase	)


   def     lowercase_   (self  :		List[str] ,						*__UpperCAmelCase  :		Optional[Any] ,						**__UpperCAmelCase  :		Union[str, Any]	)		->       int:
         """simple docstring"""



         UpperCAmelCase__			=						None
         if _tqdm_active:
               return tqdm_lib.tqdm.set_lock(*__UpperCAmelCase ,						**__UpperCAmelCase	)





   def     lowercase_   (self  :		Union[str, Any]	)		->       str:
         """simple docstring"""



         if _tqdm_active:
               return tqdm_lib.tqdm.get_lock()




UpperCamelCase__       							=      _tqdm_cls()




def      lowerCAmelCase_				(  )  ->       bool:







      '''simple docstring'''


      global _tqdm_active
      return bool(_tqdm_active							)




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







      '''simple docstring'''


      global _tqdm_active
      UpperCAmelCase__			=						True







def      lowerCAmelCase_				(  )  ->       Dict:







      '''simple docstring'''


      global _tqdm_active
      UpperCAmelCase__			=						False




import os
import pickle
import unittest
from collections import UserDict, namedtuple

import torch

from accelerate.test_utils.testing import require_cuda, require_torch_min_version
from accelerate.test_utils.training import RegressionModel
from accelerate.utils import (
    convert_outputs_to_fpaa,
    extract_model_from_parallel,
    find_device,
    listify,
    patch_environment,
    recursively_apply,
    send_to_device,
)


UpperCamelCase__       							=      namedtuple('ExampleNamedTuple', 'a b c')





class   A							(       unittest.TestCase  ):


   def     lowercase_   (self  :		str	)		->       List[Any]:
         """simple docstring"""



         UpperCAmelCase__			=						torch.randn(5 ,						2	)
         UpperCAmelCase__			=						torch.device("cuda"	) if torch.cuda.is_available() else torch.device("cpu"	)

         UpperCAmelCase__			=						send_to_device(__UpperCAmelCase ,						__UpperCAmelCase	)
         self.assertTrue(torch.equal(resulta.cpu() ,						__UpperCAmelCase	)	)

         UpperCAmelCase__			=						send_to_device((tensor, [tensor, tensor], 1) ,						__UpperCAmelCase	)
         self.assertIsInstance(__UpperCAmelCase ,						__UpperCAmelCase	)
         self.assertTrue(torch.equal(resulta[0].cpu() ,						__UpperCAmelCase	)	)
         self.assertIsInstance(resulta[1] ,						__UpperCAmelCase	)
         self.assertTrue(torch.equal(resulta[1][0].cpu() ,						__UpperCAmelCase	)	)
         self.assertTrue(torch.equal(resulta[1][1].cpu() ,						__UpperCAmelCase	)	)
         self.assertEqual(resulta[2] ,						1	)

         UpperCAmelCase__			=						send_to_device({"a": tensor, "b": [tensor, tensor], "c": 1} ,						__UpperCAmelCase	)
         self.assertIsInstance(__UpperCAmelCase ,						__UpperCAmelCase	)
         self.assertTrue(torch.equal(resulta["a"].cpu() ,						__UpperCAmelCase	)	)
         self.assertIsInstance(resulta["b"] ,						__UpperCAmelCase	)
         self.assertTrue(torch.equal(resulta["b"][0].cpu() ,						__UpperCAmelCase	)	)
         self.assertTrue(torch.equal(resulta["b"][1].cpu() ,						__UpperCAmelCase	)	)
         self.assertEqual(resulta["c"] ,						1	)

         UpperCAmelCase__			=						send_to_device(ExampleNamedTuple(a=__UpperCAmelCase ,						b=[tensor, tensor] ,						c=1	) ,						__UpperCAmelCase	)
         self.assertIsInstance(__UpperCAmelCase ,						__UpperCAmelCase	)
         self.assertTrue(torch.equal(resulta.a.cpu() ,						__UpperCAmelCase	)	)
         self.assertIsInstance(resulta.b ,						__UpperCAmelCase	)
         self.assertTrue(torch.equal(resulta.b[0].cpu() ,						__UpperCAmelCase	)	)
         self.assertTrue(torch.equal(resulta.b[1].cpu() ,						__UpperCAmelCase	)	)
         self.assertEqual(resulta.c ,						1	)

         UpperCAmelCase__			=						send_to_device(UserDict({"a": tensor, "b": [tensor, tensor], "c": 1}	) ,						__UpperCAmelCase	)
         self.assertIsInstance(__UpperCAmelCase ,						__UpperCAmelCase	)
         self.assertTrue(torch.equal(resulta["a"].cpu() ,						__UpperCAmelCase	)	)
         self.assertIsInstance(resulta["b"] ,						__UpperCAmelCase	)
         self.assertTrue(torch.equal(resulta["b"][0].cpu() ,						__UpperCAmelCase	)	)
         self.assertTrue(torch.equal(resulta["b"][1].cpu() ,						__UpperCAmelCase	)	)
         self.assertEqual(resulta["c"] ,						1	)


   def     lowercase_   (self  :		int	)		->       int:
         """simple docstring"""



         with self.assertRaises(__UpperCAmelCase	) as cm:
               UpperCAmelCase__			=						recursively_apply(torch.tensor ,						(torch.tensor(1	), 1) ,						error_on_other_type=__UpperCAmelCase	)
         self.assertEqual(
             str(cm.exception	) ,						"Unsupported types (<class 'int'>) passed to `tensor`. Only nested list/tuple/dicts of objects that are valid for `is_torch_tensor` should be passed." ,						)


   def     lowercase_   (self  :		List[str]	)		->       Any:
         """simple docstring"""



         UpperCAmelCase__			=						torch.tensor([1, 2, 3, 4, 5]	)
         self.assertEqual(listify(__UpperCAmelCase	) ,						[1, 2, 3, 4, 5]	)

         UpperCAmelCase__			=						torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]]	)
         self.assertEqual(listify(__UpperCAmelCase	) ,						[[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]]	)

         UpperCAmelCase__			=						torch.tensor([[[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]], [[1_1, 1_2, 1_3, 1_4, 1_5], [1_6, 1_7, 1_8, 1_9, 2_0]]]	)
         self.assertEqual(
             listify(__UpperCAmelCase	) ,						[[[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]], [[1_1, 1_2, 1_3, 1_4, 1_5], [1_6, 1_7, 1_8, 1_9, 2_0]]]	)


   def     lowercase_   (self  :		Union[str, Any]	)		->       Dict:
         """simple docstring"""



         with patch_environment(aa=1 ,						BB=2	):
               self.assertEqual(os.environ.get("AA"	) ,						"1"	)
               self.assertEqual(os.environ.get("BB"	) ,						"2"	)

         self.assertNotIn("AA" ,						os.environ	)
         self.assertNotIn("BB" ,						os.environ	)


   def     lowercase_   (self  :		int	)		->       str:
         """simple docstring"""



         UpperCAmelCase__			=						RegressionModel()
         UpperCAmelCase__			=						model.forward
         UpperCAmelCase__			=						convert_outputs_to_fpaa(model.forward	)
         UpperCAmelCase__			=						extract_model_from_parallel(__UpperCAmelCase ,						keep_fpaa_wrapper=__UpperCAmelCase	)
         UpperCAmelCase__			=						pickle.dumps(__UpperCAmelCase	)


   @require_cuda
   def     lowercase_   (self  :		List[str]	)		->       Union[str, Any]:
         """simple docstring"""



         UpperCAmelCase__			=						RegressionModel()
         UpperCAmelCase__			=						model.forward
         UpperCAmelCase__			=						torch.cuda.amp.autocast(dtype=torch.floataa	)(model.forward	)
         UpperCAmelCase__			=						convert_outputs_to_fpaa(model.forward	)
         UpperCAmelCase__			=						extract_model_from_parallel(__UpperCAmelCase ,						keep_fpaa_wrapper=__UpperCAmelCase	)
         UpperCAmelCase__			=						pickle.dumps(__UpperCAmelCase	)


   @require_cuda
   @require_torch_min_version(version="2.0"	)
   def     lowercase_   (self  :		Optional[int]	)		->       Union[str, Any]:
         """simple docstring"""



         UpperCAmelCase__			=						RegressionModel()
         UpperCAmelCase__			=						model.forward
         UpperCAmelCase__			=						torch.cuda.amp.autocast(dtype=torch.floataa	)(model.forward	)
         UpperCAmelCase__			=						convert_outputs_to_fpaa(model.forward	)
         UpperCAmelCase__			=						torch.compile(model.forward ,						backend="inductor"	)
         UpperCAmelCase__			=						torch.randn(4 ,						1_0	).cuda()
         UpperCAmelCase__			=						model(__UpperCAmelCase	)


   def     lowercase_   (self  :		List[Any]	)		->       Union[str, Any]:
         """simple docstring"""



         UpperCAmelCase__			=						RegressionModel()
         # could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
         UpperCAmelCase__			=						torch.nn.parallel.DataParallel(__UpperCAmelCase	)
         UpperCAmelCase__			=						extract_model_from_parallel(__UpperCAmelCase	)

         self.assertEqual(__UpperCAmelCase ,						__UpperCAmelCase	)


   @require_torch_min_version(version="2.0"	)
   def     lowercase_   (self  :		Union[str, Any]	)		->       Union[str, Any]:
         """simple docstring"""



         UpperCAmelCase__			=						RegressionModel()
         UpperCAmelCase__			=						torch.compile(__UpperCAmelCase	)

         # could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
         UpperCAmelCase__			=						torch.nn.parallel.DataParallel(__UpperCAmelCase	)
         UpperCAmelCase__			=						torch.compile(__UpperCAmelCase	)
         UpperCAmelCase__			=						extract_model_from_parallel(__UpperCAmelCase	)

         self.assertEqual(compiled_model._orig_mod ,						compiled_model_unwrapped._orig_mod	)





   def     lowercase_   (self  :		int	)		->       int:
         """simple docstring"""



         self.assertEqual(find_device([1, "a", torch.tensor([1, 2, 3]	)]	) ,						torch.device("cpu"	)	)
         self.assertEqual(find_device({"a": 1, "b": torch.tensor([1, 2, 3]	)}	) ,						torch.device("cpu"	)	)
         self.assertIsNone(find_device([1, "a"]	)	)



