




"""simple docstring"""







from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


SCREAMING_SNAKE_CASE_      =					logging.get_logger(__name__)

class 							snake_case_ (				lowerCamelCase_ ):







   """simple docstring"""




   A_										=					['''input_values''', '''attention_mask''']







   def __init__(		self					,				lowerCamelCase_=1					,				lowerCamelCase_=1_6_0_0_0					,				lowerCamelCase_=1_2_8					,				lowerCamelCase_=1_0_2_4					,				lowerCamelCase_=0.0					,				lowerCamelCase_=True					,				lowerCamelCase_=-4.267_7393					,				lowerCamelCase_=4.568_9974					,				lowerCamelCase_=False					,				**lowerCamelCase_					,				)							->      int:
       super().__init__(feature_size=A_					,				sampling_rate=A_					,				padding_value=A_					,				**A_)
       UpperCamelCase										=  num_mel_bins
       UpperCamelCase										=  max_length
       UpperCamelCase										=  do_normalize
       UpperCamelCase										=  mean
       UpperCamelCase										=  std
       UpperCamelCase										=  return_attention_mask







   def      UpperCAmelCase__ (		self					,				lowerCamelCase_					,				lowerCamelCase_					,				)							->      Union[str, Any]:
       UpperCamelCase										=  torch.from_numpy(A_).unsqueeze(0)
       UpperCamelCase										=  ta_kaldi.fbank(
           A_					,				htk_compat=A_					,				sample_frequency=self.sampling_rate					,				use_energy=A_					,				window_type='''hanning'''					,				num_mel_bins=self.num_mel_bins					,				dither=0.0					,				frame_shift=1_0					,				)

       UpperCamelCase										=  fbank.shape[0]
       UpperCamelCase										=  max_length - n_frames

       # pad or truncate, depending on difference
       if difference > 0:
           UpperCamelCase										=  torch.nn.ZeroPadad((0, 0, 0, difference))
           UpperCamelCase										=  pad_module(A_)
       elif difference < 0:
           UpperCamelCase										=  fbank[0:max_length, :]

       UpperCamelCase										=  fbank.numpy()

       return fbank







   def      UpperCAmelCase__ (		self					,				lowerCamelCase_)							->      List[Any]:
       return (input_values - (self.mean)) / (self.std * 2)







   def __call__(		self					,				lowerCamelCase_					,				lowerCamelCase_ = None					,				lowerCamelCase_ = None					,				**lowerCamelCase_					,				)							->      str:

       if sampling_rate is not None:
           if sampling_rate != self.sampling_rate:
               raise ValueError(
                   F'The model corresponding to this feature extractor: {self} was trained using a sampling rate of'
                   F' {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with'
                   F' {self.sampling_rate} and not {sampling_rate}.')
       else:
           logger.warning(
               '''It is strongly recommended to pass the `sampling_rate` argument to this function. '''
               '''Failing to do so can result in silent errors that might be hard to debug.''')

       UpperCamelCase										=  isinstance(A_					,				np.ndarray) and len(raw_speech.shape) > 1
       if is_batched_numpy and len(raw_speech.shape) > 2:
           raise ValueError(F'Only mono-channel audio is supported for input to {self}')
       UpperCamelCase										=  is_batched_numpy or (
           isinstance(A_					,				(list, tuple)) and (isinstance(raw_speech[0]					,				(np.ndarray, tuple, list)))
       )

       if is_batched:
           UpperCamelCase										=  [np.asarray(A_					,				dtype=np.floataa) for speech in raw_speech]
       elif not is_batched and not isinstance(A_					,				np.ndarray):
           UpperCamelCase										=  np.asarray(A_					,				dtype=np.floataa)
       elif isinstance(A_					,				np.ndarray) and raw_speech.dtype is np.dtype(np.floataa):
           UpperCamelCase										=  raw_speech.astype(np.floataa)

       # always return batch
       if not is_batched:
           UpperCamelCase										=  [raw_speech]

       # extract fbank features and pad/truncate to max_length
       UpperCamelCase										=  [self._extract_fbank_features(A_					,				max_length=self.max_length) for waveform in raw_speech]

       # convert into BatchFeature
       UpperCamelCase										=  BatchFeature({'''input_values''': features})

       # make sure list is in array format
       UpperCamelCase										=  padded_inputs.get('''input_values''')
       if isinstance(input_values[0]					,				A_):
           UpperCamelCase										=  [np.asarray(A_					,				dtype=np.floataa) for feature in input_values]

       # normalization
       if self.do_normalize:
           UpperCamelCase										=  [self.normalize(A_) for feature in input_values]

       if return_tensors is not None:
           UpperCamelCase										=  padded_inputs.convert_to_tensors(A_)

       return padded_inputs





"""simple docstring"""







import os
import sys
from pathlib import Path

import pytest

from datasets import Dataset, IterableDataset
from datasets.distributed import split_dataset_by_node

from .utils import execute_subprocess_async, get_torch_dist_unique_port, require_torch





def     __snake_case   (							):


    """simple docstring"""







    UpperCamelCase										=  Dataset.from_dict({'''i''': range(17						)}						)
    UpperCamelCase										=  len(_lowercase						)
    UpperCamelCase										=  3
    UpperCamelCase										=  [
        split_dataset_by_node(_lowercase ,rank=_lowercase ,world_size=_lowercase						) for rank in range(_lowercase						)
    ]
    assert sum(len(_lowercase						) for ds in datasets_per_rank						) == full_size
    assert len({tuple(x.values()						) for ds in datasets_per_rank for x in ds}						) == full_size





