File size: 4,617 Bytes
0b2f980
ed7b89b
0b2f980
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
ed7b89b
0b2f980
 
 
 
 
 
 
 
 
 
 
 
ed7b89b
0b2f980
 
 
 
 
 
 
 
 
 
 
 
 
ed7b89b
0b2f980
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
import os
import gradio as gr
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 = "[email protected]: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:
        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:
        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:
        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)