





"""simple docstring"""





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 					_lowerCAmelCase   (     a       ):
       """simple docstring"""




       __magic_name__     :"DiagonalGaussianDistribution"






class 					_lowerCAmelCase   (     a  ,   a       ):
       """simple docstring"""




       __magic_name__     :Union[str, Any]		  =					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_82_15    , ):



         '''simple docstring'''

         super().__init__()

         # pass init params to Encoder
         lowerCAmelCase__			:Optional[Any]											=							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__			:Any											=							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__			:Dict											=							nn.Convad(2 * latent_channels    , 2 * latent_channels    , 1      )
         lowerCAmelCase__			:List[str]											=							nn.Convad(__UpperCAmelCase    , __UpperCAmelCase    , 1      )

         lowerCAmelCase__			:List[str]											=							False
         lowerCAmelCase__			:Any											=							False

         # only relevant if vae tiling is enabled
         lowerCAmelCase__			:Tuple											=							self.config.sample_size
         lowerCAmelCase__			:Optional[Any]											=							(
             self.config.sample_size[0]
             if isinstance(self.config.sample_size    , (list, tuple)      )
             else self.config.sample_size
         )
         lowerCAmelCase__			:Tuple											=							int(sample_size / (2 ** (len(self.config.block_out_channels      ) - 1))      )
         lowerCAmelCase__			:List[Any]											=							0.25






       def      snake_case						(    self    , __UpperCAmelCase    , __UpperCAmelCase=False      ):



         '''simple docstring'''

         if isinstance(__UpperCAmelCase    , (Encoder, Decoder)      ):
           lowerCAmelCase__			:Tuple											=							value






       def      snake_case						(    self    , __UpperCAmelCase = True      ):



         '''simple docstring'''

         lowerCAmelCase__			:Optional[Any]											=							use_tiling






       def      snake_case						(    self      ):



         '''simple docstring'''

         self.enable_tiling(__UpperCAmelCase      )






       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:Optional[int]											=							True






       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:str											=							False






       @property
       # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:Optional[Any]											=							{}

         def fn_recursive_add_processors(__UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase      ):
           if hasattr(__UpperCAmelCase    , 'set_processor'      ):
             lowerCAmelCase__			:str											=							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      snake_case						(    self    , __UpperCAmelCase      ):



         '''simple docstring'''

         lowerCAmelCase__			:Dict											=							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      snake_case						(    self      ):



         '''simple docstring'''

         self.set_attn_processor(AttnProcessor()      )






       @apply_forward_hook
       def      snake_case						(    self    , __UpperCAmelCase    , __UpperCAmelCase = True      ):



         '''simple docstring'''

         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__			:Tuple											=							[self.encoder(__UpperCAmelCase      ) for x_slice in x.split(1      )]
           lowerCAmelCase__			:Dict											=							torch.cat(__UpperCAmelCase      )
         else:
           lowerCAmelCase__			:Optional[Any]											=							self.encoder(__UpperCAmelCase      )

         lowerCAmelCase__			:Dict											=							self.quant_conv(__UpperCAmelCase      )
         lowerCAmelCase__			:List[str]											=							DiagonalGaussianDistribution(__UpperCAmelCase      )

         if not return_dict:
           return (posterior,)

         return AutoencoderKLOutput(latent_dist=__UpperCAmelCase      )






       def      snake_case						(    self    , __UpperCAmelCase    , __UpperCAmelCase = True      ):



         '''simple docstring'''

         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__			:List[Any]											=							self.post_quant_conv(__UpperCAmelCase      )
         lowerCAmelCase__			:Optional[Any]											=							self.decoder(__UpperCAmelCase      )

         if not return_dict:
           return (dec,)

         return DecoderOutput(sample=__UpperCAmelCase      )






       @apply_forward_hook
       def      snake_case						(    self    , __UpperCAmelCase    , __UpperCAmelCase = True      ):



         '''simple docstring'''

         if self.use_slicing and z.shape[0] > 1:
           lowerCAmelCase__			:str											=							[self._decode(__UpperCAmelCase      ).sample for z_slice in z.split(1      )]
           lowerCAmelCase__			:str											=							torch.cat(__UpperCAmelCase      )
         else:
           lowerCAmelCase__			:int											=							self._decode(__UpperCAmelCase      ).sample

         if not return_dict:
           return (decoded,)

         return DecoderOutput(sample=__UpperCAmelCase      )






       def      snake_case						(    self    , __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase      ):



         '''simple docstring'''

         lowerCAmelCase__			:Union[str, Any]											=							min(a.shape[2]    , b.shape[2]    , __UpperCAmelCase      )
         for y in range(__UpperCAmelCase      ):
           lowerCAmelCase__			:Optional[Any]											=							a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
         return b






       def      snake_case						(    self    , __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase      ):



         '''simple docstring'''

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






       def      snake_case						(    self    , __UpperCAmelCase    , __UpperCAmelCase = True      ):



         '''simple docstring'''

         lowerCAmelCase__			:List[Any]											=							int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)      )
         lowerCAmelCase__			:Optional[Any]											=							int(self.tile_latent_min_size * self.tile_overlap_factor      )
         lowerCAmelCase__			:Any											=							self.tile_latent_min_size - blend_extent

         # Split the image into 512x512 tiles and encode them separately.
         lowerCAmelCase__			:Optional[int]											=							[]
         for i in range(0    , x.shape[2]    , __UpperCAmelCase      ):
           lowerCAmelCase__			:Dict											=							[]
           for j in range(0    , x.shape[3]    , __UpperCAmelCase      ):
             lowerCAmelCase__			:Union[str, Any]											=							x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
             lowerCAmelCase__			:int											=							self.encoder(__UpperCAmelCase      )
             lowerCAmelCase__			:List[str]											=							self.quant_conv(__UpperCAmelCase      )
             row.append(__UpperCAmelCase      )
           rows.append(__UpperCAmelCase      )
         lowerCAmelCase__			:int											=							[]
         for i, row in enumerate(__UpperCAmelCase      ):
           lowerCAmelCase__			:Tuple											=							[]
           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__			:Tuple											=							self.blend_v(rows[i - 1][j]    , __UpperCAmelCase    , __UpperCAmelCase      )
             if j > 0:
               lowerCAmelCase__			:Union[str, Any]											=							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__			:List[Any]											=							torch.cat(__UpperCAmelCase    , dim=2      )
         lowerCAmelCase__			:int											=							DiagonalGaussianDistribution(__UpperCAmelCase      )

         if not return_dict:
           return (posterior,)

         return AutoencoderKLOutput(latent_dist=__UpperCAmelCase      )






       def      snake_case						(    self    , __UpperCAmelCase    , __UpperCAmelCase = True      ):



         '''simple docstring'''

         lowerCAmelCase__			:Dict											=							int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)      )
         lowerCAmelCase__			:List[str]											=							int(self.tile_sample_min_size * self.tile_overlap_factor      )
         lowerCAmelCase__			:List[str]											=							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__			:Any											=							[]
         for i in range(0    , z.shape[2]    , __UpperCAmelCase      ):
           lowerCAmelCase__			:int											=							[]
           for j in range(0    , z.shape[3]    , __UpperCAmelCase      ):
             lowerCAmelCase__			:Any											=							z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
             lowerCAmelCase__			:Dict											=							self.post_quant_conv(__UpperCAmelCase      )
             lowerCAmelCase__			:Dict											=							self.decoder(__UpperCAmelCase      )
             row.append(__UpperCAmelCase      )
           rows.append(__UpperCAmelCase      )
         lowerCAmelCase__			:List[Any]											=							[]
         for i, row in enumerate(__UpperCAmelCase      ):
           lowerCAmelCase__			:Optional[Any]											=							[]
           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__			:str											=							self.blend_v(rows[i - 1][j]    , __UpperCAmelCase    , __UpperCAmelCase      )
             if j > 0:
               lowerCAmelCase__			:Union[str, Any]											=							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__			:Dict											=							torch.cat(__UpperCAmelCase    , dim=2      )
         if not return_dict:
           return (dec,)

         return DecoderOutput(sample=__UpperCAmelCase      )






       def      snake_case						(    self    , __UpperCAmelCase    , __UpperCAmelCase = False    , __UpperCAmelCase = True    , __UpperCAmelCase = None    , ):



         '''simple docstring'''

         lowerCAmelCase__			:Tuple											=							sample
         lowerCAmelCase__			:List[str]											=							self.encode(__UpperCAmelCase      ).latent_dist
         if sample_posterior:
           lowerCAmelCase__			:Tuple											=							posterior.sample(generator=__UpperCAmelCase      )
         else:
           lowerCAmelCase__			:List[Any]											=							posterior.mode()
         lowerCAmelCase__			:List[Any]											=							self.decode(__UpperCAmelCase      ).sample

         if not return_dict:
           return (dec,)

         return DecoderOutput(sample=__UpperCAmelCase      )







