


'''simple docstring'''



import dataclasses
import re
import string
from typing import Any, Dict, Iterator, List, Mapping, Optional, Sequence, Tuple

import numpy as np

from . import residue_constants


__UpperCAmelCase      :Tuple			    =       Mapping[str, np.ndarray]
__UpperCAmelCase      :List[Any]			    =       Mapping[str, Any]  # Is a nested dict.
__UpperCAmelCase      :Optional[Any]			    =       0.01



@dataclasses.dataclass(frozen=_a	)
class      a    :


     """simple docstring"""






     SCREAMING_SNAKE_CASE	:			np.ndarray  # [num_res, num_atom_type, 3]

     # Amino-acid type for each residue represented as an integer between 0 and
     # 20, where 20 is 'X'.
     SCREAMING_SNAKE_CASE	:			np.ndarray  # [num_res]

     # Binary float mask to indicate presence of a particular atom. 1.0 if an atom
     # is present and 0.0 if not. This should be used for loss masking.
     SCREAMING_SNAKE_CASE	:			np.ndarray  # [num_res, num_atom_type]

     # Residue index as used in PDB. It is not necessarily continuous or 0-indexed.
     SCREAMING_SNAKE_CASE	:			np.ndarray  # [num_res]

     # B-factors, or temperature factors, of each residue (in sq. angstroms units),
     # representing the displacement of the residue from its ground truth mean
     # value.
     SCREAMING_SNAKE_CASE	:			np.ndarray  # [num_res, num_atom_type]

     # Chain indices for multi-chain predictions
     SCREAMING_SNAKE_CASE	:			Optional[np.ndarray]							       =       None

     # Optional remark about the protein. Included as a comment in output PDB
     # files
     SCREAMING_SNAKE_CASE	:			Optional[str]							       =       None

     # Templates used to generate this protein (prediction-only)
     SCREAMING_SNAKE_CASE	:			Optional[Sequence[str]]							       =       None

     # Chain corresponding to each parent
     SCREAMING_SNAKE_CASE	:			Optional[Sequence[int]]							       =       None




