


"""simple docstring"""


# Copyright 2021 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse
import os

from accelerate.utils import ComputeEnvironment

from .cluster import get_cluster_input
from .config_args import cache_dir, default_config_file, default_yaml_config_file, load_config_from_file  # noqa: F401
from .config_utils import _ask_field, _ask_options, _convert_compute_environment  # noqa: F401
from .sagemaker import get_sagemaker_input


__lowerCamelCase      :List[Any]	    =		'Launches a series of prompts to create and save a `default_config.yaml` configuration file for your training system. Should always be ran first on your machine'




def   snake_case		(     )   ->      List[str]:
      lowerCamelCase					:					Any        =	_ask_options(
          """In which compute environment are you running?"""			,				["""This machine""", """AWS (Amazon SageMaker)"""]			,				_convert_compute_environment			,				)
      if compute_environment == ComputeEnvironment.AMAZON_SAGEMAKER:
            lowerCamelCase					:					str        =	get_sagemaker_input()
      else:
            lowerCamelCase					:					Dict        =	get_cluster_input()
      return config




def   snake_case		(     UpperCamelCase__       :						Optional[int]=None    )   ->      Optional[Any]:
      if subparsers is not None:
            lowerCamelCase					:					Any        =	subparsers.add_parser("""config"""			,				description=UpperCamelCase__    )
      else:
            lowerCamelCase					:					Optional[Any]        =	argparse.ArgumentParser("""Accelerate config command"""			,				description=UpperCamelCase__    )

      parser.add_argument(
          """--config_file"""			,				default=UpperCamelCase__			,				help=(
              """The path to use to store the config file. Will default to a file named default_config.yaml in the cache """
              """location, which is the content of the environment `HF_HOME` suffixed with 'accelerate', or if you don't have """
              """such an environment variable, your cache directory ('~/.cache' or the content of `XDG_CACHE_HOME`) suffixed """
              """with 'huggingface'."""
          )			,				)

      if subparsers is not None:
            parser.set_defaults(func=UpperCamelCase__    )
      return parser




def   snake_case		(     UpperCamelCase__       :						Union[str, Any]    )   ->      Optional[Any]:
      lowerCamelCase					:					Dict        =	get_user_input()
      if args.config_file is not None:
            lowerCamelCase					:					Tuple        =	args.config_file
      else:
            if not os.path.isdir(UpperCamelCase__    ):
                  os.makedirs(UpperCamelCase__    )
            lowerCamelCase					:					str        =	default_yaml_config_file

      if config_file.endswith(""".json"""    ):
            config.to_json_file(UpperCamelCase__    )
      else:
            config.to_yaml_file(UpperCamelCase__    )
      print(F'accelerate configuration saved at {config_file}'    )




def   snake_case		(     )   ->      Tuple:
      lowerCamelCase					:					List[str]        =	config_command_parser()
      lowerCamelCase					:					List[str]        =	parser.parse_args()
      config_command(UpperCamelCase__    )


if __name__ == "__main__":
    main()








"""simple docstring"""


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


__lowerCamelCase      :Any	    =		False


