ferrin_google/
video.rs

1//! Veo video generation (`predictLongRunning` operations).
2
3use base64::Engine;
4use ferrin_provider_util::headers::is_same_origin;
5use ferrin_provider_util::http::ResponseHandlers;
6use ferrin_provider_util::http::get;
7use ferrin_provider_util::http::json_response_handler;
8use ferrin_provider_util::http::post_json;
9use ferrin_spec::FileData;
10use ferrin_spec::JsonObject;
11use ferrin_spec::JsonValue;
12use ferrin_spec::MediaType;
13use ferrin_spec::ModelId;
14use ferrin_spec::ProviderId;
15use ferrin_spec::ResponseMetadata;
16use ferrin_spec::error::InvalidResponseDataError;
17use ferrin_spec::error::ProviderError;
18use ferrin_spec::shared::Warning;
19use ferrin_spec::video_model::FrameType;
20use ferrin_spec::video_model::VideoAspectRatio;
21use ferrin_spec::video_model::VideoData;
22use ferrin_spec::video_model::VideoFile;
23use ferrin_spec::video_model::VideoModel;
24use ferrin_spec::video_model::VideoOptions;
25use ferrin_spec::video_model::VideoResult;
26use ferrin_spec::video_model::VideoStartOptions;
27use ferrin_spec::video_model::VideoStartResult;
28use ferrin_spec::video_model::VideoStatusOptions;
29use ferrin_spec::video_model::VideoStatusResult;
30use secrecy::ExposeSecret;
31use serde::Deserialize;
32use serde_json::json;
33use url::Url;
34
35use crate::api_types::RpcStatus;
36use crate::config::CANONICAL_OPTIONS_KEY;
37use crate::config::SharedConfig;
38use crate::error::failed_response_handler;
39use crate::output::OutputMapper;
40
41/// Maximum videos per call.
42pub const MAX_VIDEOS_PER_CALL: usize = 4;
43
44/// Option keys consumed by this crate; every other key under `google` is
45/// passed through into `parameters`.
46const CONSUMED_OPTION_KEYS: [&str; 5] = [
47    "pollIntervalMs",
48    "pollTimeoutMs",
49    "personGeneration",
50    "negativePrompt",
51    "referenceImages",
52];
53
54/// A long-running operation.
55#[derive(Debug, Clone, Default, Deserialize)]
56#[serde(rename_all = "camelCase")]
57pub struct Operation {
58    /// Operation name (`models/veo-x/operations/...`).
59    #[serde(default)]
60    pub name: Option<String>,
61    /// Whether the operation finished.
62    #[serde(default)]
63    pub done: Option<bool>,
64    /// Error of a failed operation.
65    #[serde(default)]
66    pub error: Option<RpcStatus>,
67    /// Response of a finished operation.
68    #[serde(default)]
69    pub response: Option<JsonValue>,
70}
71
72/// Serializes a duration in seconds, as an integer when it has no fractional
73/// part (the API declares `durationSeconds` as an integer).
74fn seconds_value(seconds: f64) -> JsonValue {
75    if seconds.is_finite() && seconds.fract() == 0.0 && seconds >= 0.0 {
76        // `seconds` is integral and non-negative, so the conversion is exact
77        // for every value the API accepts.
78        #[allow(
79            clippy::cast_possible_truncation,
80            clippy::cast_sign_loss,
81            reason = "guarded by the integral and non-negative checks above"
82        )]
83        return JsonValue::from(seconds as u64);
84    }
85    json!(seconds)
86}
87
88fn image_object(file: &VideoFile, warnings: &mut Vec<Warning>) -> Option<JsonValue> {
89    let media_type = file
90        .media_type
91        .as_ref()
92        .map_or("image/png", MediaType::as_str)
93        .to_owned();
94    match &file.data {
95        FileData::Url { url } if url.scheme() == "gs" => {
96            Some(json!({"gcsUri": url.as_str(), "mimeType": "image/png"}))
97        }
98        FileData::Url { .. } | FileData::Reference { .. } => {
99            warnings.push(Warning::unsupported_with_details(
100                "URL-based image input",
101                "Google Generative AI video models require base64-encoded images or GCS URIs. URL will be ignored.",
102            ));
103            None
104        }
105        FileData::Bytes { data } => Some(json!({
106            "bytesBase64Encoded": base64::engine::general_purpose::STANDARD.encode(data),
107            "mimeType": media_type,
108        })),
109        FileData::Text { text } => Some(json!({
110            "bytesBase64Encoded": base64::engine::general_purpose::STANDARD.encode(text.as_bytes()),
111            "mimeType": media_type,
112        })),
113        #[allow(unreachable_patterns, reason = "FileData is non-exhaustive")]
114        _ => None,
115    }
116}
117
118fn reference_image(reference: &JsonValue) -> JsonValue {
119    if let Some(bytes) = reference
120        .get("bytesBase64Encoded")
121        .and_then(JsonValue::as_str)
122        .filter(|bytes| !bytes.is_empty())
123    {
124        return json!({"image": {"bytesBase64Encoded": bytes, "mimeType": "image/png"}, "referenceType": "asset"});
125    }
126    if let Some(uri) = reference
127        .get("gcsUri")
128        .and_then(JsonValue::as_str)
129        .filter(|uri| !uri.is_empty())
130    {
131        return json!({"image": {"gcsUri": uri, "mimeType": "image/png"}, "referenceType": "asset"});
132    }
133    reference.clone()
134}
135
136/// A prepared `predictLongRunning` request.
137#[derive(Debug, Clone, PartialEq)]
138pub struct PreparedVideoRequest {
139    /// Request body (`{instances, parameters}`).
140    pub body: JsonValue,
141    /// Warnings.
142    pub warnings: Vec<Warning>,
143}
144
145/// Veo video model.
146#[derive(Debug, Clone)]
147pub struct GoogleVideoModel {
148    config: SharedConfig,
149    provider: ProviderId,
150    model_id: ModelId,
151}
152
153impl GoogleVideoModel {
154    /// Creates the model.
155    #[must_use]
156    pub fn new(config: SharedConfig, model_id: impl Into<ModelId>) -> Self {
157        Self {
158            provider: ProviderId::new(config.name.clone()),
159            config,
160            model_id: model_id.into(),
161        }
162    }
163
164    /// Builds the request body for `options`.
165    #[must_use]
166    pub fn prepare_request(&self, options: &VideoOptions) -> PreparedVideoRequest {
167        let mut warnings = Vec::new();
168        let google: JsonObject = options
169            .provider_options
170            .get(self.config.options_key())
171            .or_else(|| options.provider_options.get(CANONICAL_OPTIONS_KEY))
172            .cloned()
173            .unwrap_or_default();
174        let mut instance = JsonObject::new();
175        if let Some(prompt) = &options.prompt {
176            instance.insert("prompt".to_owned(), JsonValue::from(prompt.as_str()));
177        }
178        let first_frame = options
179            .frame_images
180            .iter()
181            .find(|frame| frame.frame_type == FrameType::FirstFrame)
182            .map(|frame| &frame.image)
183            .or(options.image.as_ref());
184        if let Some(image) = first_frame.and_then(|file| image_object(file, &mut warnings)) {
185            instance.insert("image".to_owned(), image);
186        }
187        let last_frame = options
188            .frame_images
189            .iter()
190            .find(|frame| frame.frame_type == FrameType::LastFrame)
191            .map(|frame| &frame.image);
192        if let Some(image) = last_frame.and_then(|file| image_object(file, &mut warnings)) {
193            instance.insert("lastFrame".to_owned(), image);
194        }
195        if options.frame_images.is_empty() && !options.input_references.is_empty() {
196            let references: Vec<JsonValue> = options
197                .input_references
198                .iter()
199                .filter_map(|file| image_object(file, &mut warnings))
200                .map(|image| json!({"image": image, "referenceType": "asset"}))
201                .collect();
202            instance.insert("referenceImages".to_owned(), JsonValue::Array(references));
203        } else if let Some(JsonValue::Array(references)) = google.get("referenceImages") {
204            instance.insert(
205                "referenceImages".to_owned(),
206                JsonValue::Array(references.iter().map(reference_image).collect()),
207            );
208        }
209        let mut parameters = JsonObject::new();
210        parameters.insert("sampleCount".to_owned(), JsonValue::from(options.n));
211        match &options.aspect_ratio {
212            Some(VideoAspectRatio::Ratio(ratio)) => {
213                parameters.insert("aspectRatio".to_owned(), JsonValue::from(ratio.to_string()));
214            }
215            Some(VideoAspectRatio::Adaptive) => {
216                parameters.insert("aspectRatio".to_owned(), JsonValue::from("adaptive"));
217            }
218            #[allow(unreachable_patterns, reason = "VideoAspectRatio is non-exhaustive")]
219            Some(_) => {}
220            None => {}
221        }
222        if let Some(resolution) = &options.resolution {
223            let mapped = match (resolution.width, resolution.height) {
224                (1280, 720) => "720p".to_owned(),
225                (1920, 1080) => "1080p".to_owned(),
226                (3840, 2160) => "4k".to_owned(),
227                _ => resolution.to_string(),
228            };
229            parameters.insert("resolution".to_owned(), JsonValue::from(mapped));
230        }
231        if let Some(duration) = options.duration.filter(|duration| *duration != 0.0) {
232            parameters.insert("durationSeconds".to_owned(), seconds_value(duration));
233        }
234        if let Some(seed) = options.seed.filter(|seed| *seed != 0) {
235            parameters.insert("seed".to_owned(), JsonValue::from(seed));
236        }
237        if options.fps.is_some() {
238            warnings.push(Warning::unsupported("fps"));
239        }
240        if options.generate_audio.is_some() {
241            warnings.push(Warning::unsupported("generateAudio"));
242        }
243        for key in ["personGeneration", "negativePrompt"] {
244            if let Some(value) = google.get(key).filter(|value| !value.is_null()) {
245                parameters.insert(key.to_owned(), value.clone());
246            }
247        }
248        for (key, value) in &google {
249            if !CONSUMED_OPTION_KEYS.contains(&key.as_str()) {
250                parameters.insert(key.clone(), value.clone());
251            }
252        }
253        PreparedVideoRequest {
254            body: json!({"instances": [instance], "parameters": parameters}),
255            warnings,
256        }
257    }
258
259    fn response_metadata(&self, headers: ferrin_spec::Headers) -> ResponseMetadata {
260        ResponseMetadata {
261            id: None,
262            timestamp: Some(chrono::Utc::now()),
263            model_id: Some(self.model_id.clone()),
264            headers: Some(headers),
265            body: None,
266        }
267    }
268
269    fn completed(
270        &self,
271        operation: &Operation,
272        headers: ferrin_spec::Headers,
273    ) -> Result<VideoStatusResult, ProviderError> {
274        let samples = operation
275            .response
276            .as_ref()
277            .and_then(|response| response.get("generateVideoResponse"))
278            .and_then(|response| response.get("generatedSamples"))
279            .and_then(JsonValue::as_array)
280            .filter(|samples| !samples.is_empty())
281            .ok_or_else(|| {
282                ProviderError::InvalidResponseData(Box::new(InvalidResponseDataError::new(
283                    "no videos in the video generation response",
284                    operation.response.clone().unwrap_or(JsonValue::Null),
285                )))
286            })?;
287        let api_key = self
288            .config
289            .headers(&ferrin_spec::Headers::new())
290            .ok()
291            .and_then(|headers| {
292                headers
293                    .get_str(crate::config::API_KEY_HEADER)
294                    .map(|key| secrecy::SecretString::from(key.to_owned()))
295            });
296        let mut videos = Vec::new();
297        let mut metadata = Vec::new();
298        for sample in samples {
299            let Some(uri) = sample
300                .get("video")
301                .and_then(|video| video.get("uri"))
302                .and_then(JsonValue::as_str)
303            else {
304                continue;
305            };
306            let Ok(mut url) = Url::parse(uri) else {
307                continue;
308            };
309            if let Some(key) = &api_key
310                && is_same_origin(&url, &self.config.base_url)
311            {
312                url.query_pairs_mut()
313                    .append_pair("key", key.expose_secret());
314            }
315            videos.push(VideoData {
316                data: FileData::Url { url },
317                media_type: MediaType::new("video/mp4"),
318            });
319            metadata.push(json!({"uri": uri}));
320        }
321        if videos.is_empty() {
322            return Err(ProviderError::InvalidResponseData(Box::new(
323                InvalidResponseDataError::new(
324                    "no valid videos in the video generation response",
325                    operation.response.clone().unwrap_or(JsonValue::Null),
326                ),
327            )));
328        }
329        let mapper = OutputMapper::new(self.config.clone(), Default::default());
330        let mut object = JsonObject::new();
331        object.insert("videos".to_owned(), JsonValue::Array(metadata));
332        Ok(VideoStatusResult::Completed {
333            videos,
334            warnings: Vec::new(),
335            provider_metadata: Some(mapper.metadata(object)),
336            response: self.response_metadata(headers),
337        })
338    }
339}
340
341impl VideoModel for GoogleVideoModel {
342    fn provider(&self) -> &ProviderId {
343        &self.provider
344    }
345
346    fn model_id(&self) -> &ModelId {
347        &self.model_id
348    }
349
350    fn max_videos_per_call(&self) -> Option<usize> {
351        Some(MAX_VIDEOS_PER_CALL)
352    }
353
354    async fn do_generate(&self, options: VideoOptions) -> Result<VideoResult, ProviderError> {
355        let _ = options;
356        Err(ProviderError::unsupported(
357            "synchronous video generation; use do_start and do_status",
358        ))
359    }
360
361    fn supports_operations(&self) -> bool {
362        true
363    }
364
365    #[tracing::instrument(skip_all, fields(model = %self.model_id))]
366    async fn do_start(
367        &self,
368        options: VideoStartOptions,
369    ) -> Result<VideoStartResult, ProviderError> {
370        let mut prepared = self.prepare_request(&options.options);
371        if options.webhook_url.is_some() {
372            prepared.warnings.push(Warning::unsupported("webhookUrl"));
373        }
374        let handlers = ResponseHandlers::new(
375            json_response_handler::<Operation>(),
376            failed_response_handler(),
377        );
378        let response = post_json(
379            self.config.transport.as_ref(),
380            self.config
381                .model_url(self.model_id.as_str(), "predictLongRunning"),
382            self.config.headers(&options.options.headers)?,
383            &prepared.body,
384            &handlers,
385            options.options.cancellation.clone(),
386        )
387        .await?;
388        let name = response.value.name.clone().ok_or_else(|| {
389            ProviderError::InvalidResponseData(Box::new(InvalidResponseDataError::new(
390                "no operation name returned from the video generation API",
391                response.raw.clone().unwrap_or(JsonValue::Null),
392            )))
393        })?;
394        Ok(VideoStartResult {
395            operation: json!({"operationName": name}),
396            warnings: prepared.warnings,
397            provider_metadata: None,
398            response: self.response_metadata(response.response_headers),
399        })
400    }
401
402    #[tracing::instrument(skip_all, fields(model = %self.model_id))]
403    async fn do_status(
404        &self,
405        options: VideoStatusOptions,
406    ) -> Result<VideoStatusResult, ProviderError> {
407        let name = options
408            .operation
409            .get("operationName")
410            .and_then(JsonValue::as_str)
411            .ok_or_else(|| {
412                ProviderError::InvalidArgument(ferrin_spec::error::InvalidArgumentError::new(
413                    "operation",
414                    "operation must contain an `operationName` string",
415                ))
416            })?;
417        let handlers = ResponseHandlers::new(
418            json_response_handler::<Operation>(),
419            failed_response_handler(),
420        );
421        let response = get(
422            self.config.transport.as_ref(),
423            self.config.url(name),
424            self.config.headers(&options.headers)?,
425            &handlers,
426            options.cancellation.clone(),
427        )
428        .await?;
429        let operation = response.value;
430        if operation.done != Some(true) {
431            return Ok(VideoStatusResult::Pending {
432                warnings: Vec::new(),
433                provider_metadata: None,
434                response: self.response_metadata(response.response_headers),
435            });
436        }
437        if let Some(error) = &operation.error {
438            return Ok(VideoStatusResult::Error {
439                error: format!(
440                    "Video generation failed: {}",
441                    error.message.as_deref().unwrap_or("unknown error")
442                ),
443                provider_metadata: None,
444                response: self.response_metadata(response.response_headers),
445            });
446        }
447        self.completed(&operation, response.response_headers)
448    }
449}
ferrin_google/video.rs

ferrin_google/
video.rs
