cjm-media-plugin-ffmpeg 0.0.23

FFmpeg-based media processing plugin for the cjm-plugin-system that provides audio extraction, segmentation, format conversion, and segment extraction.

cjm-media-plugin-ffmpeg

Install

pip install cjm_media_plugin_ffmpeg

Project Structure

nbs/
├── utils/ (5)
│   ├── availability.ipynb  # Detect whether the `ffmpeg` binary is installed on this system.
│   ├── codec.ipynb         # Map audio container formats to the ffmpeg codec used to encode them.
│   ├── probe.ipynb         # Probe media files for metadata (duration, ...) via ffprobe.
│   ├── progress.ipynb      # Run ffmpeg subprocess commands with a progress bar and optional callback.
│   └── segments.ipynb      # Extract temporal segments from audio files via ffmpeg stream-copy.
├── meta.ipynb    # Metadata introspection for the FFmpeg media processing plugin used by `cjm-ctl` to generate the registration manifest.
└── plugin.ipynb  # FFmpeg-based media processing plugin implementing the `MediaProcessingPlugin` interface.

Total: 7 notebooks across 1 directory

Module Dependencies

graph LR
    meta["meta<br/>Metadata"]
    plugin["plugin<br/>FFmpeg Processing Plugin"]
    utils_availability["utils.availability<br/>FFmpeg Availability"]
    utils_codec["utils.codec<br/>Audio Codec Map"]
    utils_probe["utils.probe<br/>Media Probing"]
    utils_progress["utils.progress<br/>FFmpeg Execution + Progress"]
    utils_segments["utils.segments<br/>Audio Segment Extraction"]

    plugin --> utils_probe
    plugin --> utils_segments
    plugin --> utils_availability
    plugin --> utils_progress
    plugin --> meta
    plugin --> utils_codec
    utils_segments --> utils_progress

7 cross-module dependencies detected

CLI Reference

No CLI commands found in this project.

Module Overview

Detailed documentation for each module in the project:

FFmpeg Availability (availability.ipynb)

Detect whether the ffmpeg binary is installed on this system.

Import

from cjm_media_plugin_ffmpeg.utils.availability import (
    FFMPEG_AVAILABLE
)

Variables

FFMPEG_AVAILABLE

Audio Codec Map (codec.ipynb)

Map audio container formats to the ffmpeg codec used to encode them.

Import

from cjm_media_plugin_ffmpeg.utils.codec import (
    get_audio_codec
)

Functions

def get_audio_codec(audio_format: str  # The desired audio format (e.g. 'mp3', 'wav')
                   ) -> str:  # The ffmpeg audio codec name ('copy' if unknown)
    "Map an audio container format to the appropriate ffmpeg codec."

Metadata (meta.ipynb)

Metadata introspection for the FFmpeg media processing plugin used by cjm-ctl to generate the registration manifest.

Import

from cjm_media_plugin_ffmpeg.meta import (
    get_plugin_metadata
)

Functions

def get_plugin_metadata() -> Dict[str, Any]:  # Plugin metadata for manifest generation
    """Return metadata required to register this plugin with the PluginManager."""
    cjm_plugin_data_dir = os.environ.get("CJM_PLUGIN_DATA_DIR")
    plugin_name = "cjm-media-plugin-ffmpeg"

    if cjm_plugin_data_dir
    "Return metadata required to register this plugin with the PluginManager."

FFmpeg Processing Plugin (plugin.ipynb)

FFmpeg-based media processing plugin implementing the MediaProcessingPlugin interface.

Import

from cjm_media_plugin_ffmpeg.plugin import (
    FFmpegPluginConfig,
    FFmpegProcessingPlugin
)

Classes

@dataclass
class FFmpegPluginConfig:
    "Configuration for the FFmpeg processing plugin."
    
    output_dir: Optional[str] = field(...)
    default_audio_format: str = field(...)
    default_audio_bitrate: str = field(...)
    prefer_stream_copy: bool = field(...)
    resampler: str = field(...)

