






# Copyright 2023 The HuggingFace Inc. 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 copy
import importlib.metadata
import json
import os
from dataclasses import dataclass
from typing import Any, Dict, Union

from packaging import version

from ..utils import is_torch_available, logging


if is_torch_available():
     import torch


lowerCamelCase__       		=  logging.get_logger(__name__)




@dataclass
class       snake_case__ :

       '''simple docstring'''



       def __init__(				self      ,    a__=False      ,    a__=False      ,    a__=6.0      ,    a__=None      ,    a__=False      ,    a__=False      ,    a__=None      ,    a__="fp4"      ,    a__=False      ,    **a__      ,    )			->   List[str]:






            '''simple docstring'''





            __snake_case      :int 							=    load_in_abit
            __snake_case      :Tuple 							=    load_in_abit
            __snake_case      :Any 							=    llm_inta_threshold
            __snake_case      :str 							=    llm_inta_skip_modules
            __snake_case      :Union[str, Any] 							=    llm_inta_enable_fpaa_cpu_offload
            __snake_case      :Optional[Any] 							=    llm_inta_has_fpaa_weight
            __snake_case      :List[str] 							=    bnb_abit_quant_type
            __snake_case      :Dict 							=    bnb_abit_use_double_quant

            if bnb_abit_compute_dtype is None:
                 __snake_case      :Any 							=    torch.floataa
            elif isinstance(a__      ,    a__     ):
                 __snake_case      :Optional[int] 							=    getattr(a__      ,    a__     )
            elif isinstance(a__      ,    torch.dtype     ):
                 __snake_case      :Optional[Any] 							=    bnb_abit_compute_dtype
            else:
                 raise ValueError("""bnb_4bit_compute_dtype must be a string or a torch.dtype"""     )

            self.post_init()



       def        __lowercase    (				self     )			->   str:






            '''simple docstring'''





            if not isinstance(self.llm_inta_threshold      ,    a__     ):
                 raise ValueError("""llm_int8_threshold must be a float"""     )

            if self.llm_inta_skip_modules is not None and not isinstance(self.llm_inta_skip_modules      ,    a__     ):
                 raise ValueError("""llm_int8_skip_modules must be a list of strings"""     )
            if not isinstance(self.llm_inta_enable_fpaa_cpu_offload      ,    a__     ):
                 raise ValueError("""llm_int8_enable_fp32_cpu_offload must be a boolean"""     )

            if not isinstance(self.llm_inta_has_fpaa_weight      ,    a__     ):
                 raise ValueError("""llm_int8_has_fp16_weight must be a boolean"""     )

            if self.bnb_abit_compute_dtype is not None and not isinstance(self.bnb_abit_compute_dtype      ,    torch.dtype     ):
                 raise ValueError("""bnb_4bit_compute_dtype must be torch.dtype"""     )

            if not isinstance(self.bnb_abit_quant_type      ,    a__     ):
                 raise ValueError("""bnb_4bit_quant_type must be a string"""     )

            if not isinstance(self.bnb_abit_use_double_quant      ,    a__     ):
                 raise ValueError("""bnb_4bit_use_double_quant must be a boolean"""     )

            if self.load_in_abit and not version.parse(importlib.metadata.version("""bitsandbytes"""     )     ) >= version.parse(
                """0.39.0"""     ):
                 raise ValueError(
                     """4 bit quantization requires bitsandbytes>=0.39.0 - please upgrade your bitsandbytes version"""     )



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






            '''simple docstring'''





            return self.load_in_abit or self.load_in_abit



       def        __lowercase    (				self     )			->   Dict:






            '''simple docstring'''





            if self.load_in_abit:
                 return "llm_int8"
            elif self.load_in_abit and self.bnb_abit_quant_type == "fp4":
                 return "fp4"
            elif self.load_in_abit and self.bnb_abit_quant_type == "nf4":
                 return "nf4"
            else:
                 return None



       @classmethod
       def        __lowercase    (				cls      ,    a__      ,    a__      ,    **a__     )			->   int:






            '''simple docstring'''





            __snake_case      :int 							=    cls(**a__     )

            __snake_case      :Tuple 							=    []
            for key, value in kwargs.items():
                 if hasattr(a__      ,    a__     ):
                      setattr(a__      ,    a__      ,    a__     )
                      to_remove.append(a__     )
            for key in to_remove:
                 kwargs.pop(a__      ,    a__     )

            if return_unused_kwargs:
                 return config, kwargs
            else:
                 return config



       def        __lowercase    (				self      ,    a__     )			->   int:






            '''simple docstring'''





            with open(a__      ,    """w"""      ,    encoding="""utf-8"""     ) as writer:
                 __snake_case      :List[Any] 							=    self.to_dict()
                 __snake_case      :str 							=    json.dumps(a__      ,    indent=2      ,    sort_keys=a__     ) + """\n"""

                 writer.write(a__     )



       def        __lowercase    (				self     )			->   Dict[str, Any]:






            '''simple docstring'''





            __snake_case      :int 							=    copy.deepcopy(self.__dict__     )
            __snake_case      :List[str] 							=    str(output["""bnb_4bit_compute_dtype"""]     ).split("""."""     )[1]

            return output



       def __repr__(				self     )			->   int:






            '''simple docstring'''





            return F'''{self.__class__.__name__} {self.to_json_string()}'''



       def        __lowercase    (				self      ,    a__ = True     )			->   str:






            '''simple docstring'''





            if use_diff is True:
                 __snake_case      :Optional[int] 							=    self.to_diff_dict()
            else:
                 __snake_case      :List[str] 							=    self.to_dict()
            return json.dumps(a__      ,    indent=2      ,    sort_keys=a__     ) + "\n"



       def        __lowercase    (				self     )			->   Dict[str, Any]:






            '''simple docstring'''





            __snake_case      :int 							=    self.to_dict()

            # get the default config dict
            __snake_case      :Optional[int] 							=    BitsAndBytesConfig().to_dict()

            __snake_case      :Optional[int] 							=    {}

            # only serialize values that differ from the default config
            for key, value in config_dict.items():
                 if value != default_config_dict[key]:
                      __snake_case      :str 							=    value

            return serializable_config_dict










