




"""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 						snake_case__       (					_lowerCamelCase				)			->str:  # picklable for multiprocessing






		"""simple docstring"""


		return x.sum()

def 						snake_case__       (					_lowerCamelCase				)			->Any:  # picklable for multiprocessing






		"""simple docstring"""


		return i + 1




@dataclass
class       lowerCAmelCase__       :


				"""simple docstring"""





				__UpperCAmelCase :     int
				__UpperCAmelCase :     str





class       lowerCAmelCase__       (			lowerCAmelCase_    ):


				"""simple docstring"""





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

						__lowercase				:		Dict												=				2
						self.assertEqual(map_nested(lowercase__    ,   lowercase__    ,   num_proc=lowercase__						)    ,   lowercase__						)
						self.assertEqual(map_nested(lowercase__    ,   lowercase__    ,   num_proc=lowercase__						)    ,   lowercase__						)
						self.assertEqual(map_nested(lowercase__    ,   lowercase__    ,   num_proc=lowercase__						)    ,   lowercase__						)
						self.assertEqual(map_nested(lowercase__    ,   lowercase__    ,   num_proc=lowercase__						)    ,   lowercase__						)
						self.assertEqual(map_nested(lowercase__    ,   lowercase__    ,   num_proc=lowercase__						)    ,   lowercase__						)
						self.assertEqual(map_nested(lowercase__    ,   lowercase__    ,   num_proc=lowercase__						)    ,   lowercase__						)
						self.assertEqual(map_nested(lowercase__    ,   lowercase__    ,   num_proc=lowercase__						)    ,   lowercase__						)
						self.assertEqual(map_nested(lowercase__    ,   lowercase__    ,   num_proc=lowercase__						)    ,   lowercase__						)

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



				def       snake_case    (     self					:	Optional[int]						):
						__lowercase				:		Tuple												=				{"a": 1, "b": 2}
						__lowercase				:		str												=				{"a": 3, "b": 4}
						__lowercase				:		int												=				{"a": 5, "b": 6}
						__lowercase				:		Any												=				sorted([("a", (1, 3, 5)), ("b", (2, 4, 6))]						)
						self.assertEqual(sorted(zip_dict(lowercase__    ,   lowercase__    ,   lowercase__						)						)    ,   lowercase__						)



				def       snake_case    (     self					:	Any						):




						class       lowerCAmelCase__       :


										"""simple docstring"""





										__UpperCAmelCase :     Union[str, Any]       =      "bar"





						__lowercase				:		Optional[Any]												=				Foo()
						self.assertEqual(foo.my_attr    ,   "bar"						)
						with temporary_assignment(lowercase__    ,   "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 						snake_case__       (					_lowerCamelCase,  _lowerCamelCase,  _lowerCamelCase				)			->Optional[Any]:






		"""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:
				__lowercase				:		str												=				{F'{i}': i for i in range(_lowerCamelCase				)}
				__lowercase				:		Optional[Any]												=				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       lowerCAmelCase__       (			lowerCAmelCase_    ):


				"""simple docstring"""





				@require_tf
				def       snake_case    (     self					:	Optional[int]						):
						import tensorflow as tf
						from tensorflow.keras import layers

						__lowercase				:		List[Any]												=				layers.Dense(2						)

						def gen_random_output():
								__lowercase				:		List[str]												=				tf.random.uniform((1, 3)						)
								return model(lowercase__						).numpy()

						with temp_seed(4_2    ,   set_tensorflow=lowercase__						):
								__lowercase				:		Union[str, Any]												=				gen_random_output()
						with temp_seed(4_2    ,   set_tensorflow=lowercase__						):
								__lowercase				:		int												=				gen_random_output()
						__lowercase				:		List[Any]												=				gen_random_output()

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



				@require_torch
				def       snake_case    (     self					:	Dict						):
						import torch

						def gen_random_output():
								__lowercase				:		Tuple												=				torch.nn.Linear(3    ,   2						)
								__lowercase				:		str												=				torch.rand(1    ,   3						)
								return model(lowercase__						).detach().numpy()

						with temp_seed(4_2    ,   set_pytorch=lowercase__						):
								__lowercase				:		Optional[int]												=				gen_random_output()
						with temp_seed(4_2    ,   set_pytorch=lowercase__						):
								__lowercase				:		List[str]												=				gen_random_output()
						__lowercase				:		Any												=				gen_random_output()

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



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

						with temp_seed(4_2						):
								__lowercase				:		List[str]												=				gen_random_output()
						with temp_seed(4_2						):
								__lowercase				:		Dict												=				gen_random_output()
						__lowercase				:		Any												=				gen_random_output()

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





@pytest.mark.parametrize("input_data",  [{}]				)
def 						snake_case__       (					_lowerCamelCase				)			->Any:






		"""simple docstring"""


		__lowercase				:		Any												=				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 						snake_case__       (					_lowerCamelCase,  _lowerCamelCase				)			->str:






		"""simple docstring"""


		__lowercase				:		Optional[int]												=				NestedDataStructure(_lowerCamelCase				).flatten()
		assert output == expected_output

def 						snake_case__       (					)			->int:






		"""simple docstring"""


		__lowercase				:		Dict												=				A(x=1,  y="foobar"				)
		__lowercase				:		Tuple												=				{"x": 1, "y": "foobar"}
		assert asdict(_lowerCamelCase				) == expected_output

		__lowercase				:		List[Any]												=				{"a": {"b": A(x=10,  y="foo"				)}, "c": [A(x=20,  y="bar"				)]}
		__lowercase				:		str												=				{"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 						snake_case__       (					_lowerCamelCase				)			->Dict:






		"""simple docstring"""


		return text.split()

def 						snake_case__       (					_lowerCamelCase				)			->List[Any]:






		"""simple docstring"""


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





def 						snake_case__       (					)			->Dict:






		"""simple docstring"""


		with Pool(2				) as pool:
				__lowercase				:		Tuple												=				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:
				__lowercase				:		str												=				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:
				__lowercase				:		str												=				[]
				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"""







# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import re

from ..utils import cached_file


# docstyle-ignore
__A   :  int					=      '\nHuman: <<task>>\n\nAssistant: '


__A   :  List[str]					=      'huggingface-tools/default-prompts'
__A   :  Tuple					=      {'chat': 'chat_prompt_template.txt', 'run': 'run_prompt_template.txt'}

def 						snake_case__       (					_lowerCamelCase,  _lowerCamelCase,  _lowerCamelCase="run"				)			->Optional[int]:






		"""simple docstring"""


		if prompt_or_repo_id is None:
				__lowercase				:		Any												=				DEFAULT_PROMPTS_REPO

		# prompt is considered a repo ID when it does not contain any kind of space
		if re.search("\\s",  _lowerCamelCase				) is not None:
				return prompt_or_repo_id

		__lowercase				:		Optional[Any]												=				cached_file(
		    _lowerCamelCase,  PROMPT_FILES[mode],  repo_type="dataset",  user_agent={"agent": agent_name}				)
		with open(_lowerCamelCase,  "r",  encoding="utf-8"				) as f:
				return f.read()
