
"""simple docstring"""





import os
import tempfile
import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_torch


if is_torch_available():
					import torch
					from torch import nn

					from transformers import (
					    Adafactor,
					    AdamW,
					    get_constant_schedule,
					    get_constant_schedule_with_warmup,
					    get_cosine_schedule_with_warmup,
					    get_cosine_with_hard_restarts_schedule_with_warmup,
					    get_inverse_sqrt_schedule,
					    get_linear_schedule_with_warmup,
					    get_polynomial_decay_schedule_with_warmup,
					)

def 					_UpperCamelCase    (			UpperCamelCase				,       UpperCamelCase=10							)			->			Optional[Any]:


			"""simple docstring"""
			__UpperCAmelCase	: Any		=					[]
			for _ in range(__snake_case							):
						lrs.append(scheduler.get_lr()[0]							)
						scheduler.step()
			return lrs

def 					_UpperCamelCase    (			UpperCamelCase				,       UpperCamelCase=10							)			->			str:


			"""simple docstring"""
			__UpperCAmelCase	: List[Any]		=					[]
			for step in range(__snake_case							):
						lrs.append(scheduler.get_lr()[0]							)
						scheduler.step()
						if step == num_steps // 2:
									with tempfile.TemporaryDirectory() as tmpdirname:
												__UpperCAmelCase	: Union[str, Any]		=					os.path.join(__snake_case				,       "schedule.bin"							)
												torch.save(scheduler.state_dict()				,       __snake_case							)

												__UpperCAmelCase	: List[Any]		=					torch.load(__snake_case							)
												scheduler.load_state_dict(__snake_case							)
			return lrs



@require_torch
class      a__				(	unittest.TestCase   ):


			def       a_     (      self      :    int      ,  UpperCamelCase_      :    Dict      ,  UpperCamelCase_      :    Union[str, Any]      ,  UpperCamelCase_      :    Optional[Any]):


						"""simple docstring"""




						self.assertEqual(len(UpperCamelCase_)      ,  len(UpperCamelCase_))
						for a, b in zip(UpperCamelCase_      ,  UpperCamelCase_):
									self.assertAlmostEqual(UpperCamelCase_      ,  UpperCamelCase_      ,  delta=UpperCamelCase_)


			def       a_     (      self      :    List[str]):


						"""simple docstring"""




						__UpperCAmelCase	: Union[str, Any]		=					torch.tensor([0.1, -0.2, -0.1]      ,  requires_grad=UpperCamelCase_)
						__UpperCAmelCase	: List[Any]		=					torch.tensor([0.4, 0.2, -0.5])
						__UpperCAmelCase	: int		=					nn.MSELoss()
						# No warmup, constant schedule, no gradient clipping
						__UpperCAmelCase	: List[str]		=					AdamW(params=[w]      ,  lr=2e-1      ,  weight_decay=0.0)
						for _ in range(100):
									__UpperCAmelCase	: str		=					criterion(UpperCamelCase_      ,  UpperCamelCase_)
									loss.backward()
									optimizer.step()
									w.grad.detach_()  # No zero_grad() function on simple tensors. we do it ourselves.
									w.grad.zero_()
						self.assertListAlmostEqual(w.tolist()      ,  [0.4, 0.2, -0.5]      ,  tol=1e-2)


			def       a_     (      self      :    str):


						"""simple docstring"""




						__UpperCAmelCase	: int		=					torch.tensor([0.1, -0.2, -0.1]      ,  requires_grad=UpperCamelCase_)
						__UpperCAmelCase	: Optional[Any]		=					torch.tensor([0.4, 0.2, -0.5])
						__UpperCAmelCase	: Optional[int]		=					nn.MSELoss()
						# No warmup, constant schedule, no gradient clipping
						__UpperCAmelCase	: Any		=					Adafactor(
						    params=[w]      ,  lr=1e-2      ,  eps=(1e-30, 1e-3)      ,  clip_threshold=1.0      ,  decay_rate=-0.8      ,  betaa=UpperCamelCase_      ,  weight_decay=0.0      ,  relative_step=UpperCamelCase_      ,  scale_parameter=UpperCamelCase_      ,  warmup_init=UpperCamelCase_      ,  )
						for _ in range(1000):
									__UpperCAmelCase	: Any		=					criterion(UpperCamelCase_      ,  UpperCamelCase_)
									loss.backward()
									optimizer.step()
									w.grad.detach_()  # No zero_grad() function on simple tensors. we do it ourselves.
									w.grad.zero_()
						self.assertListAlmostEqual(w.tolist()      ,  [0.4, 0.2, -0.5]      ,  tol=1e-2)






