



'''simple docstring'''





import gc
import random
import unittest

import numpy as np
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModel,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import AutoencoderKL, DDIMScheduler, DDPMScheduler, StableUnCLIPImgaImgPipeline, UNetaDConditionModel
from diffusers.pipelines.pipeline_utils import DiffusionPipeline
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    floats_tensor,
    load_image,
    load_numpy,
    require_torch_gpu,
    skip_mps,
    slow,
    torch_device,
)

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()


class 				snake_case (UpperCamelCase		,      UpperCamelCase		,      UpperCamelCase		,      unittest.TestCase   ):
       lowerCAmelCase__  :str						   =					StableUnCLIPImgaImgPipeline
       lowerCAmelCase__  :Union[str, Any]						   =					TEXT_GUIDED_IMAGE_VARIATION_PARAMS
       lowerCAmelCase__  :str						   =					TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
       lowerCAmelCase__  :Tuple						   =					frozenset(
           []   )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
       lowerCAmelCase__  :Dict						   =					frozenset([]   )



       def 							_a      (				self      )  ->							Union[str, Any]:
           lowercase__		       =						32
           lowercase__		       =						embedder_hidden_size

           # image encoding components

           lowercase__		       =						CLIPImageProcessor(crop_size=32     ,size=32      )

           torch.manual_seed(0      )
           lowercase__		       =						CLIPVisionModelWithProjection(
               CLIPVisionConfig(
                   hidden_size=UpperCAmelCase_     ,projection_dim=UpperCAmelCase_     ,num_hidden_layers=5     ,num_attention_heads=4     ,image_size=32     ,intermediate_size=37     ,patch_size=1     ,)      )

           # regular denoising components

           torch.manual_seed(0      )
           lowercase__		       =						StableUnCLIPImageNormalizer(embedding_dim=UpperCAmelCase_      )
           lowercase__		       =						DDPMScheduler(beta_schedule="squaredcos_cap_v2"      )

           torch.manual_seed(0      )
           lowercase__		       =						CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"      )

           torch.manual_seed(0      )
           lowercase__		       =						CLIPTextModel(
               CLIPTextConfig(
                   bos_token_id=0     ,eos_token_id=2     ,hidden_size=UpperCAmelCase_     ,projection_dim=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     ,)      )

           torch.manual_seed(0      )
           lowercase__		       =						UNetaDConditionModel(
               sample_size=32     ,in_channels=4     ,out_channels=4     ,down_block_types=("CrossAttnDownBlock2D", "DownBlock2D")     ,up_block_types=("UpBlock2D", "CrossAttnUpBlock2D")     ,block_out_channels=(32, 64)     ,attention_head_dim=(2, 4)     ,class_embed_type="projection"     ,projection_class_embeddings_input_dim=embedder_projection_dim * 2     ,cross_attention_dim=UpperCAmelCase_     ,layers_per_block=1     ,upcast_attention=UpperCAmelCase_     ,use_linear_projection=UpperCAmelCase_     ,)

           torch.manual_seed(0      )
           lowercase__		       =						DDIMScheduler(
               beta_schedule="scaled_linear"     ,beta_start=0.0_00_85     ,beta_end=0.0_12     ,prediction_type="v_prediction"     ,set_alpha_to_one=UpperCAmelCase_     ,steps_offset=1     ,)

           torch.manual_seed(0      )
           lowercase__		       =						AutoencoderKL()

           lowercase__		       =						{
               # image encoding components
               "feature_extractor": feature_extractor,
               "image_encoder": image_encoder.eval(),
               # image noising components
               "image_normalizer": image_normalizer.eval(),
               "image_noising_scheduler": image_noising_scheduler,
               # regular denoising components
               "tokenizer": tokenizer,
               "text_encoder": text_encoder.eval(),
               "unet": unet.eval(),
               "scheduler": scheduler,
               "vae": vae.eval(),
           }

           return components



       def 							_a      (				self     ,UpperCAmelCase_     ,UpperCAmelCase_=0     ,UpperCAmelCase_=True      )  ->							int:
           if str(UpperCAmelCase_      ).startswith("mps"      ):
               lowercase__		       =						torch.manual_seed(UpperCAmelCase_      )
           else:
               lowercase__		       =						torch.Generator(device=UpperCAmelCase_      ).manual_seed(UpperCAmelCase_      )

           lowercase__		       =						floats_tensor((1, 3, 32, 32)     ,rng=random.Random(UpperCAmelCase_      )      ).to(UpperCAmelCase_      )

           if pil_image:
               lowercase__		       =						input_image * 0.5 + 0.5
               lowercase__		       =						input_image.clamp(0     ,1      )
               lowercase__		       =						input_image.cpu().permute(0     ,2     ,3     ,1      ).float().numpy()
               lowercase__		       =						DiffusionPipeline.numpy_to_pil(UpperCAmelCase_      )[0]

           return {
               "prompt": "An anime racoon running a marathon",
               "image": input_image,
               "generator": generator,
               "num_inference_steps": 2,
               "output_type": "np",
           }



       @skip_mps
       def 							_a      (				self      )  ->							Any:
           lowercase__		       =						"cpu"  # ensure determinism for the device-dependent torch.Generator
           lowercase__		       =						self.get_dummy_components()
           lowercase__		       =						StableUnCLIPImgaImgPipeline(**UpperCAmelCase_      )
           lowercase__		       =						sd_pipe.to(UpperCAmelCase_      )
           sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_      )

           lowercase__		       =						self.get_dummy_inputs(UpperCAmelCase_      )
           inputs.update({"image_embeds": None}      )
           lowercase__		       =						sd_pipe(**UpperCAmelCase_      ).images
           lowercase__		       =						image[0, -3:, -3:, -1]

           assert image.shape == (1, 32, 32, 3)
           lowercase__		       =						np.array([0.38_72, 0.72_24, 0.56_01, 0.47_41, 0.68_72, 0.58_14, 0.46_36, 0.38_67, 0.50_78]      )

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



       def 							_a      (				self      )  ->							Union[str, Any]:
           lowercase__		       =						torch_device in ["cpu", "mps"]

           self._test_attention_slicing_forward_pass(test_max_difference=UpperCAmelCase_      )



       def 							_a      (				self      )  ->							List[Any]:
           lowercase__		       =						torch_device in ["cpu", "mps"]

           self._test_inference_batch_single_identical(test_max_difference=UpperCAmelCase_      )







       @unittest.skipIf(
           torch_device != "cuda" or not is_xformers_available()     ,reason="XFormers attention is only available with CUDA and `xformers` installed"     ,)
       def 							_a      (				self      )  ->							str:
           self._test_xformers_attention_forwardGenerator_pass(test_max_difference=UpperCAmelCase_      )





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



       def 							_a      (				self      )  ->							int:
           # clean up the VRAM after each test
           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()



       def 							_a      (				self      )  ->							List[str]:
           lowercase__		       =						load_image(
               "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"      )

           lowercase__		       =						load_numpy(
               "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_img2img_anime_turtle_fp16.npy"      )

           lowercase__		       =						StableUnCLIPImgaImgPipeline.from_pretrained(
               "fusing/stable-unclip-2-1-l-img2img"     ,torch_dtype=torch.floataa      )
           pipe.to(UpperCAmelCase_      )
           pipe.set_progress_bar_config(disable=UpperCAmelCase_      )
           # stable unclip will oom when integration tests are run on a V100,
           # so turn on memory savings
           pipe.enable_attention_slicing()
           pipe.enable_sequential_cpu_offload()

           lowercase__		       =						torch.Generator(device="cpu"      ).manual_seed(0      )
           lowercase__		       =						pipe(UpperCAmelCase_     ,"anime turle"     ,generator=UpperCAmelCase_     ,output_type="np"      )

           lowercase__		       =						output.images[0]

           assert image.shape == (768, 768, 3)

           assert_mean_pixel_difference(UpperCAmelCase_     ,UpperCAmelCase_      )



       def 							_a      (				self      )  ->							str:
           lowercase__		       =						load_image(
               "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"      )

           lowercase__		       =						load_numpy(
               "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_h_img2img_anime_turtle_fp16.npy"      )

           lowercase__		       =						StableUnCLIPImgaImgPipeline.from_pretrained(
               "fusing/stable-unclip-2-1-h-img2img"     ,torch_dtype=torch.floataa      )
           pipe.to(UpperCAmelCase_      )
           pipe.set_progress_bar_config(disable=UpperCAmelCase_      )
           # stable unclip will oom when integration tests are run on a V100,
           # so turn on memory savings
           pipe.enable_attention_slicing()
           pipe.enable_sequential_cpu_offload()

           lowercase__		       =						torch.Generator(device="cpu"      ).manual_seed(0      )
           lowercase__		       =						pipe(UpperCAmelCase_     ,"anime turle"     ,generator=UpperCAmelCase_     ,output_type="np"      )

           lowercase__		       =						output.images[0]

           assert image.shape == (768, 768, 3)

           assert_mean_pixel_difference(UpperCAmelCase_     ,UpperCAmelCase_      )







       def 							_a      (				self      )  ->							List[str]:
           lowercase__		       =						load_image(
               "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"      )

           torch.cuda.empty_cache()
           torch.cuda.reset_max_memory_allocated()
           torch.cuda.reset_peak_memory_stats()

           lowercase__		       =						StableUnCLIPImgaImgPipeline.from_pretrained(
               "fusing/stable-unclip-2-1-h-img2img"     ,torch_dtype=torch.floataa      )
           lowercase__		       =						pipe.to(UpperCAmelCase_      )
           pipe.set_progress_bar_config(disable=UpperCAmelCase_      )
           pipe.enable_attention_slicing()
           pipe.enable_sequential_cpu_offload()

           lowercase__		       =						pipe(
               UpperCAmelCase_     ,"anime turtle"     ,num_inference_steps=2     ,output_type="np"     ,)

           lowercase__		       =						torch.cuda.max_memory_allocated()
           # make sure that less than 7 GB is allocated
           assert mem_bytes < 7 * 10**9









