


from unittest.mock import patch

import pyspark

from datasets.packaged_modules.spark.spark import (
    Spark,
    SparkExamplesIterable,
    _generate_iterable_examples,
)

from ..utils import (
    require_dill_gt_0_3_2,
    require_not_windows,
)






def 			UpperCamelCase (     __magic_name__					:							int   ,   __magic_name__					:							Union[str, Any]					)       ->  Dict:
		"""simple docstring"""




		lowercase__					      =		[]
		for part_id in partition_order:
				lowercase__					      =		df.where(f'''SPARK_PARTITION_ID() = {part_id}'''					).collect()
				for row_idx, row in enumerate(__magic_name__					):
						expected_row_ids_and_row_dicts.append((f'''{part_id}_{row_idx}''', row.asDict())					)
		return expected_row_ids_and_row_dicts






@require_not_windows
@require_dill_gt_0_3_2
def 			UpperCamelCase (     )       ->  Optional[int]:
		"""simple docstring"""




		lowercase__					      =		pyspark.sql.SparkSession.builder.master("""local[*]"""					).appName("""pyspark"""					).getOrCreate()
		lowercase__					      =		spark.range(100					).repartition(1					)
		lowercase__					      =		Spark(__magic_name__					)
		# The id ints will be converted to Pyarrow int64s, so each row will be 8 bytes. Setting a max_shard_size of 16 means
		# that each partition can hold 2 rows.
		spark_builder._repartition_df_if_needed(max_shard_size=16					)
		# Given that the dataframe has 100 rows and each partition has 2 rows, we expect 50 partitions.
		assert spark_builder.df.rdd.getNumPartitions() == 50






@require_not_windows
@require_dill_gt_0_3_2
def 			UpperCamelCase (     )       ->  List[Any]:
		"""simple docstring"""




		lowercase__					      =		pyspark.sql.SparkSession.builder.master("""local[*]"""					).appName("""pyspark"""					).getOrCreate()
		lowercase__					      =		spark.range(10					).repartition(2					)
		lowercase__					      =		[1, 0]
		lowercase__					      =		_generate_iterable_examples(__magic_name__   ,   __magic_name__					)  # Reverse the partitions.
		lowercase__					      =		_get_expected_row_ids_and_row_dicts_for_partition_order(__magic_name__   ,   __magic_name__					)

		for i, (row_id, row_dict) in enumerate(generate_fn()					):
				lowercase__      ,     lowercase__					      =		expected_row_ids_and_row_dicts[i]
				assert row_id == expected_row_id
				assert row_dict == expected_row_dict






@require_not_windows
@require_dill_gt_0_3_2
def 			UpperCamelCase (     )       ->  Tuple:
		"""simple docstring"""




		lowercase__					      =		pyspark.sql.SparkSession.builder.master("""local[*]"""					).appName("""pyspark"""					).getOrCreate()
		lowercase__					      =		spark.range(10					).repartition(1					)
		lowercase__					      =		SparkExamplesIterable(__magic_name__					)
		assert it.n_shards == 1
		for i, (row_id, row_dict) in enumerate(__magic_name__					):
				assert row_id == f'''0_{i}'''
				assert row_dict == {"id": i}






@require_not_windows
@require_dill_gt_0_3_2
def 			UpperCamelCase (     )       ->  Dict:
		"""simple docstring"""




		lowercase__					      =		pyspark.sql.SparkSession.builder.master("""local[*]"""					).appName("""pyspark"""					).getOrCreate()
		lowercase__					      =		spark.range(30					).repartition(3					)
		# Mock the generator so that shuffle reverses the partition indices.
		with patch("""numpy.random.Generator"""					) as generator_mock:
				lowercase__					      =		lambda __magic_name__					: x.reverse()
				lowercase__					      =		_get_expected_row_ids_and_row_dicts_for_partition_order(__magic_name__   ,   [2, 1, 0]					)

				lowercase__					      =		SparkExamplesIterable(__magic_name__					).shuffle_data_sources(__magic_name__					)
				assert shuffled_it.n_shards == 3
				for i, (row_id, row_dict) in enumerate(__magic_name__					):
						lowercase__      ,     lowercase__					      =		expected_row_ids_and_row_dicts[i]
						assert row_id == expected_row_id
						assert row_dict == expected_row_dict