def      _a     (						_lowercase :     str  ):





     '''simple docstring'''






     __UpperCAmelCase     :			Optional[int]  							=      R'''(\[[A-Z]+\]\n)'''
     __UpperCAmelCase     :			List[str]  							=      [tag.strip() for tag in re.split(_lowercase      , _lowercase  ) if len(_lowercase  ) > 0]
     __UpperCAmelCase     :			Iterator[Tuple[str, List[str]]]  							=      zip(tags[0::2]      , [l.split('''\n'''  ) for l in tags[1::2]]  )

     __UpperCAmelCase     :			List[str]  							=      ["N", "CA", "C"]
     __UpperCAmelCase     :			Optional[int]  							=      None
     __UpperCAmelCase     :			Tuple  							=      None
     __UpperCAmelCase     :			Dict  							=      None
     for g in groups:
          if "[PRIMARY]" == g[0]:
               __UpperCAmelCase     :			Optional[Any]  							=      g[1][0].strip()
               for i in range(len(_lowercase  )  ):
                    if seq[i] not in residue_constants.restypes:
                         __UpperCAmelCase     :			str  							=      '''X'''  # FIXME: strings are immutable
               __UpperCAmelCase     :			Optional[Any]  							=      np.array(
                   [residue_constants.restype_order.get(_lowercase      , residue_constants.restype_num  ) for res_symbol in seq]  )
          elif "[TERTIARY]" == g[0]:
               __UpperCAmelCase     :			List[List[float]]  							=      []
               for axis in range(3  ):
                    tertiary.append(list(map(_lowercase      , g[1][axis].split()  )  )  )
               __UpperCAmelCase     :			List[str]  							=      np.array(_lowercase  )
               __UpperCAmelCase     :			Union[str, Any]  							=      np.zeros((len(tertiary[0]  ) // 3, residue_constants.atom_type_num, 3)  ).astype(np.floataa  )
               for i, atom in enumerate(_lowercase  ):
                    __UpperCAmelCase     :			Optional[int]  							=      np.transpose(tertiary_np[:, i::3]  )
               atom_positions *= PICO_TO_ANGSTROM
          elif "[MASK]" == g[0]:
               __UpperCAmelCase     :			Optional[Any]  							=      np.array(list(map({'''-''': 0, '''+''': 1}.get      , g[1][0].strip()  )  )  )
               __UpperCAmelCase     :			Optional[int]  							=      np.zeros(
                   (
                       len(_lowercase  ),
                       residue_constants.atom_type_num,
                   )  ).astype(np.floataa  )
               for i, atom in enumerate(_lowercase  ):
                    __UpperCAmelCase     :			Dict  							=      1
               atom_mask *= mask[..., None]

     assert aatype is not None

     return Protein(
         atom_positions=_lowercase      , atom_mask=_lowercase      , aatype=_lowercase      , residue_index=np.arange(len(_lowercase  )  )      , b_factors=_lowercase      , )




def      _a     (						_lowercase :     Protein      , _lowercase :     int = 0  ):





     '''simple docstring'''






     __UpperCAmelCase     :			List[str]  							=      []

     __UpperCAmelCase     :			Optional[int]  							=      prot.remark
     if remark is not None:
          pdb_headers.append(F'REMARK {remark}'  )

     __UpperCAmelCase     :			int  							=      prot.parents
     __UpperCAmelCase     :			str  							=      prot.parents_chain_index
     if parents is not None and parents_chain_index is not None:
          __UpperCAmelCase     :			List[Any]  							=      [p for i, p in zip(_lowercase      , _lowercase  ) if i == chain_id]

     if parents is None or len(_lowercase  ) == 0:
          __UpperCAmelCase     :			str  							=      ['''N/A''']

     pdb_headers.append(F'PARENT {" ".join(_lowercase  )}'  )

     return pdb_headers




def      _a     (						_lowercase :     Protein      , _lowercase :     str  ):





     '''simple docstring'''






     __UpperCAmelCase     :			List[str]  							=      []
     __UpperCAmelCase     :			Optional[Any]  							=      pdb_str.split('''\n'''  )

     __UpperCAmelCase     :			Tuple  							=      prot.remark
     if remark is not None:
          out_pdb_lines.append(F'REMARK {remark}'  )

     __UpperCAmelCase     :			List[List[str]]
     if prot.parents is not None and len(prot.parents  ) > 0:
          __UpperCAmelCase     :			List[Any]  							=      []
          if prot.parents_chain_index is not None:
               __UpperCAmelCase     :			Dict[str, List[str]]  							=      {}
               for p, i in zip(prot.parents      , prot.parents_chain_index  ):
                    parent_dict.setdefault(str(_lowercase  )      , []  )
                    parent_dict[str(_lowercase  )].append(_lowercase  )

               __UpperCAmelCase     :			List[str]  							=      max([int(_lowercase  ) for chain_idx in parent_dict]  )
               for i in range(max_idx + 1  ):
                    __UpperCAmelCase     :			Tuple  							=      parent_dict.get(str(_lowercase  )      , ['''N/A''']  )
                    parents_per_chain.append(_lowercase  )
          else:
               parents_per_chain.append(list(prot.parents  )  )
     else:
          __UpperCAmelCase     :			str  							=      [['''N/A''']]

     def make_parent_line(_lowercase :     Sequence[str]  ) -> str:
          return F'PARENT {" ".join(_lowercase  )}'

     out_pdb_lines.append(make_parent_line(parents_per_chain[0]  )  )

     __UpperCAmelCase     :			int  							=      0
     for i, l in enumerate(_lowercase  ):
          if "PARENT" not in l and "REMARK" not in l:
               out_pdb_lines.append(_lowercase  )
          if "TER" in l and "END" not in lines[i + 1]:
               chain_counter += 1
               if not chain_counter >= len(_lowercase  ):
                    __UpperCAmelCase     :			List[str]  							=      parents_per_chain[chain_counter]
               else:
                    __UpperCAmelCase     :			Dict  							=      ['''N/A''']

               out_pdb_lines.append(make_parent_line(_lowercase  )  )

     return "\n".join(_lowercase  )




def      _a     (						_lowercase :     Protein  ):





     '''simple docstring'''






     __UpperCAmelCase     :			Dict  							=      residue_constants.restypes + ['''X''']

     def res_atoa(_lowercase :     int  ) -> str:
          return residue_constants.restype_atoa.get(restypes[r]      , '''UNK'''  )

     __UpperCAmelCase     :			Any  							=      residue_constants.atom_types

     __UpperCAmelCase     :			List[str]  							=      []

     __UpperCAmelCase     :			int  							=      prot.atom_mask
     __UpperCAmelCase     :			Union[str, Any]  							=      prot.aatype
     __UpperCAmelCase     :			Optional[int]  							=      prot.atom_positions
     __UpperCAmelCase     :			List[str]  							=      prot.residue_index.astype(np.intaa  )
     __UpperCAmelCase     :			Any  							=      prot.b_factors
     __UpperCAmelCase     :			Dict  							=      prot.chain_index

     if np.any(aatype > residue_constants.restype_num  ):
          raise ValueError('''Invalid aatypes.'''  )

     __UpperCAmelCase     :			List[Any]  							=      get_pdb_headers(_lowercase  )
     if len(_lowercase  ) > 0:
          pdb_lines.extend(_lowercase  )

     __UpperCAmelCase     :			List[str]  							=      aatype.shape[0]
     __UpperCAmelCase     :			int  							=      1
     __UpperCAmelCase     :			Any  							=      0
     __UpperCAmelCase     :			Union[str, Any]  							=      string.ascii_uppercase
     __UpperCAmelCase     :			str  							=      None
     # Add all atom sites.
     for i in range(_lowercase  ):
          __UpperCAmelCase     :			List[Any]  							=      res_atoa(aatype[i]  )
          for atom_name, pos, mask, b_factor in zip(_lowercase      , atom_positions[i]      , atom_mask[i]      , b_factors[i]  ):
               if mask < 0.5:
                    continue

               __UpperCAmelCase     :			Union[str, Any]  							=      '''ATOM'''
               __UpperCAmelCase     :			Dict  							=      atom_name if len(_lowercase  ) == 4 else F' {atom_name}'
               __UpperCAmelCase     :			Any  							=      ''''''
               __UpperCAmelCase     :			Any  							=      ''''''
               __UpperCAmelCase     :			int  							=      1.00
               __UpperCAmelCase     :			List[str]  							=      atom_name[0]  # Protein supports only C, N, O, S, this works.
               __UpperCAmelCase     :			Tuple  							=      ''''''

               __UpperCAmelCase     :			Any  							=      '''A'''
               if chain_index is not None:
                    __UpperCAmelCase     :			Optional[Any]  							=      chain_tags[chain_index[i]]

               # PDB is a columnar format, every space matters here!
               __UpperCAmelCase     :			Dict  							=      (
                   F'{record_type:<6}{atom_index:>5} {name:<4}{alt_loc:>1}'
                   F'{res_name_a:>3} {chain_tag:>1}'
                   F'{residue_index[i]:>4}{insertion_code:>1}   '
                   F'{pos[0]:>8.3f}{pos[1]:>8.3f}{pos[2]:>8.3f}'
                   F'{occupancy:>6.2f}{b_factor:>6.2f}          '
                   F'{element:>2}{charge:>2}'
               )
               pdb_lines.append(_lowercase  )
               atom_index += 1

          __UpperCAmelCase     :			Any  							=      i == n - 1
          if chain_index is not None:
               if i != n - 1 and chain_index[i + 1] != prev_chain_index:
                    __UpperCAmelCase     :			List[str]  							=      True
                    __UpperCAmelCase     :			Dict  							=      chain_index[i + 1]

          if should_terminate:
               # Close the chain.
               __UpperCAmelCase     :			List[Any]  							=      '''TER'''
               __UpperCAmelCase     :			Optional[Any]  							=      (
                   F'{chain_end:<6}{atom_index:>5}      {res_atoa(aatype[i]  ):>3} {chain_tag:>1}{residue_index[i]:>4}'
               )
               pdb_lines.append(_lowercase  )
               atom_index += 1

               if i != n - 1:
                    # "prev" is a misnomer here. This happens at the beginning of
                    # each new chain.
                    pdb_lines.extend(get_pdb_headers(_lowercase      , _lowercase  )  )

     pdb_lines.append('''END'''  )
     pdb_lines.append(''''''  )
     return "\n".join(_lowercase  )




def      _a     (						_lowercase :     Protein  ):





     '''simple docstring'''






     return residue_constants.STANDARD_ATOM_MASK[prot.aatype]




def      _a     (						_lowercase :     FeatureDict      , _lowercase :     ModelOutput      , _lowercase :     Optional[np.ndarray] = None      , _lowercase :     Optional[np.ndarray] = None      , _lowercase :     Optional[str] = None      , _lowercase :     Optional[Sequence[str]] = None      , _lowercase :     Optional[Sequence[int]] = None      , ):





     '''simple docstring'''






     return Protein(
         aatype=features['''aatype''']      , atom_positions=result['''final_atom_positions''']      , atom_mask=result['''final_atom_mask''']      , residue_index=features['''residue_index'''] + 1      , b_factors=b_factors if b_factors is not None else np.zeros_like(result['''final_atom_mask''']  )      , chain_index=_lowercase      , remark=_lowercase      , parents=_lowercase      , parents_chain_index=_lowercase      , )




'''simple docstring'''



import gc
import random
import unittest

import torch

from diffusers import (
    IFImgaImgPipeline,
    IFImgaImgSuperResolutionPipeline,
    IFInpaintingPipeline,
    IFInpaintingSuperResolutionPipeline,
    IFPipeline,
    IFSuperResolutionPipeline,
)
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import floats_tensor, load_numpy, require_torch_gpu, skip_mps, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference
from . import IFPipelineTesterMixin



@skip_mps
class      a    (		_a     ,  _a     ,  unittest.TestCase	):


     """simple docstring"""






     SCREAMING_SNAKE_CASE	:			Dict							       =       IFPipeline
     SCREAMING_SNAKE_CASE	:			Optional[int]							       =       TEXT_TO_IMAGE_PARAMS - {"width", "height", "latents"}
     SCREAMING_SNAKE_CASE	:			Optional[int]							       =       TEXT_TO_IMAGE_BATCH_PARAMS
     SCREAMING_SNAKE_CASE	:			str							       =       PipelineTesterMixin.required_optional_params - {"latents"}

     def 			lowerCamelCase__			(    self					:		Tuple      )						->							Optional[Any]:
          return self._get_dummy_components()

     def 			lowerCamelCase__			(    self					:		Optional[Any]							,			snake_case					:		Optional[int]							,			snake_case					:		List[Any]=0      )						->							Optional[Any]:
          if str(snake_case      ).startswith('''mps'''      ):
               __UpperCAmelCase     :			Optional[Any]  							=      torch.manual_seed(snake_case      )
          else:
               __UpperCAmelCase     :			Dict  							=      torch.Generator(device=snake_case      ).manual_seed(snake_case      )

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

          return inputs

     def 			lowerCamelCase__			(    self					:		List[str]      )						->							Optional[int]:
          self._test_save_load_optional_components()

     @unittest.skipIf(torch_device != '''cuda'''							,			reason='''float16 requires CUDA'''      )
     def 			lowerCamelCase__			(    self					:		Dict      )						->							Union[str, Any]:
          # Due to non-determinism in save load of the hf-internal-testing/tiny-random-t5 text encoder
          super().test_save_load_floataa(expected_max_diff=1E-1      )

     def 			lowerCamelCase__			(    self					:		List[str]      )						->							Tuple:
          self._test_attention_slicing_forward_pass(expected_max_diff=1E-2      )

     def 			lowerCamelCase__			(    self					:		Any      )						->							List[str]:
          self._test_save_load_local()

     def 			lowerCamelCase__			(    self					:		Union[str, Any]      )						->							str:
          self._test_inference_batch_single_identical(
              expected_max_diff=1E-2							,			)






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



@slow
@require_torch_gpu
class      a    (		unittest.TestCase	):


     """simple docstring"""






     def 			lowerCamelCase__			(    self					:		Dict      )						->							Dict:
          # clean up the VRAM after each test
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()

     def 			lowerCamelCase__			(    self					:		Any      )						->							Tuple:
          # if

          __UpperCAmelCase     :			Union[str, Any]  							=      IFPipeline.from_pretrained('''DeepFloyd/IF-I-XL-v1.0'''							,			variant='''fp16'''							,			torch_dtype=torch.floataa      )

          __UpperCAmelCase     :			List[str]  							=      IFSuperResolutionPipeline.from_pretrained(
              '''DeepFloyd/IF-II-L-v1.0'''							,			variant='''fp16'''							,			torch_dtype=torch.floataa							,			text_encoder=snake_case							,			tokenizer=snake_case      )

          # pre compute text embeddings and remove T5 to save memory

          pipe_a.text_encoder.to('''cuda'''      )

          __UpperCAmelCase							,       __UpperCAmelCase     :			Tuple  							=      pipe_a.encode_prompt('''anime turtle'''							,			device='''cuda'''      )

          del pipe_a.tokenizer
          del pipe_a.text_encoder
          gc.collect()

          __UpperCAmelCase     :			Any  							=      None
          __UpperCAmelCase     :			Optional[Any]  							=      None

          pipe_a.enable_model_cpu_offload()
          pipe_a.enable_model_cpu_offload()

          pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()      )
          pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()      )

          self._test_if(snake_case							,			snake_case							,			snake_case							,			snake_case      )

          pipe_a.remove_all_hooks()
          pipe_a.remove_all_hooks()

          # img2img

          __UpperCAmelCase     :			Union[str, Any]  							=      IFImgaImgPipeline(**pipe_a.components      )
          __UpperCAmelCase     :			int  							=      IFImgaImgSuperResolutionPipeline(**pipe_a.components      )

          pipe_a.enable_model_cpu_offload()
          pipe_a.enable_model_cpu_offload()

          pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()      )
          pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()      )

          self._test_if_imgaimg(snake_case							,			snake_case							,			snake_case							,			snake_case      )

          pipe_a.remove_all_hooks()
          pipe_a.remove_all_hooks()

          # inpainting

          __UpperCAmelCase     :			List[str]  							=      IFInpaintingPipeline(**pipe_a.components      )
          __UpperCAmelCase     :			List[str]  							=      IFInpaintingSuperResolutionPipeline(**pipe_a.components      )

          pipe_a.enable_model_cpu_offload()
          pipe_a.enable_model_cpu_offload()

          pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()      )
          pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()      )

          self._test_if_inpainting(snake_case							,			snake_case							,			snake_case							,			snake_case      )

     def 			lowerCamelCase__			(    self					:		List[str]							,			snake_case					:		Any							,			snake_case					:		Union[str, Any]							,			snake_case					:		Optional[int]							,			snake_case					:		str      )						->							Optional[int]:
          # pipeline 1

          _start_torch_memory_measurement()

          __UpperCAmelCase     :			Dict  							=      torch.Generator(device='''cpu'''      ).manual_seed(0      )
          __UpperCAmelCase     :			List[str]  							=      pipe_a(
              prompt_embeds=snake_case							,			negative_prompt_embeds=snake_case							,			num_inference_steps=2							,			generator=snake_case							,			output_type='''np'''							,			)

          __UpperCAmelCase     :			List[Any]  							=      output.images[0]

          assert image.shape == (64, 64, 3)

          __UpperCAmelCase     :			List[str]  							=      torch.cuda.max_memory_allocated()
          assert mem_bytes < 13 * 10**9

          __UpperCAmelCase     :			Optional[Any]  							=      load_numpy(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if.npy'''      )
          assert_mean_pixel_difference(snake_case							,			snake_case      )

          # pipeline 2

          _start_torch_memory_measurement()

          __UpperCAmelCase     :			int  							=      torch.Generator(device='''cpu'''      ).manual_seed(0      )

          __UpperCAmelCase     :			Optional[int]  							=      floats_tensor((1, 3, 64, 64)							,			rng=random.Random(0      )      ).to(snake_case      )

          __UpperCAmelCase     :			List[Any]  							=      pipe_a(
              prompt_embeds=snake_case							,			negative_prompt_embeds=snake_case							,			image=snake_case							,			generator=snake_case							,			num_inference_steps=2							,			output_type='''np'''							,			)

          __UpperCAmelCase     :			Optional[Any]  							=      output.images[0]

          assert image.shape == (256, 256, 3)

          __UpperCAmelCase     :			List[str]  							=      torch.cuda.max_memory_allocated()
          assert mem_bytes < 4 * 10**9

          __UpperCAmelCase     :			Tuple  							=      load_numpy(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_superresolution_stage_II.npy'''      )
          assert_mean_pixel_difference(snake_case							,			snake_case      )

     def 			lowerCamelCase__			(    self					:		Optional[int]							,			snake_case					:		Optional[int]							,			snake_case					:		Union[str, Any]							,			snake_case					:		str							,			snake_case					:		Dict      )						->							str:
          # pipeline 1

          _start_torch_memory_measurement()

          __UpperCAmelCase     :			Optional[Any]  							=      floats_tensor((1, 3, 64, 64)							,			rng=random.Random(0      )      ).to(snake_case      )

          __UpperCAmelCase     :			Dict  							=      torch.Generator(device='''cpu'''      ).manual_seed(0      )

          __UpperCAmelCase     :			Dict  							=      pipe_a(
              prompt_embeds=snake_case							,			negative_prompt_embeds=snake_case							,			image=snake_case							,			num_inference_steps=2							,			generator=snake_case							,			output_type='''np'''							,			)

          __UpperCAmelCase     :			int  							=      output.images[0]

          assert image.shape == (64, 64, 3)

          __UpperCAmelCase     :			Any  							=      torch.cuda.max_memory_allocated()
          assert mem_bytes < 10 * 10**9

          __UpperCAmelCase     :			Optional[Any]  							=      load_numpy(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_img2img.npy'''      )
          assert_mean_pixel_difference(snake_case							,			snake_case      )

          # pipeline 2

          _start_torch_memory_measurement()

          __UpperCAmelCase     :			Optional[int]  							=      torch.Generator(device='''cpu'''      ).manual_seed(0      )

          __UpperCAmelCase     :			Optional[Any]  							=      floats_tensor((1, 3, 256, 256)							,			rng=random.Random(0      )      ).to(snake_case      )
          __UpperCAmelCase     :			List[str]  							=      floats_tensor((1, 3, 64, 64)							,			rng=random.Random(0      )      ).to(snake_case      )

          __UpperCAmelCase     :			List[str]  							=      pipe_a(
              prompt_embeds=snake_case							,			negative_prompt_embeds=snake_case							,			image=snake_case							,			original_image=snake_case							,			generator=snake_case							,			num_inference_steps=2							,			output_type='''np'''							,			)

          __UpperCAmelCase     :			int  							=      output.images[0]

          assert image.shape == (256, 256, 3)

          __UpperCAmelCase     :			int  							=      torch.cuda.max_memory_allocated()
          assert mem_bytes < 4 * 10**9

          __UpperCAmelCase     :			Any  							=      load_numpy(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_img2img_superresolution_stage_II.npy'''      )
          assert_mean_pixel_difference(snake_case							,			snake_case      )






     def 			lowerCamelCase__			(    self					:		str							,			snake_case					:		Union[str, Any]							,			snake_case					:		Optional[int]							,			snake_case					:		Optional[Any]							,			snake_case					:		Optional[Any]      )						->							Optional[int]:
          # pipeline 1

          _start_torch_memory_measurement()

          __UpperCAmelCase     :			List[Any]  							=      floats_tensor((1, 3, 64, 64)							,			rng=random.Random(0      )      ).to(snake_case      )
          __UpperCAmelCase     :			List[str]  							=      floats_tensor((1, 3, 64, 64)							,			rng=random.Random(1      )      ).to(snake_case      )

          __UpperCAmelCase     :			Optional[Any]  							=      torch.Generator(device='''cpu'''      ).manual_seed(0      )
          __UpperCAmelCase     :			Dict  							=      pipe_a(
              prompt_embeds=snake_case							,			negative_prompt_embeds=snake_case							,			image=snake_case							,			mask_image=snake_case							,			num_inference_steps=2							,			generator=snake_case							,			output_type='''np'''							,			)

          __UpperCAmelCase     :			Union[str, Any]  							=      output.images[0]

          assert image.shape == (64, 64, 3)

          __UpperCAmelCase     :			Tuple  							=      torch.cuda.max_memory_allocated()
          assert mem_bytes < 10 * 10**9

          __UpperCAmelCase     :			Tuple  							=      load_numpy(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_inpainting.npy'''      )
          assert_mean_pixel_difference(snake_case							,			snake_case      )

          # pipeline 2

          _start_torch_memory_measurement()

          __UpperCAmelCase     :			int  							=      torch.Generator(device='''cpu'''      ).manual_seed(0      )

          __UpperCAmelCase     :			Tuple  							=      floats_tensor((1, 3, 64, 64)							,			rng=random.Random(0      )      ).to(snake_case      )
          __UpperCAmelCase     :			Any  							=      floats_tensor((1, 3, 256, 256)							,			rng=random.Random(0      )      ).to(snake_case      )
          __UpperCAmelCase     :			Tuple  							=      floats_tensor((1, 3, 256, 256)							,			rng=random.Random(1      )      ).to(snake_case      )

          __UpperCAmelCase     :			Union[str, Any]  							=      pipe_a(
              prompt_embeds=snake_case							,			negative_prompt_embeds=snake_case							,			image=snake_case							,			mask_image=snake_case							,			original_image=snake_case							,			generator=snake_case							,			num_inference_steps=2							,			output_type='''np'''							,			)

          __UpperCAmelCase     :			List[Any]  							=      output.images[0]

          assert image.shape == (256, 256, 3)

          __UpperCAmelCase     :			Any  							=      torch.cuda.max_memory_allocated()
          assert mem_bytes < 4 * 10**9

          __UpperCAmelCase     :			str  							=      load_numpy(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_inpainting_superresolution_stage_II.npy'''      )
          assert_mean_pixel_difference(snake_case							,			snake_case      )




def      _a     (						):





     '''simple docstring'''






     torch.cuda.empty_cache()
     torch.cuda.reset_max_memory_allocated()
     torch.cuda.reset_peak_memory_stats()