





"""simple docstring"""




import json
import os
from typing import Optional, Tuple

import regex as re

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


a		:				List[Any]  			=    logging.get_logger(__name__)

a		:				Optional[Any]  			=    {
    '''vocab_file''': '''vocab.json''',
    '''merges_file''': '''merges.txt''',
}

a		:				Any  			=    {
    '''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'''},
}

a		:				Union[str, Any]  			=    {
    '''ctrl''': 256,
}

a		:				List[Any]  			=    {
    '''Pregnancy''': 168629,
    '''Christianity''': 7675,
    '''Explain''': 106423,
    '''Fitness''': 63440,
    '''Saving''': 63163,
    '''Ask''': 27171,
    '''Ass''': 95985,
    '''Joke''': 163509,
    '''Questions''': 45622,
    '''Thoughts''': 49605,
    '''Retail''': 52342,
    '''Feminism''': 164338,
    '''Writing''': 11992,
    '''Atheism''': 192263,
    '''Netflix''': 48616,
    '''Computing''': 39639,
    '''Opinion''': 43213,
    '''Alone''': 44967,
    '''Funny''': 58917,
    '''Gaming''': 40358,
    '''Human''': 4088,
    '''India''': 1331,
    '''Joker''': 77138,
    '''Diet''': 36206,
    '''Legal''': 11859,
    '''Norman''': 4939,
    '''Tip''': 72689,
    '''Weight''': 52343,
    '''Movies''': 46273,
    '''Running''': 23425,
    '''Science''': 2090,
    '''Horror''': 37793,
    '''Confession''': 60572,
    '''Finance''': 12250,
    '''Politics''': 16360,
    '''Scary''': 191985,
    '''Support''': 12654,
    '''Technologies''': 32516,
    '''Teenage''': 66160,
    '''Event''': 32769,
    '''Learned''': 67460,
    '''Notion''': 182770,
    '''Wikipedia''': 37583,
    '''Books''': 6665,
    '''Extract''': 76050,
    '''Confessions''': 102701,
    '''Conspiracy''': 75932,
    '''Links''': 63674,
    '''Narcissus''': 150425,
    '''Relationship''': 54766,
    '''Relationships''': 134796,
    '''Reviews''': 41671,
    '''News''': 4256,
    '''Translation''': 26820,
    '''multilingual''': 128406,
}






def    _SCREAMING_SNAKE_CASE	(			_lowercase		:					str      )				->Any:






	'''simple docstring'''






	a   :	Any			=       set()
	a   :	Optional[int]			=       word[0]
	for char in word[1:]:
		pairs.add((prev_char, char)      )
		a   :	Optional[Any]			=       char

	a   :	List[Any]			=       set(_lowercase      )
	return pairs







class  __UpperCamelCase       (		a__		):
							lowerCamelCase     :      Union[str, Any]         =VOCAB_FILES_NAMES
							lowerCamelCase     :      Any         =PRETRAINED_VOCAB_FILES_MAP
							lowerCamelCase     :      Union[str, Any]         =PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							lowerCamelCase     :      Union[str, Any]         =CONTROL_CODES






							def __init__(     self   ,      lowerCAmelCase__   ,      lowerCAmelCase__   ,      lowerCAmelCase__="<unk>"   ,      **lowerCAmelCase__   )			->				List[str]:
								super().__init__(unk_token=lowerCAmelCase__   ,      **lowerCAmelCase__   )

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






							@property
							def 					__a     (     self   )			->				List[str]:
								return len(self.encoder   )






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






							def 					__a     (     self   ,      lowerCAmelCase__   )			->				Optional[int]:
								if token in self.cache:
									return self.cache[token]
								a   :	int			=       tuple(lowerCAmelCase__   )
								a   :	Optional[int]			=       tuple(list(word[:-1]   ) + [word[-1] + "</w>"]   )
								a   :	Dict			=       get_pairs(lowerCAmelCase__   )

								if not pairs:
									return token

								while True:
									a   :	Union[str, Any]			=       min(lowerCAmelCase__   ,      key=lambda lowerCAmelCase__   : self.bpe_ranks.get(lowerCAmelCase__   ,      float("inf"   )   )   )
									if bigram not in self.bpe_ranks:
										break
									a, a   :	Dict			=       bigram
									a   :	Union[str, Any]			=       []
									a   :	int			=       0
									while i < len(lowerCAmelCase__   ):
										try:
											a   :	List[str]			=       word.index(lowerCAmelCase__   ,      lowerCAmelCase__   )
										except ValueError:
											new_word.extend(word[i:]   )
											break
										else:
											new_word.extend(word[i:j]   )
											a   :	List[str]			=       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
									a   :	int			=       tuple(lowerCAmelCase__   )
									a   :	Union[str, Any]			=       new_word
									if len(lowerCAmelCase__   ) == 1:
										break
									else:
										a   :	Optional[Any]			=       get_pairs(lowerCAmelCase__   )
								a   :	str			=       "@@ ".join(lowerCAmelCase__   )
								a   :	List[Any]			=       word[:-4]
								a   :	int			=       word
								return word






							def 					__a     (     self   ,      lowerCAmelCase__   )			->				Any:
								a   :	str			=       []

								a   :	List[Any]			=       re.findall(R"\S+\n?"   ,      lowerCAmelCase__   )

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






							def 					__a     (     self   ,      lowerCAmelCase__   )			->				Optional[Any]:
								return self.encoder.get(lowerCAmelCase__   ,      self.encoder.get(self.unk_token   )   )






							def 					__a     (     self   ,      lowerCAmelCase__   )			->				str:
								return self.decoder.get(lowerCAmelCase__   ,      self.unk_token   )






							def 					__a     (     self   ,      lowerCAmelCase__   )			->				int:
								a   :	List[str]			=       " ".join(lowerCAmelCase__   ).replace("@@ "   ,      ""   ).strip()
								return out_string






							def 					__a     (     self   ,      lowerCAmelCase__   ,      lowerCAmelCase__ = None   )			->				Tuple[str]:
								if not os.path.isdir(lowerCAmelCase__   ):
									logger.error(f"""Vocabulary path ({save_directory}) should be a directory"""   )
									return
								a   :	Optional[int]			=       os.path.join(
								    lowerCAmelCase__   ,      (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]   )
								a   :	List[str]			=       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"   )

								a   :	Tuple			=       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 lowerCAmelCase__   : 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!"   )
											a   :	Dict			=       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)









