






import collections
import importlib.util
import os
import re
from pathlib import Path


A__	:			int			   = 'src/transformers'


# Matches is_xxx_available()
A__	:			Any			   = re.compile(r'is\_([a-z_]*)_available()')
# Catches a one-line _import_struct = {xxx}
A__	:			Any			   = re.compile(r'^_import_structure\s+=\s+\{([^\}]+)\}')
# Catches a line with a key-values pattern: "bla": ["foo", "bar"]
A__	:			Dict			   = re.compile(r'\s+"\S*":\s+\[([^\]]*)\]')
# Catches a line if not is_foo_available
A__	:			int			   = re.compile(r'^\s*if\s+not\s+is\_[a-z_]*\_available\(\)')
# Catches a line _import_struct["bla"].append("foo")
A__	:			Optional[Any]			   = re.compile(r'^\s*_import_structure\["\S*"\]\.append\("(\S*)"\)')
# Catches a line _import_struct["bla"].extend(["foo", "bar"]) or _import_struct["bla"] = ["foo", "bar"]
A__	:			List[Any]			   = re.compile(r'^\s*_import_structure\[\S*\](?:\.extend\(|\s*=\s+)\[([^\]]*)\]')
# Catches a line with an object between quotes and a comma:     "MyModel",
A__	:			int			   = re.compile('^\s+"([^"]+)",')
# Catches a line with objects between brackets only:    ["foo", "bar"],
A__	:			Optional[Any]			   = re.compile('^\s+\[([^\]]+)\]')
# Catches a line with from foo import bar, bla, boo
A__	:			Any			   = re.compile(r'\s+from\s+\S*\s+import\s+([^\(\s].*)\n')
# Catches a line with try:
A__	:			Tuple			   = re.compile(r'^\s*try:')
# Catches a line with else:
A__	:			Any			   = re.compile(r'^\s*else:')






def   a			(						lowerCamelCase_				):






				'''simple docstring'''

				if _re_test_backend.search(__A				) is None:
								return None
				lowercase__    =			[b[0] for b in _re_backend.findall(__A				)]
				backends.sort()
				return "_and_".join(__A				)






