Is this HLS / dash convert system reasonable?
my videos look low-res, are the bitrates ok?
Create HLS (master .m3u8) and MPEG-DASH (.mpd) streaming packages from a single
source file using ffmpeg. Generates a sensible ABR ladder and CMAF-compatible
fragmented MP4 segments.
Usage (from project root):
python -m website_utils.hls_dash_converter \
--input C:\\path\\to\\ULC_REEL_2023_1080p.mp4 \
--output-dir C:\\Users\\GeorgeAdamon\\Documents\\GitHub\\ULC\\automation\\src\\python\\local\\assets\\video \
This will create the following files:
\stream.m3u8
\stream.mpd
And their corresponding segment folders/files alongside them.
- Requires ffmpeg available in PATH. Prints a helpful error if not found.
- By default prepares the following ladder (height → max bitrate):
1080p ~ 5000k, 720p ~ 2800k, 540p ~ 1800k, 360p ~ 1000k, 240p ~ 500k
You can prune resolutions above the source height automatically.
- Uses 4s segments, aligned GOPs, and AAC stereo audio.
"""
from future import annotations
import argparse
import os
import shutil
import subprocess
import sys
import json
from typing import List, Tuple
def _check_ffmpeg() -> str:
"""Return ffmpeg path, raise SystemExit with message if not available."""
ffmpeg = shutil.which(“ffmpeg”)
if not ffmpeg:
sys.exit(
“ffmpeg was not found in PATH. Please install ffmpeg and ensure it is accessible.\n”
“Windows (scoop): scoop install ffmpeg\n”
“Windows (choco): choco install ffmpeg\n”
“macOS (brew): brew install ffmpeg\n”
“Linux (apt): sudo apt-get install ffmpeg\n”
)
return ffmpeg
def _probe_height(ffprobe: str, input_path: str) -> int:
"""Probe input video height using ffprobe. Returns 0 if not available."""
try:
out = subprocess.check_output(
[
ffprobe,
“-v”,
“error”,
“-select_streams”,
“v:0”,
“-show_entries”,
“stream=height”,
“-of”,
“json”,
input_path,
],
stderr=subprocess.STDOUT,
)
data = json.loads(out.decode(“utf-8”, “ignore”))
streams = data.get(“streams”, [])
if streams:
return int(streams[0].get(“height”) or 0)
except Exception:
pass
return 0
def _probe_has_audio(ffprobe: str, input_path: str) -> bool:
"""Return True if the input has at least one audio stream.
Falls back to True if ffprobe is unavailable to preserve previous behavior,
but our callers will only pass this when ffprobe exists.
out = subprocess.check_output(
stderr=subprocess.STDOUT,
data = json.loads(out.decode("utf-8", "ignore"))
streams = data.get("streams", [])
# If probing fails, assume audio may exist to keep compatibility
def _build_ladder(src_height: int) -> List[Tuple[int, int, int]]:
"""Return a ladder as list of tuples: (height, video_bitrate_k, maxrate_k).
Prunes entries above the source height if src_height > 0.
ladder = [l for l in ladder if l[0] <= src_height]
# Ensure at least the lowest rung exists
ladder = [(240, 500, 800)]
def run_hls(
ffmpeg: str,
input_path: str,
out_dir: str,
basename: str,
ladder: List[Tuple[int, int, int]],
audio_bitrate: int = 128,
has_audio: bool = True,
) -> None:
"""Generate multi-variant HLS master and segments."""
master_path = os.path.join(out_dir, f”{basename}.m3u8”)
# Prepare filter and maps
args: List[str] = [ffmpeg, "-y", "-i", input_path]
# Common encoder settings
"-x264opts", "keyint=96:min-keyint=96:no-scenecut",
"-b:a", f"{audio_bitrate}k",
for idx, (h, br, maxr) in enumerate(ladder):
scale = f"scale=-2:{h}:flags=bicubic"
vf_filters.append(f"[{idx}:v]format=yuv420p,{scale}[v{idx}]")
# We'll use -filter_complex to generate scaled outputs from the single input video stream
# Map input video stream to [0:v]
filter_complex_parts = []
for idx, (h, br, maxr) in enumerate(ladder):
# Take the first video stream [0:v]
filter_complex_parts.append(f"[0:v]scale=-2:{h}:flags=bicubic[v{idx}]")
filter_complex = ";".join(filter_complex_parts)
args += ["-filter_complex", filter_complex]
for idx, (h, br, maxr) in enumerate(ladder):
"-bufsize", f"{maxr*2}k",
# We will construct var_stream_map below once we know has_audio
# If audio is present, map it once so all variants can reference a:0
# Build var_stream_map according to presence of audio
# We created N video streams first (v:0..v:N-1) then 1 audio stream (a:0)
for idx, (h, _br, _maxr) in enumerate(ladder):
var_stream_map.append(f"v:{idx},a:0,name:{h}p")
for idx, (h, _br, _maxr) in enumerate(ladder):
var_stream_map.append(f"v:{idx},name:{h}p")
"-hls_playlist_type", "vod",
"-hls_flags", "independent_segments+split_by_time",
"-master_pl_name", f"{basename}.m3u8",
"-var_stream_map", " ".join(var_stream_map),
os.path.join(out_dir, f"{basename}_%v.m3u8"),
os.makedirs(out_dir, exist_ok=True)
os.makedirs(out_dir, exist_ok=True)
subprocess.run(args, check=True, capture_output=True)
except subprocess.CalledProcessError as e:
sys.stderr.write("\n[HLS] ffmpeg command failed. Full command:\n" + " ".join(args) + "\n\n")
sys.stderr.write(e.stderr.decode("utf-8", "ignore"))
def run_dash(
ffmpeg: str,
input_path: str,
out_dir: str,
basename: str,
ladder: List[Tuple[int, int, int]],
audio_bitrate: int = 128,
has_audio: bool = True,
) -> None:
"""Generate multi-bitrate DASH MPD and segments."""
mpd_path = os.path.join(out_dir, f”{basename}.mpd”)
args: List[str] = [ffmpeg, "-y", "-i", input_path]
# We'll create multiple outputs using filter_complex similar to HLS
filter_complex_parts = []
for idx, (h, br, maxr) in enumerate(ladder):
filter_complex_parts.append(f"[0:v]scale=-2:{h}:flags=bicubic[v{idx}]")
filter_complex = ";".join(filter_complex_parts)
args += ["-filter_complex", filter_complex]
# Map video variants and (optionally) audio
for idx, (h, br, maxr) in enumerate(ladder):
f"-maxrate:v:{idx}", f"{maxr}k",
f"-bufsize:v:{idx}", f"{maxr*2}k",
"-x264opts", "keyint=96:min-keyint=96:no-scenecut",
"-b:a", f"{audio_bitrate}k",
"-init_seg_name", f"{basename}_init_$RepresentationID$.m4s",
"-media_seg_name", f"{basename}_chunk_$RepresentationID$_$Number$.m4s",
("id=0,streams=v id=1,streams=a" if has_audio else "id=0,streams=v"),
os.makedirs(out_dir, exist_ok=True)
subprocess.run(args, check=True, capture_output=True)
except subprocess.CalledProcessError as e:
sys.stderr.write("\n[DASH] ffmpeg command failed. Full command:\n" + " ".join(args) + "\n\n")
sys.stderr.write(e.stderr.decode("utf-8", "ignore"))
def main() -> None:
parser = argparse.ArgumentParser(description=“Create HLS and DASH outputs with ffmpeg.”)
parser.add_argument(“—input”, required=True, help=“Path to input video file (e.g., .mp4)”)
parser.add_argument(“—output-dir”, required=True, help=“Directory to place outputs”)
parser.add_argument(“—basename”, default=“stream”, help=“Base name for manifest files (default: stream)”)
parser.add_argument(“—audio-kbps”, type=int, default=128, help=“Audio bitrate per rendition in kbps”)
args = parser.parse_args()
ffprobe = shutil.which("ffprobe") or ffmpeg # ffprobe often alongside ffmpeg
input_path = os.path.abspath(args.input)
out_dir = os.path.abspath(args.output_dir)
if not os.path.isfile(input_path):
sys.exit(f"Input file not found: {input_path}")
src_h = _probe_height(ffprobe, input_path) if ffprobe else 0
has_audio = _probe_has_audio(ffprobe, input_path) if ffprobe else True
ladder = _build_ladder(src_h)
print(f"Detected source height: {src_h or 'unknown'}")
for h, br, maxr in ladder:
print(f" - {h}p @ {br}k (max {maxr}k)")
print(f"Audio stream detected: {has_audio}")
# print("\nGenerating HLS…")
# run_hls(ffmpeg, input_path, out_dir, basename, ladder, audio_bitrate=args.audio_kbps, has_audio=has_audio)
# print(f"HLS master: {os.path.join(out_dir, basename + '.m3u8')}")
print("\nGenerating DASH…")
run_dash(ffmpeg, input_path, out_dir, basename, ladder, audio_bitrate=args.audio_kbps, has_audio=has_audio)
print(f"DASH MPD: {os.path.join(out_dir, basename + '.mpd')}")
if name == “main”:
main()