





"""simple docstring"""




from manim import *




class 					_a		( _lowerCAmelCase							):






				def 							snake_case      (					self						:       int				)      -> Union[str, Any]:




										'''simple docstring'''



										_UpperCamelCase     :   Optional[Any]      =       Rectangle(height=0.5,					width=0.5				)
										_UpperCamelCase     :   List[str]      =       Rectangle(height=0.46,					width=0.46				).set_stroke(width=0				)
										_UpperCamelCase     :   Tuple      =       Rectangle(height=0.25,					width=0.25				)

										_UpperCamelCase     :   str      =       [mem.copy() for i in range(6				)]
										_UpperCamelCase     :   Optional[int]      =       [mem.copy() for i in range(6				)]
										_UpperCamelCase     :   List[Any]      =       VGroup(*lowerCAmelCase__				).arrange(lowerCAmelCase__,					buff=0				)
										_UpperCamelCase     :   int      =       VGroup(*lowerCAmelCase__				).arrange(lowerCAmelCase__,					buff=0				)
										_UpperCamelCase     :   Optional[Any]      =       VGroup(lowerCAmelCase__,					lowerCAmelCase__				).arrange(lowerCAmelCase__,					buff=0				)
										_UpperCamelCase     :   Union[str, Any]      =       Text('''CPU''',					font_size=2_4				)
										_UpperCamelCase     :   Optional[Any]      =       Group(lowerCAmelCase__,					lowerCAmelCase__				).arrange(lowerCAmelCase__,					buff=0.5,					aligned_edge=lowerCAmelCase__				)
										cpu.move_to([-2.5, -0.5, 0]				)
										self.add(lowerCAmelCase__				)

										_UpperCamelCase     :   Union[str, Any]      =       [mem.copy() for i in range(4				)]
										_UpperCamelCase     :   List[str]      =       VGroup(*lowerCAmelCase__				).arrange(lowerCAmelCase__,					buff=0				)
										_UpperCamelCase     :   Optional[int]      =       Text('''GPU''',					font_size=2_4				)
										_UpperCamelCase     :   List[str]      =       Group(lowerCAmelCase__,					lowerCAmelCase__				).arrange(lowerCAmelCase__,					buff=0.5,					aligned_edge=lowerCAmelCase__				)
										gpu.move_to([-1, -1, 0]				)
										self.add(lowerCAmelCase__				)

										_UpperCamelCase     :   str      =       [mem.copy() for i in range(6				)]
										_UpperCamelCase     :   Any      =       VGroup(*lowerCAmelCase__				).arrange(lowerCAmelCase__,					buff=0				)

										_UpperCamelCase     :   Any      =       Text('''Model''',					font_size=2_4				)
										_UpperCamelCase     :   List[str]      =       Group(lowerCAmelCase__,					lowerCAmelCase__				).arrange(lowerCAmelCase__,					buff=0.5,					aligned_edge=lowerCAmelCase__				)
										model.move_to([3, -1.0, 0]				)
										self.add(lowerCAmelCase__				)

										_UpperCamelCase     :   str      =       []
										_UpperCamelCase     :   Union[str, Any]      =       []

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

																_UpperCamelCase     :   Any      =       Rectangle(height=0.46 / 4,					width=0.46 / 3				).set_stroke(width=0.0				).set_fill(lowerCAmelCase__,					opacity=0.7				)

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

										self.add(*lowerCAmelCase__,					*lowerCAmelCase__				)

										_UpperCamelCase     :   Optional[Any]      =       [meta_mem.copy() for i in range(6				)]
										_UpperCamelCase     :   int      =       [meta_mem.copy() for i in range(6				)]
										_UpperCamelCase     :   Tuple      =       VGroup(*lowerCAmelCase__				).arrange(lowerCAmelCase__,					buff=0				)
										_UpperCamelCase     :   Any      =       VGroup(*lowerCAmelCase__				).arrange(lowerCAmelCase__,					buff=0				)
										_UpperCamelCase     :   Any      =       VGroup(lowerCAmelCase__,					lowerCAmelCase__				).arrange(lowerCAmelCase__,					buff=0				)
										_UpperCamelCase     :   List[Any]      =       Text('''Disk''',					font_size=2_4				)
										_UpperCamelCase     :   int      =       Group(lowerCAmelCase__,					lowerCAmelCase__				).arrange(lowerCAmelCase__,					buff=0.5,					aligned_edge=lowerCAmelCase__				)
										disk.move_to([-4.0, -1.25, 0]				)
										self.add(lowerCAmelCase__,					lowerCAmelCase__				)

										_UpperCamelCase     :   List[str]      =       []

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

										self.add(*lowerCAmelCase__				)

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

										_UpperCamelCase     :   int      =       MarkupText(
										    f"""<b>Key:</b>\n\n<span fgcolor='{YELLOW}'>●</span> Empty Model""",					font_size=1_8,					)

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

										self.add(lowerCAmelCase__,					lowerCAmelCase__				)

										_UpperCamelCase     :   str      =       MarkupText(
										    f"""<span fgcolor='{BLUE}'>●</span> Checkpoint""",					font_size=1_8,					)

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

										_UpperCamelCase     :   Any      =       MarkupText(f"""The offloaded weights are all sent to the CPU.""",					font_size=2_4				)
										step_a.move_to([2, 2, 0]				)

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

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

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

																						self.remove(
																						    *lowerCAmelCase__,					*lowerCAmelCase__,					)

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

										_UpperCamelCase     :   Dict      =       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=2_4,					)
										step_a.move_to([2, 2, 0]				)

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

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




