



"""simple docstring"""




from dataclasses import dataclass
from typing import Optional

import numpy as np
import torch
import torch.nn as nn

from ..utils import BaseOutput, is_torch_version, randn_tensor
from .attention_processor import SpatialNorm
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block






@dataclass
class       UpperCAmelCase_    (							_UpperCamelCase		):
      __SCREAMING_SNAKE_CASE		:  torch.FloatTensor







class       UpperCAmelCase_    (							nn.Module		):



      def __init__(	self	:      str	,      A	:      Optional[Any]=3	,      A	:      Optional[Any]=3	,      A	:      Tuple=("DownEncoderBlock2D",)	,      A	:      Optional[int]=(6_4,)	,      A	:      str=2	,      A	:      Any=3_2	,      A	:      Optional[int]="silu"	,      A	:      Union[str, Any]=True	,      ):
           super().__init__()
           _UpperCAmelCase     :      List[Any]														=       layers_per_block

           _UpperCAmelCase     :      Optional[Any]														=       torch.nn.Convad(
               A	,      block_out_channels[0]	,      kernel_size=3	,      stride=1	,      padding=1	,      )

           _UpperCAmelCase     :      Optional[int]														=       None
           _UpperCAmelCase     :      Optional[int]														=       nn.ModuleList([]		)

           # down
           _UpperCAmelCase     :      str														=       block_out_channels[0]
           for i, down_block_type in enumerate(A		):
                _UpperCAmelCase     :      List[Any]														=       output_channel
                _UpperCAmelCase     :      Tuple														=       block_out_channels[i]
                _UpperCAmelCase     :      List[Any]														=       i == len(A		) - 1

                _UpperCAmelCase     :      Optional[Any]														=       get_down_block(
                    A	,      num_layers=self.layers_per_block	,      in_channels=A	,      out_channels=A	,      add_downsample=not is_final_block	,      resnet_eps=1e-6	,      downsample_padding=0	,      resnet_act_fn=A	,      resnet_groups=A	,      attention_head_dim=A	,      temb_channels=A	,      )
                self.down_blocks.append(A		)

           # mid
           _UpperCAmelCase     :      str														=       UNetMidBlockaD(
               in_channels=block_out_channels[-1]	,      resnet_eps=1e-6	,      resnet_act_fn=A	,      output_scale_factor=1	,      resnet_time_scale_shift="default"	,      attention_head_dim=block_out_channels[-1]	,      resnet_groups=A	,      temb_channels=A	,      )

           # out
           _UpperCAmelCase     :      Dict														=       nn.GroupNorm(num_channels=block_out_channels[-1]	,      num_groups=A	,      eps=1e-6		)
           _UpperCAmelCase     :      Optional[int]														=       nn.SiLU()

           _UpperCAmelCase     :      List[Any]														=       2 * out_channels if double_z else out_channels
           _UpperCAmelCase     :      Dict														=       nn.Convad(block_out_channels[-1]	,      A	,      3	,      padding=1		)

           _UpperCAmelCase     :      Dict														=       False



      def 		snake_case_   (	self	:      Union[str, Any]	,      A	:      int		):
           _UpperCAmelCase     :      List[Any]														=       x
           _UpperCAmelCase     :      List[Any]														=       self.conv_in(A		)

           if self.training and self.gradient_checkpointing:

                def create_custom_forward(A	:      List[Any]		):
                     def custom_forward(*A	:      Tuple		):
                          return module(*A		)

                     return custom_forward

                # down
                if is_torch_version(">="	,      "1.11.0"		):
                     for down_block in self.down_blocks:
                          _UpperCAmelCase     :      Optional[int]														=       torch.utils.checkpoint.checkpoint(
                              create_custom_forward(A		)	,      A	,      use_reentrant=A		)
                     # middle
                     _UpperCAmelCase     :      Any														=       torch.utils.checkpoint.checkpoint(
                         create_custom_forward(self.mid_block		)	,      A	,      use_reentrant=A		)
                else:
                     for down_block in self.down_blocks:
                          _UpperCAmelCase     :      Any														=       torch.utils.checkpoint.checkpoint(create_custom_forward(A		)	,      A		)
                     # middle
                     _UpperCAmelCase     :      str														=       torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block		)	,      A		)

           else:
                # down
                for down_block in self.down_blocks:
                     _UpperCAmelCase     :      List[Any]														=       down_block(A		)

                # middle
                _UpperCAmelCase     :      Optional[int]														=       self.mid_block(A		)

           # post-process
           _UpperCAmelCase     :      Union[str, Any]														=       self.conv_norm_out(A		)
           _UpperCAmelCase     :      List[str]														=       self.conv_act(A		)
           _UpperCAmelCase     :      Any														=       self.conv_out(A		)

           return sample







