如何实现让Python代码块或直播流数据捕获函数持续运行30分钟？

Hey there! Let's break down solutions for your two Python needs clearly and practically:

We’ll cover two common scenarios here:

• Scenario A: Repeat a task until 30 minutes pass
  If you need your code to keep executing a specific action (like processing data or logging) for the full duration, use the time module to track elapsed time and loop until the 30-minute mark:

import time

# Convert 30 minutes to seconds for easier calculation
run_duration = 30 * 60
start_time = time.time()

while time.time() - start_time < run_duration:
    # Replace this line with your actual code block
    print("Running task...")
    # Optional: Add a small sleep to avoid overwhelming CPU
    time.sleep(1)

• Scenario B: Keep the program idle for 30 minutes
  If you just need the program to stay active without executing tasks (e.g., waiting for an external event), use time.sleep() directly:

import time

# Pause execution for 30 minutes
time.sleep(30 * 60)
print("30 minutes done—resuming program!")

Live stream capture is usually a blocking process, so we need to control its runtime without freezing the program. Here are two reliable methods:

Method 1: Add Time Checks to the Capture Loop #

If you’ve written your own capture loop, integrate a time check to stop after 30 minutes:

import time
import cv2  # Example using OpenCV for stream capture

def capture_live_stream(stream_url):
    run_duration = 30 * 60
    start_time = time.time()
    
    # Initialize stream connection
    stream = cv2.VideoCapture(stream_url)
    
    while time.time() - start_time < run_duration:
        ret, frame = stream.read()
        if not ret:
            print("Stream interrupted—stopping capture early")
            break
        
        # Add your frame processing logic here (e.g., save to disk)
        cv2.imwrite("latest_frame.jpg", frame)
        
        # Optional: Match stream frame rate to avoid excess processing
        time.sleep(0.03)  ~30 FPS
    
    # Clean up resources
    stream.release()
    print("Capture stopped after 30 minutes")

# Call the function with your stream URL
capture_live_stream("rtmp://your-stream-source-url")

Method 2: Use Threading + Stop Flag (For Blocking Library Functions) #

If your capture function is from a third-party library and can’t be modified, run it in a thread and use a flag to signal it to stop after 30 minutes:

import time
import threading
import cv2

stop_capture = False

def capture_live_stream(stream_url):
    global stop_capture
    stream = cv2.VideoCapture(stream_url)
    
    while not stop_capture:
        ret, frame = stream.read()
        if not ret:
            break
        # Process frame as needed
        cv2.imwrite("latest_frame.jpg", frame)
    
    stream.release()
    print("Capture terminated")

# Start capture in a background thread
stream_thread = threading.Thread(target=capture_live_stream, args=("rtmp://your-stream-source-url",))
stream_thread.start()

# Wait 30 minutes, then trigger stop
time.sleep(30 * 60)
stop_capture = True
stream_thread.join()

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