video_encoder/worker/encoder.py

import subprocess
import os
import shutil
from pathlib import Path
from typing import List, Dict
from ..config import EncodingConfig
import logging

logger = logging.getLogger(__name__)

class FFmpegEncoder:
    def __init__(self, input_path: str, output_dir: str):
        # Verify FFmpeg is available
        if not shutil.which("ffmpeg"):
            raise RuntimeError("FFmpeg not found in PATH")
        self.input_path = input_path
        self.output_dir = output_dir
        self.base_name = Path(input_path).stem
        os.makedirs(output_dir, exist_ok=True)

    def generate_commands(self) -> List[List[str]]:
        """Generate FFmpeg commands for all resolutions"""
        commands = []
        for res in EncodingConfig.RESOLUTIONS:
            output_path = os.path.join(
                self.output_dir,
                f"{self.base_name}_{res['name']}.m3u8"
            )
            
            cmd = [
                "ffmpeg", "-i", self.input_path,
                "-vf", f"scale={res['width']}:{res['height']}",
                "-c:v", res["codec"],
                "-profile:v", res["profile"],
                "-preset", res["preset"],
                "-b:v", res["video_bitrate"],
                "-c:a", "aac",
                "-b:a", res["audio_bitrate"],
                "-f", "hls",
                "-hls_time", "6",
                "-hls_playlist_type", "vod",
                "-hls_segment_filename", 
                os.path.join(self.output_dir, f"{self.base_name}_{res['name']}_%03d.ts"),
                output_path
            ]
            commands.append(cmd)
        return commands

def encode(self) -> str:
    """Execute encoding commands and return master playlist path"""
    master_playlist = os.path.join(self.output_dir, f"{self.base_name}_master.m3u8")

    with open(master_playlist, "w") as f:
        f.write("#EXTM3U\n")
        for res in reversed(EncodingConfig.RESOLUTIONS):
            f.write(f"#EXT-X-STREAM-INF:BANDWIDTH={res['video_bitrate'].replace('k', '000')},"
                    f"RESOLUTION={res['width']}x{res['height']}\n")
            f.write(f"{self.base_name}_{res['name']}.m3u8\n")

    for cmd in self.generate_commands():
        logger.info(f"Executing FFmpeg command: {' '.join(cmd)}")
        result = subprocess.run(cmd, capture_output=True)
        logger.debug(f"FFmpeg command: {' '.join(cmd)}")
        logger.debug(f"FFmpeg output: {result.stdout.decode()}")
        logger.debug(f"FFmpeg error: {result.stderr.decode()}")
        if result.returncode != 0:
            logger.error(f"FFmpeg command failed with return code {result.returncode}")
            raise subprocess.CalledProcessError(result.returncode, cmd)
        result.check_returncode()

    return master_playlist