



'''simple docstring'''

import time
from dataclasses import dataclass
from multiprocessing import Pool
from unittest import TestCase
from unittest.mock import patch

import multiprocess
import numpy as np
import pytest

from datasets.utils.py_utils import (
    NestedDataStructure,
    asdict,
    iflatmap_unordered,
    map_nested,
    temp_seed,
    temporary_assignment,
    zip_dict,
)

from .utils import require_tf, require_torch



def 	__lowerCamelCase	( lowerCAmelCase_    )     -> Any:  # picklable for multiprocessing
		return x.sum()



def 	__lowerCamelCase	( lowerCAmelCase_    )     -> Union[str, Any]:  # picklable for multiprocessing
		return i + 1

@dataclass
class       __magic_name__		:
							lowerCAmelCase			:       int
							lowerCAmelCase			:       str

class       __magic_name__		(       _UpperCamelCase  ):







							def 				__lowercase     (				self   :			Dict						):
									_a	: List[str]								=				{}
									_a	: Optional[int]								=				[]
									_a	: Dict								=				1
									_a	: Union[str, Any]								=				[1, 2]
									_a	: str								=				{'a': 1, 'b': 2}
									_a	: List[str]								=				{'a': [1, 2], 'b': [3, 4]}
									_a	: Any								=				{'a': {'1': 1}, 'b': 2}
									_a	: Dict								=				{'a': 1, 'b': 2, 'c': 3, 'd': 4}
									_a	: Tuple								=				{}
									_a	: Optional[int]								=				[]
									_a	: int								=				2
									_a	: int								=				[2, 3]
									_a	: Any								=				{'a': 2, 'b': 3}
									_a	: int								=				{'a': [2, 3], 'b': [4, 5]}
									_a	: int								=				{'a': {'1': 2}, 'b': 3}
									_a	: Optional[int]								=				{'a': 2, 'b': 3, 'c': 4, 'd': 5}
									self.assertEqual(map_nested(_UpperCAmelCase      ,_UpperCAmelCase						)      ,_UpperCAmelCase						)
									self.assertEqual(map_nested(_UpperCAmelCase      ,_UpperCAmelCase						)      ,_UpperCAmelCase						)
									self.assertEqual(map_nested(_UpperCAmelCase      ,_UpperCAmelCase						)      ,_UpperCAmelCase						)
									self.assertEqual(map_nested(_UpperCAmelCase      ,_UpperCAmelCase						)      ,_UpperCAmelCase						)
									self.assertEqual(map_nested(_UpperCAmelCase      ,_UpperCAmelCase						)      ,_UpperCAmelCase						)
									self.assertEqual(map_nested(_UpperCAmelCase      ,_UpperCAmelCase						)      ,_UpperCAmelCase						)
									self.assertEqual(map_nested(_UpperCAmelCase      ,_UpperCAmelCase						)      ,_UpperCAmelCase						)
									self.assertEqual(map_nested(_UpperCAmelCase      ,_UpperCAmelCase						)      ,_UpperCAmelCase						)

									_a	: int								=				2
									self.assertEqual(map_nested(_UpperCAmelCase      ,_UpperCAmelCase      ,num_proc=_UpperCAmelCase						)      ,_UpperCAmelCase						)
									self.assertEqual(map_nested(_UpperCAmelCase      ,_UpperCAmelCase      ,num_proc=_UpperCAmelCase						)      ,_UpperCAmelCase						)
									self.assertEqual(map_nested(_UpperCAmelCase      ,_UpperCAmelCase      ,num_proc=_UpperCAmelCase						)      ,_UpperCAmelCase						)
									self.assertEqual(map_nested(_UpperCAmelCase      ,_UpperCAmelCase      ,num_proc=_UpperCAmelCase						)      ,_UpperCAmelCase						)
									self.assertEqual(map_nested(_UpperCAmelCase      ,_UpperCAmelCase      ,num_proc=_UpperCAmelCase						)      ,_UpperCAmelCase						)
									self.assertEqual(map_nested(_UpperCAmelCase      ,_UpperCAmelCase      ,num_proc=_UpperCAmelCase						)      ,_UpperCAmelCase						)
									self.assertEqual(map_nested(_UpperCAmelCase      ,_UpperCAmelCase      ,num_proc=_UpperCAmelCase						)      ,_UpperCAmelCase						)
									self.assertEqual(map_nested(_UpperCAmelCase      ,_UpperCAmelCase      ,num_proc=_UpperCAmelCase						)      ,_UpperCAmelCase						)

									_a	: Any								=				{'a': np.eye(2						), 'b': np.zeros(3						), 'c': np.ones(2						)}
									_a	: Dict								=				{'a': 2, 'b': 0, 'c': 2}
									_a	: int								=				{
									    'a': np.eye(2						).astype(_UpperCAmelCase						),
									    'b': np.zeros(3						).astype(_UpperCAmelCase						),
									    'c': np.ones(2						).astype(_UpperCAmelCase						),
									}
									self.assertEqual(map_nested(_UpperCAmelCase      ,_UpperCAmelCase      ,map_numpy=_UpperCAmelCase						)      ,_UpperCAmelCase						)
									self.assertEqual(
									    {k: v.tolist() for k, v in map_nested(_UpperCAmelCase      ,_UpperCAmelCase      ,map_numpy=_UpperCAmelCase						).items()}      ,{k: v.tolist() for k, v in expected_map_nested_sna_int.items()}      ,)
									self.assertEqual(map_nested(_UpperCAmelCase      ,_UpperCAmelCase      ,map_numpy=_UpperCAmelCase      ,num_proc=_UpperCAmelCase						)      ,_UpperCAmelCase						)
									self.assertEqual(
									    {k: v.tolist() for k, v in map_nested(_UpperCAmelCase      ,_UpperCAmelCase      ,map_numpy=_UpperCAmelCase      ,num_proc=_UpperCAmelCase						).items()}      ,{k: v.tolist() for k, v in expected_map_nested_sna_int.items()}      ,)
									with self.assertRaises(_UpperCAmelCase						):  # can't pickle a local lambda
											map_nested(lambda _UpperCAmelCase						: x + 1      ,_UpperCAmelCase      ,num_proc=_UpperCAmelCase						)







							def 				__lowercase     (				self   :			Dict						):
									_a	: Tuple								=				{'a': 1, 'b': 2}
									_a	: Dict								=				{'a': 3, 'b': 4}
									_a	: List[Any]								=				{'a': 5, 'b': 6}
									_a	: List[Any]								=				sorted([('a', (1, 3, 5)), ('b', (2, 4, 6))]						)
									self.assertEqual(sorted(zip_dict(_UpperCAmelCase      ,_UpperCAmelCase      ,_UpperCAmelCase						)						)      ,_UpperCAmelCase						)







							def 				__lowercase     (				self   :			str						):

									class       __magic_name__		:
																lowerCAmelCase			:       Tuple  							=   'bar'
									_a	: Tuple								=				Foo()
									self.assertEqual(foo.my_attr      ,'bar'						)
									with temporary_assignment(_UpperCAmelCase      ,'my_attr'      ,'BAR'						):
											self.assertEqual(foo.my_attr      ,'BAR'						)
									self.assertEqual(foo.my_attr      ,'bar'						)



