Python wrapper around FFmpeg to turn VR180/360 half-equirect videos into flat 2D HEVC (H.265), with configurable quality, HW/SW encoder selection, and per-file logging for macOS M2/M3/M4.

vr180_to_flat_h265_batch_hvc1.py

#!/usr/bin/env python3
"""
vr180_to_flat_h265_batch_hvc1.py
VR180 SBS (or mono) -> Flat 2D HEVC (H.265) batch encoder for macOS (Apple Silicon).

What this does
- Reads video files from an input folder (optionally recursive).
- If the source is VR180 SBS, it crops the left eye (half-width) then uses FFmpeg `v360` to flatten.
- Encodes to HEVC using:
  - Hardware: `hevc_videotoolbox` (fast), with SW fallback to `libx265` if HW fails.
  - Or software-only: `libx265`.
- Writes MP4 outputs tagged as **hvc1** for better Apple/QuickTime compatibility (FFmpeg often defaults to `hev1`). [web:24][web:8]

Quick start
❯ python ~/Desktop/script/vr180_to_flat_h265_batch_hvc1.py -i ~/Desktop/source -o ~/Desktop/output -t --test-seconds 5 --overwrite --encoder hw --no-v360-hfov

Example output
ℹ Found 1 video file(s)
ℹ Output: /Users/you/Desktop/output/vr_to_flat_h265_hvc1_YYYYMMDDHHMMSS
ℹ Encoder Mode: HW
ℹ HW Quality: 95 | SW CRF: 18 | Preset: slow
▶ Encoding input.mp4 → input_FLAT_480p_hvc1.mp4 [HW] (src: 28265kbps)...
✓ input_FLAT_480p_hvc1.mp4 [HW] success

✓ Done in 00:00:05 | Files: OK=1 FAIL=0
✓ Logs and outputs: /Users/you/Desktop/output/vr_to_flat_h265_hvc1_YYYYMMDDHHMMSS

Notes
- `--no-v360-hfov` maximizes compatibility across FFmpeg builds (some builds omit `hfov`). [web:12]
- `-tag:v hvc1` is the key QuickTime compatibility fix for HEVC in MP4. [web:24][web:8]
"""

from __future__ import annotations

import argparse
import json
import re
import shutil
import subprocess
import sys
import time
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
from typing import Dict, List, Tuple, Optional, Set

# =========================
# CONFIGURATION (edit)
# =========================

DEFAULT_INPUT_DIR = Path.home() / "Desktop" / "source"

# RES_LADDER = [2160, 1440, 1080, 720, 540, 480, 360]
RES_LADDER = [720]
DEFAULT_MAX_ONLY = True

TEST_MAX_HEIGHT = 480

# If your FFmpeg build supports v360 hfov/vfov, this is a reasonable default.
DEFAULT_V360_HFOV = 90.0

PITCH_VARIANTS = [-10, -20, -30]

# H.265 QUALITY CONFIG
HW_HEVC_QUALITY = 95
HW_HEVC_TARGET_BITRATE = None  # e.g. "50M" or "50000k" or None
HW_HEVC_MAX_BITRATE = None     # e.g. "200M" or None

DEFAULT_X265_CRF = 18
DEFAULT_X265_PRESET = "slow"
SW_HEVC_TARGET_BITRATE = None  # e.g. "50M" or "50000k" or None
SW_HEVC_MAX_BITRATE = None     # e.g. "100M" or None

AUDIO_CODEC = "aac"  # "aac" or "flac"
AUDIO_RATE = 48000
AUDIO_BITRATE = "256k"

CONTAINER_EXT = ".mp4"
MOV_FASTSTART = True

# Add common input formats (you can extend this)
VIDEO_EXTS = {".mkv", ".mp4", ".m4v", ".mov", ".webm", ".avi", ".ts", ".m2ts"}

# Output filename marker to reflect the `hvc1` tagging fix
OUT_TAG_SUFFIX = "_hvc1"

# =========================
# Data structures
# =========================

@dataclass(frozen=True)
class TestSegment:
    clip: int
    start: int
    dur: int
    region: str

# =========================
# Utilities
# =========================

def runstamp_compact() -> str:
    return time.strftime("%Y%m%d%H%M%S", time.localtime())

