
import contextlib
from multiprocessing import Pool, RLock

from tqdm.auto import tqdm

from ..utils import experimental, logging


lowerCamelCase__          =		logging.get_logger(__name__)







class      A__							:
				lowercase			    =  None






@experimental
def  lowerCAmelCase__			(  SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_     )				->		Union[str, Any]:
			if ParallelBackendConfig.backend_name is None:
						return _map_with_multiprocessing_pool(
						    SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_     )

			return _map_with_joblib(SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_     )
def  lowerCAmelCase__			(  SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_     )				->		List[str]:
			lowerCAmelCase__      :      Union[str, Any]         =		num_proc if num_proc <= len(SCREAMING_SNAKE_CASE_     ) else len(SCREAMING_SNAKE_CASE_     )
			lowerCAmelCase__      :      List[str]         =		[]  # We organize the splits ourselve (contiguous splits)
			for index in range(SCREAMING_SNAKE_CASE_     ):
						lowerCAmelCase__      :      List[str]         =		len(SCREAMING_SNAKE_CASE_     ) // num_proc
						lowerCAmelCase__      :      Union[str, Any]         =		len(SCREAMING_SNAKE_CASE_     ) % num_proc
						lowerCAmelCase__      :      int         =		div * index + min(SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_     )
						lowerCAmelCase__      :      Dict         =		start + div + (1 if index < mod else 0)
						split_kwds.append((function, iterable[start:end], types, index, disable_tqdm, desc)     )

			if len(SCREAMING_SNAKE_CASE_     ) != sum(len(i[1]     ) for i in split_kwds     ):
						raise ValueError(
						    F'''Error dividing inputs iterable among processes. '''
						    F'''Total number of objects {len(SCREAMING_SNAKE_CASE_     )}, '''
						    F'''length: {sum(len(i[1]     ) for i in split_kwds     )}'''     )

			logger.info(
			    F'''Spawning {num_proc} processes for {len(SCREAMING_SNAKE_CASE_     )} objects in slices of {[len(i[1]     ) for i in split_kwds]}'''     )
			lowerCAmelCase__      ,				lowerCAmelCase__      :      Tuple         =		None, None
			if not disable_tqdm:
						lowerCAmelCase__      ,				lowerCAmelCase__      :      Tuple         =		(RLock(),), tqdm.set_lock
			with Pool(SCREAMING_SNAKE_CASE_       ,							initargs=SCREAMING_SNAKE_CASE_       ,							initializer=SCREAMING_SNAKE_CASE_     ) as pool:
						lowerCAmelCase__      :      List[str]         =		pool.map(SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_     )
			logger.info(F'''Finished {num_proc} processes'''     )
			lowerCAmelCase__      :      str         =		[obj for proc_res in mapped for obj in proc_res]
			logger.info(F'''Unpacked {len(SCREAMING_SNAKE_CASE_     )} objects'''     )

			return mapped
def  lowerCAmelCase__			(  SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_     )				->		Any:
			# progress bar is not yet supported for _map_with_joblib, because tqdm couldn't accurately be applied to joblib,
			# and it requires monkey-patching joblib internal classes which is subject to change
			import joblib

			with joblib.parallel_backend(ParallelBackendConfig.backend_name       ,							n_jobs=SCREAMING_SNAKE_CASE_     ):
						return joblib.Parallel()(
						    joblib.delayed(SCREAMING_SNAKE_CASE_     )((function, obj, types, None, True, None)     ) for obj in iterable     )







@experimental
@contextlib.contextmanager
def  lowerCAmelCase__			(  SCREAMING_SNAKE_CASE_     )				->		Any:
			lowerCAmelCase__      :      Optional[int]         =		backend_name

			if backend_name == "spark":
						from joblibspark import register_spark

						register_spark()

						# TODO: call create_cache_and_write_probe if "download" in steps
						# TODO: raise NotImplementedError when Dataset.map etc is called

			try:
						yield
			finally:
						lowerCAmelCase__      :      Tuple         =		None

import unittest
from typing import Tuple

import torch

from diffusers.utils import floats_tensor, randn_tensor, torch_all_close, torch_device
from diffusers.utils.testing_utils import require_torch







