from typing import Optional
# Import the OpenCV library for video processing
# Import the time library for timing
# Import the file operation library for directory deletion
# Import the enum library for defining frame extraction strategies
# Import the Base64 encoding library for converting images to Base64 strings
# Import the operating system library for file path operations
# Install SDK:pip install 'volcengine-python-sdk[ark]'
from volcenginesdkarkruntime import Ark
from volcenginesdkarkruntime._exceptions import ArkAPIError
# Define the frame extraction strategy enumeration class
# Fixed interval frame extraction strategy, e.g., extract one frame every 1 second
CONSTANT_INTERVAL = "constant_interval"
# Even interval frame extraction strategy, extract frames evenly from the entire video based on the set maximum number of frames
EVEN_INTERVAL = "even_interval"
extraction_strategy: Optional[Strategy] = Strategy.EVEN_INTERVAL,
interval_in_seconds: Optional[float] = 1,
max_frames: Optional[int] = 10,
keyframe_naming_template: str = "frame_{:04d}.jpg",
"""Extract frames from video according to the specified strategy
video_file_path (str): Path to the video file
output_dir (str): Output directory
extraction_strategy (Optional[Strategy], optional): Frame extraction strategy.
Constant interval (e.g., extract one frame every 1 second) or
Even interval (extract frames evenly from the entire video based on the set maximum number of frames)
Defaults to constant interval (1 frame per second).
interval_in_seconds (Optional[float], optional): Time interval for constant interval frame extraction. Defaults to 1 second per frame.
max_frames (Optional[int], optional): Maximum number of frames to extract. Defaults to 10 frames.
keyframe_naming_template (_type_, optional): Naming template for extracted key frame images
list[str]: List of paths to the extracted key frame images
# Check if the output directory exists; create it if it does not
if not os.path.exists(output_dir):
# Open the video file using OpenCV
cap = cv2.VideoCapture(video_file_path)
# Get the frame rate of the video
fps = cap.get(cv2.CAP_PROP_FPS)
# Get the total number of frames in the video
length = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
# Select the frame interval based on the strategy
if extraction_strategy == Strategy.CONSTANT_INTERVAL:
# Calculate the frame interval for constant interval extraction
frame_interval = int(fps * interval_in_seconds)
elif extraction_strategy == Strategy.EVEN_INTERVAL:
# Calculate the frame interval for even interval extraction
frame_interval = int(length / max_frames)
# Throw an exception if the strategy is invalid
raise ValueError("Invalid extraction strategy")
# Initialize the frame counter
# Initialize the list of key frames
# Loop to read video frames
# Exit the loop if frame reading fails
# Check if the current frame is a key frame
if frame_count % frame_interval == 0:
# Generate the filename for the key frame
image_path = os.path.join(
output_dir, keyframe_naming_template.format(len(keyframes))
# Save the key frame as an image
# Add the key frame path to the list
keyframes.append(image_path)
# Increment the frame counter
# Exit the loop if the maximum number of key frames is reached
if len(keyframes) >= max_frames:
print("[Video Frame Extraction Completed]")
print("[Number of Extracted Frames]", len(keyframes))
# Return the list of key frame paths
Resize the image to fit the specified dimensions.
image (numpy.ndarray): Input image in numpy array format.
numpy.ndarray: Resized image.
# Get the original height and width of the image
height, width = image.shape[:2]
# Determine the target dimensions based on the image's aspect ratio
target_height, target_width = 480, 640
target_height, target_width = 640, 480
# Return the original image if its dimensions are already less than or equal to the target dimensions
if height <= target_height and width <= target_width:
# Calculate the new height and width while maintaining the image's aspect ratio
if height / target_height < width / target_width:
new_width = target_width
new_height = int(height * (new_width / width))
new_height = target_height
new_width = int(width * (new_height / height))
return cv2.resize(image, (new_width, new_height))
# Define a method to resize the image at the specified path to an appropriate size and convert it to Base64 encoding
def encode_image(image_path: str) -> str:
Encode the image at the specified path
image_path (str): Path to the image file
str: Encoded image string
image = cv2.imread(image_path)
image_resized = resize(image)
# Encode the image in JPEG format
_, encoded_image = cv2.imencode(".jpg", image_resized)
# Convert the encoded image to a Base64 string
return base64.b64encode(encoded_image).decode("utf-8")
def construct_messages(image_paths: list[str], prompt: str) -> list[dict]:
Construct a list of messages containing text and images.
image_paths (list[str]): List of image file paths.
prompt (str): Text prompt.
list[dict]: List of messages containing text and images.
print("[Assembling Request Parameters]")
# Initialize the content list with a text message
# Iterate through the list of image paths
for image_path in image_paths:
# Construct an image URL message for each image path
# Convert the image to a data URL using Base64 encoding
"url": f"data:image/jpeg;base64,{encode_image(image_path)}",
# Specify the image detail level as low
# Return the list of messages containing text and images
if __name__ == "__main__":
# Replace with the path to your video file
video_path = "<VIDEO_FILE_PATH>"
# API Key retrieved from environment variables; configure your API Key in advance
ark_api_key = os.environ.get("ARK_API_KEY")
# Replace <MODEL> with the Model ID of your model
# Replace with your text prompt; this is for example use only
prompt = "Describe the video content"
# Delete historical frames
if os.path.exists("videoFrames"):
shutil.rmtree("videoFrames")
selected_images = preprocess_video(
video_file_path=video_path,
output_dir="videoFrames",
# Two frame extraction strategies
# 1. CONSTANT_INTERVAL: Extract frames at a fixed time interval; adjust the interval as needed
# 2. EVEN_INTERVAL: Extract a fixed total number of frames evenly; adjust the total number as needed
extraction_strategy=Strategy.CONSTANT_INTERVAL,
# extraction_strategy=Strategy.EVEN_INTERVAL,
# Frame interval for constant interval extraction strategy
# Maximum number of frames to extract from the video
print("[Frame Image Compression Completed]")
preprocess_time_cost = time.time() - since
client = Ark(api_key=ark_api_key)
response = client.chat.completions.create(
messages=construct_messages(
print("[Video Understanding Result]\n", response.choices[0].message.content)
print("\n[Total Tokens Consumed]", response.usage.total_tokens)
print("- Input Tokens:", response.usage.prompt_tokens)
print("- Output Tokens:", response.usage.completion_tokens)
except ArkAPIError as e:
api_time_cost = time.time() - since
print("\n[End-to-End Total Time Consumption]", round(preprocess_time_cost + api_time_cost, 2))
print("- Video Preprocessing Time:", round(preprocess_time_cost, 2))
print("- Result Generation Time:", round(api_time_cost, 2))