






"""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_	:				Dict										= Mapping[str, np.ndarray]
UpperCamelCase_	:				List[Any]										= Mapping[str, Any]  # Is a nested dict.
UpperCamelCase_	:				List[Any]										= 0.01


@dataclasses.dataclass(frozen=lowerCAmelCase						)
class      _lowercase :
      _a   :     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'.
      _a   :     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.
      _a   :     np.ndarray  # [num_res, num_atom_type]

      # Residue index as used in PDB. It is not necessarily continuous or 0-indexed.
      _a   :     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.
      _a   :     np.ndarray  # [num_res, num_atom_type]

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

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

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

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



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

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

      assert aatype is not None

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



def     __lowercase					(    a :				Protein       ,		a :				int = 0 )					->					List[str]:
      __snake_case      :    List[str]  			=[]

      __snake_case      :    int  			=prot.remark
      if remark is not None:
            pdb_headers.append(f'''REMARK {remark}''' )

      __snake_case      :    Any  			=prot.parents
      __snake_case      :    Union[str, Any]  			=prot.parents_chain_index
      if parents is not None and parents_chain_index is not None:
            __snake_case      :    Tuple  			=[p for i, p in zip(a       ,		a ) if i == chain_id]

      if parents is None or len(a ) == 0:
            __snake_case      :    Union[str, Any]  			=['''N/A''']

      pdb_headers.append(f'''PARENT {" ".join(a )}''' )

      return pdb_headers



def     __lowercase					(    a :				Protein       ,		a :				str )					->					str:
      __snake_case      :    List[str]  			=[]
      __snake_case      :    str  			=pdb_str.split('''\n''' )

      __snake_case      :    str  			=prot.remark
      if remark is not None:
            out_pdb_lines.append(f'''REMARK {remark}''' )

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

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

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

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

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

                  out_pdb_lines.append(make_parent_line(a ) )

      return "\n".join(a )



def     __lowercase					(    a :				Protein )					->					str:
      __snake_case      :    int  			=residue_constants.restypes + ['''X''']

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

      __snake_case      :    Dict  			=residue_constants.atom_types

      __snake_case      :    List[str]  			=[]

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

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

      __snake_case      :    int  			=get_pdb_headers(a )
      if len(a ) > 0:
            pdb_lines.extend(a )

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

                  __snake_case      :    Any  			='''ATOM'''
                  __snake_case      :    Union[str, Any]  			=atom_name if len(a ) == 4 else f''' {atom_name}'''
                  __snake_case      :    Tuple  			=''''''
                  __snake_case      :    str  			=''''''
                  __snake_case      :    List[str]  			=1.0_0
                  __snake_case      :    Dict  			=atom_name[0]  # Protein supports only C, N, O, S, this works.
                  __snake_case      :    Dict  			=''''''

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

                  # PDB is a columnar format, every space matters here!
                  __snake_case      :    int  			=(
                      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(a )
                  atom_index += 1

            __snake_case      :    Any  			=i == n - 1
            if chain_index is not None:
                  if i != n - 1 and chain_index[i + 1] != prev_chain_index:
                        __snake_case      :    Tuple  			=True
                        __snake_case      :    Tuple  			=chain_index[i + 1]

            if should_terminate:
                  # Close the chain.
                  __snake_case      :    Tuple  			='''TER'''
                  __snake_case      :    Any  			=(
                      f'''{chain_end:<6}{atom_index:>5}      {res_atoa(aatype[i] ):>3} {chain_tag:>1}{residue_index[i]:>4}'''
                  )
                  pdb_lines.append(a )
                  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(a       ,		a ) )

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



def     __lowercase					(    a :				Protein )					->					np.ndarray:
      return residue_constants.STANDARD_ATOM_MASK[prot.aatype]



def     __lowercase					(    a :				FeatureDict       ,		a :				ModelOutput       ,		a :				Optional[np.ndarray] = None       ,		a :				Optional[np.ndarray] = None       ,		a :				Optional[str] = None       ,		a :				Optional[Sequence[str]] = None       ,		a :				Optional[Sequence[int]] = None       ,		)					->					Protein:
      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=a       ,		remark=a       ,		parents=a       ,		parents_chain_index=a       ,		)










"""simple docstring"""





import tempfile
import unittest

import numpy as np

from diffusers import (
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    OnnxStableDiffusionPipeline,
    PNDMScheduler,
)
from diffusers.utils.testing_utils import is_onnx_available, nightly, require_onnxruntime, require_torch_gpu

from ..test_pipelines_onnx_common import OnnxPipelineTesterMixin


if is_onnx_available():
    import onnxruntime as ort


class      _lowercase (     lowerCAmelCase		,  unittest.TestCase						):
      _a   :     Tuple								=   '''hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline'''


      def _UpperCamelCase						(    self				:	Dict							,   a				:	Optional[Any]=0						):





            """simple docstring"""







            __snake_case      :    List[str]  			=np.random.RandomState(a						)
            __snake_case      :    Union[str, Any]  			={
                '''prompt''': '''A painting of a squirrel eating a burger''',
                '''generator''': generator,
                '''num_inference_steps''': 2,
                '''guidance_scale''': 7.5,
                '''output_type''': '''numpy''',
            }
            return inputs


      def _UpperCamelCase						(    self				:	str						):





            """simple docstring"""







            __snake_case      :    Dict  			=OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint							,   provider='''CPUExecutionProvider'''						)
            pipe.set_progress_bar_config(disable=a						)

            __snake_case      :    Tuple  			=self.get_dummy_inputs()
            __snake_case      :    List[str]  			=pipe(**a						).images
            __snake_case      :    int  			=image[0, -3:, -3:, -1]

            assert image.shape == (1, 1_2_8, 1_2_8, 3)
            __snake_case      :    str  			=np.array([0.6_5_0_7_2, 0.5_8_4_9_2, 0.4_8_2_1_9, 0.5_5_5_2_1, 0.5_3_1_8_0, 0.5_5_9_3_9, 0.5_0_6_9_7, 0.3_9_8_0_0, 0.4_6_4_5_5]						)

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


      def _UpperCamelCase						(    self				:	Union[str, Any]						):





            """simple docstring"""







            __snake_case      :    Dict  			=OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint							,   provider='''CPUExecutionProvider'''						)
            __snake_case      :    Dict  			=PNDMScheduler.from_config(pipe.scheduler.config							,   skip_prk_steps=a						)
            pipe.set_progress_bar_config(disable=a						)

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

            assert image.shape == (1, 1_2_8, 1_2_8, 3)
            __snake_case      :    Optional[Any]  			=np.array([0.6_5_8_6_3, 0.5_9_4_2_5, 0.4_9_3_2_6, 0.5_6_3_1_3, 0.5_3_8_7_5, 0.5_6_6_2_7, 0.5_1_0_6_5, 0.3_9_7_7_7, 0.4_6_3_3_0]						)

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


      def _UpperCamelCase						(    self				:	Optional[int]						):





            """simple docstring"""







            __snake_case      :    Any  			=OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint							,   provider='''CPUExecutionProvider'''						)
            __snake_case      :    Any  			=LMSDiscreteScheduler.from_config(pipe.scheduler.config						)
            pipe.set_progress_bar_config(disable=a						)

            __snake_case      :    int  			=self.get_dummy_inputs()
            __snake_case      :    List[str]  			=pipe(**a						).images
            __snake_case      :    str  			=image[0, -3:, -3:, -1]

            assert image.shape == (1, 1_2_8, 1_2_8, 3)
            __snake_case      :    int  			=np.array([0.5_3_7_5_5, 0.6_0_7_8_6, 0.4_7_4_0_2, 0.4_9_4_8_8, 0.5_1_8_6_9, 0.4_9_8_1_9, 0.4_7_9_8_5, 0.3_8_9_5_7, 0.4_4_2_7_9]						)

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


      def _UpperCamelCase						(    self				:	List[str]						):





            """simple docstring"""







            __snake_case      :    Optional[int]  			=OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint							,   provider='''CPUExecutionProvider'''						)
            __snake_case      :    Optional[Any]  			=EulerDiscreteScheduler.from_config(pipe.scheduler.config						)
            pipe.set_progress_bar_config(disable=a						)

            __snake_case      :    str  			=self.get_dummy_inputs()
            __snake_case      :    int  			=pipe(**a						).images
            __snake_case      :    str  			=image[0, -3:, -3:, -1]

            assert image.shape == (1, 1_2_8, 1_2_8, 3)
            __snake_case      :    Union[str, Any]  			=np.array([0.5_3_7_5_5, 0.6_0_7_8_6, 0.4_7_4_0_2, 0.4_9_4_8_8, 0.5_1_8_6_9, 0.4_9_8_1_9, 0.4_7_9_8_5, 0.3_8_9_5_7, 0.4_4_2_7_9]						)

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


      def _UpperCamelCase						(    self				:	List[Any]						):





            """simple docstring"""







            __snake_case      :    Any  			=OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint							,   provider='''CPUExecutionProvider'''						)
            __snake_case      :    List[str]  			=EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config						)
            pipe.set_progress_bar_config(disable=a						)

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

            assert image.shape == (1, 1_2_8, 1_2_8, 3)
            __snake_case      :    Tuple  			=np.array([0.5_3_8_1_7, 0.6_0_8_1_2, 0.4_7_3_8_4, 0.4_9_5_3_0, 0.5_1_8_9_4, 0.4_9_8_1_4, 0.4_7_9_8_4, 0.3_8_9_5_8, 0.4_4_2_7_1]						)

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


      def _UpperCamelCase						(    self				:	Dict						):





            """simple docstring"""







            __snake_case      :    Tuple  			=OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint							,   provider='''CPUExecutionProvider'''						)
            __snake_case      :    Any  			=DPMSolverMultistepScheduler.from_config(pipe.scheduler.config						)
            pipe.set_progress_bar_config(disable=a						)

            __snake_case      :    Tuple  			=self.get_dummy_inputs()
            __snake_case      :    Tuple  			=pipe(**a						).images
            __snake_case      :    Optional[Any]  			=image[0, -3:, -3:, -1]

            assert image.shape == (1, 1_2_8, 1_2_8, 3)
            __snake_case      :    Dict  			=np.array([0.5_3_8_9_5, 0.6_0_8_0_8, 0.4_7_9_3_3, 0.4_9_6_0_8, 0.5_1_8_8_6, 0.4_9_9_5_0, 0.4_8_0_5_3, 0.3_8_9_5_7, 0.4_4_2_0_0]						)

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


      def _UpperCamelCase						(    self				:	int						):





            """simple docstring"""







            __snake_case      :    Dict  			=OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint							,   provider='''CPUExecutionProvider'''						)
            pipe.set_progress_bar_config(disable=a						)

            __snake_case      :    Optional[int]  			=self.get_dummy_inputs()
            __snake_case      :    Any  			=3 * [inputs['''prompt''']]

            # forward
            __snake_case      :    Any  			=pipe(**a						)
            __snake_case      :    str  			=output.images[0, -3:, -3:, -1]

            __snake_case      :    Tuple  			=self.get_dummy_inputs()
            __snake_case      :    Any  			=3 * [inputs.pop('''prompt'''						)]

            __snake_case      :    Optional[Any]  			=pipe.tokenizer(
                a							,   padding='''max_length'''							,   max_length=pipe.tokenizer.model_max_length							,   truncation=a							,   return_tensors='''np'''							,   )
            __snake_case      :    List[Any]  			=text_inputs['''input_ids''']

            __snake_case      :    str  			=pipe.text_encoder(input_ids=text_inputs.astype(np.intaa						)						)[0]

            __snake_case      :    Optional[Any]  			=prompt_embeds

            # forward
            __snake_case      :    Dict  			=pipe(**a						)
            __snake_case      :    Any  			=output.images[0, -3:, -3:, -1]

            assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()						).max() < 1e-4







      def _UpperCamelCase						(    self				:	Any						):





            """simple docstring"""







            __snake_case      :    Dict  			=OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint							,   provider='''CPUExecutionProvider'''						)
            pipe.set_progress_bar_config(disable=a						)

            __snake_case      :    List[Any]  			=self.get_dummy_inputs()
            __snake_case      :    Optional[Any]  			=3 * ['''this is a negative prompt''']
            __snake_case      :    List[str]  			=negative_prompt
            __snake_case      :    str  			=3 * [inputs['''prompt''']]

            # forward
            __snake_case      :    int  			=pipe(**a						)
            __snake_case      :    Union[str, Any]  			=output.images[0, -3:, -3:, -1]

            __snake_case      :    Tuple  			=self.get_dummy_inputs()
            __snake_case      :    Union[str, Any]  			=3 * [inputs.pop('''prompt'''						)]

            __snake_case      :    Optional[int]  			=[]
            for p in [prompt, negative_prompt]:
                  __snake_case      :    Optional[int]  			=pipe.tokenizer(
                      a							,   padding='''max_length'''							,   max_length=pipe.tokenizer.model_max_length							,   truncation=a							,   return_tensors='''np'''							,   )
                  __snake_case      :    Optional[Any]  			=text_inputs['''input_ids''']

                  embeds.append(pipe.text_encoder(input_ids=text_inputs.astype(np.intaa						)						)[0]						)

            __snake_case			,				__snake_case      :    Optional[Any]  			=embeds

            # forward
            __snake_case      :    Any  			=pipe(**a						)
            __snake_case      :    Union[str, Any]  			=output.images[0, -3:, -3:, -1]

            assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()						).max() < 1e-4



@nightly
@require_onnxruntime
@require_torch_gpu
class      _lowercase (     unittest.TestCase						):


      @property
      def _UpperCamelCase						(    self				:	List[Any]						):





            """simple docstring"""







            return (
                "CUDAExecutionProvider",
                {
                    "gpu_mem_limit": "15000000000",  # 15GB
                    "arena_extend_strategy": "kSameAsRequested",
                },
            )


      @property
      def _UpperCamelCase						(    self				:	Union[str, Any]						):





            """simple docstring"""







            __snake_case      :    List[str]  			=ort.SessionOptions()
            __snake_case      :    Optional[int]  			=False
            return options


      def _UpperCamelCase						(    self				:	Any						):





            """simple docstring"""







            __snake_case      :    List[Any]  			=OnnxStableDiffusionPipeline.from_pretrained(
                '''CompVis/stable-diffusion-v1-4'''							,   revision='''onnx'''							,   safety_checker=a							,   feature_extractor=a							,   provider=self.gpu_provider							,   sess_options=self.gpu_options							,   )
            sd_pipe.set_progress_bar_config(disable=a						)

            __snake_case      :    List[str]  			='''A painting of a squirrel eating a burger'''
            np.random.seed(0						)
            __snake_case      :    Tuple  			=sd_pipe([prompt]							,   guidance_scale=6.0							,   num_inference_steps=1_0							,   output_type='''np'''						)
            __snake_case      :    Union[str, Any]  			=output.images

            __snake_case      :    Optional[Any]  			=image[0, -3:, -3:, -1]

            assert image.shape == (1, 5_1_2, 5_1_2, 3)
            __snake_case      :    Any  			=np.array([0.0_4_5_2, 0.0_3_9_0, 0.0_0_8_7, 0.0_3_5_0, 0.0_6_1_7, 0.0_3_6_4, 0.0_5_4_4, 0.0_5_2_3, 0.0_7_2_0]						)

            assert np.abs(image_slice.flatten() - expected_slice						).max() < 1e-3


      def _UpperCamelCase						(    self				:	Optional[Any]						):





            """simple docstring"""







            __snake_case      :    List[str]  			=DDIMScheduler.from_pretrained(
                '''runwayml/stable-diffusion-v1-5'''							,   subfolder='''scheduler'''							,   revision='''onnx'''						)
            __snake_case      :    List[Any]  			=OnnxStableDiffusionPipeline.from_pretrained(
                '''runwayml/stable-diffusion-v1-5'''							,   revision='''onnx'''							,   scheduler=a							,   safety_checker=a							,   feature_extractor=a							,   provider=self.gpu_provider							,   sess_options=self.gpu_options							,   )
            sd_pipe.set_progress_bar_config(disable=a						)

            __snake_case      :    Optional[Any]  			='''open neural network exchange'''
            __snake_case      :    Optional[int]  			=np.random.RandomState(0						)
            __snake_case      :    int  			=sd_pipe([prompt]							,   guidance_scale=7.5							,   num_inference_steps=1_0							,   generator=a							,   output_type='''np'''						)
            __snake_case      :    Union[str, Any]  			=output.images
            __snake_case      :    str  			=image[0, -3:, -3:, -1]

            assert image.shape == (1, 5_1_2, 5_1_2, 3)
            __snake_case      :    Union[str, Any]  			=np.array([0.2_8_6_7, 0.1_9_7_4, 0.1_4_8_1, 0.7_2_9_4, 0.7_2_5_1, 0.6_6_6_7, 0.4_1_9_4, 0.5_6_4_2, 0.6_4_8_6]						)

            assert np.abs(image_slice.flatten() - expected_slice						).max() < 1e-3


      def _UpperCamelCase						(    self				:	Tuple						):





            """simple docstring"""







            __snake_case      :    List[str]  			=LMSDiscreteScheduler.from_pretrained(
                '''runwayml/stable-diffusion-v1-5'''							,   subfolder='''scheduler'''							,   revision='''onnx'''						)
            __snake_case      :    int  			=OnnxStableDiffusionPipeline.from_pretrained(
                '''runwayml/stable-diffusion-v1-5'''							,   revision='''onnx'''							,   scheduler=a							,   safety_checker=a							,   feature_extractor=a							,   provider=self.gpu_provider							,   sess_options=self.gpu_options							,   )
            sd_pipe.set_progress_bar_config(disable=a						)

            __snake_case      :    Optional[int]  			='''open neural network exchange'''
            __snake_case      :    Optional[Any]  			=np.random.RandomState(0						)
            __snake_case      :    Any  			=sd_pipe([prompt]							,   guidance_scale=7.5							,   num_inference_steps=1_0							,   generator=a							,   output_type='''np'''						)
            __snake_case      :    Optional[int]  			=output.images
            __snake_case      :    Any  			=image[0, -3:, -3:, -1]

            assert image.shape == (1, 5_1_2, 5_1_2, 3)
            __snake_case      :    Optional[int]  			=np.array([0.2_3_0_6, 0.1_9_5_9, 0.1_5_9_3, 0.6_5_4_9, 0.6_3_9_4, 0.5_4_0_8, 0.5_0_6_5, 0.6_0_1_0, 0.6_1_6_1]						)

            assert np.abs(image_slice.flatten() - expected_slice						).max() < 1e-3


      def _UpperCamelCase						(    self				:	Union[str, Any]						):





            """simple docstring"""







            __snake_case      :    Union[str, Any]  			=0

            def test_callback_fn(a				:	int							,   a				:	int							,   a				:	np.ndarray						) -> None:
                  __snake_case      :    Dict  			=True
                  nonlocal number_of_steps
                  number_of_steps += 1
                  if step == 0:
                        assert latents.shape == (1, 4, 6_4, 6_4)
                        __snake_case      :    Union[str, Any]  			=latents[0, -3:, -3:, -1]
                        __snake_case      :    str  			=np.array(
                            [-0.6_7_7_2, -0.3_8_3_5, -1.2_4_5_6, 0.1_9_0_5, -1.0_9_7_4, 0.6_9_6_7, -1.9_3_5_3, 0.0_1_7_8, 1.0_1_6_7]						)

                        assert np.abs(latents_slice.flatten() - expected_slice						).max() < 1e-3
                  elif step == 5:
                        assert latents.shape == (1, 4, 6_4, 6_4)
                        __snake_case      :    List[Any]  			=latents[0, -3:, -3:, -1]
                        __snake_case      :    List[Any]  			=np.array(
                            [-0.3_3_5_1, 0.2_2_4_1, -0.1_8_3_7, -0.2_3_2_5, -0.6_5_7_7, 0.3_3_9_3, -0.0_2_4_1, 0.5_8_9_9, 1.3_8_7_5]						)

                        assert np.abs(latents_slice.flatten() - expected_slice						).max() < 1e-3

            __snake_case      :    str  			=False

            __snake_case      :    int  			=OnnxStableDiffusionPipeline.from_pretrained(
                '''runwayml/stable-diffusion-v1-5'''							,   revision='''onnx'''							,   safety_checker=a							,   feature_extractor=a							,   provider=self.gpu_provider							,   sess_options=self.gpu_options							,   )
            pipe.set_progress_bar_config(disable=a						)

            __snake_case      :    List[Any]  			='''Andromeda galaxy in a bottle'''

            __snake_case      :    Optional[int]  			=np.random.RandomState(0						)
            pipe(
                prompt=a							,   num_inference_steps=5							,   guidance_scale=7.5							,   generator=a							,   callback=a							,   callback_steps=1							,   )
            assert test_callback_fn.has_been_called
            assert number_of_steps == 6







      def _UpperCamelCase						(    self				:	List[Any]						):





            """simple docstring"""







            __snake_case      :    Optional[Any]  			=OnnxStableDiffusionPipeline.from_pretrained(
                '''runwayml/stable-diffusion-v1-5'''							,   revision='''onnx'''							,   safety_checker=a							,   feature_extractor=a							,   provider=self.gpu_provider							,   sess_options=self.gpu_options							,   )
            assert isinstance(a							,   a						)
            assert pipe.safety_checker is None

            __snake_case      :    int  			=pipe('''example prompt'''							,   num_inference_steps=2						).images[0]
            assert image is not None

            # check that there's no error when saving a pipeline with one of the models being None
            with tempfile.TemporaryDirectory() as tmpdirname:
                  pipe.save_pretrained(a						)
                  __snake_case      :    List[Any]  			=OnnxStableDiffusionPipeline.from_pretrained(a						)

            # sanity check that the pipeline still works
            assert pipe.safety_checker is None
            __snake_case      :    Any  			=pipe('''example prompt'''							,   num_inference_steps=2						).images[0]
            assert image is not None


