Merge and encode dashcam videos stored on an SD card

dashcam-encode.py

#!/usr/bin/env python3

"""
Concatenate and encode dashcam videos stored on an SD card.

Assumes the videos are stored as *xxxx.avi / *xxxx.mp4 / *xxxx.mov, where xxxx is a
sequential index. This should be compatible with most dashcam SoC manufacturers.

System requirements and dependencies:
-------------------------------------

* Cross-platform (tested on Windows, Linux, and macOS).
* Requires ffmpeg [https://ffmpeg.org/] to be installed and available in PATH.
* No other packages are required. Only built-in Python modules are used.

Usage guide:
------------

The tool can operate in two modes. Run "dashcam-encode.py -h" for usage details.

1) "trips" mode: encodes all videos in the input folder, organizing them into trips
   according to file dates. Each logical trip will be encoded as a single output video.

   Run "dashcam-encode.py trips -h" for usage details.

   Example (default trip gap = 3 hours):

   Filename                               Date           Inferred Trip   Output File
   /DCIM
   |---/Video
   |-------/20230112152345_000001.MP4     Jan 12, 15:23  | => Trip 1  |  Trip 1.mp4
   |-------/20230112152445_000002.MP4     Jan 12, 15:24  | => Trip 1  |
   |-------/20230112193820_000003.MP4     Jan 12, 19:38  | => Trip 2  |  Trip 2.mp4
   |-------/20230113101200_000004.MP4     Jan 13, 10:12  | => Trip 3  |  Trip 3.mp4
   |-------/20230113101300_000005.MP4     Jan 13, 10:13  | >= Trip 3  |

2) "range" mode: merges all raw videos between the specified start/end index and
   encodes them as a single output video.

   Run "dashcam-encode.py range -h" for usage details.

   Example ("range 3 6"):

   /DCIM
   |---/Video
   |-------/20230112152345_000001.MP4
   |-------/20230112152445_000003.MP4     <= These videos
   |-------/20230112193820_000004.MP4     <= will be
   |-------/20230113101200_000006.MP4     <= merged
   |-------/20230113101300_000007.MP4


Usage examples:
---------------

Encode all raw video files in the default location, group them into trips,
and save each trip as a separate output video file:
> dashcam-encode.py trips

Encode all raw video files in the specified location, group them into trips,
and save each trip as a separate output video file:
> dashcam-encode.py trips --raw-video-folder "E:\\DCIM\\DCIMA"

Encode all raw video files in the specified location, group them into trips
where trips should be at least 8 hours apart, encode 3 trips in parallel:
> dashcam-encode.py trips --min-trip-gap-hours 8 --job-count 3

Encode raw video files labeled from #15 to #319 (in the default location)
and save them as a single output video file:
> dashcam-encode.py range 15 319

Encode raw video files labeled from #15 to #319 (in the specified custom location)
and save them as a single output video file named "Road Trip.mp4"
> dashcam-encode.py range 15 319 --output-name "Road Trip" --raw-video-folder "E:\\Video"
"""

import argparse
import dataclasses
import datetime
import logging
import multiprocessing
import os
import sys
import subprocess
import tempfile
import typing as t


# Default settings (can be overridden in the command line).
DEFAULT_VIDEO_FOLDER_PATH = "F:\\DCIM\\Movie"
DEFAULT_MIN_TRIP_GAP_HOURS = 3
DEFAULT_JOB_COUNT = 2

# Extensions of the video files to expect in the input folder.
EXPECTED_VIDEO_EXTENSIONS = (".avi", ".mp4", ".mov")

# Codec/quality settings for the output videos.
OUTPUT_VIDEO_CODEC_PARAMETERS = "-c:v libx265 -crf 30 -preset fast"
OUTPUT_AUDIO_CODEC_PARAMETERS = "-c:a aac -b:a 128k"
OUTPUT_FORMAT = "mp4"


# pylint: disable=logging-fstring-interpolation
logging.basicConfig(
    level=logging.INFO, format="%(asctime)s | %(levelname)8s | %(message)s"
)
LOGGER = logging.getLogger(__name__)


@dataclasses.dataclass
class RawVideoSegment:
    """
    Represents a single raw video file (e.g. a 1-minute segment) in the input folder.
    """

    index: int
    filename: str


