
"""simple docstring"""






from manim import *





class 							snake_case       (     SCREAMING_SNAKE_CASE_		):







   def    UpperCAmelCase__	( self)   ->str:
      a_						   =     Rectangle(height=0.5       ,							width=0.5)
      a_						   =     Rectangle(height=0.46       ,							width=0.46).set_stroke(width=0)
      a_						   =     Rectangle(height=0.25       ,							width=0.25)

      a_						   =     [mem.copy() for i in range(6)]
      a_						   =     [mem.copy() for i in range(6)]
      a_						   =     VGroup(*UpperCAmelCase_).arrange(UpperCAmelCase_       ,							buff=0)
      a_						   =     VGroup(*UpperCAmelCase_).arrange(UpperCAmelCase_       ,							buff=0)
      a_						   =     VGroup(UpperCAmelCase_       ,							UpperCAmelCase_).arrange(UpperCAmelCase_       ,							buff=0)
      a_						   =     Text("CPU"       ,							font_size=24)
      a_						   =     Group(UpperCAmelCase_       ,							UpperCAmelCase_).arrange(UpperCAmelCase_       ,							buff=0.5       ,							aligned_edge=UpperCAmelCase_)
      cpu.move_to([-2.5, -0.5, 0])
      self.add(UpperCAmelCase_)

      a_						   =     [mem.copy() for i in range(4)]
      a_						   =     VGroup(*UpperCAmelCase_).arrange(UpperCAmelCase_       ,							buff=0)
      a_						   =     Text("GPU"       ,							font_size=24)
      a_						   =     Group(UpperCAmelCase_       ,							UpperCAmelCase_).arrange(UpperCAmelCase_       ,							buff=0.5       ,							aligned_edge=UpperCAmelCase_)
      gpu.move_to([-1, -1, 0])
      self.add(UpperCAmelCase_)

      a_						   =     [mem.copy() for i in range(6)]
      a_						   =     VGroup(*UpperCAmelCase_).arrange(UpperCAmelCase_       ,							buff=0)

      a_						   =     Text("Model"       ,							font_size=24)
      a_						   =     Group(UpperCAmelCase_       ,							UpperCAmelCase_).arrange(UpperCAmelCase_       ,							buff=0.5       ,							aligned_edge=UpperCAmelCase_)
      model.move_to([3, -1.0, 0])
      self.add(UpperCAmelCase_)

      a_						   =     []
      a_						   =     []

      for i, rect in enumerate(UpperCAmelCase_):
         rect.set_stroke(UpperCAmelCase_)

         a_						   =     Rectangle(height=0.46 / 4       ,							width=0.46 / 3).set_stroke(width=0.0).set_fill(UpperCAmelCase_       ,							opacity=0.7)

         if i == 0:
            cpu_target.next_to(cpu_left_col_base[0].get_corner(DOWN + LEFT)       ,							buff=0.02       ,							direction=UpperCAmelCase_)
            cpu_target.set_x(cpu_target.get_x() + 0.1)
         elif i == 3:
            cpu_target.next_to(model_cpu_arr[0]       ,							direction=UpperCAmelCase_       ,							buff=0.0)
         else:
            cpu_target.next_to(model_cpu_arr[i - 1]       ,							direction=UpperCAmelCase_       ,							buff=0.0)
         self.add(UpperCAmelCase_)
         model_cpu_arr.append(UpperCAmelCase_)

      self.add(*UpperCAmelCase_       ,							*UpperCAmelCase_)

      a_						   =     [meta_mem.copy() for i in range(6)]
      a_						   =     [meta_mem.copy() for i in range(6)]
      a_						   =     VGroup(*UpperCAmelCase_).arrange(UpperCAmelCase_       ,							buff=0)
      a_						   =     VGroup(*UpperCAmelCase_).arrange(UpperCAmelCase_       ,							buff=0)
      a_						   =     VGroup(UpperCAmelCase_       ,							UpperCAmelCase_).arrange(UpperCAmelCase_       ,							buff=0)
      a_						   =     Text("Disk"       ,							font_size=24)
      a_						   =     Group(UpperCAmelCase_       ,							UpperCAmelCase_).arrange(UpperCAmelCase_       ,							buff=0.5       ,							aligned_edge=UpperCAmelCase_)
      disk.move_to([-4.0, -1.25, 0])
      self.add(UpperCAmelCase_       ,							UpperCAmelCase_)

      a_						   =     []

      for i in range(6):
         a_						   =     fill.copy().set_fill(UpperCAmelCase_       ,							opacity=0.8)
         target.move_to(disk_left_col_base[i]).scale(0.5)
         cpu_disk_arr.append(UpperCAmelCase_)

      self.add(*UpperCAmelCase_)

      a_						   =     Square(side_length=2.2)
      key.move_to([-5, 2, 0])

      a_						   =     MarkupText(
          F'''<b>Key:</b>\n\n<span fgcolor=\'{YELLOW}\'>●</span> Empty Model'''       ,							font_size=18       ,							)

      key_text.move_to([-5, 2.4, 0])

      self.add(UpperCAmelCase_       ,							UpperCAmelCase_)

      a_						   =     MarkupText(
          F'''<span fgcolor=\'{BLUE}\'>●</span> Checkpoint'''       ,							font_size=18       ,							)

      blue_text.next_to(UpperCAmelCase_       ,							DOWN * 2.4       ,							aligned_edge=key_text.get_left())
      self.add(UpperCAmelCase_)

      a_						   =     MarkupText(F'''The offloaded weights are all sent to the CPU.'''       ,							font_size=24)
      step_a.move_to([2, 2, 0])

      self.play(Write(UpperCAmelCase_       ,							run_time=3))

      for i in range(6):
         a_						   =     cpu_disk_arr[i]
         a_						   =     rect.copy().set_fill(UpperCAmelCase_       ,							opacity=0.8).scale(2.0)
         cpa.generate_target()
         cpa.target.move_to(model_base[i])

         if i == 0:
            rect.set_fill(UpperCAmelCase_       ,							opacity=0.8)
            rect.generate_target()
            rect.target.move_to(cpu_left_col_base[0]).scale(2.0)

            self.remove(
                *UpperCAmelCase_       ,							*UpperCAmelCase_       ,							)

         else:
            rect.generate_target()
            rect.target.move_to(cpu_left_col_base[i]).scale(2.0)
         self.play(MoveToTarget(UpperCAmelCase_)       ,							MoveToTarget(UpperCAmelCase_)       ,							model_base[i].animate.set_stroke(UpperCAmelCase_))
      self.play(FadeOut(UpperCAmelCase_))

      a_						   =     MarkupText(
          F'''Finally, hooks are added to each weight in the model\nto transfer the weights from CPU to GPU\n\t\tand back when needed.'''       ,							font_size=24       ,							)
      step_a.move_to([2, 2, 0])

      self.play(Write(UpperCAmelCase_       ,							run_time=3))

      a_						   =     []
      a_						   =     []
      for i in range(6):
         a_						   =     Arrow(start=UpperCAmelCase_       ,							end=UpperCAmelCase_       ,							color=UpperCAmelCase_       ,							buff=0.5)
         a.next_to(model_base[i].get_left()       ,							UpperCAmelCase_       ,							buff=0.2)
         arrows.append(UpperCAmelCase_)
         animations.append(Write(UpperCAmelCase_))
      self.play(*UpperCAmelCase_)
      self.wait()




