

'''simple docstring'''
# This code is adapted from OpenAI's release
# https://github.com/openai/human-eval/blob/master/human_eval/execution.py

import contextlib
import faulthandler
import io
import multiprocessing
import os
import platform
import signal
import tempfile




def 	UpperCamelCase   (       lowercase_						:    Union[str, Any]     ,					lowercase_						:    List[str]     ,					lowercase_						:    List[str]     ,					lowercase_						:    Optional[Any]	)		->	List[Any]:




  '''simple docstring'''



  lowercase									=multiprocessing.Manager()
  lowercase									=manager.list()

  lowercase									=multiprocessing.Process(target=lowercase_     ,					args=(check_program, result, timeout)	)
  p.start()
  p.join(timeout=timeout + 1	)
  if p.is_alive():
    p.kill()

  if not result:
    result.append('''timed out'''	)

  return {
      "task_id": task_id,
      "passed": result[0] == "passed",
      "result": result[0],
      "completion_id": completion_id,
  }




def 	UpperCamelCase   (       lowercase_						:    int     ,					lowercase_						:    Optional[int]     ,					lowercase_						:    List[str]	)		->	List[str]:




  '''simple docstring'''



  with create_tempdir():
    # These system calls are needed when cleaning up tempdir.
    import os
    import shutil

    lowercase									=shutil.rmtree
    lowercase									=os.rmdir
    lowercase									=os.chdir

    # Disable functionalities that can make destructive changes to the test.
    reliability_guard()

    # Run program.
    try:
      lowercase									={}
      with swallow_io():
        with time_limit(lowercase_	):
          exec(lowercase_     ,					lowercase_	)
      result.append('''passed'''	)
    except TimeoutException:
      result.append('''timed out'''	)
    except BaseException as e:
      result.append(f'failed: {e}'	)

    # Needed for cleaning up.
    lowercase									=rmtree
    lowercase									=rmdir
    lowercase									=chdir




@contextlib.contextmanager
def 	UpperCamelCase   (       lowercase_						:    List[str]	)		->	Optional[int]:




  '''simple docstring'''



  def signal_handler(lowercase_						:    Optional[int]     ,					lowercase_						:    str	):
    raise TimeoutException('''Timed out!'''	)

  signal.setitimer(signal.ITIMER_REAL     ,					lowercase_	)
  signal.signal(signal.SIGALRM     ,					lowercase_	)
  try:
    yield
  finally:
    signal.setitimer(signal.ITIMER_REAL     ,					0	)




@contextlib.contextmanager
def 	UpperCamelCase   (       )		->	Dict:




  '''simple docstring'''



  lowercase									=WriteOnlyStringIO()
  with contextlib.redirect_stdout(lowercase_	):
    with contextlib.redirect_stderr(lowercase_	):
      with redirect_stdin(lowercase_	):
        yield




@contextlib.contextmanager
def 	UpperCamelCase   (       )		->	List[Any]:




  '''simple docstring'''



  with tempfile.TemporaryDirectory() as dirname:
    with chdir(lowercase_	):
      yield dirname

class 						__magic_name__   (							__SCREAMING_SNAKE_CASE ):
 pass

class 						__magic_name__   (							io.StringIO ):







 def 						_A(       self  , *snake_case_  , **snake_case_   ):
   raise OSError







 def 						_A(       self  , *snake_case_  , **snake_case_   ):
   raise OSError







 def 						_A(       self  , *snake_case_  , **snake_case_   ):
   raise OSError







 def 						_A(       self  , *snake_case_  , **snake_case_   ):
   return False

class 						__magic_name__   (							contextlib._RedirectStream ):  # type: ignore
 UpperCamelCase__     =       'stdin'




@contextlib.contextmanager
def 	UpperCamelCase   (       lowercase_						:    List[str]	)		->	Optional[Any]:




  '''simple docstring'''



  if root == ".":
    yield
    return
  lowercase									=os.getcwd()
  os.chdir(lowercase_	)
  try:
    yield
  except BaseException as exc:
    raise exc
  finally:
    os.chdir(lowercase_	)




def 	UpperCamelCase   (       lowercase_						:    str=None	)		->	Tuple:




  '''simple docstring'''



  if maximum_memory_bytes is not None:
    import resource

    resource.setrlimit(resource.RLIMIT_AS     ,					(maximum_memory_bytes, maximum_memory_bytes)	)
    resource.setrlimit(resource.RLIMIT_DATA     ,					(maximum_memory_bytes, maximum_memory_bytes)	)
    if not platform.uname().system == "Darwin":
      resource.setrlimit(resource.RLIMIT_STACK     ,					(maximum_memory_bytes, maximum_memory_bytes)	)

  faulthandler.disable()

  import builtins

  lowercase									=None
  lowercase									=None

  import os

  lowercase									='''1'''

  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None

  import shutil

  lowercase									=None
  lowercase									=None
  lowercase									=None

  import subprocess

  lowercase									=None  # type: ignore

  lowercase									=None

  import sys

  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None
  lowercase									=None





