





'''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


lowercase__      ,				lowercase__      ,				lowercase__       : Optional[int]         =       False, False, False


@dataclass
class 					SCREAMING_SNAKE_CASE       :
     lowerCAmelCase						     =			None
     lowerCAmelCase						     =			True
     lowerCAmelCase						     =			True
     lowerCAmelCase						     =			None
     # Automatically constructed
     lowerCAmelCase						     =			"dict"
     lowerCAmelCase						     =			pa.struct({'''bytes''': pa.binary(), '''path''': pa.string()}				)
     lowerCAmelCase						     =			field(default='''Audio'''   ,						init=a__   ,						repr=a__				)


     def __call__(		self):

          '''simple docstring'''






          return self.pa_type


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

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

                    __A		:					Optional[Any]						=			BytesIO(bytes())
                    sf.write(_UpperCAmelCase		,      _UpperCAmelCase		,      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 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase = None):

          '''simple docstring'''






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

          __A					,__A		:					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

          __A		:					Tuple						=			xsplitext(_UpperCAmelCase)[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:
               __A		:					List[str]						=			token_per_repo_id or {}
               __A		:					Optional[int]						=			path.split('::')[-1]
               try:
                    __A		:					str						=			string_to_dict(_UpperCAmelCase		,      config.HUB_DATASETS_URL)['repo_id']
                    __A		:					Union[str, Any]						=			token_per_repo_id[repo_id]
               except (ValueError, KeyError):
                    __A		:					List[Any]						=			None

               with xopen(_UpperCAmelCase		,      'rb'		,      use_auth_token=_UpperCAmelCase) as f:
                    __A					,__A		:					Optional[Any]						=			sf.read(_UpperCAmelCase)

          else:
               __A					,__A		:					Tuple						=			sf.read(_UpperCAmelCase)

          __A		:					Union[str, Any]						=			array.T
          if self.mono:
               __A		:					Optional[Any]						=			librosa.to_mono(_UpperCAmelCase)
          if self.sampling_rate and self.sampling_rate != sampling_rate:
               __A		:					List[Any]						=			librosa.resample(_UpperCAmelCase		,      orig_sr=_UpperCAmelCase		,      target_sr=self.sampling_rate)
               __A		:					Optional[Any]						=			self.sampling_rate

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


     def 				SCREAMING_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 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






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


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






          @no_op_if_value_is_null
          def path_to_bytes(_UpperCAmelCase):
               with xopen(_UpperCAmelCase		,      'rb') as f:
                    __A		:					Union[str, Any]						=			f.read()
               return bytes_

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






'''simple docstring'''




from __future__ import annotations

import math


class 					SCREAMING_SNAKE_CASE       :


     def __init__(		self		,      _UpperCAmelCase):

          '''simple docstring'''






          __A		:					int						=			size
          # approximate the overall size of segment tree with given value
          __A		:					Optional[Any]						=			[0 for i in range(0		,      4 * size)]
          # create array to store lazy update
          __A		:					Optional[Any]						=			[0 for i in range(0		,      4 * size)]
          __A		:					str						=			[0 for i in range(0		,      4 * size)]  # flag for lazy update


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






          return idx * 2


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






          return idx * 2 + 1


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase):

          '''simple docstring'''






          if left_element == right_element:
               __A		:					List[Any]						=			a[left_element - 1]
          else:
               __A		:					List[str]						=			(left_element + right_element) // 2
               self.build(self.left(_UpperCAmelCase)		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase)
               self.build(self.right(_UpperCAmelCase)		,      mid + 1		,      _UpperCAmelCase		,      _UpperCAmelCase)
               __A		:					Any						=			max(
                   self.segment_tree[self.left(_UpperCAmelCase)]		,      self.segment_tree[self.right(_UpperCAmelCase)])


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase):

          '''simple docstring'''






          if self.flag[idx] is True:
               __A		:					Optional[Any]						=			self.lazy[idx]
               __A		:					Optional[Any]						=			False
               if left_element != right_element:
                    __A		:					List[Any]						=			self.lazy[idx]
                    __A		:					Dict						=			self.lazy[idx]
                    __A		:					Tuple						=			True
                    __A		:					Union[str, Any]						=			True

          if right_element < a or left_element > b:
               return True
          if left_element >= a and right_element <= b:
               __A		:					Optional[int]						=			val
               if left_element != right_element:
                    __A		:					Tuple						=			val
                    __A		:					Any						=			val
                    __A		:					Tuple						=			True
                    __A		:					Union[str, Any]						=			True
               return True
          __A		:					str						=			(left_element + right_element) // 2
          self.update(self.left(_UpperCAmelCase)		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase)
          self.update(self.right(_UpperCAmelCase)		,      mid + 1		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase)
          __A		:					int						=			max(
              self.segment_tree[self.left(_UpperCAmelCase)]		,      self.segment_tree[self.right(_UpperCAmelCase)])
          return True


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase):

          '''simple docstring'''






          if self.flag[idx] is True:
               __A		:					Union[str, Any]						=			self.lazy[idx]
               __A		:					List[str]						=			False
               if left_element != right_element:
                    __A		:					Union[str, Any]						=			self.lazy[idx]
                    __A		:					Optional[int]						=			self.lazy[idx]
                    __A		:					str						=			True
                    __A		:					Union[str, Any]						=			True
          if right_element < a or left_element > b:
               return -math.inf
          if left_element >= a and right_element <= b:
               return self.segment_tree[idx]
          __A		:					Any						=			(left_element + right_element) // 2
          __A		:					int						=			self.query(self.left(_UpperCAmelCase)		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase)
          __A		:					Union[str, Any]						=			self.query(self.right(_UpperCAmelCase)		,      mid + 1		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase)
          return max(_UpperCAmelCase		,      _UpperCAmelCase)


     def __str__(		self):

          '''simple docstring'''






          return str([self.query(1		,      1		,      self.size		,      _UpperCAmelCase		,      _UpperCAmelCase) for i in range(1		,      self.size + 1)])


if __name__ == "__main__":
  lowercase__       : Union[str, Any]         =       [1, 2, -4, 7, 3, -5, 6, 11, -20, 9, 14, 15, 5, 2, -8]
  lowercase__       : str         =       15
  lowercase__       : List[Any]         =       SegmentTree(size)
  segt.build(1, 1, size, A)
  print(segt.query(1, 1, size, 4, 6))
  print(segt.query(1, 1, size, 7, 11))
  print(segt.query(1, 1, size, 7, 12))
  segt.update(1, 1, size, 1, 3, 1_11)
  print(segt.query(1, 1, size, 1, 15))
  segt.update(1, 1, size, 7, 8, 2_35)
  print(segt)