#!/usr/bin/env python3 """ hls-encode — encode any video into a multi-bitrate HLS VOD ladder. hls-encode input.video [outdir] [seconds] outdir defaults to the input's basename; the third argument encodes only the first N seconds, for test runs. Everything is derived from the source, so landscape, vertical, square, 4:3, and anamorphic inputs all work without editing the script. --- how the ladder is picked -------------------------------------------------- Renditions are named by their SHORT side (1080, 720, 480 by default), not by width, because that is the one number that means the same thing for a 1920x1080 podcast and a 1080x1920 phone clip. The long side follows from the source's display aspect ratio, rounded to an even number since yuv420p needs both dimensions even. Nothing is ever upscaled: a tier larger than the source is clamped to the source, and tiers that collide after clamping collapse into one. A 640x360 input therefore gets a single 360-tall rendition rather than three copies of the same picture at three bitrates. Bitrates scale with the picture instead of being hardcoded. Each tier carries a bits-per-pixel-per-frame figure, and maxrate = bpp x width x height x fps. The defaults are back-derived from the fixed 1920x1080 ladder these numbers replace, so a 1080p landscape source still comes out at 5000/2800/1200k. Note the default top tier is 1080, so a 4K source is delivered at 1080p. Pass --tiers 2160,1080,720 to keep it. --- subtitles ----------------------------------------------------------------- Off by default. hls-subs(1) adds a WebVTT rendition to a finished stream in seconds without redoing the ladder, and gets three things right that muxing captions in here does not: ffmpeg will not share one subtitle stream between variants, so it is mapped once per rendition and the caption menu lists it three times; all three come out DEFAULT=NO, so captions never switch themselves on; and the muxer writes no X-TIMESTAMP-MAP, which RFC 8216 calls for on MPEG-TS, so cues can drift by the mux delay. If you want them inline anyway: hls-encode --subs auto in.mov out # the .vtt beside the input hls-encode --subs other.vtt in.mov out """ import argparse import json import shutil import subprocess import sys from fractions import Fraction from pathlib import Path # Short side, then how that tier is encoded. bpp is bits per pixel per frame: # maxrate = bpp * w * h * fps, which is what makes the ladder resolution- # agnostic. The three values are what the old fixed 5000/2800/1200k ladder # worked out to at 1920x1080, 1280x720 and 854x480, 30fps. TIERS = [ # short side, crf, profile, bpp (1080, 21, "high", 0.080), (720, 22, "high", 0.101), (480, 23, "main", 0.098), ] SEG = 6 # segment length in seconds FPS_MAX = 60 # ceiling when keeping the source rate # The source rate is kept by default, because halving it is the one thing here # that visibly degrades the picture and it is not obvious from the output that # it happened: every playlist and timestamp still measures perfectly clean, the # video is just half as smooth. `--fps 30` is right for talking heads and wrong # for anything that moves. def die(msg): sys.exit(f"hls-encode: {msg}") def even(x): """Nearest even integer, minimum 2. yuv420p needs both dimensions even.""" return max(2, int(round(x / 2)) * 2) def probe(path): """Display dimensions, source fps, and the audio stream (None if silent).""" out = subprocess.run( ["ffprobe", "-v", "error", "-show_streams", "-of", "json", str(path)], capture_output=True, text=True, ) if out.returncode != 0: die(f"ffprobe failed on {path.name}:\n{out.stderr.strip()}") streams = json.loads(out.stdout).get("streams", []) video = next((s for s in streams if s.get("codec_type") == "video"), None) if not video: die(f"no video stream in {path.name}") audio = next((s for s in streams if s.get("codec_type") == "audio"), None) w, h = int(video["width"]), int(video["height"]) # Anamorphic sources store a non-square sample aspect, so the picture is # wider (or narrower) than its pixel count says. Work in what it should # *look* like, and stamp SAR 1 on the way out. sar = video.get("sample_aspect_ratio") or "1:1" if sar not in ("0:1", "1:1", "N/A"): try: num, den = (int(v) for v in sar.split(":")) if num > 0 and den > 0: w = even(w * num / den) except ValueError: pass # ffmpeg autorotates in the filter graph, so the filters see the display # orientation rather than the stored one. Match that, or a rotated vertical # clip gets a landscape ladder. rot = 0 for sd in video.get("side_data_list") or []: if "rotation" in sd: rot = int(float(sd["rotation"])) if abs(rot) % 180 == 90: w, h = h, w fps = Fraction(0) for key in ("r_frame_rate", "avg_frame_rate"): try: fps = Fraction(video.get(key) or "0/0") except (ZeroDivisionError, ValueError): fps = Fraction(0) if fps > 0: break if fps <= 0: fps = Fraction(30) # unreadable rate; 30 is the safer guess return w, h, fps, audio def ladder(src_w, src_h, fps, tiers): """One rendition per tier, clamped to the source and deduped by dimensions.""" short = min(src_w, src_h) out, seen = [], set() for target, crf, profile, bpp in tiers: side = min(target, short) # clamp rather than skip, so a 900-tall f = side / short # source still gets a top rendition at 900 w, h = even(src_w * f), even(src_h * f) if (w, h) in seen: continue seen.add((w, h)) maxrate = max(200, round(bpp * w * h * fps / 100_000) * 100) out.append({ "w": w, "h": h, "crf": crf, "profile": profile, "maxrate": maxrate, "bufsize": maxrate * 2, "capped": side < target, }) return out def build_command(inp, rends, fps, seg, audio, subs, duration, preset): gop = max(1, int(round(seg * fps))) n = len(rends) acodec = audio["codec_name"] if audio else None cmd = ["ffmpeg", "-y", "-i", str(inp)] if duration: cmd += ["-t", str(duration)] if subs: cmd += ["-i", str(subs)] # One fps conversion ahead of the split, so it is not repeated per # rendition. It also pins a variable-frame-rate source to a constant rate, # which is what the forced keyframes below assume. chain = f"[0:v]fps={fps},split={n}" + "".join(f"[v{i}]" for i in range(n)) + ";" for i, r in enumerate(rends): chain += f"[v{i}]scale=w={r['w']}:h={r['h']}:flags=lanczos,setsar=1[v{i}out];" cmd += ["-filter_complex", chain] for i, r in enumerate(rends): cmd += [ "-map", f"[v{i}out]", f"-c:v:{i}", "libx264", f"-preset:v:{i}", preset, f"-profile:v:{i}", r["profile"], f"-crf:v:{i}", str(r["crf"]), f"-maxrate:v:{i}", f"{r['maxrate']}k", f"-bufsize:v:{i}", f"{r['bufsize']}k", ] cmd += ["-pix_fmt", "yuv420p"] # Segment boundaries have to land on a keyframe in every rendition, or a # player switching bitrate mid-stream splices into the middle of a GOP. cmd += ["-g", str(gop), "-keyint_min", str(gop), "-sc_threshold", "0"] for i in range(n): cmd += [f"-force_key_frames:v:{i}", f"expr:gte(t,n_forced*{seg})"] if acodec: cmd += ["-map", "a:0"] if acodec == "aac": cmd += ["-c:a", "copy"] # already what HLS wants else: cmd += ["-c:a", "aac", "-b:a", "192k"] if int(audio.get("channels") or 2) > 2: cmd += ["-ac", "2"] if subs: cmd += ["-map", "1:s"] * n + ["-c:s", "webvtt"] # Every rendition points at one shared audio group, so the audio is stored # and fetched once instead of per bitrate. parts = [] for i in range(n): v = f"v:{i}" if acodec: v += ",agroup:aud" if subs: v += f",s:{i},sgroup:subs" parts.append(v) if acodec: parts.append("a:0,agroup:aud,name:audio,default:yes") # Start the MPEG-TS timeline at 0. # # ffmpeg's mpegts muxer otherwise applies a default preload/delay, so the # first packet lands ~1.4 s in rather than at PTS 0. That offset is invisible # in playback — players normalise it — but it poisons WebVTT captions, # because X-TIMESTAMP-MAP then has to carry it and players disagree on the # sign. Safari subtracts the mapped value from an already-normalised # timeline, so a *correctly* measured map fires every cue ~1.4 s early. # # With the timeline starting at 0 the map is MPEGTS:0, which means the same # thing under either reading, and media time equals source time equals # transcript time. The class of bug goes away instead of being compensated # for in a second tool. cmd += ["-muxdelay", "0", "-muxpreload", "0"] cmd += [ "-f", "hls", "-hls_time", str(seg), "-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(parts), "stream_%v/playlist.m3u8", ] return cmd def main(): p = argparse.ArgumentParser( prog="hls-encode", description="Encode any video into a multi-bitrate HLS VOD ladder.", epilog="Renditions are named by their short side, so vertical and " "landscape sources both work. Nothing is upscaled.", ) p.add_argument("input", help="source video") p.add_argument("outdir", nargs="?", help="output directory (default: the input's basename)") p.add_argument("seconds", nargs="?", help="encode only the first N seconds") p.add_argument("-t", "--duration", help="same as the positional seconds argument") p.add_argument("--tiers", default=",".join(str(t[0]) for t in TIERS), help="short-side heights, largest first (default: %(default)s)") p.add_argument("--fps", type=float, default=None, help=f"output frame rate (default: the source rate, capped at " f"{FPS_MAX}). Pass 30 for talking heads") p.add_argument("--seg", type=float, default=SEG, help="segment length in seconds (default: %(default)s)") p.add_argument("--preset", default="slow", help="x264 preset (default: %(default)s)") p.add_argument("--subs", default="none", help="'none' (default), 'auto' for the .vtt beside the input, or " "a path. Prefer hls-subs(1) — see the header comment") p.add_argument("-f", "--force", action="store_true", help="overwrite an existing stream in outdir") p.add_argument("-n", "--dry-run", action="store_true", help="print the ffmpeg command without running it") args = p.parse_args() for tool in ("ffmpeg", "ffprobe"): if not shutil.which(tool): die(f"{tool} not found in PATH") inp = Path(args.input).resolve() if not inp.is_file(): die(f"no such file: {args.input}") outdir = Path(args.outdir) if args.outdir else Path(inp.stem) duration = args.duration or args.seconds subs = None if args.subs != "none": cand = inp.with_suffix(".vtt") if args.subs == "auto" else Path(args.subs) if not cand.is_file(): die(f"--subs auto: no {cand.name} beside the input" if args.subs == "auto" else f"no such file: {cand}") subs = cand.resolve() try: heights = [int(t) for t in args.tiers.split(",") if t.strip()] except ValueError: die(f"--tiers wants a comma-separated list of heights, got {args.tiers!r}") if not heights: die("--tiers is empty") # A longer tier list reuses the last tier's crf/profile/bpp; a shorter one # just takes the first few. tiers = [(t, *TIERS[min(i, len(TIERS) - 1)][1:]) for i, t in enumerate(heights)] w, h, src_fps, audio = probe(inp) fps = min(args.fps or FPS_MAX, float(src_fps)) rends = ladder(w, h, fps, tiers) if outdir.exists() and any(outdir.glob("stream_*")) and not args.force: die(f"{outdir}/ already holds a stream — pass --force to overwrite it") orient = "vertical" if h > w else ("square" if h == w else "landscape") acodec = audio["codec_name"] if audio else None print(f"source {inp.name}: {w}x{h} {orient}, {float(src_fps):g}fps, " f"audio {acodec or 'none'}") print(f"output {outdir}/ ({fps:g}fps, {args.seg:g}s segments" f"{', first ' + str(duration) + 's' if duration else ''})") if fps < float(src_fps) - 0.01: print(f" ^ frame rate reduced from {float(src_fps):g} — this is " f"the one setting that makes motion look choppy") if subs: print(f"subtitles {subs.name} (mapped per rendition; hls-subs avoids that)") for i, r in enumerate(rends): note = " [capped to source]" if r["capped"] else "" dims = f"{r['w']}x{r['h']}" print(f" stream_{i} {dims:<11} crf {r['crf']} " f"{r['profile']:<4} maxrate {r['maxrate']}k{note}") if len(rends) < len(heights): dropped = len(heights) - len(rends) print(f" {dropped} tier(s) dropped: the source is too small " f"to tell them apart") cmd = build_command(inp, rends, fps, args.seg, audio, subs, duration, args.preset) if args.dry_run: print("\n" + " ".join( f"'{c}'" if (" " in c or ";" in c) else c for c in cmd )) return # ffmpeg's hls muxer creates the stream_N directories itself, but only once # the parent exists. outdir.mkdir(parents=True, exist_ok=True) for i in range(len(rends)): (outdir / f"stream_{i}").mkdir(exist_ok=True) print() rc = subprocess.run(cmd, cwd=outdir).returncode if rc != 0: die(f"ffmpeg exited {rc}") print(f"\nwrote {outdir}/master.m3u8") print(f"\ndemo it:\n (cd {outdir} && python3 -m http.server 8971) &" f"\n open -a Safari http://localhost:8971/master.m3u8") if __name__ == "__main__": main()