


"""simple docstring"""







import inspect
from typing import Optional, Union

import numpy as np
import PIL
import torch
from torch.nn import functional as F
from torchvision import transforms
from transformers import CLIPFeatureExtractor, CLIPModel, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    DPMSolverMultistepScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput
from diffusers.utils import (
    PIL_INTERPOLATION,
    randn_tensor,
)


def 					lowercase      (    A_  ,       A_  ,       A_	)->				Optional[int]:
  '''simple docstring'''







  if isinstance(lowercase_  ,       torch.Tensor	):
    return image
  elif isinstance(lowercase_  ,       PIL.Image.Image	):
    a						:		Optional[int]											=      [image]

  if isinstance(image[0]  ,       PIL.Image.Image	):
    a						:		Dict											=      [np.array(i.resize((w, h)  ,       resample=PIL_INTERPOLATION["lanczos"]	)	)[None, :] for i in image]
    a						:		str											=      np.concatenate(lowercase_  ,       axis=0	)
    a						:		Any											=      np.array(lowercase_	).astype(np.floataa	) / 255.0
    a						:		str											=      image.transpose(0  ,       3  ,       1  ,       2	)
    a						:		List[str]											=      2.0 * image - 1.0
    a						:		str											=      torch.from_numpy(lowercase_	)
  elif isinstance(image[0]  ,       torch.Tensor	):
    a						:		Optional[int]											=      torch.cat(lowercase_  ,       dim=0	)
  return image


def 					lowercase      (    A_  ,       A_  ,       A_  ,       A_=0.9_9_9_5	)->				Any:
  '''simple docstring'''







  if not isinstance(lowercase_  ,       np.ndarray	):
    a						:		Tuple											=      True
    a						:		int											=      va.device
    a						:		Union[str, Any]											=      va.cpu().numpy()
    a						:		Optional[Any]											=      va.cpu().numpy()

  a						:		Optional[int]											=      np.sum(va * va / (np.linalg.norm(lowercase_	) * np.linalg.norm(lowercase_	))	)
  if np.abs(lowercase_	) > DOT_THRESHOLD:
    a						:		Optional[Any]											=      (1 - t) * va + t * va
  else:
    a						:		int											=      np.arccos(lowercase_	)
    a						:		Dict											=      np.sin(lowercase_	)
    a						:		Optional[Any]											=      theta_a * t
    a						:		str											=      np.sin(lowercase_	)
    a						:		List[str]											=      np.sin(theta_a - theta_t	) / sin_theta_a
    a						:		List[Any]											=      sin_theta_t / sin_theta_a
    a						:		Tuple											=      sa * va + sa * va

  if inputs_are_torch:
    a						:		int											=      torch.from_numpy(lowercase_	).to(lowercase_	)

  return va


def 					lowercase      (    A_  ,       A_	)->				Union[str, Any]:
  '''simple docstring'''







  a						:		Union[str, Any]											=      F.normalize(lowercase_  ,       dim=-1	)
  a						:		Dict											=      F.normalize(lowercase_  ,       dim=-1	)
  return (x - y).norm(dim=-1	).div(2	).arcsin().pow(2	).mul(2	)


def 					lowercase      (    A_  ,       A_	)->				Union[str, Any]:
  '''simple docstring'''







  for param in model.parameters():
    a						:		str											=      value







