Convert a video to HLS without re-encode, or re-encode it in multiple resolutions in a single HLS playlist

convert-to-hls.py

"""
https://www.mux.com/articles/how-to-convert-mp4-to-hls-format-with-ffmpeg-a-step-by-step-guide
https://www.funvisiontutorials.com/2024/10/guide-to-hls-live-streaming-with-ffmpeg.html
"""
import argparse
import subprocess
from pathlib import Path
from pprint import pprint
from typing import TypedDict


class ResolutionParam(TypedDict):
    height: int
    video_bitrate: str
    video_maxrate: str
    video_bufsize: str
    audio_bitrate: str


# TODO: Think about the bitrates instead of blindingly doubling everything
RESOLUTIONS_PARAMS: dict[str, ResolutionParam] = {
    "4k": {
        "height": 2160,
        "video_bitrate": "11200k",
        "video_maxrate": "11984k",
        "video_bufsize": "16800k",
        "audio_bitrate": "256k",
    },
    "1080p": {
        "height": 1080,
        "video_bitrate": "5600k",
        "video_maxrate": "5992k",
        "video_bufsize": "8400k",
        "audio_bitrate": "192k",
    },
    "720p": {
        "height": 720,
        "video_bitrate": "2800k",
        "video_maxrate": "2996k",
        "video_bufsize": "4200k",
        "audio_bitrate": "128k",
    },
    "480p": {
        "height": 480,
        "video_bitrate": "1400k",
        "video_maxrate": "1498k",
        "video_bufsize": "2100k",
        "audio_bitrate": "96k",
    },
    "360p": {
        "height": 360,
        "video_bitrate": "700k",
        "video_maxrate": "749k",
        "video_bufsize": "1050k",
        "audio_bitrate": "72k",
    },
}


def parse_args(args: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "video",
        help="The video file to convert",
        type=Path,
    )
    parser.add_argument(
        "-r",
        "--resolution",
        help="""Resolution to re-encode to (can be specified multiple times).
        If none specified, will convert to HLS as-is.""",
        choices=RESOLUTIONS_PARAMS.keys(),
        action="append",
        dest="resolutions",
    )
    parsed_args = parser.parse_args(args)
    if parsed_args.resolutions is not None:
        parsed_args.resolutions = set(parsed_args.resolutions)
    return parsed_args


def convert(file_path: Path, resolutions: set[str] | None = None) -> None:
    command = ["ffmpeg", "-i", str(file_path)]

    if resolutions is None:
        command.extend(
            [
                "-c",
                "copy",
                "-hls_time",
                "10",
                "-hls_list_size",
                "0",
                f"{file_path.stem}.m3u8",
            ]
        )
    else:
        command.append("-filter_complex")
        command.extend(generate_ffmpeg_resolutions_args(resolutions))
        command.extend(
            [
                "-f",
                "hls",
                "-hls_time",
                "10",
                "-hls_playlist_type",
                "vod",
                "-hls_flags",
                "independent_segments",
                "-hls_segment_type",
                "mpegts",
                "-hls_segment_filename",
                "stream_%v/data%03d.ts",
                "-master_pl_name",
                "master.m3u8",
                "-var_stream_map",
                " ".join([f"v:{i},a:{i}" for i in range(len(resolutions))]),
                "stream_%v/playlist.m3u8",
            ]
        )

    subprocess.run(command)


def generate_ffmpeg_resolutions_args(resolutions: set[str]) -> list[str]:
    split_arg = f"[0:v]split={len(resolutions)}"
    split_arg_2: list[str] = []
    video_args = []
    audio_args = []

    for i, (resolution, parameters) in enumerate(filter(lambda r: (r[0] in resolutions), RESOLUTIONS_PARAMS.items())):
            split_arg += f"[v{parameters['height']}]"
            split_arg_2.append(
                f"[v{parameters['height']}]scale=-2:{parameters['height']}[v{parameters['height']}out]"
            )
            video_args.extend(
                [
                    "-map",
                    f"[v{parameters['height']}out]",
                    f"-c:v:{i}",
                    "libx264",
                    f"-b:v:{i}",
                    parameters["video_bitrate"],
                    f"-maxrate:v:{i}",
                    parameters["video_maxrate"],
                    f"-bufsize:v:{i}",
                    parameters["video_bufsize"],
                ]
            )
            audio_args.extend(
                [
                    "-map",
                    "a:0",
                    "-c:a",
                    "aac",
                    f"-b:a:{i}",
                    parameters["audio_bitrate"],
                    "-ac",
                    "2",
                ]
            )

    final_args = [
        "; ".join([split_arg, *split_arg_2]),
        *video_args,
        *audio_args,
    ]
    return final_args


def main() -> int:
    args = parse_args()
    pprint(args)

    convert(args.video, args.resolutions)

    return 0


def run() -> None:
    raise SystemExit(main())