
import tempfile

import numpy as np
import torch
from transformers import AutoTokenizer, TaEncoderModel

from diffusers import DDPMScheduler, UNetaDConditionModel
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.pipelines.deepfloyd_if import IFWatermarker
from diffusers.utils.testing_utils import torch_device

from ..test_pipelines_common import to_np



class 		UpperCamelCase__  :
   def       lowerCAmelCase			(self :					str   ):
    torch.manual_seed(0   )
    __a :  Optional[Any]						=  TaEncoderModel.from_pretrained('''hf-internal-testing/tiny-random-t5'''   )

    torch.manual_seed(0   )
    __a :  str						=  AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-t5'''   )

    torch.manual_seed(0   )
    __a :  str						=  UNetaDConditionModel(
        sample_size=3_2					,				layers_per_block=1					,				block_out_channels=[3_2, 6_4]					,				down_block_types=[
            '''ResnetDownsampleBlock2D''',
            '''SimpleCrossAttnDownBlock2D''',
        ]					,				mid_block_type='''UNetMidBlock2DSimpleCrossAttn'''					,				up_block_types=['''SimpleCrossAttnUpBlock2D''', '''ResnetUpsampleBlock2D''']					,				in_channels=3					,				out_channels=6					,				cross_attention_dim=3_2					,				encoder_hid_dim=3_2					,				attention_head_dim=8					,				addition_embed_type='''text'''					,				addition_embed_type_num_heads=2					,				cross_attention_norm='''group_norm'''					,				resnet_time_scale_shift='''scale_shift'''					,				act_fn='''gelu'''					,				)
    unet.set_attn_processor(AttnAddedKVProcessor()   )  # For reproducibility tests

    torch.manual_seed(0   )
    __a :  List[str]						=  DDPMScheduler(
        num_train_timesteps=1_0_0_0					,				beta_schedule='''squaredcos_cap_v2'''					,				beta_start=0.0001					,				beta_end=0.02					,				thresholding=snake_case_					,				dynamic_thresholding_ratio=0.95					,				sample_max_value=1.0					,				prediction_type='''epsilon'''					,				variance_type='''learned_range'''					,				)

    torch.manual_seed(0   )
    __a :  Any						=  IFWatermarker()

    return {
        "text_encoder": text_encoder,
        "tokenizer": tokenizer,
        "unet": unet,
        "scheduler": scheduler,
        "watermarker": watermarker,
        "safety_checker": None,
        "feature_extractor": None,
    }
   def       lowerCAmelCase			(self :					Optional[Any]   ):
    torch.manual_seed(0   )
    __a :  Dict						=  TaEncoderModel.from_pretrained('''hf-internal-testing/tiny-random-t5'''   )

    torch.manual_seed(0   )
    __a :  Dict						=  AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-t5'''   )

    torch.manual_seed(0   )
    __a :  Optional[Any]						=  UNetaDConditionModel(
        sample_size=3_2					,				layers_per_block=[1, 2]					,				block_out_channels=[3_2, 6_4]					,				down_block_types=[
            '''ResnetDownsampleBlock2D''',
            '''SimpleCrossAttnDownBlock2D''',
        ]					,				mid_block_type='''UNetMidBlock2DSimpleCrossAttn'''					,				up_block_types=['''SimpleCrossAttnUpBlock2D''', '''ResnetUpsampleBlock2D''']					,				in_channels=6					,				out_channels=6					,				cross_attention_dim=3_2					,				encoder_hid_dim=3_2					,				attention_head_dim=8					,				addition_embed_type='''text'''					,				addition_embed_type_num_heads=2					,				cross_attention_norm='''group_norm'''					,				resnet_time_scale_shift='''scale_shift'''					,				act_fn='''gelu'''					,				class_embed_type='''timestep'''					,				mid_block_scale_factor=1.414					,				time_embedding_act_fn='''gelu'''					,				time_embedding_dim=3_2					,				)
    unet.set_attn_processor(AttnAddedKVProcessor()   )  # For reproducibility tests

    torch.manual_seed(0   )
    __a :  List[str]						=  DDPMScheduler(
        num_train_timesteps=1_0_0_0					,				beta_schedule='''squaredcos_cap_v2'''					,				beta_start=0.0001					,				beta_end=0.02					,				thresholding=snake_case_					,				dynamic_thresholding_ratio=0.95					,				sample_max_value=1.0					,				prediction_type='''epsilon'''					,				variance_type='''learned_range'''					,				)

    torch.manual_seed(0   )
    __a :  int						=  DDPMScheduler(
        num_train_timesteps=1_0_0_0					,				beta_schedule='''squaredcos_cap_v2'''					,				beta_start=0.0001					,				beta_end=0.02					,				)

    torch.manual_seed(0   )
    __a :  int						=  IFWatermarker()

    return {
        "text_encoder": text_encoder,
        "tokenizer": tokenizer,
        "unet": unet,
        "scheduler": scheduler,
        "image_noising_scheduler": image_noising_scheduler,
        "watermarker": watermarker,
        "safety_checker": None,
        "feature_extractor": None,
    }
   def       lowerCAmelCase			(self :					Any   ):
    __a :  Optional[Any]						=  self.get_dummy_components()
    __a :  str						=  self.pipeline_class(**snake_case_   )
    pipe.to(snake_case_   )
    pipe.set_progress_bar_config(disable=snake_case_   )

    __a :  Tuple						=  self.get_dummy_inputs(snake_case_   )

    __a :  int						=  inputs['''prompt''']
    __a :  Tuple						=  inputs['''generator''']
    __a :  int						=  inputs['''num_inference_steps''']
    __a :  List[str]						=  inputs['''output_type''']

    if "image" in inputs:
     __a :  Optional[int]						=  inputs['''image''']
    else:
     __a :  List[str]						=  None

    if "mask_image" in inputs:
     __a :  Union[str, Any]						=  inputs['''mask_image''']
    else:
     __a :  Union[str, Any]						=  None

    if "original_image" in inputs:
     __a :  Any						=  inputs['''original_image''']
    else:
     __a :  Union[str, Any]						=  None

    __a				,       __a :  int						=  pipe.encode_prompt(snake_case_   )

    # inputs with prompt converted to embeddings
    __a :  int						=  {
        '''prompt_embeds''': prompt_embeds,
        '''negative_prompt_embeds''': negative_prompt_embeds,
        '''generator''': generator,
        '''num_inference_steps''': num_inference_steps,
        '''output_type''': output_type,
    }

    if image is not None:
     __a :  Any						=  image

    if mask_image is not None:
     __a :  int						=  mask_image

    if original_image is not None:
     __a :  Optional[Any]						=  original_image

    # set all optional components to None
    for optional_component in pipe._optional_components:
     setattr(snake_case_					,				snake_case_					,				snake_case_   )

    __a :  Optional[Any]						=  pipe(**snake_case_   )[0]

    with tempfile.TemporaryDirectory() as tmpdir:
     pipe.save_pretrained(snake_case_   )
     __a :  str						=  self.pipeline_class.from_pretrained(snake_case_   )
     pipe_loaded.to(snake_case_   )
     pipe_loaded.set_progress_bar_config(disable=snake_case_   )

    pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()   )  # For reproducibility tests

    for optional_component in pipe._optional_components:
     self.assertTrue(
         getattr(snake_case_					,				snake_case_   ) is None					,				f"`{optional_component}` did not stay set to None after loading."					,				)

    __a :  List[Any]						=  self.get_dummy_inputs(snake_case_   )

    __a :  Dict						=  inputs['''generator''']
    __a :  Dict						=  inputs['''num_inference_steps''']
    __a :  Tuple						=  inputs['''output_type''']

    # inputs with prompt converted to embeddings
    __a :  Union[str, Any]						=  {
        '''prompt_embeds''': prompt_embeds,
        '''negative_prompt_embeds''': negative_prompt_embeds,
        '''generator''': generator,
        '''num_inference_steps''': num_inference_steps,
        '''output_type''': output_type,
    }

    if image is not None:
     __a :  Any						=  image

    if mask_image is not None:
     __a :  Dict						=  mask_image

    if original_image is not None:
     __a :  Any						=  original_image

    __a :  str						=  pipe_loaded(**snake_case_   )[0]

    __a :  int						=  np.abs(to_np(snake_case_   ) - to_np(snake_case_   )   ).max()
    self.assertLess(snake_case_					,				1E-4   )







   def       lowerCAmelCase			(self :					Optional[int]   ):
    __a :  int						=  self.get_dummy_components()
    __a :  Optional[int]						=  self.pipeline_class(**snake_case_   )
    pipe.to(snake_case_   )
    pipe.set_progress_bar_config(disable=snake_case_   )

    __a :  str						=  self.get_dummy_inputs(snake_case_   )
    __a :  List[str]						=  pipe(**snake_case_   )[0]

    with tempfile.TemporaryDirectory() as tmpdir:
     pipe.save_pretrained(snake_case_   )
     __a :  Optional[int]						=  self.pipeline_class.from_pretrained(snake_case_   )
     pipe_loaded.to(snake_case_   )
     pipe_loaded.set_progress_bar_config(disable=snake_case_   )

    pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()   )  # For reproducibility tests

    __a :  Optional[int]						=  self.get_dummy_inputs(snake_case_   )
    __a :  Dict						=  pipe_loaded(**snake_case_   )[0]

    __a :  str						=  np.abs(to_np(snake_case_   ) - to_np(snake_case_   )   ).max()
    self.assertLess(snake_case_					,				1E-4   )


def    __UpperCamelCase     (				lowerCAmelCase__							:				int = 5_0_0_0_0_0_0_0  ):
 __a :  int						=  set()
 __a :  str						=  int((limit - 2_4) ** (1 / 2)  )

 __a :  int						=  set(range(3  ,   prime_square_limit + 1  ,   2  )  )
 primes.add(2  )
 for p in range(3  ,   prime_square_limit + 1  ,   2  ):
  if p not in primes:
   continue
  primes.difference_update(set(range(p * p  ,   prime_square_limit + 1  ,   lowerCAmelCase__  )  )  )

 for primea in primes:
  __a :  Union[str, Any]						=  primea * primea
  for primea in primes:
   __a :  Union[str, Any]						=  primea * primea * primea
   if square + cube >= limit - 1_6:
    break
   for primea in primes:
    __a :  int						=  primea * primea * primea * primea
    __a :  Union[str, Any]						=  square + cube + tetr
    if total >= limit:
     break
    ret.add(lowerCAmelCase__  )

 return len(lowerCAmelCase__  )


if __name__ == "__main__":
      print(F"""{solution() = }""")