'''simple docstring'''





# Copyright 2022 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
import platform

import numpy as np
import psutil
import torch

from accelerate import __version__ as version
from accelerate.commands.config import default_config_file, load_config_from_file

from ..utils import is_npu_available, is_xpu_available




def        lowerCamelCase  (				_snake_case  :						str=None       ):





    '''simple docstring'''


    if subparsers is not None:
        lowercase__		       =						subparsers.add_parser("env"       )
    else:
        lowercase__		       =						argparse.ArgumentParser("Accelerate env command"       )

    parser.add_argument(
        "--config_file"     ,default=_snake_case     ,help="The config file to use for the default values in the launching script."       )

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




def        lowerCamelCase  (				_snake_case  :						Tuple       ):





    '''simple docstring'''


    lowercase__		       =						torch.__version__
    lowercase__		       =						torch.cuda.is_available()
    lowercase__		       =						is_xpu_available()
    lowercase__		       =						is_npu_available()

    lowercase__		       =						"Not found"
    # Get the default from the config file.
    if args.config_file is not None or os.path.isfile(_snake_case       ):
        lowercase__		       =						load_config_from_file(args.config_file       ).to_dict()

    lowercase__		       =						{
        "`Accelerate` version": version,
        "Platform": platform.platform(),
        "Python version": platform.python_version(),
        "Numpy version": np.__version__,
        "PyTorch version (GPU?)": f'''{pt_version} ({pt_cuda_available})''',
        "PyTorch XPU available": str(_snake_case       ),
        "PyTorch NPU available": str(_snake_case       ),
        "System RAM": f'''{psutil.virtual_memory().total / 1_024 ** 3:.2f} GB''',
    }
    if pt_cuda_available:
        lowercase__		       =						torch.cuda.get_device_name()

    print("\nCopy-and-paste the text below in your GitHub issue\n"       )
    print("\n".join([f'''- {prop}: {val}''' for prop, val in info.items()]       )       )

    print("- `Accelerate` default config:" if args.config_file is None else "- `Accelerate` config passed:"       )
    lowercase__		       =						(
        "\n".join([f'''\t- {prop}: {val}''' for prop, val in accelerate_config.items()]       )
        if isinstance(_snake_case     ,_snake_case       )
        else f'''\t{accelerate_config}'''
    )
    print(_snake_case       )

    lowercase__		       =						accelerate_config

    return info




def        lowerCamelCase  (				):





    '''simple docstring'''


    lowercase__		       =						env_command_parser()
    lowercase__		       =						parser.parse_args()
    env_command(_snake_case       )
    return 0


if __name__ == "__main__":
 raise SystemExit(main())




