






from dataclasses import dataclass
from typing import Dict, 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 .attention_processor import AttentionProcessor, AttnProcessor
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, DiagonalGaussianDistribution, Encoder



@dataclass
class 							_UpperCamelCase			(a_			):
      snake_case_  =							42



class 							_UpperCamelCase			(a_							,		a_			):
      snake_case_  =							True


      @register_to_config
      def __init__(	self					,						__UpperCamelCase = 3					,						__UpperCamelCase = 3					,						__UpperCamelCase = ("DownEncoderBlock2D",)					,						__UpperCamelCase = ("UpDecoderBlock2D",)					,						__UpperCamelCase = (6_4,)					,						__UpperCamelCase = 1					,						__UpperCamelCase = "silu"					,						__UpperCamelCase = 4					,						__UpperCamelCase = 3_2					,						__UpperCamelCase = 3_2					,						__UpperCamelCase = 0.1_8_2_1_5					,						)->	Any:
           super().__init__()

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

           # pass init params to Decoder
           __lowerCAmelCase								=					Decoder(
               in_channels=__UpperCamelCase					,						out_channels=__UpperCamelCase					,						up_block_types=__UpperCamelCase					,						block_out_channels=__UpperCamelCase					,						layers_per_block=__UpperCamelCase					,						norm_num_groups=__UpperCamelCase					,						act_fn=__UpperCamelCase					,						)

           __lowerCAmelCase								=					nn.Convad(2 * latent_channels					,						2 * latent_channels					,						1	)
           __lowerCAmelCase								=					nn.Convad(__UpperCamelCase					,						__UpperCamelCase					,						1	)

           __lowerCAmelCase								=					False
           __lowerCAmelCase								=					False

           # only relevant if vae tiling is enabled
           __lowerCAmelCase								=					self.config.sample_size
           __lowerCAmelCase								=					(
               self.config.sample_size[0]
               if isinstance(self.config.sample_size					,						(list, tuple)	)
               else self.config.sample_size
           )
           __lowerCAmelCase								=					int(sample_size / (2 ** (len(self.config.block_out_channels	) - 1))	)
           __lowerCAmelCase								=					0.2_5


      def  __UpperCAmelCase		(	self					,						__UpperCamelCase					,						__UpperCamelCase=False	)->	Optional[Any]:
           if isinstance(__UpperCamelCase					,						(Encoder, Decoder)	):
                __lowerCAmelCase								=					value


      def  __UpperCAmelCase		(	self					,						__UpperCamelCase = True	)->	Optional[Any]:
           __lowerCAmelCase								=					use_tiling


      def  __UpperCAmelCase		(	self	)->	Union[str, Any]:
           self.enable_tiling(__UpperCamelCase	)


      def  __UpperCAmelCase		(	self	)->	Union[str, Any]:
           __lowerCAmelCase								=					True


      def  __UpperCAmelCase		(	self	)->	int:
           __lowerCAmelCase								=					False


      @property
      # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
      def  __UpperCAmelCase		(	self	)->	Dict[str, AttentionProcessor]:
           __lowerCAmelCase								=					{}

           def fn_recursive_add_processors(__UpperCamelCase					,						__UpperCamelCase					,						__UpperCamelCase	):
                if hasattr(__UpperCamelCase					,						"set_processor"	):
                     __lowerCAmelCase								=					module.processor

                for sub_name, child in module.named_children():
                     fn_recursive_add_processors(F"""{name}.{sub_name}"""					,						__UpperCamelCase					,						__UpperCamelCase	)

                return processors

           for name, module in self.named_children():
                fn_recursive_add_processors(__UpperCamelCase					,						__UpperCamelCase					,						__UpperCamelCase	)

           return processors


      def  __UpperCAmelCase		(	self					,						__UpperCamelCase	)->	List[Any]:
           __lowerCAmelCase								=					len(self.attn_processors.keys()	)

           if isinstance(__UpperCamelCase					,						__UpperCamelCase	) and len(__UpperCamelCase	) != count:
                raise ValueError(
                    F"""A dict of processors was passed, but the number of processors {len(__UpperCamelCase	)} does not match the"""
                    F""" number of attention layers: {count}. Please make sure to pass {count} processor classes."""	)

           def fn_recursive_attn_processor(__UpperCamelCase					,						__UpperCamelCase					,						__UpperCamelCase	):
                if hasattr(__UpperCamelCase					,						"set_processor"	):
                     if not isinstance(__UpperCamelCase					,						__UpperCamelCase	):
                          module.set_processor(__UpperCamelCase	)
                     else:
                          module.set_processor(processor.pop(F"""{name}.processor"""	)	)

                for sub_name, child in module.named_children():
                     fn_recursive_attn_processor(F"""{name}.{sub_name}"""					,						__UpperCamelCase					,						__UpperCamelCase	)

           for name, module in self.named_children():
                fn_recursive_attn_processor(__UpperCamelCase					,						__UpperCamelCase					,						__UpperCamelCase	)


      def  __UpperCAmelCase		(	self	)->	List[str]:
           self.set_attn_processor(AttnProcessor()	)


      @apply_forward_hook
      def  __UpperCAmelCase		(	self					,						__UpperCamelCase					,						__UpperCamelCase = True	)->	AutoencoderKLOutput:
           if self.use_tiling and (x.shape[-1] > self.tile_sample_min_size or x.shape[-2] > self.tile_sample_min_size):
                return self.tiled_encode(__UpperCamelCase					,						return_dict=__UpperCamelCase	)

           if self.use_slicing and x.shape[0] > 1:
                __lowerCAmelCase								=					[self.encoder(__UpperCamelCase	) for x_slice in x.split(1	)]
                __lowerCAmelCase								=					torch.cat(__UpperCamelCase	)
           else:
                __lowerCAmelCase								=					self.encoder(__UpperCamelCase	)

           __lowerCAmelCase								=					self.quant_conv(__UpperCamelCase	)
           __lowerCAmelCase								=					DiagonalGaussianDistribution(__UpperCamelCase	)

           if not return_dict:
                return (posterior,)

           return AutoencoderKLOutput(latent_dist=__UpperCamelCase	)


      def  __UpperCAmelCase		(	self					,						__UpperCamelCase					,						__UpperCamelCase = True	)->	Union[DecoderOutput, torch.FloatTensor]:
           if self.use_tiling and (z.shape[-1] > self.tile_latent_min_size or z.shape[-2] > self.tile_latent_min_size):
                return self.tiled_decode(__UpperCamelCase					,						return_dict=__UpperCamelCase	)

           __lowerCAmelCase								=					self.post_quant_conv(__UpperCamelCase	)
           __lowerCAmelCase								=					self.decoder(__UpperCamelCase	)

           if not return_dict:
                return (dec,)

           return DecoderOutput(sample=__UpperCamelCase	)


      @apply_forward_hook
      def  __UpperCAmelCase		(	self					,						__UpperCamelCase					,						__UpperCamelCase = True	)->	Union[DecoderOutput, torch.FloatTensor]:
           if self.use_slicing and z.shape[0] > 1:
                __lowerCAmelCase								=					[self._decode(__UpperCamelCase	).sample for z_slice in z.split(1	)]
                __lowerCAmelCase								=					torch.cat(__UpperCamelCase	)
           else:
                __lowerCAmelCase								=					self._decode(__UpperCamelCase	).sample

           if not return_dict:
                return (decoded,)

           return DecoderOutput(sample=__UpperCamelCase	)


      def  __UpperCAmelCase		(	self					,						__UpperCamelCase					,						__UpperCamelCase					,						__UpperCamelCase	)->	Optional[Any]:
           __lowerCAmelCase								=					min(a.shape[2]					,						b.shape[2]					,						__UpperCamelCase	)
           for y in range(__UpperCamelCase	):
                __lowerCAmelCase								=					a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
           return b


      def  __UpperCAmelCase		(	self					,						__UpperCamelCase					,						__UpperCamelCase					,						__UpperCamelCase	)->	Tuple:
           __lowerCAmelCase								=					min(a.shape[3]					,						b.shape[3]					,						__UpperCamelCase	)
           for x in range(__UpperCamelCase	):
                __lowerCAmelCase								=					a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
           return b


      def  __UpperCAmelCase		(	self					,						__UpperCamelCase					,						__UpperCamelCase = True	)->	AutoencoderKLOutput:
           __lowerCAmelCase								=					int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)	)
           __lowerCAmelCase								=					int(self.tile_latent_min_size * self.tile_overlap_factor	)
           __lowerCAmelCase								=					self.tile_latent_min_size - blend_extent

           # Split the image into 512x512 tiles and encode them separately.
           __lowerCAmelCase								=					[]
           for i in range(0					,						x.shape[2]					,						__UpperCamelCase	):
                __lowerCAmelCase								=					[]
                for j in range(0					,						x.shape[3]					,						__UpperCamelCase	):
                     __lowerCAmelCase								=					x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
                     __lowerCAmelCase								=					self.encoder(__UpperCamelCase	)
                     __lowerCAmelCase								=					self.quant_conv(__UpperCamelCase	)
                     row.append(__UpperCamelCase	)
                rows.append(__UpperCamelCase	)
           __lowerCAmelCase								=					[]
           for i, row in enumerate(__UpperCamelCase	):
                __lowerCAmelCase								=					[]
                for j, tile in enumerate(__UpperCamelCase	):
                     # blend the above tile and the left tile
                     # to the current tile and add the current tile to the result row
                     if i > 0:
                          __lowerCAmelCase								=					self.blend_v(rows[i - 1][j]					,						__UpperCamelCase					,						__UpperCamelCase	)
                     if j > 0:
                          __lowerCAmelCase								=					self.blend_h(row[j - 1]					,						__UpperCamelCase					,						__UpperCamelCase	)
                     result_row.append(tile[:, :, :row_limit, :row_limit]	)
                result_rows.append(torch.cat(__UpperCamelCase					,						dim=3	)	)

           __lowerCAmelCase								=					torch.cat(__UpperCamelCase					,						dim=2	)
           __lowerCAmelCase								=					DiagonalGaussianDistribution(__UpperCamelCase	)

           if not return_dict:
                return (posterior,)

           return AutoencoderKLOutput(latent_dist=__UpperCamelCase	)


      def  __UpperCAmelCase		(	self					,						__UpperCamelCase					,						__UpperCamelCase = True	)->	Union[DecoderOutput, torch.FloatTensor]:
           __lowerCAmelCase								=					int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)	)
           __lowerCAmelCase								=					int(self.tile_sample_min_size * self.tile_overlap_factor	)
           __lowerCAmelCase								=					self.tile_sample_min_size - blend_extent

           # Split z into overlapping 64x64 tiles and decode them separately.
           # The tiles have an overlap to avoid seams between tiles.
           __lowerCAmelCase								=					[]
           for i in range(0					,						z.shape[2]					,						__UpperCamelCase	):
                __lowerCAmelCase								=					[]
                for j in range(0					,						z.shape[3]					,						__UpperCamelCase	):
                     __lowerCAmelCase								=					z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
                     __lowerCAmelCase								=					self.post_quant_conv(__UpperCamelCase	)
                     __lowerCAmelCase								=					self.decoder(__UpperCamelCase	)
                     row.append(__UpperCamelCase	)
                rows.append(__UpperCamelCase	)
           __lowerCAmelCase								=					[]
           for i, row in enumerate(__UpperCamelCase	):
                __lowerCAmelCase								=					[]
                for j, tile in enumerate(__UpperCamelCase	):
                     # blend the above tile and the left tile
                     # to the current tile and add the current tile to the result row
                     if i > 0:
                          __lowerCAmelCase								=					self.blend_v(rows[i - 1][j]					,						__UpperCamelCase					,						__UpperCamelCase	)
                     if j > 0:
                          __lowerCAmelCase								=					self.blend_h(row[j - 1]					,						__UpperCamelCase					,						__UpperCamelCase	)
                     result_row.append(tile[:, :, :row_limit, :row_limit]	)
                result_rows.append(torch.cat(__UpperCamelCase					,						dim=3	)	)

           __lowerCAmelCase								=					torch.cat(__UpperCamelCase					,						dim=2	)
           if not return_dict:
                return (dec,)

           return DecoderOutput(sample=__UpperCamelCase	)


      def  __UpperCAmelCase		(	self					,						__UpperCamelCase					,						__UpperCamelCase = False					,						__UpperCamelCase = True					,						__UpperCamelCase = None					,						)->	Union[DecoderOutput, torch.FloatTensor]:
           __lowerCAmelCase								=					sample
           __lowerCAmelCase								=					self.encode(__UpperCamelCase	).latent_dist
           if sample_posterior:
                __lowerCAmelCase								=					posterior.sample(generator=__UpperCamelCase	)
           else:
                __lowerCAmelCase								=					posterior.mode()
           __lowerCAmelCase								=					self.decode(__UpperCamelCase	).sample

           if not return_dict:
                return (dec,)

           return DecoderOutput(sample=__UpperCamelCase	)