def   a			(						lowerCamelCase_				):






				'''simple docstring'''

				with open(__A		,   '''r'''		,   encoding='''utf-8'''		,   newline='''\n'''				) as f:
								lowercase__    =			f.readlines()

				lowercase__    =			0
				while line_index < len(__A				) and not lines[line_index].startswith('''_import_structure = {'''				):
								line_index += 1

				# If this is a traditional init, just return.
				if line_index >= len(__A				):
								return None

				# First grab the objects without a specific backend in _import_structure
				lowercase__    =			[]
				while not lines[line_index].startswith('''if TYPE_CHECKING'''				) and find_backend(lines[line_index]				) is None:
								lowercase__    =			lines[line_index]
								# If we have everything on a single line, let's deal with it.
								if _re_one_line_import_struct.search(__A				):
												lowercase__    =			_re_one_line_import_struct.search(__A				).groups()[0]
												lowercase__    =			re.findall('''\[([^\]]+)\]'''		,   __A				)
												for imp in imports:
																objects.extend([obj[1:-1] for obj in imp.split(''', '''				)]				)
												line_index += 1
												continue
								lowercase__    =			_re_import_struct_key_value.search(__A				)
								if single_line_import_search is not None:
												lowercase__    =			[obj[1:-1] for obj in single_line_import_search.groups()[0].split(''', '''				) if len(__A				) > 0]
												objects.extend(__A				)
								elif line.startswith(''' ''' * 8 + '''"'''				):
												objects.append(line[9:-3]				)
								line_index += 1

				lowercase__    =			{'none': objects}
				# Let's continue with backend-specific objects in _import_structure
				while not lines[line_index].startswith('''if TYPE_CHECKING'''				):
								# If the line is an if not is_backend_available, we grab all objects associated.
								lowercase__    =			find_backend(lines[line_index]				)
								# Check if the backend declaration is inside a try block:
								if _re_try.search(lines[line_index - 1]				) is None:
												lowercase__    =			None

								if backend is not None:
												line_index += 1

												# Scroll until we hit the else block of try-except-else
												while _re_else.search(lines[line_index]				) is None:
																line_index += 1

												line_index += 1

												lowercase__    =			[]
												# Until we unindent, add backend objects to the list
												while len(lines[line_index]				) <= 1 or lines[line_index].startswith(''' ''' * 4				):
																lowercase__    =			lines[line_index]
																if _re_import_struct_add_one.search(__A				) is not None:
																				objects.append(_re_import_struct_add_one.search(__A				).groups()[0]				)
																elif _re_import_struct_add_many.search(__A				) is not None:
																				lowercase__    =			_re_import_struct_add_many.search(__A				).groups()[0].split(''', '''				)
																				lowercase__    =			[obj[1:-1] for obj in imports if len(__A				) > 0]
																				objects.extend(__A				)
																elif _re_between_brackets.search(__A				) is not None:
																				lowercase__    =			_re_between_brackets.search(__A				).groups()[0].split(''', '''				)
																				lowercase__    =			[obj[1:-1] for obj in imports if len(__A				) > 0]
																				objects.extend(__A				)
																elif _re_quote_object.search(__A				) is not None:
																				objects.append(_re_quote_object.search(__A				).groups()[0]				)
																elif line.startswith(''' ''' * 8 + '''"'''				):
																				objects.append(line[9:-3]				)
																elif line.startswith(''' ''' * 12 + '''"'''				):
																				objects.append(line[13:-3]				)
																line_index += 1

												lowercase__    =			objects
								else:
												line_index += 1

    # At this stage we are in the TYPE_CHECKING part, first grab the objects without a specific backend
				lowercase__    =			[]
				while (
				    line_index < len(__A				)
				    and find_backend(lines[line_index]				) is None
				    and not lines[line_index].startswith('''else'''				)
				):
								lowercase__    =			lines[line_index]
								lowercase__    =			_re_import.search(__A				)
								if single_line_import_search is not None:
												objects.extend(single_line_import_search.groups()[0].split(''', '''				)				)
								elif line.startswith(''' ''' * 8				):
												objects.append(line[8:-2]				)
								line_index += 1

				lowercase__    =			{'none': objects}
				# Let's continue with backend-specific objects
				while line_index < len(__A				):
								# If the line is an if is_backend_available, we grab all objects associated.
								lowercase__    =			find_backend(lines[line_index]				)
								# Check if the backend declaration is inside a try block:
								if _re_try.search(lines[line_index - 1]				) is None:
												lowercase__    =			None

								if backend is not None:
												line_index += 1

												# Scroll until we hit the else block of try-except-else
												while _re_else.search(lines[line_index]				) is None:
																line_index += 1

												line_index += 1

												lowercase__    =			[]
												# Until we unindent, add backend objects to the list
												while len(lines[line_index]				) <= 1 or lines[line_index].startswith(''' ''' * 8				):
																lowercase__    =			lines[line_index]
																lowercase__    =			_re_import.search(__A				)
																if single_line_import_search is not None:
																				objects.extend(single_line_import_search.groups()[0].split(''', '''				)				)
																elif line.startswith(''' ''' * 12				):
																				objects.append(line[12:-2]				)
																line_index += 1

												lowercase__    =			objects
								else:
												line_index += 1

				return import_dict_objects, type_hint_objects






def   a			(						lowerCamelCase_		,   lowerCamelCase_				):






				'''simple docstring'''

				def find_duplicates(lowerCamelCase_				):
								return [k for k, v in collections.Counter(__A				).items() if v > 1]

				if list(import_dict_objects.keys()				) != list(type_hint_objects.keys()				):
								return ["Both sides of the init do not have the same backends!"]

				lowercase__    =			[]
				for key in import_dict_objects.keys():
								lowercase__    =			find_duplicates(import_dict_objects[key]				)
								if duplicate_imports:
												errors.append(F"""Duplicate _import_structure definitions for: {duplicate_imports}"""				)
								lowercase__    =			find_duplicates(type_hint_objects[key]				)
								if duplicate_type_hints:
												errors.append(F"""Duplicate TYPE_CHECKING objects for: {duplicate_type_hints}"""				)

								if sorted(set(import_dict_objects[key]				)				) != sorted(set(type_hint_objects[key]				)				):
												lowercase__    =			'base imports' if key == 'none' else F"""{key} backend"""
												errors.append(F"""Differences for {name}:"""				)
												for a in type_hint_objects[key]:
																if a not in import_dict_objects[key]:
																				errors.append(F"""  {a} in TYPE_HINT but not in _import_structure."""				)
												for a in import_dict_objects[key]:
																if a not in type_hint_objects[key]:
																				errors.append(F"""  {a} in _import_structure but not in TYPE_HINT."""				)
				return errors






