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  _lowerCamelCase					(   a_				: Optional[Any]						,			a_				: str						,			a_				: Optional[Any]):
			if isinstance(a_						,			torch.Tensor):
						return image
			elif isinstance(a_						,			PIL.Image.Image):
						lowerCamelCase						:List[Any]						  =     [image]

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





def  _lowerCamelCase					(   a_				: int						,			a_				: Dict						,			a_				: List[str]						,			a_				: Tuple=0.9_995):
			if not isinstance(a_						,			np.ndarray):
						lowerCamelCase						:List[str]						  =     True
						lowerCamelCase						:int						  =     va.device
						lowerCamelCase						:Union[str, Any]						  =     va.cpu().numpy()
						lowerCamelCase						:List[str]						  =     va.cpu().numpy()

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

			if inputs_are_torch:
						lowerCamelCase						:Tuple						  =     torch.from_numpy(a_).to(a_)

			return va





def  _lowerCamelCase					(   a_				: Optional[int]						,			a_				: Dict):
			lowerCamelCase						:Dict						  =     F.normalize(a_						,			dim=-1)
			lowerCamelCase						:int						  =     F.normalize(a_						,			dim=-1)
			return (x - y).norm(dim=-1).div(2).arcsin().pow(2).mul(2)





def  _lowerCamelCase					(   a_				: Any						,			a_				: List[Any]):
			for param in model.parameters():
						lowerCamelCase						:List[str]						  =     value


