Update app.py

app.py

@@ -17,7 +17,161 @@ async def get_camera_motions():
         print(f"Error fetching camera motions: {str(e)}")
         return []
 
-# 나머지 함수들 (generate_video, text_to_video, image_to_video)은 변경하지 않습니다...
+async def generate_video(client, prompt, loop=False, aspect_ratio="16:9", camera_motion=None, extend_id=None, reverse_extend_id=None, interpolate_ids=None, progress=gr.Progress()):
+    generation_params = {
+        "prompt": prompt,
+        "loop": loop,
+        "aspect_ratio": aspect_ratio
+    }
+
+    if camera_motion:
+        generation_params["prompt"] += f" {camera_motion}"
+
+    if extend_id:
+        generation_params["keyframes"] = {
+            "frame0": {
+                "type": "generation",
+                "id": extend_id
+            }
+        }
+    elif reverse_extend_id:
+        generation_params["keyframes"] = {
+            "frame1": {
+                "type": "generation",
+                "id": reverse_extend_id
+            }
+        }
+    elif interpolate_ids:
+        generation_params["keyframes"] = {
+            "frame0": {
+                "type": "generation",
+                "id": interpolate_ids[0]
+            },
+            "frame1": {
+                "type": "generation",
+                "id": interpolate_ids[1]
+            }
+        }
+
+    progress(0, desc="Initiating video generation...")
+    generation = await client.generations.create(**generation_params)
+    
+    progress(0.1, desc="Video generation started. Waiting for completion...")
+    
+    # Poll for completion
+    start_time = asyncio.get_event_loop().time()
+    while True:
+        status = await client.generations.get(id=generation.id)
+        if status.state == "completed":
+            break
+        elif status.state == "failed":
+            raise Exception("Video generation failed")
+        
+        # Update progress based on time elapsed (assuming 60 seconds total)
+        elapsed_time = asyncio.get_event_loop().time() - start_time
+        progress_value = min(0.1 + (elapsed_time / 60) * 0.8, 0.9)
+        progress(progress_value, desc="Generating video...")
+        
+        await asyncio.sleep(5)
+
+    progress(0.9, desc="Downloading generated video...")
+    
+    # Download the video
+    video_url = status.assets.video
+    async with aiohttp.ClientSession() as session:
+        async with session.get(video_url) as resp:
+            if resp.status == 200:
+                file_name = f"luma_ai_generated_{generation.id}.mp4"
+                with open(file_name, 'wb') as fd:
+                    while True:
+                        chunk = await resp.content.read(1024)
+                        if not chunk:
+                            break
+                        fd.write(chunk)
+    
+    progress(1.0, desc="Video generation complete!")
+    return file_name
+
+async def text_to_video(prompt, loop, aspect_ratio, camera_motion, extend_id, reverse_extend_id, interpolate_id1, interpolate_id2, progress=gr.Progress()):
+    api_key = os.getenv("LMGEN_KEY")
+    if not api_key:
+        raise gr.Error("LMGEN_KEY environment variable is not set.")
+    
+    client = AsyncLumaAI(auth_token=api_key)
+    
+    try:
+        interpolate_ids = None
+        if interpolate_id1 and interpolate_id2:
+            interpolate_ids = [interpolate_id1, interpolate_id2]
+
+        video_path = await generate_video(
+            client, prompt, loop, aspect_ratio, camera_motion, 
+            extend_id, reverse_extend_id, interpolate_ids, progress
+        )
+        return video_path, ""
+    except Exception as e:
+        return None, f"An error occurred: {str(e)}"
+
+async def image_to_video(prompt, image_url, loop, aspect_ratio, camera_motion, progress=gr.Progress()):
+    api_key = os.getenv("LMGEN_KEY")
+    if not api_key:
+        raise gr.Error("LMGEN_KEY environment variable is not set.")
+    
+    client = AsyncLumaAI(auth_token=api_key)
+    
+    try:
+        generation_params = {
+            "prompt": prompt + (f" {camera_motion}" if camera_motion else ""),
+            "loop": loop,
+            "aspect_ratio": aspect_ratio,
+            "keyframes": {
+                "frame0": {
+                    "type": "image",
+                    "url": image_url
+                }
+            }
+        }
+        
+        progress(0, desc="Initiating video generation from image...")
+        generation = await client.generations.create(**generation_params)
+        
+        progress(0.1, desc="Video generation started. Waiting for completion...")
+        
+        # Poll for completion
+        start_time = asyncio.get_event_loop().time()
+        while True:
+            status = await client.generations.get(id=generation.id)
+            if status.state == "completed":
+                break
+            elif status.state == "failed":
+                raise Exception("Video generation failed")
+            
+            # Update progress based on time elapsed (assuming 60 seconds total)
+            elapsed_time = asyncio.get_event_loop().time() - start_time
+            progress_value = min(0.1 + (elapsed_time / 60) * 0.8, 0.9)
+            progress(progress_value, desc="Generating video...")
+            
+            await asyncio.sleep(5)
+
+        progress(0.9, desc="Downloading generated video...")
+        
+        # Download the video
+        video_url = status.assets.video
+        async with aiohttp.ClientSession() as session:
+            async with session.get(video_url) as resp:
+                if resp.status == 200:
+                    file_name = f"luma_ai_generated_{generation.id}.mp4"
+                    with open(file_name, 'wb') as fd:
+                        while True:
+                            chunk = await resp.content.read(1024)
+                            if not chunk:
+                                break
+                            fd.write(chunk)
+        
+        progress(1.0, desc="Video generation complete!")
+        return file_name, ""
+    except Exception as e:
+        return None, f"An error occurred: {str(e)}"
 
 with gr.Blocks() as demo:
     gr.Markdown("# Luma AI Text-to-Video Demo")