

'''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 						UpperCamelCase	(			a		,						a		,						a			)				->     Tuple:






					'''simple docstring'''



					if isinstance(A__		,						torch.Tensor			):
										return image
					elif isinstance(A__		,						PIL.Image.Image			):
										__magic_name__						       =    [image]

					if isinstance(image[0]		,						PIL.Image.Image			):
										__magic_name__						       =    [np.array(i.resize((w, h)		,						resample=PIL_INTERPOLATION['''lanczos''']			)			)[None, :] for i in image]
										__magic_name__						       =    np.concatenate(A__		,						axis=0			)
										__magic_name__						       =    np.array(A__			).astype(np.floataa			) / 2_55.0
										__magic_name__						       =    image.transpose(0		,						3		,						1		,						2			)
										__magic_name__						       =    2.0 * image - 1.0
										__magic_name__						       =    torch.from_numpy(A__			)
					elif isinstance(image[0]		,						torch.Tensor			):
										__magic_name__						       =    torch.cat(A__		,						dim=0			)
					return image




def 						UpperCamelCase	(			a		,						a		,						a		,						a=0.99_95			)				->     List[Any]:






					'''simple docstring'''



					if not isinstance(A__		,						np.ndarray			):
										__magic_name__						       =    True
										__magic_name__						       =    va.device
										__magic_name__						       =    va.cpu().numpy()
										__magic_name__						       =    va.cpu().numpy()

					__magic_name__						       =    np.sum(va * va / (np.linalg.norm(A__			) * np.linalg.norm(A__			))			)
					if np.abs(A__			) > DOT_THRESHOLD:
										__magic_name__						       =    (1 - t) * va + t * va
					else:
										__magic_name__						       =    np.arccos(A__			)
										__magic_name__						       =    np.sin(A__			)
										__magic_name__						       =    theta_a * t
										__magic_name__						       =    np.sin(A__			)
										__magic_name__						       =    np.sin(theta_a - theta_t			) / sin_theta_a
										__magic_name__						       =    sin_theta_t / sin_theta_a
										__magic_name__						       =    sa * va + sa * va

					if inputs_are_torch:
										__magic_name__						       =    torch.from_numpy(A__			).to(A__			)

					return va




def 						UpperCamelCase	(			a		,						a			)				->     List[str]:






					'''simple docstring'''
					__magic_name__						       =    F.normalize(A__		,						dim=-1			)
					__magic_name__						       =    F.normalize(A__		,						dim=-1			)
					return (x - y).norm(dim=-1			).div(2			).arcsin().pow(2			).mul(2			)




def 						UpperCamelCase	(			a		,						a			)				->     Optional[Any]:






					'''simple docstring'''



					for param in model.parameters():
										__magic_name__						       =    value






class   _SCREAMING_SNAKE_CASE   (   __lowerCamelCase     ):



	def __init__(   self		:   Tuple			,				a__		:   AutoencoderKL			,				a__		:   CLIPTextModel			,				a__		:   CLIPModel			,				a__		:   CLIPTokenizer			,				a__		:   UNetaDConditionModel			,				a__		:   Union[PNDMScheduler, LMSDiscreteScheduler, DDIMScheduler, DPMSolverMultistepScheduler]			,				a__		:   CLIPFeatureExtractor			,				a__		:   Optional[int]=None			,				a__		:   Union[str, Any]=None			,				a__		:   List[str]=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_			,				)
						__magic_name__						       =    (
						    feature_extractor.size
						    if isinstance(feature_extractor.size			,				UpperCamelCase_      )
						    else feature_extractor.size['''shortest_edge''']
						)
						__magic_name__						       =    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		:   List[Any]			,				a__		:   Optional[Union[str, int]] = "auto"      ):
						if slice_size == "auto":
											# half the attention head size is usually a good trade-off between
											# speed and memory
											__magic_name__						       =    self.unet.config.attention_head_dim // 2
						self.unet.set_attention_slice(UpperCamelCase_      )



	def 			snake_case__  (   self		:   str      ):
						self.enable_attention_slicing(UpperCamelCase_      )



	def 			snake_case__  (   self		:   Tuple      ):
						set_requires_grad(self.vae			,				UpperCamelCase_      )



	def 			snake_case__  (   self		:   Tuple      ):
						set_requires_grad(self.vae			,				UpperCamelCase_      )



	def 			snake_case__  (   self		:   List[Any]      ):
						set_requires_grad(self.unet			,				UpperCamelCase_      )



	def 			snake_case__  (   self		:   List[Any]      ):
						set_requires_grad(self.unet			,				UpperCamelCase_      )



	def 			snake_case__  (   self		:   List[str]			,				a__		:   Dict			,				a__		:   List[str]			,				a__		:   str      ):
						# get the original timestep using init_timestep
						__magic_name__						       =    min(int(num_inference_steps * strength      )			,				UpperCamelCase_      )

						__magic_name__						       =    max(num_inference_steps - init_timestep			,				0      )
						__magic_name__						       =    self.scheduler.timesteps[t_start:]

						return timesteps, num_inference_steps - t_start



	def 			snake_case__  (   self		:   str			,				a__		:   str			,				a__		:   int			,				a__		:   Union[str, Any]			,				a__		:   Optional[int]			,				a__		:   Dict			,				a__		:   Optional[Any]=None      ):
						if not isinstance(UpperCamelCase_			,				torch.Tensor      ):
											raise ValueError(F'''`image` has to be of type `torch.Tensor` but is {type(UpperCamelCase_      )}'''      )

						__magic_name__						       =    image.to(device=UpperCamelCase_			,				dtype=UpperCamelCase_      )

						if isinstance(UpperCamelCase_			,				UpperCamelCase_      ):
											__magic_name__						       =    [
											    self.vae.encode(image[i : i + 1]      ).latent_dist.sample(generator[i]      ) for i in range(UpperCamelCase_      )
											]
											__magic_name__						       =    torch.cat(UpperCamelCase_			,				dim=0      )
						else:
											__magic_name__						       =    self.vae.encode(UpperCamelCase_      ).latent_dist.sample(UpperCamelCase_      )

						# Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
						__magic_name__						       =    0.18_215 * init_latents
						__magic_name__						       =    init_latents.repeat_interleave(UpperCamelCase_			,				dim=0      )

						__magic_name__						       =    randn_tensor(init_latents.shape			,				generator=UpperCamelCase_			,				device=UpperCamelCase_			,				dtype=UpperCamelCase_      )

						# get latents
						__magic_name__						       =    self.scheduler.add_noise(UpperCamelCase_			,				UpperCamelCase_			,				UpperCamelCase_      )
						__magic_name__						       =    init_latents

						return latents



	def 			snake_case__  (   self		:   int			,				a__		:   Any      ):
						__magic_name__						       =    self.coca_transform(UpperCamelCase_      ).unsqueeze(0      )
						with torch.no_grad(), torch.cuda.amp.autocast():
											__magic_name__						       =    self.coca_model.generate(transformed_image.to(device=self.device			,				dtype=self.coca_model.dtype      )      )
						__magic_name__						       =    self.coca_tokenizer.decode(generated[0].cpu().numpy()      )
						return generated.split('''<end_of_text>'''      )[0].replace('''<start_of_text>'''			,				''''''      ).rstrip(''' .,'''      )



	def 			snake_case__  (   self		:   List[str]			,				a__		:   Optional[Any]			,				a__		:   Optional[int]      ):
						__magic_name__						       =    self.feature_extractor.preprocess(UpperCamelCase_      )
						__magic_name__						       =    torch.from_numpy(clip_image_input['''pixel_values'''][0]      ).unsqueeze(0      ).to(self.device      ).half()
						__magic_name__						       =    self.clip_model.get_image_features(UpperCamelCase_      )
						__magic_name__						       =    image_embeddings_clip / image_embeddings_clip.norm(p=2			,				dim=-1			,				keepdim=UpperCamelCase_      )
						__magic_name__						       =    image_embeddings_clip.repeat_interleave(UpperCamelCase_			,				dim=0      )
						return image_embeddings_clip



	@torch.enable_grad()
	def 			snake_case__  (   self		:   int			,				a__		:   Union[str, Any]			,				a__		:   List[Any]			,				a__		:   int			,				a__		:   List[Any]			,				a__		:   Optional[Any]			,				a__		:   Any			,				a__		:   List[Any]			,				):
						__magic_name__						       =    latents.detach().requires_grad_()

						__magic_name__						       =    self.scheduler.scale_model_input(UpperCamelCase_			,				UpperCamelCase_      )

						# predict the noise residual
						__magic_name__						       =    self.unet(UpperCamelCase_			,				UpperCamelCase_			,				encoder_hidden_states=UpperCamelCase_      ).sample

						if isinstance(self.scheduler			,				(PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)      ):
											__magic_name__						       =    self.scheduler.alphas_cumprod[timestep]
											__magic_name__						       =    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
											__magic_name__						       =    (latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

											__magic_name__						       =    torch.sqrt(UpperCamelCase_      )
											__magic_name__						       =    pred_original_sample * (fac) + latents * (1 - fac)
						elif isinstance(self.scheduler			,				UpperCamelCase_      ):
											__magic_name__						       =    self.scheduler.sigmas[index]
											__magic_name__						       =    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
						__magic_name__						       =    1 / 0.18_215 * sample
						__magic_name__						       =    self.vae.decode(UpperCamelCase_      ).sample
						__magic_name__						       =    (image / 2 + 0.5).clamp(0			,				1      )

						__magic_name__						       =    transforms.Resize(self.feature_extractor_size      )(UpperCamelCase_      )
						__magic_name__						       =    self.normalize(UpperCamelCase_      ).to(latents.dtype      )

						__magic_name__						       =    self.clip_model.get_image_features(UpperCamelCase_      )
						__magic_name__						       =    image_embeddings_clip / image_embeddings_clip.norm(p=2			,				dim=-1			,				keepdim=UpperCamelCase_      )

						__magic_name__						       =    spherical_dist_loss(UpperCamelCase_			,				UpperCamelCase_      ).mean() * clip_guidance_scale

						__magic_name__						       =    -torch.autograd.grad(UpperCamelCase_			,				UpperCamelCase_      )[0]

						if isinstance(self.scheduler			,				UpperCamelCase_      ):
											__magic_name__						       =    latents.detach() + grads * (sigma**2)
											__magic_name__						       =    noise_pred_original
						else:
											__magic_name__						       =    noise_pred_original - torch.sqrt(UpperCamelCase_      ) * grads
						return noise_pred, latents



	@torch.no_grad()
	def __call__(   self		:   Union[str, Any]			,				a__		:   Union[torch.FloatTensor, PIL.Image.Image]			,				a__		:   Union[torch.FloatTensor, PIL.Image.Image]			,				a__		:   Optional[str] = None			,				a__		:   Optional[str] = None			,				a__		:   Optional[int] = 512			,				a__		:   Optional[int] = 512			,				a__		:   float = 0.6			,				a__		:   Optional[int] = 50			,				a__		:   Optional[float] = 7.5			,				a__		:   Optional[int] = 1			,				a__		:   float = 0.0			,				a__		:   Optional[float] = 100			,				a__		:   Optional[torch.Generator] = None			,				a__		:   Optional[str] = "pil"			,				a__		:   bool = True			,				a__		:   float = 0.8			,				a__		:   float = 0.1			,				a__		:   float = 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:
											__magic_name__						       =    [generator] + [None] * (batch_size - 1)

						__magic_name__						       =    [
						    ('''model''', self.coca_model is None),
						    ('''tokenizer''', self.coca_tokenizer is None),
						    ('''transform''', self.coca_transform is None),
						]
						__magic_name__						       =    [x[0] for x in coca_is_none if x[1]]
						__magic_name__						       =    ''', '''.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.'''      )
											__magic_name__						       =    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.'''      )
											__magic_name__						       =    self.get_image_description(UpperCamelCase_      )

						# get prompt text embeddings for content and style
						__magic_name__						       =    self.tokenizer(
						    UpperCamelCase_			,				padding='''max_length'''			,				max_length=self.tokenizer.model_max_length			,				truncation=UpperCamelCase_			,				return_tensors='''pt'''			,				)
						__magic_name__						       =    self.text_encoder(content_text_input.input_ids.to(self.device      )      )[0]

						__magic_name__						       =    self.tokenizer(
						    UpperCamelCase_			,				padding='''max_length'''			,				max_length=self.tokenizer.model_max_length			,				truncation=UpperCamelCase_			,				return_tensors='''pt'''			,				)
						__magic_name__						       =    self.text_encoder(style_text_input.input_ids.to(self.device      )      )[0]

						__magic_name__						       =    slerp(UpperCamelCase_			,				UpperCamelCase_			,				UpperCamelCase_      )

						# duplicate text embeddings for each generation per prompt
						__magic_name__						       =    text_embeddings.repeat_interleave(UpperCamelCase_			,				dim=0      )

						# set timesteps
						__magic_name__						       =    '''offset''' in set(inspect.signature(self.scheduler.set_timesteps      ).parameters.keys()      )
						__magic_name__						       =    {}
						if accepts_offset:
											__magic_name__						       =    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      )

						__magic_name__       ,  __magic_name__						       =    self.get_timesteps(UpperCamelCase_			,				UpperCamelCase_			,				self.device      )
						__magic_name__						       =    timesteps[:1].repeat(UpperCamelCase_      )

						# Preprocess image
						__magic_name__						       =    preprocess(UpperCamelCase_			,				UpperCamelCase_			,				UpperCamelCase_      )
						__magic_name__						       =    self.prepare_latents(
						    UpperCamelCase_			,				UpperCamelCase_			,				UpperCamelCase_			,				text_embeddings.dtype			,				self.device			,				UpperCamelCase_      )

						__magic_name__						       =    preprocess(UpperCamelCase_			,				UpperCamelCase_			,				UpperCamelCase_      )
						__magic_name__						       =    self.prepare_latents(
						    UpperCamelCase_			,				UpperCamelCase_			,				UpperCamelCase_			,				text_embeddings.dtype			,				self.device			,				UpperCamelCase_      )

						__magic_name__						       =    slerp(UpperCamelCase_			,				UpperCamelCase_			,				UpperCamelCase_      )

						if clip_guidance_scale > 0:
											__magic_name__						       =    self.get_clip_image_embeddings(UpperCamelCase_			,				UpperCamelCase_      )
											__magic_name__						       =    self.get_clip_image_embeddings(UpperCamelCase_			,				UpperCamelCase_      )
											__magic_name__						       =    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.
						__magic_name__						       =    guidance_scale > 1.0
						# get unconditional embeddings for classifier free guidance
						if do_classifier_free_guidance:
											__magic_name__						       =    content_text_input.input_ids.shape[-1]
											__magic_name__						       =    self.tokenizer(['''''']			,				padding='''max_length'''			,				max_length=UpperCamelCase_			,				return_tensors='''pt'''      )
											__magic_name__						       =    self.text_encoder(uncond_input.input_ids.to(self.device      )      )[0]
											# duplicate unconditional embeddings for each generation per prompt
											__magic_name__						       =    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
											__magic_name__						       =    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`.
						__magic_name__						       =    (batch_size, self.unet.config.in_channels, height // 8, width // 8)
						__magic_name__						       =    text_embeddings.dtype
						if latents is None:
											if self.device.type == "mps":
																# randn does not work reproducibly on mps
																__magic_name__						       =    torch.randn(UpperCamelCase_			,				generator=UpperCamelCase_			,				device='''cpu'''			,				dtype=UpperCamelCase_      ).to(
																    self.device      )
											else:
																__magic_name__						       =    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}'''      )
											__magic_name__						       =    latents.to(self.device      )

						# scale the initial noise by the standard deviation required by the scheduler
						__magic_name__						       =    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]
						__magic_name__						       =    '''eta''' in set(inspect.signature(self.scheduler.step      ).parameters.keys()      )
						__magic_name__						       =    {}
						if accepts_eta:
											__magic_name__						       =    eta

						# check if the scheduler accepts generator
						__magic_name__						       =    '''generator''' in set(inspect.signature(self.scheduler.step      ).parameters.keys()      )
						if accepts_generator:
											__magic_name__						       =    generator

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

																# predict the noise residual
																__magic_name__						       =    self.unet(UpperCamelCase_			,				UpperCamelCase_			,				encoder_hidden_states=UpperCamelCase_      ).sample

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

																# perform clip guidance
																if clip_guidance_scale > 0:
																					__magic_name__						       =    (
																					    text_embeddings.chunk(2      )[1] if do_classifier_free_guidance else text_embeddings
																					)
																					__magic_name__       ,  __magic_name__						       =    self.cond_fn(
																					    UpperCamelCase_			,				UpperCamelCase_			,				UpperCamelCase_			,				UpperCamelCase_			,				UpperCamelCase_			,				UpperCamelCase_			,				UpperCamelCase_			,				)

																# compute the previous noisy sample x_t -> x_t-1
																__magic_name__						       =    self.scheduler.step(UpperCamelCase_			,				UpperCamelCase_			,				UpperCamelCase_			,				**UpperCamelCase_      ).prev_sample

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
						__magic_name__						       =    1 / 0.18_215 * latents
						__magic_name__						       =    self.vae.decode(UpperCamelCase_      ).sample

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

						if output_type == "pil":
											__magic_name__						       =    self.numpy_to_pil(UpperCamelCase_      )

						if not return_dict:
											return (image, None)

						return StableDiffusionPipelineOutput(images=UpperCamelCase_			,				nsfw_content_detected=UpperCamelCase_      )









