





"""simple docstring"""





import gc
import math
import unittest

import torch

from diffusers import UNetaDModel
from diffusers.utils import floats_tensor, logging, slow, torch_all_close, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from .test_modeling_common import ModelTesterMixin, UNetTesterMixin


__A											=		logging.get_logger(__name__)

enable_full_determinism()






class 					_lowerCAmelCase   (     a  ,   a  ,   unittest.TestCase       ):
       """simple docstring"""




       __magic_name__     :Optional[Any]		  =					UNetaDModel
       __magic_name__     :int		  =					"""sample"""






       @property
       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:Dict											=							4
         lowerCAmelCase__			:List[Any]											=							3
         lowerCAmelCase__			:Dict											=							(3_2, 3_2)

         lowerCAmelCase__			:int											=							floats_tensor((batch_size, num_channels) + sizes      ).to(__UpperCAmelCase      )
         lowerCAmelCase__			:Optional[int]											=							torch.tensor([1_0]      ).to(__UpperCAmelCase      )

         return {"sample": noise, "timestep": time_step}






       @property
       def      snake_case						(    self      ):



         '''simple docstring'''

         return (3, 3_2, 3_2)






       @property
       def      snake_case						(    self      ):



         '''simple docstring'''

         return (3, 3_2, 3_2)






       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:List[Any]											=							{
             'block_out_channels': (3_2, 6_4),
             'down_block_types': ('DownBlock2D', 'AttnDownBlock2D'),
             'up_block_types': ('AttnUpBlock2D', 'UpBlock2D'),
             'attention_head_dim': 3,
             'out_channels': 3,
             'in_channels': 3,
             'layers_per_block': 2,
             'sample_size': 3_2,
         }
         lowerCAmelCase__			:int											=							self.dummy_input
         return init_dict, inputs_dict






