如何将经OpenCV后处理的帧写回Amazon Kinesis Video Stream？

Got it, let's tackle how to write your processed OpenCV frames back to Kinesis Video Stream (KVS). Replacing imshow() isn't just a one-for-one swap—KVS deals with streaming media that needs proper encoding and chunking, so here's a practical breakdown:

Key Background First #

Kinesis Video Streams doesn't accept raw individual frames directly. It expects streaming media chunks (usually in MKV format) encoded with a supported codec like H.264. So your workflow needs to:

1. Take your processed OpenCV frames (with face detection rectangles)
2. Encode them into a valid video stream
3. Upload the encoded chunks to KVS using its PutMedia API

Step-by-Step Implementation #

1. Install Required Libraries #

You'll need a few extra tools for encoding and KVS interactions:

pip install opencv-python boto3 av

• av: Handles video encoding more reliably for streaming than OpenCV's built-in VideoWriter
• boto3: AWS SDK for Python (you already have this for reading KVS)

2. Initialize KVS Clients #

First, set up the KVS clients to get the upload endpoint and prepare the PutMedia connection:

import cv2
import av
import boto3
from botocore.config import Config
import time

# Configure your AWS region and stream name
REGION = "us-east-1"
STREAM_NAME = "your-target-kvs-stream-name"

# Get KVS data endpoint for PUT_MEDIA API
kvs_client = boto3.client(
    "kinesisvideo",
    region_name=REGION,
    config=Config(signature_version="v4")
)
endpoint_response = kvs_client.get_data_endpoint(
    StreamName=STREAM_NAME,
    APIName="PUT_MEDIA"
)
put_media_endpoint = endpoint_response["DataEndpoint"]

# Initialize the media client for uploading
kvs_media_client = boto3.client(
    "kinesis-video-media",
    endpoint_url=put_media_endpoint,
    region_name=REGION,
    config=Config(signature_version="v4")
)

3. Encode and Upload Processed Frames #

Replace your imshow() call with this encoding/upload logic. This example assumes you already have a loop reading frames from KVS and applying face detection/rectangle drawing:

# Match your input stream's resolution and frame rate (critical for KVS compatibility)
FRAME_WIDTH = 1920
FRAME_HEIGHT = 1080
FPS = 30

# Set up H.264 encoding to MKV (KVS's preferred format)
container = av.open(":", mode="w", format="matroska")
video_stream = container.add_stream("h264", rate=FPS)
video_stream.codec_context.width = FRAME_WIDTH
video_stream.codec_context.height = FRAME_HEIGHT
video_stream.codec_context.pix_fmt = "yuv420p"  # Standard for H.264

# Start the PutMedia connection to KVS
start_timestamp = int(time.time() * 1000)
put_media_response = kvs_media_client.put_media(
    StreamName=STREAM_NAME,
    ProducerStartTimestamp=start_timestamp,
    Payload=b""  # Initialize the connection with empty data
)
payload_stream = put_media_response["Payload"]

try:
    # Replace this loop with your existing KVS frame-reading/processing logic
    while True:
        # Example: Read a frame from your input KVS stream (replace with your code)
        ret, raw_frame = your_kvs_frame_reader.read()
        if not ret:
            break

        # --- Your existing processing here ---
        # e.g., run AWS Rekognition face detection, draw rectangles with OpenCV
        processed_frame = raw_frame  # Replace with your actual processed frame

        # Convert OpenCV's BGR frame to RGB (required for av encoding)
        rgb_frame = cv2.cvtColor(processed_frame, cv2.COLOR_BGR2RGB)
        av_frame = av.VideoFrame.from_ndarray(rgb_frame, format="rgb24")

        # Encode the frame and send chunks to KVS
        for packet in video_stream.encode(av_frame):
            payload_stream.write(packet.to_bytes())

        # Throttle to match your frame rate
        time.sleep(1 / FPS)

finally:
    # Send any remaining encoded data
    for packet in video_stream.encode():
        payload_stream.write(packet.to_bytes())
    # Clean up connections
    payload_stream.close()
    container.close()

Important Notes #

• IAM Permissions: Make sure your execution role has kinesisvideo:PutMedia and kinesisvideo:GetDataEndpoint permissions attached.
• Stream Consistency: Keep the resolution, frame rate, and codec consistent with your input stream—mismatches can cause KVS to reject the data or produce unplayable streams.
• Error Handling: In production, add retry logic for network failures, and handle PutMedia connection timeouts. The example above is simplified for clarity.
• Alternative SDK: If you want to avoid manual encoding, check out the Kinesis Video Streams Producer SDK for Python (community-maintained). It handles encoding, chunking, and retries out of the box.

内容的提问来源于stack exchange，提问作者Dhrumil Sheth