






import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionInstructPixaPixPipeline,
    UNetaDConditionModel,
)
from diffusers.image_processor import VaeImageProcessor
from diffusers.utils import floats_tensor, load_image, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()






class 						UpperCAmelCase__ (							__snake_case     ,				__snake_case     ,				__snake_case     ,				unittest.TestCase							):
		__snake_case      :       str     =   StableDiffusionInstructPixaPixPipeline
		__snake_case      :       Tuple     =   TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"height", "width", "cross_attention_kwargs"}
		__snake_case      :       Dict     =   TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
		__snake_case      :       Tuple     =   IMAGE_TO_IMAGE_IMAGE_PARAMS
		__snake_case      :       int     =   IMAGE_TO_IMAGE_IMAGE_PARAMS







		def  A__      (					self		):
						torch.manual_seed(0		)
						_A	:				Dict   =      UNetaDConditionModel(
						    block_out_channels=(32, 64)       ,layers_per_block=2       ,sample_size=32       ,in_channels=8       ,out_channels=4       ,down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')       ,up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')       ,cross_attention_dim=32       ,)
						_A	:				Union[str, Any]   =      PNDMScheduler(skip_prk_steps=A__		)
						torch.manual_seed(0		)
						_A	:				Optional[int]   =      AutoencoderKL(
						    block_out_channels=[32, 64]       ,in_channels=3       ,out_channels=3       ,down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']       ,up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']       ,latent_channels=4       ,)
						torch.manual_seed(0		)
						_A	:				Tuple   =      CLIPTextConfig(
						    bos_token_id=0       ,eos_token_id=2       ,hidden_size=32       ,intermediate_size=37       ,layer_norm_eps=1E-05       ,num_attention_heads=4       ,num_hidden_layers=5       ,pad_token_id=1       ,vocab_size=1000       ,)
						_A	:				List[Any]   =      CLIPTextModel(A__		)
						_A	:				Any   =      CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''		)

						_A	:				Union[str, Any]   =      {
						    '''unet''': unet,
						    '''scheduler''': scheduler,
						    '''vae''': vae,
						    '''text_encoder''': text_encoder,
						    '''tokenizer''': tokenizer,
						    '''safety_checker''': None,
						    '''feature_extractor''': None,
						}
						return components







		def  A__      (					self       ,A__       ,A__=0		):
						_A	:				List[str]   =      floats_tensor((1, 3, 32, 32)       ,rng=random.Random(A__		)		).to(A__		)
						_A	:				Union[str, Any]   =      image.cpu().permute(0       ,2       ,3       ,1		)[0]
						_A	:				List[Any]   =      Image.fromarray(np.uinta(A__		)		).convert('''RGB'''		)
						if str(A__		).startswith('''mps'''		):
										_A	:				Tuple   =      torch.manual_seed(A__		)
						else:
										_A	:				Optional[int]   =      torch.Generator(device=A__		).manual_seed(A__		)
						_A	:				Optional[int]   =      {
						    '''prompt''': '''A painting of a squirrel eating a burger''',
						    '''image''': image,
						    '''generator''': generator,
						    '''num_inference_steps''': 2,
						    '''guidance_scale''': 6.0,
						    '''image_guidance_scale''': 1,
						    '''output_type''': '''numpy''',
						}
						return inputs







		def  A__      (					self		):
						_A	:				int   =      '''cpu'''  # ensure determinism for the device-dependent torch.Generator
						_A	:				Any   =      self.get_dummy_components()
						_A	:				str   =      StableDiffusionInstructPixaPixPipeline(**A__		)
						_A	:				str   =      sd_pipe.to(A__		)
						sd_pipe.set_progress_bar_config(disable=A__		)

						_A	:				List[str]   =      self.get_dummy_inputs(A__		)
						_A	:				Any   =      sd_pipe(**A__		).images
						_A	:				Union[str, Any]   =      image[0, -3:, -3:, -1]
						assert image.shape == (1, 32, 32, 3)
						_A	:				Optional[int]   =      np.array([0.75_26, 0.37_50, 0.45_47, 0.61_17, 0.58_66, 0.50_16, 0.43_27, 0.56_42, 0.48_15]		)

						assert np.abs(image_slice.flatten() - expected_slice		).max() < 1E-3







		def  A__      (					self		):
						_A	:				Union[str, Any]   =      '''cpu'''  # ensure determinism for the device-dependent torch.Generator
						_A	:				Dict   =      self.get_dummy_components()
						_A	:				Any   =      StableDiffusionInstructPixaPixPipeline(**A__		)
						_A	:				Optional[int]   =      sd_pipe.to(A__		)
						sd_pipe.set_progress_bar_config(disable=A__		)

						_A	:				Optional[Any]   =      self.get_dummy_inputs(A__		)
						_A	:				Any   =      '''french fries'''
						_A	:				List[Any]   =      sd_pipe(**A__       ,negative_prompt=A__		)
						_A	:				List[Any]   =      output.images
						_A	:				Dict   =      image[0, -3:, -3:, -1]

						assert image.shape == (1, 32, 32, 3)
						_A	:				Dict   =      np.array([0.75_11, 0.36_42, 0.45_53, 0.62_36, 0.57_97, 0.50_13, 0.43_43, 0.56_11, 0.48_31]		)

						assert np.abs(image_slice.flatten() - expected_slice		).max() < 1E-3







		def  A__      (					self		):
						_A	:				int   =      '''cpu'''  # ensure determinism for the device-dependent torch.Generator
						_A	:				Union[str, Any]   =      self.get_dummy_components()
						_A	:				List[Any]   =      StableDiffusionInstructPixaPixPipeline(**A__		)
						_A	:				Optional[int]   =      sd_pipe.to(A__		)
						sd_pipe.set_progress_bar_config(disable=A__		)

						_A	:				Dict   =      self.get_dummy_inputs(A__		)
						_A	:				Optional[int]   =      [inputs['''prompt''']] * 2

						_A	:				List[Any]   =      np.array(inputs['''image''']		).astype(np.floataa		) / 2_55.0
						_A	:				Dict   =      torch.from_numpy(A__		).unsqueeze(0		).to(A__		)
						_A	:				Any   =      image / 2 + 0.5
						_A	:				Tuple   =      image.permute(0       ,3       ,1       ,2		)
						_A	:				Dict   =      image.repeat(2       ,1       ,1       ,1		)

						_A	:				List[str]   =      sd_pipe(**A__		).images
						_A	:				Optional[Any]   =      image[-1, -3:, -3:, -1]

						assert image.shape == (2, 32, 32, 3)
						_A	:				Union[str, Any]   =      np.array([0.58_12, 0.57_48, 0.52_22, 0.59_08, 0.56_95, 0.71_74, 0.68_04, 0.55_23, 0.55_79]		)

						assert np.abs(image_slice.flatten() - expected_slice		).max() < 1E-3







		def  A__      (					self		):
						_A	:				List[Any]   =      '''cpu'''  # ensure determinism for the device-dependent torch.Generator
						_A	:				List[str]   =      self.get_dummy_components()
						_A	:				str   =      EulerAncestralDiscreteScheduler(
						    beta_start=0.0_00_85       ,beta_end=0.0_12       ,beta_schedule='''scaled_linear'''		)
						_A	:				Optional[Any]   =      StableDiffusionInstructPixaPixPipeline(**A__		)
						_A	:				str   =      sd_pipe.to(A__		)
						sd_pipe.set_progress_bar_config(disable=A__		)

						_A	:				Dict   =      self.get_dummy_inputs(A__		)
						_A	:				Dict   =      sd_pipe(**A__		).images
						_A	:				Optional[Any]   =      image[0, -3:, -3:, -1]

						_A	:				Tuple   =      [round(A__       ,4		) for x in image_slice.flatten().tolist()]
						print(''','''.join([str(A__		) for x in slice]		)		)

						assert image.shape == (1, 32, 32, 3)
						_A	:				Union[str, Any]   =      np.array([0.74_17, 0.38_42, 0.47_32, 0.57_76, 0.58_91, 0.51_39, 0.40_52, 0.56_73, 0.49_86]		)

						assert np.abs(image_slice.flatten() - expected_slice		).max() < 1E-3







		def  A__      (					self		):
						super().test_inference_batch_single_identical(expected_max_diff=3E-3		)







		def  A__      (					self		):
						_A	:				Union[str, Any]   =      self.get_dummy_components()
						_A	:				List[str]   =      StableDiffusionInstructPixaPixPipeline(**A__		)
						_A	:				List[str]   =      VaeImageProcessor(do_resize=A__       ,do_normalize=A__		)
						_A	:				Optional[Any]   =      pipe.to(A__		)
						pipe.set_progress_bar_config(disable=A__		)

						_A	:				Dict   =      pipe(**self.get_dummy_inputs_by_type(A__       ,input_image_type='''pt'''		)		)[0]

						_A	:				Optional[Any]   =      components['''vae''']
						_A	:				Optional[int]   =      self.get_dummy_inputs_by_type(A__       ,input_image_type='''pt'''		)

						for image_param in self.image_latents_params:
										if image_param in inputs.keys():
														_A	:				Tuple   =      vae.encode(inputs[image_param]		).latent_dist.mode()

						_A	:				Union[str, Any]   =      pipe(**A__		)[0]

						_A	:				Union[str, Any]   =      np.abs(out - out_latents_inputs		).max()
						self.assertLess(A__       ,1E-4       ,'''passing latents as image input generate different result from passing image'''		)






