


'''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    _snake_case  (	A					,       A=10			)		->       Any:
							lowerCAmelCase__			  =      []
							for _ in range(A			):
														lrs.append(scheduler.get_lr()[0]			)
														scheduler.step()
							return lrs


def    _snake_case  (	A					,       A=10			)		->       Optional[Any]:
							lowerCAmelCase__			  =      []
							for step in range(A			):
														lrs.append(scheduler.get_lr()[0]			)
														scheduler.step()
														if step == num_steps // 2:
																					with tempfile.TemporaryDirectory() as tmpdirname:
																												lowerCAmelCase__			  =      os.path.join(A					,       '''schedule.bin'''			)
																												torch.save(scheduler.state_dict()					,       A			)

																												lowerCAmelCase__			  =      torch.load(A			)
																												scheduler.load_state_dict(A			)
							return lrs




@require_torch
class 						a__							(				unittest.TestCase    ):





							'''simple docstring'''






							def     __SCREAMING_SNAKE_CASE				(						self   ,       lowerCamelCase_   ,       lowerCamelCase_   ,       lowerCamelCase_  )			->				Any:
														self.assertEqual(len(lowerCamelCase_  )   ,       len(lowerCamelCase_  )  )
														for a, b in zip(lowerCamelCase_   ,       lowerCamelCase_  ):
																					self.assertAlmostEqual(lowerCamelCase_   ,       lowerCamelCase_   ,       delta=lowerCamelCase_  )







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Optional[Any]:
														lowerCAmelCase__			  =      torch.tensor([0.1, -0.2, -0.1]   ,       requires_grad=lowerCamelCase_  )
														lowerCAmelCase__			  =      torch.tensor([0.4, 0.2, -0.5]  )
														lowerCAmelCase__			  =      nn.MSELoss()
														# No warmup, constant schedule, no gradient clipping
														lowerCAmelCase__			  =      AdamW(params=[w]   ,       lr=2e-1   ,       weight_decay=0.0  )
														for _ in range(1_00  ):
																					lowerCAmelCase__			  =      criterion(lowerCamelCase_   ,       lowerCamelCase_  )
																					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     __SCREAMING_SNAKE_CASE				(						self  )			->				Union[str, Any]:
														lowerCAmelCase__			  =      torch.tensor([0.1, -0.2, -0.1]   ,       requires_grad=lowerCamelCase_  )
														lowerCAmelCase__			  =      torch.tensor([0.4, 0.2, -0.5]  )
														lowerCAmelCase__			  =      nn.MSELoss()
														# No warmup, constant schedule, no gradient clipping
														lowerCAmelCase__			  =      Adafactor(
														    params=[w]   ,       lr=1e-2   ,       eps=(1e-30, 1e-3)   ,       clip_threshold=1.0   ,       decay_rate=-0.8   ,       betaa=lowerCamelCase_   ,       weight_decay=0.0   ,       relative_step=lowerCamelCase_   ,       scale_parameter=lowerCamelCase_   ,       warmup_init=lowerCamelCase_   ,       )
														for _ in range(10_00  ):
																					lowerCAmelCase__			  =      criterion(lowerCamelCase_   ,       lowerCamelCase_  )
																					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    ):





							'''simple docstring'''






							lowercase__     :  List[str]      		= nn.Linear(5_0							,							5_0    ) if is_torch_available() else None
							lowercase__     :  Tuple      		= AdamW(m.parameters()							,							lr=1_0.0    ) if is_torch_available() else None
							lowercase__     :  int      		= 1_0







							def     __SCREAMING_SNAKE_CASE				(						self   ,       lowerCamelCase_   ,       lowerCamelCase_   ,       lowerCamelCase_   ,       lowerCamelCase_=None  )			->				Optional[Any]:
														self.assertEqual(len(lowerCamelCase_  )   ,       len(lowerCamelCase_  )  )
														for a, b in zip(lowerCamelCase_   ,       lowerCamelCase_  ):
																					self.assertAlmostEqual(lowerCamelCase_   ,       lowerCamelCase_   ,       delta=lowerCamelCase_   ,       msg=lowerCamelCase_  )







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				List[Any]:
														lowerCAmelCase__			  =      {'''num_warmup_steps''': 2, '''num_training_steps''': 10}
														# schedulers doct format
														# function: (sched_args_dict, expected_learning_rates)
														lowerCAmelCase__			  =      {
														    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():
																					lowerCAmelCase__  ,   lowerCAmelCase__			  =      data

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

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




class 						a__							:





							'''simple docstring'''






							def __init__(						self   ,       lowerCamelCase_  )			->				Optional[Any]:
														lowerCAmelCase__			  =      fn







							def __call__(						self   ,       *lowerCamelCase_   ,       **lowerCamelCase_  )			->				Optional[int]:
														return self.fn(*lowerCamelCase_   ,       **lowerCamelCase_  )







							@classmethod
							def     __SCREAMING_SNAKE_CASE				(						self   ,       lowerCamelCase_  )			->				Dict:
														lowerCAmelCase__			  =      list(map(self   ,       scheduler.lr_lambdas  )  )



'''simple docstring'''






from dataclasses import dataclass
from typing import Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, apply_forward_hook
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, Encoder, VectorQuantizer




@dataclass
class 						a__							(				a__    ):





							'''simple docstring'''






							lowercase__     :  torch.FloatTensor




class 						a__							(				a__							,							a__    ):





							'''simple docstring'''






							@register_to_config
							def __init__(						self   ,       lowerCamelCase_ = 3   ,       lowerCamelCase_ = 3   ,       lowerCamelCase_ = ("DownEncoderBlock2D",)   ,       lowerCamelCase_ = ("UpDecoderBlock2D",)   ,       lowerCamelCase_ = (64,)   ,       lowerCamelCase_ = 1   ,       lowerCamelCase_ = "silu"   ,       lowerCamelCase_ = 3   ,       lowerCamelCase_ = 32   ,       lowerCamelCase_ = 2_56   ,       lowerCamelCase_ = 32   ,       lowerCamelCase_ = None   ,       lowerCamelCase_ = 0.18_215   ,       lowerCamelCase_ = "group"   ,       )			->				Union[str, Any]:
														super().__init__()

														# pass init params to Encoder
														lowerCAmelCase__			  =      Encoder(
														    in_channels=lowerCamelCase_   ,       out_channels=lowerCamelCase_   ,       down_block_types=lowerCamelCase_   ,       block_out_channels=lowerCamelCase_   ,       layers_per_block=lowerCamelCase_   ,       act_fn=lowerCamelCase_   ,       norm_num_groups=lowerCamelCase_   ,       double_z=lowerCamelCase_   ,       )

														lowerCAmelCase__			  =      vq_embed_dim if vq_embed_dim is not None else latent_channels

														lowerCAmelCase__			  =      nn.Convad(lowerCamelCase_   ,       lowerCamelCase_   ,       1  )
														lowerCAmelCase__			  =      VectorQuantizer(lowerCamelCase_   ,       lowerCamelCase_   ,       beta=0.25   ,       remap=lowerCamelCase_   ,       sane_index_shape=lowerCamelCase_  )
														lowerCAmelCase__			  =      nn.Convad(lowerCamelCase_   ,       lowerCamelCase_   ,       1  )

														# pass init params to Decoder
														lowerCAmelCase__			  =      Decoder(
														    in_channels=lowerCamelCase_   ,       out_channels=lowerCamelCase_   ,       up_block_types=lowerCamelCase_   ,       block_out_channels=lowerCamelCase_   ,       layers_per_block=lowerCamelCase_   ,       act_fn=lowerCamelCase_   ,       norm_num_groups=lowerCamelCase_   ,       norm_type=lowerCamelCase_   ,       )







							@apply_forward_hook
							def     __SCREAMING_SNAKE_CASE				(						self   ,       lowerCamelCase_   ,       lowerCamelCase_ = True  )			->				VQEncoderOutput:
														lowerCAmelCase__			  =      self.encoder(lowerCamelCase_  )
														lowerCAmelCase__			  =      self.quant_conv(lowerCamelCase_  )

														if not return_dict:
																					return (h,)

														return VQEncoderOutput(latents=lowerCamelCase_  )







							@apply_forward_hook
							def     __SCREAMING_SNAKE_CASE				(						self   ,       lowerCamelCase_   ,       lowerCamelCase_ = False   ,       lowerCamelCase_ = True  )			->				Union[DecoderOutput, torch.FloatTensor]:
														# also go through quantization layer
														if not force_not_quantize:
																					lowerCAmelCase__  ,   lowerCAmelCase__  ,   lowerCAmelCase__			  =      self.quantize(lowerCamelCase_  )
														else:
																					lowerCAmelCase__			  =      h
														lowerCAmelCase__			  =      self.post_quant_conv(lowerCamelCase_  )
														lowerCAmelCase__			  =      self.decoder(lowerCamelCase_   ,       quant if self.config.norm_type == '''spatial''' else None  )

														if not return_dict:
																					return (dec,)

														return DecoderOutput(sample=lowerCamelCase_  )







							def     __SCREAMING_SNAKE_CASE				(						self   ,       lowerCamelCase_   ,       lowerCamelCase_ = True  )			->				Union[DecoderOutput, torch.FloatTensor]:
														lowerCAmelCase__			  =      sample
														lowerCAmelCase__			  =      self.encode(lowerCamelCase_  ).latents
														lowerCAmelCase__			  =      self.decode(lowerCamelCase_  ).sample

														if not return_dict:
																					return (dec,)

														return DecoderOutput(sample=lowerCamelCase_  )