@dataclasses.dataclass
class Trip:
    """
    Video segments, logically grouped into a single trip
    that should be encoded as a single file.
    """

    id: str
    date: datetime.datetime
    raw_segments: t.List[RawVideoSegment]

    def get_full_trip_name(self) -> str:
        trip_date_formatted = self.date.strftime("%Y-%m-%d")
        return f"{trip_date_formatted} {self.id}"

    def __str__(self):
        trip_date_formatted = self.date.strftime("%Y-%m-%d")
        start_index = self.raw_segments[0].index
        end_index = self.raw_segments[-1].index
        return f"{self.id} / Date: {trip_date_formatted} / Segments: {start_index}-{end_index}"


@dataclasses.dataclass
class TripEncodeJobDefinition:
    """
    Parameters for a single encoding job in the parallel encoding pool.
    """

    trip: Trip
    raw_video_folder: str


def parse_command_line_args(args: t.List[str]) -> argparse.Namespace:
    """
    Parse the arguments passed via the command line.
    """
    parser = argparse.ArgumentParser(
        description="Concatenate and encode dashcam videos stored on an SD card."
    )
    subparsers = parser.add_subparsers(dest="command", metavar="COMMAND")

    parser_trips_cmd = subparsers.add_parser(
        name="trips",
        help=(
            "Encode all videos in the input folder, organizing them into trips "
            "according to file dates."
        ),
    )
    parser_trips_cmd.add_argument(
        "--min-trip-gap-hours",
        metavar="TG",
        help=(
            "The minimum date difference (in hours) between consecutive files "
            "for them to be considered separate trips."
        ),
        type=int,
        required=False,
        default=DEFAULT_MIN_TRIP_GAP_HOURS,
    )
    parser_trips_cmd.add_argument(
        "--job-count",
        metavar="JC",
        help=(
            "The maximum number of trips allowed to be encoded in parallel. "
            "Adjust this number to change system resource utilization."
        ),
        type=int,
        required=False,
        default=DEFAULT_JOB_COUNT,
    )

    parser_range_cmd = subparsers.add_parser(
        name="range",
        help=(
            "Concatenate input videos in the specified range "
            "and encode them into a single output video"
        ),
    )
    parser_range_cmd.add_argument(
        "start_index",
        metavar="START-INDEX",
        help=(
            "Index of the first video (inclusive). "
            "Example: for 20230112152345_000007.MP4, specify 7."
        ),
        type=int,
    )
    parser_range_cmd.add_argument(
        "end_index",
        metavar="END-INDEX",
        help=(
            "Index of the last video (inclusive). "
            "Example: for 20230112170446_000094.MP4, specify 94."
        ),
        type=int,
    )
    parser_range_cmd.add_argument(
        "--output-name",
        metavar="NAME",
        help=(
            "Name of the output video (without extension). "
            "If omitted, a name will be generated automatically."
        ),
        type=str,
        required=False,
    )

    for subparser in [parser_trips_cmd, parser_range_cmd]:
        subparser.add_argument(
            "--raw-video-folder",
            metavar="PATH",
            help=(
                f"Path to the raw video folder on the SD card, such as "
                f"'{DEFAULT_VIDEO_FOLDER_PATH}'."
            ),
            type=str,
            required=False,
            default=DEFAULT_VIDEO_FOLDER_PATH,
        )

    parsed_args = parser.parse_args(args)
    return parsed_args


def video_index_from_filename(filename: str) -> int:
    """
    Extract the numeric index of the video from its filename.
    E.g., "F:\\DCIM\\Movie\\20230112170046_000090.MP4 -> 90"
    """
    basename_without_ext, _ = os.path.splitext(os.path.basename(filename))
    index = int(basename_without_ext[-4:])
    return index


def collect_raw_video_segments(
    raw_folder_path,
    start_index=None,
    end_index=None,
) -> t.List[RawVideoSegment]:
    """
    Scan the raw video folder for files matching the input criteria.
    """
    LOGGER.info(f"Collecting files from '{raw_folder_path}'")

    filenames = [
        filename
        for filename in os.listdir(raw_folder_path)
        if os.path.splitext(filename)[-1].lower() in EXPECTED_VIDEO_EXTENSIONS
    ]
    LOGGER.info(f"Found {len(filenames)} files with supported extensions")

    filenames = [os.path.join(raw_folder_path, filename) for filename in filenames]
    segments = [
        RawVideoSegment(index=video_index_from_filename(filename), filename=filename)
        for filename in sorted(filenames, key=video_index_from_filename)
    ]
    if start_index is not None and end_index is not None:
        segments = [
            segment for segment in segments if start_index <= segment.index <= end_index
        ]

    if not segments:
        raise RuntimeError("Could not find any files matching the input criteria")

    LOGGER.info(f"Collected {len(segments)} files matching the input criteria")
    return segments


