


"""simple docstring"""






import random
import shutil
import tempfile
import unittest

import numpy as np

from transformers import TaTokenizer, TaTokenizerFast
from transformers.testing_utils import require_sentencepiece, require_torch
from transformers.utils.import_utils import is_speech_available, is_torch_available


if is_torch_available():
   pass

if is_speech_available():
   from transformers import EncodecFeatureExtractor, MusicgenProcessor


__snake_case												=							random.Random()
def    A_ ( _lowerCAmelCase  :       str,  _lowerCAmelCase  :       Optional[Any]=1.0,  _lowerCAmelCase  :       int=None,  _lowerCAmelCase  :       Optional[int]=None							):






     """simple docstring"""

     if rng is None:
          _a     						=					global_rng

     _a     						=					[]
     for batch_idx in range(shape[0]							):
          values.append([]							)
          for _ in range(shape[1]							):
               values[-1].append(rng.random() * scale							)

     return values


@require_torch
@require_sentencepiece
class      __lowerCamelCase	(       unittest.TestCase					):

     '''simple docstring'''


     def 				_UpperCAmelCase   (   self				)							->				Tuple:
          _a     						=					'''facebook/musicgen-small'''
          _a     						=					tempfile.mkdtemp()


     def 				_UpperCAmelCase   (   self     ,					**__UpperCAmelCase				)							->				int:
          return TaTokenizer.from_pretrained(self.checkpoint     ,					**__UpperCAmelCase				)


     def 				_UpperCAmelCase   (   self     ,					**__UpperCAmelCase				)							->				List[Any]:
          return EncodecFeatureExtractor.from_pretrained(self.checkpoint     ,					**__UpperCAmelCase				)


     def 				_UpperCAmelCase   (   self				)							->				List[str]:
          shutil.rmtree(self.tmpdirname				)


     def 				_UpperCAmelCase   (   self				)							->				Dict:
          _a     						=					self.get_tokenizer()
          _a     						=					self.get_feature_extractor()

          _a     						=					MusicgenProcessor(tokenizer=__UpperCAmelCase     ,					feature_extractor=__UpperCAmelCase				)

          processor.save_pretrained(self.tmpdirname				)
          _a     						=					MusicgenProcessor.from_pretrained(self.tmpdirname				)

          self.assertEqual(processor.tokenizer.get_vocab()     ,					tokenizer.get_vocab()				)
          self.assertIsInstance(processor.tokenizer     ,					__UpperCAmelCase				)

          self.assertEqual(processor.feature_extractor.to_json_string()     ,					feature_extractor.to_json_string()				)
          self.assertIsInstance(processor.feature_extractor     ,					__UpperCAmelCase				)


     def 				_UpperCAmelCase   (   self				)							->				List[Any]:
          _a     						=					MusicgenProcessor(tokenizer=self.get_tokenizer()     ,					feature_extractor=self.get_feature_extractor()				)
          processor.save_pretrained(self.tmpdirname				)

          _a     						=					self.get_tokenizer(bos_token='''(BOS)'''     ,					eos_token='''(EOS)'''				)
          _a     						=					self.get_feature_extractor(do_normalize=__UpperCAmelCase     ,					padding_value=1.0				)

          _a     						=					MusicgenProcessor.from_pretrained(
              self.tmpdirname     ,					bos_token='''(BOS)'''     ,					eos_token='''(EOS)'''     ,					do_normalize=__UpperCAmelCase     ,					padding_value=1.0				)

          self.assertEqual(processor.tokenizer.get_vocab()     ,					tokenizer_add_kwargs.get_vocab()				)
          self.assertIsInstance(processor.tokenizer     ,					__UpperCAmelCase				)

          self.assertEqual(processor.feature_extractor.to_json_string()     ,					feature_extractor_add_kwargs.to_json_string()				)
          self.assertIsInstance(processor.feature_extractor     ,					__UpperCAmelCase				)


     def 				_UpperCAmelCase   (   self				)							->				Optional[Any]:
          _a     						=					self.get_feature_extractor()
          _a     						=					self.get_tokenizer()

          _a     						=					MusicgenProcessor(tokenizer=__UpperCAmelCase     ,					feature_extractor=__UpperCAmelCase				)

          _a     						=					floats_list((3, 1000)				)

          _a     						=					feature_extractor(__UpperCAmelCase     ,					return_tensors='''np'''				)
          _a     						=					processor(__UpperCAmelCase     ,					return_tensors='''np'''				)

          for key in input_feat_extract.keys():
               self.assertAlmostEqual(input_feat_extract[key].sum()     ,					input_processor[key].sum()     ,					delta=1e-2				)


     def 				_UpperCAmelCase   (   self				)							->				List[str]:
          _a     						=					self.get_feature_extractor()
          _a     						=					self.get_tokenizer()

          _a     						=					MusicgenProcessor(tokenizer=__UpperCAmelCase     ,					feature_extractor=__UpperCAmelCase				)

          _a     						=					'''This is a test string'''

          _a     						=					processor(text=__UpperCAmelCase				)

          _a     						=					tokenizer(__UpperCAmelCase				)

          for key in encoded_tok.keys():
               self.assertListEqual(encoded_tok[key]     ,					encoded_processor[key]				)


     def 				_UpperCAmelCase   (   self				)							->				Any:
          _a     						=					self.get_feature_extractor()
          _a     						=					self.get_tokenizer()

          _a     						=					MusicgenProcessor(tokenizer=__UpperCAmelCase     ,					feature_extractor=__UpperCAmelCase				)

          _a     						=					[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

          _a     						=					processor.batch_decode(sequences=__UpperCAmelCase				)
          _a     						=					tokenizer.batch_decode(__UpperCAmelCase				)

          self.assertListEqual(__UpperCAmelCase     ,					__UpperCAmelCase				)


     def 				_UpperCAmelCase   (   self				)							->				List[str]:
          _a     						=					self.get_feature_extractor()
          _a     						=					self.get_tokenizer()

          _a     						=					MusicgenProcessor(tokenizer=__UpperCAmelCase     ,					feature_extractor=__UpperCAmelCase				)

          self.assertListEqual(
              processor.model_input_names     ,					feature_extractor.model_input_names     ,					msg='''`processor` and `feature_extractor` model input names do not match'''     ,					)





     def 				_UpperCAmelCase   (   self				)							->				int:
          _a     						=					self.get_feature_extractor(padding_side='''left'''				)
          _a     						=					self.get_tokenizer()

          _a     						=					MusicgenProcessor(tokenizer=__UpperCAmelCase     ,					feature_extractor=__UpperCAmelCase				)

          _a     						=					[floats_list((1, x)				)[0] for x in range(5     ,					20     ,					5				)]
          _a     						=					processor(__UpperCAmelCase				).padding_mask

          _a     						=					np.asarray(floats_list((3, 20)				)				)[:, None, :]
          _a     						=					processor.batch_decode(__UpperCAmelCase     ,					padding_mask=__UpperCAmelCase				)

          self.assertIsInstance(__UpperCAmelCase     ,					__UpperCAmelCase				)

          for audio in decoded_audios:
               self.assertIsInstance(__UpperCAmelCase     ,					np.ndarray				)

          self.assertTrue(decoded_audios[0].shape == (1, 10)				)
          self.assertTrue(decoded_audios[1].shape == (1, 15)				)
          self.assertTrue(decoded_audios[2].shape == (1, 20)				)



"""simple docstring"""






import numpy as np
import torch
import torch.nn as nn
from transformers import CLIPConfig, CLIPVisionModel, PreTrainedModel

from ...utils import logging


__snake_case												=							logging.get_logger(__name__)
def    A_ ( _lowerCAmelCase  :       Optional[int],  _lowerCAmelCase  :       Optional[Any]							):






     """simple docstring"""

     _a     						=					nn.functional.normalize(_lowerCAmelCase							)
     _a     						=					nn.functional.normalize(_lowerCAmelCase							)
     return torch.mm(_lowerCAmelCase,  normalized_text_embeds.t()							)


class      __lowerCamelCase	(       a__					):

     '''simple docstring'''
     A_							:						str     =		CLIPConfig

     A_							:						List[str]     =		['CLIPEncoderLayer']


     def __init__(   self     ,					__UpperCAmelCase				)							->				List[Any]:
          super().__init__(__UpperCAmelCase				)

          _a     						=					CLIPVisionModel(config.vision_config				)
          _a     						=					nn.Linear(config.vision_config.hidden_size     ,					config.projection_dim     ,					bias=__UpperCAmelCase				)

          _a     						=					nn.Parameter(torch.ones(17     ,					config.projection_dim				)     ,					requires_grad=__UpperCAmelCase				)
          _a     						=					nn.Parameter(torch.ones(3     ,					config.projection_dim				)     ,					requires_grad=__UpperCAmelCase				)

          _a     						=					nn.Parameter(torch.ones(17				)     ,					requires_grad=__UpperCAmelCase				)
          _a     						=					nn.Parameter(torch.ones(3				)     ,					requires_grad=__UpperCAmelCase				)


     @torch.no_grad()
     def 				_UpperCAmelCase   (   self     ,					__UpperCAmelCase     ,					__UpperCAmelCase				)							->				Optional[Any]:
          _a     						=					self.vision_model(__UpperCAmelCase				)[1]  # pooled_output
          _a     						=					self.visual_projection(__UpperCAmelCase				)

          # we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
          _a     						=					cosine_distance(__UpperCAmelCase     ,					self.special_care_embeds				).cpu().float().numpy()
          _a     						=					cosine_distance(__UpperCAmelCase     ,					self.concept_embeds				).cpu().float().numpy()

          _a     						=					[]
          _a     						=					image_embeds.shape[0]
          for i in range(__UpperCAmelCase				):
               _a     						=					{'''special_scores''': {}, '''special_care''': [], '''concept_scores''': {}, '''bad_concepts''': []}

               # increase this value to create a stronger `nfsw` filter
               # at the cost of increasing the possibility of filtering benign images
               _a     						=					0.0

               for concept_idx in range(len(special_cos_dist[0]				)				):
                    _a     						=					special_cos_dist[i][concept_idx]
                    _a     						=					self.special_care_embeds_weights[concept_idx].item()
                    _a     						=					round(concept_cos - concept_threshold + adjustment     ,					3				)
                    if result_img["special_scores"][concept_idx] > 0:
                         result_img["special_care"].append({concept_idx, result_img['''special_scores'''][concept_idx]}				)
                         _a     						=					0.01

               for concept_idx in range(len(cos_dist[0]				)				):
                    _a     						=					cos_dist[i][concept_idx]
                    _a     						=					self.concept_embeds_weights[concept_idx].item()
                    _a     						=					round(concept_cos - concept_threshold + adjustment     ,					3				)
                    if result_img["concept_scores"][concept_idx] > 0:
                         result_img["bad_concepts"].append(__UpperCAmelCase				)

               result.append(__UpperCAmelCase				)

          _a     						=					[len(res['''bad_concepts''']				) > 0 for res in result]

          for idx, has_nsfw_concept in enumerate(__UpperCAmelCase				):
               if has_nsfw_concept:
                    if torch.is_tensor(__UpperCAmelCase				) or torch.is_tensor(images[0]				):
                         _a     						=					torch.zeros_like(images[idx]				)  # black image
                    else:
                         _a     						=					np.zeros(images[idx].shape				)  # black image

          if any(__UpperCAmelCase				):
               logger.warning(
                   '''Potential NSFW content was detected in one or more images. A black image will be returned instead.'''
                   ''' Try again with a different prompt and/or seed.'''				)

          return images, has_nsfw_concepts





     @torch.no_grad()
     def 				_UpperCAmelCase   (   self     ,					__UpperCAmelCase     ,					__UpperCAmelCase				)							->				Any:
          _a     						=					self.vision_model(__UpperCAmelCase				)[1]  # pooled_output
          _a     						=					self.visual_projection(__UpperCAmelCase				)

          _a     						=					cosine_distance(__UpperCAmelCase     ,					self.special_care_embeds				)
          _a     						=					cosine_distance(__UpperCAmelCase     ,					self.concept_embeds				)

          # increase this value to create a stronger `nsfw` filter
          # at the cost of increasing the possibility of filtering benign images
          _a     						=					0.0

          _a     						=					special_cos_dist - self.special_care_embeds_weights + adjustment
          # special_scores = special_scores.round(decimals=3)
          _a     						=					torch.any(special_scores > 0     ,					dim=1				)
          _a     						=					special_care * 0.01
          _a     						=					special_adjustment.unsqueeze(1				).expand(-1     ,					cos_dist.shape[1]				)

          _a     						=					(cos_dist - self.concept_embeds_weights) + special_adjustment
          # concept_scores = concept_scores.round(decimals=3)
          _a     						=					torch.any(concept_scores > 0     ,					dim=1				)

          _a     						=					0.0  # black image

          return images, has_nsfw_concepts