def   a			(						):






				'''simple docstring'''
				lowercase__    =			[]
				for root, _, files in os.walk(__A				):
								if "__init__.py" in files:
												lowercase__    =			os.path.join(__A		,   '''__init__.py'''				)
												lowercase__    =			parse_init(__A				)
												if objects is not None:
																lowercase__    =			analyze_results(*__A				)
																if len(__A				) > 0:
																				lowercase__    =			F"""Problem in {fname}, both halves do not define the same objects.\n{errors[0]}"""
																				failures.append('''\n'''.join(__A				)				)
				if len(__A				) > 0:
								raise ValueError('''\n\n'''.join(__A				)				)






def   a			(						):






				'''simple docstring'''
				lowercase__    =			[]
				for path, directories, files in os.walk(__A				):
								for folder in directories:
												# Ignore private modules
												if folder.startswith('''_'''				):
																directories.remove(__A				)
																continue
												# Ignore leftovers from branches (empty folders apart from pycache)
												if len(list((Path(__A				) / folder).glob('''*.py'''				)				)				) == 0:
																continue
												lowercase__    =			str((Path(__A				) / folder).relative_to(__A				)				)
												lowercase__    =			short_path.replace(os.path.sep		,   '''.'''				)
												submodules.append(__A				)
								for fname in files:
												if fname == "__init__.py":
																continue
												lowercase__    =			str((Path(__A				) / fname).relative_to(__A				)				)
												lowercase__    =			short_path.replace('''.py'''		,   ''''''				).replace(os.path.sep		,   '''.'''				)
												if len(submodule.split('''.'''				)				) == 1:
																submodules.append(__A				)
				return submodules


A__	:			Dict			   = [
    'convert_pytorch_checkpoint_to_tf2',
    'modeling_flax_pytorch_utils',
]






def   a			(						):






				'''simple docstring'''
				lowercase__    =			importlib.util.spec_from_file_location(
				    '''transformers'''		,   os.path.join(__A		,   '''__init__.py'''				)		,   submodule_search_locations=[PATH_TO_TRANSFORMERS]		,   )
				lowercase__    =			spec.loader.load_module()

				lowercase__    =			[
				    module
				    for module in get_transformers_submodules()
				    if module not in IGNORE_SUBMODULES and module not in transformers._import_structure.keys()
				]
				if len(__A				) > 0:
								lowercase__    =			'\n'.join(F"""- {module}""" for module in module_not_registered				)
								raise ValueError(
								    '''The following submodules are not properly registered in the main init of Transformers:\n'''
								    F"""{list_of_modules}\n"""
								    '''Make sure they appear somewhere in the keys of `_import_structure` with an empty list as value.'''				)


if __name__ == "__main__":
							check_all_inits()
							check_submodules()








import random
import unittest

import numpy as np
import torch

from diffusers import (
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    OnnxStableDiffusionUpscalePipeline,
    PNDMScheduler,
)
from diffusers.utils import floats_tensor
from diffusers.utils.testing_utils import (
    is_onnx_available,
    load_image,
    nightly,
    require_onnxruntime,
    require_torch_gpu,
)

from ..test_pipelines_onnx_common import OnnxPipelineTesterMixin


if is_onnx_available():
   import onnxruntime as ort