@require_not_windows
@require_dill_gt_0_3_2
def 			UpperCamelCase (     )       ->  Tuple:
		"""simple docstring"""




		lowercase__					      =		pyspark.sql.SparkSession.builder.master("""local[*]"""					).appName("""pyspark"""					).getOrCreate()
		lowercase__					      =		spark.range(20					).repartition(4					)

		# Partitions 0 and 2
		lowercase__					      =		SparkExamplesIterable(__magic_name__					).shard_data_sources(worker_id=0   ,   num_workers=2					)
		assert shard_it_a.n_shards == 2
		lowercase__					      =		_get_expected_row_ids_and_row_dicts_for_partition_order(__magic_name__   ,   [0, 2]					)
		for i, (row_id, row_dict) in enumerate(__magic_name__					):
				lowercase__      ,     lowercase__					      =		expected_row_ids_and_row_dicts_a[i]
				assert row_id == expected_row_id
				assert row_dict == expected_row_dict

		# Partitions 1 and 3
		lowercase__					      =		SparkExamplesIterable(__magic_name__					).shard_data_sources(worker_id=1   ,   num_workers=2					)
		assert shard_it_a.n_shards == 2
		lowercase__					      =		_get_expected_row_ids_and_row_dicts_for_partition_order(__magic_name__   ,   [1, 3]					)
		for i, (row_id, row_dict) in enumerate(__magic_name__					):
				lowercase__      ,     lowercase__					      =		expected_row_ids_and_row_dicts_a[i]
				assert row_id == expected_row_id
				assert row_dict == expected_row_dict







@require_not_windows
@require_dill_gt_0_3_2
def 			UpperCamelCase (     )       ->  List[Any]:
		"""simple docstring"""




		lowercase__					      =		pyspark.sql.SparkSession.builder.master("""local[*]"""					).appName("""pyspark"""					).getOrCreate()
		lowercase__					      =		spark.range(100					).repartition(1					)
		lowercase__					      =		Spark(__magic_name__					)
		# Choose a small max_shard_size for maximum partitioning.
		spark_builder._repartition_df_if_needed(max_shard_size=1					)
		# The new number of partitions should not be greater than the number of rows.
		assert spark_builder.df.rdd.getNumPartitions() == 100





import gc
import unittest

from parameterized import parameterized

from diffusers import FlaxUNetaDConditionModel
from diffusers.utils import is_flax_available
from diffusers.utils.testing_utils import load_hf_numpy, require_flax, slow


if is_flax_available():
	import jax
	import jax.numpy as jnp



