Skip to content

Misc

Miscellaneous functions.

process_video(source_path, target_path, callback, target_fps=1, codec='mp4v')

Process a video frame-by-frame, applying a callback function to each frame and saving the results to a new video. This version includes a progress bar and allows codec selection.

Parameters:

Name Type Description Default
source_path str

Path to the source video file.

 required
target_path str

Path to save the processed video.

 required
callback Callable[[ndarray, int], ndarray]

A function that takes a video frame and its index as input and returns the processed frame.

 required
codec str

Codec used to encode the processed video. Default is "avc1".

 'mp4v'
Source code in PytorchWildlife/utils/misc.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def process_video(
    source_path: str,
    target_path: str,
    callback: Callable[[np.ndarray, int], np.ndarray],
    target_fps: int = 1,
    codec: str = "mp4v"
) -> None:
    """
    Process a video frame-by-frame, applying a callback function to each frame and saving the results 
    to a new video. This version includes a progress bar and allows codec selection.

    Args:
        source_path (str): 
            Path to the source video file.
        target_path (str): 
            Path to save the processed video.
        callback (Callable[[np.ndarray, int], np.ndarray]): 
            A function that takes a video frame and its index as input and returns the processed frame.
        codec (str, optional): 
            Codec used to encode the processed video. Default is "avc1".
    """
    source_video_info = VideoInfo.from_video_path(video_path=source_path)

    if source_video_info.fps > target_fps:
        stride = int(source_video_info.fps / target_fps)
        source_video_info.fps = target_fps
    else:
        stride = 1

    with VideoSink(target_path=target_path, video_info=source_video_info, codec=codec) as sink:
        with tqdm(total=int(source_video_info.total_frames / stride)) as pbar: 
            for index, frame in enumerate(
                get_video_frames_generator(source_path=source_path, stride=stride)
            ):
                result_frame = callback(frame, index)
                sink.write_frame(frame=cv2.cvtColor(result_frame, cv2.COLOR_RGB2BGR))
                pbar.update(1)