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					:						Optional[Any]		)						->    List[Any]:
         """simple docstring"""



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





   def       lowercase_				(self					:						List[Any]	,  __UpperCAmelCase					:						Dict		)						->    Union[str, Any]:
         """simple docstring"""



         return self.lineara(self.batchnorm(self.lineara(__UpperCAmelCase		)		)		)





class 					A							(    unittest.TestCase					):


   def       lowercase_				(self					:						Optional[int]		)						->    Optional[int]:
         """simple docstring"""



         UpperCAmelCase__								=  ModelForTest()
         with TemporaryDirectory() as tmp_dir:
               offload_state_dict(__UpperCAmelCase	,  model.state_dict()		)
               UpperCAmelCase__								=  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"]:
                     UpperCAmelCase__								=  os.path.join(__UpperCAmelCase	,  f"""{key}.dat"""		)
                     self.assertTrue(os.path.isfile(__UpperCAmelCase		)		)
                     # TODO: add tests on the fact weights are properly loaded


   def       lowercase_				(self					:						Any		)						->    Union[str, Any]:
         """simple docstring"""



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

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

                     UpperCAmelCase__								=  load_offloaded_weight(__UpperCAmelCase	,  index["weight"]		)
                     self.assertTrue(torch.equal(__UpperCAmelCase	,  __UpperCAmelCase		)		)


   def       lowercase_				(self					:						Tuple		)						->    Optional[Any]:
         """simple docstring"""



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

         with TemporaryDirectory() as tmp_dir:
               offload_state_dict(__UpperCAmelCase	,  __UpperCAmelCase		)
               UpperCAmelCase__								=  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]		)		)

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

         with TemporaryDirectory() as tmp_dir:
               offload_state_dict(__UpperCAmelCase	,  __UpperCAmelCase		)
               UpperCAmelCase__								=  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
               UpperCAmelCase__								=  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       lowercase_				(self					:						Optional[int]		)						->    Union[str, Any]:
         """simple docstring"""



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

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




import os
import zipfile

import pytest

from datasets.utils.extract import (
    BzipaExtractor,
    Extractor,
    GzipExtractor,
    LzaExtractor,
    SevenZipExtractor,
    TarExtractor,
    XzExtractor,
    ZipExtractor,
    ZstdExtractor,
)

from .utils import require_lza, require_pyazr, require_zstandard




@pytest.mark.parametrize(
    "compression_format, is_archive",  [
        ("7z", True),
        ("bz2", False),
        ("gzip", False),
        ("lz4", False),
        ("tar", True),
        ("xz", False),
        ("zip", True),
        ("zstd", False),
    ],  )
def        lowerCAmelCase_			(					__A,  __A,  __A,  __A,  __A,  __A,  __A,  __A,  __A,  __A,  __A,  __A,  )					->    Any:







      '''simple docstring'''


      UpperCAmelCase__								=  {
          "7z": (seven_zip_file, SevenZipExtractor),
          "bz2": (bza_file, BzipaExtractor),
          "gzip": (gz_file, GzipExtractor),
          "lz4": (lza_file, LzaExtractor),
          "tar": (tar_file, TarExtractor),
          "xz": (xz_file, XzExtractor),
          "zip": (zip_file, ZipExtractor),
          "zstd": (zstd_file, ZstdExtractor),
      }
      UpperCAmelCase__		,				UpperCAmelCase__								=  input_paths_and_base_extractors[compression_format]
      if input_path is None:
            UpperCAmelCase__								=  f"""for '{compression_format}' compression_format, """
            if compression_format == "7z":
                  reason += require_pyazr.kwargs["reason"]
            elif compression_format == "lz4":
                  reason += require_lza.kwargs["reason"]
            elif compression_format == "zstd":
                  reason += require_zstandard.kwargs["reason"]
            pytest.skip(__A							)
      assert base_extractor.is_extractable(__A							)
      UpperCAmelCase__								=  tmp_path / ("extracted" if is_archive else "extracted.txt")
      base_extractor.extract(__A,  __A							)
      if is_archive:
            assert output_path.is_dir()
            for file_path in output_path.iterdir():
                  assert file_path.name == text_file.name
                  UpperCAmelCase__								=  file_path.read_text(encoding="utf-8"							)
      else:
            UpperCAmelCase__								=  output_path.read_text(encoding="utf-8"							)
      UpperCAmelCase__								=  text_file.read_text(encoding="utf-8"							)
      assert extracted_file_content == expected_file_content




@pytest.mark.parametrize(
    "compression_format, is_archive",  [
        ("7z", True),
        ("bz2", False),
        ("gzip", False),
        ("lz4", False),
        ("tar", True),
        ("xz", False),
        ("zip", True),
        ("zstd", False),
    ],  )