'''simple docstring'''
import random
from typing import List, Optional, Union

import numpy as np
import torch

from ..state import AcceleratorState
from .constants import CUDA_DISTRIBUTED_TYPES
from .dataclasses import DistributedType, RNGType
from .imports import is_npu_available, is_tpu_available, is_xpu_available


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







def 			_UpperCamelCase			(  _a						:						int					,  _a						:						bool = False      ):




    """simple docstring"""



    if device_specific:
        seed += AcceleratorState().process_index
    random.seed(_a      )
    np.random.seed(_a      )
    torch.manual_seed(_a      )
    if is_xpu_available():
        torch.xpu.manual_seed_all(_a      )
    elif is_npu_available():
        torch.npu.manual_seed_all(_a      )
    else:
        torch.cuda.manual_seed_all(_a      )
    # ^^ safe to call this function even if cuda is not available
    if is_tpu_available():
        xm.set_rng_state(_a      )







def 			_UpperCamelCase			(  _a						:						Optional[RNGType] = None					,  _a						:						Optional[torch.Generator] = None      ):




    """simple docstring"""



    if rng_type == RNGType.TORCH:
        __UpperCamelCase						:						Union[str, Any]		     =	torch.get_rng_state()
    elif rng_type == RNGType.CUDA:
        __UpperCamelCase						:						int		     =	torch.cuda.get_rng_state()
    elif rng_type == RNGType.XLA:
        assert is_tpu_available(), "Can't synchronize XLA seeds on an environment without TPUs."
        __UpperCamelCase						:						Dict		     =	torch.tensor(xm.get_rng_state()      )
    elif rng_type == RNGType.NPU:
        assert is_npu_available(), "Can't synchronize NPU seeds on an environment without NPUs."
        __UpperCamelCase						:						Optional[Any]		     =	torch.npu.get_rng_state()
    elif rng_type == RNGType.XPU:
        assert is_xpu_available(), "Can't synchronize XPU seeds on an environment without XPUs."
        __UpperCamelCase						:						str		     =	torch.xpu.get_rng_state()
    elif rng_type == RNGType.GENERATOR:
        assert generator is not None, "Need a generator to synchronize its seed."
        __UpperCamelCase						:						List[Any]		     =	generator.get_state()

    # Broadcast the rng state from device 0 to other devices
    __UpperCamelCase						:						List[Any]		     =	AcceleratorState()
    if state.distributed_type == DistributedType.TPU:
        __UpperCamelCase						:						Union[str, Any]		     =	rng_state.to(xm.xla_device()      )
        xm.collective_broadcast([rng_state]      )
        xm.mark_step()
        __UpperCamelCase						:						str		     =	rng_state.cpu()
    elif (
        state.distributed_type in CUDA_DISTRIBUTED_TYPES
        or state.distributed_type == DistributedType.MULTI_NPU
        or state.distributed_type == DistributedType.MULTI_XPU
    ):
        __UpperCamelCase						:						List[Any]		     =	rng_state.to(state.device      )
        torch.distributed.broadcast(_a					,  0      )
        __UpperCamelCase						:						Optional[Any]		     =	rng_state.cpu()
    elif state.distributed_type == DistributedType.MULTI_CPU:
        torch.distributed.broadcast(_a					,  0      )

    # Set the broadcast rng state
    if rng_type == RNGType.TORCH:
        torch.set_rng_state(_a      )
    elif rng_type == RNGType.CUDA:
        torch.cuda.set_rng_state(_a      )
    elif rng_type == RNGType.NPU:
        torch.npu.set_rng_state(_a      )
    elif rng_type == RNGType.XPU:
        torch.xpu.set_rng_state(_a      )
    elif rng_type == RNGType.XLA:
        xm.set_rng_state(rng_state.item()      )
    elif rng_type == RNGType.GENERATOR:
        generator.set_state(_a      )







def 			_UpperCamelCase			(  _a						:						List[Union[str, RNGType]]					,  _a						:						Optional[torch.Generator] = None      ):




    """simple docstring"""



    for rng_type in rng_types:
        synchronize_rng_state(RNGType(_a      )					,  generator=_a      )