def     __snake_case   (							):


    """simple docstring"""







    def gen():
        return ({"i": i} for i in range(17						))

    UpperCamelCase										=  3
    UpperCamelCase										=  IterableDataset.from_generator(_lowercase						)
    UpperCamelCase										=  len(list(_lowercase						)						)
    UpperCamelCase										=  [
        split_dataset_by_node(_lowercase ,rank=_lowercase ,world_size=_lowercase						) for rank in range(_lowercase						)
    ]
    assert sum(len(list(_lowercase						)						) for ds in datasets_per_rank						) == full_size
    assert len({tuple(x.values()						) for ds in datasets_per_rank for x in ds}						) == full_size





@pytest.mark.parametrize('''shards_per_node''' ,[1, 2, 3]						)
def     __snake_case   (							_lowercase						):


    """simple docstring"""







    def gen(_lowercase						):
        for shard in shards:
            yield from ({"i": i, "shard": shard} for i in range(17						))

    UpperCamelCase										=  3
    UpperCamelCase										=  shards_per_node * world_size
    UpperCamelCase										=  {'''shards''': [f'shard_{shard_idx}.txt' for shard_idx in range(_lowercase						)]}
    UpperCamelCase										=  IterableDataset.from_generator(_lowercase ,gen_kwargs=_lowercase						)
    UpperCamelCase										=  len(list(_lowercase						)						)
    assert full_ds.n_shards == world_size * shards_per_node
    UpperCamelCase										=  [
        split_dataset_by_node(_lowercase ,rank=_lowercase ,world_size=_lowercase						) for rank in range(_lowercase						)
    ]
    assert [ds.n_shards for ds in datasets_per_rank] == [shards_per_node] * world_size
    assert sum(len(list(_lowercase						)						) for ds in datasets_per_rank						) == full_size
    assert len({tuple(x.values()						) for ds in datasets_per_rank for x in ds}						) == full_size





def     __snake_case   (							):


    """simple docstring"""







    def gen():
        return ({"i": i} for i in range(17						))

    UpperCamelCase										=  2
    UpperCamelCase										=  IterableDataset.from_generator(_lowercase						)
    UpperCamelCase										=  len(list(_lowercase						)						)

    UpperCamelCase										=  split_dataset_by_node(_lowercase ,rank=0 ,world_size=_lowercase						).shuffle(seed=42						)
    assert len(list(_lowercase						)						) == 1 + full_size // world_size
    with pytest.raises(_lowercase						):
        split_dataset_by_node(_lowercase ,rank=0 ,world_size=_lowercase						).shuffle()

    UpperCamelCase										=  split_dataset_by_node(full_ds.shuffle(seed=42						) ,rank=0 ,world_size=_lowercase						)
    assert len(list(_lowercase						)						) == 1 + full_size // world_size
    with pytest.raises(_lowercase						):
        split_dataset_by_node(full_ds.shuffle() ,rank=0 ,world_size=_lowercase						)





@pytest.mark.parametrize('''streaming''' ,[False, True]						)
@require_torch
@pytest.mark.skipif(os.name == '''nt''' ,reason='''execute_subprocess_async doesn\'t support windows'''						)
@pytest.mark.integration
def     __snake_case   (							_lowercase						):


    """simple docstring"""







    UpperCamelCase										=  2
    UpperCamelCase										=  get_torch_dist_unique_port()
    UpperCamelCase										=  Path(_lowercase						).resolve().parent / '''distributed_scripts''' / '''launch_torch_distributed.py'''
    UpperCamelCase										=  f'\n        -m torch.distributed.launch\n        --nproc_per_node={nproc_per_node}\n        --master_port={master_port}\n        {test_script}\n    '.split()
    UpperCamelCase										=  f'\n        --streaming={streaming}\n    '.split()
    UpperCamelCase										=  [sys.executable] + distributed_args + args
    execute_subprocess_async(_lowercase ,env=os.environ.copy()						)





@pytest.mark.parametrize(
    '''nproc_per_node, num_workers''' ,[
        (2, 2),  # each node has 2 shards and each worker has 1 shards
        (3, 2),  # each node uses all the shards but skips examples, and each worker has 2 shards
    ] ,)
@require_torch
@pytest.mark.skipif(os.name == '''nt''' ,reason='''execute_subprocess_async doesn\'t support windows'''						)
@pytest.mark.integration
def     __snake_case   (							_lowercase ,_lowercase						):


    """simple docstring"""







    UpperCamelCase										=  True
    UpperCamelCase										=  get_torch_dist_unique_port()
    UpperCamelCase										=  Path(_lowercase						).resolve().parent / '''distributed_scripts''' / '''launch_torch_distributed.py'''
    UpperCamelCase										=  f'\n        -m torch.distributed.launch\n        --nproc_per_node={nproc_per_node}\n        --master_port={master_port}\n        {test_script}\n    '.split()
    UpperCamelCase										=  f'\n        --streaming={streaming}\n        --num_workers={num_workers}\n    '.split()
    UpperCamelCase										=  [sys.executable] + distributed_args + args
    execute_subprocess_async(_lowercase ,env=os.environ.copy()						)