class 					_lowerCAmelCase   (     a  ,   a  ,   unittest.TestCase       ):
       """simple docstring"""




       __magic_name__     :Dict		  =					UNetaDModel
       __magic_name__     :Dict		  =					"""sample"""






       @property
       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:str											=							4
         lowerCAmelCase__			:int											=							4
         lowerCAmelCase__			:List[Any]											=							(3_2, 3_2)

         lowerCAmelCase__			:List[str]											=							floats_tensor((batch_size, num_channels) + sizes      ).to(__UpperCAmelCase      )
         lowerCAmelCase__			:Optional[Any]											=							torch.tensor([1_0]      ).to(__UpperCAmelCase      )

         return {"sample": noise, "timestep": time_step}






       @property
       def      snake_case						(    self      ):



         '''simple docstring'''

         return (4, 3_2, 3_2)






       @property
       def      snake_case						(    self      ):



         '''simple docstring'''

         return (4, 3_2, 3_2)






       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:List[str]											=							{
             'sample_size': 3_2,
             'in_channels': 4,
             'out_channels': 4,
             'layers_per_block': 2,
             'block_out_channels': (3_2, 6_4),
             'attention_head_dim': 3_2,
             'down_block_types': ('DownBlock2D', 'DownBlock2D'),
             'up_block_types': ('UpBlock2D', 'UpBlock2D'),
         }
         lowerCAmelCase__			:Optional[int]											=							self.dummy_input
         return init_dict, inputs_dict






       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__							,				lowerCAmelCase__			:str											=							UNetaDModel.from_pretrained('fusing/unet-ldm-dummy-update'    , output_loading_info=__UpperCAmelCase      )

         self.assertIsNotNone(__UpperCAmelCase      )
         self.assertEqual(len(loading_info['missing_keys']      )    , 0      )

         model.to(__UpperCAmelCase      )
         lowerCAmelCase__			:Union[str, Any]											=							model(**self.dummy_input      ).sample

         assert image is not None, "Make sure output is not None"






       @unittest.skipIf(torch_device != 'cuda'    , 'This test is supposed to run on GPU'      )
       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__							,				lowerCAmelCase__			:Optional[int]											=							UNetaDModel.from_pretrained('fusing/unet-ldm-dummy-update'    , output_loading_info=__UpperCAmelCase      )
         model.to(__UpperCAmelCase      )
         lowerCAmelCase__			:Optional[int]											=							model(**self.dummy_input      ).sample

         assert image is not None, "Make sure output is not None"






       @unittest.skipIf(torch_device != 'cuda'    , 'This test is supposed to run on GPU'      )
       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__							,				lowerCAmelCase__			:str											=							UNetaDModel.from_pretrained('fusing/unet-ldm-dummy-update'    , output_loading_info=__UpperCAmelCase      )
         model_accelerate.to(__UpperCAmelCase      )
         model_accelerate.eval()

         lowerCAmelCase__			:List[str]											=							torch.randn(
             1    , model_accelerate.config.in_channels    , model_accelerate.config.sample_size    , model_accelerate.config.sample_size    , generator=torch.manual_seed(0      )    , )
         lowerCAmelCase__			:Dict											=							noise.to(__UpperCAmelCase      )
         lowerCAmelCase__			:Tuple											=							torch.tensor([1_0] * noise.shape[0]      ).to(__UpperCAmelCase      )

         lowerCAmelCase__			:Any											=							model_accelerate(__UpperCAmelCase    , __UpperCAmelCase      )['sample']

         # two models don't need to stay in the device at the same time
         del model_accelerate
         torch.cuda.empty_cache()
         gc.collect()

         lowerCAmelCase__							,				lowerCAmelCase__			:List[str]											=							UNetaDModel.from_pretrained(
             'fusing/unet-ldm-dummy-update'    , output_loading_info=__UpperCAmelCase    , low_cpu_mem_usage=__UpperCAmelCase      )
         model_normal_load.to(__UpperCAmelCase      )
         model_normal_load.eval()
         lowerCAmelCase__			:List[Any]											=							model_normal_load(__UpperCAmelCase    , __UpperCAmelCase      )['sample']

         assert torch_all_close(__UpperCAmelCase    , __UpperCAmelCase    , rtol=1E-3      )






       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:Optional[Any]											=							UNetaDModel.from_pretrained('fusing/unet-ldm-dummy-update'      )
         model.eval()
         model.to(__UpperCAmelCase      )

         lowerCAmelCase__			:List[str]											=							torch.randn(
             1    , model.config.in_channels    , model.config.sample_size    , model.config.sample_size    , generator=torch.manual_seed(0      )    , )
         lowerCAmelCase__			:str											=							noise.to(__UpperCAmelCase      )
         lowerCAmelCase__			:Any											=							torch.tensor([1_0] * noise.shape[0]      ).to(__UpperCAmelCase      )

         with torch.no_grad():
           lowerCAmelCase__			:Union[str, Any]											=							model(__UpperCAmelCase    , __UpperCAmelCase      ).sample

         lowerCAmelCase__			:str											=							output[0, -1, -3:, -3:].flatten().cpu()
         # fmt: off
         lowerCAmelCase__			:Union[str, Any]											=							torch.tensor([-13.32_58, -20.11_00, -15.98_73, -17.66_17, -23.05_96, -17.94_19, -13.36_75, -16.18_89, -12.38_00]      )
         # fmt: on

         self.assertTrue(torch_all_close(__UpperCAmelCase    , __UpperCAmelCase    , rtol=1E-3      )      )