def format_time_sec(sec: float) -> str:
    s = max(0, int(sec))
    h = s // 3600
    m = (s % 3600) // 60
    r = s % 60
    return f"{h:02d}:{m:02d}:{r:02d}"

def has_cmd(name: str) -> bool:
    return shutil.which(name) is not None

def runcmd(
    cmd: List[str],
    *,
    capture: bool = False,
    check: bool = True,
    stdout=None,
    stderr=None,
) -> subprocess.CompletedProcess:
    if capture:
        return subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=check)
    return subprocess.run(cmd, stdout=stdout, stderr=stderr, check=check)

def check_dependencies() -> None:
    for tool in ("ffmpeg", "ffprobe"):
        if not has_cmd(tool):
            raise RuntimeError(f"Missing tool '{tool}'. Install via: brew install ffmpeg")

def ffprobe_json(path: Path) -> Dict:
    p = runcmd(
        ["ffprobe", "-v", "error", "-print_format", "json", "-show_format", "-show_streams", str(path)],
        capture=True,
        check=True,
    )
    return json.loads(p.stdout)

def get_video_wh(probe: Dict) -> Tuple[int, int]:
    streams = probe.get("streams") or []
    vstreams = [s for s in streams if s.get("codec_type") == "video"]
    if not vstreams:
        raise RuntimeError("No video stream found.")
    w = int(vstreams[0].get("width") or 0)
    h = int(vstreams[0].get("height") or 0)
    if w <= 0 or h <= 0:
        raise RuntimeError("Could not determine width/height via ffprobe.")
    return w, h

def get_bitrate_kbps(path: Path) -> Optional[int]:
    probe = ffprobe_json(path)
    fmt = probe.get("format") or {}
    bitratestr = fmt.get("bit_rate")
    if bitratestr:
        try:
            return int(bitratestr) // 1000
        except Exception:
            return None
    return None

def sanitize_stem(s: str, maxlen: int = 80) -> str:
    s = (s or "").strip()
    s = re.sub(r"\s+", "_", s)
    s = re.sub(r"[^A-Za-z0-9._-]+", "", s)
    s = s.strip("._-")
    if not s:
        s = "file"
    return s[:maxlen]

def itervideo_files(srcdir: Path, *, recursive: bool) -> List[Path]:
    if recursive:
        files = [p for p in srcdir.rglob("*") if p.is_file() and p.suffix.lower() in VIDEO_EXTS]
    else:
        files = [p for p in srcdir.iterdir() if p.is_file() and p.suffix.lower() in VIDEO_EXTS]
    return sorted(files)

def guess_in_stereo(srcw: int, srch: int, filename: str, mode: str) -> str:
    if mode in ("sbs", "mono"):
        return mode

    name = filename.lower()
    if any(x in name for x in ("lr", "l_r", "sbs", "sidebyside", "left_right", "leftright")):
        return "sbs"

    # Typical VR180 SBS files like 5400x2700 (2:1), treat as SBS.
    if srch > 0:
        ar = srcw / float(srch)
        if 1.85 <= ar <= 2.15:
            return "sbs"
    return "mono"

def pick_output_heights(srch: int, *, maxonly: bool, caph: Optional[int]) -> List[int]:
    ladder = list(RES_LADDER)
    if caph is not None:
        ladder = [h for h in ladder if h <= caph]
    candidates = [h for h in ladder if h <= srch]
    if not candidates:
        return [min(ladder)] if ladder else [srch]
    if maxonly:
        return [max(candidates)]
    return candidates

# =========================
# FFmpeg capability probing
# =========================

@lru_cache(maxsize=1)
def ffmpeg_v360_supported_options() -> Set[str]:
    """
    Parses `ffmpeg -hide_banner -h filter=v360` output and returns a set of option names.
    Cached so it's only executed once per script run (was previously once per encode). [web:12]
    """
    try:
        p = runcmd(["ffmpeg", "-hide_banner", "-h", "filter=v360"], capture=True, check=True)
        text = (p.stdout or "") + "\n" + (p.stderr or "")
    except Exception:
        return set()

    opts: Set[str] = set()
    for line in text.splitlines():
        line = line.strip()
        if not line or line.startswith("Filter") or line.startswith("v360"):
            continue
        m = re.match(r"^([A-Za-z0-9_]+)\s+<", line)
        if m:
            opts.add(m.group(1))
    return opts

