


"""simple docstring"""



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						(						lowercase  :  Tuple						,							lowercase  :  Union[str, Any]						,							lowercase  :  List[Any]						,							lowercase  :  Optional[int]						,							lowercase  :  List[Any]						,							lowercase  :  Dict						,							lowercase  :  str						,							lowercase  :  Any						,							lowercase  :  Optional[Any]						,							lowercase  :  int						,							lowercase  :  str						,							lowercase  :  Union[str, Any]						,							)     ->						List[str]:


    """simple docstring"""





    snake_case   :				List[Any]				 =   {
        "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),
    }
    snake_case ,snake_case   :				List[str]				 =   input_paths_and_base_extractors[compression_format]
    if input_path is None:
        snake_case   :				Union[str, Any]				 =   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(lowercase			)
    assert base_extractor.is_extractable(lowercase			)
    snake_case   :				Tuple				 =   tmp_path / ("extracted" if is_archive else "extracted.txt")
    base_extractor.extract(lowercase						,							lowercase			)
    if is_archive:
        assert output_path.is_dir()
        for file_path in output_path.iterdir():
            assert file_path.name == text_file.name
            snake_case   :				Dict				 =   file_path.read_text(encoding="utf-8"			)
    else:
        snake_case   :				Optional[Any]				 =   output_path.read_text(encoding="utf-8"			)
    snake_case   :				Optional[Any]				 =   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						(						lowercase  :  Union[str, Any]						,							lowercase  :  Optional[int]						,							lowercase  :  Dict						,							lowercase  :  str						,							lowercase  :  int						,							lowercase  :  List[str]						,							lowercase  :  List[Any]						,							lowercase  :  List[str]						,							lowercase  :  List[Any]						,							lowercase  :  int						,							lowercase  :  Optional[Any]						,							lowercase  :  List[str]						,							)     ->						List[Any]:


    """simple docstring"""





    snake_case   :				List[Any]				 =   {
        "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,
    }
    snake_case   :				str				 =   input_paths[compression_format]
    if input_path is None:
        snake_case   :				Optional[Any]				 =   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(lowercase			)
    snake_case   :				Optional[int]				 =   Extractor.infer_extractor_format(lowercase			)
    assert extractor_format is not None
    snake_case   :				List[str]				 =   tmp_path / ("extracted" if is_archive else "extracted.txt")
    Extractor.extract(lowercase						,							lowercase						,							lowercase			)
    if is_archive:
        assert output_path.is_dir()
        for file_path in output_path.iterdir():
            assert file_path.name == text_file.name
            snake_case   :				Union[str, Any]				 =   file_path.read_text(encoding="utf-8"			)
    else:
        snake_case   :				Tuple				 =   output_path.read_text(encoding="utf-8"			)
    snake_case   :				Any				 =   text_file.read_text(encoding="utf-8"			)
    assert extracted_file_content == expected_file_content







@pytest.fixture
def   __lowerCAmelCase						(						lowercase  :  Tuple						,							lowercase  :  Optional[Any]			)     ->						int:


    """simple docstring"""





    import tarfile

    snake_case   :				Optional[int]				 =   tmp_path / "data_dot_dot"
    directory.mkdir()
    snake_case   :				List[str]				 =   directory / "tar_file_with_dot_dot.tar"
    with tarfile.TarFile(lowercase						,							"w"			) as f:
        f.add(lowercase						,							arcname=os.path.join(".."						,							text_file.name			)			)
    return path







