

"""simple docstring"""



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






									"""simple docstring"""


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





				def 				UpperCamelCase	(     self: Union[str, Any]    ,						UpperCamelCase: List[Any]       ):






									"""simple docstring"""


									return self.lineara(self.batchnorm(self.lineara(UpperCamelCase       )       )       )



class 					a						(       unittest.TestCase				):



				"""simple docstring"""



				def 				UpperCamelCase	(     self: Union[str, Any]       ):






									"""simple docstring"""


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

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

				def 				UpperCamelCase	(     self: Optional[int]       ):






									"""simple docstring"""


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

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

																			A__										=  load_offloaded_weight(UpperCamelCase    ,						index["""weight"""]       )
																			self.assertTrue(torch.equal(UpperCamelCase    ,						UpperCamelCase       )       )

				def 				UpperCamelCase	(     self: Optional[int]       ):






									"""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(UpperCamelCase    ,						UpperCamelCase       )
														A__										=  OffloadedWeightsLoader(state_dict=UpperCamelCase    ,						save_folder=UpperCamelCase       )

														# Every key is there with the right value
														self.assertEqual(sorted(UpperCamelCase       )    ,						sorted(state_dict.keys()       )       )
														for key, param in state_dict.items():
																			self.assertTrue(torch.allclose(UpperCamelCase    ,						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(UpperCamelCase    ,						UpperCamelCase       )
														A__										=  OffloadedWeightsLoader(state_dict=UpperCamelCase    ,						save_folder=UpperCamelCase       )

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

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

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





				def 				UpperCamelCase	(     self: str       ):






									"""simple docstring"""


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

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









"""simple docstring"""



# Copyright 2023 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.
from typing import TYPE_CHECKING

# rely on isort to merge the imports
from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tokenizers_available, is_torch_available


SCREAMING_SNAKE_CASE_					:	List[str]							  =						{'configuration_mra': ['MRA_PRETRAINED_CONFIG_ARCHIVE_MAP', 'MraConfig']}

try:
 if not is_torch_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 SCREAMING_SNAKE_CASE_					:	List[Any]							  =						[
     'MRA_PRETRAINED_MODEL_ARCHIVE_LIST',
     'MraForMaskedLM',
     'MraForMultipleChoice',
     'MraForQuestionAnswering',
     'MraForSequenceClassification',
     'MraForTokenClassification',
     'MraLayer',
     'MraModel',
     'MraPreTrainedModel',
 ]


if TYPE_CHECKING:
 from .configuration_mra import MRA_PRETRAINED_CONFIG_ARCHIVE_MAP, MraConfig

 try:
  if not is_torch_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_mra import (
      MRA_PRETRAINED_MODEL_ARCHIVE_LIST,
      MraForMaskedLM,
      MraForMultipleChoice,
      MraForQuestionAnswering,
      MraForSequenceClassification,
      MraForTokenClassification,
      MraLayer,
      MraModel,
      MraPreTrainedModel,
  )
else:
 import sys

 SCREAMING_SNAKE_CASE_					:	List[str]							  =						_LazyModule(__name__, globals()['__file__'], _import_structure)