class 					_lowerCAmelCase   (     a  ,   a  ,   unittest.TestCase       ):
       """simple docstring"""




       __magic_name__     :Dict		  =					UNetaDModel
       __magic_name__     :Optional[Any]		  =					"""sample"""






       @property
       def      snake_case						(    self    , __UpperCAmelCase=(3_2, 3_2)      ):



         '''simple docstring'''

         lowerCAmelCase__			:List[Any]											=							4
         lowerCAmelCase__			:Union[str, Any]											=							3

         lowerCAmelCase__			:List[Any]											=							floats_tensor((batch_size, num_channels) + sizes      ).to(__UpperCAmelCase      )
         lowerCAmelCase__			:Tuple											=							torch.tensor(batch_size * [1_0]      ).to(dtype=torch.intaa    , device=__UpperCAmelCase      )

         return {"sample": noise, "timestep": time_step}






       @property
       def      snake_case						(    self      ):



         '''simple docstring'''

         return (3, 3_2, 3_2)






       @property
       def      snake_case						(    self      ):



         '''simple docstring'''

         return (3, 3_2, 3_2)






       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:Optional[Any]											=							{
             'block_out_channels': [3_2, 6_4, 6_4, 6_4],
             'in_channels': 3,
             'layers_per_block': 1,
             'out_channels': 3,
             'time_embedding_type': 'fourier',
             'norm_eps': 1E-6,
             'mid_block_scale_factor': math.sqrt(2.0      ),
             'norm_num_groups': None,
             'down_block_types': [
                 'SkipDownBlock2D',
                 'AttnSkipDownBlock2D',
                 'SkipDownBlock2D',
                 'SkipDownBlock2D',
             ],
             'up_block_types': [
                 'SkipUpBlock2D',
                 'SkipUpBlock2D',
                 'AttnSkipUpBlock2D',
                 'SkipUpBlock2D',
             ],
         }
         lowerCAmelCase__			:Optional[Any]											=							self.dummy_input
         return init_dict, inputs_dict






       @slow
       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__							,				lowerCAmelCase__			:Optional[int]											=							UNetaDModel.from_pretrained('google/ncsnpp-celebahq-256'    , output_loading_info=__UpperCAmelCase      )
         self.assertIsNotNone(__UpperCAmelCase      )
         self.assertEqual(len(loading_info['missing_keys']      )    , 0      )

         model.to(__UpperCAmelCase      )
         lowerCAmelCase__			:Optional[int]											=							self.dummy_input
         lowerCAmelCase__			:List[Any]											=							floats_tensor((4, 3) + (2_5_6, 2_5_6)      ).to(__UpperCAmelCase      )
         lowerCAmelCase__			:List[str]											=							noise
         lowerCAmelCase__			:int											=							model(**__UpperCAmelCase      )

         assert image is not None, "Make sure output is not None"






       @slow
       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:Dict											=							UNetaDModel.from_pretrained('google/ncsnpp-celebahq-256'      )
         model.to(__UpperCAmelCase      )

         lowerCAmelCase__			:Union[str, Any]											=							4
         lowerCAmelCase__			:List[Any]											=							3
         lowerCAmelCase__			:List[str]											=							(2_5_6, 2_5_6)

         lowerCAmelCase__			:Union[str, Any]											=							torch.ones((batch_size, num_channels) + sizes      ).to(__UpperCAmelCase      )
         lowerCAmelCase__			:Dict											=							torch.tensor(batch_size * [1E-4]      ).to(__UpperCAmelCase      )

         with torch.no_grad():
           lowerCAmelCase__			:Union[str, Any]											=							model(__UpperCAmelCase    , __UpperCAmelCase      ).sample

         lowerCAmelCase__			:Union[str, Any]											=							output[0, -3:, -3:, -1].flatten().cpu()
         # fmt: off
         lowerCAmelCase__			:List[str]											=							torch.tensor([-48_42.86_91, -64_99.66_31, -38_00.19_53, -79_78.26_86, -1_09_80.71_29, -2_00_28.85_35, 81_48.28_22, 23_42.29_05, 5_67.76_08]      )
         # fmt: on

         self.assertTrue(torch_all_close(__UpperCAmelCase    , __UpperCAmelCase    , rtol=1E-2      )      )






       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:Any											=							UNetaDModel.from_pretrained('fusing/ncsnpp-ffhq-ve-dummy-update'      )
         model.to(__UpperCAmelCase      )

         lowerCAmelCase__			:Union[str, Any]											=							4
         lowerCAmelCase__			:List[Any]											=							3
         lowerCAmelCase__			:Tuple											=							(3_2, 3_2)

         lowerCAmelCase__			:Tuple											=							torch.ones((batch_size, num_channels) + sizes      ).to(__UpperCAmelCase      )
         lowerCAmelCase__			:List[Any]											=							torch.tensor(batch_size * [1E-4]      ).to(__UpperCAmelCase      )

         with torch.no_grad():
           lowerCAmelCase__			:str											=							model(__UpperCAmelCase    , __UpperCAmelCase      ).sample

         lowerCAmelCase__			:Optional[Any]											=							output[0, -3:, -3:, -1].flatten().cpu()
         # fmt: off
         lowerCAmelCase__			:Optional[int]											=							torch.tensor([-0.03_25, -0.09_00, -0.08_69, -0.03_32, -0.07_25, -0.02_70, -0.01_01, 0.02_27, 0.02_56]      )
         # fmt: on

         self.assertTrue(torch_all_close(__UpperCAmelCase    , __UpperCAmelCase    , rtol=1E-2      )      )






       def      snake_case						(    self      ):



         '''simple docstring'''

         pass







"""simple docstring"""





from multiprocessing import Lock, Pipe, Process

