


"""simple docstring"""
import argparse
import glob
import importlib.util
import os
import re

import black
from doc_builder.style_doc import style_docstrings_in_code


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_copies.py
__SCREAMING_SNAKE_CASE							:      List[str]						=      'src/diffusers'
__SCREAMING_SNAKE_CASE							:      List[Any]						=      '.'


# This is to make sure the diffusers module imported is the one in the repo.
__SCREAMING_SNAKE_CASE							:      List[Any]						=      importlib.util.spec_from_file_location(
    'diffusers',
    os.path.join(DIFFUSERS_PATH, '__init__.py'),
    submodule_search_locations=[DIFFUSERS_PATH],
)
__SCREAMING_SNAKE_CASE							:      Any						=      spec.loader.load_module()







def   _a			( _SCREAMING_SNAKE_CASE			,	_SCREAMING_SNAKE_CASE			)      ->			Dict:
							return line.startswith(lowerCAmelCase__			) or len(lowerCAmelCase__			) <= 1 or re.search(r"""^\s*\)(\s*->.*:|:)\s*$"""			,	lowerCAmelCase__			) is not None







def   _a			( _SCREAMING_SNAKE_CASE			)      ->			str:
							snake_case_          =						object_name.split("""."""			)
							snake_case_          =						0

							# First let's find the module where our object lives.
							snake_case_          =						parts[i]
							while i < len(lowerCAmelCase__			) and not os.path.isfile(os.path.join(lowerCAmelCase__			,	f"""{module}.py"""			)			):
														i += 1
														if i < len(lowerCAmelCase__			):
																					snake_case_          =						os.path.join(lowerCAmelCase__			,	parts[i]			)
							if i >= len(lowerCAmelCase__			):
														raise ValueError(f"""`object_name` should begin with the name of a module of diffusers but got {object_name}."""			)

							with open(os.path.join(lowerCAmelCase__			,	f"""{module}.py"""			)			,	"""r"""			,	encoding="""utf-8"""			,	newline="""\n"""			) as f:
														snake_case_          =						f.readlines()

							# Now let's find the class / func in the code!
							snake_case_          =						''''''
							snake_case_          =						0
							for name in parts[i + 1 :]:
														while (
														    line_index < len(lowerCAmelCase__			) and re.search(rf"""^{indent}(class|def)\s+{name}(\(|\:)"""			,	lines[line_index]			) is None
														):
																					line_index += 1
														indent += "    "
														line_index += 1

							if line_index >= len(lowerCAmelCase__			):
														raise ValueError(f""" {object_name} does not match any function or class in {module}."""			)

							# We found the beginning of the class / func, now let's find the end (when the indent diminishes).
							snake_case_          =						line_index
							while line_index < len(lowerCAmelCase__			) and _should_continue(lines[line_index]			,	lowerCAmelCase__			):
														line_index += 1
							# Clean up empty lines at the end (if any).
							while len(lines[line_index - 1]			) <= 1:
														line_index -= 1

							snake_case_          =						lines[start_index:line_index]
							return "".join(lowerCAmelCase__			)


__SCREAMING_SNAKE_CASE							:      Optional[int]						=      re.compile(R'^(\s*)#\s*Copied from\s+diffusers\.(\S+\.\S+)\s*($|\S.*$)')
__SCREAMING_SNAKE_CASE							:      List[str]						=      re.compile(R'^\s*(\S+)->(\S+)(\s+.*|$)')
__SCREAMING_SNAKE_CASE							:      List[Any]						=      re.compile(R'<FILL\s+[^>]*>')







def   _a			( _SCREAMING_SNAKE_CASE			)      ->			List[Any]:
							snake_case_          =						code.split("""\n"""			)
							snake_case_          =						0
							while idx < len(lowerCAmelCase__			) and len(lines[idx]			) == 0:
														idx += 1
							if idx < len(lowerCAmelCase__			):
														return re.search(r"""^(\s*)\S"""			,	lines[idx]			).groups()[0]
							return ""







