def process_video()

in data-analytics/minigolf-demo/cloud_function/video_processing.py [0:0]

• 41 lines of code
• 10 McCabe index (conditional complexity)

def process_video(video_file, user_id):
    """
    Processes a video to track the golf ball, detect shots, and upload data to BigQuery.

    Args:
        video_file: The path to the video file.
        user_id: The ID of the user who uploaded the video.
    """
    cap = cv2.VideoCapture(video_file)  # Open the video
    tracker = cv2.legacy.TrackerCSRT_create()  # Create a CSRT tracker
    ret, frame = cap.read()  # Read the first frame
    if ret:
        # Generate a unique filename for the image
        image_filename = f"BG_{user_id}.jpg"

        # Save the first frame as a JPEG image
        cv2.imwrite(f"/tmp/{image_filename}", frame)

    tracker.init(frame, BALL)  # Initialize the tracker with the ball's initial position

    frame_number = 0
    num_shots = 0
    dist_history = deque(maxlen=30)  # Store recent distances for movement detection
    status_history = deque(maxlen=30)  # Store recent movement statuses (True/False)

    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break  # End of video

        success, bbox = tracker.update(frame)

        if success:
            frame_number += 1
            x, y, w, h = bbox
            center_x, center_y = x + w // 2, y + h // 2
            time_now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

            distance = calculate_distance(center_x, center_y)
            is_moving = check_if_moving(dist_history, distance)
            dist_history.append(distance)

            # Detect a new shot when movement starts after a stationary period
            if not any(status_history) and is_moving:
                num_shots += 1

            status_history.append(is_moving)

            # Prepare tracking data for BigQuery
            tracking_data = {
                "created_time": time_now,
                "user_id": user_id,
                "frame_number": frame_number,
                "x": int(center_x),
                "y": int(center_y),
                "distance": f"{distance:.2f}",
                "is_moving": is_moving,
                "shot_number": num_shots,
            }

            insert_data("tracking", tracking_data)

    cap.release()  # Release video capture resources
    cv2.destroyAllWindows()