





'''simple docstring'''






from typing import Optional, Union

from ...file_utils import TensorType
from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, TruncationStrategy







class       lowerCamelCase       (    __UpperCAmelCase	):
					_SCREAMING_SNAKE_CASE				 =     "MarkupLMFeatureExtractor"
					_SCREAMING_SNAKE_CASE				 =     ("MarkupLMTokenizer", "MarkupLMTokenizerFast")
					_SCREAMING_SNAKE_CASE				 =     True


					def __call__(       self						:	Dict							,  __snake_case						:	Tuple=None							,  __snake_case						:	List[Any]=None							,  __snake_case						:	str=None							,  __snake_case						:	Any=None							,  __snake_case						:	List[Any]=None							,  __snake_case						:	bool = True							,  __snake_case						:	Union[bool, str, PaddingStrategy] = False							,  __snake_case						:	Union[bool, str, TruncationStrategy] = None							,  __snake_case						:	Optional[int] = None							,  __snake_case						:	int = 0							,  __snake_case						:	Optional[int] = None							,  __snake_case						:	Optional[bool] = None							,  __snake_case						:	Optional[bool] = None							,  __snake_case						:	bool = False							,  __snake_case						:	bool = False							,  __snake_case						:	bool = False							,  __snake_case						:	bool = False							,  __snake_case						:	bool = True							,  __snake_case						:	Optional[Union[str, TensorType]] = None							,  **__snake_case						:	Any							,  ):





											'''simple docstring'''





											if self.parse_html:
																	if html_strings is None:
																							raise ValueError('Make sure to pass HTML strings in case `parse_html` is set to `True`'							)

																	if nodes is not None or xpaths is not None or node_labels is not None:
																							raise ValueError(
																							    'Please don\'t pass nodes, xpaths nor node labels in case `parse_html` is set to `True`'							)

																	_snake_case: int	       =		self.feature_extractor(__snake_case							)
																	_snake_case: Tuple	       =		features['nodes']
																	_snake_case: int	       =		features['xpaths']
											else:
																	if html_strings is not None:
																							raise ValueError('You have passed HTML strings but `parse_html` is set to `False`.'							)
																	if nodes is None or xpaths is None:
																							raise ValueError('Make sure to pass nodes and xpaths in case `parse_html` is set to `False`'							)

        # # second, apply the tokenizer
											if questions is not None and self.parse_html:
																	if isinstance(__snake_case							,  __snake_case							):
																							_snake_case: Optional[Any]	       =		[questions]  # add batch dimension (as the feature extractor always adds a batch dimension)

											_snake_case: List[str]	       =		self.tokenizer(
											    text=questions if questions is not None else nodes							,  text_pair=nodes if questions is not None else None							,  xpaths=__snake_case							,  node_labels=__snake_case							,  add_special_tokens=__snake_case							,  padding=__snake_case							,  truncation=__snake_case							,  max_length=__snake_case							,  stride=__snake_case							,  pad_to_multiple_of=__snake_case							,  return_token_type_ids=__snake_case							,  return_attention_mask=__snake_case							,  return_overflowing_tokens=__snake_case							,  return_special_tokens_mask=__snake_case							,  return_offsets_mapping=__snake_case							,  return_length=__snake_case							,  verbose=__snake_case							,  return_tensors=__snake_case							,  **__snake_case							,  )

											return encoded_inputs


					def  SCREAMING_SNAKE_CASE_		(       self						:	Union[str, Any]							,  *__snake_case						:	Any							,  **__snake_case						:	Tuple							):





											'''simple docstring'''





											return self.tokenizer.batch_decode(*__snake_case							,  **__snake_case							)


					def  SCREAMING_SNAKE_CASE_		(       self						:	Dict							,  *__snake_case						:	Dict							,  **__snake_case						:	str							):





											'''simple docstring'''





											return self.tokenizer.decode(*__snake_case							,  **__snake_case							)




					@property
					def  SCREAMING_SNAKE_CASE_		(       self						:	str							):





											'''simple docstring'''





											_snake_case: int	       =		self.tokenizer.model_input_names
											return tokenizer_input_names












'''simple docstring'''






import tempfile
import unittest

import numpy as np
import torch

from diffusers import DiffusionPipeline
from diffusers.models.attention_processor import Attention, AttnAddedKVProcessor