import importlib
import os

import fsspec
import pytest
from fsspec import register_implementation
from fsspec.registry import _registry as _fsspec_registry

from datasets.filesystems import COMPRESSION_FILESYSTEMS, HfFileSystem, extract_path_from_uri, is_remote_filesystem

from .utils import require_lza, require_zstandard






def 	UpperCamelCase	(							snake_case__  : int	):




     '''simple docstring'''






     assert "mock" in _fsspec_registry
     assert "bz2" in _fsspec_registry






def 	UpperCamelCase	(							):




     '''simple docstring'''






     assert "mock" not in _fsspec_registry
     assert "bz2" in _fsspec_registry






def 	UpperCamelCase	(							):




     '''simple docstring'''






     __snake_case      :Tuple 							=    """mock-s3-bucket"""
     __snake_case      :Optional[Any] 							=    f'''s3://{mock_bucket}'''
     __snake_case      :Optional[int] 							=    extract_path_from_uri(snake_case__	)
     assert dataset_path.startswith("""s3://"""	) is False

     __snake_case      :Union[str, Any] 							=    """./local/path"""
     __snake_case      :List[str] 							=    extract_path_from_uri(snake_case__	)
     assert dataset_path == new_dataset_path






def 	UpperCamelCase	(							snake_case__  : str	):




     '''simple docstring'''






     __snake_case      :List[str] 							=    is_remote_filesystem(snake_case__	)
     assert is_remote is True

     __snake_case      :Any 							=    fsspec.filesystem("""file"""	)

     __snake_case      :Optional[Any] 							=    is_remote_filesystem(snake_case__	)
     assert is_remote is False






