Update app.py

Added uplink support

app.py

@@ -1,153 +1,115 @@
+import os
 import gradio as gr
-from gradio import Theme
-
-import os, cv2, torch, time, random
-import numpy as np
-from moviepy.editor import *
-import boto3
-from botocore.client import Config
-from botocore.exceptions import NoCredentialsError
-
-from diffusers import DiffusionPipeline, EulerAncestralDiscreteScheduler
-from PIL import Image
-from yt_dlp import YoutubeDL
-
-pipe = DiffusionPipeline.from_pretrained("timbrooks/instruct-pix2pix", torch_dtype=torch.float16, safety_checker=None)
-pipe.scheduler = EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config)
-pipe.enable_xformers_memory_efficient_attention()
-pipe.unet.to(memory_format=torch.channels_last)
-pipe = pipe.to("CPU")
-
-s3_access_key = "juj22qxqxql7u2pl6nomgxu3ip7a"
-s3_secret_key = "j3uwidtozhboy5vczhymhzkkjsaumznnqlzck5zjs5qxgsung4ukk"
-s3_endpoint = "https://gateway.storjshare.io"
-
-s3 = boto3.client("s3", aws_access_key_id=s3_access_key, aws_secret_access_key=s3_secret_key, endpoint_url=s3_endpoint, config=Config(signature_version="s3v4"))
-
-my_theme = Theme.from_hub("bethecloud/storj_theme")
-
-def download_video(url):
-    ydl_opts = {'overwrites':True, 'format':'bestvideo[ext=mp4]+bestaudio[ext=m4a]/mp4', 'outtmpl':'/content/video.mp4'}
-    with YoutubeDL(ydl_opts) as ydl:
-        ydl.download(url)
-        return f"/content/video.mp4"
-
-def pix2pix(prompt, text_guidance_scale, image_guidance_scale, image, steps, neg_prompt="", width=512, height=512, seed=0):
-    if seed == 0:
-        seed = random.randint(0, 2147483647)
-    generator = torch.Generator("cuda").manual_seed(seed)
+from modelscope.pipelines import pipeline
+from modelscope.outputs import OutputKeys
+import tempfile
+import uplink_python
+from uplink_python.uplink import Uplink
+from uplink_python.module_classes import Permission, SharePrefix, Config
+import datetime
+
+# Define the text to video synthesis pipeline
+p = pipeline("text-to-video-synthesis", "damo/text-to-video-synthesis")
+
+# Storj configuration information
+MY_API_KEY = "1H9SGajiNtFMJtjoseMKuhiLpuPHq3Sy7of6eNUjs8yeRE1DkmRF4XJsTm1ExXLbsxqJ4t5fRZobh4tyFmEdocNCFsJhukakve5C4V59TEEzqTaLVdjFc7ovCkmuRGotR9c2aqDdt5RThRTtKcXaVE5uCtuFQ9tZvnsX"
+MY_SATELLITE = "12EayRS2V1kEsWESU9QMRseFhdxYxKicsiFmxrsLZHeLUtdps3S@us1.storj.io:7777"
+MY_BUCKET = "huggingface-demo"
+MY_ENCRYPTION_PASSPHRASE = "."
+
+# Initialize Storj Uplink
+uplink = Uplink()
+
+# Implement the generate_video function
+def generate_video(text_input: str, duration: int) -> str:
+    test_text = {'text': text_input, 'duration': duration}
+    output_video_path = p(test_text,)[OutputKeys.OUTPUT_VIDEO]
+
+    access = uplink.request_access_with_passphrase(MY_SATELLITE, MY_API_KEY, MY_ENCRYPTION_PASSPHRASE)
+    project = access.open_project()
+    project.ensure_bucket(MY_BUCKET)
+    video_key = os.path.basename(output_video_path)
+    with open(output_video_path, "rb") as f:
+        upload = project.upload_object(MY_BUCKET, video_key)
+        upload.write_file(f)
+        upload.commit()
+
+    permissions = Permission(allow_download=True)
+    shared_prefix = [SharePrefix(bucket=MY_BUCKET, prefix=video_key)]
+    shared_access = access.share(permissions, shared_prefix)
+    serialized_shared_access = shared_access.serialize()
+    deserialized_shared_access = uplink.parse_access(serialized_shared_access)
+    project_with_shared_access = deserialized_shared_access.open_project()
+
+    expiration = datetime.timedelta(hours=1)
+    expires_at = (datetime.datetime.utcnow() + expiration).timestamp()
+    storj_video_url = f"{MY_SATELLITE}/{MY_BUCKET}/{video_key}?access={serialized_shared_access}&expiresAt={expires_at}"
+    download = project_with_shared_access.download_object(MY_BUCKET, video_key)
+
+    chunk_size = 1024 * 1024
+    temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
     try:
