

'''simple docstring'''







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        A__   :



    def        snake_case_      (       self							)    ->   List[Any]:





           '''simple docstring'''
           torch.manual_seed(0							)
           A_						       =   TaEncoderModel.from_pretrained("""hf-internal-testing/tiny-random-t5"""							)

           torch.manual_seed(0							)
           A_						       =   AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-t5"""							)

           torch.manual_seed(0							)
           A_						       =   UNetaDConditionModel(
               sample_size=32		,					layers_per_block=1		,					block_out_channels=[32, 64]		,					down_block_types=[
                   """ResnetDownsampleBlock2D""",
                   """SimpleCrossAttnDownBlock2D""",
               ]		,					mid_block_type="""UNetMidBlock2DSimpleCrossAttn"""		,					up_block_types=["""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""]		,					in_channels=3		,					out_channels=6		,					cross_attention_dim=32		,					encoder_hid_dim=32		,					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_						       =   DDPMScheduler(
               num_train_timesteps=1000		,					beta_schedule="""squaredcos_cap_v2"""		,					beta_start=0.0001		,					beta_end=0.02		,					thresholding=UpperCamelCase__		,					dynamic_thresholding_ratio=0.95		,					sample_max_value=1.0		,					prediction_type="""epsilon"""		,					variance_type="""learned_range"""		,					)

           torch.manual_seed(0							)
           A_						       =   IFWatermarker()

           return {
               "text_encoder": text_encoder,
               "tokenizer": tokenizer,
               "unet": unet,
               "scheduler": scheduler,
               "watermarker": watermarker,
               "safety_checker": None,
               "feature_extractor": None,
           }



    def        snake_case_      (       self							)    ->   Dict:





           '''simple docstring'''
           torch.manual_seed(0							)
           A_						       =   TaEncoderModel.from_pretrained("""hf-internal-testing/tiny-random-t5"""							)

           torch.manual_seed(0							)
           A_						       =   AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-t5"""							)

           torch.manual_seed(0							)
           A_						       =   UNetaDConditionModel(
               sample_size=32		,					layers_per_block=[1, 2]		,					block_out_channels=[32, 64]		,					down_block_types=[
                   """ResnetDownsampleBlock2D""",
                   """SimpleCrossAttnDownBlock2D""",
               ]		,					mid_block_type="""UNetMidBlock2DSimpleCrossAttn"""		,					up_block_types=["""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""]		,					in_channels=6		,					out_channels=6		,					cross_attention_dim=32		,					encoder_hid_dim=32		,					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=32		,					)
           unet.set_attn_processor(AttnAddedKVProcessor()							)  # For reproducibility tests

           torch.manual_seed(0							)
           A_						       =   DDPMScheduler(
               num_train_timesteps=1000		,					beta_schedule="""squaredcos_cap_v2"""		,					beta_start=0.0001		,					beta_end=0.02		,					thresholding=UpperCamelCase__		,					dynamic_thresholding_ratio=0.95		,					sample_max_value=1.0		,					prediction_type="""epsilon"""		,					variance_type="""learned_range"""		,					)

           torch.manual_seed(0							)
           A_						       =   DDPMScheduler(
               num_train_timesteps=1000		,					beta_schedule="""squaredcos_cap_v2"""		,					beta_start=0.0001		,					beta_end=0.02		,					)

           torch.manual_seed(0							)
           A_						       =   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        snake_case_      (       self							)    ->   Optional[Any]:





           '''simple docstring'''
           A_						       =   self.get_dummy_components()
           A_						       =   self.pipeline_class(**UpperCamelCase__							)
           pipe.to(UpperCamelCase__							)
           pipe.set_progress_bar_config(disable=UpperCamelCase__							)

           A_						       =   self.get_dummy_inputs(UpperCamelCase__							)

           A_						       =   inputs["""prompt"""]
           A_						       =   inputs["""generator"""]
           A_						       =   inputs["""num_inference_steps"""]
           A_						       =   inputs["""output_type"""]

           if "image" in inputs:
                  A_						       =   inputs["""image"""]
           else:
                  A_						       =   None

           if "mask_image" in inputs:
                  A_						       =   inputs["""mask_image"""]
           else:
                  A_						       =   None

           if "original_image" in inputs:
                  A_						       =   inputs["""original_image"""]
           else:
                  A_						       =   None

           A_       ,      A_						       =   pipe.encode_prompt(UpperCamelCase__							)

           # inputs with prompt converted to embeddings
           A_						       =   {
               """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_						       =   image

           if mask_image is not None:
                  A_						       =   mask_image

           if original_image is not None:
                  A_						       =   original_image

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

           A_						       =   pipe(**UpperCamelCase__							)[0]

           with tempfile.TemporaryDirectory() as tmpdir:
                  pipe.save_pretrained(UpperCamelCase__							)
                  A_						       =   self.pipeline_class.from_pretrained(UpperCamelCase__							)
                  pipe_loaded.to(UpperCamelCase__							)
                  pipe_loaded.set_progress_bar_config(disable=UpperCamelCase__							)

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

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

           A_						       =   self.get_dummy_inputs(UpperCamelCase__							)

           A_						       =   inputs["""generator"""]
           A_						       =   inputs["""num_inference_steps"""]
           A_						       =   inputs["""output_type"""]

           # inputs with prompt converted to embeddings
           A_						       =   {
               """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_						       =   image

           if mask_image is not None:
                  A_						       =   mask_image

           if original_image is not None:
                  A_						       =   original_image

           A_						       =   pipe_loaded(**UpperCamelCase__							)[0]

           A_						       =   np.abs(to_np(UpperCamelCase__							) - to_np(UpperCamelCase__							)							).max()
           self.assertLess(UpperCamelCase__		,					1e-4							)







    def        snake_case_      (       self							)    ->   int:





           '''simple docstring'''
           A_						       =   self.get_dummy_components()
           A_						       =   self.pipeline_class(**UpperCamelCase__							)
           pipe.to(UpperCamelCase__							)
           pipe.set_progress_bar_config(disable=UpperCamelCase__							)

           A_						       =   self.get_dummy_inputs(UpperCamelCase__							)
           A_						       =   pipe(**UpperCamelCase__							)[0]

           with tempfile.TemporaryDirectory() as tmpdir:
                  pipe.save_pretrained(UpperCamelCase__							)
                  A_						       =   self.pipeline_class.from_pretrained(UpperCamelCase__							)
                  pipe_loaded.to(UpperCamelCase__							)
                  pipe_loaded.set_progress_bar_config(disable=UpperCamelCase__							)

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

           A_						       =   self.get_dummy_inputs(UpperCamelCase__							)
           A_						       =   pipe_loaded(**UpperCamelCase__							)[0]

           A_						       =   np.abs(to_np(UpperCamelCase__							) - to_np(UpperCamelCase__							)							).max()
           self.assertLess(UpperCamelCase__		,					1e-4							)



'''simple docstring'''







import csv

import tweepy

# Twitter API credentials
__lowerCamelCase									=     ''''''
__lowerCamelCase									=     ''''''
__lowerCamelCase									=     ''''''
__lowerCamelCase									=     ''''''
def    UpperCAmelCase__     (   UpperCAmelCase__					)      ->      None:
       # authorize twitter, initialize tweepy
       A_						       =   tweepy.OAuthHandler(UpperCAmelCase__,							UpperCAmelCase__					)
       auth.set_access_token(UpperCAmelCase__,							UpperCAmelCase__					)
       A_						       =   tweepy.API(UpperCAmelCase__					)

       # initialize a list to hold all the tweepy Tweets
       A_						       =   []

       # make initial request for most recent tweets (200 is the maximum allowed count)
       A_						       =   api.user_timeline(screen_name=UpperCAmelCase__,							count=2_00					)

       # save most recent tweets
       alltweets.extend(UpperCAmelCase__					)

       # save the id of the oldest tweet less one
       A_						       =   alltweets[-1].id - 1

       # keep grabbing tweets until there are no tweets left to grab
       while len(UpperCAmelCase__					) > 0:
              print(F'''getting tweets before {oldest}'''					)

              # all subsequent requests use the max_id param to prevent duplicates
              A_						       =   api.user_timeline(
                  screen_name=UpperCAmelCase__,							count=2_00,							max_id=UpperCAmelCase__					)

              # save most recent tweets
              alltweets.extend(UpperCAmelCase__					)

              # update the id of the oldest tweet less one
              A_						       =   alltweets[-1].id - 1

              print(F'''...{len(UpperCAmelCase__					)} tweets downloaded so far'''					)

       # transform the tweepy tweets into a 2D array that will populate the csv
       A_						       =   [[tweet.id_str, tweet.created_at, tweet.text] for tweet in alltweets]

       # write the csv
       with open(F'''new_{screen_name}_tweets.csv''',							"""w"""					) as f:
              A_						       =   csv.writer(UpperCAmelCase__					)
              writer.writerow(["""id""", """created_at""", """text"""]					)
              writer.writerows(UpperCAmelCase__					)


if __name__ == "__main__":
      # pass in the username of the account you want to download
      get_all_tweets('''FirePing32''')