'''simple docstring'''




import importlib
import inspect
import json
import os
import re
import shutil
import sys
from pathlib import Path
from typing import Dict, Optional, Union
from urllib import request

from huggingface_hub import HfFolder, cached_download, hf_hub_download, model_info
from packaging import version

from .. import __version__
from . import DIFFUSERS_DYNAMIC_MODULE_NAME, HF_MODULES_CACHE, logging


_lowerCAmelCase         =						(
    "https://raw.githubusercontent.com/huggingface/diffusers/{revision}/examples/community/{pipeline}.py"
)


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




def 						UpperCamelCase	(			)				->     Optional[Any]:






     '''simple docstring'''
     __magic_name__						       =    '''https://pypi.org/pypi/diffusers/json'''
     __magic_name__						       =    json.loads(request.urlopen(a			).read()			)['''releases'''].keys()
     return sorted(a		,						key=lambda a			: version.Version(a			)			)




def 						UpperCamelCase	(			)				->     Any:






     '''simple docstring'''
     # This function has already been executed if HF_MODULES_CACHE already is in the Python path.
     if HF_MODULES_CACHE in sys.path:
          return

     sys.path.append(a			)
     os.makedirs(a		,						exist_ok=a			)
     __magic_name__						       =    Path(a			) / '''__init__.py'''
     if not init_path.exists():
          init_path.touch()




