





import os
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen, xsplitext
from ..table import array_cast
from ..utils.py_utils import no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
       from .features import FeatureType


_lowerCamelCase       ,	_lowerCamelCase       ,	_lowerCamelCase			       =      False, False, False







@dataclass
class 			__A       :




   """simple docstring"""







   UpperCAmelCase__						=      None
   UpperCAmelCase__						=      True
   UpperCAmelCase__						=      True
   UpperCAmelCase__						=      None
   # Automatically constructed
   UpperCAmelCase__						=      """dict"""
   UpperCAmelCase__						=      pa.struct({"""bytes""": pa.binary(), """path""": pa.string()}  )
   UpperCAmelCase__						=      field(default="""Audio""" ,init=lowerCamelCase__ ,repr=lowerCamelCase__  )


   def __call__(		self):



          """simple docstring"""





          return self.pa_type


   def 				__snake_case			(		self      ,							a__):



          """simple docstring"""





          try:
                 import soundfile as sf  # soundfile is a dependency of librosa, needed to decode audio files.
          except ImportError as err:
                 raise ImportError('''To support encoding audio data, please install \'soundfile\'.''') from err
          if isinstance(__a      ,							__a):
                 return {"bytes": None, "path": value}
          elif isinstance(__a      ,							__a):
                 return {"bytes": value, "path": None}
          elif "array" in value:
                 # convert the audio array to wav bytes
                 _lowerCamelCase							:     Any	=		BytesIO()
                 sf.write(__a      ,							value['''array''']      ,							value['''sampling_rate''']      ,							format='''wav''')
                 return {"bytes": buffer.getvalue(), "path": None}
          elif value.get('''path''') is not None and os.path.isfile(value['''path''']):
                 # we set "bytes": None to not duplicate the data if they're already available locally
                 if value["path"].endswith('''pcm'''):
                        # "PCM" only has raw audio bytes
                        if value.get('''sampling_rate''') is None:
                               # At least, If you want to convert "PCM-byte" to "WAV-byte", you have to know sampling rate
                               raise KeyError('''To use PCM files, please specify a \'sampling_rate\' in Audio object''')
                        if value.get('''bytes'''):
                               # If we already had PCM-byte, we don`t have to make "read file, make bytes" (just use it!)
                               _lowerCamelCase							:     Any	=		np.frombuffer(value['''bytes''']      ,							dtype=np.intaa).astype(np.floataa) / 3_2767
                        else:
                               _lowerCamelCase							:     Optional[Any]	=		np.memmap(value['''path''']      ,							dtype='''h'''      ,							mode='''r''').astype(np.floataa) / 3_2767

                        _lowerCamelCase							:     int	=		BytesIO(bytes())
                        sf.write(__a      ,							__a      ,							value['''sampling_rate''']      ,							format='''wav''')
                        return {"bytes": buffer.getvalue(), "path": None}
                 else:
                        return {"bytes": None, "path": value.get('''path''')}
          elif value.get('''bytes''') is not None or value.get('''path''') is not None:
                 # store the audio bytes, and path is used to infer the audio format using the file extension
                 return {"bytes": value.get('''bytes'''), "path": value.get('''path''')}
          else:
                 raise ValueError(
                     F"""An audio sample should have one of \'path\' or \'bytes\' but they are missing or None in {value}.""")


   def 				__snake_case			(		self      ,							a__      ,							a__ = None):



          """simple docstring"""





          if not self.decode:
                 raise RuntimeError('''Decoding is disabled for this feature. Please use Audio(decode=True) instead.''')

          _lowerCamelCase,  _lowerCamelCase							:     List[Any]	=		(value['''path'''], BytesIO(value['''bytes'''])) if value['''bytes'''] is not None else (value['''path'''], None)
          if path is None and file is None:
                 raise ValueError(F"""An audio sample should have one of \'path\' or \'bytes\' but both are None in {value}.""")

          try:
                 import librosa
                 import soundfile as sf
          except ImportError as err:
                 raise ImportError('''To support decoding audio files, please install \'librosa\' and \'soundfile\'.''') from err

          _lowerCamelCase							:     List[str]	=		xsplitext(__a)[1][1:].lower() if path is not None else None
          if not config.IS_OPUS_SUPPORTED and audio_format == "opus":
                 raise RuntimeError(
                     '''Decoding \'opus\' files requires system library \'libsndfile\'>=1.0.31, '''
                     '''You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. ''')
          elif not config.IS_MP3_SUPPORTED and audio_format == "mp3":
                 raise RuntimeError(
                     '''Decoding \'mp3\' files requires system library \'libsndfile\'>=1.1.0, '''
                     '''You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. ''')

          if file is None:
                 _lowerCamelCase							:     Optional[int]	=		token_per_repo_id or {}
                 _lowerCamelCase							:     str	=		path.split('''::''')[-1]
                 try:
                        _lowerCamelCase							:     Tuple	=		string_to_dict(__a      ,							config.HUB_DATASETS_URL)['''repo_id''']
                        _lowerCamelCase							:     List[Any]	=		token_per_repo_id[repo_id]
                 except (ValueError, KeyError):
                        _lowerCamelCase							:     str	=		None

                 with xopen(__a      ,							'''rb'''      ,							use_auth_token=__a) as f:
                        _lowerCamelCase,  _lowerCamelCase							:     Optional[int]	=		sf.read(__a)

          else:
                 _lowerCamelCase,  _lowerCamelCase							:     Union[str, Any]	=		sf.read(__a)

          _lowerCamelCase							:     Dict	=		array.T
          if self.mono:
                 _lowerCamelCase							:     int	=		librosa.to_mono(__a)
          if self.sampling_rate and self.sampling_rate != sampling_rate:
                 _lowerCamelCase							:     List[Any]	=		librosa.resample(__a      ,							orig_sr=__a      ,							target_sr=self.sampling_rate)
                 _lowerCamelCase							:     Tuple	=		self.sampling_rate

          return {"path": path, "array": array, "sampling_rate": sampling_rate}


   def 				__snake_case			(		self):



          """simple docstring"""





          from .features import Value

          if self.decode:
                 raise ValueError('''Cannot flatten a decoded Audio feature.''')
          return {
              "bytes": Value('''binary'''),
              "path": Value('''string'''),
          }


   def 				__snake_case			(		self      ,							a__):



          """simple docstring"""





          if pa.types.is_string(storage.type):
                 _lowerCamelCase							:     Dict	=		pa.array([None] * len(__a)      ,							type=pa.binary())
                 _lowerCamelCase							:     str	=		pa.StructArray.from_arrays([bytes_array, storage]      ,							['''bytes''', '''path''']      ,							mask=storage.is_null())
          elif pa.types.is_binary(storage.type):
                 _lowerCamelCase							:     Dict	=		pa.array([None] * len(__a)      ,							type=pa.string())
                 _lowerCamelCase							:     Optional[int]	=		pa.StructArray.from_arrays([storage, path_array]      ,							['''bytes''', '''path''']      ,							mask=storage.is_null())
          elif pa.types.is_struct(storage.type) and storage.type.get_all_field_indices('''array'''):
                 _lowerCamelCase							:     List[str]	=		pa.array([Audio().encode_example(__a) if x is not None else None for x in storage.to_pylist()])
          elif pa.types.is_struct(storage.type):
                 if storage.type.get_field_index('''bytes''') >= 0:
                        _lowerCamelCase							:     str	=		storage.field('''bytes''')
                 else:
                        _lowerCamelCase							:     List[str]	=		pa.array([None] * len(__a)      ,							type=pa.binary())
                 if storage.type.get_field_index('''path''') >= 0:
                        _lowerCamelCase							:     str	=		storage.field('''path''')
                 else:
                        _lowerCamelCase							:     int	=		pa.array([None] * len(__a)      ,							type=pa.string())
                 _lowerCamelCase							:     Tuple	=		pa.StructArray.from_arrays([bytes_array, path_array]      ,							['''bytes''', '''path''']      ,							mask=storage.is_null())
          return array_cast(__a      ,							self.pa_type)



   def 				__snake_case			(		self      ,							a__):



          """simple docstring"""





          @no_op_if_value_is_null
          def path_to_bytes(a__):
                 with xopen(__a      ,							'''rb''') as f:
                        _lowerCamelCase							:     int	=		f.read()
                 return bytes_

          _lowerCamelCase							:     int	=		pa.array(
              [
                  (path_to_bytes(x['''path''']) if x['''bytes'''] is None else x['''bytes''']) if x is not None else None
                  for x in storage.to_pylist()
              ]      ,							type=pa.binary()      ,							)
          _lowerCamelCase							:     Tuple	=		pa.array(
              [os.path.basename(__a) if path is not None else None for path in storage.field('''path''').to_pylist()]      ,							type=pa.string()      ,							)
          _lowerCamelCase							:     Dict	=		pa.StructArray.from_arrays([bytes_array, path_array]      ,							['''bytes''', '''path''']      ,							mask=bytes_array.is_null())
          return array_cast(__a      ,							self.pa_type)








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(      lowercase_      ):  # picklable for multiprocessing
       return x.sum()




