
"""simple docstring"""






import json
import os
from functools import lru_cache
from typing import Dict, List, Optional, Tuple, Union

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...tokenization_utils_base import BatchEncoding, EncodedInput
from ...utils import PaddingStrategy, logging


lowerCamelCase__ :       Tuple						     =			logging.get_logger(__name__)


lowerCamelCase__ :       Tuple						     =			{"vocab_file": "vocab.json", "merges_file": "merges.txt"}

# See all LED models at https://huggingface.co/models?filter=LED
lowerCamelCase__ :       int						     =			{
    "vocab_file": {
        "allenai/led-base-16384": "https://huggingface.co/allenai/led-base-16384/resolve/main/vocab.json",
    },
    "merges_file": {
        "allenai/led-base-16384": "https://huggingface.co/allenai/led-base-16384/resolve/main/merges.txt",
    },
    "tokenizer_file": {
        "allenai/led-base-16384": "https://huggingface.co/allenai/led-base-16384/resolve/main/tokenizer.json",
    },
}

lowerCamelCase__ :       List[str]						     =			{
    "allenai/led-base-16384": 16384,
}





@lru_cache()
# Copied from transformers.models.bart.tokenization_bart.bytes_to_unicode
def __A						(   )->							Tuple:




		'''simple docstring'''






		SCREAMING_SNAKE_CASE       :     Any				= (
		    list(range(ord('''!'''     )		,		ord('''~'''     ) + 1     )     ) + list(range(ord('''¡'''     )		,		ord('''¬'''     ) + 1     )     ) + list(range(ord('''®'''     )		,		ord('''ÿ'''     ) + 1     )     )
		)
		SCREAMING_SNAKE_CASE       :     int				= bs[:]
		SCREAMING_SNAKE_CASE       :     Union[str, Any]				= 0
		for b in range(2**8     ):
				if b not in bs:
						bs.append(a_     )
						cs.append(2**8 + n     )
						n += 1
		SCREAMING_SNAKE_CASE       :     Optional[int]				= [chr(a_     ) for n in cs]
		return dict(zip(a_		,		a_     )     )





def __A						(   a_ :							Any     )->							str:




		'''simple docstring'''






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




