
"""simple docstring"""


import io
import os
import unicodedata
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

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


_lowercase							      = logging.get_logger(__name__)

_lowercase							      = '''▁'''

_lowercase							      = {'''vocab_file''': '''vocab.txt''', '''sentencepiece_model_ckpt''': '''sentencepiece.bpe.model'''}

_lowercase							      = {
    '''sentencepiece_model_file''': '''sentencepiece.bpe.model''',
    '''vocab_file''': '''vocab.txt''',
}

_lowercase							      = {
    '''vocab_file''': {
        '''ernie-m-base''': '''https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/vocab.txt''',
        '''ernie-m-large''': '''https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/vocab.txt''',
    },
    '''sentencepiece_model_file''': {
        '''ernie-m-base''': '''https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/sentencepiece.bpe.model''',
        '''ernie-m-large''': '''https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/sentencepiece.bpe.model''',
    },
}

_lowercase							      = {
    '''ernie-m-base''': 5_14,
    '''ernie-m-large''': 5_14,
}

_lowercase							      = {
    '''ernie-m-base''': {'''do_lower_case''': False},
    '''ernie-m-large''': {'''do_lower_case''': False},
}







class  lowerCAmelCase_ (					_lowercase			):



						'''simple docstring'''


						_lowerCamelCase:			List[str]					=			["input_ids"]

						_lowerCamelCase:			Any					=			VOCAB_FILES_NAMES
						_lowerCamelCase:			Union[str, Any]					=			PRETRAINED_INIT_CONFIGURATION
						_lowerCamelCase:			Optional[int]					=			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
						_lowerCamelCase:			Union[str, Any]					=			PRETRAINED_VOCAB_FILES_MAP
						_lowerCamelCase:			str					=			RESOURCE_FILES_NAMES


						def __init__(       self   :					List[Any]     ,A_   :					int     ,A_   :					Tuple=None     ,A_   :					List[str]=False     ,A_   :					Union[str, Any]="utf8"     ,A_   :					List[Any]="[UNK]"     ,A_   :					Optional[int]="[SEP]"     ,A_   :					str="[PAD]"     ,A_   :					int="[CLS]"     ,A_   :					str="[MASK]"     ,A_   :					Optional[Dict[str, Any]] = None     ,**A_   :					str     ,)   ->      None:
							# Mask token behave like a normal word, i.e. include the space before it and
							# is included in the raw text, there should be a match in a non-normalized sentence.

							A        =      {} if sp_model_kwargs is None else sp_model_kwargs
							super().__init__(
							    do_lower_case=A_     ,unk_token=A_     ,sep_token=A_     ,pad_token=A_     ,cls_token=A_     ,mask_token=A_     ,vocab_file=A_     ,encoding=A_     ,sp_model_kwargs=self.sp_model_kwargs     ,**A_     ,)
							A        =      do_lower_case
							A        =      sentencepiece_model_ckpt
							A        =      spm.SentencePieceProcessor(**self.sp_model_kwargs					)
							self.sp_model.Load(A_					)

							# to mimic paddlenlp.transformers.ernie_m.tokenizer.ErnieMTokenizer functioning
							if vocab_file is not None:
								A        =      self.load_vocab(filepath=A_					)
							else:
								A        =      {self.sp_model.id_to_piece(A_					): id for id in range(self.sp_model.get_piece_size()					)}
							A        =      {v: k for k, v in self.vocab.items()}


						def   _SCREAMING_SNAKE_CASE					(       self   :					Optional[int]     ,A_   :					List[str]					)   ->      Any:
							if text is None:
								return None

							A        =      self.tokenize(A_					)
							A      ,					A        =      '', []

							for i, ch in enumerate(A_					):
								if ch in self.SP_CHAR_MAPPING:
									A        =      self.SP_CHAR_MAPPING.get(A_					)
								else:
									A        =      unicodedata.normalize('NFKC'     ,A_					)
								if self.is_whitespace(A_					):
									continue
								normalized_text += ch
								char_mapping.extend([i] * len(A_					)					)

							A      ,					A      ,					A        =      normalized_text, [], 0

							if self.do_lower_case:
								A        =      text.lower()

							for token in split_tokens:
								if token[:1] == "▁":
									A        =      token[1:]
								A        =      text[offset:].index(A_					) + offset
								A        =      start + len(A_					)

								token_mapping.append((char_mapping[start], char_mapping[end - 1] + 1)					)
								A        =      end
							return token_mapping


						@property
						def   _SCREAMING_SNAKE_CASE					(       self   :					List[Any]					)   ->      List[str]:
							return len(self.vocab					)


						def   _SCREAMING_SNAKE_CASE					(       self   :					Any					)   ->      Optional[int]:
							return dict(self.vocab     ,**self.added_tokens_encoder					)


						def __getstate__(       self   :					Optional[int]					)   ->      Optional[int]:
							A        =      self.__dict__.copy()
							A        =      None
							return state


						def __setstate__(       self   :					Any     ,A_   :					Any					)   ->      Optional[int]:
							A        =      d

							# for backward compatibility
							if not hasattr(self     ,'sp_model_kwargs'					):
								A        =      {}

							A        =      spm.SentencePieceProcessor(**self.sp_model_kwargs					)
							self.sp_model.Load(self.sentencepiece_model_ckpt					)


						def   _SCREAMING_SNAKE_CASE					(       self   :					str     ,A_   :					Union[str, Any]					)   ->      int:
							return "".join((self.SP_CHAR_MAPPING.get(A_     ,A_					) for c in text)					)


						def   _SCREAMING_SNAKE_CASE					(       self   :					List[str]     ,A_   :					Union[str, Any]     ,A_   :					List[Any]=False     ,A_   :					Any=64     ,A_   :					int=0.1					)   ->      str:

							if self.sp_model_kwargs.get('enable_sampling'					) is True:
								A        =      True
							if self.sp_model_kwargs.get('alpha'					) is not None:
								A        =      self.sp_model_kwargs.get('alpha'					)
							if self.sp_model_kwargs.get('nbest_size'					) is not None:
								A        =      self.sp_model_kwargs.get('nbest_size'					)

							if not enable_sampling:
								A        =      self.sp_model.EncodeAsPieces(A_					)
							else:
								A        =      self.sp_model.SampleEncodeAsPieces(A_     ,A_     ,A_					)
							A        =      []
							for pi, piece in enumerate(A_					):
								if piece == SPIECE_UNDERLINE:
									if not pieces[pi + 1].startswith(A_					) and pi != 0:
										new_pieces.append(A_					)
										continue
									else:
										continue
								A        =      0
								for i, chunk in enumerate(A_					):
									if chunk == SPIECE_UNDERLINE:
										continue
									if self.is_ch_char(A_					) or self.is_punct(A_					):
										if i > lst_i and piece[lst_i:i] != SPIECE_UNDERLINE:
											new_pieces.append(piece[lst_i:i]					)
										new_pieces.append(A_					)
										A        =      i + 1
									elif chunk.isdigit() and i > 0 and not piece[i - 1].isdigit():
										if i > lst_i and piece[lst_i:i] != SPIECE_UNDERLINE:
											new_pieces.append(piece[lst_i:i]					)
										A        =      i
									elif not chunk.isdigit() and i > 0 and piece[i - 1].isdigit():
										if i > lst_i and piece[lst_i:i] != SPIECE_UNDERLINE:
											new_pieces.append(piece[lst_i:i]					)
										A        =      i
								if len(A_					) > lst_i:
									new_pieces.append(piece[lst_i:]					)
							return new_pieces


						def   _SCREAMING_SNAKE_CASE					(       self   :					int     ,A_   :					Dict					)   ->      Tuple:
							A        =      ''.join(A_					).replace(A_     ,' '					).strip()
							return out_string


						def   _SCREAMING_SNAKE_CASE					(       self   :					Any     ,A_   :					Tuple					)   ->      List[Any]:
							A        =      self.convert_ids_to_tokens(A_					)
							A        =      ''.join(A_					).replace(A_     ,' '					).strip()
							return out_string


						def   _SCREAMING_SNAKE_CASE					(       self   :					Optional[Any]     ,A_   :					str					)   ->      List[Any]:
							return self.vocab.get(A_     ,self.vocab.get(self.unk_token					)					)


						def   _SCREAMING_SNAKE_CASE					(       self   :					List[str]     ,A_   :					int					)   ->      Optional[Any]:
							return self.reverse_vocab.get(A_     ,self.unk_token					)


						def   _SCREAMING_SNAKE_CASE					(       self   :					List[Any]     ,A_   :					int     ,A_   :					int=None					)   ->      List[str]:
							if token_ids_a is None:
								return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
							A        =      [self.cls_token_id]
							A        =      [self.sep_token_id]
							return _cls + token_ids_a + _sep + _sep + token_ids_a + _sep


						def   _SCREAMING_SNAKE_CASE					(       self   :					Tuple     ,A_   :					str     ,A_   :					Dict=None					)   ->      List[Any]:
							if offset_mapping_a is None:
								return [(0, 0)] + offset_mapping_a + [(0, 0)]

							return [(0, 0)] + offset_mapping_a + [(0, 0), (0, 0)] + offset_mapping_a + [(0, 0)]


						def   _SCREAMING_SNAKE_CASE					(       self   :					Tuple     ,A_   :					str     ,A_   :					List[str]=None     ,A_   :					Tuple=False					)   ->      Union[str, Any]:

							if already_has_special_tokens:
								if token_ids_a is not None:
									raise ValueError(
									    'You should not supply a second sequence if the provided sequence of '
									    'ids is already formatted with special tokens for the model.'					)
								return [1 if x in [self.sep_token_id, self.cls_token_id] else 0 for x in token_ids_a]

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


						def   _SCREAMING_SNAKE_CASE					(       self   :					List[str]     ,A_   :					List[int]     ,A_   :					Optional[List[int]] = None					)   ->      List[int]:
							# called when `add_special_tokens` is True, so align with `build_inputs_with_special_tokens` method
							if token_ids_a is None:
								# [CLS] X [SEP]
								return (len(A_					) + 2) * [0]

							# [CLS] A [SEP] [SEP] B [SEP]
							return [0] * (len(A_					) + 1) + [1] * (len(A_					) + 3)


						def   _SCREAMING_SNAKE_CASE					(       self   :					List[Any]     ,A_   :					Optional[int]					)   ->      Optional[int]:
							if "\u4e00" <= char <= "\u9fff":
								return True
							return False


						def   _SCREAMING_SNAKE_CASE					(       self   :					Optional[Any]     ,A_   :					int					)   ->      Dict:
							if ("a" <= char <= "z") or ("A" <= char <= "Z"):
								return True
							return False


						def   _SCREAMING_SNAKE_CASE					(       self   :					Any     ,A_   :					List[Any]					)   ->      List[Any]:
							if char in ",;:.?!~，；：。？！《》【】":
								return True
							return False


						def   _SCREAMING_SNAKE_CASE					(       self   :					List[str]     ,A_   :					str					)   ->      Optional[int]:
							if char == " " or char == "\t" or char == "\n" or char == "\r":
								return True
							if len(A_					) == 1:
								A        =      unicodedata.category(A_					)
								if cat == "Zs":
									return True
							return False


						def   _SCREAMING_SNAKE_CASE					(       self   :					Optional[Any]     ,A_   :					Tuple					)   ->      Any:
							A        =      {}
							with io.open(A_     ,'r'     ,encoding='utf-8'					) as f:
								for index, line in enumerate(A_					):
									A        =      line.rstrip('\n'					)
									A        =      int(A_					)

							return token_to_idx




						def   _SCREAMING_SNAKE_CASE					(       self   :					Union[str, Any]     ,A_   :					str     ,A_   :					Optional[str] = None					)   ->      Tuple[str]:
							A        =      0
							if os.path.isdir(A_					):
								A        =      os.path.join(
								    A_     ,(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']					)
							else:
								A        =      (filename_prefix + '-' if filename_prefix else '') + save_directory
							with open(A_     ,'w'     ,encoding='utf-8'					) as writer:
								for token, token_index in sorted(self.vocab.items()     ,key=lambda A_					: kv[1]					):
									if index != token_index:
										logger.warning(
										    F'Saving vocabulary to {vocab_file}: vocabulary indices are not consecutive.'
										    ' Please check that the vocabulary is not corrupted!'					)
										A        =      token_index
									writer.write(token + '\n'					)
									index += 1

							A        =      os.path.join(A_     ,'sentencepiece.bpe.model'					)
							with open(A_     ,'wb'					) as fi:
								A        =      self.sp_model.serialized_model_proto()
								fi.write(A_					)

							return (vocab_file,)

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[str]       =			logging.get_logger(__name__)




class    SCREAMING_SNAKE_CASE__     (	UpperCAmelCase					):





	'''simple docstring'''



	def __init__(  self	:    Dict			,  lowercase	:    Union[List[ControlNetModel], Tuple[ControlNetModel]]	):


				'''simple docstring'''


				super().__init__()
				_snake_case		=			nn.ModuleList(lowercase	)



	def  A							(  self	:    Optional[int]			,  lowercase	:    torch.FloatTensor			,  lowercase	:    Union[torch.Tensor, float, int]			,  lowercase	:    torch.Tensor			,  lowercase	:    List[torch.tensor]			,  lowercase	:    List[float]			,  lowercase	:    Optional[torch.Tensor] = None			,  lowercase	:    Optional[torch.Tensor] = None			,  lowercase	:    Optional[torch.Tensor] = None			,  lowercase	:    Optional[Dict[str, Any]] = None			,  lowercase	:    bool = False			,  lowercase	:    bool = True			,  ):


				'''simple docstring'''


				for i, (image, scale, controlnet) in enumerate(zip(lowercase			,  lowercase			,  self.nets	)	):
							_snake_case	,					_snake_case		=			controlnet(
							    lowercase			,  lowercase			,  lowercase			,  lowercase			,  lowercase			,  lowercase			,  lowercase			,  lowercase			,  lowercase			,  lowercase			,  lowercase			,  )

							# merge samples
							if i == 0:
										_snake_case	,					_snake_case		=			down_samples, mid_sample
							else:
										_snake_case		=			[
										    samples_prev + samples_curr
										    for samples_prev, samples_curr in zip(lowercase			,  lowercase	)
										]
										mid_block_res_sample += mid_sample

				return down_block_res_samples, mid_block_res_sample



	def  A							(  self	:    Dict			,  lowercase	:    Union[str, os.PathLike]			,  lowercase	:    bool = True			,  lowercase	:    Callable = None			,  lowercase	:    bool = False			,  lowercase	:    Optional[str] = None			,  ):


				'''simple docstring'''


				_snake_case		=			0
				_snake_case		=			save_directory
				for controlnet in self.nets:
							controlnet.save_pretrained(
							    lowercase			,  is_main_process=lowercase			,  save_function=lowercase			,  safe_serialization=lowercase			,  variant=lowercase			,  )

							idx += 1
							_snake_case		=			model_path_to_save + f'''_{idx}'''





	@classmethod
	def  A							(  cls	:    Any			,  lowercase	:    Optional[Union[str, os.PathLike]]			,  **lowercase	:    List[str]	):


				'''simple docstring'''


				_snake_case		=			0
				_snake_case		=			[]

				# 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`, ...
				_snake_case		=			pretrained_model_path
				while os.path.isdir(lowercase	):
							_snake_case		=			ControlNetModel.from_pretrained(lowercase			,  **lowercase	)
							controlnets.append(lowercase	)

							idx += 1
							_snake_case		=			pretrained_model_path + f'''_{idx}'''

				logger.info(f'''{len(lowercase	)} controlnets loaded from {pretrained_model_path}.'''	)

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

				return cls(lowercase	)