





'''simple docstring'''





from typing import List, Optional, Union

import torch
from transformers import (
    XLMRobertaTokenizer,
)

from ...models import UNetaDConditionModel, VQModel
from ...pipelines import DiffusionPipeline
from ...pipelines.pipeline_utils import ImagePipelineOutput
from ...schedulers import DDIMScheduler, DDPMScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)
from .text_encoder import MultilingualCLIP


__A             =       logging.get_logger(__name__)  # pylint: disable=invalid-name

__A             =       "\n    Examples:\n        ```py\n        >>> from diffusers import KandinskyPipeline, KandinskyPriorPipeline\n        >>> import torch\n\n        >>> pipe_prior = KandinskyPriorPipeline.from_pretrained(\"kandinsky-community/Kandinsky-2-1-prior\")\n        >>> pipe_prior.to(\"cuda\")\n\n        >>> prompt = \"red cat, 4k photo\"\n        >>> out = pipe_prior(prompt)\n        >>> image_emb = out.image_embeds\n        >>> negative_image_emb = out.negative_image_embeds\n\n        >>> pipe = KandinskyPipeline.from_pretrained(\"kandinsky-community/kandinsky-2-1\")\n        >>> pipe.to(\"cuda\")\n\n        >>> image = pipe(\n        ...     prompt,\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=negative_image_emb,\n        ...     height=768,\n        ...     width=768,\n        ...     num_inference_steps=100,\n        ... ).images\n\n        >>> image[0].save(\"cat.png\")\n        ```\n"





def   lowerCAmelCase_				( __a			,					__a			,					__a=8	)  ->	Union[str, Any]:







  """simple docstring"""





  lowerCamelCase__:   List[str]    						=h // scale_factor**2
  if h % scale_factor**2 != 0:
    new_h += 1
  lowerCamelCase__:   Optional[int]    						=w // scale_factor**2
  if w % scale_factor**2 != 0:
    new_w += 1
  return new_h * scale_factor, new_w * scale_factor