@slow
@require_torch_gpu
class 						UpperCAmelCase__ (							unittest.TestCase							):







		def  A__      (					self		):
						super().tearDown()
						gc.collect()
						torch.cuda.empty_cache()







		def  A__      (					self       ,A__=0		):
						_A	:				Optional[int]   =      torch.manual_seed(A__		)
						_A	:				str   =      load_image(
						    '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/stable_diffusion_pix2pix/example.jpg'''		)
						_A	:				Optional[Any]   =      {
						    '''prompt''': '''turn him into a cyborg''',
						    '''image''': image,
						    '''generator''': generator,
						    '''num_inference_steps''': 3,
						    '''guidance_scale''': 7.5,
						    '''image_guidance_scale''': 1.0,
						    '''output_type''': '''numpy''',
						}
						return inputs







		def  A__      (					self		):
						_A	:				Union[str, Any]   =      StableDiffusionInstructPixaPixPipeline.from_pretrained(
						    '''timbrooks/instruct-pix2pix'''       ,safety_checker=A__		)
						pipe.to(A__		)
						pipe.set_progress_bar_config(disable=A__		)
						pipe.enable_attention_slicing()

						_A	:				Any   =      self.get_inputs()
						_A	:				Tuple   =      pipe(**A__		).images
						_A	:				Union[str, Any]   =      image[0, -3:, -3:, -1].flatten()

						assert image.shape == (1, 512, 512, 3)
						_A	:				List[Any]   =      np.array([0.59_02, 0.60_15, 0.60_27, 0.59_83, 0.60_92, 0.60_61, 0.57_65, 0.57_85, 0.55_55]		)

						assert np.abs(expected_slice - image_slice		).max() < 1E-3







		def  A__      (					self		):
						_A	:				Optional[Any]   =      StableDiffusionInstructPixaPixPipeline.from_pretrained(
						    '''timbrooks/instruct-pix2pix'''       ,safety_checker=A__		)
						_A	:				List[str]   =      LMSDiscreteScheduler.from_config(pipe.scheduler.config		)
						pipe.to(A__		)
						pipe.set_progress_bar_config(disable=A__		)
						pipe.enable_attention_slicing()

						_A	:				Any   =      self.get_inputs()
						_A	:				List[Any]   =      pipe(**A__		).images
						_A	:				List[str]   =      image[0, -3:, -3:, -1].flatten()

						assert image.shape == (1, 512, 512, 3)
						_A	:				Union[str, Any]   =      np.array([0.65_78, 0.68_17, 0.69_72, 0.67_61, 0.68_56, 0.69_16, 0.64_28, 0.65_16, 0.63_01]		)

						assert np.abs(expected_slice - image_slice		).max() < 1E-3







		def  A__      (					self		):
						_A	:				Optional[Any]   =      StableDiffusionInstructPixaPixPipeline.from_pretrained(
						    '''timbrooks/instruct-pix2pix'''       ,safety_checker=A__		)
						_A	:				Any   =      DDIMScheduler.from_config(pipe.scheduler.config		)
						pipe.to(A__		)
						pipe.set_progress_bar_config(disable=A__		)
						pipe.enable_attention_slicing()

						_A	:				str   =      self.get_inputs()
						_A	:				Optional[Any]   =      pipe(**A__		).images
						_A	:				Dict   =      image[0, -3:, -3:, -1].flatten()

						assert image.shape == (1, 512, 512, 3)
						_A	:				int   =      np.array([0.38_28, 0.38_34, 0.38_18, 0.37_92, 0.38_65, 0.37_52, 0.37_92, 0.38_47, 0.37_53]		)

						assert np.abs(expected_slice - image_slice		).max() < 1E-3







		def  A__      (					self		):
						_A	:				Union[str, Any]   =      0

						def callback_fn(A__       ,A__       ,A__		) -> None:
										_A	:				Union[str, Any]   =      True
										nonlocal number_of_steps
										number_of_steps += 1
										if step == 1:
														_A	:				int   =      latents.detach().cpu().numpy()
														assert latents.shape == (1, 4, 64, 64)
														_A	:				List[str]   =      latents[0, -3:, -3:, -1]
														_A	:				Any   =      np.array([-0.24_63, -0.46_44, -0.97_56, 1.51_76, 1.44_14, 0.78_66, 0.98_97, 0.85_21, 0.79_83]		)

														assert np.abs(latents_slice.flatten() - expected_slice		).max() < 5E-2
										elif step == 2:
														_A	:				str   =      latents.detach().cpu().numpy()
														assert latents.shape == (1, 4, 64, 64)
														_A	:				str   =      latents[0, -3:, -3:, -1]
														_A	:				Optional[int]   =      np.array([-0.26_44, -0.46_26, -0.96_53, 1.51_76, 1.45_51, 0.76_86, 0.98_05, 0.84_52, 0.81_15]		)

														assert np.abs(latents_slice.flatten() - expected_slice		).max() < 5E-2

						_A	:				Union[str, Any]   =      False

						_A	:				Dict   =      StableDiffusionInstructPixaPixPipeline.from_pretrained(
						    '''timbrooks/instruct-pix2pix'''       ,safety_checker=A__       ,torch_dtype=torch.floataa		)
						_A	:				List[Any]   =      pipe.to(A__		)
						pipe.set_progress_bar_config(disable=A__		)
						pipe.enable_attention_slicing()

						_A	:				int   =      self.get_inputs()
						pipe(**A__       ,callback=A__       ,callback_steps=1		)
						assert callback_fn.has_been_called
						assert number_of_steps == 3







		def  A__      (					self		):
						torch.cuda.empty_cache()
						torch.cuda.reset_max_memory_allocated()
						torch.cuda.reset_peak_memory_stats()

						_A	:				str   =      StableDiffusionInstructPixaPixPipeline.from_pretrained(
						    '''timbrooks/instruct-pix2pix'''       ,safety_checker=A__       ,torch_dtype=torch.floataa		)
						_A	:				Union[str, Any]   =      pipe.to(A__		)
						pipe.set_progress_bar_config(disable=A__		)
						pipe.enable_attention_slicing(1		)
						pipe.enable_sequential_cpu_offload()

						_A	:				Tuple   =      self.get_inputs()
						_A	:				int   =      pipe(**A__		)

						_A	:				str   =      torch.cuda.max_memory_allocated()
						# make sure that less than 2.2 GB is allocated
						assert mem_bytes < 2.2 * 10**9







		def  A__      (					self		):
						_A	:				Any   =      self.get_inputs()
						# resize to resolution that is divisible by 8 but not 16 or 32
						_A	:				int   =      inputs['''image'''].resize((504, 504)		)

						_A	:				List[str]   =      '''timbrooks/instruct-pix2pix'''
						_A	:				Any   =      StableDiffusionInstructPixaPixPipeline.from_pretrained(
						    A__       ,safety_checker=A__       ,)
						pipe.to(A__		)
						pipe.set_progress_bar_config(disable=A__		)
						pipe.enable_attention_slicing()

						_A	:				Optional[Any]   =      pipe(**A__		)
						_A	:				Dict   =      output.images[0]

						_A	:				Dict   =      image[255:258, 383:386, -1]

						assert image.shape == (504, 504, 3)
						_A	:				Union[str, Any]   =      np.array([0.27_26, 0.25_29, 0.26_64, 0.26_55, 0.26_41, 0.26_42, 0.25_91, 0.26_49, 0.25_90]		)

						assert np.abs(image_slice.flatten() - expected_slice		).max() < 5E-3