@pytest.fixture
def   __lowerCAmelCase						(						lowercase  :  Any			)     ->						List[str]:


    """simple docstring"""





    import tarfile

    snake_case   :				Tuple				 =   tmp_path / "data_sym_link"
    directory.mkdir()
    snake_case   :				Dict				 =   directory / "tar_file_with_sym_link.tar"
    os.symlink(".."						,							directory / "subdir"						,							target_is_directory=lowercase			)
    with tarfile.TarFile(lowercase						,							"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						(						lowercase  :  List[Any]						,							lowercase  :  Tuple						,							lowercase  :  int						,							lowercase  :  Optional[int]						,							lowercase  :  int						,							lowercase  :  int			)     ->						List[str]:


    """simple docstring"""





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







def   __lowerCAmelCase						(						lowercase  :  Tuple			)     ->						int:


    """simple docstring"""





    snake_case   :				Tuple				 =   tmpdir / "not_a_zip_file"
    # From: https://github.com/python/cpython/pull/5053
    snake_case   :				List[str]				 =   (
        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(lowercase			)
    assert zipfile.is_zipfile(str(lowercase			)			)  # is a false positive for `zipfile`
    assert not ZipExtractor.is_extractable(lowercase			)  # but we're right









"""simple docstring"""



import copy
import os
from typing import Union

from ...configuration_utils import PretrainedConfig
from ...utils import logging


__snake_case             =   logging.get_logger(__name__)

__snake_case             =   {
    """BAAI/AltCLIP""": """https://huggingface.co/BAAI/AltCLIP/resolve/main/config.json""",
    # See all AltCLIP models at https://huggingface.co/models?filter=altclip
}







class _lowerCAmelCase    (				snake_case_     ):
   __UpperCAmelCase			:	Tuple							  =  '''altclip_text_model'''







   def __init__(     self						,     UpperCamelCase__=25_0002						,     UpperCamelCase__=1024						,     UpperCamelCase__=24						,     UpperCamelCase__=16						,     UpperCamelCase__=4096						,     UpperCamelCase__="gelu"						,     UpperCamelCase__=0.1						,     UpperCamelCase__=0.1						,     UpperCamelCase__=514						,     UpperCamelCase__=1						,     UpperCamelCase__=0.02						,     UpperCamelCase__=0.02						,     UpperCamelCase__=1e-05						,     UpperCamelCase__=1						,     UpperCamelCase__=0						,     UpperCamelCase__=2						,     UpperCamelCase__="absolute"						,     UpperCamelCase__=True						,     UpperCamelCase__=768						,     **UpperCamelCase__						,     )     ->		Optional[int]:

       '''simple docstring'''




       super().__init__(pad_token_id=UpperCamelCase__						,     bos_token_id=UpperCamelCase__						,     eos_token_id=UpperCamelCase__						,     **UpperCamelCase__						)

       snake_case   :				Any				 =   vocab_size
       snake_case   :				List[Any]				 =   hidden_size
       snake_case   :				Optional[int]				 =   num_hidden_layers
       snake_case   :				Optional[Any]				 =   num_attention_heads
       snake_case   :				Dict				 =   hidden_act
       snake_case   :				Dict				 =   intermediate_size
       snake_case   :				int				 =   hidden_dropout_prob
       snake_case   :				Optional[int]				 =   attention_probs_dropout_prob
       snake_case   :				Union[str, Any]				 =   max_position_embeddings
       snake_case   :				Optional[int]				 =   type_vocab_size
       snake_case   :				Dict				 =   initializer_range
       snake_case   :				int				 =   initializer_factor
       snake_case   :				Union[str, Any]				 =   layer_norm_eps
       snake_case   :				List[Any]				 =   position_embedding_type
       snake_case   :				Any				 =   use_cache
       snake_case   :				str				 =   project_dim







class _lowerCAmelCase    (				snake_case_     ):
   __UpperCAmelCase			:	Tuple							  =  '''altclip_vision_model'''







   def __init__(     self						,     UpperCamelCase__=768						,     UpperCamelCase__=3072						,     UpperCamelCase__=512						,     UpperCamelCase__=12						,     UpperCamelCase__=12						,     UpperCamelCase__=3						,     UpperCamelCase__=224						,     UpperCamelCase__=32						,     UpperCamelCase__="quick_gelu"						,     UpperCamelCase__=1e-5						,     UpperCamelCase__=0.0						,     UpperCamelCase__=0.02						,     UpperCamelCase__=1.0						,     **UpperCamelCase__						,     )     ->		str:

       '''simple docstring'''




       super().__init__(**UpperCamelCase__						)

       snake_case   :				Optional[int]				 =   hidden_size
       snake_case   :				str				 =   intermediate_size
       snake_case   :				List[str]				 =   projection_dim
       snake_case   :				Optional[Any]				 =   num_hidden_layers
       snake_case   :				Optional[int]				 =   num_attention_heads
       snake_case   :				str				 =   num_channels
       snake_case   :				List[str]				 =   patch_size
       snake_case   :				List[Any]				 =   image_size
       snake_case   :				Union[str, Any]				 =   initializer_range
       snake_case   :				Optional[Any]				 =   initializer_factor
       snake_case   :				Any				 =   attention_dropout
       snake_case   :				Dict				 =   layer_norm_eps
       snake_case   :				List[str]				 =   hidden_act







   @classmethod
   def 			lowerCamelCase		(     cls						,     UpperCamelCase__						,     **UpperCamelCase__						)     ->		"PretrainedConfig":

       '''simple docstring'''




       cls._set_token_in_kwargs(UpperCamelCase__						)

       snake_case ,snake_case   :				str				 =   cls.get_config_dict(UpperCamelCase__						,     **UpperCamelCase__						)

       # get the vision config dict if we are loading from AltCLIPConfig
       if config_dict.get("model_type"						) == "altclip":
           snake_case   :				Optional[Any]				 =   config_dict["vision_config"]

       if "model_type" in config_dict and hasattr(cls						,     "model_type"						) and config_dict["model_type"] != cls.model_type:
           logger.warning(
               F'You are using a model of type {config_dict["model_type"]} to instantiate a model of type '
               F'{cls.model_type}. This is not supported for all configurations of models and can yield errors.'						)

       return cls.from_dict(UpperCamelCase__						,     **UpperCamelCase__						)







class _lowerCAmelCase    (				snake_case_     ):
   __UpperCAmelCase			:	str							  =  '''altclip'''
   __UpperCAmelCase			:	Optional[Any]							  =  True







   def __init__(     self						,     UpperCamelCase__=None						,     UpperCamelCase__=None						,     UpperCamelCase__=768						,     UpperCamelCase__=2.6592						,     **UpperCamelCase__						)     ->		Any:

       '''simple docstring'''




       snake_case   :				List[str]				 =   kwargs.pop("text_config_dict"						,     UpperCamelCase__						)
       snake_case   :				Union[str, Any]				 =   kwargs.pop("vision_config_dict"						,     UpperCamelCase__						)

       super().__init__(**UpperCamelCase__						)

       # Instead of simply assigning `[text|vision]_config_dict` to `[text|vision]_config`, we use the values in
       # `[text|vision]_config_dict` to update the values in `[text|vision]_config`. The values should be same in most
       # cases, but we don't want to break anything regarding `_config_dict` that existed before commit `8827e1b2`.
       if text_config_dict is not None:
           if text_config is None:
               snake_case   :				List[str]				 =   {}

           # This is the complete result when using `text_config_dict`.
           snake_case   :				Dict				 =   AltCLIPTextConfig(**UpperCamelCase__						).to_dict()

           # Give a warning if the values exist in both `_text_config_dict` and `text_config` but being different.
           for key, value in _text_config_dict.items():
               if key in text_config and value != text_config[key] and key not in ["transformers_version"]:
                   # If specified in `text_config_dict`
                   if key in text_config_dict:
                       snake_case   :				Optional[Any]				 =   (
                           F'`{key}` is found in both `text_config_dict` and `text_config` but with different values. '
                           F'The value `text_config_dict["{key}"]` will be used instead.'
                       )
                   # If inferred from default argument values (just to be super careful)
                   else:
                       snake_case   :				Any				 =   (
                           F'`text_config_dict` is provided which will be used to initialize `AltCLIPTextConfig`. The '
                           F'value `text_config["{key}"]` will be overriden.'
                       )
                   logger.warning(UpperCamelCase__						)

            # Update all values in `text_config` with the ones in `_text_config_dict`.
           text_config.update(_text_config_dict						)

       if vision_config_dict is not None:
           if vision_config is None:
               snake_case   :				Union[str, Any]				 =   {}

           # This is the complete result when using `vision_config_dict`.
           snake_case   :				int				 =   AltCLIPVisionConfig(**UpperCamelCase__						).to_dict()
           # convert keys to string instead of integer
           if "id2label" in _vision_config_dict:
               snake_case   :				Optional[int]				 =   {
                   str(UpperCamelCase__						): value for key, value in _vision_config_dict["id2label"].items()
               }

           # Give a warning if the values exist in both `_vision_config_dict` and `vision_config` but being different.
           for key, value in _vision_config_dict.items():
               if key in vision_config and value != vision_config[key] and key not in ["transformers_version"]:
                   # If specified in `vision_config_dict`
                   if key in vision_config_dict:
                       snake_case   :				int				 =   (
                           F'`{key}` is found in both `vision_config_dict` and `vision_config` but with different '
                           F'values. The value `vision_config_dict["{key}"]` will be used instead.'
                       )
                   # If inferred from default argument values (just to be super careful)
                   else:
                       snake_case   :				Optional[Any]				 =   (
                           F'`vision_config_dict` is provided which will be used to initialize `AltCLIPVisionConfig`. '
                           F'The value `vision_config["{key}"]` will be overriden.'
                       )
                   logger.warning(UpperCamelCase__						)

            # Update all values in `vision_config` with the ones in `_vision_config_dict`.
           vision_config.update(_vision_config_dict						)

       if text_config is None:
           snake_case   :				Optional[int]				 =   {}
           logger.info("`text_config` is `None`. Initializing the `AltCLIPTextConfig` with default values."						)

       if vision_config is None:
           snake_case   :				Dict				 =   {}
           logger.info("`vision_config` is `None`. initializing the `AltCLIPVisionConfig` with default values."						)

       snake_case   :				Dict				 =   AltCLIPTextConfig(**UpperCamelCase__						)
       snake_case   :				Tuple				 =   AltCLIPVisionConfig(**UpperCamelCase__						)

       snake_case   :				int				 =   projection_dim
       snake_case   :				List[str]				 =   logit_scale_init_value
       snake_case   :				int				 =   1.0







   @classmethod
   def 			lowerCamelCase		(     cls						,     UpperCamelCase__						,     UpperCamelCase__						,     **UpperCamelCase__						)     ->		Union[str, Any]:

       '''simple docstring'''




       return cls(text_config=text_config.to_dict()						,     vision_config=vision_config.to_dict()						,     **UpperCamelCase__						)







   def 			lowerCamelCase		(     self						)     ->		Optional[Any]:

       '''simple docstring'''




       snake_case   :				Tuple				 =   copy.deepcopy(self.__dict__						)
       snake_case   :				Optional[int]				 =   self.text_config.to_dict()
       snake_case   :				str				 =   self.vision_config.to_dict()
       snake_case   :				Optional[int]				 =   self.__class__.model_type
       return output





