


import random
import unittest

import numpy as np
import torch

from diffusers import (
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    OnnxStableDiffusionUpscalePipeline,
    PNDMScheduler,
)
from diffusers.utils import floats_tensor
from diffusers.utils.testing_utils import (
    is_onnx_available,
    load_image,
    nightly,
    require_onnxruntime,
    require_torch_gpu,
)

from ..test_pipelines_onnx_common import OnnxPipelineTesterMixin


if is_onnx_available():
    import onnxruntime as ort


class     lowercase__  (					__SCREAMING_SNAKE_CASE					,							unittest.TestCase		):
  # TODO: is there an appropriate internal test set?
  A__=  'ssube/stable-diffusion-x4-upscaler-onnx'






  def 			_UpperCAmelCase  (					self      :					Dict		,						_lowercase      :					Any=0 ):




         """simple docstring"""







         UpperCAmelCase__										=     floats_tensor((1, 3, 1_28, 1_28)		,						rng=random.Random(_lowercase ) )
         UpperCAmelCase__										=     torch.manual_seed(_lowercase )
         UpperCAmelCase__										=     {
             "prompt": "A painting of a squirrel eating a burger",
             "image": image,
             "generator": generator,
             "num_inference_steps": 3,
             "guidance_scale": 7.5,
             "output_type": "numpy",
         }
         return inputs






  def 			_UpperCAmelCase  (					self      :					Optional[int] ):




         """simple docstring"""







         UpperCAmelCase__										=     OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint		,						provider="CPUExecutionProvider" )
         pipe.set_progress_bar_config(disable=_lowercase )

         UpperCAmelCase__										=     self.get_dummy_inputs()
         UpperCAmelCase__										=     pipe(**_lowercase ).images
         UpperCAmelCase__										=     image[0, -3:, -3:, -1].flatten()

         # started as 128, should now be 512
         assert image.shape == (1, 5_12, 5_12, 3)
         UpperCAmelCase__										=     np.array(
             [0.6_9_7_4_7_8_2, 0.6_8_9_0_2_0_9_3, 0.7_0_1_3_5_8_8_5, 0.7_5_8_3_6_1_8, 0.7_8_0_4_5_4_5, 0.7_8_5_4_9_1_2, 0.7_8_6_6_7_4_2_6, 0.7_8_7_4_3_8_6_3, 0.7_8_0_7_0_2_2_3] )
         assert np.abs(image_slice - expected_slice ).max() < 1E-1






  def 			_UpperCAmelCase  (					self      :					List[Any] ):




         """simple docstring"""







         UpperCAmelCase__										=     OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint		,						provider="CPUExecutionProvider" )
         UpperCAmelCase__										=     PNDMScheduler.from_config(pipe.scheduler.config		,						skip_prk_steps=_lowercase )
         pipe.set_progress_bar_config(disable=_lowercase )

         UpperCAmelCase__										=     self.get_dummy_inputs()
         UpperCAmelCase__										=     pipe(**_lowercase ).images
         UpperCAmelCase__										=     image[0, -3:, -3:, -1]

         assert image.shape == (1, 5_12, 5_12, 3)
         UpperCAmelCase__										=     np.array(
             [0.6_8_9_8_8_9_2, 0.5_9_2_4_0_5_5_6, 0.5_2_4_9_9_5_2_7, 0.5_8_8_6_6_2_1_5, 0.5_2_2_5_8_2_3_5, 0.5_2_5_7_2_7_1_5, 0.6_2_4_1_4_4_7_3, 0.6_1_7_4_3_8_7, 0.6_2_1_4_9_6_4] )
         assert np.abs(image_slice.flatten() - expected_slice ).max() < 1E-1






  def 			_UpperCAmelCase  (					self      :					List[str] ):




         """simple docstring"""







         UpperCAmelCase__										=     OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint		,						provider="CPUExecutionProvider" )
         UpperCAmelCase__										=     DPMSolverMultistepScheduler.from_config(pipe.scheduler.config )
         pipe.set_progress_bar_config(disable=_lowercase )

         UpperCAmelCase__										=     self.get_dummy_inputs()
         UpperCAmelCase__										=     pipe(**_lowercase ).images
         UpperCAmelCase__										=     image[0, -3:, -3:, -1]

         assert image.shape == (1, 5_12, 5_12, 3)
         UpperCAmelCase__										=     np.array(
             [0.7_6_5_9_2_7_8, 0.7_6_4_3_7_6_6_4, 0.7_5_5_7_9_1_0_7, 0.7_6_9_1_1_1_6, 0.7_7_6_6_6_9_8_6, 0.7_7_2_7_6_7_2, 0.7_7_5_8_6_6_4, 0.7_8_1_2_2_2_6, 0.7_6_9_4_2_5_1_5] )

         assert np.abs(image_slice.flatten() - expected_slice ).max() < 1E-1






  def 			_UpperCAmelCase  (					self      :					Optional[Any] ):




         """simple docstring"""







         UpperCAmelCase__										=     OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint		,						provider="CPUExecutionProvider" )
         UpperCAmelCase__										=     EulerDiscreteScheduler.from_config(pipe.scheduler.config )
         pipe.set_progress_bar_config(disable=_lowercase )

         UpperCAmelCase__										=     self.get_dummy_inputs()
         UpperCAmelCase__										=     pipe(**_lowercase ).images
         UpperCAmelCase__										=     image[0, -3:, -3:, -1]

         assert image.shape == (1, 5_12, 5_12, 3)
         UpperCAmelCase__										=     np.array(
             [0.6_9_7_4_7_8_2, 0.6_8_9_0_2_0_9_3, 0.7_0_1_3_5_8_8_5, 0.7_5_8_3_6_1_8, 0.7_8_0_4_5_4_5, 0.7_8_5_4_9_1_2, 0.7_8_6_6_7_4_2_6, 0.7_8_7_4_3_8_6_3, 0.7_8_0_7_0_2_2_3] )
         assert np.abs(image_slice.flatten() - expected_slice ).max() < 1E-1






  def 			_UpperCAmelCase  (					self      :					Dict ):




         """simple docstring"""







         UpperCAmelCase__										=     OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint		,						provider="CPUExecutionProvider" )
         UpperCAmelCase__										=     EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config )
         pipe.set_progress_bar_config(disable=_lowercase )

         UpperCAmelCase__										=     self.get_dummy_inputs()
         UpperCAmelCase__										=     pipe(**_lowercase ).images
         UpperCAmelCase__										=     image[0, -3:, -3:, -1]

         assert image.shape == (1, 5_12, 5_12, 3)
         UpperCAmelCase__										=     np.array(
             [0.7_7_4_2_4_4_9_6, 0.7_7_3_6_0_1, 0.7_6_4_5_2_8_8, 0.7_7_6_9_5_9_8, 0.7_7_7_2_7_3_9, 0.7_7_3_8_6_8_8, 0.7_8_1_8_7_2_3_3, 0.7_7_8_7_9_5_8_4, 0.7_6_7_0_4_3] )

         assert np.abs(image_slice.flatten() - expected_slice ).max() < 1E-1