"""simple docstring"""




import flax.linen as nn
import jax
import jax.numpy as jnp


class  __UpperCamelCase       (		nn.Module		):
							lowerCamelCase     :      int
							lowerCamelCase     :      jnp.dtype         =jnp.floataa






							def 					__a     (     self   )			->				Tuple:
								a   :	str			=       nn.Conv(
								    self.out_channels   ,      kernel_size=(3, 3)   ,      strides=(1, 1)   ,      padding=((1, 1), (1, 1))   ,      dtype=self.dtype   ,      )






							def __call__(     self   ,      lowerCAmelCase__   )			->				Optional[Any]:
								a, a, a, a   :	List[str]			=       hidden_states.shape
								a   :	List[Any]			=       jax.image.resize(
								    lowerCAmelCase__   ,      shape=(batch, height * 2, width * 2, channels)   ,      method="nearest"   ,      )
								a   :	List[str]			=       self.conv(lowerCAmelCase__   )
								return hidden_states


class  __UpperCamelCase       (		nn.Module		):
							lowerCamelCase     :      int
							lowerCamelCase     :      jnp.dtype         =jnp.floataa






							def 					__a     (     self   )			->				Dict:
								a   :	Optional[Any]			=       nn.Conv(
								    self.out_channels   ,      kernel_size=(3, 3)   ,      strides=(2, 2)   ,      padding=((1, 1), (1, 1))   ,      dtype=self.dtype   ,      )






							def __call__(     self   ,      lowerCAmelCase__   )			->				Tuple:
								# pad = ((0, 0), (0, 1), (0, 1), (0, 0))  # pad height and width dim
								# hidden_states = jnp.pad(hidden_states, pad_width=pad)
								a   :	Tuple			=       self.conv(lowerCAmelCase__   )
								return hidden_states


class  __UpperCamelCase       (		nn.Module		):
							lowerCamelCase     :      int
							lowerCamelCase     :      int         =None
							lowerCamelCase     :      float         =0.0
							lowerCamelCase     :      bool         =None
							lowerCamelCase     :      jnp.dtype         =jnp.floataa






							def 					__a     (     self   )			->				int:
								a   :	Dict			=       self.in_channels if self.out_channels is None else self.out_channels

								a   :	Union[str, Any]			=       nn.GroupNorm(num_groups=32   ,      epsilon=1E-5   )
								a   :	List[Any]			=       nn.Conv(
								    lowerCAmelCase__   ,      kernel_size=(3, 3)   ,      strides=(1, 1)   ,      padding=((1, 1), (1, 1))   ,      dtype=self.dtype   ,      )

								a   :	List[Any]			=       nn.Dense(lowerCAmelCase__   ,      dtype=self.dtype   )

								a   :	Union[str, Any]			=       nn.GroupNorm(num_groups=32   ,      epsilon=1E-5   )
								a   :	Optional[int]			=       nn.Dropout(self.dropout_prob   )
								a   :	Dict			=       nn.Conv(
								    lowerCAmelCase__   ,      kernel_size=(3, 3)   ,      strides=(1, 1)   ,      padding=((1, 1), (1, 1))   ,      dtype=self.dtype   ,      )

								a   :	Union[str, Any]			=       self.in_channels != out_channels if self.use_nin_shortcut is None else self.use_nin_shortcut

								a   :	List[str]			=       None
								if use_nin_shortcut:
									a   :	Optional[Any]			=       nn.Conv(
									    lowerCAmelCase__   ,      kernel_size=(1, 1)   ,      strides=(1, 1)   ,      padding="VALID"   ,      dtype=self.dtype   ,      )






							def __call__(     self   ,      lowerCAmelCase__   ,      lowerCAmelCase__   ,      lowerCAmelCase__=True   )			->				str:
								a   :	int			=       hidden_states
								a   :	Tuple			=       self.norma(lowerCAmelCase__   )
								a   :	Any			=       nn.swish(lowerCAmelCase__   )
								a   :	int			=       self.conva(lowerCAmelCase__   )

								a   :	int			=       self.time_emb_proj(nn.swish(lowerCAmelCase__   )   )
								a   :	Tuple			=       jnp.expand_dims(jnp.expand_dims(lowerCAmelCase__   ,      1   )   ,      1   )
								a   :	Dict			=       hidden_states + temb

								a   :	str			=       self.norma(lowerCAmelCase__   )
								a   :	List[Any]			=       nn.swish(lowerCAmelCase__   )
								a   :	List[str]			=       self.dropout(lowerCAmelCase__   ,      lowerCAmelCase__   )
								a   :	List[str]			=       self.conva(lowerCAmelCase__   )

								if self.conv_shortcut is not None:
									a   :	Tuple			=       self.conv_shortcut(lowerCAmelCase__   )

								return hidden_states + residual


