code: 
  # Copyright (c) 2019 Uber Technologies, Inc.
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  # http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  import argparse
  import csv
  import json
  import math
  import time
  
  import numpy as np
  import torch
  import torch.optim as optim
  import torch.utils.data as data
  from nltk.tokenize.treebank import TreebankWordDetokenizer
  from pplm_classification_head import ClassificationHead
  from torch import nn
  from torchtext import data as torchtext_data
  from torchtext import datasets
  from tqdm import tqdm, trange
  
  from transformers import GPTaLMHeadModel, GPTaTokenizer
  
  
  torch.manual_seed(0)
  np.random.seed(0)
  UpperCamelCase_	:    Optional[Any]  					=       1e-10
  UpperCamelCase_	:    str  					=       """This is incredible! I love it, this is the best chicken I have ever had."""
  UpperCamelCase_	:    Optional[int]  					=       100
  
  
  
  
  class      __lowercase	(	nn.Module					):
  
  
  
  
  
  
  	def __init__(self							:   List[Any]			,  snake_case							:   int			,  snake_case							:   Optional[Any]="gpt2-medium"			,  snake_case							:   Dict=False			,  snake_case							:   int="cpu"						)   -> Dict:
  							super().__init__()
  							_lowercase					:  Optional[Any]    			=      GPTaTokenizer.from_pretrained(snake_case						)
  							_lowercase					:  Optional[Any]    			=      GPTaLMHeadModel.from_pretrained(snake_case						)
  							_lowercase					:  Optional[int]    			=      self.encoder.transformer.config.hidden_size
  							_lowercase					:  Optional[Any]    			=      ClassificationHead(class_size=snake_case			,  embed_size=self.embed_size						)
  							_lowercase					:  Dict    			=      cached_mode
  							_lowercase					:  List[str]    			=      device
  
  
  
  
  
  
  	def 			_a(self							:   str						)   -> Union[str, Any]:
  							return self.classifier_head
  
  
  
  
  
  
  	def 			_a(self							:   Optional[Any]						)   -> List[str]:
  							for param in self.encoder.parameters():
  													_lowercase					:  List[str]    			=      False
  							self.classifier_head.train()
  
  
  
  
  
  
  	def 			_a(self							:   Dict			,  snake_case							:   Dict						)   -> int:
  							_lowercase					:  List[Any]    			=      x.ne(0						).unsqueeze(2						).repeat(1			,  1			,  self.embed_size						).float().to(self.device						).detach()
  							_lowercase					:  Union[str, Any]    			=      self.encoder.transformer(snake_case						)["last_hidden_state"]
  							_lowercase					:  List[Any]    			=      hidden * mask
  							_lowercase					:  List[str]    			=      torch.sum(snake_case			,  dim=1						) / (torch.sum(snake_case			,  dim=1						).detach() + EPSILON)
  							return avg_hidden
  
  
  
  
  
  
  	def 			_a(self							:   Any			,  snake_case							:   Union[str, Any]						)   -> Tuple:
  							if self.cached_mode:
  													_lowercase					:  str    			=      x.to(self.device						)
  							else:
  													_lowercase					:  Dict    			=      self.avg_representation(x.to(self.device						)						)
  
  							_lowercase					:  str    			=      self.classifier_head(snake_case						)
  							_lowercase					:  List[str]    			=      nn.functional.log_softmax(snake_case			,  dim=-1						)
  
  							return probs
  
  
  
  
  class      __lowercase	(	data.Dataset					):
  
  
  
  
  
  
  	def __init__(self							:   Optional[Any]			,  snake_case							:   Optional[int]			,  snake_case							:   Any						)   -> List[str]:
  							_lowercase					:  List[Any]    			=      X
  							_lowercase					:  Tuple    			=      y
  
  
  
  
  
  
  	def __len__(self							:   Dict						)   -> Tuple:
  							return len(self.X						)
  
  
  
  
  
  
  	def __getitem__(self							:   List[Any]			,  snake_case							:   Optional[int]						)   -> Optional[int]:
  							_lowercase					:  Optional[Any]    			=      {}
  							_lowercase					:  Optional[Any]    			=      self.X[index]
  							_lowercase					:  Any    			=      self.y[index]
  							return data
  
  
  
  def 		UpperCamelCase    (					_UpperCAmelCase     :      str     )      -> int:
  
  
  
  
  
  
  						'''simple docstring'''
  
  						def pad_sequences(_UpperCAmelCase     :      Optional[int]     ):
  												_lowercase					:  Union[str, Any]    			=      [len(_UpperCAmelCase     ) for seq in sequences]
  
  												_lowercase					:  Dict    			=      torch.zeros(len(_UpperCAmelCase     )   , max(_UpperCAmelCase     )     ).long()  # padding value = 0
  
  												for i, seq in enumerate(_UpperCAmelCase     ):
  																		_lowercase					:  List[str]    			=      lengths[i]
  																		_lowercase					:  Optional[int]    			=      seq[:end]
  
  												return padded_sequences, lengths
  
  						_lowercase					:  Dict    			=      {}
  						for key in data[0].keys():
  												_lowercase					:  Any    			=      [d[key] for d in data]
  
  						_lowercase			,						_lowercase					:  Tuple    			=      pad_sequences(item_info["X"]     )
  						_lowercase					:  Tuple    			=      torch.tensor(item_info["y"]   , dtype=torch.long     )
  
  						return x_batch, y_batch
  
  
  
  def 		UpperCamelCase    (					_UpperCAmelCase     :      Dict     )      -> List[str]:
  
  
  
  
  
  
  						'''simple docstring'''
  
  						_lowercase					:  Tuple    			=      {}
  						for key in data[0].keys():
  												_lowercase					:  Union[str, Any]    			=      [d[key] for d in data]
  
  						_lowercase					:  int    			=      torch.cat(item_info["X"]   , 0     )
  						_lowercase					:  List[Any]    			=      torch.tensor(item_info["y"]   , dtype=torch.long     )
  
  						return x_batch, y_batch
  
  
  
  def 		UpperCamelCase    (					_UpperCAmelCase     :      Any   , _UpperCAmelCase     :      int   , _UpperCAmelCase     :      Union[str, Any]   , _UpperCAmelCase     :      str=0   , _UpperCAmelCase     :      int=10   , _UpperCAmelCase     :      Optional[Any]="cpu"     )      -> Dict:
  
  
  
  
  
  
  						'''simple docstring'''
  
  						_lowercase					:  List[Any]    			=      0
  						discriminator.train_custom()
  						for batch_idx, (input_t, target_t) in enumerate(_UpperCAmelCase     ):
  												_lowercase			,						_lowercase					:  Any    			=      input_t.to(_UpperCAmelCase     ), target_t.to(_UpperCAmelCase     )
  
  												optimizer.zero_grad()
  
  												_lowercase					:  Optional[int]    			=      discriminator(_UpperCAmelCase     )
  												_lowercase					:  Optional[int]    			=      nn.functional.nll_loss(_UpperCAmelCase   , _UpperCAmelCase     )
  												loss.backward(retain_graph=_UpperCAmelCase     )
  												optimizer.step()
  
  												samples_so_far += len(_UpperCAmelCase     )
  
  												if batch_idx % log_interval == 0:
  																		print(
  																		    "Train Epoch: {} [{}/{} ({:.0f}%)]\tLoss: {:.6f}".format(
  																		        epoch + 1   , _UpperCAmelCase   , len(data_loader.dataset     )   , 100 * samples_so_far / len(data_loader.dataset     )   , loss.item()   , )     )
  
  
  
  def 		UpperCamelCase    (					_UpperCAmelCase     :      str   , _UpperCAmelCase     :      Optional[int]   , _UpperCAmelCase     :      List[Any]="cpu"     )      -> List[str]:
  
  
  
  
  
  
  						'''simple docstring'''
  
  						discriminator.eval()
  						_lowercase					:  str    			=      0
  						_lowercase					:  Optional[int]    			=      0
  						with torch.no_grad():
  												for input_t, target_t in data_loader:
  																		_lowercase			,						_lowercase					:  Dict    			=      input_t.to(_UpperCAmelCase     ), target_t.to(_UpperCAmelCase     )
  																		_lowercase					:  Union[str, Any]    			=      discriminator(_UpperCAmelCase     )
  																		# sum up batch loss
  																		test_loss += nn.functional.nll_loss(_UpperCAmelCase   , _UpperCAmelCase   , reduction="sum"     ).item()
  																		# get the index of the max log-probability
  																		_lowercase					:  List[Any]    			=      output_t.argmax(dim=1   , keepdim=_UpperCAmelCase     )
  																		correct += pred_t.eq(target_t.view_as(_UpperCAmelCase     )     ).sum().item()
  
  						test_loss /= len(data_loader.dataset     )
  
  						print(
  						    "Performance on test set: Average loss: {:.4f}, Accuracy: {}/{} ({:.0f}%)".format(
  						        _UpperCAmelCase   , _UpperCAmelCase   , len(data_loader.dataset     )   , 100.0 * correct / len(data_loader.dataset     )     )     )
  
  
  
  def 		UpperCamelCase    (					_UpperCAmelCase     :      Optional[Any]   , _UpperCAmelCase     :      Tuple   , _UpperCAmelCase     :      int   , _UpperCAmelCase     :      str=False   , _UpperCAmelCase     :      str="cpu"     )      -> Union[str, Any]:
  
  
  
  
  
  
  						'''simple docstring'''
  
  						_lowercase					:  Optional[Any]    			=      model.tokenizer.encode(_UpperCAmelCase     )
  						_lowercase					:  str    			=      torch.tensor([input_t]   , dtype=torch.long   , device=_UpperCAmelCase     )
  						if cached:
  												_lowercase					:  Any    			=      model.avg_representation(_UpperCAmelCase     )
  
  						_lowercase					:  List[Any]    			=      model(_UpperCAmelCase     ).data.cpu().numpy().flatten().tolist()
  						print("Input sentence:"   , _UpperCAmelCase     )
  						print(
  						    "Predictions:"   , ", ".join("{}: {:.4f}".format(_UpperCAmelCase   , math.exp(_UpperCAmelCase     )     ) for c, log_prob in zip(_UpperCAmelCase   , _UpperCAmelCase     )     )   , )
  
  
  
  def 		UpperCamelCase    (					_UpperCAmelCase     :      Optional[int]   , _UpperCAmelCase     :      Optional[int]   , _UpperCAmelCase     :      Optional[int]   , _UpperCAmelCase     :      Optional[Any]=False   , _UpperCAmelCase     :      Dict="cpu"     )      -> List[Any]:
  
  
  
  
  
  
  						'''simple docstring'''
  
  						_lowercase					:  List[str]    			=      torch.utils.data.DataLoader(dataset=_UpperCAmelCase   , batch_size=_UpperCAmelCase   , collate_fn=_UpperCAmelCase     )
  
  						_lowercase					:  Dict    			=      []
  						_lowercase					:  List[Any]    			=      []
  						for batch_idx, (x, y) in enumerate(tqdm(_UpperCAmelCase   , ascii=_UpperCAmelCase     )     ):
  												with torch.no_grad():
  																		_lowercase					:  int    			=      x.to(_UpperCAmelCase     )
  																		_lowercase					:  Tuple    			=      discriminator.avg_representation(_UpperCAmelCase     ).cpu().detach()
  																		_lowercase					:  int    			=      torch.unbind(avg_rep.unsqueeze(1     )     )
  																		xs += avg_rep_list
  																		ys += y.cpu().numpy().tolist()
  
  						_lowercase					:  str    			=      torch.utils.data.DataLoader(
  						    dataset=Dataset(_UpperCAmelCase   , _UpperCAmelCase     )   , batch_size=_UpperCAmelCase   , shuffle=_UpperCAmelCase   , collate_fn=_UpperCAmelCase     )
  
  						return data_loader
  
  
  
  def 		UpperCamelCase    (					_UpperCAmelCase     :      str   , _UpperCAmelCase     :      int=None   , _UpperCAmelCase     :      List[str]="gpt2-medium"   , _UpperCAmelCase     :      List[Any]=10   , _UpperCAmelCase     :      List[str]=64   , _UpperCAmelCase     :      Tuple=10   , _UpperCAmelCase     :      int=False   , _UpperCAmelCase     :      List[str]=False   , _UpperCAmelCase     :      Union[str, Any]=False   , )      -> List[Any]:
  
  
  
  
  
  
  						'''simple docstring'''
  
  						_lowercase					:  Tuple    			=      "cuda" if torch.cuda.is_available() and not no_cuda else "cpu"
  
  						print("Preprocessing {} dataset...".format(_UpperCAmelCase     )     )
  						_lowercase					:  List[str]    			=      time.time()
  
  						if dataset == "SST":
  												_lowercase					:  int    			=      ["positive", "negative", "very positive", "very negative", "neutral"]
  												_lowercase					:  List[str]    			=      {c: i for i, c in enumerate(_UpperCAmelCase     )}
  
  												_lowercase					:  Optional[Any]    			=      Discriminator(
  												    class_size=len(_UpperCAmelCase     )   , pretrained_model=_UpperCAmelCase   , cached_mode=_UpperCAmelCase   , device=_UpperCAmelCase     ).to(_UpperCAmelCase     )
  
  												_lowercase					:  Dict    			=      torchtext_data.Field()
  												_lowercase					:  str    			=      torchtext_data.Field(sequential=_UpperCAmelCase     )
  												_lowercase			,						_lowercase			,						_lowercase					:  int    			=      datasets.SST.splits(
  												    _UpperCAmelCase   , _UpperCAmelCase   , fine_grained=_UpperCAmelCase   , train_subtrees=_UpperCAmelCase   , )
  
  												_lowercase					:  Union[str, Any]    			=      []
  												_lowercase					:  Union[str, Any]    			=      []
  												for i in trange(len(_UpperCAmelCase     )   , ascii=_UpperCAmelCase     ):
  																		_lowercase					:  Any    			=      TreebankWordDetokenizer().detokenize(vars(train_data[i]     )["text"]     )
  																		_lowercase					:  Tuple    			=      discriminator.tokenizer.encode(_UpperCAmelCase     )
  																		_lowercase					:  Union[str, Any]    			=      torch.tensor([5_0256] + seq   , device=_UpperCAmelCase   , dtype=torch.long     )
  																		x.append(_UpperCAmelCase     )
  																		y.append(classaidx[vars(train_data[i]     )["label"]]     )
  												_lowercase					:  Tuple    			=      Dataset(_UpperCAmelCase   , _UpperCAmelCase     )
  
  												_lowercase					:  Tuple    			=      []
  												_lowercase					:  List[str]    			=      []
  												for i in trange(len(_UpperCAmelCase     )   , ascii=_UpperCAmelCase     ):
  																		_lowercase					:  Optional[Any]    			=      TreebankWordDetokenizer().detokenize(vars(test_data[i]     )["text"]     )
  																		_lowercase					:  Tuple    			=      discriminator.tokenizer.encode(_UpperCAmelCase     )
  																		_lowercase					:  List[str]    			=      torch.tensor([5_0256] + seq   , device=_UpperCAmelCase   , dtype=torch.long     )
  																		test_x.append(_UpperCAmelCase     )
  																		test_y.append(classaidx[vars(test_data[i]     )["label"]]     )
  												_lowercase					:  Union[str, Any]    			=      Dataset(_UpperCAmelCase   , _UpperCAmelCase     )
  
  												_lowercase					:  Optional[Any]    			=      {
  												    "class_size": len(_UpperCAmelCase     ),
  												    "embed_size": discriminator.embed_size,
  												    "pretrained_model": pretrained_model,
  												    "class_vocab": classaidx,
  												    "default_class": 2,
  												}
  
  						elif dataset == "clickbait":
  												_lowercase					:  int    			=      ["non_clickbait", "clickbait"]
  												_lowercase					:  List[str]    			=      {c: i for i, c in enumerate(_UpperCAmelCase     )}
  
  												_lowercase					:  Union[str, Any]    			=      Discriminator(
  												    class_size=len(_UpperCAmelCase     )   , pretrained_model=_UpperCAmelCase   , cached_mode=_UpperCAmelCase   , device=_UpperCAmelCase     ).to(_UpperCAmelCase     )
  
  												with open("datasets/clickbait/clickbait_train_prefix.txt"     ) as f:
  																		_lowercase					:  List[Any]    			=      []
  																		for i, line in enumerate(_UpperCAmelCase     ):
  																								try:
  																														data.append(eval(_UpperCAmelCase     )     )
  																								except Exception:
  																														print("Error evaluating line {}: {}".format(_UpperCAmelCase   , _UpperCAmelCase     )     )
  																														continue
  												_lowercase					:  Any    			=      []
  												_lowercase					:  Any    			=      []
  												with open("datasets/clickbait/clickbait_train_prefix.txt"     ) as f:
  																		for i, line in enumerate(tqdm(_UpperCAmelCase   , ascii=_UpperCAmelCase     )     ):
  																								try:
  																														_lowercase					:  List[Any]    			=      eval(_UpperCAmelCase     )
  																														_lowercase					:  Any    			=      discriminator.tokenizer.encode(d["text"]     )
  
  																														if len(_UpperCAmelCase     ) < max_length_seq:
  																																				_lowercase					:  int    			=      torch.tensor([5_0256] + seq   , device=_UpperCAmelCase   , dtype=torch.long     )
  																														else:
  																																				print("Line {} is longer than maximum length {}".format(_UpperCAmelCase   , _UpperCAmelCase     )     )
  																																				continue
  																														x.append(_UpperCAmelCase     )
  																														y.append(d["label"]     )
  																								except Exception:
  																														print("Error evaluating / tokenizing line {}, skipping it".format(_UpperCAmelCase     )     )
  																														pass
  
  												_lowercase					:  int    			=      Dataset(_UpperCAmelCase   , _UpperCAmelCase     )
  												_lowercase					:  List[Any]    			=      int(0.9 * len(_UpperCAmelCase     )     )
  												_lowercase					:  Union[str, Any]    			=      len(_UpperCAmelCase     ) - train_size
  												_lowercase			,						_lowercase					:  Optional[Any]    			=      torch.utils.data.random_split(_UpperCAmelCase   , [train_size, test_size]     )
  
  												_lowercase					:  Any    			=      {
  												    "class_size": len(_UpperCAmelCase     ),
  												    "embed_size": discriminator.embed_size,
  												    "pretrained_model": pretrained_model,
  												    "class_vocab": classaidx,
  												    "default_class": 1,
  												}
  
  						elif dataset == "toxic":
  												_lowercase					:  Optional[Any]    			=      ["non_toxic", "toxic"]
  												_lowercase					:  Optional[Any]    			=      {c: i for i, c in enumerate(_UpperCAmelCase     )}
  
  												_lowercase					:  Any    			=      Discriminator(
  												    class_size=len(_UpperCAmelCase     )   , pretrained_model=_UpperCAmelCase   , cached_mode=_UpperCAmelCase   , device=_UpperCAmelCase     ).to(_UpperCAmelCase     )
  
  												_lowercase					:  Union[str, Any]    			=      []
  												_lowercase					:  List[str]    			=      []
  												with open("datasets/toxic/toxic_train.txt"     ) as f:
  																		for i, line in enumerate(tqdm(_UpperCAmelCase   , ascii=_UpperCAmelCase     )     ):
  																								try:
  																														_lowercase					:  Dict    			=      eval(_UpperCAmelCase     )
  																														_lowercase					:  Dict    			=      discriminator.tokenizer.encode(d["text"]     )
  
  																														if len(_UpperCAmelCase     ) < max_length_seq:
  																																				_lowercase					:  str    			=      torch.tensor([5_0256] + seq   , device=_UpperCAmelCase   , dtype=torch.long     )
  																														else:
  																																				print("Line {} is longer than maximum length {}".format(_UpperCAmelCase   , _UpperCAmelCase     )     )
  																																				continue
  																														x.append(_UpperCAmelCase     )
  																														y.append(int(np.sum(d["label"]     ) > 0     )     )
  																								except Exception:
  																														print("Error evaluating / tokenizing line {}, skipping it".format(_UpperCAmelCase     )     )
  																														pass
  
  												_lowercase					:  Union[str, Any]    			=      Dataset(_UpperCAmelCase   , _UpperCAmelCase     )
  												_lowercase					:  List[str]    			=      int(0.9 * len(_UpperCAmelCase     )     )
  												_lowercase					:  int    			=      len(_UpperCAmelCase     ) - train_size
  												_lowercase			,						_lowercase					:  Union[str, Any]    			=      torch.utils.data.random_split(_UpperCAmelCase   , [train_size, test_size]     )
  
  												_lowercase					:  Any    			=      {
  												    "class_size": len(_UpperCAmelCase     ),
  												    "embed_size": discriminator.embed_size,
  												    "pretrained_model": pretrained_model,
  												    "class_vocab": classaidx,
  												    "default_class": 0,
  												}
  
  						else:  # if dataset == "generic":
  												# This assumes the input dataset is a TSV with the following structure:
  												# class \t text
  
  												if dataset_fp is None:
  																		raise ValueError("When generic dataset is selected, dataset_fp needs to be specified aswell."     )
  
  												_lowercase					:  Tuple    			=      set()
  												with open(_UpperCAmelCase     ) as f:
  																		_lowercase					:  Any    			=      csv.reader(_UpperCAmelCase   , delimiter="\t"     )
  																		for row in tqdm(_UpperCAmelCase   , ascii=_UpperCAmelCase     ):
  																								if row:
  																														classes.add(row[0]     )
  
  												_lowercase					:  Optional[Any]    			=      sorted(_UpperCAmelCase     )
  												_lowercase					:  str    			=      {c: i for i, c in enumerate(_UpperCAmelCase     )}
  
  												_lowercase					:  Optional[Any]    			=      Discriminator(
  												    class_size=len(_UpperCAmelCase     )   , pretrained_model=_UpperCAmelCase   , cached_mode=_UpperCAmelCase   , device=_UpperCAmelCase     ).to(_UpperCAmelCase     )
  
  												_lowercase					:  Optional[int]    			=      []
  												_lowercase					:  List[Any]    			=      []
  												with open(_UpperCAmelCase     ) as f:
  																		_lowercase					:  Optional[int]    			=      csv.reader(_UpperCAmelCase   , delimiter="\t"     )
  																		for i, row in enumerate(tqdm(_UpperCAmelCase   , ascii=_UpperCAmelCase     )     ):
  																								if row:
  																														_lowercase					:  Tuple    			=      row[0]
  																														_lowercase					:  List[Any]    			=      row[1]
  
  																														try:
  																																				_lowercase					:  Tuple    			=      discriminator.tokenizer.encode(_UpperCAmelCase     )
  																																				if len(_UpperCAmelCase     ) < max_length_seq:
  																																										_lowercase					:  Any    			=      torch.tensor([5_0256] + seq   , device=_UpperCAmelCase   , dtype=torch.long     )
  
  																																				else:
  																																										print("Line {} is longer than maximum length {}".format(_UpperCAmelCase   , _UpperCAmelCase     )     )
  																																										continue
  
  																																				x.append(_UpperCAmelCase     )
  																																				y.append(classaidx[label]     )
  
  																														except Exception:
  																																				print("Error tokenizing line {}, skipping it".format(_UpperCAmelCase     )     )
  																																				pass
  
  												_lowercase					:  Tuple    			=      Dataset(_UpperCAmelCase   , _UpperCAmelCase     )
  												_lowercase					:  Any    			=      int(0.9 * len(_UpperCAmelCase     )     )
  												_lowercase					:  Dict    			=      len(_UpperCAmelCase     ) - train_size
  												_lowercase			,						_lowercase					:  List[str]    			=      torch.utils.data.random_split(_UpperCAmelCase   , [train_size, test_size]     )
  
  												_lowercase					:  List[Any]    			=      {
  												    "class_size": len(_UpperCAmelCase     ),
  												    "embed_size": discriminator.embed_size,
  												    "pretrained_model": pretrained_model,
  												    "class_vocab": classaidx,
  												    "default_class": 0,
  												}
  
  						_lowercase					:  Tuple    			=      time.time()
  						print("Preprocessed {} data points".format(len(_UpperCAmelCase     ) + len(_UpperCAmelCase     )     )     )
  						print("Data preprocessing took: {:.3f}s".format(end - start     )     )
  
  						if cached:
  												print("Building representation cache..."     )
  
  												_lowercase					:  Any    			=      time.time()
  
  												_lowercase					:  List[str]    			=      get_cached_data_loader(_UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , shuffle=_UpperCAmelCase   , device=_UpperCAmelCase     )
  
  												_lowercase					:  List[Any]    			=      get_cached_data_loader(_UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , device=_UpperCAmelCase     )
  
  												_lowercase					:  Optional[Any]    			=      time.time()
  												print("Building representation cache took: {:.3f}s".format(end - start     )     )
  
  						else:
  												_lowercase					:  List[str]    			=      torch.utils.data.DataLoader(
  												    dataset=_UpperCAmelCase   , batch_size=_UpperCAmelCase   , shuffle=_UpperCAmelCase   , collate_fn=_UpperCAmelCase     )
  												_lowercase					:  int    			=      torch.utils.data.DataLoader(dataset=_UpperCAmelCase   , batch_size=_UpperCAmelCase   , collate_fn=_UpperCAmelCase     )
  
  						if save_model:
  												with open("{}_classifier_head_meta.json".format(_UpperCAmelCase     )   , "w"     ) as meta_file:
  																		json.dump(_UpperCAmelCase   , _UpperCAmelCase     )
  
  						_lowercase					:  Tuple    			=      optim.Adam(discriminator.parameters()   , lr=0.0_001     )
  
  						for epoch in range(_UpperCAmelCase     ):
  												_lowercase					:  Tuple    			=      time.time()
  												print("\nEpoch"   , epoch + 1     )
  
  												train_epoch(
  												    discriminator=_UpperCAmelCase   , data_loader=_UpperCAmelCase   , optimizer=_UpperCAmelCase   , epoch=_UpperCAmelCase   , log_interval=_UpperCAmelCase   , device=_UpperCAmelCase   , )
  												evaluate_performance(data_loader=_UpperCAmelCase   , discriminator=_UpperCAmelCase   , device=_UpperCAmelCase     )
  
  												_lowercase					:  Tuple    			=      time.time()
  												print("Epoch took: {:.3f}s".format(end - start     )     )
  
  												print("\nExample prediction"     )
  												predict(_UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , cached=_UpperCAmelCase   , device=_UpperCAmelCase     )
  
  												if save_model:
  																		# torch.save(discriminator.state_dict(),
  																		#           "{}_discriminator_{}.pt".format(
  																		#               args.dataset, epoch + 1
  																		#               ))
  																		torch.save(
  																		    discriminator.get_classifier().state_dict()   , "{}_classifier_head_epoch_{}.pt".format(_UpperCAmelCase   , epoch + 1     )   , )
  
  
  if __name__ == "__main__":
  							UpperCamelCase_	:    int  					=       argparse.ArgumentParser(description="""Train a discriminator on top of GPT-2 representations""")
  							parser.add_argument(
  							    """--dataset""",
  							    type=str,
  							    default="""SST""",
  							    choices=("""SST""", """clickbait""", """toxic""", """generic"""),
  							    help=(
  							        """dataset to train the discriminator on."""
  							        """In case of generic, the dataset is expected"""
  							        """to be a TSBV file with structure: class \\t text"""
  							    ),
  							)
  							parser.add_argument(
  							    """--dataset_fp""",
  							    type=str,
  							    default="""""",
  							    help="""File path of the dataset to use. Needed only in case of generic datadset""",
  							)
  							parser.add_argument(
  							    """--pretrained_model""", type=str, default="""gpt2-medium""", help="""Pretrained model to use as encoder"""
  							)
  							parser.add_argument("""--epochs""", type=int, default=10, metavar="""N""", help="""Number of training epochs""")
  							parser.add_argument(
  							    """--batch_size""", type=int, default=64, metavar="""N""", help="""input batch size for training (default: 64)"""
  							)
  							parser.add_argument(
  							    """--log_interval""",
  							    type=int,
  							    default=10,
  							    metavar="""N""",
  							    help="""how many batches to wait before logging training status""",
  							)
  							parser.add_argument("""--save_model""", action="""store_true""", help="""whether to save the model""")
  							parser.add_argument("""--cached""", action="""store_true""", help="""whether to cache the input representations""")
  							parser.add_argument("""--no_cuda""", action="""store_true""", help="""use to turn off cuda""")
  							UpperCamelCase_	:    Optional[Any]  					=       parser.parse_args()
  
  							train_discriminator(**(vars(args)))
  
  
  
  
  
