



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

from utils import logger




class 					snake_case__						(       lowerCAmelCase_     ):






     """simple docstring"""


     def __init__(       self  :  Tuple, _snake_case  :  str, _snake_case  :  Union[str, Any]					)   ->Tuple:
            snake_case__							:	Union[str, Any]   	=	params

            snake_case__							:	Optional[Any]   	=	np.array(_snake_case					)
            snake_case__							:	Tuple   	=	np.array([len(_snake_case					) 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  :  Tuple, _snake_case  :  List[str]					)   ->str:
            return (self.token_ids[index], self.lengths[index])






     def __len__(       self  :  Union[str, Any]					)   ->str:
            return len(self.lengths					)






     def    lowercase_   (       self  :  int					)   ->Optional[int]:
            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    lowercase_   (       self  :  Dict					)   ->Optional[int]:
            snake_case__							:	Dict   	=	self.params.max_model_input_size
            snake_case__							:	int   	=	self.lengths > max_len
            logger.info(F'''Splitting {sum(_snake_case					)} too long sequences.'''					)

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

            snake_case__							:	Dict   	=	[]
            snake_case__							:	Any   	=	[]
            if self.params.mlm:
                   snake_case__	,				snake_case__							:	str   	=	self.params.special_tok_ids['cls_token'], self.params.special_tok_ids['sep_token']
            else:
                   snake_case__	,				snake_case__							:	Dict   	=	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:
                          snake_case__							:	int   	=	[]
                          for sub_s in divide_chunks(seq_, max_len - 2					):
                                 if sub_s[0] != cls_id:
                                        snake_case__							:	List[Any]   	=	np.insert(_snake_case, 0, _snake_case					)
                                 if sub_s[-1] != sep_id:
                                        snake_case__							:	Optional[Any]   	=	np.insert(_snake_case, len(_snake_case					), _snake_case					)
                                 assert len(_snake_case					) <= max_len
                                 assert (sub_s[0] == cls_id) and (sub_s[-1] == sep_id), sub_s
                                 sub_seqs.append(_snake_case					)

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

            snake_case__							:	int   	=	np.array(_snake_case					)
            snake_case__							:	Union[str, Any]   	=	np.array(_snake_case					)






     def    lowercase_   (       self  :  Optional[int]					)   ->Optional[int]:
            snake_case__							:	int   	=	len(self					)
            snake_case__							:	Dict   	=	self.lengths > 1_1
            snake_case__							:	List[Any]   	=	self.token_ids[indices]
            snake_case__							:	Any   	=	self.lengths[indices]
            snake_case__							:	Dict   	=	len(self					)
            logger.info(F'''Remove {init_size - new_size} too short (<=11 tokens) sequences.'''					)






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






     def    lowercase_   (       self  :  List[str]					)   ->Optional[int]:
            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    lowercase_   (       self  :  Optional[int], _snake_case  :  int					)   ->List[Any]:
            snake_case__							:	Any   	=	[t[0] for t in batch]
            snake_case__							:	Optional[Any]   	=	[t[1] for t in batch]
            assert len(_snake_case					) == len(_snake_case					)

            # Max for paddings
            snake_case__							:	Tuple   	=	max(_snake_case					)

            # Pad token ids
            if self.params.mlm:
                   snake_case__							:	Dict   	=	self.params.special_tok_ids['pad_token']
            else:
                   snake_case__							:	List[Any]   	=	self.params.special_tok_ids['unk_token']
            snake_case__							:	Union[str, Any]   	=	[list(t.astype(_snake_case					)					) + [pad_idx] * (max_seq_len_ - len(_snake_case					)) for t in token_ids]
            assert len(tk_					) == len(_snake_case					)
            assert all(len(_snake_case					) == max_seq_len_ for t in tk_					)

            snake_case__							:	int   	=	torch.tensor(tk_					)  # (bs, max_seq_len_)
            snake_case__							:	Optional[int]   	=	torch.tensor(_snake_case					)  # (bs)
            return tk_t, lg_t











from collections import UserDict
from typing import Union

import numpy as np
import requests

from ..utils import (
    add_end_docstrings,
    logging,
)
from .audio_classification import ffmpeg_read
from .base import PIPELINE_INIT_ARGS, Pipeline


a_   :List[Any]        =      logging.get_logger(__name__)




@add_end_docstrings(lowerCAmelCase_     )
class 					snake_case__						(       lowerCAmelCase_     ):






     """simple docstring"""


     def __init__(       self  :  Optional[Any], **_snake_case  :  str					)   ->Dict:
            super().__init__(**_snake_case					)

            if self.framework != "pt":
                   raise ValueError(F'''The {self.__class__} is only available in PyTorch.'''					)
            # No specific FOR_XXX available yet






     def __call__(       self  :  Union[str, Any], _snake_case  :  Union[np.ndarray, bytes, str], **_snake_case  :  Tuple					)   ->Dict:
            return super().__call__(_snake_case, **_snake_case					)






     def    lowercase_   (       self  :  Tuple, **_snake_case  :  Any					)   ->Union[str, Any]:
            snake_case__							:	str   	=	{}
            if "candidate_labels" in kwargs:
                   snake_case__							:	str   	=	kwargs['candidate_labels']
            if "hypothesis_template" in kwargs:
                   snake_case__							:	str   	=	kwargs['hypothesis_template']

            return preprocess_params, {}, {}






     def    lowercase_   (       self  :  Dict, _snake_case  :  str, _snake_case  :  Optional[int]=None, _snake_case  :  List[str]="This is a sound of {}."					)   ->int:
            if isinstance(_snake_case, _snake_case					):
                   if audio.startswith('http://'					) or audio.startswith('https://'					):
                          # We need to actually check for a real protocol, otherwise it's impossible to use a local file
                          # like http_huggingface_co.png
                          snake_case__							:	List[Any]   	=	requests.get(_snake_case					).content
                   else:
                          with open(_snake_case, 'rb'					) as f:
                                 snake_case__							:	Union[str, Any]   	=	f.read()

            if isinstance(_snake_case, _snake_case					):
                   snake_case__							:	List[Any]   	=	ffmpeg_read(_snake_case, self.feature_extractor.sampling_rate					)

            if not isinstance(_snake_case, np.ndarray					):
                   raise ValueError('We expect a numpy ndarray as input'					)
            if len(audio.shape					) != 1:
                   raise ValueError('We expect a single channel audio input for ZeroShotAudioClassificationPipeline'					)

            snake_case__							:	Tuple   	=	self.feature_extractor(
                [audio], sampling_rate=self.feature_extractor.sampling_rate, return_tensors='pt'					)
            snake_case__							:	int   	=	candidate_labels
            snake_case__							:	int   	=	[hypothesis_template.format(_snake_case					) for x in candidate_labels]
            snake_case__							:	Optional[int]   	=	self.tokenizer(_snake_case, return_tensors=self.framework, padding=_snake_case					)
            snake_case__							:	List[Any]   	=	[text_inputs]
            return inputs






     def    lowercase_   (       self  :  Optional[int], _snake_case  :  Optional[Any]					)   ->int:
            snake_case__							:	Optional[int]   	=	model_inputs.pop('candidate_labels'					)
            snake_case__							:	str   	=	model_inputs.pop('text_inputs'					)
            if isinstance(text_inputs[0], _snake_case					):
                   snake_case__							:	Optional[Any]   	=	text_inputs[0]
            else:
                   # Batching case.
                   snake_case__							:	int   	=	text_inputs[0][0]

            snake_case__							:	Any   	=	self.model(**_snake_case, **_snake_case					)

            snake_case__							:	List[Any]   	=	{
                'candidate_labels': candidate_labels,
                'logits': outputs.logits_per_audio,
            }
            return model_outputs






     def    lowercase_   (       self  :  Union[str, Any], _snake_case  :  str					)   ->List[str]:
            snake_case__							:	int   	=	model_outputs.pop('candidate_labels'					)
            snake_case__							:	List[Any]   	=	model_outputs['logits'][0]

            if self.framework == "pt":
                   snake_case__							:	Tuple   	=	logits.softmax(dim=0					)
                   snake_case__							:	Union[str, Any]   	=	probs.tolist()
            else:
                   raise ValueError('`tf` framework not supported.'					)

            snake_case__							:	Union[str, Any]   	=	[
                {'score': score, 'label': candidate_label}
                for score, candidate_label in sorted(zip(_snake_case, _snake_case					), key=lambda _snake_case					: -x[0]					)
            ]
            return result