# =========================
# Core FFmpeg builders
# =========================

def build_ffmpeg_filter_vr2flat(
    *,
    inw: int,
    inh: int,
    outw: int,
    outh: int,
    stereomode: str,
    projection: str,
    v360_hfov: Optional[float],
    pitch: float = 0.0,
) -> str:
    inproj = "hequirect" if projection == "hequirect" else "equirect"

    if stereomode == "sbs":
        cropfilter = f"crop=w={inw//2}:h={inh}:x=0:y=0"
    else:
        cropfilter = "copy"

    v360_opts_supported = ffmpeg_v360_supported_options()

    v360opts = [f"input={inproj}", "output=flat", f"pitch={pitch}", "yaw=0", "roll=0"]

    if v360_hfov is not None and "hfov" in v360_opts_supported:
        v360opts.append(f"hfov={v360_hfov}")

    v360arg = ":".join(v360opts)

    filtergraph = (
        f"[0:v]{cropfilter},"
        f"v360={v360arg},"
        f"scale={outw}:{outh}:force_original_aspect_ratio=decrease,"
        f"pad={outw}:{outh}:(ow-iw)/2:(oh-ih)/2[padded]"
    )
    return filtergraph

def build_hw_hevc_video_args(targetbitrate: Optional[str] = None) -> List[str]:
    # `-pix_fmt yuv420p` is often a good “plays everywhere” choice on Apple apps. [web:22][web:25]
    args = ["-c:v", "hevc_videotoolbox", "-q:v", str(HW_HEVC_QUALITY), "-color_range", "tv", "-pix_fmt", "yuv420p"]
    if targetbitrate:
        args.extend(["-b:v", targetbitrate])
    if HW_HEVC_MAX_BITRATE:
        args.extend(["-maxrate", HW_HEVC_MAX_BITRATE])
    return args

def build_sw_hevc_video_args(crf: Optional[int] = None, targetbitrate: Optional[str] = None) -> List[str]:
    args = ["-c:v", "libx265", "-preset", DEFAULT_X265_PRESET, "-color_range", "tv", "-pix_fmt", "yuv420p"]
    if targetbitrate:
        args.extend(["-b:v", targetbitrate])
        if SW_HEVC_MAX_BITRATE:
            args.extend(["-maxrate", SW_HEVC_MAX_BITRATE])
    else:
        crfval = crf if crf is not None else DEFAULT_X265_CRF
        args.extend(["-crf", str(crfval)])
    return args

def build_audio_args() -> List[str]:
    args = ["-c:a", AUDIO_CODEC, "-ar", str(AUDIO_RATE)]
    if AUDIO_CODEC == "aac":
        args.extend(["-b:a", AUDIO_BITRATE])
    return args

