Script Fails When Processing Subsequent Video Chunks

[KostiantynBoiar]

Description

I encountered a problem when trying to use the VideoSplitter class to split a video file into smaller chunks based on a specified size. The script fails during the process of creating or saving video chunks, and the output files are not generated as expected. Below is a detailed explanation of the problem.

Expected Behavior

The script should successfully split the input video into smaller chunks, each not exceeding the specified maximum size (max_chunk_size_mb).
All chunks should be saved in the specified output directory (video_chunks).

Actual Behavior

• The script successfully processes the first chunk.
• It fails on the second chunk (and potentially subsequent chunks) with the following error:

Error processing part 2: 'NoneType' object has no attribute 'stdout'

Environment

OS: Win10
Python Version: 3.12.1
Dependencies:
moviepy version: 2.0.0

The code of my class:

import os
from typing import Optional
from dataclasses import dataclass
from moviepy import *
import traceback

@dataclass
class VideoMetadata:
    """Class to store video metadata"""
    duration: float
    fps: float
    width: int
    height: int
    size_mb: float

class VideoSplitter:
    """Class for splitting video files into parts of a specified size"""
    
    def __init__(self, input_file: str, max_chunk_size_mb: float = 2, output_dir: str = "video_chunks"):
        """
        Splitter initialization
        
        Args:
            input_file: path to the input video file
            max_chunk_size_mb: maximum size of each part in MB
            output_dir: directory to save the parts
        """
        self.input_file = input_file
        self.max_chunk_size_mb = max_chunk_size_mb
        self.output_dir = output_dir
        self.video: Optional[VideoFileClip] = None
        self.metadata: Optional[VideoMetadata] = None

    def _validate_input_file(self) -> bool:
        """Check if the input file exists"""
        if not os.path.exists(self.input_file):
            raise FileNotFoundError(f"File {self.input_file} not found!")
        return True

    def _prepare_output_directory(self) -> None:
        """Create the output directory"""
        os.makedirs(self.output_dir, exist_ok=True)

    def _load_video(self) -> None:
        """Load video and collect metadata"""
        self.video = VideoFileClip(self.input_file)
        size_mb = os.path.getsize(self.input_file) / (1024 * 1024)
        
        self.metadata = VideoMetadata(
            duration=self.video.duration,
            fps=self.video.fps,
            width=self.video.w,
            height=self.video.h,
            size_mb=size_mb
        )

    def _calculate_chunk_duration(self) -> float:
        """
        Calculate the duration of each chunk based on the desired size
        and the overall size/duration of the original file
        """
        # Use proportion: if the entire file size_mb lasts duration seconds,
        # a chunk of max_chunk_size_mb will last x seconds
        chunk_duration = (self.max_chunk_size_mb * self.metadata.duration) / self.metadata.size_mb
        return chunk_duration

    def _generate_output_filename(self, chunk_number: int) -> str:
        """Generate the filename for a video chunk"""
        base_name = os.path.splitext(os.path.basename(self.input_file))[0]
        return os.path.join(self.output_dir, 
                          f"{base_name}_part_{chunk_number}.mp4")

    def split(self) -> None:
        """Main method to split the video into parts."""
        try:
            # Validate input file and prepare
            self._validate_input_file()
            self._prepare_output_directory()
            self._load_video()

            # Get metadata and calculate chunk durations
            chunk_duration = self._calculate_chunk_duration()
            num_chunks = int(self.metadata.duration / chunk_duration) + 1

            print(f"The video will be split into {num_chunks} parts, each lasting: {chunk_duration:.2f} seconds.")
            
            # Iterate through chunks
            for i in range(num_chunks):
                start_time = i * chunk_duration
                end_time = min((i + 1) * chunk_duration, self.metadata.duration)

                print(f"Processing part {i+1}/{num_chunks} from {start_time:.2f} to {end_time:.2f} seconds...")
                output_file = self._generate_output_filename(i + 1)

                try:
                    # Create a subclip
                    subclip = self.video.with_subclip(start_time, end_time)
                    if not subclip:
                        raise ValueError(f"Failed to create subclip {i+1}.")
                    
                    # Write the subclip
                    subclip.write_videofile(
                        output_file,
                        codec='libx264',
                        audio_codec='aac',
                        temp_audiofile=f"temp_audio_{i}.mp4",
                        remove_temp=True,
                        threads=4
                    )
                except Exception as e:
                    print(f"Error processing part {i+1}: {e}")
                    print(traceback.format_exc())
                    raise
                finally:
                    # Safely close the subclip
                    if 'subclip' in locals() and subclip:
                        subclip.close()

                # Check the size of the created file
                try:
                    actual_size = os.path.getsize(output_file) / (1024 * 1024)
                    print(f"Part {i+1} saved: {output_file} (size: {actual_size:.2f} MB).")
                except Exception as e:
                    print(f"Error checking the size of part {i+1}: {e}")
                    raise

        except Exception as e:
            print(f"General error while splitting video: {e}")
            print(traceback.format_exc())
            raise
        finally:
            # Safely close the video file
            if self.video:
                try:
                    self.video.close()
                except Exception as e:
                    print(f"Error closing video: {e}")


    def get_metadata(self) -> Optional[VideoMetadata]:
        """Retrieve video metadata"""
        return self.metadata

