





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       (       _UpperCamelCase       ):

  """simple docstring"""


  a_              =					42





class 							_UpperCAmelCase       (       _UpperCamelCase						,       _UpperCamelCase       ):

  """simple docstring"""


  a_              =					True






  @register_to_config
  def __init__(							self					:						Tuple   ,    lowerCAmelCase_					:						int = 3   ,    lowerCAmelCase_					:						int = 3   ,    lowerCAmelCase_					:						Tuple[str] = ("DownEncoderBlock2D",)   ,    lowerCAmelCase_					:						Tuple[str] = ("UpDecoderBlock2D",)   ,    lowerCAmelCase_					:						Tuple[int] = (6_4,)   ,    lowerCAmelCase_					:						int = 1   ,    lowerCAmelCase_					:						str = "silu"   ,    lowerCAmelCase_					:						int = 4   ,    lowerCAmelCase_					:						int = 3_2   ,    lowerCAmelCase_					:						int = 3_2   ,    lowerCAmelCase_					:						float = 0.1_82_15   ,    )				->		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_   ,    )

     # 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_   ,    norm_num_groups=lowerCAmelCase_   ,    act_fn=lowerCAmelCase_   ,    )

     __lowerCAmelCase				      =      nn.Convad(2 * latent_channels   ,    2 * latent_channels   ,    1	)
     __lowerCAmelCase				      =      nn.Convad(lowerCAmelCase_   ,    lowerCAmelCase_   ,    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.25






  def 							lowercase						(							self					:						Dict   ,    lowerCAmelCase_					:						List[str]   ,    lowerCAmelCase_					:						Union[str, Any]=False	)				->		int:
     if isinstance(lowerCAmelCase_   ,    (Encoder, Decoder)	):
        __lowerCAmelCase				      =      value






  def 							lowercase						(							self					:						Dict   ,    lowerCAmelCase_					:						bool = True	)				->		Optional[int]:
     __lowerCAmelCase				      =      use_tiling






  def 							lowercase						(							self					:						Optional[Any]	)				->		Any:
     self.enable_tiling(lowerCAmelCase_	)






  def 							lowercase						(							self					:						List[str]	)				->		Optional[int]:
     __lowerCAmelCase				      =      True






  def 							lowercase						(							self					:						Dict	)				->		Optional[int]:
     __lowerCAmelCase				      =      False






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

     def fn_recursive_add_processors(lowerCAmelCase_					:						str   ,    lowerCAmelCase_					:						torch.nn.Module   ,    lowerCAmelCase_					:						Dict[str, AttentionProcessor]	):
        if hasattr(lowerCAmelCase_   ,    'set_processor'	):
           __lowerCAmelCase				      =      module.processor

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

        return processors

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

     return processors






  def 							lowercase						(							self					:						Optional[Any]   ,    lowerCAmelCase_					:						Union[AttentionProcessor, Dict[str, AttentionProcessor]]	)				->		str:
     __lowerCAmelCase				      =      len(self.attn_processors.keys()	)

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

     def fn_recursive_attn_processor(lowerCAmelCase_					:						str   ,    lowerCAmelCase_					:						torch.nn.Module   ,    lowerCAmelCase_					:						Optional[int]	):
        if hasattr(lowerCAmelCase_   ,    'set_processor'	):
           if not isinstance(lowerCAmelCase_   ,    lowerCAmelCase_	):
              module.set_processor(lowerCAmelCase_	)
           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}"""   ,    lowerCAmelCase_   ,    lowerCAmelCase_	)

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






  def 							lowercase						(							self					:						Dict	)				->		Optional[Any]:
     self.set_attn_processor(AttnProcessor()	)






  @apply_forward_hook
  def 							lowercase						(							self					:						Union[str, Any]   ,    lowerCAmelCase_					:						torch.FloatTensor   ,    lowerCAmelCase_					:						bool = 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(lowerCAmelCase_   ,    return_dict=lowerCAmelCase_	)

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

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

     if not return_dict:
        return (posterior,)

     return AutoencoderKLOutput(latent_dist=lowerCAmelCase_	)






  def 							lowercase						(							self					:						str   ,    lowerCAmelCase_					:						torch.FloatTensor   ,    lowerCAmelCase_					:						bool = 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(lowerCAmelCase_   ,    return_dict=lowerCAmelCase_	)

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

     if not return_dict:
        return (dec,)

     return DecoderOutput(sample=lowerCAmelCase_	)






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

     if not return_dict:
        return (decoded,)

     return DecoderOutput(sample=lowerCAmelCase_	)






  def 							lowercase						(							self					:						List[str]   ,    lowerCAmelCase_					:						Any   ,    lowerCAmelCase_					:						Union[str, Any]   ,    lowerCAmelCase_					:						Optional[int]	)				->		Any:
     __lowerCAmelCase				      =      min(a.shape[2]   ,    b.shape[2]   ,    lowerCAmelCase_	)
     for y in range(lowerCAmelCase_	):
        __lowerCAmelCase				      =      a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
     return b






  def 							lowercase						(							self					:						List[Any]   ,    lowerCAmelCase_					:						Union[str, Any]   ,    lowerCAmelCase_					:						Any   ,    lowerCAmelCase_					:						Tuple	)				->		Tuple:
     __lowerCAmelCase				      =      min(a.shape[3]   ,    b.shape[3]   ,    lowerCAmelCase_	)
     for x in range(lowerCAmelCase_	):
        __lowerCAmelCase				      =      a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
     return b






  def 							lowercase						(							self					:						int   ,    lowerCAmelCase_					:						torch.FloatTensor   ,    lowerCAmelCase_					:						bool = 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]   ,    lowerCAmelCase_	):
        __lowerCAmelCase				      =      []
        for j in range(0   ,    x.shape[3]   ,    lowerCAmelCase_	):
           __lowerCAmelCase				      =      x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
           __lowerCAmelCase				      =      self.encoder(lowerCAmelCase_	)
           __lowerCAmelCase				      =      self.quant_conv(lowerCAmelCase_	)
           row.append(lowerCAmelCase_	)
        rows.append(lowerCAmelCase_	)
     __lowerCAmelCase				      =      []
     for i, row in enumerate(lowerCAmelCase_	):
        __lowerCAmelCase				      =      []
        for j, tile in enumerate(lowerCAmelCase_	):
           # 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]   ,    lowerCAmelCase_   ,    lowerCAmelCase_	)
           if j > 0:
              __lowerCAmelCase				      =      self.blend_h(row[j - 1]   ,    lowerCAmelCase_   ,    lowerCAmelCase_	)
           result_row.append(tile[:, :, :row_limit, :row_limit]	)
        result_rows.append(torch.cat(lowerCAmelCase_   ,    dim=3	)	)

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

     if not return_dict:
        return (posterior,)

     return AutoencoderKLOutput(latent_dist=lowerCAmelCase_	)






  def 							lowercase						(							self					:						List[Any]   ,    lowerCAmelCase_					:						torch.FloatTensor   ,    lowerCAmelCase_					:						bool = 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]   ,    lowerCAmelCase_	):
        __lowerCAmelCase				      =      []
        for j in range(0   ,    z.shape[3]   ,    lowerCAmelCase_	):
           __lowerCAmelCase				      =      z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
           __lowerCAmelCase				      =      self.post_quant_conv(lowerCAmelCase_	)
           __lowerCAmelCase				      =      self.decoder(lowerCAmelCase_	)
           row.append(lowerCAmelCase_	)
        rows.append(lowerCAmelCase_	)
     __lowerCAmelCase				      =      []
     for i, row in enumerate(lowerCAmelCase_	):
        __lowerCAmelCase				      =      []
        for j, tile in enumerate(lowerCAmelCase_	):
           # 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]   ,    lowerCAmelCase_   ,    lowerCAmelCase_	)
           if j > 0:
              __lowerCAmelCase				      =      self.blend_h(row[j - 1]   ,    lowerCAmelCase_   ,    lowerCAmelCase_	)
           result_row.append(tile[:, :, :row_limit, :row_limit]	)
        result_rows.append(torch.cat(lowerCAmelCase_   ,    dim=3	)	)

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

     return DecoderOutput(sample=lowerCAmelCase_	)






  def 							lowercase						(							self					:						Tuple   ,    lowerCAmelCase_					:						torch.FloatTensor   ,    lowerCAmelCase_					:						bool = False   ,    lowerCAmelCase_					:						bool = True   ,    lowerCAmelCase_					:						Optional[torch.Generator] = None   ,    )				->		Union[DecoderOutput, torch.FloatTensor]:
     __lowerCAmelCase				      =      sample
     __lowerCAmelCase				      =      self.encode(lowerCAmelCase_	).latent_dist
     if sample_posterior:
        __lowerCAmelCase				      =      posterior.sample(generator=lowerCAmelCase_	)
     else:
        __lowerCAmelCase				      =      posterior.mode()
     __lowerCAmelCase				      =      self.decode(lowerCAmelCase_	).sample

     if not return_dict:
        return (dec,)

     return DecoderOutput(sample=lowerCAmelCase_	)





import unittest

from knapsack import greedy_knapsack as kp




class 							lowerCamelCase   (  unittest.TestCase ):


	def  A(    self):
							__UpperCAmelCase			:							Optional[Any]	    =	[1_0, 2_0, 3_0, 4_0, 5_0, 6_0]
							__UpperCAmelCase			:							str	    =	[2, 4, 6, 8, 1_0, 1_2]
							__UpperCAmelCase			:							List[Any]	    =	1_0_0
							self.assertEqual(kp.calc_profit(lowercase__				,     lowercase__				,     lowercase__)				,     2_1_0)


	def  A(    self):
							self.assertRaisesRegex(lowercase__				,     '''max_weight must greater than zero.''')


	def  A(    self):
							self.assertRaisesRegex(lowercase__				,     '''Weight can not be negative.''')


	def  A(    self):
							self.assertRaisesRegex(lowercase__				,     '''Profit can not be negative.''')


	def  A(    self):
							self.assertRaisesRegex(lowercase__				,     '''max_weight must greater than zero.''')


	def  A(    self):
							self.assertRaisesRegex(
							    lowercase__				,     '''The length of profit and weight must be same.''')


if __name__ == "__main__":
				unittest.main()

