




'''simple docstring'''






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

from ...utils import logging


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




def 		a_(			lowerCamelCase ,					lowerCamelCase   )					->Optional[Any]:






       '''simple docstring'''




       lowercase	:		List[Any]    =   nn.functional.normalize(lowerCamelCase   )
       lowercase	:		Union[str, Any]    =   nn.functional.normalize(lowerCamelCase   )
       return torch.mm(lowerCamelCase ,					normalized_text_embeds.t()   )





class 	SCREAMING_SNAKE_CASE_		(							lowerCamelCase_					):
 __snake_case						:  Optional[int]       =   CLIPConfig

 __snake_case						:  List[Any]       =   ["CLIPEncoderLayer"]



 def __init__(	self ,		A			):






        '''simple docstring'''




        super().__init__(A			)

        lowercase	:		Dict    =   CLIPVisionModel(config.vision_config			)
        lowercase	:		Union[str, Any]    =   nn.Linear(config.vision_config.hidden_size ,		config.projection_dim ,		bias=A			)

        lowercase	:		Tuple    =   nn.Parameter(torch.ones(17 ,		config.projection_dim			) ,		requires_grad=A			)
        lowercase	:		Union[str, Any]    =   nn.Parameter(torch.ones(3 ,		config.projection_dim			) ,		requires_grad=A			)

        lowercase	:		Optional[Any]    =   nn.Parameter(torch.ones(17			) ,		requires_grad=A			)
        lowercase	:		List[str]    =   nn.Parameter(torch.ones(3			) ,		requires_grad=A			)



 @torch.no_grad()
 def  SCREAMING_SNAKE_CASE(	self ,		A ,		A			):






        '''simple docstring'''




        lowercase	:		Dict    =   self.vision_model(A			)[1]  # pooled_output
        lowercase	:		Optional[Any]    =   self.visual_projection(A			)

        # we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
        lowercase	:		Optional[int]    =   cosine_distance(A ,		self.special_care_embeds			).cpu().float().numpy()
        lowercase	:		Optional[Any]    =   cosine_distance(A ,		self.concept_embeds			).cpu().float().numpy()

        lowercase	:		Union[str, Any]    =   []
        lowercase	:		Optional[Any]    =   image_embeds.shape[0]
        for i in range(A			):
               lowercase	:		List[Any]    =   {'''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
               lowercase	:		Dict    =   0.0

               for concept_idx in range(len(special_cos_dist[0]			)			):
                      lowercase	:		int    =   special_cos_dist[i][concept_idx]
                      lowercase	:		Dict    =   self.special_care_embeds_weights[concept_idx].item()
                      lowercase	:		Optional[Any]    =   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]}			)
                             lowercase	:		str    =   0.0_1

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

               result.append(A			)

        lowercase	:		Dict    =   [len(res['''bad_concepts''']			) > 0 for res in result]

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

        if any(A			):
               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  SCREAMING_SNAKE_CASE(	self ,		A ,		A			):






        '''simple docstring'''




        lowercase	:		Tuple    =   self.vision_model(A			)[1]  # pooled_output
        lowercase	:		Tuple    =   self.visual_projection(A			)

        lowercase	:		str    =   cosine_distance(A ,		self.special_care_embeds			)
        lowercase	:		Optional[Any]    =   cosine_distance(A ,		self.concept_embeds			)

        # increase this value to create a stronger `nsfw` filter
        # at the cost of increasing the possibility of filtering benign images
        lowercase	:		Union[str, Any]    =   0.0

        lowercase	:		int    =   special_cos_dist - self.special_care_embeds_weights + adjustment
        # special_scores = special_scores.round(decimals=3)
        lowercase	:		List[str]    =   torch.any(special_scores > 0 ,		dim=1			)
        lowercase	:		str    =   special_care * 0.0_1
        lowercase	:		Optional[Any]    =   special_adjustment.unsqueeze(1			).expand(-1 ,		cos_dist.shape[1]			)

        lowercase	:		Any    =   (cos_dist - self.concept_embeds_weights) + special_adjustment
        # concept_scores = concept_scores.round(decimals=3)
        lowercase	:		Optional[int]    =   torch.any(concept_scores > 0 ,		dim=1			)

        lowercase	:		Any    =   0.0  # black image

        return images, has_nsfw_concepts










'''simple docstring'''






from manim import *

class 	SCREAMING_SNAKE_CASE_		(							lowerCamelCase_					):



 def  SCREAMING_SNAKE_CASE(	self			):






        '''simple docstring'''




        lowercase	:		str    =   Rectangle(height=0.5 ,		width=0.5			)
        lowercase	:		Optional[Any]    =   Rectangle(height=0.4_6 ,		width=0.4_6			).set_stroke(width=0			)
        lowercase	:		str    =   Rectangle(height=0.2_5 ,		width=0.2_5			)

        lowercase	:		Any    =   [mem.copy() for i in range(6			)]
        lowercase	:		List[str]    =   [mem.copy() for i in range(6			)]
        lowercase	:		List[Any]    =   VGroup(*A			).arrange(A ,		buff=0			)
        lowercase	:		Any    =   VGroup(*A			).arrange(A ,		buff=0			)
        lowercase	:		str    =   VGroup(A ,		A			).arrange(A ,		buff=0			)
        lowercase	:		int    =   Text('''CPU''' ,		font_size=24			)
        lowercase	:		Dict    =   Group(A ,		A			).arrange(A ,		buff=0.5 ,		aligned_edge=A			)
        cpu.move_to([-2.5, -0.5, 0]			)
        self.add(A			)

        lowercase	:		List[str]    =   [mem.copy() for i in range(4			)]
        lowercase	:		Optional[Any]    =   VGroup(*A			).arrange(A ,		buff=0			)
        lowercase	:		Optional[int]    =   Text('''GPU''' ,		font_size=24			)
        lowercase	:		Optional[int]    =   Group(A ,		A			).arrange(A ,		buff=0.5 ,		aligned_edge=A			)
        gpu.move_to([-1, -1, 0]			)
        self.add(A			)

        lowercase	:		str    =   [mem.copy() for i in range(6			)]
        lowercase	:		int    =   VGroup(*A			).arrange(A ,		buff=0			)

        lowercase	:		int    =   Text('''Model''' ,		font_size=24			)
        lowercase	:		Tuple    =   Group(A ,		A			).arrange(A ,		buff=0.5 ,		aligned_edge=A			)
        model.move_to([3, -1.0, 0]			)
        self.add(A			)

        lowercase	:		str    =   []
        lowercase	:		int    =   []

        for i, rect in enumerate(A			):
               rect.set_stroke(A			)

               lowercase	:		List[Any]    =   Rectangle(height=0.4_6 / 4 ,		width=0.4_6 / 3			).set_stroke(width=0.0			).set_fill(A ,		opacity=0.7			)

               if i == 0:
                      cpu_target.next_to(cpu_left_col_base[0].get_corner(DOWN + LEFT			) ,		buff=0.0_2 ,		direction=A			)
                      cpu_target.set_x(cpu_target.get_x() + 0.1			)
               elif i == 3:
                      cpu_target.next_to(model_cpu_arr[0] ,		direction=A ,		buff=0.0			)
               else:
                      cpu_target.next_to(model_cpu_arr[i - 1] ,		direction=A ,		buff=0.0			)
               self.add(A			)
               model_cpu_arr.append(A			)

        self.add(*A ,		*A			)

        lowercase	:		Optional[int]    =   [meta_mem.copy() for i in range(6			)]
        lowercase	:		List[str]    =   [meta_mem.copy() for i in range(6			)]
        lowercase	:		int    =   VGroup(*A			).arrange(A ,		buff=0			)
        lowercase	:		int    =   VGroup(*A			).arrange(A ,		buff=0			)
        lowercase	:		Any    =   VGroup(A ,		A			).arrange(A ,		buff=0			)
        lowercase	:		Union[str, Any]    =   Text('''Disk''' ,		font_size=24			)
        lowercase	:		Dict    =   Group(A ,		A			).arrange(A ,		buff=0.5 ,		aligned_edge=A			)
        disk.move_to([-4.0, -1.2_5, 0]			)
        self.add(A ,		A			)

        lowercase	:		List[str]    =   []

        for i in range(6			):
               lowercase	:		Tuple    =   fill.copy().set_fill(A ,		opacity=0.8			)
               target.move_to(disk_left_col_base[i]			).scale(0.5			)
               cpu_disk_arr.append(A			)

        self.add(*A			)

        lowercase	:		Any    =   Square(side_length=2.2			)
        key.move_to([-5, 2, 0]			)

        lowercase	:		Tuple    =   MarkupText(
            f"""<b>Key:</b>\n\n<span fgcolor='{YELLOW}'>●</span> Empty Model""" ,		font_size=18 ,		)

        key_text.move_to([-5, 2.4, 0]			)

        self.add(A ,		A			)

        lowercase	:		List[Any]    =   MarkupText(
            f"""<span fgcolor='{BLUE}'>●</span> Checkpoint""" ,		font_size=18 ,		)

        blue_text.next_to(A ,		DOWN * 2.4 ,		aligned_edge=key_text.get_left()			)
        self.add(A			)

        lowercase	:		Optional[Any]    =   MarkupText(f"""The offloaded weights are all sent to the CPU.""" ,		font_size=24			)
        step_a.move_to([2, 2, 0]			)

        self.play(Write(A ,		run_time=3			)			)

        for i in range(6			):
               lowercase	:		List[str]    =   cpu_disk_arr[i]
               lowercase	:		Optional[Any]    =   rect.copy().set_fill(A ,		opacity=0.8			).scale(2.0			)
               cpa.generate_target()
               cpa.target.move_to(model_base[i]			)

               if i == 0:
                      rect.set_fill(A ,		opacity=0.8			)
                      rect.generate_target()
                      rect.target.move_to(cpu_left_col_base[0]			).scale(2.0			)

                      self.remove(
                          *A ,		*A ,		)

               else:
                      rect.generate_target()
                      rect.target.move_to(cpu_left_col_base[i]			).scale(2.0			)
               self.play(MoveToTarget(A			) ,		MoveToTarget(A			) ,		model_base[i].animate.set_stroke(A			)			)
        self.play(FadeOut(A			)			)

        lowercase	:		Optional[Any]    =   MarkupText(
            f"""Finally, hooks are added to each weight in the model\nto transfer the weights from CPU to GPU\n\t\tand back when needed.""" ,		font_size=24 ,		)
        step_a.move_to([2, 2, 0]			)

        self.play(Write(A ,		run_time=3			)			)

        lowercase	:		Tuple    =   []
        lowercase	:		List[Any]    =   []
        for i in range(6			):
               lowercase	:		List[str]    =   Arrow(start=A ,		end=A ,		color=A ,		buff=0.5			)
               a.next_to(model_base[i].get_left() ,		A ,		buff=0.2			)
               arrows.append(A			)
               animations.append(Write(A			)			)
        self.play(*A			)
        self.wait()