class 				lowercase__(				_UpperCAmelCase   ):
	'''simple docstring'''

	UpperCamelCase				=  VOCAB_FILES_NAMES
	UpperCamelCase				=  PRETRAINED_VOCAB_FILES_MAP
	UpperCamelCase				=  PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	UpperCamelCase				=  ["""input_ids""", """attention_mask"""]





	def __init__(						self    :Union[str, Any]		, lowerCamelCase_    :Optional[int]		, lowerCamelCase_    :int		, lowerCamelCase_    :int="replace"		, lowerCamelCase_    :Tuple="<s>"		, lowerCamelCase_    :List[str]="</s>"		, lowerCamelCase_    :List[str]="</s>"		, lowerCamelCase_    :Tuple="<s>"		, lowerCamelCase_    :List[str]="<unk>"		, lowerCamelCase_    :Dict="<pad>"		, lowerCamelCase_    :Any="<mask>"		, lowerCamelCase_    :int=False		, **lowerCamelCase_    :Union[str, Any]		, )							->			List[Any]:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     List[str]				= AddedToken(lowerCamelCase_		, lstrip=lowerCamelCase_		, rstrip=lowerCamelCase_					) if isinstance(lowerCamelCase_		, lowerCamelCase_					) else bos_token
			SCREAMING_SNAKE_CASE       :     Optional[Any]				= AddedToken(lowerCamelCase_		, lstrip=lowerCamelCase_		, rstrip=lowerCamelCase_					) if isinstance(lowerCamelCase_		, lowerCamelCase_					) else eos_token
			SCREAMING_SNAKE_CASE       :     Dict				= AddedToken(lowerCamelCase_		, lstrip=lowerCamelCase_		, rstrip=lowerCamelCase_					) if isinstance(lowerCamelCase_		, lowerCamelCase_					) else sep_token
			SCREAMING_SNAKE_CASE       :     Optional[Any]				= AddedToken(lowerCamelCase_		, lstrip=lowerCamelCase_		, rstrip=lowerCamelCase_					) if isinstance(lowerCamelCase_		, lowerCamelCase_					) else cls_token
			SCREAMING_SNAKE_CASE       :     Optional[Any]				= AddedToken(lowerCamelCase_		, lstrip=lowerCamelCase_		, rstrip=lowerCamelCase_					) if isinstance(lowerCamelCase_		, lowerCamelCase_					) else unk_token
			SCREAMING_SNAKE_CASE       :     Union[str, Any]				= AddedToken(lowerCamelCase_		, lstrip=lowerCamelCase_		, rstrip=lowerCamelCase_					) if isinstance(lowerCamelCase_		, lowerCamelCase_					) else pad_token

			# Mask token behave like a normal word, i.e. include the space before it
			SCREAMING_SNAKE_CASE       :     int				= AddedToken(lowerCamelCase_		, lstrip=lowerCamelCase_		, rstrip=lowerCamelCase_					) if isinstance(lowerCamelCase_		, lowerCamelCase_					) else mask_token

			super().__init__(
			    errors=lowerCamelCase_		, bos_token=lowerCamelCase_		, eos_token=lowerCamelCase_		, unk_token=lowerCamelCase_		, sep_token=lowerCamelCase_		, cls_token=lowerCamelCase_		, pad_token=lowerCamelCase_		, mask_token=lowerCamelCase_		, add_prefix_space=lowerCamelCase_		, **lowerCamelCase_		, )

			with open(lowerCamelCase_		, encoding='''utf-8'''					) as vocab_handle:
					SCREAMING_SNAKE_CASE       :     Optional[Any]				= json.load(lowerCamelCase_					)
			SCREAMING_SNAKE_CASE       :     Dict				= {v: k for k, v in self.encoder.items()}
			SCREAMING_SNAKE_CASE       :     str				= errors  # how to handle errors in decoding
			SCREAMING_SNAKE_CASE       :     List[str]				= bytes_to_unicode()
			SCREAMING_SNAKE_CASE       :     Dict				= {v: k for k, v in self.byte_encoder.items()}
			with open(lowerCamelCase_		, encoding='''utf-8'''					) as merges_handle:
					SCREAMING_SNAKE_CASE       :     List[str]				= merges_handle.read().split('''\n'''					)[1:-1]
			SCREAMING_SNAKE_CASE       :     Union[str, Any]				= [tuple(merge.split()					) for merge in bpe_merges]
			SCREAMING_SNAKE_CASE       :     Union[str, Any]				= dict(zip(lowerCamelCase_		, range(len(lowerCamelCase_					)					)					)					)
			SCREAMING_SNAKE_CASE       :     Dict				= {}
			SCREAMING_SNAKE_CASE       :     int				= add_prefix_space

			# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
			SCREAMING_SNAKE_CASE       :     Dict				= re.compile(R'''\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'''					)





	@property
	# Copied from transformers.models.bart.tokenization_bart.BartTokenizer.vocab_size
	def   __lowerCAmelCase (						self    :Optional[Any]					)							->			str:







			'''simple docstring'''




			return len(self.encoder					)





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







			'''simple docstring'''




			return dict(self.encoder		, **self.added_tokens_encoder					)





	def   __lowerCAmelCase (						self    :List[str]		, lowerCamelCase_    :int					)							->			List[str]:







			'''simple docstring'''




			if token in self.cache:
					return self.cache[token]
			SCREAMING_SNAKE_CASE       :     List[Any]				= tuple(lowerCamelCase_					)
			SCREAMING_SNAKE_CASE       :     Tuple				= get_pairs(lowerCamelCase_					)

			if not pairs:
					return token

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





	def   __lowerCAmelCase (						self    :Tuple		, lowerCamelCase_    :str					)							->			Tuple:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     str				= []
			for token in re.findall(self.pat		, lowerCamelCase_					):
					SCREAMING_SNAKE_CASE       :     Optional[Any]				= ''''''.join(
					    self.byte_encoder[b] for b in token.encode('''utf-8'''					)					)  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
					bpe_tokens.extend(bpe_token for bpe_token in self.bpe(lowerCamelCase_					).split(''' '''					)					)
			return bpe_tokens





	def   __lowerCAmelCase (						self    :List[Any]		, lowerCamelCase_    :Optional[int]					)							->			Dict:







			'''simple docstring'''




			return self.encoder.get(lowerCamelCase_		, self.encoder.get(self.unk_token					)					)





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







			'''simple docstring'''




			return self.decoder.get(lowerCamelCase_					)





	def   __lowerCAmelCase (						self    :Tuple		, lowerCamelCase_    :Dict					)							->			Union[str, Any]:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     Optional[Any]				= ''''''.join(lowerCamelCase_					)
			SCREAMING_SNAKE_CASE       :     Any				= bytearray([self.byte_decoder[c] for c in text]					).decode('''utf-8'''		, errors=self.errors					)
			return text





	def   __lowerCAmelCase (						self    :Optional[Any]		, lowerCamelCase_    :str		, lowerCamelCase_    :Optional[str] = None					)							->			Tuple[str]:







			'''simple docstring'''




			if not os.path.isdir(lowerCamelCase_					):
					logger.error(f"Vocabulary path ({save_directory}) should be a directory"					)
					return
			SCREAMING_SNAKE_CASE       :     List[Any]				= os.path.join(
			    lowerCamelCase_		, (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']					)
			SCREAMING_SNAKE_CASE       :     Union[str, Any]				= 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'''					)

			SCREAMING_SNAKE_CASE       :     Any				= 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!'''					)
									SCREAMING_SNAKE_CASE       :     Union[str, Any]				= token_index
							writer.write(''' '''.join(lowerCamelCase_					) + '''\n'''					)
							index += 1

			return vocab_file, merge_file





	def   __lowerCAmelCase (						self    :Optional[Any]		, lowerCamelCase_    :List[int]		, lowerCamelCase_    :Optional[List[int]] = None					)							->			List[int]:







			'''simple docstring'''




			if token_ids_a is None:
					return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
			SCREAMING_SNAKE_CASE       :     str				= [self.cls_token_id]
			SCREAMING_SNAKE_CASE       :     str				= [self.sep_token_id]
			return cls + token_ids_a + sep + sep + token_ids_a + sep





	def   __lowerCAmelCase (						self    :Union[str, Any]		, lowerCamelCase_    :List[int]		, lowerCamelCase_    :Optional[List[int]] = None		, lowerCamelCase_    :bool = False					)							->			List[int]:







			'''simple docstring'''




			if already_has_special_tokens:
					return super().get_special_tokens_mask(
					    token_ids_a=lowerCamelCase_		, token_ids_a=lowerCamelCase_		, already_has_special_tokens=lowerCamelCase_					)

			if token_ids_a is None:
					return [1] + ([0] * len(lowerCamelCase_					)) + [1]
			return [1] + ([0] * len(lowerCamelCase_					)) + [1, 1] + ([0] * len(lowerCamelCase_					)) + [1]





	def   __lowerCAmelCase (						self    :Tuple		, lowerCamelCase_    :List[int]		, lowerCamelCase_    :Optional[List[int]] = None					)							->			List[int]:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     List[str]				= [self.sep_token_id]
			SCREAMING_SNAKE_CASE       :     Any				= [self.cls_token_id]

			if token_ids_a is None:
					return len(cls + token_ids_a + sep					) * [0]
			return len(cls + token_ids_a + sep + sep + token_ids_a + sep					) * [0]





	def   __lowerCAmelCase (						self    :int		, lowerCamelCase_    :int		, lowerCamelCase_    :List[Any]=False		, **lowerCamelCase_    :List[Any]					)							->			Optional[Any]:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     List[str]				= kwargs.pop('''add_prefix_space'''		, self.add_prefix_space					)
			if (is_split_into_words or add_prefix_space) and (len(lowerCamelCase_					) > 0 and not text[0].isspace()):
					SCREAMING_SNAKE_CASE       :     List[Any]				= ''' ''' + text
			return (text, kwargs)







	def   __lowerCAmelCase (						self    :int		, lowerCamelCase_    :Union[Dict[str, EncodedInput], BatchEncoding]		, lowerCamelCase_    :Optional[int] = None		, lowerCamelCase_    :PaddingStrategy = PaddingStrategy.DO_NOT_PAD		, lowerCamelCase_    :Optional[int] = None		, lowerCamelCase_    :Optional[bool] = None		, )							->			dict:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     Optional[int]				= super()._pad(
			    encoded_inputs=lowerCamelCase_		, max_length=lowerCamelCase_		, padding_strategy=lowerCamelCase_		, pad_to_multiple_of=lowerCamelCase_		, return_attention_mask=lowerCamelCase_		, )

			# Load from model defaults
			if return_attention_mask is None:
					SCREAMING_SNAKE_CASE       :     List[Any]				= '''attention_mask''' in self.model_input_names

			if return_attention_mask and "global_attention_mask" in encoded_inputs:
					SCREAMING_SNAKE_CASE       :     int				= encoded_inputs[self.model_input_names[0]]
					# `global_attention_mask` need to have the same length as other (sequential) inputs.
					SCREAMING_SNAKE_CASE       :     str				= len(encoded_inputs['''global_attention_mask''']					) != len(lowerCamelCase_					)

					if needs_to_be_padded:
							SCREAMING_SNAKE_CASE       :     Any				= len(lowerCamelCase_					) - len(encoded_inputs['''global_attention_mask''']					)

							if self.padding_side == "right":
									# Use `-1` since `0` in `global_attention_mask` means `local attention` instead of `not to attend`
									SCREAMING_SNAKE_CASE       :     Any				= (
									    encoded_inputs['''global_attention_mask'''] + [-1] * difference
									)
							elif self.padding_side == "left":
									SCREAMING_SNAKE_CASE       :     Any				= [-1] * difference + encoded_inputs[
									    '''global_attention_mask'''
									]
							else:
									raise ValueError('''Invalid padding strategy:''' + str(self.padding_side					)					)

			return encoded_inputs







"""simple docstring"""






import os
from typing import Any, Callable, Dict, List, Optional, Tuple, Union

import torch
from torch import nn

from ...models.controlnet import ControlNetModel, ControlNetOutput
from ...models.modeling_utils import ModelMixin
from ...utils import logging


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




class 				lowercase__(				_UpperCAmelCase   ):
	'''simple docstring'''

	def __init__(						self    :List[str]		, lowerCamelCase_    :Union[List[ControlNetModel], Tuple[ControlNetModel]]					)							->			Union[str, Any]:







			'''simple docstring'''




			super().__init__()
			SCREAMING_SNAKE_CASE       :     Optional[Any]				= nn.ModuleList(lowerCamelCase_					)





	def   __lowerCAmelCase (						self    :Any		, lowerCamelCase_    :torch.FloatTensor		, lowerCamelCase_    :Union[torch.Tensor, float, int]		, lowerCamelCase_    :torch.Tensor		, lowerCamelCase_    :List[torch.tensor]		, lowerCamelCase_    :List[float]		, lowerCamelCase_    :Optional[torch.Tensor] = None		, lowerCamelCase_    :Optional[torch.Tensor] = None		, lowerCamelCase_    :Optional[torch.Tensor] = None		, lowerCamelCase_    :Optional[Dict[str, Any]] = None		, lowerCamelCase_    :bool = False		, lowerCamelCase_    :bool = True		, )							->			Union[ControlNetOutput, Tuple]:







			'''simple docstring'''




			for i, (image, scale, controlnet) in enumerate(zip(lowerCamelCase_		, lowerCamelCase_		, self.nets					)					):
					SCREAMING_SNAKE_CASE,					SCREAMING_SNAKE_CASE       :     List[Any]				= controlnet(
					    lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_		, )

					# merge samples
					if i == 0:
							SCREAMING_SNAKE_CASE,					SCREAMING_SNAKE_CASE       :     int				= down_samples, mid_sample
					else:
							SCREAMING_SNAKE_CASE       :     Optional[int]				= [
							    samples_prev + samples_curr
							    for samples_prev, samples_curr in zip(lowerCamelCase_		, lowerCamelCase_					)
							]
							mid_block_res_sample += mid_sample

			return down_block_res_samples, mid_block_res_sample





	def   __lowerCAmelCase (						self    :Any		, lowerCamelCase_    :Union[str, os.PathLike]		, lowerCamelCase_    :bool = True		, lowerCamelCase_    :Callable = None		, lowerCamelCase_    :bool = False		, lowerCamelCase_    :Optional[str] = None		, )							->			Optional[int]:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     Optional[Any]				= 0
			SCREAMING_SNAKE_CASE       :     Any				= save_directory
			for controlnet in self.nets:
					controlnet.save_pretrained(
					    lowerCamelCase_		, is_main_process=lowerCamelCase_		, save_function=lowerCamelCase_		, safe_serialization=lowerCamelCase_		, variant=lowerCamelCase_		, )

					idx += 1
					SCREAMING_SNAKE_CASE       :     Union[str, Any]				= model_path_to_save + f"_{idx}"







	@classmethod
	def   __lowerCAmelCase (						cls    :Dict		, lowerCamelCase_    :Optional[Union[str, os.PathLike]]		, **lowerCamelCase_    :Tuple					)							->			List[Any]:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     str				= 0
			SCREAMING_SNAKE_CASE       :     Optional[int]				= []

			# load controlnet and append to list until no controlnet directory exists anymore
			# first controlnet has to be saved under `./mydirectory/controlnet` to be compliant with `DiffusionPipeline.from_prertained`
			# second, third, ... controlnets have to be saved under `./mydirectory/controlnet_1`, `./mydirectory/controlnet_2`, ...
			SCREAMING_SNAKE_CASE       :     Dict				= pretrained_model_path
			while os.path.isdir(lowerCamelCase_					):
					SCREAMING_SNAKE_CASE       :     Tuple				= ControlNetModel.from_pretrained(lowerCamelCase_		, **lowerCamelCase_					)
					controlnets.append(lowerCamelCase_					)

					idx += 1
					SCREAMING_SNAKE_CASE       :     Union[str, Any]				= pretrained_model_path + f"_{idx}"

			logger.info(f"{len(lowerCamelCase_					)} controlnets loaded from {pretrained_model_path}."					)

			if len(lowerCamelCase_					) == 0:
					raise ValueError(
					    f"No ControlNets found under {os.path.dirname(lowerCamelCase_					)}. Expected at least {pretrained_model_path + '_0'}."					)

			return cls(lowerCamelCase_					)





