



"""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 PaddingStrategy, TensorType, logging


lowercase__    :  str							    =	logging.get_logger(__name__)




class        _UpperCAmelCase       (			lowerCAmelCase__):
 _lowerCAmelCase   :			Dict							=		["""input_features""", """attention_mask"""]

 def __init__(     self		:       Optional[int]  ,     lowercase_		:       List[str]=80  ,     lowercase_		:       Optional[int]=16000  ,     lowercase_		:       Dict=80  ,     lowercase_		:       Dict=0.0  ,     lowercase_		:       Dict=True  ,     lowercase_		:       List[str]=True  ,     lowercase_		:       Optional[Any]=True  ,     **lowercase_		:       Any  ,     ):
     super().__init__(feature_size=lowercase_  ,     sampling_rate=lowercase_  ,     padding_value=lowercase_  ,     **lowercase_				)
     snake_case_ :     Optional[Any]				=	num_mel_bins
     snake_case_ :     Tuple				=	do_ceptral_normalize
     snake_case_ :     str				=	normalize_means
     snake_case_ :     List[str]				=	normalize_vars
     snake_case_ :     Optional[int]				=	True

 def 						_snake_case					(     self		:       Any  ,     lowercase_		:       np.ndarray  ,     ):
     snake_case_ :     Dict				=	waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
     snake_case_ :     str				=	torch.from_numpy(lowercase_				).unsqueeze(0				)
     snake_case_ :     Any				=	ta_kaldi.fbank(lowercase_  ,     num_mel_bins=self.num_mel_bins  ,     sample_frequency=self.sampling_rate				)
     return features.numpy()

 @staticmethod
 def 						_snake_case					(     lowercase_		:       np.ndarray  ,     lowercase_		:       int  ,     lowercase_		:       Optional[bool] = True  ,     lowercase_		:       Optional[bool] = True  ,     lowercase_		:       float = 0.0  ,     ):
     # make sure we normalize float32 arrays
     if normalize_means:
         snake_case_ :     Dict				=	x[:input_length].mean(axis=0				)
         snake_case_ :     List[Any]				=	np.subtract(lowercase_  ,     lowercase_				)
     if normalize_vars:
         snake_case_ :     List[str]				=	x[:input_length].std(axis=0				)
         snake_case_ :     List[str]				=	np.divide(lowercase_  ,     lowercase_				)

     if input_length < x.shape[0]:
         snake_case_ :     Union[str, Any]				=	padding_value

     # make sure array is in float32
     snake_case_ :     int				=	x.astype(np.floataa				)

     return x

 def 						_snake_case					(     self		:       List[Any]  ,     lowercase_		:       List[np.ndarray]  ,     lowercase_		:       Optional[np.ndarray] = None				):
     snake_case_ :     Any				=	attention_mask.sum(-1				) if attention_mask is not None else [x.shape[0] for x in input_features]
     return [
         self.utterance_cmvn(lowercase_  ,     lowercase_  ,     self.normalize_means  ,     self.normalize_vars  ,     self.padding_value				)
         for x, n in zip(lowercase_  ,     lowercase_				)
     ]






 def __call__(     self		:       int  ,     lowercase_		:       Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]  ,     lowercase_		:       Union[bool, str, PaddingStrategy] = False  ,     lowercase_		:       Optional[int] = None  ,     lowercase_		:       bool = False  ,     lowercase_		:       Optional[int] = None  ,     lowercase_		:       Optional[Union[str, TensorType]] = None  ,     lowercase_		:       Optional[int] = None  ,     lowercase_		:       Optional[bool] = None  ,     **lowercase_		:       Optional[int]  ,     ):

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

     snake_case_ :     Any				=	isinstance(lowercase_  ,     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}"				)
     snake_case_ :     str				=	is_batched_numpy or (
         isinstance(lowercase_  ,     (list, tuple)				) and (isinstance(raw_speech[0]  ,     (np.ndarray, tuple, list)				))
     )

     if is_batched:
         snake_case_ :     Optional[int]				=	[np.asarray(lowercase_  ,     dtype=np.floataa				) for speech in raw_speech]
     elif not is_batched and not isinstance(lowercase_  ,     np.ndarray				):
         snake_case_ :     Dict				=	np.asarray(lowercase_  ,     dtype=np.floataa				)
     elif isinstance(lowercase_  ,     np.ndarray				) and raw_speech.dtype is np.dtype(np.floataa				):
         snake_case_ :     List[str]				=	raw_speech.astype(np.floataa				)

     # always return batch
     if not is_batched:
         snake_case_ :     Optional[int]				=	[raw_speech]

     # extract fbank features
     snake_case_ :     Union[str, Any]				=	[self._extract_fbank_features(lowercase_				) for waveform in raw_speech]

     # convert into correct format for padding
     snake_case_ :     Tuple				=	BatchFeature({'''input_features''': features}				)

     snake_case_ :     Optional[Any]				=	self.pad(
         lowercase_  ,     padding=lowercase_  ,     max_length=lowercase_  ,     truncation=lowercase_  ,     pad_to_multiple_of=lowercase_  ,     return_attention_mask=lowercase_  ,     **lowercase_  ,     )

     # make sure list is in array format
     snake_case_ :     Tuple				=	padded_inputs.get('''input_features'''				)
     if isinstance(input_features[0]  ,     lowercase_				):
         snake_case_ :     Optional[Any]				=	[np.asarray(lowercase_  ,     dtype=np.floataa				) for feature in input_features]

     snake_case_ :     Union[str, Any]				=	padded_inputs.get('''attention_mask'''				)
     if attention_mask is not None:
         snake_case_ :     List[str]				=	[np.asarray(lowercase_  ,     dtype=np.intaa				) for array in attention_mask]

     # Utterance-level cepstral mean and variance normalization
     if self.do_ceptral_normalize:
         snake_case_ :     Any				=	(
             np.array(lowercase_  ,     dtype=np.intaa				)
             if self._get_padding_strategies(lowercase_  ,     max_length=lowercase_				) is not PaddingStrategy.DO_NOT_PAD
             else None
         )
         snake_case_ :     List[str]				=	self.normalize(
             padded_inputs['''input_features''']  ,     attention_mask=lowercase_				)

     if return_tensors is not None:
         snake_case_ :     List[Any]				=	padded_inputs.convert_to_tensors(lowercase_				)

     return padded_inputs










"""simple docstring"""

import os
from datetime import datetime as dt

from github import Github


lowercase__    :  int							    =	[
    '''good first issue''',
    '''good second issue''',
    '''good difficult issue''',
    '''enhancement''',
    '''new pipeline/model''',
    '''new scheduler''',
    '''wip''',
]




def __lowercase			(							):
    snake_case_ :     Optional[Any]				=	Github(os.environ['''GITHUB_TOKEN''']    )
    snake_case_ :     Any				=	g.get_repo('''huggingface/diffusers'''    )
    snake_case_ :     Any				=	repo.get_issues(state='''open'''    )

    for issue in open_issues:
        snake_case_ :     str				=	sorted(issue.get_comments()		,			key=lambda _a    : i.created_at		,			reverse=_a    )
        snake_case_ :     Dict				=	comments[0] if len(_a    ) > 0 else None
        if (
            last_comment is not None
            and last_comment.user.login == "github-actions[bot]"
            and (dt.utcnow() - issue.updated_at).days > 7
            and (dt.utcnow() - issue.created_at).days >= 30
            and not any(label.name.lower() in LABELS_TO_EXEMPT for label in issue.get_labels()    )
        ):
            # Closes the issue after 7 days of inactivity since the Stalebot notification.
            issue.edit(state='''closed'''    )
        elif (
            "stale" in issue.get_labels()
            and last_comment is not None
            and last_comment.user.login != "github-actions[bot]"
        ):
            # Opens the issue if someone other than Stalebot commented.
            issue.edit(state='''open'''    )
            issue.remove_from_labels('''stale'''    )
        elif (
            (dt.utcnow() - issue.updated_at).days > 23
            and (dt.utcnow() - issue.created_at).days >= 30
            and not any(label.name.lower() in LABELS_TO_EXEMPT for label in issue.get_labels()    )
        ):
            # Post a Stalebot notification after 23 days of inactivity.
            issue.create_comment(
                '''This issue has been automatically marked as stale because it has not had '''
                '''recent activity. If you think this still needs to be addressed '''
                '''please comment on this thread.\n\nPlease note that issues that do not follow the '''
                '''[contributing guidelines](https://github.com/huggingface/diffusers/blob/main/CONTRIBUTING.md) '''
                '''are likely to be ignored.'''    )
            issue.add_to_labels('''stale'''    )


if __name__ == "__main__":
     main()