-        image = Image.open(image)
-        ratio = min(height / image.height, width / image.width)
-        image = image.resize((int(image.width * ratio), int(image.height * ratio)), Image.LANCZOS)
-        result = pipe(
-            prompt,
-            negative_prompt=neg_prompt,
-            image=image,
-            num_inference_steps=int(steps),
-            image_guidance_scale=image_guidance_scale,
-            guidance_scale=text_guidance_scale,
-            generator=generator,
-        )
-        return result.images, result.nsfw_content_detected, seed
+        with open(temp_file.name, "wb") as dst:
+            while True:
+                try:
+                    chunk = download.read(chunk_size)  # Read a chunk from the download
+                except Exception as e:
+                    print(f"Error while reading from download: {e}")
+                    break
+
+                if not chunk:
+                    break
+
+                dst.write(chunk)  # Write the chunk to the destination file
     except Exception as e:
-        return None, None, error_str(e)
-
-def error_str(error, title="Error"):
-    return (
-        f"""#### {title}
-            {error}"""
-        if error
-        else ""
-    )
-
-def get_frames(video_in):
-    frames = []
-    clip = VideoFileClip(video_in)
-    if clip.fps > 30:
-        print("vide rate is over 30, resetting to 30")
-        clip_resized = clip.resize(height=512)
-        clip_resized.write_videofile("video_resized.mp4", fps=30, verbose=False)
+        print(f"Error while writing to file: {e}")
+
+    temp_file.seek(0)
+    project.close()
+
+    return temp_file.name, storj_video_url
+
+# Define the tweet_video function
+def tweet_video(text_input: str, duration: int, button_click: bool) -> str:
+    if button_click:
+        tweet_text = f"Check out this amazing video generated by Text-to-Video Synthesis! {text_input}"
+        tweet_url = f"https://twitter.com/intent/tweet?text={tweet_text}"
+        return tweet_url
     else:
