Create app.py

app.py

@@ -0,0 +1,154 @@
+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("cuda")
+
+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)
+    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
+    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)
+    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)
+%cd /content/temp