"""simple docstring"""







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:  Union[str, Any]			     =				namedtuple("""ExampleNamedTuple""", """a b c""")





class   UpperCamelCase	(     unittest.TestCase       ):




	"""simple docstring"""






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

			_lowercase				: int     =   send_to_device(UpperCAmelCase_				,UpperCAmelCase_   )
			self.assertTrue(torch.equal(resulta.cpu()				,UpperCAmelCase_   )   )

			_lowercase				: int     =   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   )

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

			_lowercase				: Optional[int]     =   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   )

			_lowercase				: str     =   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    lowerCamelCase__					(     self   ):
			with self.assertRaises(UpperCAmelCase_   ) as cm:
					_lowercase				: str     =   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    lowerCamelCase__					(     self   ):
			_lowercase				: Optional[int]     =   torch.tensor([1, 2, 3, 4, 5]   )
			self.assertEqual(listify(UpperCAmelCase_   )				,[1, 2, 3, 4, 5]   )

			_lowercase				: Optional[int]     =   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]]   )

			_lowercase				: Optional[Any]     =   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    lowerCamelCase__					(     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    lowerCamelCase__					(     self   ):
			_lowercase				: Optional[Any]     =   RegressionModel()
			_lowercase				: List[Any]     =   model.forward
			_lowercase				: Optional[Any]     =   convert_outputs_to_fpaa(model.forward   )
			_lowercase				: Any     =   extract_model_from_parallel(UpperCAmelCase_				,keep_fpaa_wrapper=UpperCAmelCase_   )
			_lowercase				: Tuple     =   pickle.dumps(UpperCAmelCase_   )






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






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






	def    lowerCamelCase__					(     self   ):
			_lowercase				: List[str]     =   RegressionModel()
			# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
			_lowercase				: Any     =   torch.nn.parallel.DataParallel(UpperCAmelCase_   )
			_lowercase				: List[Any]     =   extract_model_from_parallel(UpperCAmelCase_   )

			self.assertEqual(UpperCAmelCase_				,UpperCAmelCase_   )






	@require_torch_min_version(version="""2.0"""   )
	def    lowerCamelCase__					(     self   ):
			_lowercase				: int     =   RegressionModel()
			_lowercase				: List[str]     =   torch.compile(UpperCAmelCase_   )

			# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
			_lowercase				: Optional[int]     =   torch.nn.parallel.DataParallel(UpperCAmelCase_   )
			_lowercase				: int     =   torch.compile(UpperCAmelCase_   )
			_lowercase				: Dict     =   extract_model_from_parallel(UpperCAmelCase_   )

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






	def    lowerCamelCase__					(     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"""]   )   )



