





"""simple docstring"""





import json
import os
from typing import Optional, Tuple

import regex as re

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


SCREAMING_SNAKE_CASE__   =  logging.get_logger(__name__)

SCREAMING_SNAKE_CASE__   =  {
    '''vocab_file''': '''vocab.json''',
    '''merges_file''': '''merges.txt''',
}

SCREAMING_SNAKE_CASE__   =  {
    '''vocab_file''': {'''ctrl''': '''https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-vocab.json'''},
    '''merges_file''': {'''ctrl''': '''https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-merges.txt'''},
}

SCREAMING_SNAKE_CASE__   =  {
    '''ctrl''': 256,
}

SCREAMING_SNAKE_CASE__   =  {
    '''Pregnancy''': 168_629,
    '''Christianity''': 7_675,
    '''Explain''': 106_423,
    '''Fitness''': 63_440,
    '''Saving''': 63_163,
    '''Ask''': 27_171,
    '''Ass''': 95_985,
    '''Joke''': 163_509,
    '''Questions''': 45_622,
    '''Thoughts''': 49_605,
    '''Retail''': 52_342,
    '''Feminism''': 164_338,
    '''Writing''': 11_992,
    '''Atheism''': 192_263,
    '''Netflix''': 48_616,
    '''Computing''': 39_639,
    '''Opinion''': 43_213,
    '''Alone''': 44_967,
    '''Funny''': 58_917,
    '''Gaming''': 40_358,
    '''Human''': 4_088,
    '''India''': 1_331,
    '''Joker''': 77_138,
    '''Diet''': 36_206,
    '''Legal''': 11_859,
    '''Norman''': 4_939,
    '''Tip''': 72_689,
    '''Weight''': 52_343,
    '''Movies''': 46_273,
    '''Running''': 23_425,
    '''Science''': 2_090,
    '''Horror''': 37_793,
    '''Confession''': 60_572,
    '''Finance''': 12_250,
    '''Politics''': 16_360,
    '''Scary''': 191_985,
    '''Support''': 12_654,
    '''Technologies''': 32_516,
    '''Teenage''': 66_160,
    '''Event''': 32_769,
    '''Learned''': 67_460,
    '''Notion''': 182_770,
    '''Wikipedia''': 37_583,
    '''Books''': 6_665,
    '''Extract''': 76_050,
    '''Confessions''': 102_701,
    '''Conspiracy''': 75_932,
    '''Links''': 63_674,
    '''Narcissus''': 150_425,
    '''Relationship''': 54_766,
    '''Relationships''': 134_796,
    '''Reviews''': 41_671,
    '''News''': 4_256,
    '''Translation''': 26_820,
    '''multilingual''': 128_406,
}


def      UpperCAmelCase__				(    SCREAMING_SNAKE_CASE	:		Optional[Any]     ):







      '''simple docstring'''




      lowerCAmelCase    		=			set()
      lowerCAmelCase    		=			word[0]
      for char in word[1:]:
            pairs.add((prev_char, char)     )
            lowerCAmelCase    		=			char

      lowerCAmelCase    		=			set(SCREAMING_SNAKE_CASE     )
      return pairs






