Spaces:
Sleeping
Sleeping
File size: 2,768 Bytes
8b4f8f9 a4f374c 8b4f8f9 21b1e7f 8b4f8f9 a4f374c 8b4f8f9 a4f374c ca7cbe1 a4f374c ca7cbe1 a4f374c ca7cbe1 a4f374c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 |
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
|