class  SCREAMING_SNAKE_CASE__  (				lowercase__						,    unittest.TestCase			):
      # TODO: is there an appropriate internal test set?
      snake_case__				:     Any        =						'''ssube/stable-diffusion-x4-upscaler-onnx'''
      def        SCREAMING_SNAKE_CASE							(       self				:       List[str]     ,							SCREAMING_SNAKE_CASE__				:       int=0     )		->			Tuple:
             a_					:		Union[str, Any]							=  floats_tensor((1, 3, 1_2_8, 1_2_8)     ,							rng=random.Random(SCREAMING_SNAKE_CASE__     )     )
             a_					:		Dict							=  torch.manual_seed(SCREAMING_SNAKE_CASE__     )
             a_					:		Union[str, Any]							=  {
                 'prompt': 'A painting of a squirrel eating a burger',
                 'image': image,
                 'generator': generator,
                 'num_inference_steps': 3,
                 'guidance_scale': 7.5,
                 'output_type': 'numpy',
             }
             return inputs
      def        SCREAMING_SNAKE_CASE							(       self				:       List[str]     )		->			Dict:
             a_					:		Dict							=  OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint     ,							provider='CPUExecutionProvider'     )
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__     )

             a_					:		Optional[int]							=  self.get_dummy_inputs()
             a_					:		int							=  pipe(**SCREAMING_SNAKE_CASE__     ).images
             a_					:		Tuple							=  image[0, -3:, -3:, -1].flatten()

             # started as 128, should now be 512
             assert image.shape == (1, 5_1_2, 5_1_2, 3)
             a_					:		List[Any]							=  np.array(
                 [0.6974782, 0.68902093, 0.70135885, 0.7583618, 0.7804545, 0.7854912, 0.78667426, 0.78743863, 0.78070223]     )
             assert np.abs(image_slice - expected_slice     ).max() < 1E-1
      def        SCREAMING_SNAKE_CASE							(       self				:       Dict     )		->			Dict:
             a_					:		Tuple							=  OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint     ,							provider='CPUExecutionProvider'     )
             a_					:		int							=  PNDMScheduler.from_config(pipe.scheduler.config     ,							skip_prk_steps=SCREAMING_SNAKE_CASE__     )
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__     )

             a_					:		List[str]							=  self.get_dummy_inputs()
             a_					:		List[str]							=  pipe(**SCREAMING_SNAKE_CASE__     ).images
             a_					:		Optional[Any]							=  image[0, -3:, -3:, -1]

             assert image.shape == (1, 5_1_2, 5_1_2, 3)
             a_					:		str							=  np.array(
                 [0.6898892, 0.59240556, 0.52499527, 0.58866215, 0.52258235, 0.52572715, 0.62414473, 0.6174387, 0.6214964]     )
             assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1E-1
      def        SCREAMING_SNAKE_CASE							(       self				:       Any     )		->			Dict:
             a_					:		Optional[Any]							=  OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint     ,							provider='CPUExecutionProvider'     )
             a_					:		Optional[Any]							=  DPMSolverMultistepScheduler.from_config(pipe.scheduler.config     )
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__     )

             a_					:		Optional[int]							=  self.get_dummy_inputs()
             a_					:		Dict							=  pipe(**SCREAMING_SNAKE_CASE__     ).images
             a_					:		str							=  image[0, -3:, -3:, -1]

             assert image.shape == (1, 5_1_2, 5_1_2, 3)
             a_					:		Optional[Any]							=  np.array(
                 [0.7659278, 0.76437664, 0.75579107, 0.7691116, 0.77666986, 0.7727672, 0.7758664, 0.7812226, 0.76942515]     )

             assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1E-1
      def        SCREAMING_SNAKE_CASE							(       self				:       List[Any]     )		->			int:
             a_					:		List[str]							=  OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint     ,							provider='CPUExecutionProvider'     )
             a_					:		int							=  EulerDiscreteScheduler.from_config(pipe.scheduler.config     )
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__     )

             a_					:		Optional[Any]							=  self.get_dummy_inputs()
             a_					:		Dict							=  pipe(**SCREAMING_SNAKE_CASE__     ).images
             a_					:		List[str]							=  image[0, -3:, -3:, -1]

             assert image.shape == (1, 5_1_2, 5_1_2, 3)
             a_					:		int							=  np.array(
                 [0.6974782, 0.68902093, 0.70135885, 0.7583618, 0.7804545, 0.7854912, 0.78667426, 0.78743863, 0.78070223]     )
             assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1E-1

      def        SCREAMING_SNAKE_CASE							(       self				:       str     )		->			Optional[Any]:
             a_					:		Tuple							=  OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint     ,							provider='CPUExecutionProvider'     )
             a_					:		Optional[int]							=  EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config     )
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__     )

             a_					:		Union[str, Any]							=  self.get_dummy_inputs()
             a_					:		List[str]							=  pipe(**SCREAMING_SNAKE_CASE__     ).images
             a_					:		List[str]							=  image[0, -3:, -3:, -1]

             assert image.shape == (1, 5_1_2, 5_1_2, 3)
             a_					:		Union[str, Any]							=  np.array(
                 [0.77424496, 0.773601, 0.7645288, 0.7769598, 0.7772739, 0.7738688, 0.78187233, 0.77879584, 0.767043]     )

             assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1E-1