class       UpperCAmelCase_    (							nn.Module		):



      def __init__(	self	:      int	,      A	:      int=3	,      A	:      Dict=3	,      A	:      int=("UpDecoderBlock2D",)	,      A	:      Union[str, Any]=(6_4,)	,      A	:      List[Any]=2	,      A	:      int=3_2	,      A	:      Optional[int]="silu"	,      A	:      Dict="group"	,      ):
           super().__init__()
           _UpperCAmelCase     :      Tuple														=       layers_per_block

           _UpperCAmelCase     :      List[Any]														=       nn.Convad(
               A	,      block_out_channels[-1]	,      kernel_size=3	,      stride=1	,      padding=1	,      )

           _UpperCAmelCase     :      Tuple														=       None
           _UpperCAmelCase     :      str														=       nn.ModuleList([]		)

           _UpperCAmelCase     :      Any														=       in_channels if norm_type == "spatial" else None

           # mid
           _UpperCAmelCase     :      List[Any]														=       UNetMidBlockaD(
               in_channels=block_out_channels[-1]	,      resnet_eps=1e-6	,      resnet_act_fn=A	,      output_scale_factor=1	,      resnet_time_scale_shift="default" if norm_type == "group" else norm_type	,      attention_head_dim=block_out_channels[-1]	,      resnet_groups=A	,      temb_channels=A	,      )

           # up
           _UpperCAmelCase     :      int														=       list(reversed(A		)		)
           _UpperCAmelCase     :      Union[str, Any]														=       reversed_block_out_channels[0]
           for i, up_block_type in enumerate(A		):
                _UpperCAmelCase     :      Optional[int]														=       output_channel
                _UpperCAmelCase     :      List[Any]														=       reversed_block_out_channels[i]

                _UpperCAmelCase     :      List[Any]														=       i == len(A		) - 1

                _UpperCAmelCase     :      Tuple														=       get_up_block(
                    A	,      num_layers=self.layers_per_block + 1	,      in_channels=A	,      out_channels=A	,      prev_output_channel=A	,      add_upsample=not is_final_block	,      resnet_eps=1e-6	,      resnet_act_fn=A	,      resnet_groups=A	,      attention_head_dim=A	,      temb_channels=A	,      resnet_time_scale_shift=A	,      )
                self.up_blocks.append(A		)
                _UpperCAmelCase     :      Dict														=       output_channel

           # out
           if norm_type == "spatial":
                _UpperCAmelCase     :      Dict														=       SpatialNorm(block_out_channels[0]	,      A		)
           else:
                _UpperCAmelCase     :      List[Any]														=       nn.GroupNorm(num_channels=block_out_channels[0]	,      num_groups=A	,      eps=1e-6		)
           _UpperCAmelCase     :      Tuple														=       nn.SiLU()
           _UpperCAmelCase     :      str														=       nn.Convad(block_out_channels[0]	,      A	,      3	,      padding=1		)

           _UpperCAmelCase     :      Union[str, Any]														=       False



      def 		snake_case_   (	self	:      List[str]	,      A	:      Optional[Any]	,      A	:      Any=None		):
           _UpperCAmelCase     :      Optional[Any]														=       z
           _UpperCAmelCase     :      Optional[int]														=       self.conv_in(A		)

           _UpperCAmelCase     :      Dict														=       next(iter(self.up_blocks.parameters()		)		).dtype
           if self.training and self.gradient_checkpointing:

                def create_custom_forward(A	:      Optional[Any]		):
                     def custom_forward(*A	:      Any		):
                          return module(*A		)

                     return custom_forward

                if is_torch_version(">="	,      "1.11.0"		):
                     # middle
                     _UpperCAmelCase     :      int														=       torch.utils.checkpoint.checkpoint(
                         create_custom_forward(self.mid_block		)	,      A	,      A	,      use_reentrant=A		)
                     _UpperCAmelCase     :      Optional[int]														=       sample.to(A		)

                     # up
                     for up_block in self.up_blocks:
                          _UpperCAmelCase     :      List[Any]														=       torch.utils.checkpoint.checkpoint(
                              create_custom_forward(A		)	,      A	,      A	,      use_reentrant=A		)
                else:
                     # middle
                     _UpperCAmelCase     :      str														=       torch.utils.checkpoint.checkpoint(
                         create_custom_forward(self.mid_block		)	,      A	,      A		)
                     _UpperCAmelCase     :      Optional[int]														=       sample.to(A		)

                     # up
                     for up_block in self.up_blocks:
                          _UpperCAmelCase     :      Union[str, Any]														=       torch.utils.checkpoint.checkpoint(create_custom_forward(A		)	,      A	,      A		)
           else:
                # middle
                _UpperCAmelCase     :      Union[str, Any]														=       self.mid_block(A	,      A		)
                _UpperCAmelCase     :      Union[str, Any]														=       sample.to(A		)

                # up
                for up_block in self.up_blocks:
                     _UpperCAmelCase     :      List[Any]														=       up_block(A	,      A		)

        # post-process
           if latent_embeds is None:
                _UpperCAmelCase     :      Optional[Any]														=       self.conv_norm_out(A		)
           else:
                _UpperCAmelCase     :      int														=       self.conv_norm_out(A	,      A		)
           _UpperCAmelCase     :      Optional[int]														=       self.conv_act(A		)
           _UpperCAmelCase     :      List[str]														=       self.conv_out(A		)

           return sample