class   _lowerCAmelCase		(     __SCREAMING_SNAKE_CASE				):







							def __init__(				self     :    Union[str, Any]	,					__snake_case     :    AutoencoderKL	,					__snake_case     :    CLIPTextModel	,					__snake_case     :    CLIPModel	,					__snake_case     :    CLIPTokenizer	,					__snake_case     :    UNetaDConditionModel	,					__snake_case     :    Union[PNDMScheduler, LMSDiscreteScheduler, DDIMScheduler, DPMSolverMultistepScheduler]	,					__snake_case     :    CLIPFeatureExtractor	,					__snake_case     :    str=None	,					__snake_case     :    Tuple=None	,					__snake_case     :    Optional[int]=None	,					):
										super().__init__()
										self.register_modules(
										    vae=__snake_case	,					text_encoder=__snake_case	,					clip_model=__snake_case	,					tokenizer=__snake_case	,					unet=__snake_case	,					scheduler=__snake_case	,					feature_extractor=__snake_case	,					coca_model=__snake_case	,					coca_tokenizer=__snake_case	,					coca_transform=__snake_case	,					)
										lowerCamelCase						:Tuple						  =     (
										    feature_extractor.size
										    if isinstance(feature_extractor.size	,					__snake_case		)
										    else feature_extractor.size['''shortest_edge''']
										)
										lowerCamelCase						:List[Any]						  =     transforms.Normalize(mean=feature_extractor.image_mean	,					std=feature_extractor.image_std		)
										set_requires_grad(self.text_encoder	,					__snake_case		)
										set_requires_grad(self.clip_model	,					__snake_case		)







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







							def 				snake_case  (				self     :    Optional[int]		):
										self.enable_attention_slicing(__snake_case		)







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







							def 				snake_case  (				self     :    Tuple		):
										set_requires_grad(self.vae	,					__snake_case		)







							def 				snake_case  (				self     :    Any		):
										set_requires_grad(self.unet	,					__snake_case		)







							def 				snake_case  (				self     :    Union[str, Any]		):
										set_requires_grad(self.unet	,					__snake_case		)







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

										lowerCamelCase						:Optional[int]						  =     max(num_inference_steps - init_timestep	,					0		)
										lowerCamelCase						:List[str]						  =     self.scheduler.timesteps[t_start:]

										return timesteps, num_inference_steps - t_start







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

										lowerCamelCase						:Dict						  =     image.to(device=__snake_case	,					dtype=__snake_case		)

										if isinstance(__snake_case	,					__snake_case		):
													lowerCamelCase						:Any						  =     [
													    self.vae.encode(image[i : i + 1]		).latent_dist.sample(generator[i]		) for i in range(__snake_case		)
													]
													lowerCamelCase						:Union[str, Any]						  =     torch.cat(__snake_case	,					dim=0		)
										else:
													lowerCamelCase						:Tuple						  =     self.vae.encode(__snake_case		).latent_dist.sample(__snake_case		)

										# Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
										lowerCamelCase						:Dict						  =     0.1_8_2_1_5 * init_latents
										lowerCamelCase						:Union[str, Any]						  =     init_latents.repeat_interleave(__snake_case	,					dim=0		)

										lowerCamelCase						:Optional[int]						  =     randn_tensor(init_latents.shape	,					generator=__snake_case	,					device=__snake_case	,					dtype=__snake_case		)

										# get latents
										lowerCamelCase						:int						  =     self.scheduler.add_noise(__snake_case	,					__snake_case	,					__snake_case		)
										lowerCamelCase						:Optional[Any]						  =     init_latents

										return latents







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







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







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

										lowerCamelCase						:Union[str, Any]						  =     self.scheduler.scale_model_input(__snake_case	,					__snake_case		)

										# predict the noise residual
										lowerCamelCase						:Dict						  =     self.unet(__snake_case	,					__snake_case	,					encoder_hidden_states=__snake_case		).sample

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

													lowerCamelCase						:Any						  =     torch.sqrt(__snake_case		)
													lowerCamelCase						:Dict						  =     pred_original_sample * (fac) + latents * (1 - fac)
										elif isinstance(self.scheduler	,					__snake_case		):
													lowerCamelCase						:Optional[Any]						  =     self.scheduler.sigmas[index]
													lowerCamelCase						:Tuple						  =     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
										lowerCamelCase						:int						  =     1 / 0.1_8_2_1_5 * sample
										lowerCamelCase						:int						  =     self.vae.decode(__snake_case		).sample
										lowerCamelCase						:Union[str, Any]						  =     (image / 2 + 0.5).clamp(0	,					1		)

										lowerCamelCase						:int						  =     transforms.Resize(self.feature_extractor_size		)(__snake_case		)
										lowerCamelCase						:str						  =     self.normalize(__snake_case		).to(latents.dtype		)

										lowerCamelCase						:List[Any]						  =     self.clip_model.get_image_features(__snake_case		)
										lowerCamelCase						:Optional[Any]						  =     image_embeddings_clip / image_embeddings_clip.norm(p=2	,					dim=-1	,					keepdim=__snake_case		)

										lowerCamelCase						:str						  =     spherical_dist_loss(__snake_case	,					__snake_case		).mean() * clip_guidance_scale

										lowerCamelCase						:Tuple						  =     -torch.autograd.grad(__snake_case	,					__snake_case		)[0]

										if isinstance(self.scheduler	,					__snake_case		):
													lowerCamelCase						:Union[str, Any]						  =     latents.detach() + grads * (sigma**2)
													lowerCamelCase						:Dict						  =     noise_pred_original
										else:
													lowerCamelCase						:Dict						  =     noise_pred_original - torch.sqrt(__snake_case		) * grads
										return noise_pred, latents







							@torch.no_grad()
							def __call__(				self     :    Tuple	,					__snake_case     :    Union[torch.FloatTensor, PIL.Image.Image]	,					__snake_case     :    Union[torch.FloatTensor, PIL.Image.Image]	,					__snake_case     :    Optional[str] = None	,					__snake_case     :    Optional[str] = None	,					__snake_case     :    Optional[int] = 512	,					__snake_case     :    Optional[int] = 512	,					__snake_case     :    float = 0.6	,					__snake_case     :    Optional[int] = 50	,					__snake_case     :    Optional[float] = 7.5	,					__snake_case     :    Optional[int] = 1	,					__snake_case     :    float = 0.0	,					__snake_case     :    Optional[float] = 100	,					__snake_case     :    Optional[torch.Generator] = None	,					__snake_case     :    Optional[str] = "pil"	,					__snake_case     :    bool = True	,					__snake_case     :    float = 0.8	,					__snake_case     :    float = 0.1	,					__snake_case     :    float = 0.1	,					):
										if isinstance(__snake_case	,					__snake_case		) and len(__snake_case		) != batch_size:
													raise ValueError(F"You have passed {batch_size} batch_size, but only {len(__snake_case		)} 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(__snake_case	,					torch.Generator		) and batch_size > 1:
													lowerCamelCase						:Optional[int]						  =     [generator] + [None] * (batch_size - 1)

										lowerCamelCase						:Optional[Any]						  =     [
										    ('''model''', self.coca_model is None),
										    ('''tokenizer''', self.coca_tokenizer is None),
										    ('''transform''', self.coca_transform is None),
										]
										lowerCamelCase						:str						  =     [x[0] for x in coca_is_none if x[1]]
										lowerCamelCase						:List[Any]						  =     ''', '''.join(__snake_case		)
										# generate prompts with coca model if prompt is None
										if content_prompt is None:
													if len(__snake_case		):
																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."		)
													lowerCamelCase						:Optional[Any]						  =     self.get_image_description(__snake_case		)
										if style_prompt is None:
													if len(__snake_case		):
																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."		)
													lowerCamelCase						:Optional[Any]						  =     self.get_image_description(__snake_case		)

										# get prompt text embeddings for content and style
										lowerCamelCase						:Optional[Any]						  =     self.tokenizer(
										    __snake_case	,					padding='''max_length'''	,					max_length=self.tokenizer.model_max_length	,					truncation=__snake_case	,					return_tensors='''pt'''	,					)
										lowerCamelCase						:str						  =     self.text_encoder(content_text_input.input_ids.to(self.device		)		)[0]

										lowerCamelCase						:Union[str, Any]						  =     self.tokenizer(
										    __snake_case	,					padding='''max_length'''	,					max_length=self.tokenizer.model_max_length	,					truncation=__snake_case	,					return_tensors='''pt'''	,					)
										lowerCamelCase						:List[Any]						  =     self.text_encoder(style_text_input.input_ids.to(self.device		)		)[0]

										lowerCamelCase						:Tuple						  =     slerp(__snake_case	,					__snake_case	,					__snake_case		)

										# duplicate text embeddings for each generation per prompt
										lowerCamelCase						:List[Any]						  =     text_embeddings.repeat_interleave(__snake_case	,					dim=0		)

										# set timesteps
										lowerCamelCase						:Any						  =     '''offset''' in set(inspect.signature(self.scheduler.set_timesteps		).parameters.keys()		)
										lowerCamelCase						:int						  =     {}
										if accepts_offset:
													lowerCamelCase						:Dict						  =     1

										self.scheduler.set_timesteps(__snake_case	,					**__snake_case		)
										# 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		)

										lowerCamelCase  ,    lowerCamelCase						:Dict						  =     self.get_timesteps(__snake_case	,					__snake_case	,					self.device		)
										lowerCamelCase						:Optional[Any]						  =     timesteps[:1].repeat(__snake_case		)

										# Preprocess image
										lowerCamelCase						:List[Any]						  =     preprocess(__snake_case	,					__snake_case	,					__snake_case		)
										lowerCamelCase						:Union[str, Any]						  =     self.prepare_latents(
										    __snake_case	,					__snake_case	,					__snake_case	,					text_embeddings.dtype	,					self.device	,					__snake_case		)

										lowerCamelCase						:str						  =     preprocess(__snake_case	,					__snake_case	,					__snake_case		)
										lowerCamelCase						:Dict						  =     self.prepare_latents(
										    __snake_case	,					__snake_case	,					__snake_case	,					text_embeddings.dtype	,					self.device	,					__snake_case		)

										lowerCamelCase						:Tuple						  =     slerp(__snake_case	,					__snake_case	,					__snake_case		)

										if clip_guidance_scale > 0:
													lowerCamelCase						:Optional[int]						  =     self.get_clip_image_embeddings(__snake_case	,					__snake_case		)
													lowerCamelCase						:Dict						  =     self.get_clip_image_embeddings(__snake_case	,					__snake_case		)
													lowerCamelCase						:List[str]						  =     slerp(
													    __snake_case	,					__snake_case	,					__snake_case		)

										# 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.
										lowerCamelCase						:Tuple						  =     guidance_scale > 1.0
										# get unconditional embeddings for classifier free guidance
										if do_classifier_free_guidance:
													lowerCamelCase						:List[Any]						  =     content_text_input.input_ids.shape[-1]
													lowerCamelCase						:Any						  =     self.tokenizer(['''''']	,					padding='''max_length'''	,					max_length=__snake_case	,					return_tensors='''pt'''		)
													lowerCamelCase						:Dict						  =     self.text_encoder(uncond_input.input_ids.to(self.device		)		)[0]
													# duplicate unconditional embeddings for each generation per prompt
													lowerCamelCase						:Any						  =     uncond_embeddings.repeat_interleave(__snake_case	,					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
													lowerCamelCase						: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`.
										lowerCamelCase						:Union[str, Any]						  =     (batch_size, self.unet.config.in_channels, height // 8, width // 8)
										lowerCamelCase						:Union[str, Any]						  =     text_embeddings.dtype
										if latents is None:
													if self.device.type == "mps":
																# randn does not work reproducibly on mps
																lowerCamelCase						:Any						  =     torch.randn(__snake_case	,					generator=__snake_case	,					device='''cpu'''	,					dtype=__snake_case		).to(
																    self.device		)
													else:
																lowerCamelCase						:Optional[int]						  =     torch.randn(__snake_case	,					generator=__snake_case	,					device=self.device	,					dtype=__snake_case		)
										else:
													if latents.shape != latents_shape:
																raise ValueError(F"Unexpected latents shape, got {latents.shape}, expected {latents_shape}"		)
													lowerCamelCase						:List[str]						  =     latents.to(self.device		)

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

										# check if the scheduler accepts generator
										lowerCamelCase						:Optional[int]						  =     '''generator''' in set(inspect.signature(self.scheduler.step		).parameters.keys()		)
										if accepts_generator:
													lowerCamelCase						:Union[str, Any]						  =     generator

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

																# predict the noise residual
																lowerCamelCase						:List[str]						  =     self.unet(__snake_case	,					__snake_case	,					encoder_hidden_states=__snake_case		).sample

																# perform classifier free guidance
																if do_classifier_free_guidance:
																			lowerCamelCase  ,    lowerCamelCase						:Optional[int]						  =     noise_pred.chunk(2		)
																			lowerCamelCase						:Dict						  =     noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

																# perform clip guidance
																if clip_guidance_scale > 0:
																			lowerCamelCase						:List[Any]						  =     (
																			    text_embeddings.chunk(2		)[1] if do_classifier_free_guidance else text_embeddings
																			)
																			lowerCamelCase  ,    lowerCamelCase						:Optional[Any]						  =     self.cond_fn(
																			    __snake_case	,					__snake_case	,					__snake_case	,					__snake_case	,					__snake_case	,					__snake_case	,					__snake_case	,					)

																# compute the previous noisy sample x_t -> x_t-1
																lowerCamelCase						:int						  =     self.scheduler.step(__snake_case	,					__snake_case	,					__snake_case	,					**__snake_case		).prev_sample

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
										lowerCamelCase						:List[str]						  =     1 / 0.1_8_2_1_5 * latents
										lowerCamelCase						:Optional[int]						  =     self.vae.decode(__snake_case		).sample

										lowerCamelCase						:Dict						  =     (image / 2 + 0.5).clamp(0	,					1		)
										lowerCamelCase						:Union[str, Any]						  =     image.cpu().permute(0	,					2	,					3	,					1		).numpy()

										if output_type == "pil":
													lowerCamelCase						:Any						  =     self.numpy_to_pil(__snake_case		)

										if not return_dict:
													return (image, None)

										return StableDiffusionPipelineOutput(images=__snake_case	,					nsfw_content_detected=__snake_case		)




import ast
import os
import re
import shutil
import tempfile
import unittest
from unittest import mock

import torch

from accelerate.test_utils.examples import compare_against_test
from accelerate.test_utils.testing import TempDirTestCase, require_trackers, run_command, slow
from accelerate.utils import write_basic_config


# DataLoaders built from `test_samples/MRPC` for quick testing
# Should mock `{script_name}.get_dataloaders` via:
# @mock.patch("{script_name}.get_dataloaders", mocked_dataloaders)

A__           =							[
    """cross_validation.py""",
    """gradient_accumulation.py""",
    """local_sgd.py""",
    """multi_process_metrics.py""",
    """memory.py""",
    """automatic_gradient_accumulation.py""",
    """fsdp_with_peak_mem_tracking.py""",
    """deepspeed_with_config_support.py""",
    """megatron_lm_gpt_pretraining.py""",
]
class   _lowerCAmelCase		(     unittest.TestCase				):







							def 				snake_case  (				self     :    Any	,					__snake_case     :    str	,					__snake_case     :    bool	,					__snake_case     :    str = None	,					__snake_case     :    list = None		):
										lowerCamelCase						:Tuple						  =     None
										lowerCamelCase						:Tuple						  =     os.path.abspath(os.path.join('''examples'''	,					'''by_feature'''		)		)
										lowerCamelCase						:Optional[int]						  =     os.path.abspath('''examples'''		)
										for item in os.listdir(__snake_case		):
													if item not in EXCLUDE_EXAMPLES:
																lowerCamelCase						:Optional[int]						  =     os.path.join(__snake_case	,					__snake_case		)
																if os.path.isfile(__snake_case		) and ".py" in item_path:
																			with self.subTest(
																			    tested_script=__snake_case	,					feature_script=__snake_case	,					tested_section='''main()''' if parser_only else '''training_function()'''	,					):
																						lowerCamelCase						:Union[str, Any]						  =     compare_against_test(
																						    os.path.join(__snake_case	,					__snake_case		)	,					__snake_case	,					__snake_case	,					__snake_case		)
																						lowerCamelCase						:int						  =     '''\n'''.join(__snake_case		)
																						if special_strings is not None:
																									for string in special_strings:
																												lowerCamelCase						:int						  =     diff.replace(__snake_case	,					''''''		)
																						self.assertEqual(__snake_case	,					''''''		)







							def 				snake_case  (				self     :    Dict		):
										self.one_complete_example('''complete_nlp_example.py'''	,					__snake_case		)
										self.one_complete_example('''complete_nlp_example.py'''	,					__snake_case		)







							def 				snake_case  (				self     :    Optional[Any]		):
										lowerCamelCase						:Union[str, Any]						  =     os.path.abspath(os.path.join('''examples'''	,					'''cv_example.py'''		)		)
										lowerCamelCase						:Optional[int]						  =     [
										    ''' ''' * 16 + '''{\n\n''',
										    ''' ''' * 20 + '''"accuracy": eval_metric["accuracy"],\n\n''',
										    ''' ''' * 20 + '''"f1": eval_metric["f1"],\n\n''',
										    ''' ''' * 20 + '''"train_loss": total_loss.item() / len(train_dataloader),\n\n''',
										    ''' ''' * 20 + '''"epoch": epoch,\n\n''',
										    ''' ''' * 16 + '''},\n\n''',
										    ''' ''' * 16 + '''step=epoch,\n''',
										    ''' ''' * 12,
										    ''' ''' * 8 + '''for step, batch in enumerate(active_dataloader):\n''',
										]
										self.one_complete_example('''complete_cv_example.py'''	,					__snake_case	,					__snake_case	,					__snake_case		)
										self.one_complete_example('''complete_cv_example.py'''	,					__snake_case	,					__snake_case	,					__snake_case		)


@mock.patch.dict(os.environ      ,   {'TESTING_MOCKED_DATALOADERS': '1'}				)
class   _lowerCAmelCase		(     __SCREAMING_SNAKE_CASE				):
							_UpperCAmelCase						    =     False







							@classmethod
							def 				snake_case  (				cls     :    Optional[Any]		):
										super().setUpClass()
										lowerCamelCase						:Any						  =     tempfile.mkdtemp()
										lowerCamelCase						:Optional[int]						  =     os.path.join(cls._tmpdir	,					'''default_config.yml'''		)

										write_basic_config(save_location=cls.configPath		)
										lowerCamelCase						:List[str]						  =     ['''accelerate''', '''launch''', '''--config_file''', cls.configPath]







							@classmethod
							def 				snake_case  (				cls     :    Dict		):
										super().tearDownClass()
										shutil.rmtree(cls._tmpdir		)







							def 				snake_case  (				self     :    int		):
										lowerCamelCase						:Any						  =     F"\n        examples/by_feature/checkpointing.py\n        --checkpointing_steps epoch\n        --output_dir {self.tmpdir}\n        ".split()
										run_command(self._launch_args + testargs		)
										self.assertTrue(os.path.exists(os.path.join(self.tmpdir	,					'''epoch_0'''		)		)		)







							def 				snake_case  (				self     :    List[Any]		):
										lowerCamelCase						:Tuple						  =     F"\n        examples/by_feature/checkpointing.py\n        --checkpointing_steps 1\n        --output_dir {self.tmpdir}\n        ".split()
										lowerCamelCase						:List[Any]						  =     run_command(self._launch_args + testargs		)
										self.assertTrue(os.path.exists(os.path.join(self.tmpdir	,					'''step_2'''		)		)		)







							def 				snake_case  (				self     :    List[str]		):
										lowerCamelCase						:Dict						  =     F"\n        examples/by_feature/checkpointing.py\n        --resume_from_checkpoint {os.path.join(self.tmpdir	,					'epoch_0'		)}\n        ".split()
										lowerCamelCase						:Dict						  =     run_command(self._launch_args + testargs	,					return_stdout=__snake_case		)
										self.assertNotIn('''epoch 0:'''	,					__snake_case		)
										self.assertIn('''epoch 1:'''	,					__snake_case		)







							def 				snake_case  (				self     :    str		):
										lowerCamelCase						:List[Any]						  =     F"\n        examples/by_feature/checkpointing.py\n        --resume_from_checkpoint {os.path.join(self.tmpdir	,					'step_2'		)}\n        ".split()
										lowerCamelCase						:Optional[int]						  =     run_command(self._launch_args + testargs	,					return_stdout=__snake_case		)
										if torch.cuda.is_available():
													lowerCamelCase						:Union[str, Any]						  =     torch.cuda.device_count()
										else:
													lowerCamelCase						:Dict						  =     1
										if num_processes > 1:
													self.assertNotIn('''epoch 0:'''	,					__snake_case		)
													self.assertIn('''epoch 1:'''	,					__snake_case		)
										else:
													self.assertIn('''epoch 0:'''	,					__snake_case		)
													self.assertIn('''epoch 1:'''	,					__snake_case		)







							@slow
							def 				snake_case  (				self     :    Any		):
										lowerCamelCase						:Tuple						  =     '''
        examples/by_feature/cross_validation.py
        --num_folds 2
        '''.split()
										with mock.patch.dict(os.environ	,					{'''TESTING_MOCKED_DATALOADERS''': '''0'''}		):
													lowerCamelCase						:Dict						  =     run_command(self._launch_args + testargs	,					return_stdout=__snake_case		)
													lowerCamelCase						:Tuple						  =     re.findall('''({.+})'''	,					__snake_case		)
													lowerCamelCase						:Optional[Any]						  =     [r for r in results if '''accuracy''' in r][-1]
													lowerCamelCase						:List[str]						  =     ast.literal_eval(__snake_case		)
													self.assertGreaterEqual(results['''accuracy''']	,					0.7_5		)







							def 				snake_case  (				self     :    int		):
										lowerCamelCase						:Dict						  =     ['''examples/by_feature/multi_process_metrics.py''']
										run_command(self._launch_args + testargs		)







							@require_trackers
							@mock.patch.dict(os.environ	,					{'''WANDB_MODE''': '''offline'''}		)
							def 				snake_case  (				self     :    Any		):
										with tempfile.TemporaryDirectory() as tmpdir:
													lowerCamelCase						:Tuple						  =     F"\n            examples/by_feature/tracking.py\n            --with_tracking\n            --project_dir {tmpdir}\n            ".split()
													run_command(self._launch_args + testargs		)
													self.assertTrue(os.path.exists(os.path.join(__snake_case	,					'''tracking'''		)		)		)







							def 				snake_case  (				self     :    Tuple		):
										lowerCamelCase						:Tuple						  =     ['''examples/by_feature/gradient_accumulation.py''']
										run_command(self._launch_args + testargs		)







							def 				snake_case  (				self     :    Optional[Any]		):
										lowerCamelCase						:int						  =     ['''examples/by_feature/local_sgd.py''']
										run_command(self._launch_args + testargs		)