@skip_mps
class 			A__	(						__lowercase ,     __lowercase ,     __lowercase ,     unittest.TestCase):


     """simple docstring"""
     snake_case__	:					Optional[Any]													=StableDiffusionAttendAndExcitePipeline
     snake_case__	:					Any													=False
     snake_case__	:					Dict													=TEXT_TO_IMAGE_PARAMS
     snake_case__	:					Any													=TEXT_TO_IMAGE_BATCH_PARAMS.union({'''token_indices'''})
     snake_case__	:					Dict													=TEXT_TO_IMAGE_IMAGE_PARAMS
     snake_case__	:					str													=TEXT_TO_IMAGE_IMAGE_PARAMS

     @classmethod
     def 		a__						(	cls:						Dict	)->    Tuple:
           super().setUpClass()
           torch.use_deterministic_algorithms(__a	)

     @classmethod
     def 		a__						(	cls:						Union[str, Any]	)->    Any:
           super().tearDownClass()
           torch.use_deterministic_algorithms(__a	)

     def 		a__						(	self:						Tuple	)->    Union[str, Any]:
           torch.manual_seed(0	)
           lowerCamelCase					:					str        =	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=__a     ,		)
           lowerCamelCase					:					Union[str, Any]        =	DDIMScheduler(
               beta_start=0.0_00_85     ,		beta_end=0.0_12     ,		beta_schedule="""scaled_linear"""     ,		clip_sample=__a     ,		set_alpha_to_one=__a     ,		)
           torch.manual_seed(0	)
           lowerCamelCase					:					Union[str, Any]        =	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=128     ,		)
           torch.manual_seed(0	)
           lowerCamelCase					:					str        =	CLIPTextConfig(
               bos_token_id=0     ,		eos_token_id=2     ,		hidden_size=32     ,		intermediate_size=37     ,		layer_norm_eps=1e-0_5     ,		num_attention_heads=4     ,		num_hidden_layers=5     ,		pad_token_id=1     ,		vocab_size=1_000     ,		hidden_act="""gelu"""     ,		projection_dim=512     ,		)
           lowerCamelCase					:					Optional[int]        =	CLIPTextModel(__a	)
           lowerCamelCase					:					str        =	CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""	)

           lowerCamelCase					:					List[str]        =	{
               """unet""": unet,
               """scheduler""": scheduler,
               """vae""": vae,
               """text_encoder""": text_encoder,
               """tokenizer""": tokenizer,
               """safety_checker""": None,
               """feature_extractor""": None,
           }

           return components

     def 		a__						(	self:						Tuple     ,		__a:						int     ,		__a:						Union[str, Any]=0	)->    Optional[Any]:
           if str(__a	).startswith("""mps"""	):
                 lowerCamelCase					:					Tuple        =	torch.manual_seed(__a	)
           else:
                 lowerCamelCase					:					str        =	torch.Generator(device=__a	).manual_seed(__a	)
           lowerCamelCase					:					Dict        =	{
               """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:						Dict	)->    str:
           lowerCamelCase					:					Tuple        =	"""cpu"""

           lowerCamelCase					:					List[str]        =	self.get_dummy_components()
           lowerCamelCase					:					List[Any]        =	self.pipeline_class(**__a	)
           pipe.to(__a	)
           pipe.set_progress_bar_config(disable=__a	)

           lowerCamelCase					:					Any        =	self.get_dummy_inputs(__a	)
           lowerCamelCase					:					Union[str, Any]        =	pipe(**__a	).images
           lowerCamelCase					:					Tuple        =	image[0, -3:, -3:, -1]

           self.assertEqual(image.shape     ,		(1, 64, 64, 3)	)
           lowerCamelCase					:					Optional[Any]        =	np.array(
               [0.63_90_53_64, 0.62_89_73_07, 0.48_59_90_17, 0.5_13_36_24, 0.5_55_00_48, 0.45_76_95_16, 0.50_32_69_73, 0.5_02_31_39, 0.45_38_44_96]	)
           lowerCamelCase					:					Optional[Any]        =	np.abs(image_slice.flatten() - expected_slice	).max()
           self.assertLessEqual(__a     ,		1e-3	)

     def 		a__						(	self:						int	)->    Optional[Any]:
           super().test_cpu_offload_forward_pass(expected_max_diff=5e-4	)

     def 		a__						(	self:						Union[str, Any]	)->    Optional[int]:
           # NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
           self._test_inference_batch_consistent(batch_sizes=[1, 2]	)

     def 		a__						(	self:						Tuple	)->    int:
           self._test_inference_batch_single_identical(batch_size=2     ,		expected_max_diff=7e-4	)

     def 		a__						(	self:						Dict	)->    List[Any]:
           super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3	)

     def 		a__						(	self:						Optional[int]	)->    Dict:
           super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5e-4	)

     def 		a__						(	self:						Any	)->    Tuple:
           super().test_save_load_local(expected_max_difference=5e-4	)







     def 		a__						(	self:						str	)->    str:
           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:						Any	)->    Tuple:
           super().setUpClass()
           torch.use_deterministic_algorithms(__a	)

     @classmethod
     def 		a__						(	cls:						Dict	)->    Optional[int]:
           super().tearDownClass()
           torch.use_deterministic_algorithms(__a	)

     def 		a__						(	self:						int	)->    Optional[int]:
           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()







     def 		a__						(	self:						int	)->    Optional[Any]:
           lowerCamelCase					:					List[Any]        =	torch.manual_seed(51	)

           lowerCamelCase					:					List[str]        =	StableDiffusionAttendAndExcitePipeline.from_pretrained(
               """CompVis/stable-diffusion-v1-4"""     ,		safety_checker=__a     ,		torch_dtype=torch.floataa	)
           pipe.to("""cuda"""	)

           lowerCamelCase					:					Dict        =	"""a painting of an elephant with glasses"""
           lowerCamelCase					:					Any        =	[5, 7]

           lowerCamelCase					:					Tuple        =	pipe(
               prompt=__a     ,		token_indices=__a     ,		guidance_scale=7.5     ,		generator=__a     ,		num_inference_steps=5     ,		max_iter_to_alter=5     ,		output_type="""numpy"""     ,		).images[0]

           lowerCamelCase					:					Union[str, Any]        =	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




