





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    ):


				def __init__(		self							):
								super().__init__()
								lowerCamelCase	 =nn.Linear(3   , 4							)
								lowerCamelCase	 =nn.BatchNormad(4							)
								lowerCamelCase	 =nn.Linear(4   , 5							)


				def _snake_case       (		self   , UpperCAmelCase_							):
								return self.lineara(self.batchnorm(self.lineara(_SCREAMING_SNAKE_CASE							)							)							)




class       __A     (						unittest.TestCase    ):


				def _snake_case       (		self							):
								lowerCamelCase	 =ModelForTest()
								with TemporaryDirectory() as tmp_dir:
												offload_state_dict(_SCREAMING_SNAKE_CASE   , model.state_dict()							)
												lowerCamelCase	 =os.path.join(_SCREAMING_SNAKE_CASE   , """index.json"""							)
												self.assertTrue(os.path.isfile(_SCREAMING_SNAKE_CASE							)							)
												# TODO: add tests on what is inside the index

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


				def _snake_case       (		self							):
								lowerCamelCase	 =[torch.floataa, torch.floataa, torch.bfloataa]

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

																lowerCamelCase	 =load_offloaded_weight(_SCREAMING_SNAKE_CASE   , index["""weight"""]							)
																self.assertTrue(torch.equal(_SCREAMING_SNAKE_CASE   , _SCREAMING_SNAKE_CASE							)							)


				def _snake_case       (		self							):
								lowerCamelCase	 =ModelForTest()
								lowerCamelCase	 =model.state_dict()
								lowerCamelCase	 ={k: v for k, v in state_dict.items() if """linear2""" not in k}
								lowerCamelCase	 ={k: v for k, v in state_dict.items() if """linear2""" in k}

								with TemporaryDirectory() as tmp_dir:
												offload_state_dict(_SCREAMING_SNAKE_CASE   , _SCREAMING_SNAKE_CASE							)
												lowerCamelCase	 =OffloadedWeightsLoader(state_dict=_SCREAMING_SNAKE_CASE   , save_folder=_SCREAMING_SNAKE_CASE							)

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

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

								with TemporaryDirectory() as tmp_dir:
												offload_state_dict(_SCREAMING_SNAKE_CASE   , _SCREAMING_SNAKE_CASE							)
												lowerCamelCase	 =OffloadedWeightsLoader(state_dict=_SCREAMING_SNAKE_CASE   , save_folder=_SCREAMING_SNAKE_CASE							)

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

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

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


				def _snake_case       (		self							):
								lowerCamelCase	 ={"""a.1""": 0, """a.10""": 1, """a.2""": 2}
								lowerCamelCase	 =extract_submodules_state_dict(_SCREAMING_SNAKE_CASE   , ["""a.1""", """a.2"""]							)
								self.assertDictEqual(_SCREAMING_SNAKE_CASE   , {"""a.1""": 0, """a.2""": 2}							)

								lowerCamelCase	 ={"""a.1.a""": 0, """a.10.a""": 1, """a.2.a""": 2}
								lowerCamelCase	 =extract_submodules_state_dict(_SCREAMING_SNAKE_CASE   , ["""a.1""", """a.2"""]							)
								self.assertDictEqual(_SCREAMING_SNAKE_CASE   , {"""a.1.a""": 0, """a.2.a""": 2}							)












from ...utils import (
    OptionalDependencyNotAvailable,
    is_torch_available,
    is_transformers_available,
    is_transformers_version,
)


try:
							if not (is_transformers_available() and is_torch_available()):
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							from ...utils.dummy_torch_and_transformers_objects import KandinskyPipeline, KandinskyPriorPipeline
else:
							from .pipeline_kandinsky import KandinskyPipeline
							from .pipeline_kandinsky_imgaimg import KandinskyImgaImgPipeline
							from .pipeline_kandinsky_inpaint import KandinskyInpaintPipeline
							from .pipeline_kandinsky_prior import KandinskyPriorPipeline, KandinskyPriorPipelineOutput
							from .text_encoder import MultilingualCLIP





