in experiments/veo-app/pages/portraits.py [0:0]
def on_click_motion_portraits(e: me.ClickEvent):
"""Create the motion portrait"""
state = me.state(PageState)
if not state.reference_image_gcs:
print("No reference image uploaded or GCS URI is missing.")
state.error_message = "Please upload a reference image first."
state.show_error_dialog = True
state.is_loading = False
yield
return
state.is_loading = True
state.show_error_dialog = False
state.error_message = ""
state.result_video = ""
state.timing = ""
state.generated_scene_direction = ""
yield
# get scene direction
base_prompt = f"""Scene direction for a motion portrait for an approximately {state.video_length} second scene.
Expand the given direction to include more facial engagement, as if the subject is looking out of the image and interested in the world outside.
Examine the picture provided to improve the scene direction.
Optionally, include is waving of hands and if necessary, and physical motion outside the frame.
Do not describe the frame. There should be no lip movement like speaking, but there can be descriptions of facial movements such as laughter, either in joy or cruelty."""
final_prompt_for_llm = base_prompt
if state.modifier_array:
modifiers_string = ", ".join(state.modifier_array)
final_prompt_for_llm += (
f"\n\nUtilize the following modifiers for the subject: {modifiers_string}."
)
final_prompt_for_llm += "\n\nScene direction:\n" # Guide for the LLM
try:
print(
f"Generating scene direction for {state.reference_image_gcs} with prompt:\n{final_prompt_for_llm}"
)
# The scene_direction returned here is what we'll use for the video model
scene_direction_for_video = generate_scene_direction(
final_prompt_for_llm,
state.reference_image_gcs,
state.reference_image_mime_type,
)
state.generated_scene_direction = (
scene_direction_for_video # Store the generated direction
)
print(f"Generated Scene Direction (for video):\n{scene_direction_for_video}")
yield
print("Lights, camera, action!")
aspect_ratio = state.aspect_ratio # @param ["16:9", "9:16"]
seed = 120
sample_count = 1
rewrite_prompt = state.auto_enhance_prompt
if rewrite_prompt:
print("Default auto-enhance prompt is ON")
duration_seconds = state.video_length
# invoke i2v
start_time = time.time() # Record the starting time
gcs_uri = ""
current_error_message = ""
print(f"I2V invoked. I see you have an image! {state.reference_image_gcs} ")
op = image_to_video(
# state.veo_prompt_input,
scene_direction_for_video,
state.reference_image_gcs,
seed,
aspect_ratio,
sample_count,
f"gs://{config.VIDEO_BUCKET}",
rewrite_prompt,
duration_seconds,
)
print(f"I2V Operation result: {op}")
print_keys(op) # Useful for debugging response structure
# Check for explicit errors in response
if op.get("done"):
if op.get("error"):
current_error_message = op["error"].get(
"message", "Unknown API error during video generation."
)
print(f"API Error Detected: {current_error_message}")
elif op.get("response"):
response_data = op["response"]
print(f"Response: {response_data}")
print_keys(op["response"])
if response_data.get(
"raiMediaFilteredCount", 0
) > 0 and response_data.get("raiMediaFilteredReasons"):
# Extract the first reason provided
filter_reason = response_data["raiMediaFilteredReasons"][0]
current_error_message = f"Content Filtered: {filter_reason}"
print(f"Filtering Detected: {current_error_message}")
elif response_data.get("generatedSamples") and response_data[
"generatedSamples"
][0].get("video", {}).get("uri"):
gcs_uri = response_data["generatedSamples"][0]["video"]["uri"]
elif response_data.get("videos") and response_data["videos"][0].get(
"gcsUri"
):
gcs_uri = response_data["videos"][0]["gcsUri"]
else:
current_error_message = (
"API reported success but no video URI was found."
)
# else:
# # Extract GCS URI from different possible locations
# if (
# "generatedSamples" in response_data
# and response_data["generatedSamples"]
# ):
# print(f"Generated Samples: {response_data["generatedSamples"]}")
# gcs_uri = (
# response_data["generatedSamples"][0]
# .get("video", {})
# .get("uri", "")
# )
# elif "videos" in response_data and response_data["videos"]:
# print(f"Videos: {response_data["videos"]}")
# gcs_uri = response_data["videos"][0].get("gcsUri", "")
if gcs_uri: # if GCS URI, set to state
state.result_video = gcs_uri
file_name = gcs_uri.split("/")[-1]
print(
f"Video generated: {gcs_uri}. To copy: gsutil cp {gcs_uri} {file_name}"
)
elif not current_error_message:
current_error_message = "API reported success but no video URI was found in the response."
else:
# Success reported, but no video URI found - treat as an error/unexpected state
current_error_message = (
"API operation completed but returned no error or response data."
)
else:
# Handle cases where 'done' is false or response structure is unexpected
current_error_message = "Video generation operation did not complete or returned an unexpected status. API response structure or operation not done."
end_time = time.time()
execution_time = end_time - start_time
state.timing = f"Generation time: {round(execution_time)} seconds"
if current_error_message:
state.error_message = current_error_message
state.show_error_dialog = True
state.result_video = ""
if gcs_uri and not current_error_message:
try:
add_video_metadata(
gcs_uri,
scene_direction_for_video,
aspect_ratio,
veo_model,
execution_time,
state.video_length,
state.reference_image_gcs,
rewrite_prompt,
error_message="",
comment="motion portrait",
)
except Exception as meta_err:
print(f"CRITICAL: Failed to store metadata: {meta_err}")
additional_meta_error = f" (Metadata storage failed: {meta_err})"
state.error_message = (
state.error_message or "Video generated but metadata failed."
) + additional_meta_error
state.show_error_dialog = True
elif not gcs_uri and not current_error_message:
state.error_message = (
state.error_message
or "Video generation completed without error, but no video was produced."
)
state.show_error_dialog = True
except Exception as err:
print(
f"Exception during motion portrait generation: {type(err).__name__}: {err}"
)
state.error_message = f"An unexpected error occurred: {err}"
state.show_error_dialog = True
state.result_video = ""
finally:
state.is_loading = False
yield
print("Motion portrait generation process finished.")
print("Cut! That's a wrap!")