





import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionAttendAndExcitePipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_numpy, skip_mps, slow
from diffusers.utils.testing_utils import require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


A   : int			=				False


@skip_mps
class   A      (SCREAMING_SNAKE_CASE       ,					SCREAMING_SNAKE_CASE       ,					SCREAMING_SNAKE_CASE       ,					unittest.TestCase   ):
			'''simple docstring'''




			__lowerCamelCase     :	Union[str, Any]          =   StableDiffusionAttendAndExcitePipeline
			__lowerCamelCase     :	Optional[int]          =   False
			__lowerCamelCase     :	List[str]          =   TEXT_TO_IMAGE_PARAMS
			__lowerCamelCase     :	Union[str, Any]          =   TEXT_TO_IMAGE_BATCH_PARAMS.union({'''token_indices'''}   )
			__lowerCamelCase     :	Any          =   TEXT_TO_IMAGE_IMAGE_PARAMS
			__lowerCamelCase     :	List[str]          =   TEXT_TO_IMAGE_IMAGE_PARAMS






			@classmethod
			def        a_	(      cls       :   Dict       )     ->     int:







					"""simple docstring"""






					super().setUpClass()
					torch.use_deterministic_algorithms(__lowerCAmelCase       )






			@classmethod
			def        a_	(      cls       :   Optional[Any]       )     ->     Optional[int]:







					"""simple docstring"""






					super().tearDownClass()
					torch.use_deterministic_algorithms(__lowerCAmelCase       )






			def        a_	(      self       :   Optional[Any]       )     ->     str:







					"""simple docstring"""






					torch.manual_seed(0       )
					A__  				=       UNetaDConditionModel(
					    block_out_channels=(32, 64)							,     layers_per_block=1							,     sample_size=32							,     in_channels=4							,     out_channels=4							,     down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")							,     up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")							,     cross_attention_dim=32							,     attention_head_dim=(2, 4)							,     use_linear_projection=__lowerCAmelCase							,     )
					A__  				=       DDIMScheduler(
					    beta_start=0.0_0_0_8_5							,     beta_end=0.0_1_2							,     beta_schedule="""scaled_linear"""							,     clip_sample=__lowerCAmelCase							,     set_alpha_to_one=__lowerCAmelCase							,     )
					torch.manual_seed(0       )
					A__  				=       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							,     sample_size=1_28							,     )
					torch.manual_seed(0       )
					A__  				=       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=10_00							,     hidden_act="""gelu"""							,     projection_dim=5_12							,     )
					A__  				=       CLIPTextModel(__lowerCAmelCase       )
					A__  				=       CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""       )

					A__  				=       {
					    """unet""": unet,
					    """scheduler""": scheduler,
					    """vae""": vae,
					    """text_encoder""": text_encoder,
					    """tokenizer""": tokenizer,
					    """safety_checker""": None,
					    """feature_extractor""": None,
					}

					return components






			def        a_	(      self       :   str							,     __lowerCAmelCase       :   Dict							,     __lowerCAmelCase       :   int=0       )     ->     Union[str, Any]:







					"""simple docstring"""






					if str(__lowerCAmelCase       ).startswith("""mps"""       ):
							A__  				=       torch.manual_seed(__lowerCAmelCase       )
					else:
							A__  				=       torch.Generator(device=__lowerCAmelCase       ).manual_seed(__lowerCAmelCase       )
					A__  				=       A__  				=       {
					    """prompt""": """a cat and a frog""",
					    """token_indices""": [2, 5],
					    """generator""": generator,
					    """num_inference_steps""": 1,
					    """guidance_scale""": 6.0,
					    """output_type""": """numpy""",
					    """max_iter_to_alter""": 2,
					    """thresholds""": {0: 0.7},
					}
					return inputs






			def        a_	(      self       :   Union[str, Any]       )     ->     str:







					"""simple docstring"""






					A__  				=       """cpu"""

					A__  				=       self.get_dummy_components()
					A__  				=       self.pipeline_class(**__lowerCAmelCase       )
					pipe.to(__lowerCAmelCase       )
					pipe.set_progress_bar_config(disable=__lowerCAmelCase       )

					A__  				=       self.get_dummy_inputs(__lowerCAmelCase       )
					A__  				=       pipe(**__lowerCAmelCase       ).images
					A__  				=       image[0, -3:, -3:, -1]

					self.assertEqual(image.shape							,     (1, 64, 64, 3)       )
					A__  				=       np.array(
					    [0.6_3_9_0_5_3_6_4, 0.6_2_8_9_7_3_0_7, 0.4_8_5_9_9_0_1_7, 0.5_1_3_3_6_2_4, 0.5_5_5_0_0_4_8, 0.4_5_7_6_9_5_1_6, 0.5_0_3_2_6_9_7_3, 0.5_0_2_3_1_3_9, 0.4_5_3_8_4_4_9_6]       )
					A__  				=       np.abs(image_slice.flatten() - expected_slice       ).max()
					self.assertLessEqual(__lowerCAmelCase							,     1e-3       )






			def        a_	(      self       :   Tuple       )     ->     Optional[Any]:







					"""simple docstring"""






					super().test_cpu_offload_forward_pass(expected_max_diff=5e-4       )






			def        a_	(      self       :   Optional[Any]       )     ->     Optional[Any]:







					"""simple docstring"""






					self._test_inference_batch_consistent(batch_sizes=[1, 2]       )






			def        a_	(      self       :   List[str]       )     ->     str:







					"""simple docstring"""






					self._test_inference_batch_single_identical(batch_size=2							,     expected_max_diff=7e-4       )






			def        a_	(      self       :   int       )     ->     Tuple:







					"""simple docstring"""






					super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3       )






			def        a_	(      self       :   Dict       )     ->     str:







					"""simple docstring"""






					super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5e-4       )






			def        a_	(      self       :   List[str]       )     ->     List[Any]:







					"""simple docstring"""






					super().test_save_load_local(expected_max_difference=5e-4       )







			def        a_	(      self       :   str       )     ->     Any:







					"""simple docstring"""






					super().test_save_load_optional_components(expected_max_difference=4e-4       )