style_context: 
  import gc
  import os
  import tempfile
  import unittest
  
  import torch
  import torch.nn as nn
  import torch.nn.functional as F
  from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer
  
  from diffusers import AutoencoderKL, DDIMScheduler, StableDiffusionPipeline, UNetaDConditionModel
  from diffusers.loaders import AttnProcsLayers, LoraLoaderMixin
  from diffusers.models.attention_processor import (
      Attention,
      AttnProcessor,
      AttnProcessor2_0,
      LoRAAttnProcessor,
      LoRAAttnProcessor2_0,
      LoRAXFormersAttnProcessor,
      XFormersAttnProcessor,
  )
  from diffusers.utils import TEXT_ENCODER_ATTN_MODULE, floats_tensor, torch_device
  
  
  
  def 		UpperCamelCase    (					_UpperCAmelCase     :      nn.Module     )      -> Any:
  
  
  
  
  
  
  						'''simple docstring'''
  
  						_lowercase					:  Optional[int]    			=      {}
  						for name in unet.attn_processors.keys():
  												_lowercase					:  Any    			=      None if name.endswith("attn1.processor"     ) else unet.config.cross_attention_dim
  												if name.startswith("mid_block"     ):
  																		_lowercase					:  Any    			=      unet.config.block_out_channels[-1]
  												elif name.startswith("up_blocks"     ):
  																		_lowercase					:  List[Any]    			=      int(name[len("up_blocks."     )]     )
  																		_lowercase					:  Any    			=      list(reversed(unet.config.block_out_channels     )     )[block_id]
  												elif name.startswith("down_blocks"     ):
  																		_lowercase					:  Union[str, Any]    			=      int(name[len("down_blocks."     )]     )
  																		_lowercase					:  Dict    			=      unet.config.block_out_channels[block_id]
  												_lowercase					:  List[str]    			=      (
  												    LoRAAttnProcessor2_0 if hasattr(_UpperCAmelCase   , "scaled_dot_product_attention"     ) else LoRAAttnProcessor
  												)
  												_lowercase					:  Optional[Any]    			=      lora_attn_processor_class(
  												    hidden_size=_UpperCAmelCase   , cross_attention_dim=_UpperCAmelCase     )
  						_lowercase					:  Dict    			=      AttnProcsLayers(_UpperCAmelCase     )
  						return lora_attn_procs, unet_lora_layers
  
  
  
  def 		UpperCamelCase    (					_UpperCAmelCase     :      nn.Module     )      -> Any:
  
  
  
  
  
  
  						'''simple docstring'''
  
  						_lowercase					:  List[str]    			=      {}
  						_lowercase					:  Dict    			=      (
  						    LoRAAttnProcessor2_0 if hasattr(_UpperCAmelCase   , "scaled_dot_product_attention"     ) else LoRAAttnProcessor
  						)
  						for name, module in text_encoder.named_modules():
  												if name.endswith(_UpperCAmelCase     ):
  																		_lowercase					:  Tuple    			=      lora_attn_processor_class(
  																		    hidden_size=module.out_proj.out_features   , cross_attention_dim=_UpperCAmelCase     )
  						return text_lora_attn_procs
  
  
  
  def 		UpperCamelCase    (					_UpperCAmelCase     :      nn.Module     )      -> Union[str, Any]:
  
  
  
  
  
  
  						'''simple docstring'''
  
  						_lowercase					:  Union[str, Any]    			=      create_text_encoder_lora_attn_procs(_UpperCAmelCase     )
  						_lowercase					:  str    			=      AttnProcsLayers(_UpperCAmelCase     )
  						return text_encoder_lora_layers
  
  
  
  def 		UpperCamelCase    (					_UpperCAmelCase     :      str   , _UpperCAmelCase     :      int=False     )      -> List[Any]:
  
  
  
  
  
  
  						'''simple docstring'''
  
  						for _, attn_proc in text_lora_attn_procs.items():
  												# set up.weights
  												for layer_name, layer_module in attn_proc.named_modules():
  																		if layer_name.endswith("_lora"     ):
  																								_lowercase					:  Tuple    			=      (
  																								    torch.randn_like(layer_module.up.weight     )
  																								    if randn_weight
  																								    else torch.zeros_like(layer_module.up.weight     )
  																								)
  																								_lowercase					:  Dict    			=      torch.nn.Parameter(_UpperCAmelCase     )
  
  
  
  
  class      __lowercase	(	unittest.TestCase					):
  
  
  
  
  
  
  	def 			_a(self							:   str						)   -> Tuple:
  							torch.manual_seed(0						)
  							_lowercase					:  int    			=      UNetaDConditionModel(
  							    block_out_channels=(32, 64)			,  layers_per_block=2			,  sample_size=32			,  in_channels=4			,  out_channels=4			,  down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")			,  up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")			,  cross_attention_dim=32			,  )
  							_lowercase					:  Union[str, Any]    			=      DDIMScheduler(
  							    beta_start=0.0_00_85			,  beta_end=0.0_12			,  beta_schedule="scaled_linear"			,  clip_sample=snake_case			,  set_alpha_to_one=snake_case			,  steps_offset=1			,  )
  							torch.manual_seed(0						)
  							_lowercase					:  int    			=      AutoencoderKL(
  							    block_out_channels=[32, 64]			,  in_channels=3			,  out_channels=3			,  down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]			,  up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]			,  latent_channels=4			,  )
  							_lowercase					:  Optional[int]    			=      CLIPTextConfig(
  							    bos_token_id=0			,  eos_token_id=2			,  hidden_size=32			,  intermediate_size=37			,  layer_norm_eps=1e-05			,  num_attention_heads=4			,  num_hidden_layers=5			,  pad_token_id=1			,  vocab_size=1000			,  )
  							_lowercase					:  List[Any]    			=      CLIPTextModel(snake_case						)
  							_lowercase					:  Dict    			=      CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"						)
  
  							_lowercase			,						_lowercase					:  List[Any]    			=      create_unet_lora_layers(snake_case						)
  							_lowercase					:  Union[str, Any]    			=      create_text_encoder_lora_layers(snake_case						)
  
  							_lowercase					:  List[Any]    			=      {
  							    "unet": unet,
  							    "scheduler": scheduler,
  							    "vae": vae,
  							    "text_encoder": text_encoder,
  							    "tokenizer": tokenizer,
  							    "safety_checker": None,
  							    "feature_extractor": None,
  							}
  							_lowercase					:  Union[str, Any]    			=      {
  							    "unet_lora_layers": unet_lora_layers,
  							    "text_encoder_lora_layers": text_encoder_lora_layers,
  							    "unet_lora_attn_procs": unet_lora_attn_procs,
  							}
  							return pipeline_components, lora_components
  
  
  
  
  
  
  	def 			_a(self							:   List[Any]						)   -> Optional[Any]:
  							_lowercase					:  str    			=      1
  							_lowercase					:  str    			=      10
  							_lowercase					:  Any    			=      4
  							_lowercase					:  List[str]    			=      (32, 32)
  
  							_lowercase					:  Union[str, Any]    			=      torch.manual_seed(0						)
  							_lowercase					:  List[str]    			=      floats_tensor((batch_size, num_channels) + sizes						)
  							_lowercase					:  Dict    			=      torch.randint(1			,  snake_case			,  size=(batch_size, sequence_length)			,  generator=snake_case						)
  
  							_lowercase					:  Tuple    			=      {
  							    "prompt": "A painting of a squirrel eating a burger",
  							    "generator": generator,
  							    "num_inference_steps": 2,
  							    "guidance_scale": 6.0,
  							    "output_type": "numpy",
  							}
  
  							return noise, input_ids, pipeline_inputs
  
  							# copied from: https://colab.research.google.com/gist/sayakpaul/df2ef6e1ae6d8c10a49d859883b10860/scratchpad.ipynb
  
  
  
  
  
  
  	def 			_a(self							:   Union[str, Any]						)   -> int:
  							_lowercase					:  Any    			=      77
  
  							_lowercase					:  Any    			=      torch.randint(2			,  56			,  size=(1, max_seq_length)			,  generator=torch.manual_seed(0						)						)
  
  							_lowercase					:  int    			=      {}
  							_lowercase					:  int    			=      inputs
  							return prepared_inputs
  
  
  
  
  
  
  	def 			_a(self							:   Union[str, Any]			,  snake_case							:   Dict						)   -> List[Any]:
  							_lowercase			,						_lowercase					:  List[Any]    			=      self.get_dummy_components()
  							LoraLoaderMixin.save_lora_weights(
  							    save_directory=snake_case			,  unet_lora_layers=lora_components["unet_lora_layers"]			,  text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]			,  )
  							self.assertTrue(os.path.isfile(os.path.join(snake_case			,  "pytorch_lora_weights.bin"						)						)						)
  
  
  
  
  
  
  	def 			_a(self							:   Any						)   -> Union[str, Any]:
  							_lowercase			,						_lowercase					:  int    			=      self.get_dummy_components()
  							_lowercase					:  List[str]    			=      StableDiffusionPipeline(**snake_case						)
  							_lowercase					:  Dict    			=      sd_pipe.to(snake_case						)
  							sd_pipe.set_progress_bar_config(disable=snake_case						)
  
  							_lowercase			,						_lowercase			,						_lowercase					:  Optional[int]    			=      self.get_dummy_inputs()
  
  							_lowercase					:  Any    			=      sd_pipe(**snake_case						).images
  							_lowercase					:  int    			=      original_images[0, -3:, -3:, -1]
  
  							with tempfile.TemporaryDirectory() as tmpdirname:
  													LoraLoaderMixin.save_lora_weights(
  													    save_directory=snake_case			,  unet_lora_layers=lora_components["unet_lora_layers"]			,  text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]			,  )
  													self.assertTrue(os.path.isfile(os.path.join(snake_case			,  "pytorch_lora_weights.bin"						)						)						)
  													sd_pipe.load_lora_weights(snake_case						)
  
  							_lowercase					:  Dict    			=      sd_pipe(**snake_case						).images
  							_lowercase					:  List[Any]    			=      lora_images[0, -3:, -3:, -1]
  
  							# Outputs shouldn't match.
  							self.assertFalse(torch.allclose(torch.from_numpy(snake_case						)			,  torch.from_numpy(snake_case						)						)						)
  
  
  
  
  
  
  	def 			_a(self							:   Optional[int]						)   -> List[Any]:
  							_lowercase			,						_lowercase					:  str    			=      self.get_dummy_components()
  							_lowercase					:  Any    			=      StableDiffusionPipeline(**snake_case						)
  							_lowercase					:  Any    			=      sd_pipe.to(snake_case						)
  							sd_pipe.set_progress_bar_config(disable=snake_case						)
  
  							_lowercase			,						_lowercase			,						_lowercase					:  Optional[int]    			=      self.get_dummy_inputs()
  
  							_lowercase					:  Optional[int]    			=      sd_pipe(**snake_case						).images
  							_lowercase					:  Any    			=      original_images[0, -3:, -3:, -1]
  
  							with tempfile.TemporaryDirectory() as tmpdirname:
  													LoraLoaderMixin.save_lora_weights(
  													    save_directory=snake_case			,  unet_lora_layers=lora_components["unet_lora_layers"]			,  text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]			,  safe_serialization=snake_case			,  )
  													self.assertTrue(os.path.isfile(os.path.join(snake_case			,  "pytorch_lora_weights.safetensors"						)						)						)
  													sd_pipe.load_lora_weights(snake_case						)
  
  							_lowercase					:  Any    			=      sd_pipe(**snake_case						).images
  							_lowercase					:  Optional[int]    			=      lora_images[0, -3:, -3:, -1]
  
  							# Outputs shouldn't match.
  							self.assertFalse(torch.allclose(torch.from_numpy(snake_case						)			,  torch.from_numpy(snake_case						)						)						)
  
  
  
  
  
  
  	def 			_a(self							:   str						)   -> Union[str, Any]:
  							_lowercase			,						_lowercase					:  Optional[Any]    			=      self.get_dummy_components()
  							_lowercase					:  int    			=      lora_components["unet_lora_attn_procs"]
  							_lowercase					:  str    			=      StableDiffusionPipeline(**snake_case						)
  							_lowercase					:  List[str]    			=      sd_pipe.to(snake_case						)
  							sd_pipe.set_progress_bar_config(disable=snake_case						)
  
  							_lowercase			,						_lowercase			,						_lowercase					:  Any    			=      self.get_dummy_inputs()
  
  							_lowercase					:  int    			=      sd_pipe(**snake_case						).images
  							_lowercase					:  List[str]    			=      original_images[0, -3:, -3:, -1]
  
  							with tempfile.TemporaryDirectory() as tmpdirname:
  													_lowercase					:  Optional[Any]    			=      sd_pipe.unet
  													unet.set_attn_processor(snake_case						)
  													unet.save_attn_procs(snake_case						)
  													self.assertTrue(os.path.isfile(os.path.join(snake_case			,  "pytorch_lora_weights.bin"						)						)						)
  													sd_pipe.load_lora_weights(snake_case						)
  
  							_lowercase					:  int    			=      sd_pipe(**snake_case						).images
  							_lowercase					:  int    			=      lora_images[0, -3:, -3:, -1]
  
  							# Outputs shouldn't match.
  							self.assertFalse(torch.allclose(torch.from_numpy(snake_case						)			,  torch.from_numpy(snake_case						)						)						)
  
  
  
  
  
  
  	def 			_a(self							:   Dict						)   -> List[str]:
  							_lowercase			,						_lowercase					:  Optional[Any]    			=      self.get_dummy_components()
  							_lowercase					:  List[Any]    			=      StableDiffusionPipeline(**snake_case						)
  
  							_lowercase					:  Tuple    			=      self.get_dummy_tokens()
  
  							# inference without lora
  							_lowercase					:  int    			=      pipe.text_encoder(**snake_case						)[0]
  							assert outputs_without_lora.shape == (1, 77, 32)
  
  							# create lora_attn_procs with zeroed out up.weights
  							_lowercase					:  str    			=      create_text_encoder_lora_attn_procs(pipe.text_encoder						)
  							set_lora_up_weights(snake_case			,  randn_weight=snake_case						)
  
  							# monkey patch
  							pipe._modify_text_encoder(snake_case						)
  
  							# verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
  							del text_attn_procs
  							gc.collect()
  
  							# inference with lora
  							_lowercase					:  Dict    			=      pipe.text_encoder(**snake_case						)[0]
  							assert outputs_with_lora.shape == (1, 77, 32)
  
  							assert torch.allclose(
  							    snake_case			,  snake_case						), "lora_up_weight are all zero, so the lora outputs should be the same to without lora outputs"
  
  							# create lora_attn_procs with randn up.weights
  							_lowercase					:  Any    			=      create_text_encoder_lora_attn_procs(pipe.text_encoder						)
  							set_lora_up_weights(snake_case			,  randn_weight=snake_case						)
  
  							# monkey patch
  							pipe._modify_text_encoder(snake_case						)
  
  							# verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
  							del text_attn_procs
  							gc.collect()
  
  							# inference with lora
  							_lowercase					:  Any    			=      pipe.text_encoder(**snake_case						)[0]
  							assert outputs_with_lora.shape == (1, 77, 32)
  
  							assert not torch.allclose(
  							    snake_case			,  snake_case						), "lora_up_weight are not zero, so the lora outputs should be different to without lora outputs"
  
  
  
  
  
  
  	def 			_a(self							:   str						)   -> Tuple:
  							_lowercase			,						_lowercase					:  List[Any]    			=      self.get_dummy_components()
  							_lowercase					:  str    			=      StableDiffusionPipeline(**snake_case						)
  
  							_lowercase					:  Dict    			=      self.get_dummy_tokens()
  
  							# inference without lora
  							_lowercase					:  Tuple    			=      pipe.text_encoder(**snake_case						)[0]
  							assert outputs_without_lora.shape == (1, 77, 32)
  
  							# create lora_attn_procs with randn up.weights
  							_lowercase					:  Optional[int]    			=      create_text_encoder_lora_attn_procs(pipe.text_encoder						)
  							set_lora_up_weights(snake_case			,  randn_weight=snake_case						)
  
  							# monkey patch
  							pipe._modify_text_encoder(snake_case						)
  
  							# verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
  							del text_attn_procs
  							gc.collect()
  
  							# inference with lora
  							_lowercase					:  Union[str, Any]    			=      pipe.text_encoder(**snake_case						)[0]
  							assert outputs_with_lora.shape == (1, 77, 32)
  
  							assert not torch.allclose(
  							    snake_case			,  snake_case						), "lora outputs should be different to without lora outputs"
  
  							# remove monkey patch
  							pipe._remove_text_encoder_monkey_patch()
  
  							# inference with removed lora
  							_lowercase					:  Union[str, Any]    			=      pipe.text_encoder(**snake_case						)[0]
  							assert outputs_without_lora_removed.shape == (1, 77, 32)
  
  							assert torch.allclose(
  							    snake_case			,  snake_case						), "remove lora monkey patch should restore the original outputs"
  
  
  
  
  
  
  	def 			_a(self							:   Optional[int]						)   -> Dict:
  							_lowercase			,						_lowercase					:  Dict    			=      self.get_dummy_components()
  							_lowercase					:  Optional[Any]    			=      StableDiffusionPipeline(**snake_case						)
  							_lowercase					:  Union[str, Any]    			=      sd_pipe.to(snake_case						)
  							sd_pipe.set_progress_bar_config(disable=snake_case						)
  
  							_lowercase			,						_lowercase			,						_lowercase					:  str    			=      self.get_dummy_inputs()
  
  							with tempfile.TemporaryDirectory() as tmpdirname:
  													LoraLoaderMixin.save_lora_weights(
  													    save_directory=snake_case			,  unet_lora_layers=lora_components["unet_lora_layers"]			,  text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]			,  )
  													self.assertTrue(os.path.isfile(os.path.join(snake_case			,  "pytorch_lora_weights.bin"						)						)						)
  													sd_pipe.load_lora_weights(snake_case						)
  
  							_lowercase					:  Dict    			=      sd_pipe(**snake_case						).images
  							_lowercase					:  int    			=      lora_images[0, -3:, -3:, -1]
  
  							_lowercase					:  List[str]    			=      sd_pipe(**snake_case			,  cross_attention_kwargs={"scale": 0.5}						).images
  							_lowercase					:  int    			=      lora_images_with_scale[0, -3:, -3:, -1]
  
  							# Outputs shouldn't match.
  							self.assertFalse(
  							    torch.allclose(torch.from_numpy(snake_case						)			,  torch.from_numpy(snake_case						)						)						)
  
  
  
  
  
  
  	def 			_a(self							:   Tuple						)   -> str:
  							with tempfile.TemporaryDirectory() as tmpdirname:
  													self.create_lora_weight_file(snake_case						)
  
  													_lowercase			,						_lowercase					:  str    			=      self.get_dummy_components()
  													_lowercase					:  str    			=      StableDiffusionPipeline(**snake_case						)
  													_lowercase					:  List[Any]    			=      sd_pipe.to(snake_case						)
  													sd_pipe.set_progress_bar_config(disable=snake_case						)
  
  													# check if vanilla attention processors are used
  													for _, module in sd_pipe.unet.named_modules():
  																			if isinstance(snake_case			,  snake_case						):
  																									self.assertIsInstance(module.processor			,  (AttnProcessor, AttnProcessor2_0)						)
  
              # load LoRA weight file
  													sd_pipe.load_lora_weights(snake_case						)
  
  													# check if lora attention processors are used
  													for _, module in sd_pipe.unet.named_modules():
  																			if isinstance(snake_case			,  snake_case						):
  																									_lowercase					:  Tuple    			=      (
  																									    LoRAAttnProcessor2_0 if hasattr(snake_case			,  "scaled_dot_product_attention"						) else LoRAAttnProcessor
  																									)
  																									self.assertIsInstance(module.processor			,  snake_case						)
  
  
  
  
  
  
  	@unittest.skipIf(torch_device != "cuda"			,  "This test is supposed to run on GPU"						)
  	def 			_a(self							:   List[str]						)   -> Union[str, Any]:
  							with tempfile.TemporaryDirectory() as tmpdirname:
  													self.create_lora_weight_file(snake_case						)
  
  													_lowercase			,						_lowercase					:  List[str]    			=      self.get_dummy_components()
  													_lowercase					:  str    			=      StableDiffusionPipeline(**snake_case						)
  													_lowercase					:  Tuple    			=      sd_pipe.to(snake_case						)
  													sd_pipe.set_progress_bar_config(disable=snake_case						)
  
  													# enable XFormers
  													sd_pipe.enable_xformers_memory_efficient_attention()
  
  													# check if xFormers attention processors are used
  													for _, module in sd_pipe.unet.named_modules():
  																			if isinstance(snake_case			,  snake_case						):
  																									self.assertIsInstance(module.processor			,  snake_case						)
  
              # load LoRA weight file
  													sd_pipe.load_lora_weights(snake_case						)
  
  													# check if lora attention processors are used
  													for _, module in sd_pipe.unet.named_modules():
  																			if isinstance(snake_case			,  snake_case						):
  																									self.assertIsInstance(module.processor			,  snake_case						)
  
  
  
  
  
  
  	@unittest.skipIf(torch_device != "cuda"			,  "This test is supposed to run on GPU"						)
  	def 			_a(self							:   List[str]						)   -> Any:
  							_lowercase			,						_lowercase					:  Tuple    			=      self.get_dummy_components()
  							_lowercase					:  List[Any]    			=      StableDiffusionPipeline(**snake_case						)
  							_lowercase					:  Union[str, Any]    			=      sd_pipe.to(snake_case						)
  							sd_pipe.set_progress_bar_config(disable=snake_case						)
  
  							_lowercase			,						_lowercase			,						_lowercase					:  List[Any]    			=      self.get_dummy_inputs()
  
  							# enable XFormers
  							sd_pipe.enable_xformers_memory_efficient_attention()
  
  							_lowercase					:  Tuple    			=      sd_pipe(**snake_case						).images
  							_lowercase					:  Any    			=      original_images[0, -3:, -3:, -1]
  
  							with tempfile.TemporaryDirectory() as tmpdirname:
  													LoraLoaderMixin.save_lora_weights(
  													    save_directory=snake_case			,  unet_lora_layers=lora_components["unet_lora_layers"]			,  text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]			,  )
  													self.assertTrue(os.path.isfile(os.path.join(snake_case			,  "pytorch_lora_weights.bin"						)						)						)
  													sd_pipe.load_lora_weights(snake_case						)
  
  							_lowercase					:  Optional[int]    			=      sd_pipe(**snake_case						).images
  							_lowercase					:  Tuple    			=      lora_images[0, -3:, -3:, -1]
  
  							# Outputs shouldn't match.
  							self.assertFalse(torch.allclose(torch.from_numpy(snake_case						)			,  torch.from_numpy(snake_case						)						)						)
  
  
  
  
  