class       UpperCAmelCase_    (							nn.Module		):



      def __init__(	self	:      Tuple	,      A	:      List[Any]	,      A	:      Tuple	,      A	:      int	,      A	:      Optional[int]=None	,      A	:      Tuple="random"	,      A	:      str=False	,      A	:      str=True		):
           super().__init__()
           _UpperCAmelCase     :      Dict														=       n_e
           _UpperCAmelCase     :      Tuple														=       vq_embed_dim
           _UpperCAmelCase     :      Optional[int]														=       beta
           _UpperCAmelCase     :      Optional[Any]														=       legacy

           _UpperCAmelCase     :      Dict														=       nn.Embedding(self.n_e	,      self.vq_embed_dim		)
           self.embedding.weight.data.uniform_(-1.0 / self.n_e	,      1.0 / self.n_e		)

           _UpperCAmelCase     :      Tuple														=       remap
           if self.remap is not None:
                self.register_buffer("used"	,      torch.tensor(np.load(self.remap		)		)		)
                _UpperCAmelCase     :      List[str]														=       self.used.shape[0]
                _UpperCAmelCase     :      Optional[int]														=       unknown_index  # "random" or "extra" or integer
                if self.unknown_index == "extra":
                     _UpperCAmelCase     :      Tuple														=       self.re_embed
                     _UpperCAmelCase     :      Dict														=       self.re_embed + 1
                print(
                    f'Remapping {self.n_e} indices to {self.re_embed} indices. '
                    f'Using {self.unknown_index} for unknown indices.'		)
           else:
                _UpperCAmelCase     :      Optional[Any]														=       n_e

           _UpperCAmelCase     :      str														=       sane_index_shape



      def 		snake_case_   (	self	:      int	,      A	:      Tuple		):
           _UpperCAmelCase     :      List[str]														=       inds.shape
           assert len(A		) > 1
           _UpperCAmelCase     :      List[Any]														=       inds.reshape(ishape[0]	,      -1		)
           _UpperCAmelCase     :      Dict														=       self.used.to(A		)
           _UpperCAmelCase     :      str														=       (inds[:, :, None] == used[None, None, ...]).long()
           _UpperCAmelCase     :      Optional[int]														=       match.argmax(-1		)
           _UpperCAmelCase     :      Optional[Any]														=       match.sum(2		) < 1
           if self.unknown_index == "random":
                _UpperCAmelCase     :      Optional[Any]														=       torch.randint(0	,      self.re_embed	,      size=new[unknown].shape		).to(device=new.device		)
           else:
                _UpperCAmelCase     :      Tuple														=       self.unknown_index
           return new.reshape(A		)



      def 		snake_case_   (	self	:      str	,      A	:      str		):
           _UpperCAmelCase     :      Tuple														=       inds.shape
           assert len(A		) > 1
           _UpperCAmelCase     :      Union[str, Any]														=       inds.reshape(ishape[0]	,      -1		)
           _UpperCAmelCase     :      Union[str, Any]														=       self.used.to(A		)
           if self.re_embed > self.used.shape[0]:  # extra token
                _UpperCAmelCase     :      int														=       0  # simply set to zero
           _UpperCAmelCase     :      List[str]														=       torch.gather(used[None, :][inds.shape[0] * [0], :]	,      1	,      A		)
           return back.reshape(A		)



      def 		snake_case_   (	self	:      Dict	,      A	:      Union[str, Any]		):
           # reshape z -> (batch, height, width, channel) and flatten
           _UpperCAmelCase     :      List[Any]														=       z.permute(0	,      2	,      3	,      1		).contiguous()
           _UpperCAmelCase     :      int														=       z.view(-1	,      self.vq_embed_dim		)

           # distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
           _UpperCAmelCase     :      List[Any]														=       torch.argmin(torch.cdist(A	,      self.embedding.weight		)	,      dim=1		)

           _UpperCAmelCase     :      Dict														=       self.embedding(A		).view(z.shape		)
           _UpperCAmelCase     :      List[str]														=       None
           _UpperCAmelCase     :      List[str]														=       None

           # compute loss for embedding
           if not self.legacy:
                _UpperCAmelCase     :      str														=       self.beta * torch.mean((z_q.detach() - z) ** 2		) + torch.mean((z_q - z.detach()) ** 2		)
           else:
                _UpperCAmelCase     :      Optional[Any]														=       torch.mean((z_q.detach() - z) ** 2		) + self.beta * torch.mean((z_q - z.detach()) ** 2		)

           # preserve gradients
           _UpperCAmelCase     :      str														=       z + (z_q - z).detach()

           # reshape back to match original input shape
           _UpperCAmelCase     :      Optional[Any]														=       z_q.permute(0	,      3	,      1	,      2		).contiguous()

           if self.remap is not None:
                _UpperCAmelCase     :      Any														=       min_encoding_indices.reshape(z.shape[0]	,      -1		)  # add batch axis
                _UpperCAmelCase     :      List[str]														=       self.remap_to_used(A		)
                _UpperCAmelCase     :      Optional[Any]														=       min_encoding_indices.reshape(-1	,      1		)  # flatten

           if self.sane_index_shape:
                _UpperCAmelCase     :      Optional[int]														=       min_encoding_indices.reshape(z_q.shape[0]	,      z_q.shape[2]	,      z_q.shape[3]		)

           return z_q, loss, (perplexity, min_encodings, min_encoding_indices)



      def 		snake_case_   (	self	:      Any	,      A	:      Optional[int]	,      A	:      Optional[Any]		):
           # shape specifying (batch, height, width, channel)
           if self.remap is not None:
                _UpperCAmelCase     :      int														=       indices.reshape(shape[0]	,      -1		)  # add batch axis
                _UpperCAmelCase     :      Tuple														=       self.unmap_to_all(A		)
                _UpperCAmelCase     :      Union[str, Any]														=       indices.reshape(-1		)  # flatten again

           # get quantized latent vectors
           _UpperCAmelCase     :      Union[str, Any]														=       self.embedding(A		)

           if shape is not None:
                _UpperCAmelCase     :      Tuple														=       z_q.view(A		)
                # reshape back to match original input shape
                _UpperCAmelCase     :      Dict														=       z_q.permute(0	,      3	,      1	,      2		).contiguous()

           return z_q