@slow
@require_flax
class 							A   (	unittest.TestCase							):
					'''simple docstring'''






					def   lowerCamelCase__					(self   :						Any				, _UpperCAmelCase   :						List[str]				, _UpperCAmelCase   :						Union[str, Any]		)						->       int:




							"""simple docstring"""


							return f'''gaussian_noise_s={seed}_shape={"_".join([str(_UpperCAmelCase		) for s in shape]		)}.npy'''

					def   lowerCamelCase__					(self   :						Any		)						->       int:




							"""simple docstring"""


							super().tearDown()
							gc.collect()

					def   lowerCamelCase__					(self   :						List[Any]				, _UpperCAmelCase   :						Any=0				, _UpperCAmelCase   :						Optional[Any]=(4, 4, 64, 64)				, _UpperCAmelCase   :						Any=False		)						->       List[Any]:




							"""simple docstring"""


							lowercase__					      =		jnp.bfloataa if fpaa else jnp.floataa
							lowercase__					      =		jnp.array(load_hf_numpy(self.get_file_format(_UpperCAmelCase				, _UpperCAmelCase		)		)				, dtype=_UpperCAmelCase		)
							return image

					def   lowerCamelCase__					(self   :						Optional[Any]				, _UpperCAmelCase   :						str=False				, _UpperCAmelCase   :						str="CompVis/stable-diffusion-v1-4"		)						->       Dict:




							"""simple docstring"""


							lowercase__					      =		jnp.bfloataa if fpaa else jnp.floataa
							lowercase__					      =		"""bf16""" if fpaa else None

							lowercase__      ,     lowercase__					      =		FlaxUNetaDConditionModel.from_pretrained(
							    _UpperCAmelCase				, subfolder="""unet"""				, dtype=_UpperCAmelCase				, revision=_UpperCAmelCase		)
							return model, params

					def   lowerCamelCase__					(self   :						List[Any]				, _UpperCAmelCase   :						int=0				, _UpperCAmelCase   :						Optional[int]=(4, 77, 768)				, _UpperCAmelCase   :						Any=False		)						->       Optional[Any]:




							"""simple docstring"""


							lowercase__					      =		jnp.bfloataa if fpaa else jnp.floataa
							lowercase__					      =		jnp.array(load_hf_numpy(self.get_file_format(_UpperCAmelCase				, _UpperCAmelCase		)		)				, dtype=_UpperCAmelCase		)
							return hidden_states

					@parameterized.expand(
					    [
					        # fmt: off
					        [83, 4, [-0.2_323, -0.1_304, 0.0_813, -0.3_093, -0.0_919, -0.1_571, -0.1_125, -0.5_806]],
					        [17, 0.55, [-0.0_831, -0.2_443, 0.0_901, -0.0_919, 0.3_396, 0.0_103, -0.3_743, 0.0_701]],
					        [8, 0.89, [-0.4_863, 0.0_859, 0.0_875, -0.1_658, 0.9_199, -0.0_114, 0.4_839, 0.4_639]],
					        [3, 1000, [-0.5_649, 0.2_402, -0.5_518, 0.1_248, 1.1_328, -0.2_443, -0.0_325, -1.0_078]],
					        # fmt: on
					    ]		)
					def   lowerCamelCase__					(self   :						int				, _UpperCAmelCase   :						str				, _UpperCAmelCase   :						Union[str, Any]				, _UpperCAmelCase   :						int		)						->       str:




							"""simple docstring"""


							lowercase__      ,     lowercase__					      =		self.get_unet_model(model_id="""CompVis/stable-diffusion-v1-4"""				, fpaa=_UpperCAmelCase		)
							lowercase__					      =		self.get_latents(_UpperCAmelCase				, fpaa=_UpperCAmelCase		)
							lowercase__					      =		self.get_encoder_hidden_states(_UpperCAmelCase				, fpaa=_UpperCAmelCase		)

							lowercase__					      =		model.apply(
							    {"""params""": params}				, _UpperCAmelCase				, jnp.array(_UpperCAmelCase				, dtype=jnp.intaa		)				, encoder_hidden_states=_UpperCAmelCase				, ).sample

							assert sample.shape == latents.shape

							lowercase__					      =		jnp.asarray(jax.device_get((sample[-1, -2:, -2:, :2].flatten())		)				, dtype=jnp.floataa		)
							lowercase__					      =		jnp.array(_UpperCAmelCase				, dtype=jnp.floataa		)

							# Found torch (float16) and flax (bfloat16) outputs to be within this tolerance, in the same hardware
							assert jnp.allclose(_UpperCAmelCase				, _UpperCAmelCase				, atol=1E-2		)


					@parameterized.expand(
					    [
					        # fmt: off
					        [83, 4, [0.1_514, 0.0_807, 0.1_624, 0.1_016, -0.1_896, 0.0_263, 0.0_677, 0.2_310]],
					        [17, 0.55, [0.1_164, -0.0_216, 0.0_170, 0.1_589, -0.3_120, 0.1_005, -0.0_581, -0.1_458]],
					        [8, 0.89, [-0.1_758, -0.0_169, 0.1_004, -0.1_411, 0.1_312, 0.1_103, -0.1_996, 0.2_139]],
					        [3, 1000, [0.1_214, 0.0_352, -0.0_731, -0.1_562, -0.0_994, -0.0_906, -0.2_340, -0.0_539]],
					        # fmt: on
					    ]		)
					def   lowerCamelCase__					(self   :						Optional[Any]				, _UpperCAmelCase   :						Tuple				, _UpperCAmelCase   :						int				, _UpperCAmelCase   :						List[str]		)						->       str:




							"""simple docstring"""


							lowercase__      ,     lowercase__					      =		self.get_unet_model(model_id="""stabilityai/stable-diffusion-2"""				, fpaa=_UpperCAmelCase		)
							lowercase__					      =		self.get_latents(_UpperCAmelCase				, shape=(4, 4, 96, 96)				, fpaa=_UpperCAmelCase		)
							lowercase__					      =		self.get_encoder_hidden_states(_UpperCAmelCase				, shape=(4, 77, 1024)				, fpaa=_UpperCAmelCase		)

							lowercase__					      =		model.apply(
							    {"""params""": params}				, _UpperCAmelCase				, jnp.array(_UpperCAmelCase				, dtype=jnp.intaa		)				, encoder_hidden_states=_UpperCAmelCase				, ).sample

							assert sample.shape == latents.shape

							lowercase__					      =		jnp.asarray(jax.device_get((sample[-1, -2:, -2:, :2].flatten())		)				, dtype=jnp.floataa		)
							lowercase__					      =		jnp.array(_UpperCAmelCase				, dtype=jnp.floataa		)

							# Found torch (float16) and flax (bfloat16) outputs to be within this tolerance, on the same hardware
							assert jnp.allclose(_UpperCAmelCase				, _UpperCAmelCase				, atol=1E-2		)