@pytest.mark.parametrize(
    'iterable_length, num_proc, expected_num_proc'		,     [
        (1, None, 1),
        (1, 1, 1),
        (2, None, 1),
        (2, 1, 1),
        (2, 2, 1),
        (2, 3, 1),
        (3, 2, 1),
        (16, 16, 16),
        (16, 17, 16),
        (17, 16, 16),
    ]		,     )
def 	__lowerCamelCase	( lowerCAmelCase_		,     lowerCAmelCase_		,     lowerCAmelCase_    )     -> Union[str, Any]:
		with patch('datasets.utils.py_utils._single_map_nested'    ) as mock_single_map_nested, patch(
		    'datasets.parallel.parallel.Pool'    ) as mock_multiprocessing_pool:
				_a	: Union[str, Any]								=				{f"""{i}""": i for i in range(lowerCAmelCase_    )}
				_a	: Dict								=				map_nested(lambda lowerCAmelCase_    : x + 10		,     lowerCAmelCase_		,     num_proc=lowerCAmelCase_		,     parallel_min_length=16    )
				if expected_num_proc == 1:
						assert mock_single_map_nested.called
						assert not mock_multiprocessing_pool.called
				else:
						assert not mock_single_map_nested.called
						assert mock_multiprocessing_pool.called
						assert mock_multiprocessing_pool.call_args[0][0] == expected_num_proc

class       __magic_name__		(       _UpperCamelCase  ):







							@require_tf
							def 				__lowercase     (				self   :			Dict						):
									import tensorflow as tf
									from tensorflow.keras import layers

									_a	: Union[str, Any]								=				layers.Dense(2						)

									def gen_random_output():
											_a	: Any								=				tf.random.uniform((1, 3)						)
											return model(_UpperCAmelCase						).numpy()

									with temp_seed(42      ,set_tensorflow=_UpperCAmelCase						):
											_a	: Union[str, Any]								=				gen_random_output()
									with temp_seed(42      ,set_tensorflow=_UpperCAmelCase						):
											_a	: Optional[Any]								=				gen_random_output()
									_a	: Union[str, Any]								=				gen_random_output()

									np.testing.assert_equal(_UpperCAmelCase      ,_UpperCAmelCase						)
									self.assertGreater(np.abs(outa - outa						).sum()      ,0						)







							@require_torch
							def 				__lowercase     (				self   :			List[Any]						):
									import torch

									def gen_random_output():
											_a	: Optional[int]								=				torch.nn.Linear(3      ,2						)
											_a	: Dict								=				torch.rand(1      ,3						)
											return model(_UpperCAmelCase						).detach().numpy()

									with temp_seed(42      ,set_pytorch=_UpperCAmelCase						):
											_a	: Dict								=				gen_random_output()
									with temp_seed(42      ,set_pytorch=_UpperCAmelCase						):
											_a	: Dict								=				gen_random_output()
									_a	: Any								=				gen_random_output()

									np.testing.assert_equal(_UpperCAmelCase      ,_UpperCAmelCase						)
									self.assertGreater(np.abs(outa - outa						).sum()      ,0						)







							def 				__lowercase     (				self   :			Optional[Any]						):
									def gen_random_output():
											return np.random.rand(1      ,3						)

									with temp_seed(42						):
											_a	: Optional[Any]								=				gen_random_output()
									with temp_seed(42						):
											_a	: Dict								=				gen_random_output()
									_a	: Tuple								=				gen_random_output()

									np.testing.assert_equal(_UpperCAmelCase      ,_UpperCAmelCase						)
									self.assertGreater(np.abs(outa - outa						).sum()      ,0						)