from __future__ import annotations

import json

import requests
from bsa import BeautifulSoup
from fake_useragent import UserAgent

_UpperCamelCase			:	Union[str, Any]								={'UserAgent': UserAgent().random}



def  a__ (__lowercase							:Optional[Any]   )		-> dict:
				_A	:				str   =      script.contents[0]
				_A	:				Dict   =      json.loads(data[data.find('''{"config"'''   ) : -1]   )
				return info["entry_data"]["ProfilePage"][0]["graphql"]["user"]






class 						UpperCAmelCase__ :







		def __init__(					self       ,A__		):
						_A	:				Any   =      f"""https://www.instagram.com/{username}/"""
						_A	:				Optional[Any]   =      self.get_json()







		def  A__      (					self		):
						_A	:				str   =      requests.get(self.url       ,headers=A__		).text
						_A	:				Dict   =      BeautifulSoup(A__       ,'''html.parser'''		).find_all('''script'''		)
						try:
										return extract_user_profile(scripts[4]		)
						except (json.decoder.JSONDecodeError, KeyError):
										return extract_user_profile(scripts[3]		)







		def __repr__(					self		):
						return f"""{self.__class__.__name__}('{self.username}')"""







		def __str__(					self		):
						return f"""{self.fullname} ({self.username}) is {self.biography}"""







		@property
		def  A__      (					self		):
						return self.user_data["username"]







		@property
		def  A__      (					self		):
						return self.user_data["full_name"]







		@property
		def  A__      (					self		):
						return self.user_data["biography"]







		@property
		def  A__      (					self		):
						return self.user_data["business_email"]







		@property
		def  A__      (					self		):
						return self.user_data["external_url"]







		@property
		def  A__      (					self		):
						return self.user_data["edge_followed_by"]["count"]







		@property
		def  A__      (					self		):
						return self.user_data["edge_follow"]["count"]







		@property
		def  A__      (					self		):
						return self.user_data["edge_owner_to_timeline_media"]["count"]







		@property
		def  A__      (					self		):
						return self.user_data["profile_pic_url_hd"]







		@property
		def  A__      (					self		):
						return self.user_data["is_verified"]







		@property
		def  A__      (					self		):
						return self.user_data["is_private"]