class 						lowercase   (   UpperCAmelCase_   ):
      _SCREAMING_SNAKE_CASE							      = VOCAB_FILES_NAMES
      _SCREAMING_SNAKE_CASE							      = PRETRAINED_VOCAB_FILES_MAP
      _SCREAMING_SNAKE_CASE							      = PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      _SCREAMING_SNAKE_CASE							      = CONTROL_CODES





      def __init__(			self ,		lowercase ,		lowercase ,		lowercase="<unk>" ,		**lowercase					)						->	Optional[Any]:
            super().__init__(unk_token=__lowerCamelCase ,		**__lowerCamelCase					)

            with open(__lowerCamelCase ,		encoding="""utf-8"""					) as vocab_handle:
                  lowerCAmelCase    		=			json.load(__lowerCamelCase					)
            lowerCAmelCase    		=			{v: k for k, v in self.encoder.items()}
            with open(__lowerCamelCase ,		encoding="""utf-8"""					) as merges_handle:
                  lowerCAmelCase    		=			merges_handle.read().split("""\n"""					)[1:-1]
            lowerCAmelCase    		=			[tuple(merge.split()					) for merge in merges]
            lowerCAmelCase    		=			dict(zip(__lowerCamelCase ,		range(len(__lowerCamelCase					)					)					)					)
            lowerCAmelCase    		=			{}





      @property
      def        _snake_case     (			self					)						->	Optional[Any]:
            return len(self.encoder					)





      def        _snake_case     (			self					)						->	List[str]:
            return dict(self.encoder ,		**self.added_tokens_encoder					)





      def        _snake_case     (			self ,		lowercase					)						->	Optional[int]:




            if token in self.cache:
                  return self.cache[token]
            lowerCAmelCase    		=			tuple(__lowerCamelCase					)
            lowerCAmelCase    		=			tuple(list(word[:-1]					) + [word[-1] + """</w>"""]					)
            lowerCAmelCase    		=			get_pairs(__lowerCamelCase					)

            if not pairs:
                  return token

            while True:
                  lowerCAmelCase    		=			min(__lowerCamelCase ,		key=lambda lowercase					: self.bpe_ranks.get(__lowerCamelCase ,		float("""inf"""					)					)					)
                  if bigram not in self.bpe_ranks:
                        break
                  lowerCAmelCase		,						lowerCAmelCase    		=			bigram
                  lowerCAmelCase    		=			[]
                  lowerCAmelCase    		=			0
                  while i < len(__lowerCamelCase					):
                        try:
                              lowerCAmelCase    		=			word.index(__lowerCamelCase ,		__lowerCamelCase					)
                        except ValueError:
                              new_word.extend(word[i:]					)
                              break
                        else:
                              new_word.extend(word[i:j]					)
                              lowerCAmelCase    		=			j

                        if word[i] == first and i < len(__lowerCamelCase					) - 1 and word[i + 1] == second:
                              new_word.append(first + second					)
                              i += 2
                        else:
                              new_word.append(word[i]					)
                              i += 1
                  lowerCAmelCase    		=			tuple(__lowerCamelCase					)
                  lowerCAmelCase    		=			new_word
                  if len(__lowerCamelCase					) == 1:
                        break
                  else:
                        lowerCAmelCase    		=			get_pairs(__lowerCamelCase					)
            lowerCAmelCase    		=			"""@@ """.join(__lowerCamelCase					)
            lowerCAmelCase    		=			word[:-4]
            lowerCAmelCase    		=			word
            return word





      def        _snake_case     (			self ,		lowercase					)						->	Optional[int]:
            lowerCAmelCase    		=			[]

            lowerCAmelCase    		=			re.findall(r"""\S+\n?""" ,		__lowerCamelCase					)

            for token in words:
                  split_tokens.extend(list(self.bpe(__lowerCamelCase					).split(""" """					)					)					)
            return split_tokens





      def        _snake_case     (			self ,		lowercase					)						->	Tuple:
            return self.encoder.get(__lowerCamelCase ,		self.encoder.get(self.unk_token					)					)





      def        _snake_case     (			self ,		lowercase					)						->	List[str]:
            return self.decoder.get(__lowerCamelCase ,		self.unk_token					)





      def        _snake_case     (			self ,		lowercase					)						->	Optional[int]:
            lowerCAmelCase    		=			""" """.join(__lowerCamelCase					).replace("""@@ """ ,		""""""					).strip()
            return out_string





      def        _snake_case     (			self ,		lowercase ,		lowercase = None					)						->	Tuple[str]:




            if not os.path.isdir(__lowerCamelCase					):
                  logger.error(f'Vocabulary path ({save_directory}) should be a directory'					)
                  return
            lowerCAmelCase    		=			os.path.join(
                __lowerCamelCase ,		(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]					)
            lowerCAmelCase    		=			os.path.join(
                __lowerCamelCase ,		(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""merges_file"""]					)

            with open(__lowerCamelCase ,		"""w""" ,		encoding="""utf-8"""					) as f:
                  f.write(json.dumps(self.encoder ,		indent=2 ,		sort_keys=__lowerCamelCase ,		ensure_ascii=__lowerCamelCase					) + """\n"""					)

            lowerCAmelCase    		=			0
            with open(__lowerCamelCase ,		"""w""" ,		encoding="""utf-8"""					) as writer:
                  writer.write("""#version: 0.2\n"""					)
                  for bpe_tokens, token_index in sorted(self.bpe_ranks.items() ,		key=lambda lowercase					: kv[1]					):
                        if index != token_index:
                              logger.warning(
                                  f'Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'
                                  """ Please check that the tokenizer is not corrupted!"""					)
                              lowerCAmelCase    		=			token_index
                        writer.write(""" """.join(__lowerCamelCase					) + """\n"""					)
                        index += 1

            return vocab_file, merge_file

      # def decode(self, token_ids, skip_special_tokens=False, clean_up_tokenization_spaces=True):
      #     filtered_tokens = ' '.join(self.convert_ids_to_tokens(token_ids, skip_special_tokens=skip_special_tokens))
      #     tokens_generated_so_far = re.sub('(@@ )', '', string=filtered_tokens)
      #     tokens_generated_so_far = re.sub('(@@ ?$)', '', string=tokens_generated_so_far)
      #     return ''.join(tokens_generated_so_far)













import os
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen, xsplitext
from ..table import array_cast
from ..utils.py_utils import no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
		from .features import FeatureType


SCREAMING_SNAKE_CASE_	,SCREAMING_SNAKE_CASE_	,SCREAMING_SNAKE_CASE_   :      Any			=   False, False, False

@dataclass
class 		snake_case_   :







	'''simple docstring'''
	__UpperCamelCase   =		None
	__UpperCamelCase   =		True
	__UpperCamelCase   =		True
	__UpperCamelCase   =		None
	# Automatically constructed
	__UpperCamelCase   =		"dict"
	__UpperCamelCase   =		pa.struct({'''bytes''': pa.binary(), '''path''': pa.string()}						)
	__UpperCamelCase   =		field(default='''Audio'''     ,     init=UpperCAmelCase_     ,     repr=UpperCAmelCase_						)




	def __call__(						self							:     Optional[Any]   )       -> List[str]:

			'''simple docstring'''




			return self.pa_type




	def    UpperCAmelCase			(						self							:     List[Any]					,      __lowerCamelCase							:     Union[str, bytes, dict]   )       -> dict:

			'''simple docstring'''




			try:
					import soundfile as sf  # soundfile is a dependency of librosa, needed to decode audio files.
			except ImportError as err:
					raise ImportError('To support encoding audio data, please install \'soundfile\'.'   ) from err
			if isinstance(__lowerCamelCase					,      __lowerCamelCase   ):
					return {"bytes": None, "path": value}
			elif isinstance(__lowerCamelCase					,      __lowerCamelCase   ):
					return {"bytes": value, "path": None}
			elif "array" in value:
					# convert the audio array to wav bytes
					__lowercase											=     BytesIO()
					sf.write(__lowerCamelCase					,      value['array']					,      value['sampling_rate']					,      format='wav'   )
					return {"bytes": buffer.getvalue(), "path": None}
			elif value.get('path'   ) is not None and os.path.isfile(value['path']   ):
					# we set "bytes": None to not duplicate the data if they're already available locally
					if value["path"].endswith('pcm'   ):
							# "PCM" only has raw audio bytes
							if value.get('sampling_rate'   ) is None:
									# At least, If you want to convert "PCM-byte" to "WAV-byte", you have to know sampling rate
									raise KeyError('To use PCM files, please specify a \'sampling_rate\' in Audio object'   )
							if value.get('bytes'   ):
									# If we already had PCM-byte, we don`t have to make "read file, make bytes" (just use it!)
									__lowercase											=     np.frombuffer(value['bytes']					,      dtype=np.intaa   ).astype(np.floataa   ) / 32_767
							else:
									__lowercase											=     np.memmap(value['path']					,      dtype='h'					,      mode='r'   ).astype(np.floataa   ) / 32_767

							__lowercase											=     BytesIO(bytes()   )
							sf.write(__lowerCamelCase					,      __lowerCamelCase					,      value['sampling_rate']					,      format='wav'   )
							return {"bytes": buffer.getvalue(), "path": None}
					else:
							return {"bytes": None, "path": value.get('path'   )}
			elif value.get('bytes'   ) is not None or value.get('path'   ) is not None:
					# store the audio bytes, and path is used to infer the audio format using the file extension
					return {"bytes": value.get('bytes'   ), "path": value.get('path'   )}
			else:
					raise ValueError(
					    F"An audio sample should have one of 'path' or 'bytes' but they are missing or None in {value}."   )




	def    UpperCAmelCase			(						self							:     Any					,      __lowerCamelCase							:     dict					,      __lowerCamelCase							:     Optional[Dict[str, Union[str, bool, None]]] = None   )       -> dict:

			'''simple docstring'''




			if not self.decode:
					raise RuntimeError('Decoding is disabled for this feature. Please use Audio(decode=True) instead.'   )

			__lowercase						,      __lowercase											=     (value['path'], BytesIO(value['bytes']   )) if value['bytes'] is not None else (value['path'], None)
			if path is None and file is None:
					raise ValueError(F"An audio sample should have one of 'path' or 'bytes' but both are None in {value}."   )

			try:
					import librosa
					import soundfile as sf
			except ImportError as err:
					raise ImportError('To support decoding audio files, please install \'librosa\' and \'soundfile\'.'   ) from err

			__lowercase											=     xsplitext(__lowerCamelCase   )[1][1:].lower() if path is not None else None
			if not config.IS_OPUS_SUPPORTED and audio_format == "opus":
					raise RuntimeError(
					    'Decoding \'opus\' files requires system library \'libsndfile\'>=1.0.31, '
					    'You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. '   )
			elif not config.IS_MP3_SUPPORTED and audio_format == "mp3":
					raise RuntimeError(
					    'Decoding \'mp3\' files requires system library \'libsndfile\'>=1.1.0, '
					    'You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. '   )

			if file is None:
					__lowercase											=     token_per_repo_id or {}
					__lowercase											=     path.split('::'   )[-1]
					try:
							__lowercase											=     string_to_dict(__lowerCamelCase					,      config.HUB_DATASETS_URL   )['repo_id']
							__lowercase											=     token_per_repo_id[repo_id]
					except (ValueError, KeyError):
							__lowercase											=     None

					with xopen(__lowerCamelCase					,      'rb'					,      use_auth_token=__lowerCamelCase   ) as f:
							__lowercase						,      __lowercase											=     sf.read(__lowerCamelCase   )

			else:
					__lowercase						,      __lowercase											=     sf.read(__lowerCamelCase   )

			__lowercase											=     array.T
			if self.mono:
					__lowercase											=     librosa.to_mono(__lowerCamelCase   )
			if self.sampling_rate and self.sampling_rate != sampling_rate:
					__lowercase											=     librosa.resample(__lowerCamelCase					,      orig_sr=__lowerCamelCase					,      target_sr=self.sampling_rate   )
					__lowercase											=     self.sampling_rate

			return {"path": path, "array": array, "sampling_rate": sampling_rate}




	def    UpperCAmelCase			(						self							:     int   )       -> Union["FeatureType", Dict[str, "FeatureType"]]:

			'''simple docstring'''




			from .features import Value

			if self.decode:
					raise ValueError('Cannot flatten a decoded Audio feature.'   )
			return {
			    "bytes": Value('binary'   ),
			    "path": Value('string'   ),
			}




	def    UpperCAmelCase			(						self							:     List[Any]					,      __lowerCamelCase							:     Union[pa.StringArray, pa.StructArray]   )       -> pa.StructArray:

			'''simple docstring'''




			if pa.types.is_string(storage.type   ):
					__lowercase											=     pa.array([None] * len(__lowerCamelCase   )					,      type=pa.binary()   )
					__lowercase											=     pa.StructArray.from_arrays([bytes_array, storage]					,      ['bytes', 'path']					,      mask=storage.is_null()   )
			elif pa.types.is_binary(storage.type   ):
					__lowercase											=     pa.array([None] * len(__lowerCamelCase   )					,      type=pa.string()   )
					__lowercase											=     pa.StructArray.from_arrays([storage, path_array]					,      ['bytes', 'path']					,      mask=storage.is_null()   )
			elif pa.types.is_struct(storage.type   ) and storage.type.get_all_field_indices('array'   ):
					__lowercase											=     pa.array([Audio().encode_example(__lowerCamelCase   ) if x is not None else None for x in storage.to_pylist()]   )
			elif pa.types.is_struct(storage.type   ):
					if storage.type.get_field_index('bytes'   ) >= 0:
							__lowercase											=     storage.field('bytes'   )
					else:
							__lowercase											=     pa.array([None] * len(__lowerCamelCase   )					,      type=pa.binary()   )
					if storage.type.get_field_index('path'   ) >= 0:
							__lowercase											=     storage.field('path'   )
					else:
							__lowercase											=     pa.array([None] * len(__lowerCamelCase   )					,      type=pa.string()   )
					__lowercase											=     pa.StructArray.from_arrays([bytes_array, path_array]					,      ['bytes', 'path']					,      mask=storage.is_null()   )
			return array_cast(__lowerCamelCase					,      self.pa_type   )






	def    UpperCAmelCase			(						self							:     Union[str, Any]					,      __lowerCamelCase							:     pa.StructArray   )       -> pa.StructArray:

			'''simple docstring'''




			@no_op_if_value_is_null
			def path_to_bytes(__lowerCamelCase							:     Any   ):
					with xopen(__lowerCamelCase					,      'rb'   ) as f:
							__lowercase											=     f.read()
					return bytes_

			__lowercase											=     pa.array(
			    [
			        (path_to_bytes(x['path']   ) if x['bytes'] is None else x['bytes']) if x is not None else None
			        for x in storage.to_pylist()
			    ]					,      type=pa.binary()					,      )
			__lowercase											=     pa.array(
			    [os.path.basename(__lowerCamelCase   ) if path is not None else None for path in storage.field('path'   ).to_pylist()]					,      type=pa.string()					,      )
			__lowercase											=     pa.StructArray.from_arrays([bytes_array, path_array]					,      ['bytes', 'path']					,      mask=bytes_array.is_null()   )
			return array_cast(__lowerCamelCase					,      self.pa_type   )




