






import numpy as np
import torch
from torch.utils.data import Dataset

from utils import logger
class 						lowerCamelCase_				(      lowerCamelCase ):







				def __init__(						self       ,					__lowerCAmelCase       ,					__lowerCAmelCase     ):
								"""simple docstring"""
								__magic_name__		:List[Any]        =      params

								__magic_name__		:Union[str, Any]        =      np.array(__lowerCAmelCase     )
								__magic_name__		:List[Any]        =      np.array([len(__lowerCAmelCase     ) for t in data]     )

								self.check()
								self.remove_long_sequences()
								self.remove_empty_sequences()
								self.remove_unknown_sequences()
								self.check()
								self.print_statistics()







				def __getitem__(						self       ,					__lowerCAmelCase     ):
								"""simple docstring"""
								return (self.token_ids[index], self.lengths[index])







				def __len__(						self     ):
								"""simple docstring"""
								return len(self.lengths     )







				def 			A	(						self     ):
								"""simple docstring"""
								assert len(self.token_ids     ) == len(self.lengths     )
								assert all(self.lengths[i] == len(self.token_ids[i]     ) for i in range(len(self.lengths     )     )     )







				def 			A	(						self     ):
								"""simple docstring"""
								__magic_name__		:Union[str, Any]        =      self.params.max_model_input_size
								__magic_name__		:Optional[int]        =      self.lengths > max_len
								logger.info(F'''Splitting {sum(__lowerCAmelCase     )} too long sequences.'''     )

								def divide_chunks(__lowerCAmelCase       ,					__lowerCAmelCase     ):
												return [l[i : i + n] for i in range(0       ,					len(__lowerCAmelCase     )       ,					__lowerCAmelCase     )]

								__magic_name__		:Union[str, Any]        =      []
								__magic_name__		:Union[str, Any]        =      []
								if self.params.mlm:
												__magic_name__		:Dict        =      self.params.special_tok_ids['cls_token'], self.params.special_tok_ids['sep_token']
								else:
												__magic_name__		:int        =      self.params.special_tok_ids['bos_token'], self.params.special_tok_ids['eos_token']

								for seq_, len_ in zip(self.token_ids       ,					self.lengths     ):
												assert (seq_[0] == cls_id) and (seq_[-1] == sep_id), seq_
												if len_ <= max_len:
																new_tok_ids.append(seq_     )
																new_lengths.append(len_     )
												else:
																__magic_name__		:Optional[int]        =      []
																for sub_s in divide_chunks(seq_       ,					max_len - 2     ):
																				if sub_s[0] != cls_id:
																								__magic_name__		:Dict        =      np.insert(__lowerCAmelCase       ,					0       ,					__lowerCAmelCase     )
																				if sub_s[-1] != sep_id:
																								__magic_name__		:Dict        =      np.insert(__lowerCAmelCase       ,					len(__lowerCAmelCase     )       ,					__lowerCAmelCase     )
																				assert len(__lowerCAmelCase     ) <= max_len
																				assert (sub_s[0] == cls_id) and (sub_s[-1] == sep_id), sub_s
																				sub_seqs.append(__lowerCAmelCase     )

																new_tok_ids.extend(__lowerCAmelCase     )
																new_lengths.extend([len(__lowerCAmelCase     ) for l in sub_seqs]     )

								__magic_name__		:str        =      np.array(__lowerCAmelCase     )
								__magic_name__		:Optional[Any]        =      np.array(__lowerCAmelCase     )







				def 			A	(						self     ):
								"""simple docstring"""
								__magic_name__		:Union[str, Any]        =      len(self     )
								__magic_name__		:List[Any]        =      self.lengths > 1_1
								__magic_name__		:Dict        =      self.token_ids[indices]
								__magic_name__		:Tuple        =      self.lengths[indices]
								__magic_name__		:Any        =      len(self     )
								logger.info(F'''Remove {init_size - new_size} too short (<=11 tokens) sequences.'''     )







				def 			A	(						self     ):
								"""simple docstring"""







								if "unk_token" not in self.params.special_tok_ids:
												return
								else:
												__magic_name__		:int        =      self.params.special_tok_ids['unk_token']
								__magic_name__		:str        =      len(self     )
								__magic_name__		:List[str]        =      np.array([np.count_nonzero(a == unk_token_id     ) for a in self.token_ids]     )
								__magic_name__		:int        =      (unk_occs / self.lengths) < 0.5
								__magic_name__		:List[str]        =      self.token_ids[indices]
								__magic_name__		:Optional[Any]        =      self.lengths[indices]
								__magic_name__		:Union[str, Any]        =      len(self     )
								logger.info(F'''Remove {init_size - new_size} sequences with a high level of unknown tokens (50%).'''     )







				def 			A	(						self     ):
								"""simple docstring"""







								if not self.params.is_master:
												return
								logger.info(F'''{len(self     )} sequences'''     )
								# data_len = sum(self.lengths)
								# nb_unique_tokens = len(Counter(list(chain(*self.token_ids))))
								# logger.info(f'{data_len} tokens ({nb_unique_tokens} unique)')

								# unk_idx = self.params.special_tok_ids['unk_token']
								# nb_unknown = sum([(t==unk_idx).sum() for t in self.token_ids])
								# logger.info(f'{nb_unknown} unknown tokens (covering {100*nb_unknown/data_len:.2f}% of the data)')







				def 			A	(						self       ,					__lowerCAmelCase     ):
								"""simple docstring"""
								__magic_name__		:Optional[Any]        =      [t[0] for t in batch]
								__magic_name__		:List[str]        =      [t[1] for t in batch]
								assert len(__lowerCAmelCase     ) == len(__lowerCAmelCase     )

								# Max for paddings
								__magic_name__		:List[str]        =      max(__lowerCAmelCase     )

								# Pad token ids
								if self.params.mlm:
												__magic_name__		:str        =      self.params.special_tok_ids['pad_token']
								else:
												__magic_name__		:Optional[int]        =      self.params.special_tok_ids['unk_token']
								__magic_name__		:Tuple        =      [list(t.astype(__lowerCAmelCase     )     ) + [pad_idx] * (max_seq_len_ - len(__lowerCAmelCase     )) for t in token_ids]
								assert len(tk_     ) == len(__lowerCAmelCase     )
								assert all(len(__lowerCAmelCase     ) == max_seq_len_ for t in tk_     )

								__magic_name__		:Union[str, Any]        =      torch.tensor(tk_     )  # (bs, max_seq_len_)
								__magic_name__		:List[str]        =      torch.tensor(__lowerCAmelCase     )  # (bs)
								return tk_t, lg_t














