File size: 4,924 Bytes
957256e 26cd782 957256e 26cd782 957256e 26cd782 957256e 26cd782 957256e 26cd782 ed5dd8e 26cd782 957256e |
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 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 |
"""
Video Service
High-level service for video generation.
Abstracts all API complexity from the UI layer.
"""
from typing import Callable, Optional, List
from dataclasses import dataclass
from ..api.client import StackNetClient
@dataclass
class GeneratedVideo:
"""Generated video result."""
video_url: str
video_path: Optional[str] = None
thumbnail_url: Optional[str] = None
duration: Optional[float] = None
prompt: Optional[str] = None
class VideoService:
"""
Service for video generation.
Provides clean interfaces for:
- Text-to-video generation
- Image-to-video animation
"""
def __init__(self, client: Optional[StackNetClient] = None):
self.client = client or StackNetClient()
async def generate_video(
self,
prompt: str,
duration: int = 10,
style: Optional[str] = None,
on_progress: Optional[Callable[[float, str], None]] = None
) -> List[GeneratedVideo]:
"""
Generate video from a text prompt using generate_video_2 tool.
Args:
prompt: Description of desired video
duration: Target duration in seconds
style: Style preset (Cinematic, Animation, etc.)
on_progress: Callback for progress updates
Returns:
List of generated videos
"""
full_prompt = prompt
if style and style != "Cinematic":
full_prompt = f"{prompt}, {style.lower()} style"
result = await self.client.submit_tool_task(
tool_name="generate_video_2",
parameters={
"prompt": full_prompt,
"duration": duration
},
on_progress=on_progress
)
if not result.success:
raise Exception(result.error or "Video generation failed")
return self._parse_video_result(result.data, prompt)
async def animate_image(
self,
image_url: str,
motion_prompt: str,
duration: int = 5,
on_progress: Optional[Callable[[float, str], None]] = None
) -> List[GeneratedVideo]:
"""
Animate a static image into video using generate_image_to_video_2 tool.
Args:
image_url: URL to source image
motion_prompt: Description of desired motion
duration: Target duration in seconds
on_progress: Progress callback
Returns:
List of animated videos
"""
result = await self.client.submit_tool_task(
tool_name="generate_image_to_video_2",
parameters={
"prompt": motion_prompt,
"image_url": image_url,
"duration": duration
},
on_progress=on_progress
)
if not result.success:
raise Exception(result.error or "Image animation failed")
return self._parse_video_result(result.data, motion_prompt)
def _parse_video_result(self, data: dict, prompt: str) -> List[GeneratedVideo]:
"""Parse API response into GeneratedVideo objects."""
videos = []
# Handle various response formats
raw_videos = data.get("videos", [])
if not raw_videos:
# Check for single video URL
video_url = (
data.get("video_url") or
data.get("videoUrl") or
data.get("url") or
data.get("content")
)
if video_url:
raw_videos = [{"url": video_url}]
for vid_data in raw_videos:
if isinstance(vid_data, str):
video_url = vid_data
else:
video_url = (
vid_data.get("url") or
vid_data.get("video_url") or
vid_data.get("videoUrl")
)
if video_url:
videos.append(GeneratedVideo(
video_url=video_url,
thumbnail_url=vid_data.get("thumbnail") if isinstance(vid_data, dict) else None,
duration=vid_data.get("duration") if isinstance(vid_data, dict) else None,
prompt=prompt
))
return videos
async def download_video(self, video: GeneratedVideo) -> str:
"""Download a video to local file."""
if video.video_path:
return video.video_path
# Determine extension from URL
url = video.video_url
if ".webm" in url:
ext = ".webm"
elif ".mov" in url:
ext = ".mov"
else:
ext = ".mp4"
filename = f"video_{hash(url) % 10000}{ext}"
video.video_path = await self.client.download_file(url, filename)
return video.video_path
def cleanup(self):
"""Clean up temporary files."""
self.client.cleanup()
|