def encode_one(
    *,
    srcfile: Path,
    rundir: Path,
    outheight: int,
    pitchvariant: float,
    encodermode: str,
    overwrite: bool,
    testmode: bool,
    testseconds: int,
    stereomode: str,
    projection: str,
    v360_hfov: Optional[float],
) -> None:
    probe = ffprobe_json(srcfile)
    srcw, srch = get_video_wh(probe)
    srcbitratekbps = get_bitrate_kbps(srcfile)

    cap = TEST_MAX_HEIGHT if testmode else None
    if cap is not None:
        outheight = min(outheight, cap)

    outw = int(outheight * srcw / srch) if srch else int(outheight * 16 / 9)
    if outw % 2 != 0:
        outw += 1

    stereo = guess_in_stereo(srcw, srch, srcfile.name, stereomode)

    filtergraph = build_ffmpeg_filter_vr2flat(
        inw=srcw,
        inh=srch,
        outw=outw,
        outh=outheight,
        stereomode=stereo,
        projection=projection,
        v360_hfov=v360_hfov,
        pitch=pitchvariant,
    )

    filestem = sanitize_stem(srcfile.stem)
    outsubdir = rundir / filestem
    outsubdir.mkdir(parents=True, exist_ok=True)

    logsdir = outsubdir / "logs"
    logsdir.mkdir(exist_ok=True)

    metadir = outsubdir / "_metadata"
    metadir.mkdir(exist_ok=True)
    (metadir / "ffprobe.json").write_text(json.dumps(probe, indent=2), encoding="utf-8")

    pitchsuffix = f"_pitch{pitchvariant:+.0f}" if pitchvariant != 0 else ""
    outfile = outsubdir / f"{filestem}_FLAT_{outheight}p{pitchsuffix}{OUT_TAG_SUFFIX}{CONTAINER_EXT}"

    if outfile.exists() and not overwrite:
        print(f"↷ Skipping {outfile.name} (exists; use --overwrite)")
        return

    logfile = logsdir / f"ffmpeg_{runstamp_compact()}.log"

    hwtarget = HW_HEVC_TARGET_BITRATE
    swtarget = SW_HEVC_TARGET_BITRATE
    if srcbitratekbps:
        bump = 5000
        derived = f"{max(srcbitratekbps, srcbitratekbps + bump)}k"
        if not hwtarget:
            hwtarget = derived
        if not swtarget:
            swtarget = derived

    containerflags: List[str] = []
    if MOV_FASTSTART and CONTAINER_EXT.lower() in (".mp4", ".mov", ".m4v"):
        containerflags.extend(["-movflags", "faststart"])

    # Force MP4 HEVC sample entry to hvc1 instead of hev1 for Apple/QuickTime. [web:24][web:8]
    if CONTAINER_EXT.lower() in (".mp4", ".mov", ".m4v"):
        containerflags.extend(["-tag:v", "hvc1"])

    def build_base_cmd() -> List[str]:
        cmd = ["ffmpeg", "-hide_banner"]
        cmd.append("-y" if overwrite else "-n")
        cmd.extend(["-i", str(srcfile)])
        if testmode:
            cmd.extend(["-t", str(testseconds)])
        cmd.extend(["-filter_complex", filtergraph, "-map", "[padded]", "-map", "0:a?"])
        return cmd

    attempts: List[Tuple[str, List[str]]] = []
    if encodermode == "hw":
        attempts.append(("HW", build_hw_hevc_video_args(hwtarget)))
        attempts.append(("SW", build_sw_hevc_video_args(None, swtarget)))
    else:
        attempts.append(("SW", build_sw_hevc_video_args(None, swtarget)))

    lasterror: Optional[Exception] = None
    for label, videoargs in attempts:
        srcbrstr = f"{srcbitratekbps}kbps" if srcbitratekbps else "unknown"
        print(f"▶ Encoding {srcfile.name} → {outfile.name} [{label}] (src: {srcbrstr})...")
        cmd = build_base_cmd()
        cmd.extend(videoargs)
        cmd.extend(build_audio_args())
        cmd.extend(containerflags)
        cmd.append(str(outfile))

        try:
            with open(logfile, "a", encoding="utf-8") as logfh:
                runcmd(cmd, stdout=logfh, stderr=logfh, check=True)
            print(f"✓ {outfile.name} [{label}] success")
            return
        except subprocess.CalledProcessError as e:
            lasterror = e
            with open(logfile, "a", encoding="utf-8") as logfh:
                logfh.write(f"\n\n--- Attempt failed ({label}) ---\n{e}\n")
            if label == "HW":
                print(f"✗ {outfile.name} [{label}] failed, falling back to SW…", file=sys.stderr)
            else:
                print(f"✗ {outfile.name} [{label}] failed.", file=sys.stderr)

    raise RuntimeError(f"FFmpeg error for {srcfile} (see {logfile} for details)") from lasterror

# =========================
# CLI
# =========================