def        lowerCAmelCase_			(					__A,  __A,  __A,  __A,  __A,  __A,  __A,  __A,  __A,  __A,  __A,  __A,  )					->    List[Any]:







      '''simple docstring'''


      UpperCAmelCase__								=  {
          "7z": seven_zip_file,
          "bz2": bza_file,
          "gzip": gz_file,
          "lz4": lza_file,
          "tar": tar_file,
          "xz": xz_file,
          "zip": zip_file,
          "zstd": zstd_file,
      }
      UpperCAmelCase__								=  input_paths[compression_format]
      if input_path is None:
            UpperCAmelCase__								=  f"""for '{compression_format}' compression_format, """
            if compression_format == "7z":
                  reason += require_pyazr.kwargs["reason"]
            elif compression_format == "lz4":
                  reason += require_lza.kwargs["reason"]
            elif compression_format == "zstd":
                  reason += require_zstandard.kwargs["reason"]
            pytest.skip(__A							)
      UpperCAmelCase__								=  Extractor.infer_extractor_format(__A							)
      assert extractor_format is not None
      UpperCAmelCase__								=  tmp_path / ("extracted" if is_archive else "extracted.txt")
      Extractor.extract(__A,  __A,  __A							)
      if is_archive:
            assert output_path.is_dir()
            for file_path in output_path.iterdir():
                  assert file_path.name == text_file.name
                  UpperCAmelCase__								=  file_path.read_text(encoding="utf-8"							)
      else:
            UpperCAmelCase__								=  output_path.read_text(encoding="utf-8"							)
      UpperCAmelCase__								=  text_file.read_text(encoding="utf-8"							)
      assert extracted_file_content == expected_file_content




@pytest.fixture
def        lowerCAmelCase_			(					__A,  __A							)					->    List[str]:







      '''simple docstring'''


      import tarfile

      UpperCAmelCase__								=  tmp_path / "data_dot_dot"
      directory.mkdir()
      UpperCAmelCase__								=  directory / "tar_file_with_dot_dot.tar"
      with tarfile.TarFile(__A,  "w"							) as f:
            f.add(__A,  arcname=os.path.join("..",  text_file.name							)							)
      return path




@pytest.fixture
def        lowerCAmelCase_			(					__A							)					->    Dict:







      '''simple docstring'''


      import tarfile

      UpperCAmelCase__								=  tmp_path / "data_sym_link"
      directory.mkdir()
      UpperCAmelCase__								=  directory / "tar_file_with_sym_link.tar"
      os.symlink("..",  directory / "subdir",  target_is_directory=__A							)
      with tarfile.TarFile(__A,  "w"							) as f:
            f.add(str(directory / "subdir"							),  arcname="subdir"							)  # str required by os.readlink on Windows and Python < 3.8
      return path




@pytest.mark.parametrize(
    "insecure_tar_file, error_log",  [("tar_file_with_dot_dot", "illegal path"), ("tar_file_with_sym_link", "Symlink")],  )
def        lowerCAmelCase_			(					__A,  __A,  __A,  __A,  __A,  __A							)					->    Dict:







      '''simple docstring'''


      UpperCAmelCase__								=  {
          "tar_file_with_dot_dot": tar_file_with_dot_dot,
          "tar_file_with_sym_link": tar_file_with_sym_link,
      }
      UpperCAmelCase__								=  insecure_tar_files[insecure_tar_file]
      UpperCAmelCase__								=  tmp_path / "extracted"
      TarExtractor.extract(__A,  __A							)
      assert caplog.text
      for record in caplog.records:
            assert record.levelname == "ERROR"
            assert error_log in record.msg







def        lowerCAmelCase_			(					__A							)					->    Any:







      '''simple docstring'''


      UpperCAmelCase__								=  tmpdir / "not_a_zip_file"
      # From: https://github.com/python/cpython/pull/5053
      UpperCAmelCase__								=  (
          B"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00"
          B"\x00\x02\x08\x06\x00\x00\x00\x99\x81\xb6'\x00\x00\x00\x15I"
          B"DATx\x01\x01\n\x00\xf5\xff\x00PK\x05\x06\x00PK\x06\x06\x07"
          B"\xac\x01N\xc6|a\r\x00\x00\x00\x00IEND\xaeB`\x82"
      )
      with not_a_zip_file.open("wb"							) as f:
            f.write(__A							)
      assert zipfile.is_zipfile(str(__A							)							)  # is a false positive for `zipfile`
      assert not ZipExtractor.is_extractable(__A							)  # but we're right



