




'''simple docstring'''







import random
import unittest

import torch

from diffusers import IFSuperResolutionPipeline
from diffusers.utils import floats_tensor
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import skip_mps, torch_device

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin
from . import IFPipelineTesterMixin

@skip_mps
class   _snake_case( UpperCAmelCase					,   UpperCAmelCase					,   unittest.TestCase			):
 __snake_case:       List[str]									=				IFSuperResolutionPipeline
 __snake_case:       Dict									=				TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'''width''', '''height'''}
 __snake_case:       Tuple									=				TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
 __snake_case:       Union[str, Any]									=				PipelineTesterMixin.required_optional_params - {'''latents'''}






 def 	_UpperCamelCase   (self					:     int						)					->      Any:






    """simple docstring"""


    return self._get_superresolution_dummy_components()






 def 	_UpperCamelCase   (self					:     str						,		a					:     Tuple						,		a					:     List[str]=0						)					->      Optional[int]:






    """simple docstring"""


    if str(a						).startswith('mps'						):
       A__         =    torch.manual_seed(a						)
    else:
       A__         =    torch.Generator(device=a						).manual_seed(a						)

    A__         =    floats_tensor((1, 3, 32, 32)						,		rng=random.Random(a						)						).to(a						)

    A__         =    {
        'prompt': 'A painting of a squirrel eating a burger',
        'image': image,
        'generator': generator,
        'num_inference_steps': 2,
        'output_type': 'numpy',
    }

    return inputs






 @unittest.skipIf(
     torch_device != 'cuda' or not is_xformers_available()						,		reason='XFormers attention is only available with CUDA and `xformers` installed'						,		)
 def 	_UpperCamelCase   (self					:     Any						)					->      Union[str, Any]:






    """simple docstring"""


    self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1e-3						)






 def 	_UpperCamelCase   (self					:     List[str]						)					->      List[Any]:






    """simple docstring"""


    self._test_save_load_optional_components()






 @unittest.skipIf(torch_device != 'cuda'						,		reason='float16 requires CUDA'						)
 def 	_UpperCamelCase   (self					:     Union[str, Any]						)					->      Union[str, Any]:






    """simple docstring"""


    super().test_save_load_floataa(expected_max_diff=1e-1						)






 def 	_UpperCamelCase   (self					:     str						)					->      Dict:






    """simple docstring"""


    self._test_attention_slicing_forward_pass(expected_max_diff=1e-2						)






 def 	_UpperCamelCase   (self					:     List[str]						)					->      str:






    """simple docstring"""


    self._test_save_load_local()






 def 	_UpperCamelCase   (self					:     List[Any]						)					->      List[str]:






    """simple docstring"""


    self._test_inference_batch_single_identical(
        expected_max_diff=1e-2						,		)








'''simple docstring'''







import os

from apache_beam.io.filesystems import FileSystems
from apache_beam.pipeline import Pipeline

from .logging import get_logger


lowerCAmelCase_    	=							2 << 2_0  # 2mb
lowerCAmelCase_    	=							get_logger(__name__)

class   _snake_case( UpperCAmelCase			):






 def 	_UpperCamelCase   (self					:     Optional[int]						)					->      Optional[Any]:






    """simple docstring"""


    A__         =    self._options.get_all_options().get('runner'						)
    return runner in [None, "DirectRunner", "PortableRunner"]







def 	_A			(   UpperCAmelCase			,UpperCAmelCase			,UpperCAmelCase=False	):






   '''simple docstring'''







   A__         =    FileSystems
   if fs.exists(UpperCAmelCase	):
      if force_upload:
         logger.info(F"""Remote path already exist: {remote_file_path}. Overwriting it as force_upload=True."""	)
      else:
         logger.info(F"""Remote path already exist: {remote_file_path}. Skipping it as force_upload=False."""	)
         return
   with fs.create(UpperCAmelCase	) as remote_file:
      with open(UpperCAmelCase			,'rb'	) as local_file:
         A__         =    local_file.read(UpperCAmelCase	)
         while chunk:
            remote_file.write(UpperCAmelCase	)
            A__         =    local_file.read(UpperCAmelCase	)







def 	_A			(   UpperCAmelCase			,UpperCAmelCase			,UpperCAmelCase=False	):






   '''simple docstring'''







   A__         =    FileSystems
   if os.path.exists(UpperCAmelCase	):
      if force_download:
         logger.info(F"""Local path already exist: {remote_file_path}. Overwriting it as force_upload=True."""	)
      else:
         logger.info(F"""Local path already exist: {remote_file_path}. Skipping it as force_upload=False."""	)
         return
   with fs.open(UpperCAmelCase	) as remote_file:
      with open(UpperCAmelCase			,'wb'	) as local_file:
         A__         =    remote_file.read(UpperCAmelCase	)
         while chunk:
            local_file.write(UpperCAmelCase	)
            A__         =    remote_file.read(UpperCAmelCase	)