'''simple docstring'''

import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel

from diffusers import DDIMScheduler, LDMPipeline, UNetaDModel, VQModel
from diffusers.utils.testing_utils import enable_full_determinism, require_torch, slow, torch_device


enable_full_determinism()

class       __magic_name__		(       unittest.TestCase  ):







							@property
							def 				__lowercase     (				self   :			int						):
									torch.manual_seed(0						)
									_a	: int								=				UNetaDModel(
									    block_out_channels=(32, 64)      ,layers_per_block=2      ,sample_size=32      ,in_channels=3      ,out_channels=3      ,down_block_types=('DownBlock2D', 'AttnDownBlock2D')      ,up_block_types=('AttnUpBlock2D', 'UpBlock2D')      ,)
									return model







							@property
							def 				__lowercase     (				self   :			Any						):
									torch.manual_seed(0						)
									_a	: int								=				VQModel(
									    block_out_channels=[32, 64]      ,in_channels=3      ,out_channels=3      ,down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']      ,up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']      ,latent_channels=3      ,)
									return model







							@property
							def 				__lowercase     (				self   :			Union[str, Any]						):
									torch.manual_seed(0						)
									_a	: Dict								=				CLIPTextConfig(
									    bos_token_id=0      ,eos_token_id=2      ,hidden_size=32      ,intermediate_size=37      ,layer_norm_eps=1E-05      ,num_attention_heads=4      ,num_hidden_layers=5      ,pad_token_id=1      ,vocab_size=1000      ,)
									return CLIPTextModel(_UpperCAmelCase						)







							def 				__lowercase     (				self   :			Dict						):
									_a	: Union[str, Any]								=				self.dummy_uncond_unet
									_a	: Union[str, Any]								=				DDIMScheduler()
									_a	: Union[str, Any]								=				self.dummy_vq_model

									_a	: List[Any]								=				LDMPipeline(unet=_UpperCAmelCase      ,vqvae=_UpperCAmelCase      ,scheduler=_UpperCAmelCase						)
									ldm.to(_UpperCAmelCase						)
									ldm.set_progress_bar_config(disable=_UpperCAmelCase						)

									_a	: Dict								=				torch.manual_seed(0						)
									_a	: int								=				ldm(generator=_UpperCAmelCase      ,num_inference_steps=2      ,output_type='numpy'						).images

									_a	: Tuple								=				torch.manual_seed(0						)
									_a	: List[Any]								=				ldm(generator=_UpperCAmelCase      ,num_inference_steps=2      ,output_type='numpy'      ,return_dict=_UpperCAmelCase						)[0]

									_a	: Any								=				image[0, -3:, -3:, -1]
									_a	: List[Any]								=				image_from_tuple[0, -3:, -3:, -1]

									assert image.shape == (1, 64, 64, 3)
									_a	: Union[str, Any]								=				np.array([0.85_12, 0.8_18, 0.64_11, 0.68_08, 0.44_65, 0.56_18, 0.46, 0.62_31, 0.51_72]						)
									_a	: Any								=				1E-2 if torch_device != 'mps' else 3E-2

									assert np.abs(image_slice.flatten() - expected_slice						).max() < tolerance
									assert np.abs(image_from_tuple_slice.flatten() - expected_slice						).max() < tolerance

@slow
@require_torch
class       __magic_name__		(       unittest.TestCase  ):







							def 				__lowercase     (				self   :			Optional[int]						):
									_a	: Dict								=				LDMPipeline.from_pretrained('CompVis/ldm-celebahq-256'						)
									ldm.to(_UpperCAmelCase						)
									ldm.set_progress_bar_config(disable=_UpperCAmelCase						)

									_a	: str								=				torch.manual_seed(0						)
									_a	: Optional[Any]								=				ldm(generator=_UpperCAmelCase      ,num_inference_steps=5      ,output_type='numpy'						).images

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

									assert image.shape == (1, 256, 256, 3)
									_a	: Union[str, Any]								=				np.array([0.43_99, 0.4_49_75, 0.4_68_25, 0.4_74, 0.43_59, 0.45_81, 0.4_50_95, 0.43_41, 0.44_47]						)
									_a	: List[str]								=				1E-2 if torch_device != 'mps' else 3E-2

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