@pytest.mark.parametrize('input_data'		,     [{}]    )
def 	__lowerCamelCase	( lowerCAmelCase_    )     -> Any:
		_a	: int								=				NestedDataStructure(lowerCAmelCase_    ).data
		assert output_data == input_data



@pytest.mark.parametrize(
    'data, expected_output'		,     [
        ({}, []),
        ([], []),
        ('foo', ['foo']),
        (['foo', 'bar'], ['foo', 'bar']),
        ([['foo', 'bar']], ['foo', 'bar']),
        ([[['foo'], ['bar']]], ['foo', 'bar']),
        ([[['foo'], 'bar']], ['foo', 'bar']),
        ({'a': 1, 'b': 2}, [1, 2]),
        ({'a': [1, 2], 'b': [3, 4]}, [1, 2, 3, 4]),
        ({'a': [[1, 2]], 'b': [[3, 4]]}, [1, 2, 3, 4]),
        ({'a': [[1, 2]], 'b': [3, 4]}, [1, 2, 3, 4]),
        ({'a': [[[1], [2]]], 'b': [[[3], [4]]]}, [1, 2, 3, 4]),
        ({'a': [[[1], [2]]], 'b': [[3, 4]]}, [1, 2, 3, 4]),
        ({'a': [[[1], [2]]], 'b': [3, 4]}, [1, 2, 3, 4]),
        ({'a': [[[1], [2]]], 'b': [3, [4]]}, [1, 2, 3, 4]),
        ({'a': {'1': 1}, 'b': 2}, [1, 2]),
        ({'a': {'1': [1]}, 'b': 2}, [1, 2]),
        ({'a': {'1': [1]}, 'b': [2]}, [1, 2]),
    ]		,     )
def 	__lowerCamelCase	( lowerCAmelCase_		,     lowerCAmelCase_    )     -> Dict:
		_a	: int								=				NestedDataStructure(lowerCAmelCase_    ).flatten()
		assert output == expected_output



def 	__lowerCamelCase	( )     -> Union[str, Any]:
		_a	: Optional[int]								=				A(x=1		,     y='foobar'    )
		_a	: int								=				{'x': 1, 'y': 'foobar'}
		assert asdict(lowerCAmelCase_    ) == expected_output

		_a	: Union[str, Any]								=				{'a': {'b': A(x=10		,     y='foo'    )}, 'c': [A(x=20		,     y='bar'    )]}
		_a	: List[Any]								=				{'a': {'b': {'x': 10, 'y': 'foo'}}, 'c': [{'x': 20, 'y': 'bar'}]}
		assert asdict(lowerCAmelCase_    ) == expected_output

		with pytest.raises(lowerCAmelCase_    ):
				asdict([1, A(x=10		,     y='foo'    )]    )



def 	__lowerCamelCase	( lowerCAmelCase_    )     -> str:
		return text.split()



def 	__lowerCamelCase	( lowerCAmelCase_    )     -> List[str]:
		yield (time.time(), content)
		time.sleep(2    )
		yield (time.time(), content)



def 	__lowerCamelCase	( )     -> List[str]:
		with Pool(2    ) as pool:
				_a	: Optional[int]								=				list(iflatmap_unordered(lowerCAmelCase_		,     _split_text		,     kwargs_iterable=[{'text': 'hello there'}] * 10    )    )
				assert out.count('hello'    ) == 10
				assert out.count('there'    ) == 10
				assert len(lowerCAmelCase_    ) == 20

		# check multiprocess from pathos (uses dill for pickling)
		with multiprocess.Pool(2    ) as pool:
				_a	: Any								=				list(iflatmap_unordered(lowerCAmelCase_		,     _split_text		,     kwargs_iterable=[{'text': 'hello there'}] * 10    )    )
				assert out.count('hello'    ) == 10
				assert out.count('there'    ) == 10
				assert len(lowerCAmelCase_    ) == 20

		# check that we get items as fast as possible
		with Pool(2    ) as pool:
				_a	: List[Any]								=				[]
				for yield_time, content in iflatmap_unordered(
				    lowerCAmelCase_		,     _aseconds_generator_of_aitems_with_timing		,     kwargs_iterable=[{'content': 'a'}, {'content': 'b'}]    ):
						assert yield_time < time.time() + 0.1, "we should each item directly after it was yielded"
						out.append(lowerCAmelCase_    )
				assert out.count('a'    ) == 2
				assert out.count('b'    ) == 2
				assert len(lowerCAmelCase_    ) == 4









'''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




