






"""simple docstring"""



import bza
import gzip
import lzma
import os
import shutil
import struct
import tarfile
import warnings
import zipfile
from abc import ABC, abstractmethod
from pathlib import Path
from typing import Dict, List, Optional, Type, Union

from .. import config
from .filelock import FileLock
from .logging import get_logger


A__			:							Dict					   =   get_logger(__name__)







class   lowercase__     :


			def __init__( self	:    int  ,			snake_case__	:    Optional[Any] = None   ):
										lowerCamelCase_			:				Any										=(
										    os.path.join(lowercase_  ,			config.EXTRACTED_DATASETS_DIR   ) if cache_dir else config.EXTRACTED_DATASETS_PATH
										)
										lowerCamelCase_			:				str										=Extractor


			def       UpperCAmelCase__		( self	:    str  ,			snake_case__	:    List[str]   ):
										from .file_utils import hash_url_to_filename

										# Path where we extract compressed archives
										# We extract in the cache dir, and get the extracted path name by hashing the original path"
										lowerCamelCase_			:				str										=os.path.abspath(lowercase_   )
										return os.path.join(self.extract_dir  ,			hash_url_to_filename(lowercase_   )   )


			def       UpperCAmelCase__		( self	:    List[str]  ,			snake_case__	:    List[Any]  ,			snake_case__	:    Union[str, Any]   ):
										return force_extract or (
										    not os.path.isfile(lowercase_   ) and not (os.path.isdir(lowercase_   ) and os.listdir(lowercase_   ))
										)


			def       UpperCAmelCase__		( self	:    int  ,			snake_case__	:    Union[str, Any]  ,			snake_case__	:    Union[str, Any] = False   ):
										lowerCamelCase_			:				Tuple										=self.extractor.infer_extractor_format(lowercase_   )
										if not extractor_format:
																	return input_path
										lowerCamelCase_			:				List[str]										=self._get_output_path(lowercase_   )
										if self._do_extract(lowercase_  ,			lowercase_   ):
																	self.extractor.extract(lowercase_  ,			lowercase_  ,			lowercase_   )
										return output_path







class   lowercase__     (				SCREAMING_SNAKE_CASE_  ):


			@classmethod
			@abstractmethod
			def       UpperCAmelCase__		( cls	:    Any  ,			snake_case__	:    Union[str, Any]  ,			**snake_case__	:    Tuple   ):
										...


			@staticmethod
			@abstractmethod
			def       UpperCAmelCase__		( snake_case__	:    str  ,			snake_case__	:    Optional[Any]   ):
										...







class   lowercase__     (				SCREAMING_SNAKE_CASE_,  SCREAMING_SNAKE_CASE_  ):
			_UpperCAmelCase						:List[bytes]       =					[]


			@staticmethod
			def       UpperCAmelCase__		( snake_case__	:    Union[str, Any]  ,			snake_case__	:    int   ):
										with open(lowercase_  ,			"rb"   ) as f:
																	return f.read(lowercase_   )


			@classmethod
			def       UpperCAmelCase__		( cls	:    List[Any]  ,			snake_case__	:    str  ,			snake_case__	:    List[str] = b""   ):
										if not magic_number:
																	lowerCamelCase_			:				List[str]										=max(len(lowercase_   ) for cls_magic_number in cls.magic_numbers   )
																	try:
																								lowerCamelCase_			:				List[Any]										=cls.read_magic_number(lowercase_  ,			lowercase_   )
																	except OSError:
																								return False
										return any(magic_number.startswith(lowercase_   ) for cls_magic_number in cls.magic_numbers   )