# lock used to ensure that two processes do not access a pipe at the same time
__A											=		Lock()
def  __A   (_SCREAMING_SNAKE_CASE     ,   _SCREAMING_SNAKE_CASE     ,   _SCREAMING_SNAKE_CASE     ,   _SCREAMING_SNAKE_CASE     ,   _SCREAMING_SNAKE_CASE     ,   _SCREAMING_SNAKE_CASE     ,   _SCREAMING_SNAKE_CASE			)					->List[Any]:





  """simple docstring"""


  global process_lock

  # we perform n swaps since after n swaps we know we are sorted
  # we *could* stop early if we are sorted already, but it takes as long to
  # find out we are sorted as it does to sort the list with this algorithm
  for i in range(0     ,   10			):
    if (i + position) % 2 == 0 and r_send is not None:
      # send your value to your right neighbor
      process_lock.acquire()
      r_send[1].send(_SCREAMING_SNAKE_CASE			)
      process_lock.release()

      # receive your right neighbor's value
      process_lock.acquire()
      lowerCAmelCase__			:Any											=							rr_cv[0].recv()
      process_lock.release()

      # take the lower value since you are on the left
      lowerCAmelCase__			:Tuple											=							min(_SCREAMING_SNAKE_CASE     ,   _SCREAMING_SNAKE_CASE			)
    elif (i + position) % 2 != 0 and l_send is not None:
      # send your value to your left neighbor
      process_lock.acquire()
      l_send[1].send(_SCREAMING_SNAKE_CASE			)
      process_lock.release()

      # receive your left neighbor's value
      process_lock.acquire()
      lowerCAmelCase__			:Optional[int]											=							lr_cv[0].recv()
      process_lock.release()

      # take the higher value since you are on the right
      lowerCAmelCase__			:Optional[int]											=							max(_SCREAMING_SNAKE_CASE     ,   _SCREAMING_SNAKE_CASE			)
    # after all swaps are performed, send the values back to main
  result_pipe[1].send(_SCREAMING_SNAKE_CASE			)
def  __A   (_SCREAMING_SNAKE_CASE			)					->Optional[int]:





  """simple docstring"""


  lowerCAmelCase__			:str											=							[]
  lowerCAmelCase__			:Optional[Any]											=							[]
  # initialize the list of pipes where the values will be retrieved
  for _ in arr:
    result_pipe.append(Pipe()			)
  # creates the processes
  # the first and last process only have one neighbor so they are made outside
  # of the loop
  lowerCAmelCase__			:List[str]											=							Pipe()
  lowerCAmelCase__			:List[Any]											=							Pipe()
  process_array_.append(
      Process(
          target=_SCREAMING_SNAKE_CASE     ,   args=(0, arr[0], None, temp_rs, None, temp_rr, result_pipe[0])     ,   )			)
  lowerCAmelCase__			:Dict											=							temp_rs
  lowerCAmelCase__			:Optional[Any]											=							temp_rr

  for i in range(1     ,   len(_SCREAMING_SNAKE_CASE			) - 1			):
    lowerCAmelCase__			:Union[str, Any]											=							Pipe()
    lowerCAmelCase__			:List[str]											=							Pipe()
    process_array_.append(
        Process(
            target=_SCREAMING_SNAKE_CASE     ,   args=(i, arr[i], temp_ls, temp_rs, temp_lr, temp_rr, result_pipe[i])     ,   )			)
    lowerCAmelCase__			:Union[str, Any]											=							temp_rs
    lowerCAmelCase__			:Any											=							temp_rr

  process_array_.append(
      Process(
          target=_SCREAMING_SNAKE_CASE     ,   args=(
              len(_SCREAMING_SNAKE_CASE			) - 1,
              arr[len(_SCREAMING_SNAKE_CASE			) - 1],
              temp_ls,
              None,
              temp_lr,
              None,
              result_pipe[len(_SCREAMING_SNAKE_CASE			) - 1],
          )     ,   )			)

  # start the processes
  for p in process_array_:
    p.start()

  # wait for the processes to end and write their values to the list
  for p in range(0     ,   len(_SCREAMING_SNAKE_CASE			)			):
    lowerCAmelCase__			:str											=							result_pipe[p][0].recv()
    process_array_[p].join()
  return arr
def  __A   ()					->List[Any]:





  """simple docstring"""


  lowerCAmelCase__			:Union[str, Any]											=							list(range(10     ,   0     ,   -1			)			)
  print('Initial List'			)
  print(*_SCREAMING_SNAKE_CASE			)
  lowerCAmelCase__			:List[str]											=							odd_even_transposition(_SCREAMING_SNAKE_CASE			)
  print('Sorted List\n'			)
  print(*_SCREAMING_SNAKE_CASE			)


if __name__ == "__main__":
  main()
