





from abc import ABC, abstractmethod
from argparse import ArgumentParser



class       A__     ( __snake_case    ):





      @staticmethod
      @abstractmethod
      def  __UpperCamelCase(				A_     ):
       '''simple docstring'''




       raise NotImplementedError()





      @abstractmethod
      def  __UpperCamelCase(				self     ):
       '''simple docstring'''




       raise NotImplementedError()







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 	A_       (  _lowerCAmelCase  )		->				Union[str, Any]:  # picklable for multiprocessing
 return x.sum()



def 	A_       (  _lowerCAmelCase  )		->				Optional[Any]:  # picklable for multiprocessing
 return i + 1



@dataclass
class       A__     :
      _UpperCAmelCase			:int
      _UpperCAmelCase			:str



class       A__     ( __snake_case    ):





      def  __UpperCamelCase(				self     ):
       '''simple docstring'''




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

       UpperCamelCase			:      List[str] 			=     2
       self.assertEqual(map_nested(A_ ,							A_ ,							num_proc=A_     ) ,							A_     )
       self.assertEqual(map_nested(A_ ,							A_ ,							num_proc=A_     ) ,							A_     )
       self.assertEqual(map_nested(A_ ,							A_ ,							num_proc=A_     ) ,							A_     )
       self.assertEqual(map_nested(A_ ,							A_ ,							num_proc=A_     ) ,							A_     )
       self.assertEqual(map_nested(A_ ,							A_ ,							num_proc=A_     ) ,							A_     )
       self.assertEqual(map_nested(A_ ,							A_ ,							num_proc=A_     ) ,							A_     )
       self.assertEqual(map_nested(A_ ,							A_ ,							num_proc=A_     ) ,							A_     )
       self.assertEqual(map_nested(A_ ,							A_ ,							num_proc=A_     ) ,							A_     )

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





      def  __UpperCamelCase(				self     ):
       '''simple docstring'''




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





      def  __UpperCamelCase(				self     ):
       '''simple docstring'''




       class       A__     :
             _UpperCAmelCase			:str							=     'bar'


       UpperCamelCase			:      List[Any] 			=     Foo()
       self.assertEqual(foo.my_attr ,							"bar"     )
       with temporary_assignment(A_ ,							"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 	A_       (  _lowerCAmelCase							,     _lowerCAmelCase							,     _lowerCAmelCase  )		->				Optional[Any]:
 with patch("datasets.utils.py_utils._single_map_nested"  ) as mock_single_map_nested, patch(
     "datasets.parallel.parallel.Pool"  ) as mock_multiprocessing_pool:
  UpperCamelCase			:      Union[str, Any] 			=     {F"""{i}""": i for i in range(_lowerCAmelCase  )}
  UpperCamelCase			:      List[str] 			=     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       A__     ( __snake_case    ):





      @require_tf
      def  __UpperCamelCase(				self     ):
       '''simple docstring'''




       import tensorflow as tf
       from tensorflow.keras import layers

       UpperCamelCase			:      int 			=     layers.Dense(2     )

       def gen_random_output():
        UpperCamelCase			:      Optional[Any] 			=     tf.random.uniform((1, 3)     )
        return model(A_     ).numpy()

       with temp_seed(42 ,							set_tensorflow=A_     ):
        UpperCamelCase			:      List[Any] 			=     gen_random_output()
       with temp_seed(42 ,							set_tensorflow=A_     ):
        UpperCamelCase			:      Dict 			=     gen_random_output()
       UpperCamelCase			:      Optional[int] 			=     gen_random_output()

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





      @require_torch
      def  __UpperCamelCase(				self     ):
       '''simple docstring'''




       import torch

       def gen_random_output():
        UpperCamelCase			:      Optional[Any] 			=     torch.nn.Linear(3 ,							2     )
        UpperCamelCase			:      Dict 			=     torch.rand(1 ,							3     )
        return model(A_     ).detach().numpy()

       with temp_seed(42 ,							set_pytorch=A_     ):
        UpperCamelCase			:      Dict 			=     gen_random_output()
       with temp_seed(42 ,							set_pytorch=A_     ):
        UpperCamelCase			:      Optional[int] 			=     gen_random_output()
       UpperCamelCase			:      List[Any] 			=     gen_random_output()

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





      def  __UpperCamelCase(				self     ):
       '''simple docstring'''




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

       with temp_seed(42     ):
        UpperCamelCase			:      Optional[Any] 			=     gen_random_output()
       with temp_seed(42     ):
        UpperCamelCase			:      Optional[Any] 			=     gen_random_output()
       UpperCamelCase			:      Optional[Any] 			=     gen_random_output()

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



@pytest.mark.parametrize("input_data"							,     [{}]  )
def 	A_       (  _lowerCAmelCase  )		->				List[Any]:
 UpperCamelCase			:      Optional[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 	A_       (  _lowerCAmelCase							,     _lowerCAmelCase  )		->				Tuple:
 UpperCamelCase			:      Dict 			=     NestedDataStructure(_lowerCAmelCase  ).flatten()
 assert output == expected_output



def 	A_       (  )		->				List[Any]:
 UpperCamelCase			:      str 			=     A(x=1							,     y="foobar"  )
 UpperCamelCase			:      Tuple 			=     {"x": 1, "y": "foobar"}
 assert asdict(_lowerCAmelCase  ) == expected_output

 UpperCamelCase			:      List[str] 			=     {"a": {"b": A(x=10							,     y="foo"  )}, "c": [A(x=20							,     y="bar"  )]}
 UpperCamelCase			:      Tuple 			=     {"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 	A_       (  _lowerCAmelCase  )		->				Tuple:
 return text.split()



def 	A_       (  _lowerCAmelCase  )		->				Dict:
 yield (time.time(), content)
 time.sleep(2  )
 yield (time.time(), content)



def 	A_       (  )		->				str:
 with Pool(2  ) as pool:
  UpperCamelCase			:      List[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 multiprocess from pathos (uses dill for pickling)
 with multiprocess.Pool(2  ) as pool:
  UpperCamelCase			:      Dict 			=     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:
  UpperCamelCase			:      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