def 						UpperCamelCase	(			a			)				->     Tuple:






     '''simple docstring'''
     init_hf_modules()
     __magic_name__						       =    Path(a			) / name
     # If the parent module does not exist yet, recursively create it.
     if not dynamic_module_path.parent.exists():
          create_dynamic_module(dynamic_module_path.parent			)
     os.makedirs(a		,						exist_ok=a			)
     __magic_name__						       =    dynamic_module_path / '''__init__.py'''
     if not init_path.exists():
          init_path.touch()




def 						UpperCamelCase	(			a			)				->     List[str]:






     '''simple docstring'''
     with open(a		,						'''r'''		,						encoding='''utf-8'''			) as f:
          __magic_name__						       =    f.read()

     # Imports of the form `import .xxx`
     __magic_name__						       =    re.findall('''^\s*import\s+\.(\S+)\s*$'''		,						a		,						flags=re.MULTILINE			)
     # Imports of the form `from .xxx import yyy`
     relative_imports += re.findall('''^\s*from\s+\.(\S+)\s+import'''		,						a		,						flags=re.MULTILINE			)
     # Unique-ify
     return list(set(a			)			)




def 						UpperCamelCase	(			a			)				->     str:






     '''simple docstring'''
     __magic_name__						       =    False
     __magic_name__						       =    [module_file]
     __magic_name__						       =    []

     # Let's recurse through all relative imports
     while not no_change:
          __magic_name__						       =    []
          for f in files_to_check:
               new_imports.extend(get_relative_imports(a			)			)

          __magic_name__						       =    Path(a			).parent
          __magic_name__						       =    [str(module_path / m			) for m in new_imports]
          __magic_name__						       =    [f for f in new_import_files if f not in all_relative_imports]
          __magic_name__						       =    [F'''{f}.py''' for f in new_import_files]

          __magic_name__						       =    len(a			) == 0
          all_relative_imports.extend(a			)

     return all_relative_imports




