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 {
232            parameters.insert("durationSeconds".to_owned(), seconds_value(duration));
233        }
234        if let Some(seed) = options.seed {
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.config.api_key().ok();
288        let mut videos = Vec::new();
289        let mut metadata = Vec::new();
290        for sample in samples {
291            let Some(uri) = sample
292                .get("video")
293                .and_then(|video| video.get("uri"))
294                .and_then(JsonValue::as_str)
295            else {
296                continue;
297            };
298            let Ok(mut url) = Url::parse(uri) else {
299                continue;
300            };
301            if let Some(key) = &api_key
302                && is_same_origin(&url, &self.config.base_url)
303            {
304                url.query_pairs_mut()
305                    .append_pair("key", key.expose_secret());
306            }
307            videos.push(VideoData {
308                data: FileData::Url { url },
309                media_type: MediaType::new("video/mp4"),
310            });
311            metadata.push(json!({"uri": uri}));
312        }
313        if videos.is_empty() {
314            return Err(ProviderError::InvalidResponseData(Box::new(
315                InvalidResponseDataError::new(
316                    "no valid videos in the video generation response",
317                    operation.response.clone().unwrap_or(JsonValue::Null),
318                ),
319            )));
320        }
321        let mapper = OutputMapper::new(self.config.clone(), Default::default());
322        let mut object = JsonObject::new();
323        object.insert("videos".to_owned(), JsonValue::Array(metadata));
324        Ok(VideoStatusResult::Completed {
325            videos,
326            warnings: Vec::new(),
327            provider_metadata: Some(mapper.metadata(object)),
328            response: self.response_metadata(headers),
329        })
330    }
331}
332
333impl VideoModel for GoogleVideoModel {
334    fn provider(&self) -> &ProviderId {
335        &self.provider
336    }
337
338    fn model_id(&self) -> &ModelId {
339        &self.model_id
340    }
341
342    fn max_videos_per_call(&self) -> Option<usize> {
343        Some(MAX_VIDEOS_PER_CALL)
344    }
345
346    async fn do_generate(&self, options: VideoOptions) -> Result<VideoResult, ProviderError> {
347        let _ = options;
348        Err(ProviderError::unsupported(
349            "synchronous video generation; use do_start and do_status",
350        ))
351    }
352
353    fn supports_operations(&self) -> bool {
354        true
355    }
356
357    #[tracing::instrument(skip_all, fields(model = %self.model_id))]
358    async fn do_start(
359        &self,
360        options: VideoStartOptions,
361    ) -> Result<VideoStartResult, ProviderError> {
362        let mut prepared = self.prepare_request(&options.options);
363        if options.webhook_url.is_some() {
364            prepared.warnings.push(Warning::unsupported("webhookUrl"));
365        }
366        let handlers = ResponseHandlers::new(
367            json_response_handler::<Operation>(),
368            failed_response_handler(),
369        );
370        let response = post_json(
371            self.config.transport.as_ref(),
372            self.config
373                .model_url(self.model_id.as_str(), "predictLongRunning"),
374            self.config.headers(&options.options.headers)?,
375            &prepared.body,
376            &handlers,
377            options.options.cancellation.clone(),
378        )
379        .await?;
380        let name = response.value.name.clone().ok_or_else(|| {
381            ProviderError::InvalidResponseData(Box::new(InvalidResponseDataError::new(
382                "no operation name returned from the video generation API",
383                response.raw.clone().unwrap_or(JsonValue::Null),
384            )))
385        })?;
386        Ok(VideoStartResult {
387            operation: json!({"operationName": name}),
388            warnings: prepared.warnings,
389            provider_metadata: None,
390            response: self.response_metadata(response.response_headers),
391        })
392    }
393
394    #[tracing::instrument(skip_all, fields(model = %self.model_id))]
395    async fn do_status(
396        &self,
397        options: VideoStatusOptions,
398    ) -> Result<VideoStatusResult, ProviderError> {
399        let name = options
400            .operation
401            .get("operationName")
402            .and_then(JsonValue::as_str)
403            .ok_or_else(|| {
404                ProviderError::InvalidArgument(ferrin_spec::error::InvalidArgumentError::new(
405                    "operation",
406                    "operation must contain an `operationName` string",
407                ))
408            })?;
409        let handlers = ResponseHandlers::new(
410            json_response_handler::<Operation>(),
411            failed_response_handler(),
412        );
413        let response = get(
414            self.config.transport.as_ref(),
415            self.config.url(name),
416            self.config.headers(&options.headers)?,
417            &handlers,
418            options.cancellation.clone(),
419        )
420        .await?;
421        let operation = response.value;
422        if operation.done != Some(true) {
423            return Ok(VideoStatusResult::Pending {
424                warnings: Vec::new(),
425                provider_metadata: None,
426                response: self.response_metadata(response.response_headers),
427            });
428        }
429        if let Some(error) = &operation.error {
430            return Ok(VideoStatusResult::Error {
431                error: format!(
432                    "Video generation failed: {}",
433                    error.message.as_deref().unwrap_or("unknown error")
434                ),
435                provider_metadata: None,
436                response: self.response_metadata(response.response_headers),
437            });
438        }
439        self.completed(&operation, response.response_headers)
440    }
441}
ferrin_google/video.rs

ferrin_google/
video.rs