def   _a			( _SCREAMING_SNAKE_CASE			)      ->			List[str]:
							snake_case_          =						len(get_indent(lowerCAmelCase__			)			) > 0
							if has_indent:
														snake_case_          =						f"""class Bla:\n{code}"""
							snake_case_          =						black.Mode(target_versions={black.TargetVersion.PYaa}			,	line_length=119			,	preview=lowerCAmelCase__			)
							snake_case_          =						black.format_str(lowerCAmelCase__			,	mode=lowerCAmelCase__			)
							snake_case_          =						style_docstrings_in_code(lowerCAmelCase__			)
							return result[len("""class Bla:\n"""			) :] if has_indent else result







def   _a			( _SCREAMING_SNAKE_CASE			,	_SCREAMING_SNAKE_CASE=False			)      ->			Optional[int]:
							with open(lowerCAmelCase__			,	"""r"""			,	encoding="""utf-8"""			,	newline="""\n"""			) as f:
														snake_case_          =						f.readlines()
							snake_case_          =						[]
							snake_case_          =						0
							# Not a for loop cause `lines` is going to change (if `overwrite=True`).
							while line_index < len(lowerCAmelCase__			):
														snake_case_          =						_re_copy_warning.search(lines[line_index]			)
														if search is None:
																					line_index += 1
																					continue

														# There is some copied code here, let's retrieve the original.
														snake_case_          =						search.groups()
														snake_case_          =						find_code_in_diffusers(lowerCAmelCase__			)
														snake_case_          =						get_indent(lowerCAmelCase__			)

														snake_case_          =						line_index + 1 if indent == theoretical_indent else line_index + 2
														snake_case_          =						theoretical_indent
														snake_case_          =						start_index

														# Loop to check the observed code, stop when indentation diminishes or if we see a End copy comment.
														snake_case_          =						True
														while line_index < len(lowerCAmelCase__			) and should_continue:
																					line_index += 1
																					if line_index >= len(lowerCAmelCase__			):
																												break
																					snake_case_          =						lines[line_index]
																					snake_case_          =						_should_continue(lowerCAmelCase__			,	lowerCAmelCase__			) and re.search(f"""^{indent}# End copy"""			,	lowerCAmelCase__			) is None
														# Clean up empty lines at the end (if any).
														while len(lines[line_index - 1]			) <= 1:
																					line_index -= 1

														snake_case_          =						lines[start_index:line_index]
														snake_case_          =						''''''.join(lowerCAmelCase__			)

														# Remove any nested `Copied from` comments to avoid circular copies
														snake_case_          =						[line for line in theoretical_code.split("""\n"""			) if _re_copy_warning.search(lowerCAmelCase__			) is None]
														snake_case_          =						'''\n'''.join(lowerCAmelCase__			)

														# Before comparing, use the `replace_pattern` on the original code.
														if len(lowerCAmelCase__			) > 0:
																					snake_case_          =						replace_pattern.replace("""with"""			,	""""""			).split(""","""			)
																					snake_case_          =						[_re_replace_pattern.search(lowerCAmelCase__			) for p in patterns]
																					for pattern in patterns:
																												if pattern is None:
																																			continue
																												snake_case_          =						pattern.groups()
																												snake_case_          =						re.sub(lowerCAmelCase__			,	lowerCAmelCase__			,	lowerCAmelCase__			)
																												if option.strip() == "all-casing":
																																			snake_case_          =						re.sub(obja.lower()			,	obja.lower()			,	lowerCAmelCase__			)
																																			snake_case_          =						re.sub(obja.upper()			,	obja.upper()			,	lowerCAmelCase__			)

																																			       # Blackify after replacement. To be able to do that, we need the header (class or function definition)
																																			       # from the previous line
																					snake_case_          =						blackify(lines[start_index - 1] + theoretical_code			)
																					snake_case_          =						theoretical_code[len(lines[start_index - 1]			) :]

														# Test for a diff and act accordingly.
														if observed_code != theoretical_code:
																					diffs.append([object_name, start_index]			)
																					if overwrite:
																												snake_case_          =						lines[:start_index] + [theoretical_code] + lines[line_index:]
																												snake_case_          =						start_index + 1

							if overwrite and len(lowerCAmelCase__			) > 0:
														# Warn the user a file has been modified.
														print(f"""Detected changes, rewriting {filename}."""			)
														with open(lowerCAmelCase__			,	"""w"""			,	encoding="""utf-8"""			,	newline="""\n"""			) as f:
																					f.writelines(lowerCAmelCase__			)
							return diffs