def 						UpperCamelCase	(			a			)				->     List[str]:






     '''simple docstring'''
     with open(a		,						'''r'''		,						encoding='''utf-8'''			) as f:
          __magic_name__						       =    f.read()

     # Imports of the form `import xxx`
     __magic_name__						       =    re.findall('''^\s*import\s+(\S+)\s*$'''		,						a		,						flags=re.MULTILINE			)
     # Imports of the form `from xxx import yyy`
     imports += re.findall('''^\s*from\s+(\S+)\s+import'''		,						a		,						flags=re.MULTILINE			)
     # Only keep the top-level module
     __magic_name__						       =    [imp.split('''.'''			)[0] for imp in imports if not imp.startswith('''.'''			)]

     # Unique-ify and test we got them all
     __magic_name__						       =    list(set(a			)			)
     __magic_name__						       =    []
     for imp in imports:
          try:
               importlib.import_module(a			)
          except ImportError:
               missing_packages.append(a			)

     if len(a			) > 0:
          raise ImportError(
              '''This modeling file requires the following packages that were not found in your environment: '''
              F'''{", ".join(a			)}. Run `pip install {" ".join(a			)}`'''			)

     return get_relative_imports(a			)




def 						UpperCamelCase	(			a		,						a			)				->     Optional[int]:






     '''simple docstring'''
     __magic_name__						       =    module_path.replace(os.path.sep		,						'''.'''			)
     __magic_name__						       =    importlib.import_module(a			)

     if class_name is None:
          return find_pipeline_class(a			)
     return getattr(a		,						a			)