class   lowercase__     (				SCREAMING_SNAKE_CASE_  ):


			@classmethod
			def       UpperCAmelCase__		( cls	:    List[str]  ,			snake_case__	:    Optional[int]  ,			**snake_case__	:    Optional[int]   ):
										return tarfile.is_tarfile(lowercase_   )


			@staticmethod
			def       UpperCAmelCase__		( snake_case__	:    Optional[Any]  ,			snake_case__	:    List[Any]   ):

										def resolved(snake_case__	:    str   ) -> str:
																	return os.path.realpath(os.path.abspath(lowercase_   )   )

										def badpath(snake_case__	:    Tuple  ,			snake_case__	:    str   ) -> bool:
																	# joinpath will ignore base if path is absolute
																	return not resolved(os.path.join(lowercase_  ,			lowercase_   )   ).startswith(lowercase_   )

										def badlink(snake_case__	:    Dict  ,			snake_case__	:    str   ) -> bool:
																	# Links are interpreted relative to the directory containing the link
																	lowerCamelCase_			:				str										=resolved(os.path.join(lowercase_  ,			os.path.dirname(info.name   )   )   )
																	return badpath(info.linkname  ,			base=lowercase_   )

										lowerCamelCase_			:				Union[str, Any]										=resolved(lowercase_   )

										for finfo in members:
																	if badpath(finfo.name  ,			lowercase_   ):
																								logger.error(F"""Extraction of {finfo.name} is blocked (illegal path)"""   )
																	elif finfo.issym() and badlink(lowercase_  ,			lowercase_   ):
																								logger.error(F"""Extraction of {finfo.name} is blocked: Symlink to {finfo.linkname}"""   )
																	elif finfo.islnk() and badlink(lowercase_  ,			lowercase_   ):
																								logger.error(F"""Extraction of {finfo.name} is blocked: Hard link to {finfo.linkname}"""   )
																	else:
																								yield finfo


			@staticmethod
			def       UpperCAmelCase__		( snake_case__	:    Union[str, Any]  ,			snake_case__	:    List[str]   ):
										os.makedirs(lowercase_  ,			exist_ok=lowercase_   )
										lowerCamelCase_			:				List[Any]										=tarfile.open(lowercase_   )
										tar_file.extractall(lowercase_  ,			members=TarExtractor.safemembers(lowercase_  ,			lowercase_   )   )
										tar_file.close()







class   lowercase__     (				SCREAMING_SNAKE_CASE_  ):
			_UpperCAmelCase						:Optional[Any]       =					[B"\x1F\x8B"]


			@staticmethod
			def       UpperCAmelCase__		( snake_case__	:    Dict  ,			snake_case__	:    Union[str, Any]   ):
										with gzip.open(lowercase_  ,			"rb"   ) as gzip_file:
																	with open(lowercase_  ,			"wb"   ) as extracted_file:
																								shutil.copyfileobj(lowercase_  ,			lowercase_   )