class FFmpegProcessingPlugin:
    def __init__(self):
        """Initialize the FFmpeg processing plugin."""
        self.logger = logging.getLogger(f"{__name__}.{type(self).__name__}")
        self.config: Optional[FFmpegPluginConfig] = None
    "FFmpeg-based media processing plugin."
    
    def __init__(self):
            """Initialize the FFmpeg processing plugin."""
            self.logger = logging.getLogger(f"{__name__}.{type(self).__name__}")
            self.config: Optional[FFmpegPluginConfig] = None
        "Initialize the FFmpeg processing plugin."
    
    def name(self) -> str:  # Plugin identifier
            return get_plugin_metadata()["name"]
    
        @property
        def version(self) -> str:  # Plugin version
    
    def version(self) -> str:  # Plugin version
            return get_plugin_metadata()["version"]
    
        @property
        def supported_media_types(self) -> List[str]:  # Supported input types
    
    def supported_media_types(self) -> List[str]:  # Supported input types
            return ["audio", "video"]
    
        def initialize(self, config: Optional[Any] = None) -> None
    
    def initialize(self, config: Optional[Any] = None) -> None:
            """Initialize plugin with configuration."""
            self.config = dict_to_config(FFmpegPluginConfig, config or {})
            meta = get_plugin_metadata()
            db_path = meta["db_path"]
            self._data_dir = os.path.dirname(db_path)
            self.storage = MediaProcessingStorage(db_path)
            self.logger.info(f"Initialized FFmpeg plugin (format={self.config.default_audio_format})")
    
        def get_config_schema(self) -> Dict[str, Any]:  # JSON Schema for UI form generation
        "Initialize plugin with configuration."
    
    def get_config_schema(self) -> Dict[str, Any]:  # JSON Schema for UI form generation
            """Return the JSON Schema for plugin configuration."""
            return dataclass_to_jsonschema(FFmpegPluginConfig)
    
        def get_current_config(self) -> Dict[str, Any]:  # Current config as dict
        "Return the JSON Schema for plugin configuration."
    
    def get_current_config(self) -> Dict[str, Any]:  # Current config as dict
            """Return the current configuration as a dictionary."""
            return config_to_dict(self.config) if self.config else {}
    
        def is_available(self) -> bool:  # Whether ffmpeg is installed
        "Return the current configuration as a dictionary."
    
    def is_available(self) -> bool:  # Whether ffmpeg is installed
            """Check if ffmpeg is available on this system."""
            return FFMPEG_AVAILABLE
    
        # ------------------------------------------------------------------
        # Helpers
        # ------------------------------------------------------------------
    
        def _get_output_dir(self,
                            output_dir: Optional[str] = None,  # Explicit output dir override
                            subdirectory: Optional[str] = None,  # Subdirectory within output dir
                           ) -> str:  # Resolved output directory path
        "Check if ffmpeg is available on this system."
    
    def execute(self,
                    action: str = "get_info",  # Action to perform
                    **kwargs
                   ) -> Dict[str, Any]:  # Action result
        "Dispatch to the `@plugin_action`-tagged handler for `action` (SG-44)."
    
    def get_info(self,
                     file_path: Union[str, Path],  # Path to media file
                    ) -> MediaMetadata:  # Probed metadata
        "Get metadata for a media file via ffprobe."
    
    def convert(self,
                    input_path: Union[str, Path],  # Source file path
                    output_format: str,  # Target format (e.g. 'mp3', 'wav')
                    **kwargs
                   ) -> str:  # Output file path
        "Convert media to a different format."
    
    def extract_segment(self,
                            input_path: Union[str, Path],  # Source audio file
                            start: float,  # Start time in seconds
                            end: float,  # End time in seconds
                            output_path: Optional[str] = None,  # Custom output path
                           ) -> str:  # Output file path
        "Extract a temporal segment from a media file."

Media Probing (probe.ipynb)

Probe media files for metadata (duration, …) via ffprobe.

Import

from cjm_media_plugin_ffmpeg.utils.probe import (
    get_media_duration
)

Functions

def get_media_duration(file_path: Path  # Path to the media file
                      ) -> Optional[float]:  # Duration in seconds, or None if it cannot be determined
    "Get the duration of a media file (seconds) via ffprobe."

FFmpeg Execution + Progress (progress.ipynb)

Run ffmpeg subprocess commands with a progress bar and optional callback.

Import

from cjm_media_plugin_ffmpeg.utils.progress import (
    parse_progress_line,
    run_ffmpeg_with_progress
)

Functions

def parse_progress_line(line: str  # A line of stderr output from ffmpeg
                        ) -> Optional[float]:  # Current time in seconds, or None if the line has no progress info
    "Parse a progress line from ffmpeg stderr output."

def run_ffmpeg_with_progress(
    cmd: List[str],  # The ffmpeg command and arguments
    total_duration: Optional[float] = None,  # Total duration in seconds for a determinate bar, else indeterminate
    description: str = "Processing",  # Description text for the progress bar
    verbose: bool = False,  # If True, prints detailed ffmpeg output
    progress_callback: Optional[Callable[[float], None]] = None  # Optional callback receiving current progress in seconds
) -> None:  # Raises FileNotFoundError or subprocess.CalledProcessError on failure
    "Run an ffmpeg command with a progress bar."

Audio Segment Extraction (segments.ipynb)

Extract temporal segments from audio files via ffmpeg stream-copy.

Import

from cjm_media_plugin_ffmpeg.utils.segments import (
    extract_audio_segment
)

Functions

def extract_audio_segment(input_path: Path,  # Path to the input audio file
                          output_path: Path,  # Path where the extracted segment is saved
                          start_time: str,  # Start time as "HH:MM:SS" or seconds
                          duration: str,  # Duration as "HH:MM:SS" or seconds
                          verbose: bool = False,  # If True, shows verbose ffmpeg output
                          pbar: bool = False,  # If True, shows a progress bar
                          copy_codec: bool = True,  # Stream-copy without re-encoding (fast)
                        ) -> None:  # Raises subprocess.CalledProcessError if extraction fails
    "Extract a temporal segment from an audio file."