# Copyright 2021 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from pathlib import Path

import torch

from ...utils import is_npu_available, is_xpu_available
from .config_args import ClusterConfig, default_json_config_file
from .config_utils import SubcommandHelpFormatter


lowerCamelCase			:					List[str]         =	'''Create a default config file for Accelerate with only a few flags set.'''





def       __lowerCAmelCase      (						__snake_case="no"					, __snake_case = default_json_config_file					, __snake_case = False       ):
     __lowerCAmelCase								=					Path(__snake_case       )
     path.parent.mkdir(parents=__snake_case					, exist_ok=__snake_case       )
     if path.exists():
          print(
              F"""Configuration already exists at {save_location}, will not override. Run `accelerate config` manually or pass a different `save_location`."""       )
          return False
     __lowerCAmelCase								=					mixed_precision.lower()
     if mixed_precision not in ["no", "fp16", "bf16", "fp8"]:
          raise ValueError(
              F"""`mixed_precision` should be one of 'no', 'fp16', 'bf16', or 'fp8'. Received {mixed_precision}"""       )
     __lowerCAmelCase								=					{
         "compute_environment": "LOCAL_MACHINE",
         "mixed_precision": mixed_precision,
     }
     if torch.cuda.is_available():
          __lowerCAmelCase								=					torch.cuda.device_count()
          __lowerCAmelCase								=					num_gpus
          __lowerCAmelCase								=					False
          if num_gpus > 1:
               __lowerCAmelCase								=					"MULTI_GPU"
          else:
               __lowerCAmelCase								=					"NO"
     elif is_xpu_available() and use_xpu:
          __lowerCAmelCase								=					torch.xpu.device_count()
          __lowerCAmelCase								=					num_xpus
          __lowerCAmelCase								=					False
          if num_xpus > 1:
               __lowerCAmelCase								=					"MULTI_XPU"
          else:
               __lowerCAmelCase								=					"NO"
     elif is_npu_available():
          __lowerCAmelCase								=					torch.npu.device_count()
          __lowerCAmelCase								=					num_npus
          __lowerCAmelCase								=					False
          if num_npus > 1:
               __lowerCAmelCase								=					"MULTI_NPU"
          else:
               __lowerCAmelCase								=					"NO"
     else:
          __lowerCAmelCase								=					0
          __lowerCAmelCase								=					True
          __lowerCAmelCase								=					1
          __lowerCAmelCase								=					"NO"
     __lowerCAmelCase								=					ClusterConfig(**__snake_case       )
     config.to_json_file(__snake_case       )
     return path