def    __UpperCAmelCase(      lowercase_      ):  # picklable for multiprocessing
       return i + 1







@dataclass
class 			__A       :




      """simple docstring"""







      UpperCAmelCase__						=      42
      UpperCAmelCase__						=      42







class 			__A       (      lowerCamelCase__  ):




      """simple docstring"""







      def 				__snake_case			(		self):



             """simple docstring"""





             _lowerCamelCase							:     Tuple	=		{}
             _lowerCamelCase							:     Dict	=		[]
             _lowerCamelCase							:     Optional[Any]	=		1
             _lowerCamelCase							:     Optional[int]	=		[1, 2]
             _lowerCamelCase							:     str	=		{'''a''': 1, '''b''': 2}
             _lowerCamelCase							:     Dict	=		{'''a''': [1, 2], '''b''': [3, 4]}
             _lowerCamelCase							:     Any	=		{'''a''': {'''1''': 1}, '''b''': 2}
             _lowerCamelCase							:     Optional[Any]	=		{'''a''': 1, '''b''': 2, '''c''': 3, '''d''': 4}
             _lowerCamelCase							:     str	=		{}
             _lowerCamelCase							:     int	=		[]
             _lowerCamelCase							:     str	=		2
             _lowerCamelCase							:     int	=		[2, 3]
             _lowerCamelCase							:     str	=		{'''a''': 2, '''b''': 3}
             _lowerCamelCase							:     Tuple	=		{'''a''': [2, 3], '''b''': [4, 5]}
             _lowerCamelCase							:     List[str]	=		{'''a''': {'''1''': 2}, '''b''': 3}
             _lowerCamelCase							:     str	=		{'''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__)

             _lowerCamelCase							:     Dict	=		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__)

             _lowerCamelCase							:     Any	=		{'''a''': np.eye(2), '''b''': np.zeros(3), '''c''': np.ones(2)}
             _lowerCamelCase							:     Optional[int]	=		{'''a''': 2, '''b''': 0, '''c''': 2}
             _lowerCamelCase							:     Optional[int]	=		{
                 '''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 				__snake_case			(		self):



             """simple docstring"""





             _lowerCamelCase							:     Dict	=		{'''a''': 1, '''b''': 2}
             _lowerCamelCase							:     Optional[int]	=		{'''a''': 3, '''b''': 4}
             _lowerCamelCase							:     int	=		{'''a''': 5, '''b''': 6}
             _lowerCamelCase							:     Optional[int]	=		sorted([('''a''', (1, 3, 5)), ('''b''', (2, 4, 6))])
             self.assertEqual(sorted(zip_dict(a__      ,							a__      ,							a__))      ,							a__)



      def 				__snake_case			(		self):



             """simple docstring"""





             class 			__A       :




                   """simple docstring"""







                   UpperCAmelCase__						=      """bar"""
             _lowerCamelCase							:     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    __UpperCAmelCase(      lowercase_ ,    lowercase_ ,    lowercase_      ):
       with patch('''datasets.utils.py_utils._single_map_nested'''      ) as mock_single_map_nested, patch(
           '''datasets.parallel.parallel.Pool'''      ) as mock_multiprocessing_pool:
              _lowerCamelCase							:     Union[str, Any]	=		{F"""{i}""": i for i in range(lowercase_      )}
              _lowerCamelCase							:     List[str]	=		map_nested(lambda lowercase_      : x + 10 ,    lowercase_ ,    num_proc=lowercase_ ,    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       (      lowerCamelCase__  ):




      """simple docstring"""







      @require_tf
      def 				__snake_case			(		self):



             """simple docstring"""





             import tensorflow as tf
             from tensorflow.keras import layers

             _lowerCamelCase							:     int	=		layers.Dense(2)

             def gen_random_output():
                    _lowerCamelCase							:     Union[str, Any]	=		tf.random.uniform((1, 3))
                    return model(a__).numpy()

             with temp_seed(42      ,							set_tensorflow=a__):
                    _lowerCamelCase							:     List[str]	=		gen_random_output()
             with temp_seed(42      ,							set_tensorflow=a__):
                    _lowerCamelCase							:     Any	=		gen_random_output()
             _lowerCamelCase							:     str	=		gen_random_output()

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


      @require_torch
      def 				__snake_case			(		self):



             """simple docstring"""





             import torch

             def gen_random_output():
                    _lowerCamelCase							:     Union[str, Any]	=		torch.nn.Linear(3      ,							2)
                    _lowerCamelCase							:     Dict	=		torch.rand(1      ,							3)
                    return model(a__).detach().numpy()

             with temp_seed(42      ,							set_pytorch=a__):
                    _lowerCamelCase							:     Any	=		gen_random_output()
             with temp_seed(42      ,							set_pytorch=a__):
                    _lowerCamelCase							:     Optional[int]	=		gen_random_output()
             _lowerCamelCase							:     Union[str, Any]	=		gen_random_output()

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



      def 				__snake_case			(		self):



             """simple docstring"""





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

             with temp_seed(42):
                    _lowerCamelCase							:     Union[str, Any]	=		gen_random_output()
             with temp_seed(42):
                    _lowerCamelCase							:     List[str]	=		gen_random_output()
             _lowerCamelCase							:     str	=		gen_random_output()

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




@pytest.mark.parametrize('''input_data''' ,    [{}]      )
def    __UpperCAmelCase(      lowercase_      ):
       _lowerCamelCase							:     List[Any]	=		NestedDataStructure(lowercase_      ).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(      lowercase_ ,    lowercase_      ):
       _lowerCamelCase							:     int	=		NestedDataStructure(lowercase_      ).flatten()
       assert output == expected_output




def    __UpperCAmelCase(      ):
       _lowerCamelCase							:     Any	=		A(x=1 ,    y='''foobar'''      )
       _lowerCamelCase							:     Union[str, Any]	=		{'''x''': 1, '''y''': '''foobar'''}
       assert asdict(lowercase_      ) == expected_output

       _lowerCamelCase							:     Optional[int]	=		{'''a''': {'''b''': A(x=10 ,    y='''foo'''      )}, '''c''': [A(x=20 ,    y='''bar'''      )]}
       _lowerCamelCase							:     Union[str, Any]	=		{'''a''': {'''b''': {'''x''': 10, '''y''': '''foo'''}}, '''c''': [{'''x''': 20, '''y''': '''bar'''}]}
       assert asdict(lowercase_      ) == expected_output

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




def    __UpperCAmelCase(      lowercase_      ):
       return text.split()




def    __UpperCAmelCase(      lowercase_      ):
       yield (time.time(), content)
       time.sleep(2      )
       yield (time.time(), content)




def    __UpperCAmelCase(      ):
       with Pool(2      ) as pool:
              _lowerCamelCase							:     Tuple	=		list(iflatmap_unordered(lowercase_ ,    _split_text ,    kwargs_iterable=[{'''text''': '''hello there'''}] * 10      )      )
              assert out.count('''hello'''      ) == 10
              assert out.count('''there'''      ) == 10
              assert len(lowercase_      ) == 20

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

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