@require_torch
class      a__				(	unittest.TestCase   ):
			lowercase_    =	nn.Linear(5_0							,    5_0   ) if is_torch_available() else None
			lowercase_    =	AdamW(m.parameters()							,    lr=1_0.0   ) if is_torch_available() else None
			lowercase_    =	1_0


			def       a_     (      self      :    str      ,  UpperCamelCase_      :    Tuple      ,  UpperCamelCase_      :    Optional[Any]      ,  UpperCamelCase_      :    Any      ,  UpperCamelCase_      :    Any=None):


						"""simple docstring"""




						self.assertEqual(len(UpperCamelCase_)      ,  len(UpperCamelCase_))
						for a, b in zip(UpperCamelCase_      ,  UpperCamelCase_):
									self.assertAlmostEqual(UpperCamelCase_      ,  UpperCamelCase_      ,  delta=UpperCamelCase_      ,  msg=UpperCamelCase_)


			def       a_     (      self      :    Union[str, Any]):


						"""simple docstring"""




						__UpperCAmelCase	: Dict		=					{"num_warmup_steps": 2, "num_training_steps": 10}
						# schedulers doct format
						# function: (sched_args_dict, expected_learning_rates)
						__UpperCAmelCase	: Dict		=					{
						    get_constant_schedule: ({}, [10.0] * self.num_steps),
						    get_constant_schedule_with_warmup: (
						        {"num_warmup_steps": 4},
						        [0.0, 2.5, 5.0, 7.5, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0],
						    ),
						    get_linear_schedule_with_warmup: (
						        {**common_kwargs},
						        [0.0, 5.0, 10.0, 8.75, 7.5, 6.25, 5.0, 3.75, 2.5, 1.25],
						    ),
						    get_cosine_schedule_with_warmup: (
						        {**common_kwargs},
						        [0.0, 5.0, 10.0, 9.61, 8.53, 6.91, 5.0, 3.08, 1.46, 0.38],
						    ),
						    get_cosine_with_hard_restarts_schedule_with_warmup: (
						        {**common_kwargs, "num_cycles": 2},
						        [0.0, 5.0, 10.0, 8.53, 5.0, 1.46, 10.0, 8.53, 5.0, 1.46],
						    ),
						    get_polynomial_decay_schedule_with_warmup: (
						        {**common_kwargs, "power": 2.0, "lr_end": 1e-7},
						        [0.0, 5.0, 10.0, 7.656, 5.625, 3.906, 2.5, 1.406, 0.625, 0.156],
						    ),
						    get_inverse_sqrt_schedule: (
						        {"num_warmup_steps": 2},
						        [0.0, 5.0, 10.0, 8.165, 7.071, 6.325, 5.774, 5.345, 5.0, 4.714],
						    ),
						}

						for scheduler_func, data in scheds.items():
									__UpperCAmelCase	: List[Any]		=					data

									__UpperCAmelCase	: List[str]		=					scheduler_func(self.optimizer      ,  **UpperCamelCase_)
									self.assertEqual(len([scheduler.get_lr()[0]])      ,  1)
									__UpperCAmelCase	: int		=					unwrap_schedule(UpperCamelCase_      ,  self.num_steps)
									self.assertListAlmostEqual(
									    UpperCamelCase_      ,  UpperCamelCase_      ,  tol=1e-2      ,  msg=F"failed for {scheduler_func} in normal scheduler"      ,  )

									__UpperCAmelCase	: str		=					scheduler_func(self.optimizer      ,  **UpperCamelCase_)
									if scheduler_func.__name__ != "get_constant_schedule":
												LambdaScheduleWrapper.wrap_scheduler(UpperCamelCase_)  # wrap to test picklability of the schedule
									__UpperCAmelCase	: Dict		=					unwrap_and_save_reload_schedule(UpperCamelCase_      ,  self.num_steps)
									self.assertListEqual(UpperCamelCase_      ,  UpperCamelCase_      ,  msg=F"failed for {scheduler_func} in save and reload")