class 						_A  (  __snake_case  ):







    """simple docstring"""




    def __init__( self    :   int    ,			__UpperCAmelCase    :   Dict    ,			__UpperCAmelCase    :   str    ,			__UpperCAmelCase    :   str    ,			__UpperCAmelCase    :   List[str]    ,			__UpperCAmelCase    :   Dict    ,			__UpperCAmelCase    :   Dict    ,			__UpperCAmelCase    :   Dict    ,			__UpperCAmelCase    :   Optional[Any]=None    ,			__UpperCAmelCase    :   int=None    ,			__UpperCAmelCase    :   Union[str, Any]=None    ,			):
      super().__init__()
      self.register_modules(
          vae=__UpperCAmelCase    ,			text_encoder=__UpperCAmelCase    ,			clip_model=__UpperCAmelCase    ,			tokenizer=__UpperCAmelCase    ,			unet=__UpperCAmelCase    ,			scheduler=__UpperCAmelCase    ,			feature_extractor=__UpperCAmelCase    ,			coca_model=__UpperCAmelCase    ,			coca_tokenizer=__UpperCAmelCase    ,			coca_transform=__UpperCAmelCase    ,			)
      a						:		Dict											=      (
          feature_extractor.size
          if isinstance(feature_extractor.size    ,			__UpperCAmelCase)
          else feature_extractor.size["shortest_edge"]
      )
      a						:		Union[str, Any]											=      transforms.Normalize(mean=feature_extractor.image_mean    ,			std=feature_extractor.image_std)
      set_requires_grad(self.text_encoder    ,			__UpperCAmelCase)
      set_requires_grad(self.clip_model    ,			__UpperCAmelCase)






    def      __snake_case							( self    :   Optional[int]    ,			__UpperCAmelCase    :   Union[str, Any] = "auto"):
      if slice_size == "auto":
        # half the attention head size is usually a good trade-off between
        # speed and memory
        a						:		str											=      self.unet.config.attention_head_dim // 2
      self.unet.set_attention_slice(__UpperCAmelCase)






    def      __snake_case							( self    :   List[Any]):
      self.enable_attention_slicing(__UpperCAmelCase)






    def      __snake_case							( self    :   Optional[int]):
      set_requires_grad(self.vae    ,			__UpperCAmelCase)






    def      __snake_case							( self    :   List[Any]):
      set_requires_grad(self.vae    ,			__UpperCAmelCase)






    def      __snake_case							( self    :   str):
      set_requires_grad(self.unet    ,			__UpperCAmelCase)






    def      __snake_case							( self    :   Dict):
      set_requires_grad(self.unet    ,			__UpperCAmelCase)






    def      __snake_case							( self    :   List[str]    ,			__UpperCAmelCase    :   Union[str, Any]    ,			__UpperCAmelCase    :   List[Any]    ,			__UpperCAmelCase    :   Union[str, Any]):
      # get the original timestep using init_timestep
      a						:		Tuple											=      min(int(num_inference_steps * strength)    ,			__UpperCAmelCase)

      a						:		Dict											=      max(num_inference_steps - init_timestep    ,			0)
      a						:		Optional[Any]											=      self.scheduler.timesteps[t_start:]

      return timesteps, num_inference_steps - t_start






    def      __snake_case							( self    :   List[str]    ,			__UpperCAmelCase    :   Tuple    ,			__UpperCAmelCase    :   Optional[int]    ,			__UpperCAmelCase    :   Dict    ,			__UpperCAmelCase    :   Dict    ,			__UpperCAmelCase    :   Any    ,			__UpperCAmelCase    :   Optional[Any]=None):
      if not isinstance(__UpperCAmelCase    ,			torch.Tensor):
        raise ValueError(f'''`image` has to be of type `torch.Tensor` but is {type(__UpperCAmelCase)}''')

      a						:		List[Any]											=      image.to(device=__UpperCAmelCase    ,			dtype=__UpperCAmelCase)

      if isinstance(__UpperCAmelCase    ,			__UpperCAmelCase):
        a						:		Union[str, Any]											=      [
            self.vae.encode(image[i : i + 1]).latent_dist.sample(generator[i]) for i in range(__UpperCAmelCase)
        ]
        a						:		List[Any]											=      torch.cat(__UpperCAmelCase    ,			dim=0)
      else:
        a						:		Optional[int]											=      self.vae.encode(__UpperCAmelCase).latent_dist.sample(__UpperCAmelCase)

      # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
      a						:		Optional[Any]											=      0.18_215 * init_latents
      a						:		Any											=      init_latents.repeat_interleave(__UpperCAmelCase    ,			dim=0)

      a						:		Tuple											=      randn_tensor(init_latents.shape    ,			generator=__UpperCAmelCase    ,			device=__UpperCAmelCase    ,			dtype=__UpperCAmelCase)

      # get latents
      a						:		List[Any]											=      self.scheduler.add_noise(__UpperCAmelCase    ,			__UpperCAmelCase    ,			__UpperCAmelCase)
      a						:		str											=      init_latents

      return latents






    def      __snake_case							( self    :   Tuple    ,			__UpperCAmelCase    :   Dict):
      a						:		Dict											=      self.coca_transform(__UpperCAmelCase).unsqueeze(0)
      with torch.no_grad(), torch.cuda.amp.autocast():
        a						:		str											=      self.coca_model.generate(transformed_image.to(device=self.device    ,			dtype=self.coca_model.dtype))
      a						:		int											=      self.coca_tokenizer.decode(generated[0].cpu().numpy())
      return generated.split("<end_of_text>")[0].replace("<start_of_text>"    ,			"").rstrip(" .,")






    def      __snake_case							( self    :   Union[str, Any]    ,			__UpperCAmelCase    :   Any    ,			__UpperCAmelCase    :   Union[str, Any]):
      a						:		Union[str, Any]											=      self.feature_extractor.preprocess(__UpperCAmelCase)
      a						:		Any											=      torch.from_numpy(clip_image_input["pixel_values"][0]).unsqueeze(0).to(self.device).half()
      a						:		Optional[int]											=      self.clip_model.get_image_features(__UpperCAmelCase)
      a						:		Union[str, Any]											=      image_embeddings_clip / image_embeddings_clip.norm(p=2    ,			dim=-1    ,			keepdim=__UpperCAmelCase)
      a						:		Optional[Any]											=      image_embeddings_clip.repeat_interleave(__UpperCAmelCase    ,			dim=0)
      return image_embeddings_clip






    @torch.enable_grad()
    def      __snake_case							( self    :   Optional[Any]    ,			__UpperCAmelCase    :   Tuple    ,			__UpperCAmelCase    :   Union[str, Any]    ,			__UpperCAmelCase    :   Optional[Any]    ,			__UpperCAmelCase    :   Tuple    ,			__UpperCAmelCase    :   Tuple    ,			__UpperCAmelCase    :   str    ,			__UpperCAmelCase    :   Optional[Any]    ,			):
      a						:		Tuple											=      latents.detach().requires_grad_()

      a						:		Any											=      self.scheduler.scale_model_input(__UpperCAmelCase    ,			__UpperCAmelCase)

      # predict the noise residual
      a						:		Optional[Any]											=      self.unet(__UpperCAmelCase    ,			__UpperCAmelCase    ,			encoder_hidden_states=__UpperCAmelCase).sample

      if isinstance(self.scheduler    ,			(PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)):
        a						:		Dict											=      self.scheduler.alphas_cumprod[timestep]
        a						:		Optional[int]											=      1 - alpha_prod_t
        # compute predicted original sample from predicted noise also called
        # "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
        a						:		str											=      (latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

        a						:		Any											=      torch.sqrt(__UpperCAmelCase)
        a						:		Tuple											=      pred_original_sample * (fac) + latents * (1 - fac)
      elif isinstance(self.scheduler    ,			__UpperCAmelCase):
        a						:		str											=      self.scheduler.sigmas[index]
        a						:		List[str]											=      latents - sigma * noise_pred
      else:
        raise ValueError(f'''scheduler type {type(self.scheduler)} not supported''')

      # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
      a						:		Any											=      1 / 0.18_215 * sample
      a						:		List[str]											=      self.vae.decode(__UpperCAmelCase).sample
      a						:		Optional[int]											=      (image / 2 + 0.5).clamp(0    ,			1)

      a						:		List[Any]											=      transforms.Resize(self.feature_extractor_size)(__UpperCAmelCase)
      a						:		Optional[Any]											=      self.normalize(__UpperCAmelCase).to(latents.dtype)

      a						:		List[str]											=      self.clip_model.get_image_features(__UpperCAmelCase)
      a						:		Union[str, Any]											=      image_embeddings_clip / image_embeddings_clip.norm(p=2    ,			dim=-1    ,			keepdim=__UpperCAmelCase)

      a						:		Dict											=      spherical_dist_loss(__UpperCAmelCase    ,			__UpperCAmelCase).mean() * clip_guidance_scale

      a						:		List[Any]											=      -torch.autograd.grad(__UpperCAmelCase    ,			__UpperCAmelCase)[0]

      if isinstance(self.scheduler    ,			__UpperCAmelCase):
        a						:		Optional[int]											=      latents.detach() + grads * (sigma**2)
        a						:		List[str]											=      noise_pred_original
      else:
        a						:		Optional[int]											=      noise_pred_original - torch.sqrt(__UpperCAmelCase) * grads
      return noise_pred, latents






    @torch.no_grad()
    def __call__( self    :   int    ,			__UpperCAmelCase    :   Optional[int]    ,			__UpperCAmelCase    :   str    ,			__UpperCAmelCase    :   Tuple = None    ,			__UpperCAmelCase    :   Tuple = None    ,			__UpperCAmelCase    :   Tuple = 512    ,			__UpperCAmelCase    :   List[Any] = 512    ,			__UpperCAmelCase    :   Any = 0.6    ,			__UpperCAmelCase    :   List[Any] = 50    ,			__UpperCAmelCase    :   List[str] = 7.5    ,			__UpperCAmelCase    :   Dict = 1    ,			__UpperCAmelCase    :   List[Any] = 0.0    ,			__UpperCAmelCase    :   str = 100    ,			__UpperCAmelCase    :   int = None    ,			__UpperCAmelCase    :   List[str] = "pil"    ,			__UpperCAmelCase    :   Any = True    ,			__UpperCAmelCase    :   Optional[int] = 0.8    ,			__UpperCAmelCase    :   List[Any] = 0.1    ,			__UpperCAmelCase    :   str = 0.1    ,			):
      if isinstance(__UpperCAmelCase    ,			__UpperCAmelCase) and len(__UpperCAmelCase) != batch_size:
        raise ValueError(f'''You have passed {batch_size} batch_size, but only {len(__UpperCAmelCase)} generators.''')

      if height % 8 != 0 or width % 8 != 0:
        raise ValueError(f'''`height` and `width` have to be divisible by 8 but are {height} and {width}.''')

      if isinstance(__UpperCAmelCase    ,			torch.Generator) and batch_size > 1:
        a						:		Any											=      [generator] + [None] * (batch_size - 1)

      a						:		List[Any]											=      [
          ("model", self.coca_model is None),
          ("tokenizer", self.coca_tokenizer is None),
          ("transform", self.coca_transform is None),
      ]
      a						:		Optional[int]											=      [x[0] for x in coca_is_none if x[1]]
      a						:		List[Any]											=      ", ".join(__UpperCAmelCase)
      # generate prompts with coca model if prompt is None
      if content_prompt is None:
        if len(__UpperCAmelCase):
          raise ValueError(
              f'''Content prompt is None and CoCa [{coca_is_none_str}] is None.'''
              f'''Set prompt or pass Coca [{coca_is_none_str}] to DiffusionPipeline.''')
        a						:		int											=      self.get_image_description(__UpperCAmelCase)
      if style_prompt is None:
        if len(__UpperCAmelCase):
          raise ValueError(
              f'''Style prompt is None and CoCa [{coca_is_none_str}] is None.'''
              f''' Set prompt or pass Coca [{coca_is_none_str}] to DiffusionPipeline.''')
        a						:		Dict											=      self.get_image_description(__UpperCAmelCase)

      # get prompt text embeddings for content and style
      a						:		Tuple											=      self.tokenizer(
          __UpperCAmelCase    ,			padding="max_length"    ,			max_length=self.tokenizer.model_max_length    ,			truncation=__UpperCAmelCase    ,			return_tensors="pt"    ,			)
      a						:		Dict											=      self.text_encoder(content_text_input.input_ids.to(self.device))[0]

      a						:		int											=      self.tokenizer(
          __UpperCAmelCase    ,			padding="max_length"    ,			max_length=self.tokenizer.model_max_length    ,			truncation=__UpperCAmelCase    ,			return_tensors="pt"    ,			)
      a						:		str											=      self.text_encoder(style_text_input.input_ids.to(self.device))[0]

      a						:		int											=      slerp(__UpperCAmelCase    ,			__UpperCAmelCase    ,			__UpperCAmelCase)

      # duplicate text embeddings for each generation per prompt
      a						:		Dict											=      text_embeddings.repeat_interleave(__UpperCAmelCase    ,			dim=0)

      # set timesteps
      a						:		Any											=      "offset" in set(inspect.signature(self.scheduler.set_timesteps).parameters.keys())
      a						:		List[str]											=      {}
      if accepts_offset:
        a						:		List[Any]											=      1

      self.scheduler.set_timesteps(__UpperCAmelCase    ,			**__UpperCAmelCase)
      # Some schedulers like PNDM have timesteps as arrays
      # It's more optimized to move all timesteps to correct device beforehand
      self.scheduler.timesteps.to(self.device)

      a							,		a						:		Union[str, Any]											=      self.get_timesteps(__UpperCAmelCase    ,			__UpperCAmelCase    ,			self.device)
      a						:		Optional[Any]											=      timesteps[:1].repeat(__UpperCAmelCase)

      # Preprocess image
      a						:		Optional[Any]											=      preprocess(__UpperCAmelCase    ,			__UpperCAmelCase    ,			__UpperCAmelCase)
      a						:		Dict											=      self.prepare_latents(
          __UpperCAmelCase    ,			__UpperCAmelCase    ,			__UpperCAmelCase    ,			text_embeddings.dtype    ,			self.device    ,			__UpperCAmelCase)

      a						:		Optional[int]											=      preprocess(__UpperCAmelCase    ,			__UpperCAmelCase    ,			__UpperCAmelCase)
      a						:		Dict											=      self.prepare_latents(
          __UpperCAmelCase    ,			__UpperCAmelCase    ,			__UpperCAmelCase    ,			text_embeddings.dtype    ,			self.device    ,			__UpperCAmelCase)

      a						:		int											=      slerp(__UpperCAmelCase    ,			__UpperCAmelCase    ,			__UpperCAmelCase)

      if clip_guidance_scale > 0:
        a						:		Tuple											=      self.get_clip_image_embeddings(__UpperCAmelCase    ,			__UpperCAmelCase)
        a						:		Optional[Any]											=      self.get_clip_image_embeddings(__UpperCAmelCase    ,			__UpperCAmelCase)
        a						:		Optional[Any]											=      slerp(
            __UpperCAmelCase    ,			__UpperCAmelCase    ,			__UpperCAmelCase)

      # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
      # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
      # corresponds to doing no classifier free guidance.
      a						:		str											=      guidance_scale > 1.0
      # get unconditional embeddings for classifier free guidance
      if do_classifier_free_guidance:
        a						:		List[Any]											=      content_text_input.input_ids.shape[-1]
        a						:		Tuple											=      self.tokenizer([""]    ,			padding="max_length"    ,			max_length=__UpperCAmelCase    ,			return_tensors="pt")
        a						:		List[str]											=      self.text_encoder(uncond_input.input_ids.to(self.device))[0]
        # duplicate unconditional embeddings for each generation per prompt
        a						:		int											=      uncond_embeddings.repeat_interleave(__UpperCAmelCase    ,			dim=0)

        # For classifier free guidance, we need to do two forward passes.
        # Here we concatenate the unconditional and text embeddings into a single batch
        # to avoid doing two forward passes
        a						:		Optional[Any]											=      torch.cat([uncond_embeddings, text_embeddings])

      # get the initial random noise unless the user supplied it

      # Unlike in other pipelines, latents need to be generated in the target device
      # for 1-to-1 results reproducibility with the CompVis implementation.
      # However this currently doesn't work in `mps`.
      a						:		List[str]											=      (batch_size, self.unet.config.in_channels, height // 8, width // 8)
      a						:		Any											=      text_embeddings.dtype
      if latents is None:
        if self.device.type == "mps":
          # randn does not work reproducibly on mps
          a						:		Optional[Any]											=      torch.randn(__UpperCAmelCase    ,			generator=__UpperCAmelCase    ,			device="cpu"    ,			dtype=__UpperCAmelCase).to(
              self.device)
        else:
          a						:		List[Any]											=      torch.randn(__UpperCAmelCase    ,			generator=__UpperCAmelCase    ,			device=self.device    ,			dtype=__UpperCAmelCase)
      else:
        if latents.shape != latents_shape:
          raise ValueError(f'''Unexpected latents shape, got {latents.shape}, expected {latents_shape}''')
        a						:		List[str]											=      latents.to(self.device)

      # scale the initial noise by the standard deviation required by the scheduler
      a						:		List[Any]											=      latents * self.scheduler.init_noise_sigma

      # prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
      # eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
      # eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
      # and should be between [0, 1]
      a						:		List[Any]											=      "eta" in set(inspect.signature(self.scheduler.step).parameters.keys())
      a						:		str											=      {}
      if accepts_eta:
        a						:		Tuple											=      eta

      # check if the scheduler accepts generator
      a						:		Any											=      "generator" in set(inspect.signature(self.scheduler.step).parameters.keys())
      if accepts_generator:
        a						:		Optional[int]											=      generator

      with self.progress_bar(total=__UpperCAmelCase):
        for i, t in enumerate(__UpperCAmelCase):
          # expand the latents if we are doing classifier free guidance
          a						:		Dict											=      torch.cat([latents] * 2) if do_classifier_free_guidance else latents
          a						:		Any											=      self.scheduler.scale_model_input(__UpperCAmelCase    ,			__UpperCAmelCase)

          # predict the noise residual
          a						:		str											=      self.unet(__UpperCAmelCase    ,			__UpperCAmelCase    ,			encoder_hidden_states=__UpperCAmelCase).sample

          # perform classifier free guidance
          if do_classifier_free_guidance:
            a							,		a						:		Tuple											=      noise_pred.chunk(2)
            a						:		str											=      noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

          # perform clip guidance
          if clip_guidance_scale > 0:
            a						:		Optional[int]											=      (
                text_embeddings.chunk(2)[1] if do_classifier_free_guidance else text_embeddings
            )
            a							,		a						:		List[str]											=      self.cond_fn(
                __UpperCAmelCase    ,			__UpperCAmelCase    ,			__UpperCAmelCase    ,			__UpperCAmelCase    ,			__UpperCAmelCase    ,			__UpperCAmelCase    ,			__UpperCAmelCase    ,			)

          # compute the previous noisy sample x_t -> x_t-1
          a						:		str											=      self.scheduler.step(__UpperCAmelCase    ,			__UpperCAmelCase    ,			__UpperCAmelCase    ,			**__UpperCAmelCase).prev_sample

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
      a						:		Dict											=      1 / 0.18_215 * latents
      a						:		str											=      self.vae.decode(__UpperCAmelCase).sample

      a						:		Dict											=      (image / 2 + 0.5).clamp(0    ,			1)
      a						:		Dict											=      image.cpu().permute(0    ,			2    ,			3    ,			1).numpy()

      if output_type == "pil":
        a						:		Union[str, Any]											=      self.numpy_to_pil(__UpperCAmelCase)

      if not return_dict:
        return (image, None)

      return StableDiffusionPipelineOutput(images=__UpperCAmelCase    ,			nsfw_content_detected=__UpperCAmelCase)










"""simple docstring"""







from typing import Optional, Union

import torch
from torch import nn
from torch.nn import BCEWithLogitsLoss, CrossEntropyLoss, MSELoss

from ...activations import ACTaFN
from ...modeling_outputs import BaseModelOutputWithPoolingAndNoAttention, ImageClassifierOutputWithNoAttention
from ...modeling_utils import PreTrainedModel
from ...utils import add_code_sample_docstrings, add_start_docstrings, add_start_docstrings_to_model_forward, logging
from .configuration_mobilenet_va import MobileNetVaConfig


__lowercase    =				logging.get_logger(__name__)


# General docstring
__lowercase    =				"""MobileNetV1Config"""

# Base docstring
__lowercase    =				"""google/mobilenet_v1_1.0_224"""
__lowercase    =				[1, 1024, 7, 7]

# Image classification docstring
__lowercase    =				"""google/mobilenet_v1_1.0_224"""
__lowercase    =				"""tabby, tabby cat"""


__lowercase    =				[
    """google/mobilenet_v1_1.0_224""",
    """google/mobilenet_v1_0.75_192""",
    # See all MobileNetV1 models at https://huggingface.co/models?filter=mobilenet_v1
]


def 					lowercase      (    A_  ,       A_  ,       A_=None	)->				int:
  '''simple docstring'''







  a						:		Union[str, Any]											=      {}

  if isinstance(A_  ,       A_	):
    a						:		Tuple											=      model.mobilenet_va
  else:
    a						:		Optional[int]											=      model

  a						:		Any											=      "MobilenetV1/Conv2d_0/"
  a						:		List[Any]											=      backbone.conv_stem.convolution.weight
  a						:		Tuple											=      backbone.conv_stem.normalization.bias
  a						:		int											=      backbone.conv_stem.normalization.weight
  a						:		Optional[int]											=      backbone.conv_stem.normalization.running_mean
  a						:		Dict											=      backbone.conv_stem.normalization.running_var

  for i in range(13	):
    a						:		Dict											=      i + 1
    a						:		str											=      i * 2

    a						:		str											=      backbone.layer[pt_index]
    a						:		Tuple											=      F'''MobilenetV1/Conv2d_{tf_index}_depthwise/'''
    a						:		Union[str, Any]											=      pointer.convolution.weight
    a						:		Optional[int]											=      pointer.normalization.bias
    a						:		Any											=      pointer.normalization.weight
    a						:		Optional[int]											=      pointer.normalization.running_mean
    a						:		Dict											=      pointer.normalization.running_var

    a						:		Dict											=      backbone.layer[pt_index + 1]
    a						:		List[str]											=      F'''MobilenetV1/Conv2d_{tf_index}_pointwise/'''
    a						:		Dict											=      pointer.convolution.weight
    a						:		Any											=      pointer.normalization.bias
    a						:		List[Any]											=      pointer.normalization.weight
    a						:		Tuple											=      pointer.normalization.running_mean
    a						:		List[str]											=      pointer.normalization.running_var

  if isinstance(A_  ,       A_	):
    a						:		Dict											=      "MobilenetV1/Logits/Conv2d_1c_1x1/"
    a						:		List[Any]											=      model.classifier.weight
    a						:		Optional[int]											=      model.classifier.bias

  return tf_to_pt_map


def 					lowercase      (    A_  ,       A_  ,       A_	)->				Optional[int]:
  '''simple docstring'''







  try:
    import numpy as np
    import tensorflow as tf
  except ImportError:
    logger.error(
        "Loading a TensorFlow models in PyTorch, requires TensorFlow to be installed. Please see "
        "https://www.tensorflow.org/install/ for installation instructions."	)
    raise

  # Load weights from TF model
  a						:		List[str]											=      tf.train.list_variables(A_	)
  a						:		Dict											=      {}
  for name, shape in init_vars:
    logger.info(F'''Loading TF weight {name} with shape {shape}'''	)
    a						:		List[str]											=      tf.train.load_variable(A_  ,       A_	)
    a						:		int											=      array

  # Build TF to PyTorch weights loading map
  a						:		int											=      _build_tf_to_pytorch_map(A_  ,       A_  ,       A_	)

  for name, pointer in tf_to_pt_map.items():
    logger.info(F'''Importing {name}'''	)
    if name not in tf_weights:
      logger.info(F'''{name} not in tf pre-trained weights, skipping'''	)
      continue

    a						:		Tuple											=      tf_weights[name]

    if "depthwise_weights" in name:
      logger.info("Transposing depthwise"	)
      a						:		Optional[Any]											=      np.transpose(A_  ,       (2, 3, 0, 1)	)
    elif "weights" in name:
      logger.info("Transposing"	)
      if len(pointer.shape	) == 2:  # copying into linear layer
        a						:		List[str]											=      array.squeeze().transpose()
      else:
        a						:		List[str]											=      np.transpose(A_  ,       (3, 2, 0, 1)	)

    if pointer.shape != array.shape:
      raise ValueError(F'''Pointer shape {pointer.shape} and array shape {array.shape} mismatched'''	)

    logger.info(F'''Initialize PyTorch weight {name} {array.shape}'''	)
    a						:		Any											=      torch.from_numpy(A_	)

    tf_weights.pop(A_  ,       A_	)
    tf_weights.pop(name + "/RMSProp"  ,       A_	)
    tf_weights.pop(name + "/RMSProp_1"  ,       A_	)
    tf_weights.pop(name + "/ExponentialMovingAverage"  ,       A_	)

  logger.info(F'''Weights not copied to PyTorch model: {", ".join(tf_weights.keys()	)}'''	)
  return model


def 					lowercase      (    A_  ,       A_	)->				torch.Tensor:
  '''simple docstring'''







  a							,		a						:		str											=      features.shape[-2:]
  a							,		a						:		Tuple											=      conv_layer.stride
  a							,		a						:		Union[str, Any]											=      conv_layer.kernel_size

  if in_height % stride_height == 0:
    a						:		Union[str, Any]											=      max(kernel_height - stride_height  ,       0	)
  else:
    a						:		Tuple											=      max(kernel_height - (in_height % stride_height)  ,       0	)

  if in_width % stride_width == 0:
    a						:		Optional[int]											=      max(kernel_width - stride_width  ,       0	)
  else:
    a						:		int											=      max(kernel_width - (in_width % stride_width)  ,       0	)

  a						:		List[Any]											=      pad_along_width // 2
  a						:		List[str]											=      pad_along_width - pad_left
  a						:		str											=      pad_along_height // 2
  a						:		Any											=      pad_along_height - pad_top

  a						:		Tuple											=      (pad_left, pad_right, pad_top, pad_bottom)
  return nn.functional.pad(A_  ,       A_  ,       "constant"  ,       0.0	)






class 						_A  (  nn.Module  ):







    """simple docstring"""




    def __init__( self    :   Dict    ,			__UpperCAmelCase    :   MobileNetVaConfig    ,			__UpperCAmelCase    :   int    ,			__UpperCAmelCase    :   int    ,			__UpperCAmelCase    :   int    ,			__UpperCAmelCase    :   Optional[int] = 1    ,			__UpperCAmelCase    :   Optional[int] = 1    ,			__UpperCAmelCase    :   bool = False    ,			__UpperCAmelCase    :   Optional[bool] = True    ,			__UpperCAmelCase    :   Optional[bool or str] = True    ,			):
      super().__init__()
      a						:		Optional[Any]											=      config

      if in_channels % groups != 0:
        raise ValueError(f'''Input channels ({in_channels}) are not divisible by {groups} groups.''')
      if out_channels % groups != 0:
        raise ValueError(f'''Output channels ({out_channels}) are not divisible by {groups} groups.''')

      a						:		Dict											=      0 if config.tf_padding else int((kernel_size - 1) / 2)

      a						:		List[str]											=      nn.Convad(
          in_channels=__UpperCAmelCase    ,			out_channels=__UpperCAmelCase    ,			kernel_size=__UpperCAmelCase    ,			stride=__UpperCAmelCase    ,			padding=__UpperCAmelCase    ,			groups=__UpperCAmelCase    ,			bias=__UpperCAmelCase    ,			padding_mode="zeros"    ,			)

      if use_normalization:
        a						:		str											=      nn.BatchNormad(
            num_features=__UpperCAmelCase    ,			eps=config.layer_norm_eps    ,			momentum=0.9_997    ,			affine=__UpperCAmelCase    ,			track_running_stats=__UpperCAmelCase    ,			)
      else:
        a						:		List[str]											=      None

      if use_activation:
        if isinstance(__UpperCAmelCase    ,			__UpperCAmelCase):
          a						:		Optional[int]											=      ACTaFN[use_activation]
        elif isinstance(config.hidden_act    ,			__UpperCAmelCase):
          a						:		Optional[Any]											=      ACTaFN[config.hidden_act]
        else:
          a						:		str											=      config.hidden_act
      else:
        a						:		Dict											=      None






    def      __snake_case							( self    :   Tuple    ,			__UpperCAmelCase    :   torch.Tensor):
      if self.config.tf_padding:
        a						:		Tuple											=      apply_tf_padding(__UpperCAmelCase    ,			self.convolution)
      a						:		Dict											=      self.convolution(__UpperCAmelCase)
      if self.normalization is not None:
        a						:		List[str]											=      self.normalization(__UpperCAmelCase)
      if self.activation is not None:
        a						:		Optional[Any]											=      self.activation(__UpperCAmelCase)
      return features






class 						_A  (  _a  ):







    """simple docstring"""




    UpperCAmelCase	:							Union[str, Any]     =	MobileNetVaConfig
    UpperCAmelCase	:							str     =	load_tf_weights_in_mobilenet_va
    UpperCAmelCase	:							List[str]     =	"""mobilenet_v1"""
    UpperCAmelCase	:							Dict     =	"""pixel_values"""
    UpperCAmelCase	:							str     =	False






    def      __snake_case							( self    :   str    ,			__UpperCAmelCase    :   Union[nn.Linear, nn.Convad]):
      if isinstance(__UpperCAmelCase    ,			(nn.Linear, nn.Convad)):
        module.weight.data.normal_(mean=0.0    ,			std=self.config.initializer_range)
        if module.bias is not None:
          module.bias.data.zero_()
      elif isinstance(__UpperCAmelCase    ,			nn.BatchNormad):
        module.bias.data.zero_()
        module.weight.data.fill_(1.0)






__lowercase    =				R"""
    This model is a PyTorch [torch.nn.Module](https://pytorch.org/docs/stable/nn.html#torch.nn.Module) subclass. Use it
    as a regular PyTorch Module and refer to the PyTorch documentation for all matter related to general usage and
    behavior.

    Parameters:
        config ([`MobileNetV1Config`]): Model configuration class with all the parameters of the model.
            Initializing with a config file does not load the weights associated with the model, only the
            configuration. Check out the [`~PreTrainedModel.from_pretrained`] method to load the model weights.
"""

__lowercase    =				R"""
    Args:
        pixel_values (`torch.FloatTensor` of shape `(batch_size, num_channels, height, width)`):
            Pixel values. Pixel values can be obtained using [`AutoImageProcessor`]. See
            [`MobileNetV1ImageProcessor.__call__`] for details.
        output_hidden_states (`bool`, *optional*):
            Whether or not to return the hidden states of all layers. See `hidden_states` under returned tensors for
            more detail.
        return_dict (`bool`, *optional*):
            Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple.
"""






@add_start_docstrings(
    """The bare MobileNetV1 model outputting raw hidden-states without any specific head on top."""   ,_a   ,)
class 						_A  (  _a  ):







    """simple docstring"""




    def __init__( self    :   Optional[int]    ,			__UpperCAmelCase    :   MobileNetVaConfig    ,			__UpperCAmelCase    :   bool = True):
      super().__init__(__UpperCAmelCase)
      a						:		List[str]											=      config

      a						:		List[Any]											=      32
      a						:		Union[str, Any]											=      max(int(depth * config.depth_multiplier)    ,			config.min_depth)

      a						:		List[str]											=      MobileNetVaConvLayer(
          __UpperCAmelCase    ,			in_channels=config.num_channels    ,			out_channels=__UpperCAmelCase    ,			kernel_size=3    ,			stride=2    ,			)

      a						:		int											=      [1, 2, 1, 2, 1, 2, 1, 1, 1, 1, 1, 2, 1]

      a						:		List[Any]											=      nn.ModuleList()
      for i in range(13):
        a						:		Optional[Any]											=      out_channels

        if strides[i] == 2 or i == 0:
          depth *= 2
          a						:		List[str]											=      max(int(depth * config.depth_multiplier)    ,			config.min_depth)

        self.layer.append(
            MobileNetVaConvLayer(
                __UpperCAmelCase    ,			in_channels=__UpperCAmelCase    ,			out_channels=__UpperCAmelCase    ,			kernel_size=3    ,			stride=strides[i]    ,			groups=__UpperCAmelCase    ,			))

        self.layer.append(
            MobileNetVaConvLayer(
                __UpperCAmelCase    ,			in_channels=__UpperCAmelCase    ,			out_channels=__UpperCAmelCase    ,			kernel_size=1    ,			))

      a						:		Optional[Any]											=      nn.AdaptiveAvgPoolad((1, 1)) if add_pooling_layer else None

      # Initialize weights and apply final processing
      self.post_init()






    def      __snake_case							( self    :   Optional[Any]    ,			__UpperCAmelCase    :   Any):
      raise NotImplementedError






    @add_start_docstrings_to_model_forward(__UpperCAmelCase)
    @add_code_sample_docstrings(
        checkpoint=_CHECKPOINT_FOR_DOC    ,			output_type=__UpperCAmelCase    ,			config_class=_CONFIG_FOR_DOC    ,			modality="vision"    ,			expected_output=_EXPECTED_OUTPUT_SHAPE    ,			)
    def      __snake_case							( self    :   List[Any]    ,			__UpperCAmelCase    :   Optional[torch.Tensor] = None    ,			__UpperCAmelCase    :   Optional[bool] = None    ,			__UpperCAmelCase    :   Optional[bool] = None    ,			):
      a						:		str											=      (
          output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
      )
      a						:		Optional[Any]											=      return_dict if return_dict is not None else self.config.use_return_dict

      if pixel_values is None:
        raise ValueError("You have to specify pixel_values")

      a						:		Optional[int]											=      self.conv_stem(__UpperCAmelCase)

      a						:		Tuple											=      () if output_hidden_states else None

      for i, layer_module in enumerate(self.layer):
        a						:		Union[str, Any]											=      layer_module(__UpperCAmelCase)

        if output_hidden_states:
          a						:		str											=      all_hidden_states + (hidden_states,)

      a						:		Dict											=      hidden_states

      if self.pooler is not None:
        a						:		Optional[int]											=      torch.flatten(self.pooler(__UpperCAmelCase)    ,			start_dim=1)
      else:
        a						:		List[str]											=      None

      if not return_dict:
        return tuple(v for v in [last_hidden_state, pooled_output, all_hidden_states] if v is not None)

      return BaseModelOutputWithPoolingAndNoAttention(
          last_hidden_state=__UpperCAmelCase    ,			pooler_output=__UpperCAmelCase    ,			hidden_states=__UpperCAmelCase    ,			)






@add_start_docstrings(
    """
    MobileNetV1 model with an image classification head on top (a linear layer on top of the pooled features), e.g. for
    ImageNet.
    """   ,_a   ,)
class 						_A  (  _a  ):







    """simple docstring"""




    def __init__( self    :   int    ,			__UpperCAmelCase    :   MobileNetVaConfig):
      super().__init__(__UpperCAmelCase)

      a						:		Dict											=      config.num_labels
      a						:		Union[str, Any]											=      MobileNetVaModel(__UpperCAmelCase)

      a						:		List[Any]											=      self.mobilenet_va.layer[-1].convolution.out_channels

      # Classifier head
      a						:		Union[str, Any]											=      nn.Dropout(config.classifier_dropout_prob    ,			inplace=__UpperCAmelCase)
      a						:		Optional[int]											=      nn.Linear(__UpperCAmelCase    ,			config.num_labels) if config.num_labels > 0 else nn.Identity()

      # Initialize weights and apply final processing
      self.post_init()






    @add_start_docstrings_to_model_forward(__UpperCAmelCase)
    @add_code_sample_docstrings(
        checkpoint=_IMAGE_CLASS_CHECKPOINT    ,			output_type=__UpperCAmelCase    ,			config_class=_CONFIG_FOR_DOC    ,			expected_output=_IMAGE_CLASS_EXPECTED_OUTPUT    ,			)
    def      __snake_case							( self    :   Dict    ,			__UpperCAmelCase    :   Optional[torch.Tensor] = None    ,			__UpperCAmelCase    :   Optional[bool] = None    ,			__UpperCAmelCase    :   Optional[torch.Tensor] = None    ,			__UpperCAmelCase    :   Optional[bool] = None    ,			):
      a						:		Union[str, Any]											=      return_dict if return_dict is not None else self.config.use_return_dict

      a						:		Dict											=      self.mobilenet_va(__UpperCAmelCase    ,			output_hidden_states=__UpperCAmelCase    ,			return_dict=__UpperCAmelCase)

      a						:		List[Any]											=      outputs.pooler_output if return_dict else outputs[1]

      a						:		List[Any]											=      self.classifier(self.dropout(__UpperCAmelCase))

      a						:		Any											=      None
      if labels is not None:
        if self.config.problem_type is None:
          if self.num_labels == 1:
            a						:		List[str]											=      "regression"
          elif self.num_labels > 1 and (labels.dtype == torch.long or labels.dtype == torch.int):
            a						:		Any											=      "single_label_classification"
          else:
            a						:		Optional[Any]											=      "multi_label_classification"

        if self.config.problem_type == "regression":
          a						:		Union[str, Any]											=      MSELoss()
          if self.num_labels == 1:
            a						:		Tuple											=      loss_fct(logits.squeeze()    ,			labels.squeeze())
          else:
            a						:		Dict											=      loss_fct(__UpperCAmelCase    ,			__UpperCAmelCase)
        elif self.config.problem_type == "single_label_classification":
          a						:		Optional[Any]											=      CrossEntropyLoss()
          a						:		Tuple											=      loss_fct(logits.view(-1    ,			self.num_labels)    ,			labels.view(-1))
        elif self.config.problem_type == "multi_label_classification":
          a						:		Dict											=      BCEWithLogitsLoss()
          a						:		List[Any]											=      loss_fct(__UpperCAmelCase    ,			__UpperCAmelCase)

      if not return_dict:
        a						:		int											=      (logits,) + outputs[2:]
        return ((loss,) + output) if loss is not None else output

      return ImageClassifierOutputWithNoAttention(
          loss=__UpperCAmelCase    ,			logits=__UpperCAmelCase    ,			hidden_states=outputs.hidden_states    ,			)