def       __lowerCAmelCase      (						__snake_case					, __snake_case       ):
     __lowerCAmelCase								=					parser.add_parser("default"					, parents=__snake_case					, help=__snake_case					, formatter_class=__snake_case       )
     parser.add_argument(
         "--config_file"					, default=__snake_case					, help=(
             "The path to use to store the config file. Will default to a file named default_config.yaml in the cache "
             "location, which is the content of the environment `HF_HOME` suffixed with 'accelerate', or if you don't have "
             "such an environment variable, your cache directory ('~/.cache' or the content of `XDG_CACHE_HOME`) suffixed "
             "with 'huggingface'."
         )					, dest="save_location"					, )

     parser.add_argument(
         "--mixed_precision"					, choices=["no", "fp16", "bf16"]					, type=__snake_case					, help="Whether or not to use mixed precision training. "
         "Choose between FP16 and BF16 (bfloat16) training. "
         "BF16 training is only supported on Nvidia Ampere GPUs and PyTorch 1.10 or later."					, default="no"					, )
     parser.set_defaults(func=__snake_case       )
     return parser





def       __lowerCAmelCase      (						__snake_case       ):
     __lowerCAmelCase								=					write_basic_config(args.mixed_precision					, args.save_location       )
     if config_file:
          print(F"""accelerate configuration saved at {config_file}"""       )






