# Lint as: python3

import hashlib
from typing import Union



def    lowerCamelCase_   (  _lowercase    )							->	bytes:
		if isinstance(_lowercase ,     _lowercase    ):
				# Data already in bytes, returns as it as
				return hash_data
		elif isinstance(_lowercase ,     _lowercase    ):
				# We keep the data as it as for it ot be later encoded to UTF-8
				# However replace `\\` with `/` for Windows compatibility
				__A							:   Dict	  =				hash_data.replace("\\" ,     "/"    )
		elif isinstance(_lowercase ,     _lowercase    ):
				__A							:   Any	  =				str(_lowercase    )
		else:
				# If data is not of the required type, raise error
				raise InvalidKeyError(_lowercase    )

		return hash_data.encode("utf-8"    )


class _a       (	lowerCAmelCase__ ):

	'''simple docstring'''



	def __init__(       self							,			__UpperCAmelCase   ):
			__A							:   Any	  =				"\nFAILURE TO GENERATE DATASET: Invalid key type detected"
			__A							:   Dict	  =				F"\nFound Key {hash_data} of type {type(__UpperCAmelCase   )}"
			__A							:   Dict	  =				"\nKeys should be either str, int or bytes type"
			super().__init__(F"{self.prefix}{self.err_msg}{self.suffix}"   )





class _a       (	lowerCAmelCase__ ):

	'''simple docstring'''



	def __init__(       self							,			__UpperCAmelCase							,			__UpperCAmelCase							,			__UpperCAmelCase=""   ):
			__A							:   Tuple	  =				key
			__A							:   str	  =				duplicate_key_indices
			__A							:   Tuple	  =				fix_msg
			__A							:   int	  =				"Found multiple examples generated with the same key"
			if len(__UpperCAmelCase   ) <= 20:
					__A							:   List[Any]	  =				F"\nThe examples at index {', '.join(__UpperCAmelCase   )} have the key {key}"
			else:
					__A							:   Dict	  =				F"\nThe examples at index {', '.join(duplicate_key_indices[:20]   )}... ({len(__UpperCAmelCase   ) - 20} more) have the key {key}"
			__A							:   Tuple	  =				"\n" + fix_msg if fix_msg else ""
			super().__init__(F"{self.prefix}{self.err_msg}{self.suffix}"   )





class _a       :

	'''simple docstring'''



	def __init__(       self							,			__UpperCAmelCase   ):
			__A							:   int	  =				hashlib.mda(_as_bytes(__UpperCAmelCase   )   )







	def 					__UpperCAmelCase(       self							,			__UpperCAmelCase   ):
			__A							:   str	  =				self._split_mda.copy()
			__A							:   Tuple	  =				_as_bytes(__UpperCAmelCase   )
			mda.update(__UpperCAmelCase   )
			# Convert to integer with hexadecimal conversion
			return int(mda.hexdigest()							,			16   )






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 ):

	'''simple docstring'''



	def 					__UpperCAmelCase(       self   ):
			__A							:   List[str]	  =				torch.randn(5							,			2   )
			__A							:   Any	  =				torch.device("cuda"   ) if torch.cuda.is_available() else torch.device("cpu"   )

			__A							:   Union[str, Any]	  =				send_to_device(__UpperCAmelCase							,			__UpperCAmelCase   )
			self.assertTrue(torch.equal(resulta.cpu()							,			__UpperCAmelCase   )   )

			__A							:   Dict	  =				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   )

			__A							:   Optional[Any]	  =				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   )

			__A							:   Any	  =				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   )

			__A							:   List[Any]	  =				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 					__UpperCAmelCase(       self   ):
			with self.assertRaises(__UpperCAmelCase   ) as cm:
					__A							:   List[Any]	  =				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 					__UpperCAmelCase(       self   ):
			__A							:   List[Any]	  =				torch.tensor([1, 2, 3, 4, 5]   )
			self.assertEqual(listify(__UpperCAmelCase   )							,			[1, 2, 3, 4, 5]   )

			__A							:   List[str]	  =				torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]   )
			self.assertEqual(listify(__UpperCAmelCase   )							,			[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]   )

			__A							:   Dict	  =				torch.tensor([[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]]   )
			self.assertEqual(
			    listify(__UpperCAmelCase   )							,			[[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]]   )

	def 					__UpperCAmelCase(       self   ):
			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 					__UpperCAmelCase(       self   ):
			__A							:   str	  =				RegressionModel()
			__A							:   str	  =				model.forward
			__A							:   List[Any]	  =				convert_outputs_to_fpaa(model.forward   )
			__A							:   Optional[int]	  =				extract_model_from_parallel(__UpperCAmelCase							,			keep_fpaa_wrapper=__UpperCAmelCase   )
			__A							:   Any	  =				pickle.dumps(__UpperCAmelCase   )

	@require_cuda
	def 					__UpperCAmelCase(       self   ):
			__A							:   Any	  =				RegressionModel()
			__A							:   str	  =				model.forward
			__A							:   Tuple	  =				torch.cuda.amp.autocast(dtype=torch.floataa   )(model.forward   )
			__A							:   List[Any]	  =				convert_outputs_to_fpaa(model.forward   )
			__A							:   Optional[Any]	  =				extract_model_from_parallel(__UpperCAmelCase							,			keep_fpaa_wrapper=__UpperCAmelCase   )
			__A							:   Any	  =				pickle.dumps(__UpperCAmelCase   )

	@require_cuda
	@require_torch_min_version(version="2.0"   )
	def 					__UpperCAmelCase(       self   ):
			__A							:   Any	  =				RegressionModel()
			__A							:   Optional[int]	  =				model.forward
			__A							:   Union[str, Any]	  =				torch.cuda.amp.autocast(dtype=torch.floataa   )(model.forward   )
			__A							:   int	  =				convert_outputs_to_fpaa(model.forward   )
			__A							:   Optional[Any]	  =				torch.compile(model.forward							,			backend="inductor"   )
			__A							:   Tuple	  =				torch.randn(4							,			10   ).cuda()
			__A							:   Union[str, Any]	  =				model(__UpperCAmelCase   )

	def 					__UpperCAmelCase(       self   ):
			__A							:   Union[str, Any]	  =				RegressionModel()
			# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
			__A							:   Dict	  =				torch.nn.parallel.DataParallel(__UpperCAmelCase   )
			__A							:   int	  =				extract_model_from_parallel(__UpperCAmelCase   )

			self.assertEqual(__UpperCAmelCase							,			__UpperCAmelCase   )

	@require_torch_min_version(version="2.0"   )
	def 					__UpperCAmelCase(       self   ):
			__A							:   Dict	  =				RegressionModel()
			__A							:   Dict	  =				torch.compile(__UpperCAmelCase   )

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

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







	def 					__UpperCAmelCase(       self   ):
			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"]   )   )