"""simple docstring"""





from collections import defaultdict
from typing import Optional

from ..image_utils import load_image
from ..utils import (
    add_end_docstrings,
    is_torch_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, ChunkPipeline


if is_torch_available():
  import torch

  from ..models.auto.modeling_auto import MODEL_FOR_MASK_GENERATION_MAPPING

__A											=		logging.get_logger(__name__)






@add_end_docstrings(a       )
class 					_lowerCAmelCase   (     a       ):
       """simple docstring"""




       def __init__(    self    , **__UpperCAmelCase      ):



         '''simple docstring'''

         super().__init__(**__UpperCAmelCase      )
         requires_backends(self    , 'vision'      )
         requires_backends(self    , 'torch'      )

         if self.framework != "pt":
           raise ValueError(F"The {self.__class__} is only available in PyTorch."      )

         self.check_model_type(__UpperCAmelCase      )






       def      snake_case						(    self    , **__UpperCAmelCase      ):



         '''simple docstring'''

         lowerCAmelCase__			:List[str]											=							{}
         lowerCAmelCase__			:Tuple											=							{}
         lowerCAmelCase__			:Any											=							{}
         # preprocess args
         if "points_per_batch" in kwargs:
           lowerCAmelCase__			:Dict											=							kwargs['points_per_batch']
         if "points_per_crop" in kwargs:
           lowerCAmelCase__			:Union[str, Any]											=							kwargs['points_per_crop']
         if "crops_n_layers" in kwargs:
           lowerCAmelCase__			:Any											=							kwargs['crops_n_layers']
         if "crop_overlap_ratio" in kwargs:
           lowerCAmelCase__			:Any											=							kwargs['crop_overlap_ratio']
         if "crop_n_points_downscale_factor" in kwargs:
           lowerCAmelCase__			:Dict											=							kwargs['crop_n_points_downscale_factor']
         # postprocess args
         if "pred_iou_thresh" in kwargs:
           lowerCAmelCase__			:Tuple											=							kwargs['pred_iou_thresh']
         if "stability_score_offset" in kwargs:
           lowerCAmelCase__			:Optional[int]											=							kwargs['stability_score_offset']
         if "mask_threshold" in kwargs:
           lowerCAmelCase__			:List[Any]											=							kwargs['mask_threshold']
         if "stability_score_thresh" in kwargs:
           lowerCAmelCase__			:Optional[Any]											=							kwargs['stability_score_thresh']
         if "crops_nms_thresh" in kwargs:
           lowerCAmelCase__			:int											=							kwargs['crops_nms_thresh']
         if "output_rle_mask" in kwargs:
           lowerCAmelCase__			:Union[str, Any]											=							kwargs['output_rle_mask']
         if "output_bboxes_mask" in kwargs:
           lowerCAmelCase__			:Optional[Any]											=							kwargs['output_bboxes_mask']
         return preprocess_kwargs, forward_params, postprocess_kwargs






       def __call__(    self    , __UpperCAmelCase    , *__UpperCAmelCase    , __UpperCAmelCase=None    , __UpperCAmelCase=None    , **__UpperCAmelCase      ):



         '''simple docstring'''

         return super().__call__(__UpperCAmelCase    , *__UpperCAmelCase    , num_workers=__UpperCAmelCase    , batch_size=__UpperCAmelCase    , **__UpperCAmelCase      )






       def      snake_case						(    self    , __UpperCAmelCase    , __UpperCAmelCase=6_4    , __UpperCAmelCase = 0    , __UpperCAmelCase = 5_1_2 / 1_5_0_0    , __UpperCAmelCase = 3_2    , __UpperCAmelCase = 1    , ):



         '''simple docstring'''

         lowerCAmelCase__			:Union[str, Any]											=							load_image(__UpperCAmelCase      )
         lowerCAmelCase__			:int											=							self.image_processor.size['longest_edge']
         lowerCAmelCase__							,				lowerCAmelCase__							,				lowerCAmelCase__							,				lowerCAmelCase__			:int											=							self.image_processor.generate_crop_boxes(
             __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase      )
         lowerCAmelCase__			:Union[str, Any]											=							self.image_processor(images=__UpperCAmelCase    , return_tensors='pt'      )

         with self.device_placement():
           if self.framework == "pt":
             lowerCAmelCase__			:Optional[int]											=							self.get_inference_context()
             with inference_context():
               lowerCAmelCase__			:Any											=							self._ensure_tensor_on_device(__UpperCAmelCase    , device=self.device      )
               lowerCAmelCase__			:Tuple											=							self.model.get_image_embeddings(model_inputs.pop('pixel_values'      )      )
               lowerCAmelCase__			:Optional[int]											=							image_embeddings

         lowerCAmelCase__			:List[Any]											=							grid_points.shape[1]
         lowerCAmelCase__			:Union[str, Any]											=							points_per_batch if points_per_batch is not None else n_points

         if points_per_batch <= 0:
           raise ValueError(
               'Cannot have points_per_batch<=0. Must be >=1 to returned batched outputs. '
               'To return all points at once, set points_per_batch to None'      )

         for i in range(0    , __UpperCAmelCase    , __UpperCAmelCase      ):
           lowerCAmelCase__			:Optional[Any]											=							grid_points[:, i : i + points_per_batch, :, :]
           lowerCAmelCase__			:List[str]											=							input_labels[:, i : i + points_per_batch]
           lowerCAmelCase__			:List[Any]											=							i == n_points - points_per_batch
           yield {
               "input_points": batched_points,
               "input_labels": labels,
               "input_boxes": crop_boxes,
               "is_last": is_last,
               **model_inputs,
           }






       def      snake_case						(    self    , __UpperCAmelCase    , __UpperCAmelCase=0.88    , __UpperCAmelCase=0.95    , __UpperCAmelCase=0    , __UpperCAmelCase=1    , ):



         '''simple docstring'''

         lowerCAmelCase__			:Any											=							model_inputs.pop('input_boxes'      )
         lowerCAmelCase__			:Optional[int]											=							model_inputs.pop('is_last'      )
         lowerCAmelCase__			:Dict											=							model_inputs.pop('original_sizes'      ).tolist()
         lowerCAmelCase__			:Dict											=							model_inputs.pop('reshaped_input_sizes'      ).tolist()

         lowerCAmelCase__			:Optional[int]											=							self.model(**__UpperCAmelCase      )

         # post processing happens here in order to avoid CPU GPU copies of ALL the masks
         lowerCAmelCase__			:int											=							model_outputs['pred_masks']
         lowerCAmelCase__			:Optional[Any]											=							self.image_processor.post_process_masks(
             __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase    , binarize=__UpperCAmelCase      )
         lowerCAmelCase__			:Any											=							model_outputs['iou_scores']
         lowerCAmelCase__							,				lowerCAmelCase__							,				lowerCAmelCase__			:Tuple											=							self.image_processor.filter_masks(
             masks[0]    , iou_scores[0]    , original_sizes[0]    , input_boxes[0]    , __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase    , )
         return {
             "masks": masks,
             "is_last": is_last,
             "boxes": boxes,
             "iou_scores": iou_scores,
         }






       def      snake_case						(    self    , __UpperCAmelCase    , __UpperCAmelCase=False    , __UpperCAmelCase=False    , __UpperCAmelCase=0.7    , ):



         '''simple docstring'''

         lowerCAmelCase__			:Dict											=							[]
         lowerCAmelCase__			:Optional[Any]											=							[]
         lowerCAmelCase__			:int											=							[]
         for model_output in model_outputs:
           all_scores.append(model_output.pop('iou_scores'      )      )
           all_masks.extend(model_output.pop('masks'      )      )
           all_boxes.append(model_output.pop('boxes'      )      )

         lowerCAmelCase__			:Dict											=							torch.cat(__UpperCAmelCase      )
         lowerCAmelCase__			:Dict											=							torch.cat(__UpperCAmelCase      )
         lowerCAmelCase__							,				lowerCAmelCase__							,				lowerCAmelCase__							,				lowerCAmelCase__			:Any											=							self.image_processor.post_process_for_mask_generation(
             __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase      )

         lowerCAmelCase__			:Tuple											=							defaultdict(__UpperCAmelCase      )
         for output in model_outputs:
           for k, v in output.items():
             extra[k].append(__UpperCAmelCase      )

         lowerCAmelCase__			:Optional[int]											=							{}
         if output_rle_mask:
           lowerCAmelCase__			:str											=							rle_mask

         if output_bboxes_mask:
           lowerCAmelCase__			:Optional[int]											=							bounding_boxes

         return {"masks": output_masks, "scores": iou_scores, **optional, **extra}