@require_torch_gpu
@slow
class   A      (unittest.TestCase   ):
			'''simple docstring'''




			@classmethod
			def        a_	(      cls       :   List[str]       )     ->     str:







					"""simple docstring"""






					super().setUpClass()
					torch.use_deterministic_algorithms(__lowerCAmelCase       )






			@classmethod
			def        a_	(      cls       :   int       )     ->     Union[str, Any]:







					"""simple docstring"""






					super().tearDownClass()
					torch.use_deterministic_algorithms(__lowerCAmelCase       )






			def        a_	(      self       :   List[Any]       )     ->     Union[str, Any]:







					"""simple docstring"""






					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()







			def        a_	(      self       :   List[str]       )     ->     List[str]:







					"""simple docstring"""






					A__  				=       torch.manual_seed(51       )

					A__  				=       StableDiffusionAttendAndExcitePipeline.from_pretrained(
					    """CompVis/stable-diffusion-v1-4"""							,     safety_checker=__lowerCAmelCase							,     torch_dtype=torch.floataa       )
					pipe.to("""cuda"""       )

					A__  				=       """a painting of an elephant with glasses"""
					A__  				=       [5, 7]

					A__  				=       pipe(
					    prompt=__lowerCAmelCase							,     token_indices=__lowerCAmelCase							,     guidance_scale=7.5							,     generator=__lowerCAmelCase							,     num_inference_steps=5							,     max_iter_to_alter=5							,     output_type="""numpy"""							,     ).images[0]

					A__  				=       load_numpy(
					    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/attend-and-excite/elephant_glasses.npy"""       )
					assert np.abs((expected_image - image).max()       ) < 5e-1










import argparse
import math
import traceback

import dateutil.parser as date_parser
import requests







def        __lowerCamelCase (							__a       :str   )		->    Optional[int]:






		"""simple docstring"""

		A__  				=       {}

		A__  				=       job["""started_at"""]
		A__  				=       job["""completed_at"""]

		A__  				=       date_parser.parse(__a   )
		A__  				=       date_parser.parse(__a   )

		A__  				=       round((end_datetime - start_datetime).total_seconds() / 60.0   )

		A__  				=       start
		A__  				=       end
		A__  				=       duration_in_min

		return job_info







def        __lowerCamelCase (							__a       :Optional[Any]   ,					__a       :List[str]=None   )		->    List[Any]:






		"""simple docstring"""

		A__  				=       None
		if token is not None:
				A__  				=       {"""Accept""": """application/vnd.github+json""", """Authorization""": F'Bearer {token}'}

		A__  				=       F'https://api.github.com/repos/huggingface/transformers/actions/runs/{workflow_run_id}/jobs?per_page=100'
		A__  				=       requests.get(__a   ,					headers=__a   ).json()
		A__  				=       {}

		try:
				job_time.update({job["""name"""]: extract_time_from_single_job(__a   ) for job in result["""jobs"""]}   )
				A__  				=       math.ceil((result["""total_count"""] - 1_0_0) / 1_0_0   )

				for i in range(__a   ):
						A__  				=       requests.get(url + F'&page={i + 2}'   ,					headers=__a   ).json()
						job_time.update({job["""name"""]: extract_time_from_single_job(__a   ) for job in result["""jobs"""]}   )

				return job_time
		except Exception:
				print(F'Unknown error, could not fetch links:\n{traceback.format_exc()}'   )

		return {}


if __name__ == "__main__":
					A   : Optional[Any]			=				argparse.ArgumentParser()
					# Required parameters
					parser.add_argument('''--workflow_run_id''', type=str, required=True, help='''A GitHub Actions workflow run id.''')
					A   : Dict			=				parser.parse_args()

					A   : List[Any]			=				get_job_time(args.workflow_run_id)
					A   : int			=				dict(sorted(job_time.items(), key=lambda item: item[1]["duration"], reverse=True))

					for k, v in job_time.items():
										print(F'''{k}: {v["duration"]}''')



