in experiments/veo-app/pages/veo.py [0:0]
def on_click_veo(e: me.ClickEvent): # pylint: disable=unused-argument
"""Veo generate request handler"""
state = me.state(PageState)
state.is_loading = True
state.show_error_dialog = False # Reset error state before starting
state.error_message = ""
state.result_video = "" # Clear previous result
state.timing = "" # Clear previous timing
yield
print(f"Lights, camera, action!:\n{state.veo_prompt_input}")
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
start_time = time.time() # Record the starting time
gcs_uri = ""
current_error_message = ""
try:
if state.reference_image_gcs:
print(f"I2V invoked. I see you have an image! {state.reference_image_gcs} ")
op = image_to_video(
state.veo_prompt_input,
state.reference_image_gcs,
seed,
aspect_ratio,
sample_count,
f"gs://{config.VIDEO_BUCKET}",
rewrite_prompt,
duration_seconds,
)
else:
print("T2V invoked.")
op = text_to_video(
state.veo_prompt_input,
seed,
aspect_ratio,
sample_count,
f"gs://{config.VIDEO_BUCKET}",
rewrite_prompt,
duration_seconds,
)
print(f"Operation result: {op}")
# Check for explicit errors in response
if op.get("done") and op.get("error"):
current_error_message = op["error"].get("message", "Unknown API error")
print(f"API Error Detected: {current_error_message}")
# No GCS URI in this case
gcs_uri = ""
elif op.get("done") and 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}")
gcs_uri = "" # No GCS URI if content was filtered
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:
file_name = gcs_uri.split("/")[-1]
print("Video generated - use the following to copy locally")
print(f"gsutil cp {gcs_uri} {file_name}")
state.result_video = gcs_uri
else:
# Success reported, but no video URI found - treat as an error/unexpected state
current_error_message = "API reported success but no video URI was found in the response."
print(f"Error: {current_error_message}")
state.result_video = "" # Ensure no video is shown
else:
# Handle cases where 'done' is false or response structure is unexpected
current_error_message = (
"Unexpected API response structure or operation not done."
)
print(f"Error: {current_error_message}")
state.result_video = ""
# Catch specific exceptions you anticipate
except ValueError as err:
print(f"ValueError caught: {err}")
current_error_message = f"Input Error: {err}"
except requests.exceptions.HTTPError as err:
print(f"HTTPError caught: {err}")
current_error_message = f"Network/API Error: {err}"
# Catch any other unexpected exceptions
except Exception as err:
print(f"Generic Exception caught: {type(err).__name__}: {err}")
current_error_message = f"An unexpected error occurred: {err}"
finally:
end_time = time.time() # Record the ending time
execution_time = end_time - start_time # Calculate the elapsed time
print(f"Execution time: {execution_time} seconds") # Print the execution time
state.timing = f"Generation time: {round(execution_time)} seconds"
# If an error occurred, update the state to show the dialog
if current_error_message:
state.error_message = current_error_message
state.show_error_dialog = True
# Ensure no result video is displayed on error
state.result_video = ""
try:
add_video_metadata(
gcs_uri,
state.veo_prompt_input,
aspect_ratio,
veo_model,
execution_time,
state.video_length,
state.reference_image_gcs,
rewrite_prompt,
error_message=current_error_message,
comment="veo2 default generation",
)
except Exception as meta_err:
# Handle potential errors during metadata storage itself
print(f"CRITICAL: Failed to store metadata: {meta_err}")
# Optionally, display another error or log this critical failure
if not state.show_error_dialog: # Avoid overwriting primary error
state.error_message = f"Failed to store video metadata: {meta_err}"
state.show_error_dialog = True
state.is_loading = False
yield
print("Cut! That's a wrap!")