def main():
    # Example usage
    splitter = VideoSplitter(
        input_file='video.mp4',
        max_chunk_size_mb=24,
        output_dir='video_chunks'
    )
    
    try:
        splitter.split()
        
        # Output metadata
        metadata = splitter.get_metadata()
        if metadata:
            print("\nVideo Information:")
            print(f"Duration: {metadata.duration:.2f} seconds")
            print(f"Size: {metadata.size_mb:.2f} MB")
            print(f"Resolution: {metadata.width}x{metadata.height}")
            print(f"FPS: {metadata.fps}")
            
    except Exception as e:
        print(f"An error occurred: {str(e)}")

if __name__ == "__main__":
    main()

The function where I get an exception:

def split(self) -> None:
        """Main method to split the video into parts."""
        try:
            # Validate input file and prepare
            self._validate_input_file()
            self._prepare_output_directory()
            self._load_video()

            # Get metadata and calculate chunk durations
            chunk_duration = self._calculate_chunk_duration()
            num_chunks = int(self.metadata.duration / chunk_duration) + 1

            print(f"The video will be split into {num_chunks} parts, each lasting: {chunk_duration:.2f} seconds.")
            
            # Iterate through chunks
            for i in range(num_chunks):
                start_time = i * chunk_duration
                end_time = min((i + 1) * chunk_duration, self.metadata.duration)

                print(f"Processing part {i+1}/{num_chunks} from {start_time:.2f} to {end_time:.2f} seconds...")
                output_file = self._generate_output_filename(i + 1)

                try:
                    # Create a subclip
                    subclip = self.video.with_subclip(start_time, end_time)
                    if not subclip:
                        raise ValueError(f"Failed to create subclip {i+1}.")
                    
                    # Write the subclip
                    subclip.write_videofile(
                        output_file,
                        codec='libx264',
                        audio_codec='aac',
                        temp_audiofile=f"temp_audio_{i}.mp4",
                        remove_temp=True,
                        threads=4
                    )
                except Exception as e:
                    print(f"Error processing part {i+1}: {e}")
                    print(traceback.format_exc())
                    raise
                finally:
                    # Safely close the subclip
                    if 'subclip' in locals() and subclip:
                        subclip.close()

                # Check the size of the created file
                try:
                    actual_size = os.path.getsize(output_file) / (1024 * 1024)
                    print(f"Part {i+1} saved: {output_file} (size: {actual_size:.2f} MB).")
                except Exception as e:
                    print(f"Error checking the size of part {i+1}: {e}")
                    raise

        except Exception as e:
            print(f"General error while splitting video: {e}")
            print(traceback.format_exc())
            raise
        finally:
            # Safely close the video file
            if self.video:
                try:
                    self.video.close()
                except Exception as e:
                    print(f"Error closing video: {e}")