@pytest.mark.parametrize("""compression_fs_class"""						,snake_case__	)
def 	UpperCamelCase	(							snake_case__  : Optional[Any]						,snake_case__  : List[str]						,snake_case__  : int						,snake_case__  : List[Any]						,snake_case__  : Tuple						,snake_case__  : str						,snake_case__  : Optional[Any]	):




     '''simple docstring'''






     __snake_case      :Optional[int] 							=    {"""gzip""": gz_file, """xz""": xz_file, """zstd""": zstd_file, """bz2""": bza_file, """lz4""": lza_file}
     __snake_case      :Optional[Any] 							=    input_paths[compression_fs_class.protocol]
     if input_path is None:
          __snake_case      :Optional[int] 							=    f'''for \'{compression_fs_class.protocol}\' compression protocol, '''
          if compression_fs_class.protocol == "lz4":
               reason += require_lza.kwargs["reason"]
          elif compression_fs_class.protocol == "zstd":
               reason += require_zstandard.kwargs["reason"]
          pytest.skip(snake_case__	)
     __snake_case      :Optional[Any] 							=    fsspec.filesystem(compression_fs_class.protocol						,fo=snake_case__	)
     assert isinstance(snake_case__						,snake_case__	)
     __snake_case      :Dict 							=    os.path.basename(snake_case__	)
     __snake_case      :Dict 							=    expected_filename[: expected_filename.rindex("""."""	)]
     assert fs.glob("""*"""	) == [expected_filename]
     with fs.open(snake_case__						,"""r"""						,encoding="""utf-8"""	) as f, open(snake_case__						,encoding="""utf-8"""	) as expected_file:
          assert f.read() == expected_file.read()






@pytest.mark.parametrize("""protocol"""						,["""zip""", """gzip"""]	)
def 	UpperCamelCase	(							snake_case__  : List[Any]						,snake_case__  : List[Any]						,snake_case__  : Dict	):




     '''simple docstring'''






     __snake_case      :Optional[int] 							=    {"""zip""": zip_jsonl_path, """gzip""": jsonl_gz_path}
     __snake_case      :List[str] 							=    compressed_file_paths[protocol]
     __snake_case      :Optional[int] 							=    """dataset.jsonl"""
     __snake_case      :Optional[int] 							=    f'''{protocol}://{member_file_path}::{compressed_file_path}'''
     __snake_case							,			*__snake_case      :Optional[Any] 							=    fsspec.get_fs_token_paths(snake_case__	)
     assert fs.isfile(snake_case__	)
     assert not fs.isfile("""non_existing_""" + member_file_path	)






@pytest.mark.integration
def 	UpperCamelCase	(							snake_case__  : Optional[int]						,snake_case__  : Dict						,snake_case__  : List[str]						,snake_case__  : Dict	):




     '''simple docstring'''






     __snake_case      :List[Any] 							=    hf_api.dataset_info(snake_case__						,token=snake_case__	)
     __snake_case      :List[str] 							=    HfFileSystem(repo_info=snake_case__						,token=snake_case__	)
     assert sorted(hffs.glob("""*"""	)	) == [".gitattributes", "data"]
     assert hffs.isdir("""data"""	)
     assert hffs.isfile(""".gitattributes"""	) and hffs.isfile("""data/text_data.txt"""	)
     with open(snake_case__	) as f:
          assert hffs.open("""data/text_data.txt"""						,"""r"""	).read() == f.read()






def 	UpperCamelCase	(							):




     '''simple docstring'''






     __snake_case      :str 							=    """bz2"""

     # Import module
     import datasets.filesystems

     # Overwrite protocol and reload
     register_implementation(snake_case__						,snake_case__						,clobber=snake_case__	)
     with pytest.warns(snake_case__	) as warning_info:
          importlib.reload(datasets.filesystems	)

     assert len(snake_case__	) == 1
     assert (
         str(warning_info[0].message	)
         == f'''A filesystem protocol was already set for {protocol} and will be overwritten.'''
     )