class   lowercase__     (				SCREAMING_SNAKE_CASE_  ):
			_UpperCAmelCase						:int       =					[
			    B"PK\x03\x04",
			    B"PK\x05\x06",  # empty archive
			    B"PK\x07\x08",  # spanned archive
			]


			@classmethod
			def       UpperCAmelCase__		( cls	:    Any  ,			snake_case__	:    int  ,			snake_case__	:    List[str] = b""   ):
										if super().is_extractable(lowercase_  ,			magic_number=lowercase_   ):
																	return True
										try:
																	# Alternative version of zipfile.is_zipfile that has less false positives, but misses executable zip archives.
																	# From: https://github.com/python/cpython/pull/5053
																	from zipfile import (
																	    _CD_SIGNATURE,
																	    _ECD_DISK_NUMBER,
																	    _ECD_DISK_START,
																	    _ECD_ENTRIES_TOTAL,
																	    _ECD_OFFSET,
																	    _ECD_SIZE,
																	    _EndRecData,
																	    sizeCentralDir,
																	    stringCentralDir,
																	    structCentralDir,
																	)

																	with open(lowercase_  ,			"rb"   ) as fp:
																								lowerCamelCase_			:				Any										=_EndRecData(lowercase_   )
																								if endrec:
																															if endrec[_ECD_ENTRIES_TOTAL] == 0 and endrec[_ECD_SIZE] == 0 and endrec[_ECD_OFFSET] == 0:
																																						return True  # Empty zipfiles are still zipfiles
																															elif endrec[_ECD_DISK_NUMBER] == endrec[_ECD_DISK_START]:
																																						fp.seek(endrec[_ECD_OFFSET]   )  # Central directory is on the same disk
																																						if fp.tell() == endrec[_ECD_OFFSET] and endrec[_ECD_SIZE] >= sizeCentralDir:
																																													lowerCamelCase_			:				List[Any]										=fp.read(lowercase_   )  # CD is where we expect it to be
																																													if len(lowercase_   ) == sizeCentralDir:
																																																				lowerCamelCase_			:				Tuple										=struct.unpack(lowercase_  ,			lowercase_   )  # CD is the right size
																																																				if centdir[_CD_SIGNATURE] == stringCentralDir:
																																																											return True  # First central directory entry  has correct magic number
																	return False
										except Exception:  # catch all errors in case future python versions change the zipfile internals
																	return False


			@staticmethod
			def       UpperCAmelCase__		( snake_case__	:    Tuple  ,			snake_case__	:    Dict   ):
										os.makedirs(lowercase_  ,			exist_ok=lowercase_   )
										with zipfile.ZipFile(lowercase_  ,			"r"   ) as zip_file:
																	zip_file.extractall(lowercase_   )
																	zip_file.close()







class   lowercase__     (				SCREAMING_SNAKE_CASE_  ):
			_UpperCAmelCase						:List[Any]       =					[B"\xFD\x37\x7A\x58\x5A\x00"]


			@staticmethod
			def       UpperCAmelCase__		( snake_case__	:    Tuple  ,			snake_case__	:    Tuple   ):
										with lzma.open(lowercase_   ) as compressed_file:
																	with open(lowercase_  ,			"wb"   ) as extracted_file:
																								shutil.copyfileobj(lowercase_  ,			lowercase_   )







class   lowercase__     (				SCREAMING_SNAKE_CASE_  ):
			_UpperCAmelCase						:Optional[int]       =					[B"Rar!\x1a\x07\x00", B"Rar!\x1a\x07\x01\x00"]  # RAR_ID  # RAR5_ID


			@staticmethod
			def       UpperCAmelCase__		( snake_case__	:    int  ,			snake_case__	:    int   ):
										if not config.RARFILE_AVAILABLE:
																	raise ImportError("Please pip install rarfile"   )
										import rarfile

										os.makedirs(lowercase_  ,			exist_ok=lowercase_   )
										lowerCamelCase_			:				Tuple										=rarfile.RarFile(lowercase_   )
										rf.extractall(lowercase_   )
										rf.close()







class   lowercase__     (				SCREAMING_SNAKE_CASE_  ):
			_UpperCAmelCase						:Dict       =					[B"\x28\xb5\x2F\xFD"]


			@staticmethod
			def       UpperCAmelCase__		( snake_case__	:    Union[str, Any]  ,			snake_case__	:    Tuple   ):
										if not config.ZSTANDARD_AVAILABLE:
																	raise ImportError("Please pip install zstandard"   )
										import zstandard as zstd

										lowerCamelCase_			:				List[str]										=zstd.ZstdDecompressor()
										with open(lowercase_  ,			"rb"   ) as ifh, open(lowercase_  ,			"wb"   ) as ofh:
																	dctx.copy_stream(lowercase_  ,			lowercase_   )







class   lowercase__     (				SCREAMING_SNAKE_CASE_  ):
			_UpperCAmelCase						:Union[str, Any]       =					[B"\x42\x5A\x68"]


			@staticmethod
			def       UpperCAmelCase__		( snake_case__	:    Optional[Any]  ,			snake_case__	:    Tuple   ):
										with bza.open(lowercase_  ,			"rb"   ) as compressed_file:
																	with open(lowercase_  ,			"wb"   ) as extracted_file:
																								shutil.copyfileobj(lowercase_  ,			lowercase_   )