@require_torch
class      A__							:






				@property
				def 							_lowerCamelCase		(						self					:      Optional[int]   ):






							'''simple docstring'''







							return self.get_dummy_input()






				@property
				def 							_lowerCamelCase		(						self					:      Dict   ):






							'''simple docstring'''







							if self.block_type == "down":
										return (4, 32, 16, 16)
							elif self.block_type == "mid":
										return (4, 32, 32, 32)
							elif self.block_type == "up":
										return (4, 32, 64, 64)

							raise ValueError(f'''\'{self.block_type}\' is not a supported block_type. Set it to \'up\', \'mid\', or \'down\'.'''   )






				def 							_lowerCamelCase		(						self					:      Optional[int]    ,     a					:      List[Any]=True    ,     a					:      Any=False    ,     a					:      Dict=False    ,     a					:      Union[str, Any]=False    ,     ):






							'''simple docstring'''







							lowerCAmelCase__      :      Tuple         =		4
							lowerCAmelCase__      :      int         =		32
							lowerCAmelCase__      :      Tuple         =		(32, 32)

							lowerCAmelCase__      :      List[Any]         =		torch.manual_seed(0   )
							lowerCAmelCase__      :      List[Any]         =		torch.device(a   )
							lowerCAmelCase__      :      str         =		(batch_size, num_channels) + sizes
							lowerCAmelCase__      :      Tuple         =		randn_tensor(a    ,     generator=a    ,     device=a   )
							lowerCAmelCase__      :      Optional[Any]         =		{'hidden_states': hidden_states}

							if include_temb:
										lowerCAmelCase__      :      int         =		128
										lowerCAmelCase__      :      List[str]         =		randn_tensor((batch_size, temb_channels)    ,     generator=a    ,     device=a   )

							if include_res_hidden_states_tuple:
										lowerCAmelCase__      :      int         =		torch.manual_seed(1   )
										lowerCAmelCase__      :      str         =		(randn_tensor(a    ,     generator=a    ,     device=a   ),)

							if include_encoder_hidden_states:
										lowerCAmelCase__      :      Any         =		floats_tensor((batch_size, 32, 32)   ).to(a   )

							if include_skip_sample:
										lowerCAmelCase__      :      Union[str, Any]         =		randn_tensor(((batch_size, 3) + sizes)    ,     generator=a    ,     device=a   )

							return dummy_input






				def 							_lowerCamelCase		(						self					:      Tuple   ):






							'''simple docstring'''







							lowerCAmelCase__      :      str         =		{
							    'in_channels': 32,
							    'out_channels': 32,
							    'temb_channels': 128,
							}
							if self.block_type == "up":
										lowerCAmelCase__      :      Union[str, Any]         =		32

							if self.block_type == "mid":
										init_dict.pop('out_channels'   )

							lowerCAmelCase__      :      Optional[int]         =		self.dummy_input
							return init_dict, inputs_dict






				def 							_lowerCamelCase		(						self					:      str    ,     a					:      List[str]   ):






							'''simple docstring'''







							lowerCAmelCase__      ,				lowerCAmelCase__      :      Union[str, Any]         =		self.prepare_init_args_and_inputs_for_common()
							lowerCAmelCase__      :      int         =		self.block_class(**a   )
							unet_block.to(a   )
							unet_block.eval()

							with torch.no_grad():
										lowerCAmelCase__      :      int         =		unet_block(**a   )

							if isinstance(a    ,     a   ):
										lowerCAmelCase__      :      List[str]         =		output[0]

							self.assertEqual(output.shape    ,     self.output_shape   )

							lowerCAmelCase__      :      List[str]         =		output[0, -1, -3:, -3:]
							lowerCAmelCase__      :      Any         =		torch.tensor(a   ).to(a   )
							assert torch_all_close(output_slice.flatten()    ,     a    ,     atol=5E-3   )






				@unittest.skipIf(torch_device == 'mps'    ,     'Training is not supported in mps'   )
				def 							_lowerCamelCase		(						self					:      List[str]   ):






							'''simple docstring'''







							lowerCAmelCase__      ,				lowerCAmelCase__      :      Union[str, Any]         =		self.prepare_init_args_and_inputs_for_common()
							lowerCAmelCase__      :      Any         =		self.block_class(**a   )
							model.to(a   )
							model.train()
							lowerCAmelCase__      :      int         =		model(**a   )

							if isinstance(a    ,     a   ):
										lowerCAmelCase__      :      Dict         =		output[0]

							lowerCAmelCase__      :      Optional[int]         =		torch.device(a   )
							lowerCAmelCase__      :      List[Any]         =		randn_tensor(output.shape    ,     device=a   )
							lowerCAmelCase__      :      List[Any]         =		torch.nn.functional.mse_loss(a    ,     a   )
							loss.backward()