from ...processing_utils import ProcessorMixin
class 						lowerCamelCase_				(      lowerCamelCase ):
				a__					       =  '''WhisperFeatureExtractor'''
				a__					       =  '''WhisperTokenizer'''







				def __init__(						self       ,					__lowerCAmelCase       ,					__lowerCAmelCase     ):
								"""simple docstring"""
								super().__init__(__lowerCAmelCase       ,					__lowerCAmelCase     )
								__magic_name__		:int        =      self.feature_extractor
								__magic_name__		:int        =      False







				def 			A	(						self       ,					__lowerCAmelCase=None       ,					__lowerCAmelCase=None       ,					__lowerCAmelCase=True     ):
								"""simple docstring"""
								return self.tokenizer.get_decoder_prompt_ids(task=__lowerCAmelCase       ,					language=__lowerCAmelCase       ,					no_timestamps=__lowerCAmelCase     )







				def __call__(						self       ,					*__lowerCAmelCase       ,					**__lowerCAmelCase     ):
								"""simple docstring"""
								# For backward compatibility
								if self._in_target_context_manager:
												return self.current_processor(*__lowerCAmelCase       ,					**__lowerCAmelCase     )

								__magic_name__		:List[Any]        =      kwargs.pop('''audio'''       ,					__lowerCAmelCase     )
								__magic_name__		:Any        =      kwargs.pop('''sampling_rate'''       ,					__lowerCAmelCase     )
								__magic_name__		:str        =      kwargs.pop('''text'''       ,					__lowerCAmelCase     )
								if len(__lowerCAmelCase     ) > 0:
												__magic_name__		:Any        =      args[0]
												__magic_name__		:Union[str, Any]        =      args[1:]

								if audio is None and text is None:
												raise ValueError('''You need to specify either an `audio` or `text` input to process.'''     )

								if audio is not None:
												__magic_name__		:Optional[int]        =      self.feature_extractor(__lowerCAmelCase       ,					*__lowerCAmelCase       ,					sampling_rate=__lowerCAmelCase       ,					**__lowerCAmelCase     )
								if text is not None:
												__magic_name__		:Optional[int]        =      self.tokenizer(__lowerCAmelCase       ,					**__lowerCAmelCase     )

								if text is None:
												return inputs

								elif audio is None:
												return encodings
								else:
												__magic_name__		:str        =      encodings['''input_ids''']
												return inputs







				def 			A	(						self       ,					*__lowerCAmelCase       ,					**__lowerCAmelCase     ):
								"""simple docstring"""
								return self.tokenizer.batch_decode(*__lowerCAmelCase       ,					**__lowerCAmelCase     )







				def 			A	(						self       ,					*__lowerCAmelCase       ,					**__lowerCAmelCase     ):
								"""simple docstring"""
								return self.tokenizer.decode(*__lowerCAmelCase       ,					**__lowerCAmelCase     )







				def 			A	(						self       ,					__lowerCAmelCase       ,					__lowerCAmelCase="np"     ):
								"""simple docstring"""
								return self.tokenizer.get_prompt_ids(__lowerCAmelCase       ,					return_tensors=__lowerCAmelCase     )