def   _a			( _SCREAMING_SNAKE_CASE = False			)      ->			Dict:
							snake_case_          =						glob.glob(os.path.join(lowerCAmelCase__			,	"""**/*.py"""			)			,	recursive=lowerCAmelCase__			)
							snake_case_          =						[]
							for filename in all_files:
														snake_case_          =						is_copy_consistent(lowerCAmelCase__			,	lowerCAmelCase__			)
														diffs += [f"""- {filename}: copy does not match {d[0]} at line {d[1]}""" for d in new_diffs]
							if not overwrite and len(lowerCAmelCase__			) > 0:
														snake_case_          =						'''\n'''.join(lowerCAmelCase__			)
														raise Exception(
														    """Found the following copy inconsistencies:\n"""
														    + diff
														    + """\nRun `make fix-copies` or `python utils/check_copies.py --fix_and_overwrite` to fix them."""			)


if __name__ == "__main__":
				__SCREAMING_SNAKE_CASE							:      str						=      argparse.ArgumentParser()
				parser.add_argument('--fix_and_overwrite', action='store_true', help='Whether to fix inconsistencies.')
				__SCREAMING_SNAKE_CASE							:      Tuple						=      parser.parse_args()

				check_copies(args.fix_and_overwrite)







"""simple docstring"""
import re
import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin





class        __A       (snake_case__):





							'''simple docstring'''

							__lowercase:						Any							 =      ["""image_processor""", """tokenizer"""]
							__lowercase:						Dict							 =      """AutoImageProcessor"""
							__lowercase:						List[Any]							 =      """AutoTokenizer"""



							def __init__(			self							: Tuple						,    UpperCAmelCase_							: int=None						,    UpperCAmelCase_							: Tuple=None						,    **UpperCAmelCase_							: str    )   ->Optional[int]:



														"""simple docstring"""



														snake_case_          =						None
														if "feature_extractor" in kwargs:
																					warnings.warn(
																					    """The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"""
																					    """ instead."""						,    UpperCAmelCase_						,    )
																					snake_case_          =						kwargs.pop("""feature_extractor"""    )

														snake_case_          =						image_processor if image_processor is not None else feature_extractor
														if image_processor is None:
																					raise ValueError("""You need to specify an `image_processor`."""    )
														if tokenizer is None:
																					raise ValueError("""You need to specify a `tokenizer`."""    )

														super().__init__(UpperCAmelCase_						,    UpperCAmelCase_    )
														snake_case_          =						self.image_processor
														snake_case_          =						False



							def __call__(			self							: Any						,    *UpperCAmelCase_							: Optional[int]						,    **UpperCAmelCase_							: str    )   ->List[str]:



														"""simple docstring"""



														if self._in_target_context_manager:
																					return self.current_processor(*UpperCAmelCase_						,    **UpperCAmelCase_    )

														snake_case_          =						kwargs.pop("""images"""						,    UpperCAmelCase_    )
														snake_case_          =						kwargs.pop("""text"""						,    UpperCAmelCase_    )
														if len(UpperCAmelCase_    ) > 0:
																					snake_case_          =						args[0]
																					snake_case_          =						args[1:]

														if images is None and text is None:
																					raise ValueError("""You need to specify either an `images` or `text` input to process."""    )

														if images is not None:
																					snake_case_          =						self.image_processor(UpperCAmelCase_						,    *UpperCAmelCase_						,    **UpperCAmelCase_    )
														if text is not None:
																					snake_case_          =						self.tokenizer(UpperCAmelCase_						,    **UpperCAmelCase_    )

														if text is None:
																					return inputs
														elif images is None:
																					return encodings
														else:
																					snake_case_          =						encodings["""input_ids"""]
																					return inputs



							def 						lowerCAmelCase    (			self							: Optional[Any]						,    *UpperCAmelCase_							: Optional[Any]						,    **UpperCAmelCase_							: Tuple    )   ->Optional[int]:



														"""simple docstring"""



														return self.tokenizer.batch_decode(*UpperCAmelCase_						,    **UpperCAmelCase_    )



							def 						lowerCAmelCase    (			self							: Optional[int]						,    *UpperCAmelCase_							: Union[str, Any]						,    **UpperCAmelCase_							: List[Any]    )   ->str:



														"""simple docstring"""



														return self.tokenizer.decode(*UpperCAmelCase_						,    **UpperCAmelCase_    )



							@contextmanager
							def 						lowerCAmelCase    (			self							: List[str]    )   ->Dict:



														"""simple docstring"""



														warnings.warn(
														    """`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your """
														    """labels by using the argument `text` of the regular `__call__` method (either in the same call as """
														    """your images inputs, or in a separate call."""    )
														snake_case_          =						True
														snake_case_          =						self.tokenizer
														yield
														snake_case_          =						self.image_processor
														snake_case_          =						False



							def 						lowerCAmelCase    (			self							: int						,    UpperCAmelCase_							: Union[str, Any]						,    UpperCAmelCase_							: str=False						,    UpperCAmelCase_							: str=None    )   ->Tuple:



														"""simple docstring"""



														if added_vocab is None:
																					snake_case_          =						self.tokenizer.get_added_vocab()

														snake_case_          =						{}

														while tokens:
																					snake_case_          =						re.search(R"""<s_(.*?)>"""						,    UpperCAmelCase_						,    re.IGNORECASE    )
																					if start_token is None:
																												break
																					snake_case_          =						start_token.group(1    )
																					snake_case_          =						re.search(RF"""</s_{key}>"""						,    UpperCAmelCase_						,    re.IGNORECASE    )
																					snake_case_          =						start_token.group()
																					if end_token is None:
																												snake_case_          =						tokens.replace(UpperCAmelCase_						,    """"""    )
																					else:
																												snake_case_          =						end_token.group()
																												snake_case_          =						re.escape(UpperCAmelCase_    )
																												snake_case_          =						re.escape(UpperCAmelCase_    )
																												snake_case_          =						re.search(F"""{start_token_escaped}(.*?){end_token_escaped}"""						,    UpperCAmelCase_						,    re.IGNORECASE    )
																												if content is not None:
																																			snake_case_          =						content.group(1    ).strip()
																																			if r"<s_" in content and r"</s_" in content:  # non-leaf node
																																										snake_case_          =						self.tokenajson(UpperCAmelCase_						,    is_inner_value=UpperCAmelCase_						,    added_vocab=UpperCAmelCase_    )
																																										if value:
																																																	if len(UpperCAmelCase_    ) == 1:
																																																								snake_case_          =						value[0]
																																																	snake_case_          =						value
																																			else:  # leaf nodes
																																										snake_case_          =						[]
																																										for leaf in content.split(R"""<sep/>"""    ):
																																																	snake_case_          =						leaf.strip()
																																																	if leaf in added_vocab and leaf[0] == "<" and leaf[-2:] == "/>":
																																																								snake_case_          =						leaf[1:-2]  # for categorical special tokens
																																																	output[key].append(UpperCAmelCase_    )
																																										if len(output[key]    ) == 1:
																																																	snake_case_          =						output[key][0]

																												snake_case_          =						tokens[tokens.find(UpperCAmelCase_    ) + len(UpperCAmelCase_    ) :].strip()
																												if tokens[:6] == r"<sep/>":  # non-leaf nodes
																																			return [output] + self.tokenajson(tokens[6:]						,    is_inner_value=UpperCAmelCase_						,    added_vocab=UpperCAmelCase_    )

														if len(UpperCAmelCase_    ):
																					return [output] if is_inner_value else output
														else:
																					return [] if is_inner_value else {"text_sequence": tokens}



							@property
							def 						lowerCAmelCase    (			self							: List[Any]    )   ->Optional[int]:



														"""simple docstring"""



														warnings.warn(
														    """`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."""						,    UpperCAmelCase_						,    )
														return self.image_processor_class



							@property
							def 						lowerCAmelCase    (			self							: List[str]    )   ->str:



														"""simple docstring"""



														warnings.warn(
														    """`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."""						,    UpperCAmelCase_						,    )
														return self.image_processor