class       UpperCAmelCase_    (							_UpperCamelCase		):



      def __init__(	self	:      Union[str, Any]	,      A	:      Any	,      A	:      int=False		):
           _UpperCAmelCase     :      Any														=       parameters
           _UpperCAmelCase							,			_UpperCAmelCase     :      Any														=       torch.chunk(A	,      2	,      dim=1		)
           _UpperCAmelCase     :      int														=       torch.clamp(self.logvar	,      -30.0	,      20.0		)
           _UpperCAmelCase     :      List[str]														=       deterministic
           _UpperCAmelCase     :      Any														=       torch.exp(0.5 * self.logvar		)
           _UpperCAmelCase     :      Dict														=       torch.exp(self.logvar		)
           if self.deterministic:
                _UpperCAmelCase     :      List[Any]														=       torch.zeros_like(
                    self.mean	,      device=self.parameters.device	,      dtype=self.parameters.dtype		)



      def 		snake_case_   (	self	:      Union[str, Any]	,      A	:      Optional[torch.Generator] = None		):
           # make sure sample is on the same device as the parameters and has same dtype
           _UpperCAmelCase     :      Tuple														=       randn_tensor(
               self.mean.shape	,      generator=A	,      device=self.parameters.device	,      dtype=self.parameters.dtype		)
           _UpperCAmelCase     :      Optional[Any]														=       self.mean + self.std * sample
           return x



      def 		snake_case_   (	self	:      Tuple	,      A	:      Tuple=None		):
           if self.deterministic:
                return torch.Tensor([0.0]		)
           else:
                if other is None:
                     return 0.5 * torch.sum(torch.pow(self.mean	,      2		) + self.var - 1.0 - self.logvar	,      dim=[1, 2, 3]		)
                else:
                     return 0.5 * torch.sum(
                         torch.pow(self.mean - other.mean	,      2		) / other.var
                         + self.var / other.var
                         - 1.0
                         - self.logvar
                         + other.logvar	,      dim=[1, 2, 3]	,      )



      def 		snake_case_   (	self	:      Tuple	,      A	:      int	,      A	:      Any=[1, 2, 3]		):
           if self.deterministic:
                return torch.Tensor([0.0]		)
           _UpperCAmelCase     :      List[Any]														=       np.log(2.0 * np.pi		)
           return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean	,      2		) / self.var	,      dim=A		)



      def 		snake_case_   (	self	:      Tuple		):
           return self.mean