def group_segments_into_trips(segments, min_trip_gap_hours) -> t.List[Trip]:
    """
    Given a list of video segments, bucket them into trips according to file date difference.
    """
    trips = []
    file_dates = [
        datetime.datetime.fromtimestamp(os.path.getmtime(segment.filename))
        for segment in segments
    ]

    # Scan the files sequentially and create a new trip
    # each time the date difference is big enough.
    prev_file_date = None
    for segment, file_date in zip(segments, file_dates):
        is_new_trip = (
            prev_file_date is None
            or (file_date - prev_file_date).total_seconds() / 3600.0
            > min_trip_gap_hours
        )
        if is_new_trip:
            new_trip_id = f"Trip {len(trips) + 1}"
            trips.append(Trip(id=new_trip_id, date=file_date, raw_segments=[]))

        trips[-1].raw_segments.append(segment)
        prev_file_date = file_date

    return trips


def generate_ffmpeg_input_file(segments) -> str:
    """
    Build an input file list for ffmpeg containing the files to be concatenated.

    Returns
    -------
    str
        The path of the generated ffmpeg input file.
    """
    with tempfile.NamedTemporaryFile(
        mode="w",
        encoding="utf-8",
        prefix="dashcam-encode-",
        suffix=".txt",
        delete=False,
    ) as fp:
        LOGGER.info(f"Writing ffmpeg file list to '{fp.name}'")
        fp.writelines(f"file '{segment.filename}'\n" for segment in segments)
        return fp.name


def run_ffmpeg(ffmpeg_input_filename, output_video_name):
    """
    Run video concatenation and encoding.
    """
    output_filename = f"{output_video_name}.{OUTPUT_FORMAT}"
    cmd = (
        # Allow ffmpeg to use absolute input paths (-safe 0)
        f'ffmpeg -f concat -safe 0 -i "{ffmpeg_input_filename}" '
        # Video and audio codec parameters
        f"{OUTPUT_VIDEO_CODEC_PARAMETERS} {OUTPUT_AUDIO_CODEC_PARAMETERS} "
        # MPEG4 output
        f'"{output_filename}"'
    )
    LOGGER.info(f"Running: {cmd}")
    subprocess.run(cmd, shell=True, check=True)


def run_trip_encoding(trip_job_def):
    """
    Run video concatenation and encoding for a single trip.
    """
    trip = trip_job_def.trip
    trip_full_name = trip.get_full_trip_name()

    ffmpeg_input_filename = generate_ffmpeg_input_file(trip.raw_segments)
    run_ffmpeg(
        ffmpeg_input_filename=ffmpeg_input_filename,
        output_video_name=trip_full_name,
    )
    LOGGER.info(f"Trip encoding completed: {trip_full_name}")


def run_trips_command(parsed_args):
    """
    Entry point for the "trips" subcommand.
    """
    segments = collect_raw_video_segments(parsed_args.raw_video_folder)
    trips = group_segments_into_trips(segments, parsed_args.min_trip_gap_hours)

    LOGGER.info(f"Discovered {len(trips)} trips")
    for trip in trips:
        LOGGER.info(trip)

    job_defs = [
        TripEncodeJobDefinition(trip, parsed_args.raw_video_folder) for trip in trips
    ]
    with multiprocessing.Pool(parsed_args.job_count) as process_pool:
        process_pool.map(run_trip_encoding, job_defs)


def run_range_command(parsed_args):
    """
    Entry point for the "range" subcommand.
    """
    segments = collect_raw_video_segments(
        raw_folder_path=parsed_args.raw_video_folder,
        start_index=parsed_args.start_index,
        end_index=parsed_args.end_index,
    )
    ffmpeg_input_filename = generate_ffmpeg_input_file(segments)
    output_name = (
        parsed_args.output_name
        if parsed_args.output_name
        else f"dashcam-{parsed_args.start_index}-{parsed_args.end_index}"
    )
    run_ffmpeg(
        ffmpeg_input_filename=ffmpeg_input_filename,
        output_video_name=output_name,
    )


def main():
    parsed_args = parse_command_line_args(sys.argv[1:])
    if parsed_args.command == "trips":
        run_trips_command(parsed_args)
    elif parsed_args.command == "range":
        run_range_command(parsed_args)


if __name__ == "__main__":
    main()