def main() -> int:
    parser = argparse.ArgumentParser(
        description="VR180 SBS → Flat 2D H.265 Batch Encoder (macOS, Apple Silicon HW/SW) with hvc1 MP4 tag",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=(
            "Examples:\n"
            "  %(prog)s -i ~/Desktop/source -o ~/Desktop/output\n"
            "  %(prog)s -o ~/Desktop/output (uses DEFAULT_INPUT_DIR)\n"
            "  %(prog)s -i input -o output -r -p\n"
            "  %(prog)s -i input -o output -t --test-seconds 30\n"
            "  %(prog)s -i input -o output --encoder sw\n"
        ),
    )

    parser.add_argument(
        "-i", "--input",
        type=Path,
        default=DEFAULT_INPUT_DIR,
        help=f"Source directory containing video files (default: {DEFAULT_INPUT_DIR})",
    )
    parser.add_argument(
        "-o", "--output",
        type=Path,
        required=True,
        help="Output directory for encoded files",
    )
    parser.add_argument("-r", "--recursive", action="store_true", help="Scan subdirectories recursively")
    parser.add_argument("-p", "--pitch-variants", action="store_true", help="Create pitch variant outputs (default max res only)")
    parser.add_argument("--stereo", choices=["auto", "sbs", "mono"], default="auto", help="Stereo mode: auto-detect (default), side-by-side, or monoscopic")
    parser.add_argument("--projection", choices=["hequirect", "full360"], default="hequirect", help="Input projection: VR180 half-equirect (default) or full equirectangular")
    parser.add_argument("-t", "--test", action="store_true", help="Test mode: limit output to 480p and first N seconds")
    parser.add_argument("--test-seconds", type=int, default=30, help="Duration of test output in seconds (default 30)")
    parser.add_argument("--overwrite", action="store_true", help="Overwrite existing output files")
    parser.add_argument("--encoder", choices=["hw", "sw"], default="hw", help="Preferred encoder: hw VideoToolbox (default, falls back to sw) or sw libx265 only")

    parser.add_argument("--v360-hfov", type=float, default=DEFAULT_V360_HFOV, help="Requested v360 hfov (degrees). Used only if your FFmpeg v360 supports hfov. (default: 90)")
    parser.add_argument("--no-v360-hfov", action="store_true", help="Do not pass any FOV option to v360 (max compatibility).")

    args = parser.parse_args()

    try:
        check_dependencies()
    except RuntimeError as e:
        print(e, file=sys.stderr)
        return 1

    if not args.input.is_dir():
        print(f"Input directory not found: {args.input}", file=sys.stderr)
        return 1

    args.output.mkdir(parents=True, exist_ok=True)

    # Updated run directory name to reflect the hvc1 tagging.
    rundir = args.output / f"vr_to_flat_h265_hvc1_{runstamp_compact()}"
    rundir.mkdir(parents=True, exist_ok=True)

    videofiles = itervideo_files(args.input, recursive=args.recursive)
    if not videofiles:
        print(f"No video files found in: {args.input}", file=sys.stderr)
        return 1

    v360_hfov = None if args.no_v360_hfov else float(args.v360_hfov)

    print(f"ℹ Found {len(videofiles)} video file(s)")
    print(f"ℹ Output: {rundir}")
    print(f"ℹ Encoder Mode: {args.encoder.upper()}")
    print(f"ℹ HW Quality: {HW_HEVC_QUALITY} | SW CRF: {DEFAULT_X265_CRF} | Preset: {DEFAULT_X265_PRESET}")

    startall = time.time()
    ok = 0
    fail = 0

    for f in videofiles:
        try:
            probe = ffprobe_json(f)
            _, srch = get_video_wh(probe)
            cap = TEST_MAX_HEIGHT if args.test else None
            heights = pick_output_heights(srch, maxonly=DEFAULT_MAX_ONLY, caph=cap)

            for h in heights:
                encode_one(
                    srcfile=f,
                    rundir=rundir,
                    outheight=h,
                    pitchvariant=0.0,
                    encodermode=args.encoder,
                    overwrite=args.overwrite,
                    testmode=args.test,
                    testseconds=args.test_seconds,
                    stereomode=args.stereo,
                    projection=args.projection,
                    v360_hfov=v360_hfov,
                )

                if args.pitch_variants:
                    for pitchv in PITCH_VARIANTS:
                        encode_one(
                            srcfile=f,
                            rundir=rundir,
                            outheight=h,
                            pitchvariant=float(pitchv),
                            encodermode=args.encoder,
                            overwrite=args.overwrite,
                            testmode=args.test,
                            testseconds=args.test_seconds,
                            stereomode=args.stereo,
                            projection=args.projection,
                            v360_hfov=v360_hfov,
                        )

            ok += 1

        except Exception as e:
            fail += 1
            print(f"✗ {e}", file=sys.stderr)

    total = format_time_sec(time.time() - startall)
    print(f"\n✓ Done in {total} | Files: OK={ok} FAIL={fail}")
    print(f"✓ Logs and outputs: {rundir}")
    return 0 if fail == 0 else 1

if __name__ == "__main__":
    raise SystemExit(main())