class   lowercase__     (				SCREAMING_SNAKE_CASE_  ):
			_UpperCAmelCase						:Union[str, Any]       =					[B"\x37\x7A\xBC\xAF\x27\x1C"]


			@staticmethod
			def       UpperCAmelCase__		( snake_case__	:    str  ,			snake_case__	:    int   ):
										if not config.PY7ZR_AVAILABLE:
																	raise ImportError("Please pip install py7zr"   )
										import pyazr

										os.makedirs(lowercase_  ,			exist_ok=lowercase_   )
										with pyazr.SevenZipFile(lowercase_  ,			"r"   ) as archive:
																	archive.extractall(lowercase_   )







class   lowercase__     (				SCREAMING_SNAKE_CASE_  ):
			_UpperCAmelCase						:Optional[Any]       =					[B"\x04\x22\x4D\x18"]


			@staticmethod
			def       UpperCAmelCase__		( snake_case__	:    Tuple  ,			snake_case__	:    Optional[Any]   ):
										if not config.LZ4_AVAILABLE:
																	raise ImportError("Please pip install lz4"   )
										import lza.frame

										with lza.frame.open(lowercase_  ,			"rb"   ) as compressed_file:
																	with open(lowercase_  ,			"wb"   ) as extracted_file:
																								shutil.copyfileobj(lowercase_  ,			lowercase_   )







class   lowercase__     :
			#  Put zip file to the last, b/c it is possible wrongly detected as zip (I guess it means: as tar or gzip)
			_UpperCAmelCase						:Dict[str, Type[BaseExtractor]]       =					{
			    "tar": TarExtractor,
			    "gzip": GzipExtractor,
			    "zip": ZipExtractor,
			    "xz": XzExtractor,
			    "rar": RarExtractor,
			    "zstd": ZstdExtractor,
			    "bz2": BzipaExtractor,
			    "7z": SevenZipExtractor,  # <Added version="2.4.0"/>
			    "lz4": LzaExtractor,  # <Added version="2.4.0"/>
			}


			@classmethod
			def       UpperCAmelCase__		( cls	:    Any   ):
										return max(
										    len(lowercase_   )
										    for extractor in cls.extractors.values()
										    if issubclass(lowercase_  ,			lowercase_   )
										    for extractor_magic_number in extractor.magic_numbers   )


			@staticmethod
			def       UpperCAmelCase__		( snake_case__	:    str  ,			snake_case__	:    List[str]   ):
										try:
																	return MagicNumberBaseExtractor.read_magic_number(lowercase_  ,			magic_number_length=lowercase_   )
										except OSError:
																	return b""


			@classmethod
			def       UpperCAmelCase__		( cls	:    Dict  ,			snake_case__	:    Union[str, Any]  ,			snake_case__	:    Union[str, Any] = False   ):
										warnings.warn(
										    "Method \'is_extractable\' was deprecated in version 2.4.0 and will be removed in 3.0.0. "
										    "Use \'infer_extractor_format\' instead."  ,			category=lowercase_  ,			)
										lowerCamelCase_			:				Dict										=cls.infer_extractor_format(lowercase_   )
										if extractor_format:
																	return True if not return_extractor else (True, cls.extractors[extractor_format])
										return False if not return_extractor else (False, None)


			@classmethod
			def       UpperCAmelCase__		( cls	:    Optional[int]  ,			snake_case__	:    Optional[int]   ):  # <Added version="2.4.0"/>
										lowerCamelCase_			:				Union[str, Any]										=cls._get_magic_number_max_length()
										lowerCamelCase_			:				Any										=cls._read_magic_number(lowercase_  ,			lowercase_   )
										for extractor_format, extractor in cls.extractors.items():
																	if extractor.is_extractable(lowercase_  ,			magic_number=lowercase_   ):
																								return extractor_format


			@classmethod
			def       UpperCAmelCase__		( cls	:    List[Any]  ,			snake_case__	:    Union[str, Any]  ,			snake_case__	:    Tuple  ,			snake_case__	:    Union[str, Any] = None  ,			snake_case__	:    Dict = "deprecated"  ,			):
										os.makedirs(os.path.dirname(lowercase_   )  ,			exist_ok=lowercase_   )
										# Prevent parallel extractions
										lowerCamelCase_			:				Dict										=str(Path(lowercase_   ).with_suffix(".lock"   )   )
										with FileLock(lowercase_   ):
																	shutil.rmtree(lowercase_  ,			ignore_errors=lowercase_   )
																	if extractor_format or extractor != "deprecated":
																								if extractor != "deprecated" or not isinstance(lowercase_  ,			lowercase_   ):  # passed as positional arg
																															warnings.warn(
																															    "Parameter \'extractor\' was deprecated in version 2.4.0 and will be removed in 3.0.0. "
																															    "Use \'extractor_format\' instead."  ,			category=lowercase_  ,			)
																															lowerCamelCase_			:				Union[str, Any]										=extractor if extractor != "deprecated" else extractor_format
																								else:
																															lowerCamelCase_			:				Tuple										=cls.extractors[extractor_format]
																								return extractor.extract(lowercase_  ,			lowercase_   )
																	else:
																								warnings.warn(
																								    "Parameter \'extractor_format\' was made required in version 2.4.0 and not passing it will raise an "
																								    "exception in 3.0.0."  ,			category=lowercase_  ,			)
																								for extractor in cls.extractors.values():
																															if extractor.is_extractable(lowercase_   ):
																																						return extractor.extract(lowercase_  ,			lowercase_   )





