

import warnings

from .state import AcceleratorState, GradientState


warnings.filterwarnings("ignore", category=UserWarning, module="torch.optim.lr_scheduler")







class 					__lowercase  :





							"""simple docstring"""

							def __init__(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase = True    ,						__UpperCAmelCase = False						)		->       Any:
														A      :					Dict      		=					scheduler
														A      :					str      		=					optimizers if isinstance(__UpperCAmelCase    ,						(list, tuple)						) else [optimizers]
														A      :					List[str]      		=					split_batches
														A      :					Dict      		=					step_with_optimizer
														A      :					Tuple      		=					GradientState()




							def       snake_case			(					self    ,						*__UpperCAmelCase    ,						**__UpperCAmelCase						)		->       Dict:
														if not self.step_with_optimizer:
																					# No link between scheduler and optimizer -> just step
																					self.scheduler.step(*__UpperCAmelCase    ,						**__UpperCAmelCase						)
																					return

														# Otherwise, first make sure the optimizer was stepped.
														if not self.gradient_state.sync_gradients:
																					if self.gradient_state.adjust_scheduler:
																												self.scheduler._step_count += 1
																					return

														for opt in self.optimizers:
																					if opt.step_was_skipped:
																												return
														if self.split_batches:
																					# Split batches -> the training dataloader batch size is not changed so one step per training step
																					self.scheduler.step(*__UpperCAmelCase    ,						**__UpperCAmelCase						)
														else:
																					# Otherwise the training dataloader batch size was multiplied by `num_processes`, so we need to do
																					# num_processes steps per training step
																					A      :					int      		=					AcceleratorState().num_processes
																					for _ in range(__UpperCAmelCase						):
																												# Special case when using OneCycle and `drop_last` was not used
																												if hasattr(self.scheduler    ,						'''total_steps'''						):
																																			if self.scheduler._step_count <= self.scheduler.total_steps:
																																										self.scheduler.step(*__UpperCAmelCase    ,						**__UpperCAmelCase						)
																												else:
																																			self.scheduler.step(*__UpperCAmelCase    ,						**__UpperCAmelCase						)




							def       snake_case			(					self						)		->       Optional[Any]:
														return self.scheduler.get_last_lr()




							def       snake_case			(					self						)		->       int:
														return self.scheduler.state_dict()




							def       snake_case			(					self    ,						__UpperCAmelCase						)		->       Any:
														self.scheduler.load_state_dict(__UpperCAmelCase						)




							def       snake_case			(					self						)		->       Dict:
														return self.scheduler.get_lr()






							def       snake_case			(					self    ,						*__UpperCAmelCase    ,						**__UpperCAmelCase						)		->       str:
														return self.scheduler.print_lr(*__UpperCAmelCase    ,						**__UpperCAmelCase						)











"""simple docstring"""




from manim import *






class       UpperCAmelCase_    (							_UpperCamelCase		):



      def 		snake_case_   (	self	:      Tuple		):
           _UpperCAmelCase     :      str														=       Rectangle(height=0.5	,      width=0.5		)
           _UpperCAmelCase     :      List[Any]														=       Rectangle(height=0.25	,      width=0.25		)
           _UpperCAmelCase     :      Tuple														=       Rectangle(height=0.46	,      width=0.46		).set_stroke(width=0		)

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

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

           _UpperCAmelCase     :      Tuple														=       [mem.copy() for i in range(6		)]
           _UpperCAmelCase     :      str														=       VGroup(*A		).arrange(A	,      buff=0		)

           _UpperCAmelCase     :      Any														=       Text("Model"	,      font_size=2_4		)
           _UpperCAmelCase     :      Any														=       Group(A	,      A		).arrange(A	,      buff=0.5	,      aligned_edge=A		)
           model.move_to([3, -1.0, 0]		)
           self.add(A		)

           _UpperCAmelCase     :      List[str]														=       []
           _UpperCAmelCase     :      Optional[int]														=       []
           _UpperCAmelCase     :      str														=       []

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

                _UpperCAmelCase     :      Dict														=       Rectangle(height=0.46 / 4	,      width=0.46 / 3		).set_stroke(width=0.0		).set_fill(A	,      opacity=0.7		)

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

           self.add(*A	,      *A	,      *A		)

           _UpperCAmelCase     :      Dict														=       [mem.copy() for i in range(6		)]
           _UpperCAmelCase     :      List[str]														=       VGroup(*A		).arrange(A	,      buff=0		)

           _UpperCAmelCase     :      List[str]														=       Text("Loaded Checkpoint"	,      font_size=2_4		)
           _UpperCAmelCase     :      Optional[int]														=       Group(A	,      A		).arrange(A	,      buff=0.5	,      aligned_edge=A		)
           checkpoint.move_to([3, 0.5, 0]		)

           self.add(A		)

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

           for i, rect in enumerate(A		):
                _UpperCAmelCase     :      str														=       fill.copy().set_fill(A	,      opacity=0.7		)
                target.move_to(A		)
                ckpt_arr.append(A		)

                _UpperCAmelCase     :      Dict														=       target.copy()
                if i < 5:
                     cpu_target.move_to(cpu_left_col_base[i + 1]		)
                else:
                     cpu_target.move_to(cpu_right_col_base[i - 5]		)
                ckpt_cpu_arr.append(A		)
           self.add(*A	,      *A		)

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

           _UpperCAmelCase     :      List[Any]														=       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(A	,      A		)

           _UpperCAmelCase     :      Tuple														=       MarkupText(
               f'<span fgcolor=\'{BLUE}\'>●</span> Checkpoint'	,      font_size=1_8	,      )

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

           _UpperCAmelCase     :      Any														=       MarkupText(
               f'Based on the passed in configuration, weights are stored in\na variety of np.memmaps on disk or to a particular device.'	,      font_size=2_4	,      )
           step_a.move_to([2, 2, 0]		)

           _UpperCAmelCase     :      int														=       [meta_mem.copy() for i in range(6		)]
           _UpperCAmelCase     :      Optional[int]														=       [meta_mem.copy() for i in range(6		)]
           _UpperCAmelCase     :      List[str]														=       VGroup(*A		).arrange(A	,      buff=0		)
           _UpperCAmelCase     :      Optional[int]														=       VGroup(*A		).arrange(A	,      buff=0		)
           _UpperCAmelCase     :      List[Any]														=       VGroup(A	,      A		).arrange(A	,      buff=0		)
           _UpperCAmelCase     :      Union[str, Any]														=       Text("Disk"	,      font_size=2_4		)
           _UpperCAmelCase     :      int														=       Group(A	,      A		).arrange(A	,      buff=0.5	,      aligned_edge=A		)
           disk.move_to([-4.0, -1.25, 0]		)
           self.play(Write(A	,      run_time=3		)	,      Write(A	,      run_time=1		)	,      Create(A	,      run_time=1		)		)

           _UpperCAmelCase     :      List[Any]														=       []
           for i, rect in enumerate(A		):
                _UpperCAmelCase     :      Any														=       rect.copy()
                target.generate_target()
                target.target.move_to(disk_left_col_base[i]		).scale(0.5		)
                animations.append(MoveToTarget(A	,      run_time=1.5		)		)
           self.play(*A		)

           self.play(FadeOut(A		)		)

           _UpperCAmelCase     :      int														=       MarkupText(f'Then, the checkpoint is removed from memory\nthrough garbage collection.'	,      font_size=2_4		)
           step_a.move_to([2, 2, 0]		)

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

           self.play(
               FadeOut(A	,      A	,      *A	,      *A		)	,      )

           self.wait()