@nightly
@require_onnxruntime
@require_torch_gpu
class  SCREAMING_SNAKE_CASE__  (				unittest.TestCase			):
      @property
      def        SCREAMING_SNAKE_CASE							(       self				:       Tuple     )		->			Union[str, Any]:
             return (
                 "CUDAExecutionProvider",
                 {
                     "gpu_mem_limit": "15000000000",  # 15GB
                     "arena_extend_strategy": "kSameAsRequested",
                 },
             )
      @property
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[Any]     )		->			List[str]:
             a_					:		List[str]							=  ort.SessionOptions()
             a_					:		int							=  False
             return options
      def        SCREAMING_SNAKE_CASE							(       self				:       List[str]     )		->			Tuple:
             a_					:		str							=  load_image(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
                 '/img2img/sketch-mountains-input.jpg'     )
             a_					:		int							=  init_image.resize((1_2_8, 1_2_8)     )
             # using the PNDM scheduler by default
             a_					:		Dict							=  OnnxStableDiffusionUpscalePipeline.from_pretrained(
                 'ssube/stable-diffusion-x4-upscaler-onnx'     ,							provider=self.gpu_provider     ,							sess_options=self.gpu_options     ,							)
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__     )

             a_					:		Tuple							=  'A fantasy landscape, trending on artstation'

             a_					:		str							=  torch.manual_seed(0     )
             a_					:		List[str]							=  pipe(
                 prompt=SCREAMING_SNAKE_CASE__     ,							image=SCREAMING_SNAKE_CASE__     ,							guidance_scale=7.5     ,							num_inference_steps=1_0     ,							generator=SCREAMING_SNAKE_CASE__     ,							output_type='np'     ,							)
             a_					:		Dict							=  output.images
             a_					:		Any							=  images[0, 2_5_5:2_5_8, 3_8_3:3_8_6, -1]

             assert images.shape == (1, 5_1_2, 5_1_2, 3)
             a_					:		str							=  np.array([0.4883, 0.4947, 0.4980, 0.4975, 0.4982, 0.4980, 0.5000, 0.5006, 0.4972]     )
             # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

             assert np.abs(image_slice.flatten() - expected_slice     ).max() < 2E-2

      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[int]     )		->			List[str]:
             a_					:		Dict							=  load_image(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
                 '/img2img/sketch-mountains-input.jpg'     )
             a_					:		List[str]							=  init_image.resize((1_2_8, 1_2_8)     )
             a_					:		Dict							=  LMSDiscreteScheduler.from_pretrained(
                 'ssube/stable-diffusion-x4-upscaler-onnx'     ,							subfolder='scheduler'     )
             a_					:		Union[str, Any]							=  OnnxStableDiffusionUpscalePipeline.from_pretrained(
                 'ssube/stable-diffusion-x4-upscaler-onnx'     ,							scheduler=SCREAMING_SNAKE_CASE__     ,							provider=self.gpu_provider     ,							sess_options=self.gpu_options     ,							)
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__     )

             a_					:		Any							=  'A fantasy landscape, trending on artstation'

             a_					:		Tuple							=  torch.manual_seed(0     )
             a_					:		Optional[Any]							=  pipe(
                 prompt=SCREAMING_SNAKE_CASE__     ,							image=SCREAMING_SNAKE_CASE__     ,							guidance_scale=7.5     ,							num_inference_steps=2_0     ,							generator=SCREAMING_SNAKE_CASE__     ,							output_type='np'     ,							)
             a_					:		str							=  output.images
             a_					:		List[Any]							=  images[0, 2_5_5:2_5_8, 3_8_3:3_8_6, -1]

             assert images.shape == (1, 5_1_2, 5_1_2, 3)
             a_					:		Tuple							=  np.array(
                 [0.50173753, 0.50223356, 0.502039, 0.50233036, 0.5023725, 0.5022601, 0.5018758, 0.50234085, 0.50241566]     )
             # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

             assert np.abs(image_slice.flatten() - expected_slice     ).max() < 2E-2