class      _SCREAMING_SNAKE_CASE		(			__SCREAMING_SNAKE_CASE					):







 '''simple docstring'''

 def __init__(self       :     int ,				UpperCAmelCase_       :     MultilingualCLIP ,				UpperCAmelCase_       :     XLMRobertaTokenizer ,				UpperCAmelCase_       :     UNetaDConditionModel ,				UpperCAmelCase_       :     Union[DDIMScheduler, DDPMScheduler] ,				UpperCAmelCase_       :     VQModel ,				) ->Optional[int]:



   '''simple docstring'''



   super().__init__()

   self.register_modules(
       text_encoder=UpperCAmelCase__ ,				tokenizer=UpperCAmelCase__ ,				unet=UpperCAmelCase__ ,				scheduler=UpperCAmelCase__ ,				movq=UpperCAmelCase__ ,				)
   lowerCamelCase__:   List[str]    						=2 ** (len(self.movq.config.block_out_channels) - 1)

 def 							SCREAMING_SNAKE_CASE_      (self       :     Union[str, Any] ,				UpperCAmelCase_       :     List[str] ,				UpperCAmelCase_       :     List[str] ,				UpperCAmelCase_       :     str ,				UpperCAmelCase_       :     List[str] ,				UpperCAmelCase_       :     List[Any] ,				UpperCAmelCase_       :     Tuple) ->Dict:



   '''simple docstring'''



   if latents is None:
     lowerCamelCase__:   List[str]    						=randn_tensor(UpperCAmelCase__ ,				generator=UpperCAmelCase__ ,				device=UpperCAmelCase__ ,				dtype=UpperCAmelCase__)
   else:
     if latents.shape != shape:
       raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {shape}""")
     lowerCamelCase__:   Dict    						=latents.to(UpperCAmelCase__)

   lowerCamelCase__:   Any    						=latents * scheduler.init_noise_sigma
   return latents

 def 							SCREAMING_SNAKE_CASE_      (self       :     Tuple ,				UpperCAmelCase_       :     Optional[int] ,				UpperCAmelCase_       :     Any ,				UpperCAmelCase_       :     Optional[Any] ,				UpperCAmelCase_       :     str ,				UpperCAmelCase_       :     Any=None ,				) ->List[Any]:



   '''simple docstring'''



   lowerCamelCase__:   int    						=len(UpperCAmelCase__) if isinstance(UpperCAmelCase__ ,				UpperCAmelCase__) else 1
   # get prompt text embeddings
   lowerCamelCase__:   Union[str, Any]    						=self.tokenizer(
       UpperCAmelCase__ ,				padding="max_length" ,				truncation=UpperCAmelCase__ ,				max_length=77 ,				return_attention_mask=UpperCAmelCase__ ,				add_special_tokens=UpperCAmelCase__ ,				return_tensors="pt" ,				)

   lowerCamelCase__:   Union[str, Any]    						=text_inputs.input_ids
   lowerCamelCase__:   Union[str, Any]    						=self.tokenizer(UpperCAmelCase__ ,				padding="longest" ,				return_tensors="pt").input_ids

   if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(UpperCAmelCase__ ,				UpperCAmelCase__):
     lowerCamelCase__:   Union[str, Any]    						=self.tokenizer.batch_decode(untruncated_ids[:, self.tokenizer.model_max_length - 1 : -1])
     logger.warning(
         "The following part of your input was truncated because CLIP can only handle sequences up to"
         F""" {self.tokenizer.model_max_length} tokens: {removed_text}""")

   lowerCamelCase__:   Tuple    						=text_input_ids.to(UpperCAmelCase__)
   lowerCamelCase__:   Optional[int]    						=text_inputs.attention_mask.to(UpperCAmelCase__)

   lowerCamelCase__						,      lowerCamelCase__:   Optional[Any]    						=self.text_encoder(
       input_ids=UpperCAmelCase__ ,				attention_mask=UpperCAmelCase__)

   lowerCamelCase__:   Tuple    						=prompt_embeds.repeat_interleave(UpperCAmelCase__ ,				dim=0)
   lowerCamelCase__:   Optional[int]    						=text_encoder_hidden_states.repeat_interleave(UpperCAmelCase__ ,				dim=0)
   lowerCamelCase__:   Tuple    						=text_mask.repeat_interleave(UpperCAmelCase__ ,				dim=0)

   if do_classifier_free_guidance:
     lowerCamelCase__:   Union[str, Any]    						=42
     if negative_prompt is None:
       lowerCamelCase__:   Dict    						=[""] * batch_size
     elif type(UpperCAmelCase__) is not type(UpperCAmelCase__):
       raise TypeError(
           F"""`negative_prompt` should be the same type to `prompt`, but got {type(UpperCAmelCase__)} !="""
           F""" {type(UpperCAmelCase__)}.""")
     elif isinstance(UpperCAmelCase__ ,				UpperCAmelCase__):
       lowerCamelCase__:   Any    						=[negative_prompt]
     elif batch_size != len(UpperCAmelCase__):
       raise ValueError(
           F"""`negative_prompt`: {negative_prompt} has batch size {len(UpperCAmelCase__)}, but `prompt`:"""
           F""" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"""
           " the batch size of `prompt`.")
     else:
       lowerCamelCase__:   Optional[int]    						=negative_prompt

     lowerCamelCase__:   Tuple    						=self.tokenizer(
         UpperCAmelCase__ ,				padding="max_length" ,				max_length=77 ,				truncation=UpperCAmelCase__ ,				return_attention_mask=UpperCAmelCase__ ,				add_special_tokens=UpperCAmelCase__ ,				return_tensors="pt" ,				)
     lowerCamelCase__:   str    						=uncond_input.input_ids.to(UpperCAmelCase__)
     lowerCamelCase__:   Optional[int]    						=uncond_input.attention_mask.to(UpperCAmelCase__)

     lowerCamelCase__						,      lowerCamelCase__:   List[str]    						=self.text_encoder(
         input_ids=UpperCAmelCase__ ,				attention_mask=UpperCAmelCase__)

     # duplicate unconditional embeddings for each generation per prompt, using mps friendly method

     lowerCamelCase__:   Union[str, Any]    						=negative_prompt_embeds.shape[1]
     lowerCamelCase__:   str    						=negative_prompt_embeds.repeat(1 ,				UpperCAmelCase__)
     lowerCamelCase__:   Any    						=negative_prompt_embeds.view(batch_size * num_images_per_prompt ,				UpperCAmelCase__)

     lowerCamelCase__:   List[str]    						=uncond_text_encoder_hidden_states.shape[1]
     lowerCamelCase__:   Tuple    						=uncond_text_encoder_hidden_states.repeat(1 ,				UpperCAmelCase__ ,				1)
     lowerCamelCase__:   Union[str, Any]    						=uncond_text_encoder_hidden_states.view(
         batch_size * num_images_per_prompt ,				UpperCAmelCase__ ,				-1)
     lowerCamelCase__:   str    						=uncond_text_mask.repeat_interleave(UpperCAmelCase__ ,				dim=0)

     # done duplicates

     # 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
     lowerCamelCase__:   Optional[int]    						=torch.cat([negative_prompt_embeds, prompt_embeds])
     lowerCamelCase__:   int    						=torch.cat([uncond_text_encoder_hidden_states, text_encoder_hidden_states])

     lowerCamelCase__:   Union[str, Any]    						=torch.cat([uncond_text_mask, text_mask])

   return prompt_embeds, text_encoder_hidden_states, text_mask

 def 							SCREAMING_SNAKE_CASE_      (self       :     int ,				UpperCAmelCase_       :     Optional[Any]=0) ->List[str]:



   '''simple docstring'''



   if is_accelerate_available():
     from accelerate import cpu_offload
   else:
     raise ImportError("Please install accelerate via `pip install accelerate`")

   lowerCamelCase__:   Union[str, Any]    						=torch.device(F"""cuda:{gpu_id}""")

   lowerCamelCase__:   List[str]    						=[
       self.unet,
       self.text_encoder,
       self.movq,
   ]
   for cpu_offloaded_model in models:
     if cpu_offloaded_model is not None:
       cpu_offload(UpperCAmelCase__ ,				UpperCAmelCase__)

 def 							SCREAMING_SNAKE_CASE_      (self       :     int ,				UpperCAmelCase_       :     Dict=0) ->List[str]:



   '''simple docstring'''



   if is_accelerate_available() and is_accelerate_version(">=" ,				"0.17.0.dev0"):
     from accelerate import cpu_offload_with_hook
   else:
     raise ImportError("`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher.")

   lowerCamelCase__:   Union[str, Any]    						=torch.device(F"""cuda:{gpu_id}""")

   if self.device.type != "cpu":
     self.to("cpu" ,				silence_dtype_warnings=UpperCAmelCase__)
     torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

   lowerCamelCase__:   int    						=None
   for cpu_offloaded_model in [self.text_encoder, self.unet, self.movq]:
     lowerCamelCase__						,      lowerCamelCase__:   List[str]    						=cpu_offload_with_hook(UpperCAmelCase__ ,				UpperCAmelCase__ ,				prev_module_hook=UpperCAmelCase__)

   if self.safety_checker is not None:
     lowerCamelCase__						,      lowerCamelCase__:   Optional[int]    						=cpu_offload_with_hook(self.safety_checker ,				UpperCAmelCase__ ,				prev_module_hook=UpperCAmelCase__)

   # We'll offload the last model manually.
   lowerCamelCase__:   List[Any]    						=hook

 @property
 # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
 def 							SCREAMING_SNAKE_CASE_      (self       :     List[Any]) ->int:



   '''simple docstring'''



   if not hasattr(self.unet ,				"_hf_hook"):
     return self.device
   for module in self.unet.modules():
     if (
         hasattr(UpperCAmelCase__ ,				"_hf_hook")
         and hasattr(module._hf_hook ,				"execution_device")
         and module._hf_hook.execution_device is not None
     ):
       return torch.device(module._hf_hook.execution_device)
   return self.device







 @torch.no_grad()
 @replace_example_docstring(UpperCAmelCase__)
 def __call__(self       :     int ,				UpperCAmelCase_       :     Union[str, List[str]] ,				UpperCAmelCase_       :     Union[torch.FloatTensor, List[torch.FloatTensor]] ,				UpperCAmelCase_       :     Union[torch.FloatTensor, List[torch.FloatTensor]] ,				UpperCAmelCase_       :     Optional[Union[str, List[str]]] = None ,				UpperCAmelCase_       :     int = 512 ,				UpperCAmelCase_       :     int = 512 ,				UpperCAmelCase_       :     int = 100 ,				UpperCAmelCase_       :     float = 4.0 ,				UpperCAmelCase_       :     int = 1 ,				UpperCAmelCase_       :     Optional[Union[torch.Generator, List[torch.Generator]]] = None ,				UpperCAmelCase_       :     Optional[torch.FloatTensor] = None ,				UpperCAmelCase_       :     Optional[str] = "pil" ,				UpperCAmelCase_       :     bool = True ,				) ->str:



   '''simple docstring'''



   if isinstance(UpperCAmelCase__ ,				UpperCAmelCase__):
     lowerCamelCase__:   Dict    						=1
   elif isinstance(UpperCAmelCase__ ,				UpperCAmelCase__):
     lowerCamelCase__:   Union[str, Any]    						=len(UpperCAmelCase__)
   else:
     raise ValueError(F"""`prompt` has to be of type `str` or `list` but is {type(UpperCAmelCase__)}""")

   lowerCamelCase__:   List[Any]    						=self._execution_device

   lowerCamelCase__:   str    						=batch_size * num_images_per_prompt
   lowerCamelCase__:   Union[str, Any]    						=guidance_scale > 1.0

   lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__:   Tuple    						=self._encode_prompt(
       UpperCAmelCase__ ,				UpperCAmelCase__ ,				UpperCAmelCase__ ,				UpperCAmelCase__ ,				UpperCAmelCase__)

   if isinstance(UpperCAmelCase__ ,				UpperCAmelCase__):
     lowerCamelCase__:   List[str]    						=torch.cat(UpperCAmelCase__ ,				dim=0)
   if isinstance(UpperCAmelCase__ ,				UpperCAmelCase__):
     lowerCamelCase__:   str    						=torch.cat(UpperCAmelCase__ ,				dim=0)

   if do_classifier_free_guidance:
     lowerCamelCase__:   int    						=image_embeds.repeat_interleave(UpperCAmelCase__ ,				dim=0)
     lowerCamelCase__:   str    						=negative_image_embeds.repeat_interleave(UpperCAmelCase__ ,				dim=0)

   lowerCamelCase__:   Optional[Any]    						=torch.cat([negative_image_embeds, image_embeds] ,				dim=0).to(
       dtype=prompt_embeds.dtype ,				device=UpperCAmelCase__)

   self.scheduler.set_timesteps(UpperCAmelCase__ ,				device=UpperCAmelCase__)
   lowerCamelCase__:   str    						=self.scheduler.timesteps

   lowerCamelCase__:   Union[str, Any]    						=self.unet.config.in_channels

   lowerCamelCase__						,      lowerCamelCase__:   Union[str, Any]    						=get_new_h_w(UpperCAmelCase__ ,				UpperCAmelCase__ ,				self.movq_scale_factor)

   # create initial latent
   lowerCamelCase__:   Tuple    						=self.prepare_latents(
       (batch_size, num_channels_latents, height, width) ,				text_encoder_hidden_states.dtype ,				UpperCAmelCase__ ,				UpperCAmelCase__ ,				UpperCAmelCase__ ,				self.scheduler ,				)

   for i, t in enumerate(self.progress_bar(UpperCAmelCase__)):
     # expand the latents if we are doing classifier free guidance
     lowerCamelCase__:   Any    						=torch.cat([latents] * 2) if do_classifier_free_guidance else latents

     lowerCamelCase__:   Any    						={"text_embeds": prompt_embeds, "image_embeds": image_embeds}
     lowerCamelCase__:   Dict    						=self.unet(
         sample=UpperCAmelCase__ ,				timestep=UpperCAmelCase__ ,				encoder_hidden_states=UpperCAmelCase__ ,				added_cond_kwargs=UpperCAmelCase__ ,				return_dict=UpperCAmelCase__ ,				)[0]

     if do_classifier_free_guidance:
       lowerCamelCase__						,      lowerCamelCase__:   str    						=noise_pred.split(latents.shape[1] ,				dim=1)
       lowerCamelCase__						,      lowerCamelCase__:   Union[str, Any]    						=noise_pred.chunk(2)
       lowerCamelCase__						,      lowerCamelCase__:   List[Any]    						=variance_pred.chunk(2)
       lowerCamelCase__:   str    						=noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
       lowerCamelCase__:   Dict    						=torch.cat([noise_pred, variance_pred_text] ,				dim=1)

     if not (
         hasattr(self.scheduler.config ,				"variance_type")
         and self.scheduler.config.variance_type in ["learned", "learned_range"]
     ):
       lowerCamelCase__						,      lowerCamelCase__:   Union[str, Any]    						=noise_pred.split(latents.shape[1] ,				dim=1)

     # compute the previous noisy sample x_t -> x_t-1
     lowerCamelCase__:   Dict    						=self.scheduler.step(
         UpperCAmelCase__ ,				UpperCAmelCase__ ,				UpperCAmelCase__ ,				generator=UpperCAmelCase__ ,				).prev_sample
   # post-processing
   lowerCamelCase__:   Any    						=self.movq.decode(UpperCAmelCase__ ,				force_not_quantize=UpperCAmelCase__)["sample"]

   if output_type not in ["pt", "np", "pil"]:
     raise ValueError(F"""Only the output types `pt`, `pil` and `np` are supported not output_type={output_type}""")

   if output_type in ["np", "pil"]:
     lowerCamelCase__:   Optional[int]    						=image * 0.5 + 0.5
     lowerCamelCase__:   Union[str, Any]    						=image.clamp(0 ,				1)
     lowerCamelCase__:   str    						=image.cpu().permute(0 ,				2 ,				3 ,				1).float().numpy()

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

   if not return_dict:
     return (image,)

   return ImagePipelineOutput(images=UpperCAmelCase__)








from typing import Dict, List, Optional, Union

import numpy as np

from transformers.utils import is_vision_available
from transformers.utils.generic import TensorType

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import (
    center_crop,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    is_valid_image,
    to_numpy_array,
    valid_images,
)
from ...utils import logging


if is_vision_available():
     import PIL

__A             =       logging.get_logger(__name__)





def   lowerCAmelCase_				( __a	)  ->	List[List[ImageInput]]:







  """simple docstring"""





  if isinstance(__a			,					(list, tuple)	) and isinstance(videos[0]			,					(list, tuple)	) and is_valid_image(videos[0][0]	):
    return videos

  elif isinstance(__a			,					(list, tuple)	) and is_valid_image(videos[0]	):
    return [videos]

  elif is_valid_image(__a	):
    return [[videos]]

  raise ValueError(F"""Could not make batched video from {videos}"""	)






class      _SCREAMING_SNAKE_CASE		(			__SCREAMING_SNAKE_CASE					):







 '''simple docstring'''
 lowercase_	=    ["pixel_values"]

 def __init__(self       :     Tuple ,				UpperCAmelCase_       :     bool = True ,				UpperCAmelCase_       :     Dict[str, int] = None ,				UpperCAmelCase_       :     PILImageResampling = PILImageResampling.BILINEAR ,				UpperCAmelCase_       :     bool = True ,				UpperCAmelCase_       :     Dict[str, int] = None ,				UpperCAmelCase_       :     bool = True ,				UpperCAmelCase_       :     Union[int, float] = 1 / 255 ,				UpperCAmelCase_       :     bool = True ,				UpperCAmelCase_       :     bool = True ,				UpperCAmelCase_       :     Optional[Union[float, List[float]]] = None ,				UpperCAmelCase_       :     Optional[Union[float, List[float]]] = None ,				**UpperCAmelCase_       :     List[Any] ,				) ->None:



   '''simple docstring'''



   super().__init__(**UpperCAmelCase_)
   lowerCamelCase__:   Optional[int]    						=size if size is not None else {"shortest_edge": 256}
   lowerCamelCase__:   Any    						=get_size_dict(UpperCAmelCase_ ,				default_to_square=UpperCAmelCase_)
   lowerCamelCase__:   Tuple    						=crop_size if crop_size is not None else {"height": 224, "width": 224}
   lowerCamelCase__:   Any    						=get_size_dict(UpperCAmelCase_ ,				param_name="crop_size")

   lowerCamelCase__:   str    						=do_resize
   lowerCamelCase__:   Any    						=size
   lowerCamelCase__:   Any    						=do_center_crop
   lowerCamelCase__:   int    						=crop_size
   lowerCamelCase__:   int    						=resample
   lowerCamelCase__:   Optional[int]    						=do_rescale
   lowerCamelCase__:   int    						=rescale_factor
   lowerCamelCase__:   Dict    						=offset
   lowerCamelCase__:   List[Any]    						=do_normalize
   lowerCamelCase__:   Optional[int]    						=image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
   lowerCamelCase__:   List[str]    						=image_std if image_std is not None else IMAGENET_STANDARD_STD

 def 							SCREAMING_SNAKE_CASE_      (self       :     Union[str, Any] ,				UpperCAmelCase_       :     np.ndarray ,				UpperCAmelCase_       :     Dict[str, int] ,				UpperCAmelCase_       :     PILImageResampling = PILImageResampling.BILINEAR ,				UpperCAmelCase_       :     Optional[Union[str, ChannelDimension]] = None ,				**UpperCAmelCase_       :     List[Any] ,				) ->np.ndarray:



   '''simple docstring'''



   lowerCamelCase__:   Dict    						=get_size_dict(UpperCAmelCase_ ,				default_to_square=UpperCAmelCase_)
   if "shortest_edge" in size:
     lowerCamelCase__:   Union[str, Any]    						=get_resize_output_image_size(UpperCAmelCase_ ,				size["shortest_edge"] ,				default_to_square=UpperCAmelCase_)
   elif "height" in size and "width" in size:
     lowerCamelCase__:   int    						=(size["height"], size["width"])
   else:
     raise ValueError(F"""Size must have 'height' and 'width' or 'shortest_edge' as keys. Got {size.keys()}""")
   return resize(UpperCAmelCase_ ,				size=UpperCAmelCase_ ,				resample=UpperCAmelCase_ ,				data_format=UpperCAmelCase_ ,				**UpperCAmelCase_)

 def 							SCREAMING_SNAKE_CASE_      (self       :     str ,				UpperCAmelCase_       :     np.ndarray ,				UpperCAmelCase_       :     Dict[str, int] ,				UpperCAmelCase_       :     Optional[Union[str, ChannelDimension]] = None ,				**UpperCAmelCase_       :     List[Any] ,				) ->np.ndarray:



   '''simple docstring'''



   lowerCamelCase__:   List[str]    						=get_size_dict(UpperCAmelCase_)
   if "height" not in size or "width" not in size:
     raise ValueError(F"""Size must have 'height' and 'width' as keys. Got {size.keys()}""")
   return center_crop(UpperCAmelCase_ ,				size=(size["height"], size["width"]) ,				data_format=UpperCAmelCase_ ,				**UpperCAmelCase_)

 def 							SCREAMING_SNAKE_CASE_      (self       :     Union[str, Any] ,				UpperCAmelCase_       :     np.ndarray ,				UpperCAmelCase_       :     Union[int, float] ,				UpperCAmelCase_       :     bool = True ,				UpperCAmelCase_       :     Optional[Union[str, ChannelDimension]] = None ,				**UpperCAmelCase_       :     Union[str, Any] ,				) ->Union[str, Any]:



   '''simple docstring'''



   lowerCamelCase__:   str    						=image.astype(np.floataa)
   if offset:
     lowerCamelCase__:   List[Any]    						=image - (scale / 2)
   return rescale(UpperCAmelCase_ ,				scale=UpperCAmelCase_ ,				data_format=UpperCAmelCase_ ,				**UpperCAmelCase_)

 def 							SCREAMING_SNAKE_CASE_      (self       :     Tuple ,				UpperCAmelCase_       :     np.ndarray ,				UpperCAmelCase_       :     Union[float, List[float]] ,				UpperCAmelCase_       :     Union[float, List[float]] ,				UpperCAmelCase_       :     Optional[Union[str, ChannelDimension]] = None ,				**UpperCAmelCase_       :     Union[str, Any] ,				) ->np.ndarray:



   '''simple docstring'''



   return normalize(UpperCAmelCase_ ,				mean=UpperCAmelCase_ ,				std=UpperCAmelCase_ ,				data_format=UpperCAmelCase_ ,				**UpperCAmelCase_)

 def 							SCREAMING_SNAKE_CASE_      (self       :     Optional[int] ,				UpperCAmelCase_       :     ImageInput ,				UpperCAmelCase_       :     bool = None ,				UpperCAmelCase_       :     Dict[str, int] = None ,				UpperCAmelCase_       :     PILImageResampling = None ,				UpperCAmelCase_       :     bool = None ,				UpperCAmelCase_       :     Dict[str, int] = None ,				UpperCAmelCase_       :     bool = None ,				UpperCAmelCase_       :     float = None ,				UpperCAmelCase_       :     bool = None ,				UpperCAmelCase_       :     bool = None ,				UpperCAmelCase_       :     Optional[Union[float, List[float]]] = None ,				UpperCAmelCase_       :     Optional[Union[float, List[float]]] = None ,				UpperCAmelCase_       :     Optional[ChannelDimension] = ChannelDimension.FIRST ,				) ->np.ndarray:



   '''simple docstring'''



   if do_resize and size is None or resample is None:
     raise ValueError("Size and resample must be specified if do_resize is True.")

   if do_center_crop and crop_size is None:
     raise ValueError("Crop size must be specified if do_center_crop is True.")

   if do_rescale and rescale_factor is None:
     raise ValueError("Rescale factor must be specified if do_rescale is True.")

   if do_normalize and (image_mean is None or image_std is None):
     raise ValueError("Image mean and std must be specified if do_normalize is True.")

   if offset and not do_rescale:
     raise ValueError("For offset, do_rescale must also be set to True.")

   # All transformations expect numpy arrays.
   lowerCamelCase__:   List[str]    						=to_numpy_array(UpperCAmelCase_)

   if do_resize:
     lowerCamelCase__:   Optional[Any]    						=self.resize(image=UpperCAmelCase_ ,				size=UpperCAmelCase_ ,				resample=UpperCAmelCase_)

   if do_center_crop:
     lowerCamelCase__:   str    						=self.center_crop(UpperCAmelCase_ ,				size=UpperCAmelCase_)

   if do_rescale:
     lowerCamelCase__:   List[str]    						=self.rescale(image=UpperCAmelCase_ ,				scale=UpperCAmelCase_ ,				offset=UpperCAmelCase_)

   if do_normalize:
     lowerCamelCase__:   Dict    						=self.normalize(image=UpperCAmelCase_ ,				mean=UpperCAmelCase_ ,				std=UpperCAmelCase_)

   lowerCamelCase__:   Any    						=to_channel_dimension_format(UpperCAmelCase_ ,				UpperCAmelCase_)
   return image







 def 							SCREAMING_SNAKE_CASE_      (self       :     Optional[Any] ,				UpperCAmelCase_       :     ImageInput ,				UpperCAmelCase_       :     bool = None ,				UpperCAmelCase_       :     Dict[str, int] = None ,				UpperCAmelCase_       :     PILImageResampling = None ,				UpperCAmelCase_       :     bool = None ,				UpperCAmelCase_       :     Dict[str, int] = None ,				UpperCAmelCase_       :     bool = None ,				UpperCAmelCase_       :     float = None ,				UpperCAmelCase_       :     bool = None ,				UpperCAmelCase_       :     bool = None ,				UpperCAmelCase_       :     Optional[Union[float, List[float]]] = None ,				UpperCAmelCase_       :     Optional[Union[float, List[float]]] = None ,				UpperCAmelCase_       :     Optional[Union[str, TensorType]] = None ,				UpperCAmelCase_       :     ChannelDimension = ChannelDimension.FIRST ,				**UpperCAmelCase_       :     List[str] ,				) ->PIL.Image.Image:



   '''simple docstring'''



   lowerCamelCase__:   Union[str, Any]    						=do_resize if do_resize is not None else self.do_resize
   lowerCamelCase__:   Optional[int]    						=resample if resample is not None else self.resample
   lowerCamelCase__:   Optional[int]    						=do_center_crop if do_center_crop is not None else self.do_center_crop
   lowerCamelCase__:   List[str]    						=do_rescale if do_rescale is not None else self.do_rescale
   lowerCamelCase__:   Dict    						=rescale_factor if rescale_factor is not None else self.rescale_factor
   lowerCamelCase__:   Union[str, Any]    						=offset if offset is not None else self.offset
   lowerCamelCase__:   Dict    						=do_normalize if do_normalize is not None else self.do_normalize
   lowerCamelCase__:   Optional[int]    						=image_mean if image_mean is not None else self.image_mean
   lowerCamelCase__:   List[Any]    						=image_std if image_std is not None else self.image_std

   lowerCamelCase__:   List[Any]    						=size if size is not None else self.size
   lowerCamelCase__:   Optional[int]    						=get_size_dict(UpperCAmelCase_ ,				default_to_square=UpperCAmelCase_)
   lowerCamelCase__:   str    						=crop_size if crop_size is not None else self.crop_size
   lowerCamelCase__:   Optional[Any]    						=get_size_dict(UpperCAmelCase_ ,				param_name="crop_size")

   if not valid_images(UpperCAmelCase_):
     raise ValueError(
         "Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, "
         "torch.Tensor, tf.Tensor or jax.ndarray.")

   lowerCamelCase__:   str    						=make_batched(UpperCAmelCase_)

   lowerCamelCase__:   Any    						=[
       [
           self._preprocess_image(
               image=UpperCAmelCase_ ,				do_resize=UpperCAmelCase_ ,				size=UpperCAmelCase_ ,				resample=UpperCAmelCase_ ,				do_center_crop=UpperCAmelCase_ ,				crop_size=UpperCAmelCase_ ,				do_rescale=UpperCAmelCase_ ,				rescale_factor=UpperCAmelCase_ ,				offset=UpperCAmelCase_ ,				do_normalize=UpperCAmelCase_ ,				image_mean=UpperCAmelCase_ ,				image_std=UpperCAmelCase_ ,				data_format=UpperCAmelCase_ ,				)
           for img in video
       ]
       for video in videos
   ]

   lowerCamelCase__:   int    						={"pixel_values": videos}
   return BatchFeature(data=UpperCAmelCase_ ,				tensor_type=UpperCAmelCase_)