class      a__				:


			def __init__(      self      :    str      ,  UpperCamelCase_      :    Optional[int]):


						"""simple docstring"""




						__UpperCAmelCase	: Any		=					fn


			def __call__(      self      :    Tuple      ,  *UpperCamelCase_      :    str      ,  **UpperCamelCase_      :    List[Any]):


						"""simple docstring"""




						return self.fn(*UpperCamelCase_      ,  **UpperCamelCase_)


			@classmethod
			def       a_     (      self      :    str      ,  UpperCamelCase_      :    Any):


						"""simple docstring"""




						__UpperCAmelCase	: Dict		=					list(map(self      ,  scheduler.lr_lambdas))








"""simple docstring"""





from manim import *



class      a__				(	__magic_name__   ):


			def       a_     (      self      :    str):


						"""simple docstring"""




						__UpperCAmelCase	: List[str]		=					Rectangle(height=0.5      ,  width=0.5)
						__UpperCAmelCase	: List[str]		=					Rectangle(height=0.25      ,  width=0.25)
						__UpperCAmelCase	: Dict		=					Rectangle(height=0.46      ,  width=0.46).set_stroke(width=0)

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

						__UpperCAmelCase	: List[str]		=					[mem.copy() for i in range(4)]
						__UpperCAmelCase	: List[Any]		=					VGroup(*UpperCamelCase_).arrange(UpperCamelCase_      ,  buff=0)
						__UpperCAmelCase	: Tuple		=					Text("GPU"      ,  font_size=24)
						__UpperCAmelCase	: Tuple		=					Group(UpperCamelCase_      ,  UpperCamelCase_).arrange(UpperCamelCase_      ,  buff=0.5      ,  aligned_edge=UpperCamelCase_)
						gpu.move_to([-1, -1, 0])
						self.add(UpperCamelCase_)

						__UpperCAmelCase	: Any		=					[mem.copy() for i in range(6)]
						__UpperCAmelCase	: int		=					VGroup(*UpperCamelCase_).arrange(UpperCamelCase_      ,  buff=0)

						__UpperCAmelCase	: Union[str, Any]		=					Text("Model"      ,  font_size=24)
						__UpperCAmelCase	: Optional[Any]		=					Group(UpperCamelCase_      ,  UpperCamelCase_).arrange(UpperCamelCase_      ,  buff=0.5      ,  aligned_edge=UpperCamelCase_)
						model.move_to([3, -1.0, 0])
						self.add(UpperCamelCase_)

						__UpperCAmelCase	: str		=					[]
						__UpperCAmelCase	: Tuple		=					[]
						__UpperCAmelCase	: Any		=					[]

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

									__UpperCAmelCase	: Optional[int]		=					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_      ,  *UpperCamelCase_)

						__UpperCAmelCase	: Optional[int]		=					[mem.copy() for i in range(6)]
						__UpperCAmelCase	: Dict		=					VGroup(*UpperCamelCase_).arrange(UpperCamelCase_      ,  buff=0)

						__UpperCAmelCase	: Tuple		=					Text("Loaded Checkpoint"      ,  font_size=24)
						__UpperCAmelCase	: Any		=					Group(UpperCamelCase_      ,  UpperCamelCase_).arrange(UpperCamelCase_      ,  buff=0.5      ,  aligned_edge=UpperCamelCase_)
						checkpoint.move_to([3, 0.5, 0])

						self.add(UpperCamelCase_)

						__UpperCAmelCase	: Optional[int]		=					[]
						__UpperCAmelCase	: int		=					[]

						for i, rect in enumerate(UpperCamelCase_):
									__UpperCAmelCase	: Optional[Any]		=					fill.copy().set_fill(UpperCamelCase_      ,  opacity=0.7)
									target.move_to(UpperCamelCase_)
									ckpt_arr.append(UpperCamelCase_)

									__UpperCAmelCase	: Optional[Any]		=					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(UpperCamelCase_)
						self.add(*UpperCamelCase_      ,  *UpperCamelCase_)

						__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=18      ,  )

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

						self.add(UpperCamelCase_      ,  UpperCamelCase_)

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

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

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

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

						self.play(FadeOut(UpperCamelCase_))

						__UpperCAmelCase	: Union[str, Any]		=					MarkupText(F"Then, the checkpoint is removed from memory\nthrough garbage collection."      ,  font_size=24)
						step_a.move_to([2, 2, 0])

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

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

						self.wait()


