"""
Backend manager for FMUS-VID.
This module manages the available video processing backends.
"""
import logging
from enum import Enum, auto
from typing import Union, Type, Dict, Any, Optional, Tuple, Callable, List
from pathlib import Path
from .backend import Backend as BackendInterface
logger = logging.getLogger(__name__)
[docs]
class Backend(Enum):
"""Enum of available backends."""
FFMPEG = auto()
MOVIEPY = auto()
PYAV = auto()
GSTREAMER = auto()
VLC = auto()
DUMMY = auto() # Added dummy backend for testing/CLI
[docs]
@staticmethod
def from_string(name: str) -> 'Backend':
"""
Convert a string to a Backend enum.
Args:
name: Backend name (case-insensitive)
Returns:
Backend enum
Raises:
ValueError: If the backend is not recognized
"""
name = name.upper()
if name == "FFMPEG":
return Backend.FFMPEG
elif name == "MOVIEPY":
return Backend.MOVIEPY
elif name == "PYAV":
return Backend.PYAV
elif name == "GSTREAMER":
return Backend.GSTREAMER
elif name == "VLC":
return Backend.VLC
elif name == "DUMMY":
return Backend.DUMMY
else:
raise ValueError(f"Unknown backend: {name}")
# Dummy backend implementation
[docs]
class DummyBackend(BackendInterface):
"""
Dummy backend for testing and CLI functionality when no real backends are available.
"""
[docs]
@staticmethod
def is_available() -> bool:
"""Always available."""
return True
[docs]
def load(self, path: Union[str, Path], **kwargs) -> Any:
"""Pretend to load a video."""
logger.warning("Using dummy backend - no real video operations will be performed")
return {"path": path, "loaded": True}
[docs]
def create(self, width: int, height: int, duration: float, fps: float,
color: Tuple[int, int, int] = (0, 0, 0), **kwargs) -> Any:
"""Create a blank video."""
logger.info(f"Dummy create: {width}x{height}, {duration}s at {fps} fps")
return {"width": width, "height": height, "duration": duration, "fps": fps, "color": color}
[docs]
def save(self, video: Any, output_path: Union[str, Path],
progress_callback: Optional[Callable[[float], None]] = None, **kwargs) -> None:
"""Pretend to save a video."""
logger.info(f"Dummy save to {output_path}")
if progress_callback:
progress_callback(1.0) # Simulate completion
[docs]
def get_info(self, video: Any) -> Dict[str, Any]:
"""Return dummy video info."""
return {
"width": 1920,
"height": 1080,
"duration": 30.0,
"fps": 30.0,
"has_audio": True,
"codec": "h264",
"audio_codec": "aac",
"bitrate": "5M"
}
# Dummy implementation of all basic operations
[docs]
def trim(self, video: Any, start: float, end: Optional[float] = None, **kwargs) -> Any:
logger.info(f"Dummy trim: {start} to {end}")
return video
[docs]
def resize(self, video: Any, width: Optional[int] = None, height: Optional[int] = None,
keep_aspect: bool = True, **kwargs) -> Any:
logger.info(f"Dummy resize: {width}x{height}, keep_aspect={keep_aspect}")
return video
[docs]
def crop(self, video: Any, x: int, y: int, width: int, height: int, **kwargs) -> Any:
logger.info(f"Dummy crop: ({x}, {y}), {width}x{height}")
return video
[docs]
def rotate(self, video: Any, degrees: float, **kwargs) -> Any:
logger.info(f"Dummy rotate: {degrees}°")
return video
[docs]
def grayscale(self, video: Any, **kwargs) -> Any:
logger.info("Dummy grayscale")
return video
[docs]
def blur(self, video: Any, radius: float, **kwargs) -> Any:
logger.info(f"Dummy blur: radius={radius}")
return video
[docs]
def brightness(self, video: Any, factor: float, **kwargs) -> Any:
logger.info(f"Dummy brightness: factor={factor}")
return video
[docs]
def contrast(self, video: Any, factor: float, **kwargs) -> Any:
logger.info(f"Dummy contrast: factor={factor}")
return video
[docs]
def mute(self, video: Any, **kwargs) -> Any:
logger.info("Dummy mute")
return video
[docs]
def volume(self, video: Any, level: float, **kwargs) -> Any:
logger.info(f"Dummy volume: level={level}")
return video
[docs]
def add_audio(self, video: Any, audio_path: Union[str, Path],
start: float = 0, volume: float = 1.0, **kwargs) -> Any:
logger.info(f"Dummy add_audio: {audio_path}, start={start}, volume={volume}")
return video
[docs]
def overlay(self, video: Any, overlay_video: Any, position=(0, 0),
start: float = 0, duration: Optional[float] = None, opacity: float = 1.0, **kwargs) -> Any:
logger.info(f"Dummy overlay: position={position}, start={start}, duration={duration}, opacity={opacity}")
return video
[docs]
def concat(self, videos: List[Any], **kwargs) -> Any:
logger.info(f"Dummy concat {len(videos)} videos")
return videos[0] if videos else None
[docs]
def grid(self, videos: List[Any], rows: int, cols: int, **kwargs) -> Any:
logger.info(f"Dummy grid: {rows}x{cols} with {len(videos)} videos")
return videos[0] if videos else None
[docs]
def add_subtitles(self, video: Any, entries: List,
font: str = "Arial", size: int = 24,
color: Union[str, Tuple[int, int, int]] = "white",
position: Optional[Tuple[int, int]] = None, **kwargs) -> Any:
logger.info(f"Dummy add_subtitles: {len(entries)} entries, font={font}, size={size}")
return video
[docs]
def add_subtitle_text(self, video: Any, entry,
font: str = "Arial", size: int = 24,
color: Union[str, Tuple[int, int, int]] = "white", **kwargs) -> Any:
logger.info(f"Dummy add_subtitle_text: font={font}, size={size}")
return video
[docs]
def get_backend(backend_name: Union[str, Backend] = "auto") -> BackendInterface:
"""
Get a video processing backend.
Args:
backend_name: Backend name or enum ("auto", "ffmpeg", "moviepy", "pyav", "gstreamer", "vlc")
Returns:
Backend instance
Raises:
ValueError: If the backend is not available
"""
# Check if a Backend enum was passed
if isinstance(backend_name, Backend):
backend_enum = backend_name
elif backend_name == "auto":
# Auto-detect the best available backend
backend_enum = _auto_detect_backend()
else:
# Convert string to enum
backend_enum = Backend.from_string(backend_name)
# Import and create the appropriate backend
try:
if backend_enum == Backend.FFMPEG:
# Use the new ffmpeg package
from .ffmpeg import FFmpegBackend
backend_instance = FFmpegBackend()
elif backend_enum == Backend.MOVIEPY:
from .moviepy import MoviePyBackend
backend_instance = MoviePyBackend()
elif backend_enum == Backend.PYAV:
from .pyav import PyAVBackend
backend_instance = PyAVBackend()
elif backend_enum == Backend.GSTREAMER:
from .gstreamer import GStreamerBackend
backend_instance = GStreamerBackend()
elif backend_enum == Backend.VLC:
from .vlc import VLCBackend
backend_instance = VLCBackend()
elif backend_enum == Backend.DUMMY:
backend_instance = DummyBackend()
else:
raise ValueError(f"Unknown backend: {backend_enum}")
except (ImportError, AttributeError, TypeError) as e:
logger.warning(f"Error importing backend {backend_enum.name}: {e}. Falling back to dummy backend.")
backend_instance = DummyBackend()
return backend_instance
def _auto_detect_backend() -> Backend:
"""
Auto-detect the best available backend.
Returns:
Backend enum
Raises:
ValueError: If no backend is available
"""
# Try backends in order of preference
backends = [
(Backend.FFMPEG, lambda: _check_backend_available(Backend.FFMPEG)),
(Backend.PYAV, lambda: _check_backend_available(Backend.PYAV)),
(Backend.MOVIEPY, lambda: _check_backend_available(Backend.MOVIEPY)),
(Backend.GSTREAMER, lambda: _check_backend_available(Backend.GSTREAMER)),
(Backend.VLC, lambda: _check_backend_available(Backend.VLC))
]
for backend_enum, check_func in backends:
try:
if check_func():
logger.info(f"Auto-detected backend: {backend_enum.name}")
return backend_enum
except Exception as e:
logger.debug(f"Error checking {backend_enum.name} availability: {e}")
# If we get here, no backend is available - use dummy backend
logger.warning("No video processing backends available. Using dummy backend instead.")
return Backend.DUMMY
def _check_backend_available(backend_enum: Backend) -> bool:
"""
Check if a backend is available.
Args:
backend_enum: Backend enum
Returns:
True if the backend is available, False otherwise
"""
if backend_enum == Backend.FFMPEG:
try:
# Use the new ffmpeg package
from .ffmpeg import FFmpegBackend
return FFmpegBackend.is_available()
except (ImportError, AttributeError, TypeError):
logger.debug("FFMPEG backend not available")
return False
elif backend_enum == Backend.MOVIEPY:
try:
from .moviepy import MoviePyBackend
return MoviePyBackend.is_available()
except (ImportError, AttributeError, TypeError):
logger.debug("MoviePy backend not available")
return False
elif backend_enum == Backend.PYAV:
try:
from .pyav import PyAVBackend
return PyAVBackend.is_available()
except (ImportError, AttributeError, TypeError):
logger.debug("PyAV backend not available")
return False
elif backend_enum == Backend.GSTREAMER:
try:
from .gstreamer import GStreamerBackend
return GStreamerBackend.is_available()
except (ImportError, AttributeError, TypeError):
logger.debug("GStreamer backend not available")
return False
elif backend_enum == Backend.VLC:
try:
from .vlc import VLCBackend
return VLCBackend.is_available()
except (ImportError, AttributeError, TypeError):
logger.debug("VLC backend not available")
return False
elif backend_enum == Backend.DUMMY:
return True
else:
return False