class       lowerCamelCase       (    unittest.TestCase	):


					def  SCREAMING_SNAKE_CASE_		(       self						:	List[Any]							,  __snake_case						:	bool = False							):





											'''simple docstring'''





											_snake_case: Any	       =		10

											if only_cross_attention:
																	_snake_case: Optional[int]	       =		12
											else:
																	# when only cross attention is not set, the cross attention dim must be the same as the query dim
																	_snake_case: Optional[int]	       =		query_dim

											return {
											    "query_dim": query_dim,
											    "cross_attention_dim": cross_attention_dim,
											    "heads": 2,
											    "dim_head": 4,
											    "added_kv_proj_dim": 6,
											    "norm_num_groups": 1,
											    "only_cross_attention": only_cross_attention,
											    "processor": AttnAddedKVProcessor(),
											}


					def  SCREAMING_SNAKE_CASE_		(       self						:	List[Any]							,  __snake_case						:	List[Any]							,  __snake_case						:	Tuple							):





											'''simple docstring'''





											_snake_case: Optional[int]	       =		2

											_snake_case: Union[str, Any]	       =		torch.rand(__snake_case							,  __snake_case							,  3							,  2							)
											_snake_case: Optional[int]	       =		torch.rand(__snake_case							,  4							,  __snake_case							)
											_snake_case: int	       =		None

											return {
											    "hidden_states": hidden_states,
											    "encoder_hidden_states": encoder_hidden_states,
											    "attention_mask": attention_mask,
											}




					def  SCREAMING_SNAKE_CASE_		(       self						:	List[Any]							):





											'''simple docstring'''





											torch.manual_seed(0							)

											_snake_case: int	       =		self.get_constructor_arguments(only_cross_attention=__snake_case							)
											_snake_case: Union[str, Any]	       =		Attention(**__snake_case							)

											self.assertTrue(attn.to_k is not None							)
											self.assertTrue(attn.to_v is not None							)

											_snake_case: Optional[int]	       =		self.get_forward_arguments(
											    query_dim=constructor_args['query_dim']							,  added_kv_proj_dim=constructor_args['added_kv_proj_dim']							)

											_snake_case: str	       =		attn(**__snake_case							)

											# only self attention

											torch.manual_seed(0							)

											_snake_case: str	       =		self.get_constructor_arguments(only_cross_attention=__snake_case							)
											_snake_case: Optional[Any]	       =		Attention(**__snake_case							)

											self.assertTrue(attn.to_k is None							)
											self.assertTrue(attn.to_v is None							)

											_snake_case: Dict	       =		self.get_forward_arguments(
											    query_dim=constructor_args['query_dim']							,  added_kv_proj_dim=constructor_args['added_kv_proj_dim']							)

											_snake_case: str	       =		attn(**__snake_case							)

											self.assertTrue((only_cross_attn_out != self_and_cross_attn_out).all()							)







class       lowerCamelCase       (    unittest.TestCase	):


					def  SCREAMING_SNAKE_CASE_		(       self						:	Tuple							):





											'''simple docstring'''





											_snake_case: str	       =		DiffusionPipeline.from_pretrained('hf-internal-testing/tiny-stable-diffusion-pipe'							,  safety_checker=__snake_case							)

											_snake_case: int	       =		pipe(
											    'foo'							,  num_inference_steps=2							,  generator=torch.Generator('cpu'							).manual_seed(0							)							,  output_type='np'							,  ).images

											# the initial conversion succeeds
											_snake_case: Dict	       =		DiffusionPipeline.from_pretrained(
											    'hf-internal-testing/tiny-stable-diffusion-pipe'							,  device_map='sequential'							,  safety_checker=__snake_case							)

											_snake_case: Any	       =		pipe(
											    'foo'							,  num_inference_steps=2							,  generator=torch.Generator('cpu'							).manual_seed(0							)							,  output_type='np'							,  ).images

											with tempfile.TemporaryDirectory() as tmpdir:
																	# save the converted model
																	pipe.save_pretrained(__snake_case							)

																	# can also load the converted weights
																	_snake_case: Union[str, Any]	       =		DiffusionPipeline.from_pretrained(__snake_case							,  device_map='sequential'							,  safety_checker=__snake_case							)

											_snake_case: Union[str, Any]	       =		pipe(
											    'foo'							,  num_inference_steps=2							,  generator=torch.Generator('cpu'							).manual_seed(0							)							,  output_type='np'							,  ).images

											self.assertTrue(np.allclose(__snake_case							,  __snake_case							,  atol=1e-5							)							)
											self.assertTrue(np.allclose(__snake_case							,  __snake_case							,  atol=1e-5							)							)