@nightly
@require_onnxruntime
@require_torch_gpu
class     lowercase__  (					unittest.TestCase		):






  @property
  def 			_UpperCAmelCase  (					self      :					Optional[int] ):




         """simple docstring"""







         return (
             "CUDAExecutionProvider",
             {
                 "gpu_mem_limit": "15000000000",  # 15GB
                 "arena_extend_strategy": "kSameAsRequested",
             },
         )






  @property
  def 			_UpperCAmelCase  (					self      :					str ):




         """simple docstring"""







         UpperCAmelCase__										=     ort.SessionOptions()
         UpperCAmelCase__										=     False
         return options






  def 			_UpperCAmelCase  (					self      :					Optional[int] ):




         """simple docstring"""







         UpperCAmelCase__										=     load_image(
             "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
             "/img2img/sketch-mountains-input.jpg" )
         UpperCAmelCase__										=     init_image.resize((1_28, 1_28) )
         # using the PNDM scheduler by default
         UpperCAmelCase__										=     OnnxStableDiffusionUpscalePipeline.from_pretrained(
             "ssube/stable-diffusion-x4-upscaler-onnx"		,						provider=self.gpu_provider		,						sess_options=self.gpu_options		,						)
         pipe.set_progress_bar_config(disable=_lowercase )

         UpperCAmelCase__										=     "A fantasy landscape, trending on artstation"

         UpperCAmelCase__										=     torch.manual_seed(0 )
         UpperCAmelCase__										=     pipe(
             prompt=_lowercase		,						image=_lowercase		,						guidance_scale=7.5		,						num_inference_steps=10		,						generator=_lowercase		,						output_type="np"		,						)
         UpperCAmelCase__										=     output.images
         UpperCAmelCase__										=     images[0, 2_55:2_58, 3_83:3_86, -1]

         assert images.shape == (1, 5_12, 5_12, 3)
         UpperCAmelCase__										=     np.array([0.4_8_8_3, 0.4_9_4_7, 0.4_9_8_0, 0.4_9_7_5, 0.4_9_8_2, 0.4_9_8_0, 0.5_0_0_0, 0.5_0_0_6, 0.4_9_7_2] )
         # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

         assert np.abs(image_slice.flatten() - expected_slice ).max() < 2E-2






  def 			_UpperCAmelCase  (					self      :					Union[str, Any] ):




         """simple docstring"""







         UpperCAmelCase__										=     load_image(
             "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
             "/img2img/sketch-mountains-input.jpg" )
         UpperCAmelCase__										=     init_image.resize((1_28, 1_28) )
         UpperCAmelCase__										=     LMSDiscreteScheduler.from_pretrained(
             "ssube/stable-diffusion-x4-upscaler-onnx"		,						subfolder="scheduler" )
         UpperCAmelCase__										=     OnnxStableDiffusionUpscalePipeline.from_pretrained(
             "ssube/stable-diffusion-x4-upscaler-onnx"		,						scheduler=_lowercase		,						provider=self.gpu_provider		,						sess_options=self.gpu_options		,						)
         pipe.set_progress_bar_config(disable=_lowercase )

         UpperCAmelCase__										=     "A fantasy landscape, trending on artstation"

         UpperCAmelCase__										=     torch.manual_seed(0 )
         UpperCAmelCase__										=     pipe(
             prompt=_lowercase		,						image=_lowercase		,						guidance_scale=7.5		,						num_inference_steps=20		,						generator=_lowercase		,						output_type="np"		,						)
         UpperCAmelCase__										=     output.images
         UpperCAmelCase__										=     images[0, 2_55:2_58, 3_83:3_86, -1]

         assert images.shape == (1, 5_12, 5_12, 3)
         UpperCAmelCase__										=     np.array(
             [0.5_0_1_7_3_7_5_3, 0.5_0_2_2_3_3_5_6, 0.5_0_2_0_3_9, 0.5_0_2_3_3_0_3_6, 0.5_0_2_3_7_2_5, 0.5_0_2_2_6_0_1, 0.5_0_1_8_7_5_8, 0.5_0_2_3_4_0_8_5, 0.5_0_2_4_1_5_6_6] )
         # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

         assert np.abs(image_slice.flatten() - expected_slice ).max() < 2E-2









import csv

import tweepy

# Twitter API credentials
A      = ""
A      = ""
A      = ""
A      = ""





def __UpperCAmelCase       (		__A			)						->					None:



       '''simple docstring'''





       UpperCAmelCase__										=     tweepy.OAuthHandler(__A ,						__A			)
       auth.set_access_token(__A ,						__A			)
       UpperCAmelCase__										=     tweepy.API(__A			)

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

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

       # save most recent tweets
       alltweets.extend(__A			)

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

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

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

              # save most recent tweets
              alltweets.extend(__A			)

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

              print(F"""...{len(__A			)} tweets downloaded so far"""			)

       # transform the tweepy tweets into a 2D array that will populate the csv
       UpperCAmelCase__										=     [[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:
              UpperCAmelCase__										=     csv.writer(__A			)
              writer.writerow(["id", "created_at", "text"]			)
              writer.writerows(__A			)


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





