






'''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 	__UpperCAmelCase       ( __magic_name__						)->    Dict:  # picklable for multiprocessing



			"""simple docstring"""



			return x.sum()

def 	__UpperCAmelCase       ( __magic_name__						)->    List[Any]:  # picklable for multiprocessing



			"""simple docstring"""



			return i + 1







@dataclass
class    A_	:
							"""simple docstring"""




							a__   		=	42
							a__   		=	42







class    A_	(a_							):
							"""simple docstring"""




							def 							_A	(		self			:Dict				)				->  Any:


										'''simple docstring'''




										snake_case_				:					List[str]   =       {}
										snake_case_				:					Optional[Any]   =       []
										snake_case_				:					str   =       1
										snake_case_				:					Optional[int]   =       [1, 2]
										snake_case_				:					Optional[int]   =       {"a": 1, "b": 2}
										snake_case_				:					List[str]   =       {"a": [1, 2], "b": [3, 4]}
										snake_case_				:					Dict   =       {"a": {"1": 1}, "b": 2}
										snake_case_				:					Any   =       {"a": 1, "b": 2, "c": 3, "d": 4}
										snake_case_				:					Optional[Any]   =       {}
										snake_case_				:					Any   =       []
										snake_case_				:					List[Any]   =       2
										snake_case_				:					Tuple   =       [2, 3]
										snake_case_				:					Any   =       {"a": 2, "b": 3}
										snake_case_				:					str   =       {"a": [2, 3], "b": [4, 5]}
										snake_case_				:					Optional[int]   =       {"a": {"1": 2}, "b": 3}
										snake_case_				:					int   =       {"a": 2, "b": 3, "c": 4, "d": 5}
										self.assertEqual(map_nested(lowerCAmelCase__       ,							lowerCAmelCase__				)       ,							lowerCAmelCase__				)
										self.assertEqual(map_nested(lowerCAmelCase__       ,							lowerCAmelCase__				)       ,							lowerCAmelCase__				)
										self.assertEqual(map_nested(lowerCAmelCase__       ,							lowerCAmelCase__				)       ,							lowerCAmelCase__				)
										self.assertEqual(map_nested(lowerCAmelCase__       ,							lowerCAmelCase__				)       ,							lowerCAmelCase__				)
										self.assertEqual(map_nested(lowerCAmelCase__       ,							lowerCAmelCase__				)       ,							lowerCAmelCase__				)
										self.assertEqual(map_nested(lowerCAmelCase__       ,							lowerCAmelCase__				)       ,							lowerCAmelCase__				)
										self.assertEqual(map_nested(lowerCAmelCase__       ,							lowerCAmelCase__				)       ,							lowerCAmelCase__				)
										self.assertEqual(map_nested(lowerCAmelCase__       ,							lowerCAmelCase__				)       ,							lowerCAmelCase__				)

										snake_case_				:					Optional[int]   =       2
										self.assertEqual(map_nested(lowerCAmelCase__       ,							lowerCAmelCase__       ,							num_proc=lowerCAmelCase__				)       ,							lowerCAmelCase__				)
										self.assertEqual(map_nested(lowerCAmelCase__       ,							lowerCAmelCase__       ,							num_proc=lowerCAmelCase__				)       ,							lowerCAmelCase__				)
										self.assertEqual(map_nested(lowerCAmelCase__       ,							lowerCAmelCase__       ,							num_proc=lowerCAmelCase__				)       ,							lowerCAmelCase__				)
										self.assertEqual(map_nested(lowerCAmelCase__       ,							lowerCAmelCase__       ,							num_proc=lowerCAmelCase__				)       ,							lowerCAmelCase__				)
										self.assertEqual(map_nested(lowerCAmelCase__       ,							lowerCAmelCase__       ,							num_proc=lowerCAmelCase__				)       ,							lowerCAmelCase__				)
										self.assertEqual(map_nested(lowerCAmelCase__       ,							lowerCAmelCase__       ,							num_proc=lowerCAmelCase__				)       ,							lowerCAmelCase__				)
										self.assertEqual(map_nested(lowerCAmelCase__       ,							lowerCAmelCase__       ,							num_proc=lowerCAmelCase__				)       ,							lowerCAmelCase__				)
										self.assertEqual(map_nested(lowerCAmelCase__       ,							lowerCAmelCase__       ,							num_proc=lowerCAmelCase__				)       ,							lowerCAmelCase__				)

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

							def 							_A	(		self			:Union[str, Any]				)				->  List[str]:


										'''simple docstring'''




										snake_case_				:					Union[str, Any]   =       {"a": 1, "b": 2}
										snake_case_				:					List[Any]   =       {"a": 3, "b": 4}
										snake_case_				:					Optional[int]   =       {"a": 5, "b": 6}
										snake_case_				:					Optional[Any]   =       sorted([("a", (1, 3, 5)), ("b", (2, 4, 6))]				)
										self.assertEqual(sorted(zip_dict(lowerCAmelCase__       ,							lowerCAmelCase__       ,							lowerCAmelCase__				)				)       ,							lowerCAmelCase__				)






							def 							_A	(		self			:Tuple				)				->  str:


										'''simple docstring'''




										class    A_	:
																	"""simple docstring"""




																	a__   		=	'''bar'''
										snake_case_				:					int   =       Foo()
										self.assertEqual(foo.my_attr       ,							"bar"				)
										with temporary_assignment(lowerCAmelCase__       ,							"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 	__UpperCAmelCase       ( __magic_name__     ,__magic_name__     ,__magic_name__						)->    int:



			"""simple docstring"""



			with patch("datasets.utils.py_utils._single_map_nested"						) as mock_single_map_nested, patch(
			    "datasets.parallel.parallel.Pool"						) as mock_multiprocessing_pool:
						snake_case_				:					Any   =       {F'''{i}''': i for i in range(__magic_name__						)}
						snake_case_				:					Any   =       map_nested(lambda __magic_name__						: x + 10     ,__magic_name__     ,num_proc=__magic_name__     ,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    A_	(a_							):
							"""simple docstring"""




							@require_tf
							def 							_A	(		self			:Any				)				->  Any:


										'''simple docstring'''




										import tensorflow as tf
										from tensorflow.keras import layers

										snake_case_				:					List[str]   =       layers.Dense(2				)

										def gen_random_output():
													snake_case_				:					List[Any]   =       tf.random.uniform((1, 3)				)
													return model(lowerCAmelCase__				).numpy()

										with temp_seed(42       ,							set_tensorflow=lowerCAmelCase__				):
													snake_case_				:					Union[str, Any]   =       gen_random_output()
										with temp_seed(42       ,							set_tensorflow=lowerCAmelCase__				):
													snake_case_				:					str   =       gen_random_output()
										snake_case_				:					str   =       gen_random_output()

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

							@require_torch
							def 							_A	(		self			:Tuple				)				->  Any:


										'''simple docstring'''




										import torch

										def gen_random_output():
													snake_case_				:					Dict   =       torch.nn.Linear(3       ,							2				)
													snake_case_				:					int   =       torch.rand(1       ,							3				)
													return model(lowerCAmelCase__				).detach().numpy()

										with temp_seed(42       ,							set_pytorch=lowerCAmelCase__				):
													snake_case_				:					Any   =       gen_random_output()
										with temp_seed(42       ,							set_pytorch=lowerCAmelCase__				):
													snake_case_				:					Any   =       gen_random_output()
										snake_case_				:					Tuple   =       gen_random_output()

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






							def 							_A	(		self			:Any				)				->  Tuple:


										'''simple docstring'''




										def gen_random_output():
													return np.random.rand(1       ,							3				)

										with temp_seed(42				):
													snake_case_				:					List[str]   =       gen_random_output()
										with temp_seed(42				):
													snake_case_				:					int   =       gen_random_output()
										snake_case_				:					Any   =       gen_random_output()

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

@pytest.mark.parametrize("input_data"     ,[{}]						)
def 	__UpperCAmelCase       ( __magic_name__						)->    Dict:



			"""simple docstring"""



			snake_case_				:					str   =       NestedDataStructure(__magic_name__						).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 	__UpperCAmelCase       ( __magic_name__     ,__magic_name__						)->    Optional[Any]:



			"""simple docstring"""



			snake_case_				:					Any   =       NestedDataStructure(__magic_name__						).flatten()
			assert output == expected_output

def 	__UpperCAmelCase       ( )->    Union[str, Any]:



			"""simple docstring"""



			snake_case_				:					Tuple   =       A(x=1     ,y="foobar"						)
			snake_case_				:					Tuple   =       {"x": 1, "y": "foobar"}
			assert asdict(__magic_name__						) == expected_output

			snake_case_				:					int   =       {"a": {"b": A(x=10     ,y="foo"						)}, "c": [A(x=20     ,y="bar"						)]}
			snake_case_				:					str   =       {"a": {"b": {"x": 10, "y": "foo"}}, "c": [{"x": 20, "y": "bar"}]}
			assert asdict(__magic_name__						) == expected_output

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

def 	__UpperCAmelCase       ( __magic_name__						)->    Tuple:



			"""simple docstring"""



			return text.split()

def 	__UpperCAmelCase       ( __magic_name__						)->    Optional[Any]:



			"""simple docstring"""



			yield (time.time(), content)
			time.sleep(2						)
			yield (time.time(), content)






def 	__UpperCAmelCase       ( )->    Tuple:



			"""simple docstring"""



			with Pool(2						) as pool:
						snake_case_				:					List[Any]   =       list(iflatmap_unordered(__magic_name__     ,_split_text     ,kwargs_iterable=[{"text": "hello there"}] * 10						)						)
						assert out.count("hello"						) == 10
						assert out.count("there"						) == 10
						assert len(__magic_name__						) == 20

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

			# check that we get items as fast as possible
			with Pool(2						) as pool:
						snake_case_				:					Any   =       []
						for yield_time, content in iflatmap_unordered(
						    __magic_name__     ,_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(__magic_name__						)
						assert out.count("a"						) == 2
						assert out.count("b"						) == 2
						assert len(__magic_name__						) == 4












'''simple docstring'''




def 	__UpperCAmelCase       ( __magic_name__						)->    list[int]:



			"""simple docstring"""



			if length <= 0 or not isinstance(__magic_name__     ,__magic_name__						):
						raise ValueError("Length must be a positive integer."						)
			return [n * (2 * n - 1) for n in range(__magic_name__						)]


if __name__ == "__main__":
		print(hexagonal_numbers(length=5))
		print(hexagonal_numbers(length=10))