"""simple docstring"""

import functools
import operator

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


SCREAMING_SNAKE_CASE_            =    logging.get_logger(__name__)

SCREAMING_SNAKE_CASE_            =    {
    '''asapp/sew-d-tiny-100k''': '''https://huggingface.co/asapp/sew-d-tiny-100k/resolve/main/config.json''',
    # See all SEW-D models at https://huggingface.co/models?filter=sew-d
}
class   _UpperCAmelCase				(		SCREAMING_SNAKE_CASE_				):
      __SCREAMING_SNAKE_CASE :				List[str]				   =						"sew-d"




      def __init__(				self ,							lowercase_=3_2 ,							lowercase_=7_6_8 ,							lowercase_=1_2 ,							lowercase_=1_2 ,							lowercase_=3_0_7_2 ,							lowercase_=2 ,							lowercase_=5_1_2 ,							lowercase_=2_5_6 ,							lowercase_=True ,							lowercase_=True ,							lowercase_=("p2c", "c2p") ,							lowercase_="layer_norm" ,							lowercase_="gelu_python" ,							lowercase_=0.1 ,							lowercase_=0.1 ,							lowercase_=0.1 ,							lowercase_=0.0 ,							lowercase_=0.1 ,							lowercase_=0.0_2 ,							lowercase_=1E-7 ,							lowercase_=1E-5 ,							lowercase_="group" ,							lowercase_="gelu" ,							lowercase_=(6_4, 1_2_8, 1_2_8, 1_2_8, 1_2_8, 2_5_6, 2_5_6, 2_5_6, 2_5_6, 5_1_2, 5_1_2, 5_1_2, 5_1_2) ,							lowercase_=(5, 2, 1, 2, 1, 2, 1, 2, 1, 2, 1, 2, 1) ,							lowercase_=(1_0, 3, 1, 3, 1, 3, 1, 3, 1, 2, 1, 2, 1) ,							lowercase_=False ,							lowercase_=1_2_8 ,							lowercase_=1_6 ,							lowercase_=True ,							lowercase_=0.0_5 ,							lowercase_=1_0 ,							lowercase_=2 ,							lowercase_=0.0 ,							lowercase_=1_0 ,							lowercase_=0 ,							lowercase_="mean" ,							lowercase_=False ,							lowercase_=False ,							lowercase_=2_5_6 ,							lowercase_=0 ,							lowercase_=1 ,							lowercase_=2 ,							**lowercase_ ,							)    ->       Dict:
          super().__init__(**lowercase_ ,							pad_token_id=lowercase_ ,							bos_token_id=lowercase_ ,							eos_token_id=lowercase_				)
          UpperCAmelCase          =       hidden_size
          UpperCAmelCase          =       feat_extract_norm
          UpperCAmelCase          =       feat_extract_activation
          UpperCAmelCase          =       list(lowercase_				)
          UpperCAmelCase          =       list(lowercase_				)
          UpperCAmelCase          =       list(lowercase_				)
          UpperCAmelCase          =       conv_bias
          UpperCAmelCase          =       num_conv_pos_embeddings
          UpperCAmelCase          =       num_conv_pos_embedding_groups
          UpperCAmelCase          =       len(self.conv_dim				)
          UpperCAmelCase          =       num_hidden_layers
          UpperCAmelCase          =       intermediate_size
          UpperCAmelCase          =       squeeze_factor
          UpperCAmelCase          =       max_position_embeddings
          UpperCAmelCase          =       position_buckets
          UpperCAmelCase          =       share_att_key
          UpperCAmelCase          =       relative_attention
          UpperCAmelCase          =       norm_rel_ebd
          UpperCAmelCase          =       list(lowercase_				)
          UpperCAmelCase          =       hidden_act
          UpperCAmelCase          =       num_attention_heads
          UpperCAmelCase          =       hidden_dropout
          UpperCAmelCase          =       attention_dropout
          UpperCAmelCase          =       activation_dropout
          UpperCAmelCase          =       feat_proj_dropout
          UpperCAmelCase          =       final_dropout
          UpperCAmelCase          =       layer_norm_eps
          UpperCAmelCase          =       feature_layer_norm_eps
          UpperCAmelCase          =       initializer_range
          UpperCAmelCase          =       vocab_size

          if (
              (len(self.conv_stride				) != self.num_feat_extract_layers)
              or (len(self.conv_kernel				) != self.num_feat_extract_layers)
              or (len(self.conv_dim				) != self.num_feat_extract_layers)
          ):
              raise ValueError(
                  'Configuration for convolutional layers is incorrect.'
                  'It is required that `len(config.conv_dim)` == `len(config.conv_stride)` == `len(config.conv_kernel)`,'
                  F"but is `len(config.conv_dim) = {len(self.conv_dim				)}`, `len(config.conv_stride)"
                  F"= {len(self.conv_stride				)}`, `len(config.conv_kernel) = {len(self.conv_kernel				)}`."				)

          # fine-tuning config parameters for SpecAugment: https://arxiv.org/abs/1904.08779
          UpperCAmelCase          =       apply_spec_augment
          UpperCAmelCase          =       mask_time_prob
          UpperCAmelCase          =       mask_time_length
          UpperCAmelCase          =       mask_time_min_masks
          UpperCAmelCase          =       mask_feature_prob
          UpperCAmelCase          =       mask_feature_length
          UpperCAmelCase          =       mask_feature_min_masks

          # ctc loss
          UpperCAmelCase          =       ctc_loss_reduction
          UpperCAmelCase          =       ctc_zero_infinity

          # sequence classification
          UpperCAmelCase          =       use_weighted_layer_sum
          UpperCAmelCase          =       classifier_proj_size




      @property
      def      a_   (				self				)    ->       Tuple:
          return functools.reduce(operator.mul ,							self.conv_stride ,							1				)