-        print("video rate is OK")
-        clip_resized = clip.resize(height=512)
-        clip_resized.write_videofile("video_resized.mp4", fps=clip.fps, verbose=False)
-    print("video resized to 512 height")
-    cap= cv2.VideoCapture("video_resized.mp4")
-    fps = cap.get(cv2.CAP_PROP_FPS)
-    print("video fps: " + str(fps))
-    i=0
-    while(cap.isOpened()):
-        ret, frame = cap.read()
-        if ret == False:
-            break
-        cv2.imwrite('in'+str(i)+'.jpg',frame)
-        frames.append('in'+str(i)+'.jpg')
-        i+=1
-    cap.release()
-    cv2.destroyAllWindows()
-    print("broke the video into frames")
-    return frames, fps
-
-def create_video(frames, fps):
-    clips = [ImageClip(m).set_duration(1 / fps) for m in frames]
-    concat_clip = concatenate_videoclips(clips, method="compose")
-    concat_clip.write_videofile("/content/output.mp4", fps=fps, verbose=False)
-    return "/content/output.mp4"
-
-
-def infer(prompt,video_in, seed_in, trim_value):
-    print(prompt)
-    break_vid = get_frames(video_in)
-    frames_list= break_vid[0]
-    fps = break_vid[1]
-    n_frame = int(trim_value*fps)
-    if n_frame >= len(frames_list):
-        print("video is shorter than the cut value")
-        n_frame = len(frames_list)
-    result_frames = []
-    print("set stop frames to: " + str(n_frame))
-    for i in frames_list[0:int(n_frame)]:
-        pix2pix_img = pix2pix(prompt,5.5,1.5,i,15,"",512,512,seed_in)
-        images = pix2pix_img[0]
-        rgb_im = images[0].convert("RGB")
-        rgb_im.save(f"out-{i}.jpg")
-        result_frames.append(f"out-{i}.jpg")
-        print("frame " + i + "/" + str(n_frame) + ": done;")
-    final_vid = create_video(result_frames, fps)
-    print("Done!")
-    return final_vid
-
-def upload_to_storj(bucket_name, file_path):
-    try:
-        file_name = os.path.basename(file_path)
-        s3.upload_file(file_path, bucket_name, file_name)
-        print(f"{file_name} uploaded to {bucket_name}")
-    except NoCredentialsError:
-        print("Credentials not available")
-
-with gr.Blocks(theme=my_theme) as demo:
-    with gr.Column(elem_id="col-container"):
-        with gr.Row():
-            with gr.Column():
-                input_text = gr.Textbox(show_label=False, value="https://link.storjshare.io/s/jwn3ljgyj4ynlg6qtqj6yrbo63ha/demo/dancing.mp4?view")
-                input_download_button = gr.Button(value="Enter a link from Storj Linkshare")
-                prompt = gr.Textbox(label="Prompt", placeholder="Enter new style of video", show_label=False, elem_id="prompt-in")
-                video_inp = gr.Video(label="Video source", source="upload", type="filepath", elem_id="input-vid")
-                input_download_button.click(download_video, inputs=[input_text], outputs=[video_inp])
-            with gr.Column():
-                video_out = gr.Video(label="Pix2pix video result", type="filepath", elem_id="video-output")
-                with gr.Row():
-                  seed_inp = gr.Slider(label="Seed", minimum=0, maximum=2147483647, step=1, value=69)
-                  trim_in = gr.Slider(label="Cut video at (s)", minimun=1, maximum=600, step=1, value=1)
-                submit_btn = gr.Button("Generate transformed video with Storj")
-        inputs = [prompt,video_inp,seed_inp, trim_in]
-        def process_and_upload(prompt, video_inp, seed_inp, trim_in):
-            final_vid = infer(prompt, video_inp, seed_inp, trim_in)
-            # Specify the name of the bucket to upload to
-            storj_bucket_name = "huggingface-demo"
-            upload_to_storj(storj_bucket_name, final_vid)
-            return final_vid
-
-        submit_btn.click(process_and_upload, inputs=inputs, outputs=[video_out])
-demo.queue().launch(debug=True, share=True, inline=False)
+        return "Hello " + text_input + "!"
+
+# Set theme
+my_theme = gr.Theme.from_hub("bethecloud/storj_theme")
+
+# Define the input components
+text_input = gr.inputs.Textbox(lines=3, placeholder="Enter a short text to generate video")
+duration = gr.inputs.Slider(minimum=1, maximum=10, default=5, label="Duration (seconds)")
+button = gr.inputs.Button(label="Share on Twitter")
+
+# Define the output components
+video_output = gr.outputs.Video(label="Generated Video")
+url_output = gr.outputs.Textbox(label="Storj Video URL")
+
+# Create the Gradio interface
+iface = gr.Interface(
+    fn=tweet_video,
+    inputs=[text_input, duration],
+    outputs=[video_output, url_output],
+    title="Text-to-Video Synthesis",
+    description="Generate a video based on the input text and share it on Twitter.",
+    theme=my_theme
+)
+
+# Define the action for the "Share on Twitter" button
+def on_button_click():
+    storj_video_url = iface.get_component("Storj Video URL").data
+    text_input = iface.get_component("Enter a short text to generate video").data
+    duration = iface.get_component("Duration (seconds)").data
+    tweet_url = tweet_video(text_input, duration, True)
+    gr.Interface.open_url(tweet_url)
+
+# Add the "Share on Twitter" button and its action to the interface
+iface.set_component("Share on Twitter", gr.Button(on_button_click, label="Share on Twitter"))
+
+# Launch the interface
+iface.launch(share=True, debug=True)