def  a__ (__lowercase							:str = "github"   )		-> None:
				import os

				if os.environ.get('''CI'''   ):
								return  # test failing on GitHub Actions
				_A	:				Optional[int]   =      InstagramUser(__lowercase   )
				assert instagram_user.user_data
				assert isinstance(instagram_user.user_data					,  __lowercase   )
				assert instagram_user.username == username
				if username != "github":
								return
				assert instagram_user.fullname == "GitHub"
				assert instagram_user.biography == "Built for developers."
				assert instagram_user.number_of_posts > 150
				assert instagram_user.number_of_followers > 12_0000
				assert instagram_user.number_of_followings > 15
				assert instagram_user.email == "support@github.com"
				assert instagram_user.website == "https://github.com/readme"
				assert instagram_user.profile_picture_url.startswith('''https://instagram.'''   )
				assert instagram_user.is_verified is True
				assert instagram_user.is_private is False


if __name__ == "__main__":
	import doctest

	doctest.testmod()
	_UpperCamelCase			:	List[Any]								=InstagramUser('github')
	print(instagram_user)
	print(f'''{instagram_user.number_of_posts = }''')
	print(f'''{instagram_user.number_of_followers = }''')
	print(f'''{instagram_user.number_of_followings = }''')
	print(f'''{instagram_user.email = }''')
	print(f'''{instagram_user.website = }''')
	print(f'''{instagram_user.profile_picture_url = }''')
	print(f'''{instagram_user.is_verified = }''')
	print(f'''{instagram_user.is_private = }''')



