双独立接收器IP网络下视频流与处理结果同步方案咨询

Great question—your timestamp-based approach is totally on the right track, and it’s actually the go-to solution for this kind of multi-receiver sync scenario. Let’s walk through a practical, reliable implementation that keeps your receivers independent (no redundant video streaming) and ensures tight sync between video frames and processed results.

The key here is leveraging the Presentation Time Stamp (PTS) embedded in every video frame of your source stream. Both receivers pull the same stream independently, so they’ll each have access to the same PTS values for every frame. This gives you a shared, frame-level time reference to align processed results with their corresponding video frames.

1. Extract & Normalize Timestamps on Both Receivers #

• Every video stream (RTSP, RTMP, HLS, etc.) includes PTS values that define exactly when each frame should be displayed. Use a media processing library like FFmpeg or GStreamer to extract these timestamps directly from the stream.
  • For FFmpeg: Use av_frame_get_best_effort_timestamp() to get the PTS, then convert it to a standardized format (e.g., milliseconds since the stream started, or Unix epoch time if the stream includes absolute timing).
  • For GStreamer: Use gst_buffer_get_pts() and convert it to your preferred time base.
• Ensure both receivers use the same time base for PTS values (e.g., all in milliseconds) to avoid alignment errors.

2. Transmit Processed Results with Timestamps #

• On your processing receiver (let’s call it Receiver A), after running your frame processing logic (object detection, analytics, etc.), package the result along with the corresponding frame’s PTS into a lightweight payload. A JSON object works perfectly here:

  {"frame_pts": 1698765432123, "processed_result": "your_output_value"}
  

• Use WebSocket for transmission (preferred for real-time, low-latency scenarios) or REST POST requests. WebSockets maintain a persistent connection, which eliminates the overhead of repeated HTTP handshakes and reduces latency—critical for syncing.

3. Sync Frames & Results on the Display Receiver #

• On your display receiver (Receiver B), maintain a small result cache queue (e.g., hold the last 50 results) to account for network jitter or minor decoding delays.
• When Receiver B decodes a frame, extract its PTS and check the cache for a matching result:
  • If a result with a PTS within a small tolerance (e.g., ±10ms) is found, display the frame and result together immediately.
  • If no match is found, briefly delay rendering the frame (up to 100ms—adjust based on your latency requirements) to wait for the result. If the result still doesn’t arrive, render the frame with a placeholder (e.g., "Result pending") to keep video playback smooth.
• Periodically clean up the cache to remove results for frames that have already been rendered, preventing memory bloat.

• Handle Network Jitter: A cache queue with enough capacity (50-100 frames) will absorb temporary spikes in network latency without breaking sync.
• Account for Clock Drift: Over long runs, minor clock differences between receivers can cause drift. To mitigate, periodically have both receivers re-calibrate their time base using the stream’s SPS/PPS headers (for H.264/H.265) or sync to a shared NTP server (optional, but useful for very long sessions).
• Validate PTS Matches: Allow a small time window for PTS matching (±10-20ms) to account for tiny differences in decoding speed between the two receivers.
• Prioritize Video Smoothness: If sync delays become too large, always prioritize uninterrupted video playback over waiting for stale results—users will notice skipped results less than stuttering video.

Receiver A (Processing & Transmission) #

import ffmpeg
import websockets
import asyncio
import json

def process_frame(frame):
    # Replace with your actual processing logic (e.g., object detection)
    return "Detected 3 objects"

async def stream_processed_results():
    # Initialize FFmpeg to read the source stream
    stream = ffmpeg.input("rtsp://your-source-stream-url")
    frame_gen = stream.output('pipe:', format='rawvideo', pix_fmt='bgr24').run_async(pipe_stdout=True)
    
    # Connect to Receiver B's WebSocket endpoint
    async with websockets.connect("ws://receiver-b-ip:8000/sync") as ws:
        while True:
            frame = frame_gen.stdout.read(1920*1080*3)  # Adjust for your resolution
            if not frame:
                break
            
            # Extract the frame's PTS (simplified example—use FFmpeg's API for accuracy)
            pts = stream.get_frame_pts()
            processed_result = process_frame(frame)
            
            # Send payload with PTS and result
            await ws.send(json.dumps({
                "frame_pts": pts,
                "processed_result": processed_result
            }))

asyncio.run(stream_processed_results())

Receiver B (Display & Sync) #

import ffmpeg
import websockets
import asyncio
import json
from collections import deque

result_cache = deque(maxlen=50)  # Cache last 50 results

async def receive_results(websocket):
    async for message in websocket:
        data = json.loads(message)
        result_cache.append( (data["frame_pts"], data["processed_result"]) )

async def start_websocket_server():
    async with websockets.serve(receive_results, "0.0.0.0", 8000):
        await asyncio.Future()  # Keep server running

async def play_and_sync():
    stream = ffmpeg.input("rtsp://your-source-stream-url")
    frame_gen = stream.output('pipe:', format='rawvideo', pix_fmt='bgr24').run_async(pipe_stdout=True)
    
    while True:
        frame = frame_gen.stdout.read(1920*1080*3)
        if not frame:
            break
        
        current_pts = stream.get_frame_pts()
        matching_result = None
        
        # Search cache for matching PTS (with tolerance)
        for pts, res in list(result_cache):
            if abs(pts - current_pts) < 15:  # 15ms tolerance
                matching_result = res
                result_cache.remove( (pts, res) )
                break
        
        # Render frame and result
        render_video_frame(frame)
        if matching_result:
            display_processed_result(matching_result)
        else:
            display_placeholder("Waiting for result...")

async def main():
    await asyncio.gather(start_websocket_server(), play_and_sync())

asyncio.run(main())

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