def 						UpperCamelCase	(			a			)				->     Tuple:






     '''simple docstring'''
     from ..pipelines import DiffusionPipeline

     __magic_name__						       =    dict(inspect.getmembers(a		,						inspect.isclass			)			)

     __magic_name__						       =    None
     for cls_name, cls in cls_members.items():
          if (
              cls_name != DiffusionPipeline.__name__
              and issubclass(cls		,						a			)
              and cls.__module__.split('''.'''			)[0] != "diffusers"
          ):
               if pipeline_class is not None:
                    raise ValueError(
                        F'''Multiple classes that inherit from {DiffusionPipeline.__name__} have been found:'''
                        F''' {pipeline_class.__name__}, and {cls_name}. Please make sure to define only one in'''
                        F''' {loaded_module}.'''			)
               __magic_name__						       =    cls

     return pipeline_class




def 						UpperCamelCase	(			a		,						a		,						a = None		,						a = False		,						a = False		,						a = None		,						a = None		,						a = None		,						a = False		,						)				->     List[Any]:






     '''simple docstring'''
     __magic_name__						       =    str(a			)

     __magic_name__						       =    os.path.join(a		,						a			)

     if os.path.isfile(a			):
          __magic_name__						       =    module_file_or_url
          __magic_name__						       =    '''local'''
     elif pretrained_model_name_or_path.count('''/'''			) == 0:
          __magic_name__						       =    get_diffusers_versions()
          # cut ".dev0"
          __magic_name__						       =    '''v''' + '''.'''.join(__version__.split('''.'''			)[:3]			)

          # retrieve github version that matches
          if revision is None:
               __magic_name__						       =    latest_version if latest_version[1:] in available_versions else '''main'''
               logger.info(F'''Defaulting to latest_version: {revision}.'''			)
          elif revision in available_versions:
               __magic_name__						       =    F'''v{revision}'''
          elif revision == "main":
               __magic_name__						       =    revision
          else:
               raise ValueError(
                   F'''`custom_revision`: {revision} does not exist. Please make sure to choose one of'''
                   F''' {", ".join(available_versions + ["main"]			)}.'''			)

          # community pipeline on GitHub
          __magic_name__						       =    COMMUNITY_PIPELINES_URL.format(revision=a		,						pipeline=a			)
          try:
               __magic_name__						       =    cached_download(
                   a		,						cache_dir=a		,						force_download=a		,						proxies=a		,						resume_download=a		,						local_files_only=a		,						use_auth_token=a		,						)
               __magic_name__						       =    '''git'''
               __magic_name__						       =    pretrained_model_name_or_path + '''.py'''
          except EnvironmentError:
               logger.error(F'''Could not locate the {module_file} inside {pretrained_model_name_or_path}.'''			)
               raise
     else:
          try:
               # Load from URL or cache if already cached
               __magic_name__						       =    hf_hub_download(
                   a		,						a		,						cache_dir=a		,						force_download=a		,						proxies=a		,						resume_download=a		,						local_files_only=a		,						use_auth_token=a		,						)
               __magic_name__						       =    os.path.join('''local'''		,						'''--'''.join(pretrained_model_name_or_path.split('''/'''			)			)			)
          except EnvironmentError:
               logger.error(F'''Could not locate the {module_file} inside {pretrained_model_name_or_path}.'''			)
               raise

    # Check we have all the requirements in our environment
     __magic_name__						       =    check_imports(a			)

     # Now we move the module inside our cached dynamic modules.
     __magic_name__						       =    DIFFUSERS_DYNAMIC_MODULE_NAME + os.path.sep + submodule
     create_dynamic_module(a			)
     __magic_name__						       =    Path(a			) / full_submodule
     if submodule == "local" or submodule == "git":
          # We always copy local files (we could hash the file to see if there was a change, and give them the name of
          # that hash, to only copy when there is a modification but it seems overkill for now).
          # The only reason we do the copy is to avoid putting too many folders in sys.path.
          shutil.copy(a		,						submodule_path / module_file			)
          for module_needed in modules_needed:
               __magic_name__						       =    F'''{module_needed}.py'''
               shutil.copy(os.path.join(a		,						a			)		,						submodule_path / module_needed			)
     else:
          # Get the commit hash
          # TODO: we will get this info in the etag soon, so retrieve it from there and not here.
          if isinstance(a		,						a			):
               __magic_name__						       =    use_auth_token
          elif use_auth_token is True:
               __magic_name__						       =    HfFolder.get_token()
          else:
               __magic_name__						       =    None

          __magic_name__						       =    model_info(a		,						revision=a		,						token=a			).sha

          # The module file will end up being placed in a subfolder with the git hash of the repo. This way we get the
          # benefit of versioning.
          __magic_name__						       =    submodule_path / commit_hash
          __magic_name__						       =    full_submodule + os.path.sep + commit_hash
          create_dynamic_module(a			)

          if not (submodule_path / module_file).exists():
               shutil.copy(a		,						submodule_path / module_file			)
          # Make sure we also have every file with relative
          for module_needed in modules_needed:
               if not (submodule_path / module_needed).exists():
                    get_cached_module_file(
                        a		,						F'''{module_needed}.py'''		,						cache_dir=a		,						force_download=a		,						resume_download=a		,						proxies=a		,						use_auth_token=a		,						revision=a		,						local_files_only=a		,						)
     return os.path.join(a		,						a			)




def 						UpperCamelCase	(			a		,						a		,						a = None		,						a = None		,						a = False		,						a = False		,						a = None		,						a = None		,						a = None		,						a = False		,						**a		,						)				->     List[Any]:






     '''simple docstring'''
     __magic_name__						       =    get_cached_module_file(
         a		,						a		,						cache_dir=a		,						force_download=a		,						resume_download=a		,						proxies=a		,						use_auth_token=a		,						revision=a		,						local_files_only=a		,						)
     return get_class_in_module(a		,						final_module.replace('''.py'''		,						''''''			)			)