"""simple docstring"""




# We ignore warnings about stepping the scheduler since we step it ourselves during gradient accumulation

import warnings

from .state import AcceleratorState, GradientState


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






class       UpperCAmelCase_    :



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



      def 		snake_case_   (	self	:      Optional[int]	,      *A	:      List[Any]	,      **A	:      Optional[int]		):
           if not self.step_with_optimizer:
                # No link between scheduler and optimizer -> just step
                self.scheduler.step(*A	,      **A		)
                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(*A	,      **A		)
           else:
                # Otherwise the training dataloader batch size was multiplied by `num_processes`, so we need to do
                # num_processes steps per training step
                _UpperCAmelCase     :      int														=       AcceleratorState().num_processes
                for _ in range(A		):
                     # 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(*A	,      **A		)
                     else:
                          self.scheduler.step(*A	,      **A		)



      def 		snake_case_   (	self	:      Union[str, Any]		):
           return self.scheduler.get_last_lr()



      def 		snake_case_   (	self	:      str		):
           return self.scheduler.state_dict()



      def 		snake_case_   (	self	:      str	,      A	:      Optional[Any]		):
           self.scheduler.load_state_dict(A		)



      def 		snake_case_   (	self	:      Tuple		):
           return self.scheduler.get_lr()



      def 		snake_case_   (	self	:      Optional[int]	,      *A	:      List[str]	,      **A	:      Optional[Any]		):
           return self.scheduler.print_lr(*A	,      **A		)
