

import unittest

import numpy as np

from diffusers import LMSDiscreteScheduler, OnnxStableDiffusionInpaintPipeline
from diffusers.utils.testing_utils import (
    is_onnx_available,
    load_image,
    nightly,
    require_onnxruntime,
    require_torch_gpu,
)

from ..test_pipelines_onnx_common import OnnxPipelineTesterMixin


if is_onnx_available():
     import onnxruntime as ort


class 			A						(							UpperCAmelCase_   ,							unittest.TestCase			):







    """simple docstring"""





    pass




@nightly
@require_onnxruntime
@require_torch_gpu
class 			A						(							unittest.TestCase			):







    """simple docstring"""





    @property
    def 						snake_case__					(				self			:	Tuple		)->   str:





        '''simple docstring'''







        return (
            "CUDAExecutionProvider",
            {
                "gpu_mem_limit": "15000000000",  # 15GB
                "arena_extend_strategy": "kSameAsRequested",
            },
        )


    @property
    def 						snake_case__					(				self			:	Optional[int]		)->   List[str]:





        '''simple docstring'''







        A__   				=       ort.SessionOptions()
        A__   				=       False
        return options


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





        '''simple docstring'''







        A__   				=       load_image(
            'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
            '/in_paint/overture-creations-5sI6fQgYIuo.png'		)
        A__   				=       load_image(
            'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
            '/in_paint/overture-creations-5sI6fQgYIuo_mask.png'		)
        A__   				=       OnnxStableDiffusionInpaintPipeline.from_pretrained(
            'runwayml/stable-diffusion-inpainting',revision='onnx',safety_checker=__lowercase,feature_extractor=__lowercase,provider=self.gpu_provider,sess_options=self.gpu_options,)
        pipe.set_progress_bar_config(disable=__lowercase		)

        A__   				=       '''A red cat sitting on a park bench'''

        A__   				=       np.random.RandomState(0		)
        A__   				=       pipe(
            prompt=__lowercase,image=__lowercase,mask_image=__lowercase,guidance_scale=7.5,num_inference_steps=1_0,generator=__lowercase,output_type='np',)
        A__   				=       output.images
        A__   				=       images[0, 2_5_5:2_5_8, 2_5_5:2_5_8, -1]

        assert images.shape == (1, 5_1_2, 5_1_2, 3)
        A__   				=       np.array([0.2_514, 0.3_007, 0.3_517, 0.1_790, 0.2_382, 0.3_167, 0.1_944, 0.2_273, 0.2_464]		)

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


    def 						snake_case__					(				self			:	Dict		)->   List[Any]:





        '''simple docstring'''







        A__   				=       load_image(
            'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
            '/in_paint/overture-creations-5sI6fQgYIuo.png'		)
        A__   				=       load_image(
            'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
            '/in_paint/overture-creations-5sI6fQgYIuo_mask.png'		)
        A__   				=       LMSDiscreteScheduler.from_pretrained(
            'runwayml/stable-diffusion-inpainting',subfolder='scheduler',revision='onnx'		)
        A__   				=       OnnxStableDiffusionInpaintPipeline.from_pretrained(
            'runwayml/stable-diffusion-inpainting',revision='onnx',scheduler=__lowercase,safety_checker=__lowercase,feature_extractor=__lowercase,provider=self.gpu_provider,sess_options=self.gpu_options,)
        pipe.set_progress_bar_config(disable=__lowercase		)

        A__   				=       '''A red cat sitting on a park bench'''

        A__   				=       np.random.RandomState(0		)
        A__   				=       pipe(
            prompt=__lowercase,image=__lowercase,mask_image=__lowercase,guidance_scale=7.5,num_inference_steps=2_0,generator=__lowercase,output_type='np',)
        A__   				=       output.images
        A__   				=       images[0, 2_5_5:2_5_8, 2_5_5:2_5_8, -1]

        assert images.shape == (1, 5_1_2, 5_1_2, 3)
        A__   				=       np.array([0.0_086, 0.0_077, 0.0_083, 0.0_093, 0.0_107, 0.0_139, 0.0_094, 0.0_097, 0.0_125]		)

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




import os
import unittest
from tempfile import TemporaryDirectory

import torch
import torch.nn as nn

from accelerate.utils import (
    OffloadedWeightsLoader,
    extract_submodules_state_dict,
    load_offloaded_weight,
    offload_state_dict,
    offload_weight,
)


