





'''simple docstring'''



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


_SCREAMING_SNAKE_CASE					,	_SCREAMING_SNAKE_CASE					,	_SCREAMING_SNAKE_CASE					=    False, False, False






@dataclass
class 							__lowercase       :







	'''simple docstring'''

	a    :				Optional[int]											=       None
	a    :				bool											=       True
	a    :				bool											=       True
	a    :				Optional[str]											=       None
	# Automatically constructed
	a    :				ClassVar[str]											=       "dict"
	a    :				ClassVar[Any]											=       pa.struct({"bytes": pa.binary(), "path": pa.string()}						)
	a    :				str											=       field(default="Audio" ,      init=lowerCAmelCase__ ,      repr=lowerCAmelCase__						)


	def __call__(self )							->							Optional[int]:







				'''simple docstring'''





				return self.pa_type


	def 						_UpperCAmelCase     (self     ,_lowerCamelCase )							->							dict:







				'''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(_lowerCamelCase     ,_lowerCamelCase ):
							return {"bytes": None, "path": value}
				elif isinstance(_lowerCamelCase     ,_lowerCamelCase ):
							return {"bytes": value, "path": None}
				elif "array" in value:
							# convert the audio array to wav bytes
							__lowercase										=							BytesIO()
							sf.write(_lowerCamelCase     ,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!)
													__lowercase										=							np.frombuffer(value['''bytes''']     ,dtype=np.intaa ).astype(np.floataa ) / 32767
										else:
													__lowercase										=							np.memmap(value['''path''']     ,dtype='''h'''     ,mode='''r''' ).astype(np.floataa ) / 32767

										__lowercase										=							BytesIO(bytes() )
										sf.write(_lowerCamelCase     ,_lowerCamelCase     ,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 						_UpperCAmelCase     (self     ,_lowerCamelCase     ,_lowerCamelCase = None )							->							dict:







				'''simple docstring'''





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

				__lowercase						,		__lowercase										=							(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

				__lowercase										=							xsplitext(_lowerCamelCase )[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:
							__lowercase										=							token_per_repo_id or {}
							__lowercase										=							path.split('''::''' )[-1]
							try:
										__lowercase										=							string_to_dict(_lowerCamelCase     ,config.HUB_DATASETS_URL )['''repo_id''']
										__lowercase										=							token_per_repo_id[repo_id]
							except (ValueError, KeyError):
										__lowercase										=							None

							with xopen(_lowerCamelCase     ,'''rb'''     ,use_auth_token=_lowerCamelCase ) as f:
										__lowercase						,		__lowercase										=							sf.read(_lowerCamelCase )

				else:
							__lowercase						,		__lowercase										=							sf.read(_lowerCamelCase )

				__lowercase										=							array.T
				if self.mono:
							__lowercase										=							librosa.to_mono(_lowerCamelCase )
				if self.sampling_rate and self.sampling_rate != sampling_rate:
							__lowercase										=							librosa.resample(_lowerCamelCase     ,orig_sr=_lowerCamelCase     ,target_sr=self.sampling_rate )
							__lowercase										=							self.sampling_rate

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


	def 						_UpperCAmelCase     (self )							->							Union["FeatureType", Dict[str, "FeatureType"]]:







				'''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 						_UpperCAmelCase     (self     ,_lowerCamelCase )							->							pa.StructArray:







				'''simple docstring'''





				if pa.types.is_string(storage.type ):
							__lowercase										=							pa.array([None] * len(_lowerCamelCase )     ,type=pa.binary() )
							__lowercase										=							pa.StructArray.from_arrays([bytes_array, storage]     ,['''bytes''', '''path''']     ,mask=storage.is_null() )
				elif pa.types.is_binary(storage.type ):
							__lowercase										=							pa.array([None] * len(_lowerCamelCase )     ,type=pa.string() )
							__lowercase										=							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''' ):
							__lowercase										=							pa.array([Audio().encode_example(_lowerCamelCase ) 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:
										__lowercase										=							storage.field('''bytes''' )
							else:
										__lowercase										=							pa.array([None] * len(_lowerCamelCase )     ,type=pa.binary() )
							if storage.type.get_field_index('''path''' ) >= 0:
										__lowercase										=							storage.field('''path''' )
							else:
										__lowercase										=							pa.array([None] * len(_lowerCamelCase )     ,type=pa.string() )
							__lowercase										=							pa.StructArray.from_arrays([bytes_array, path_array]     ,['''bytes''', '''path''']     ,mask=storage.is_null() )
				return array_cast(_lowerCamelCase     ,self.pa_type )


	def 						_UpperCAmelCase     (self     ,_lowerCamelCase )							->							pa.StructArray:







				'''simple docstring'''





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

				__lowercase										=							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()     ,)
				__lowercase										=							pa.array(
				    [os.path.basename(_lowerCamelCase ) if path is not None else None for path in storage.field('''path''' ).to_pylist()]     ,type=pa.string()     ,)
				__lowercase										=							pa.StructArray.from_arrays([bytes_array, path_array]     ,['''bytes''', '''path''']     ,mask=bytes_array.is_null() )
				return array_cast(_lowerCamelCase     ,self.pa_type )












'''simple docstring'''



from __future__ import annotations

from typing import Any






class 							__lowercase       :







	'''simple docstring'''

	def __init__(self     ,_lowerCamelCase )							->							None:







				'''simple docstring'''





				__lowercase										=							num_of_nodes
				__lowercase										=							[]
				__lowercase										=							{}


	def 						_UpperCAmelCase     (self     ,_lowerCamelCase     ,_lowerCamelCase     ,_lowerCamelCase )							->							None:







				'''simple docstring'''





				self.m_edges.append([u_node, v_node, weight] )


	def 						_UpperCAmelCase     (self     ,_lowerCamelCase )							->							int:







				'''simple docstring'''





				if self.m_component[u_node] == u_node:
							return u_node
				return self.find_component(self.m_component[u_node] )


	def 						_UpperCAmelCase     (self     ,_lowerCamelCase )							->							None:







				'''simple docstring'''





				if self.m_component[u_node] != u_node:
							for k in self.m_component:
										__lowercase										=							self.find_component(_lowerCamelCase )


	def 						_UpperCAmelCase     (self     ,_lowerCamelCase     ,_lowerCamelCase     ,_lowerCamelCase )							->							None:







				'''simple docstring'''





				if component_size[u_node] <= component_size[v_node]:
							__lowercase										=							v_node
							component_size[v_node] += component_size[u_node]
							self.set_component(_lowerCamelCase )

				elif component_size[u_node] >= component_size[v_node]:
							__lowercase										=							self.find_component(_lowerCamelCase )
							component_size[u_node] += component_size[v_node]
							self.set_component(_lowerCamelCase )


	def 						_UpperCAmelCase     (self )							->							None:







				'''simple docstring'''





				__lowercase										=							[]
				__lowercase										=							0

				__lowercase										=							[-1] * self.m_num_of_nodes

				# A list of components (initialized to all of the nodes)
				for node in range(self.m_num_of_nodes ):
							self.m_component.update({node: node} )
							component_size.append(1 )

				__lowercase										=							self.m_num_of_nodes

				while num_of_components > 1:
							for edge in self.m_edges:
										__lowercase						,		__lowercase						,		__lowercase										=							edge

										__lowercase										=							self.m_component[u]
										__lowercase										=							self.m_component[v]

										if u_component != v_component:

													for component in (u_component, v_component):
																if (
																    minimum_weight_edge[component] == -1
																    or minimum_weight_edge[component][2] > w
																):
																			__lowercase										=							[u, v, w]

							for edge in minimum_weight_edge:
										if isinstance(_lowerCamelCase     ,_lowerCamelCase ):
													__lowercase						,		__lowercase						,		__lowercase										=							edge

													__lowercase										=							self.m_component[u]
													__lowercase										=							self.m_component[v]

													if u_component != v_component:
																mst_weight += w
																self.union(_lowerCamelCase     ,_lowerCamelCase     ,_lowerCamelCase )
																print(f"Added edge [{u} - {v}]\nAdded weight: {w}\n" )
																num_of_components -= 1

							__lowercase										=							[-1] * self.m_num_of_nodes
				print(f"The total weight of the minimal spanning tree is: {mst_weight}" )





def    _lowerCAmelCase						(				):
			pass


if __name__ == "__main__":
	import doctest

	doctest.testmod()





