import os
from pathlib import Path
from typing import Dict, List

import fire
import torch

from transformers import AutoModelForSeqaSeqLM, AutoTokenizer
from transformers.utils.logging import get_logger


UpperCamelCase      			=     get_logger(__name__)



def    lowerCamelCase_   (  _lowercase ,     _lowercase    )							->	str:
		if text.startswith(_lowercase    ):
				return text[len(_lowercase    ) :]
		return text  # or whatever



def    lowerCamelCase_   (  _lowercase    )							->	Optional[int]:
		return {remove_prefix(_lowercase ,     "model."    ): v for k, v in sd.items()}



def    lowerCamelCase_   (  _lowercase    )							->	Tuple:
		__A							:   Optional[int]	  =				{}
		for k in state_dicts[0].keys():
				__A							:   Optional[Any]	  =				[sd[k] for sd in state_dicts]
				__A							:   Tuple	  =				sum(_lowercase    ) / len(_lowercase    )
				assert isinstance(_lowercase ,     torch.Tensor    )
				__A							:   List[str]	  =				new_t
		return new_sd



def    lowerCamelCase_   (  _lowercase ,     _lowercase ,     _lowercase    )							->	None:
		__A							:   Optional[int]	  =				AutoModelForSeqaSeqLM.from_pretrained(_lowercase    )
		if os.path.isfile(_lowercase    ):
				__A							:   str	  =				[pl_ckpt_path]
		else:
				assert os.path.isdir(_lowercase    )
				__A							:   str	  =				list(Path(_lowercase    ).glob("*.ckpt"    )    )
				assert ckpt_files, F"could not find any ckpt files inside the {pl_ckpt_path} directory"

		if len(_lowercase    ) > 1:
				logger.info(F"averaging the weights of {ckpt_files}"    )

		__A							:   Tuple	  =				[sanitize(torch.load(_lowercase ,     map_location="cpu"    )["state_dict"]    ) for x in ckpt_files]
		__A							:   Optional[int]	  =				average_state_dicts(_lowercase    )

		__A  ,       __A							:   Optional[int]	  =				hf_model.load_state_dict(_lowercase ,     strict=_lowercase    )
		assert not missing, F"missing keys: {missing}"
		hf_model.save_pretrained(_lowercase    )
		try:
				__A							:   Union[str, Any]	  =				AutoTokenizer.from_pretrained(_lowercase    )
				tok.save_pretrained(_lowercase    )
		except Exception:
				pass
				# dont copy tokenizer if cant


if __name__ == "__main__":
					fire.Fire(convert_pl_to_hf)






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"]   )   )