class 			A						(							nn.Module			):







    """simple docstring"""





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





        '''simple docstring'''







        super().__init__()
        A__   				=       nn.Linear(3,4		)
        A__   				=       nn.BatchNormad(4		)
        A__   				=       nn.Linear(4,5		)


    def 						snake_case__					(				self			:	Dict,lowercase_			:	Union[str, Any]		)->   Tuple:





        '''simple docstring'''







        return self.lineara(self.batchnorm(self.lineara(lowercase_		)		)		)




class 			A						(							unittest.TestCase			):







    """simple docstring"""





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





        '''simple docstring'''







        A__   				=       ModelForTest()
        with TemporaryDirectory() as tmp_dir:
            offload_state_dict(lowercase_,model.state_dict()		)
            A__   				=       os.path.join(lowercase_,'index.json'		)
            self.assertTrue(os.path.isfile(lowercase_		)		)
            # TODO: add tests on what is inside the index

            for key in ["linear1.weight", "linear1.bias", "linear2.weight", "linear2.bias"]:
                A__   				=       os.path.join(lowercase_,F'{key}.dat'		)
                self.assertTrue(os.path.isfile(lowercase_		)		)
                # TODO: add tests on the fact weights are properly loaded


    def 						snake_case__					(				self			:	List[Any]		)->   Optional[int]:





        '''simple docstring'''







        A__   				=       [torch.floataa, torch.floataa, torch.bfloataa]

        for dtype in dtypes:
            A__   				=       torch.randn(2,3,dtype=lowercase_		)
            with TemporaryDirectory() as tmp_dir:
                A__   				=       offload_weight(lowercase_,'weight',lowercase_,{}		)
                A__   				=       os.path.join(lowercase_,'weight.dat'		)
                self.assertTrue(os.path.isfile(lowercase_		)		)
                self.assertDictEqual(lowercase_,{'weight': {'shape': [2, 3], 'dtype': str(lowercase_		).split('.'		)[1]}}		)

                A__   				=       load_offloaded_weight(lowercase_,index['weight']		)
                self.assertTrue(torch.equal(lowercase_,lowercase_		)		)


    def 						snake_case__					(				self			:	Optional[int]		)->   Optional[Any]:





        '''simple docstring'''







        A__   				=       ModelForTest()
        A__   				=       model.state_dict()
        A__   				=       {k: v for k, v in state_dict.items() if 'linear2' not in k}
        A__   				=       {k: v for k, v in state_dict.items() if 'linear2' in k}

        with TemporaryDirectory() as tmp_dir:
            offload_state_dict(lowercase_,lowercase_		)
            A__   				=       OffloadedWeightsLoader(state_dict=lowercase_,save_folder=lowercase_		)

            # Every key is there with the right value
            self.assertEqual(sorted(lowercase_		),sorted(state_dict.keys()		)		)
            for key, param in state_dict.items():
                self.assertTrue(torch.allclose(lowercase_,weight_map[key]		)		)

        A__   				=       {k: v for k, v in state_dict.items() if 'weight' in k}
        A__   				=       {k: v for k, v in state_dict.items() if 'weight' not in k}

        with TemporaryDirectory() as tmp_dir:
            offload_state_dict(lowercase_,lowercase_		)
            A__   				=       OffloadedWeightsLoader(state_dict=lowercase_,save_folder=lowercase_		)

            # Every key is there with the right value
            self.assertEqual(sorted(lowercase_		),sorted(state_dict.keys()		)		)
            for key, param in state_dict.items():
                self.assertTrue(torch.allclose(lowercase_,weight_map[key]		)		)

        with TemporaryDirectory() as tmp_dir:
            offload_state_dict(lowercase_,lowercase_		)
            # Duplicates are removed
            A__   				=       OffloadedWeightsLoader(state_dict=lowercase_,save_folder=lowercase_		)

            # Every key is there with the right value
            self.assertEqual(sorted(lowercase_		),sorted(state_dict.keys()		)		)
            for key, param in state_dict.items():
                self.assertTrue(torch.allclose(lowercase_,weight_map[key]		)		)


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





        '''simple docstring'''







        A__   				=       {'a.1': 0, 'a.10': 1, 'a.2': 2}
        A__   				=       extract_submodules_state_dict(lowercase_,['a.1', 'a.2']		)
        self.assertDictEqual(lowercase_,{'a.1': 0, 'a.2': 2}		)

        A__   				=       {'a.1.a': 0, 'a.10.a': 1, 'a.2.a': 2}
        A__   				=       extract_submodules_state_dict(lowercase_,['a.1', 'a.2']		)
        self.assertDictEqual(lowercase_,{'a.1.a': 0, 'a.2.a': 2}		)

