Metadata-Version: 2.4
Name: cjm-media-plugin-ffmpeg
Version: 0.0.22
Summary: FFmpeg-based media processing plugin for the cjm-plugin-system that provides audio extraction, segmentation, format conversion, and segment extraction.
Author-email: "Christian J. Mills" <9126128+cj-mills@users.noreply.github.com>
License: Apache-2.0
Project-URL: Repository, https://github.com/cj-mills/cjm-media-plugin-ffmpeg
Project-URL: Documentation, https://cj-mills.github.io/cjm-media-plugin-ffmpeg
Keywords: nbdev,jupyter,notebook,python
Classifier: Natural Language :: English
Classifier: Intended Audience :: Developers
Classifier: Development Status :: 3 - Alpha
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Requires-Python: >=3.12
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: cjm_media_plugin_system>=0.0.21
Requires-Dist: tqdm
Dynamic: license-file

# cjm-media-plugin-ffmpeg


<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->

## Install

``` bash
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

``` mermaid
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

``` python
from cjm_media_plugin_ffmpeg.utils.availability import (
    FFMPEG_AVAILABLE
)
```

#### Variables

``` python
FFMPEG_AVAILABLE
```

### Audio Codec Map (`codec.ipynb`)

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

#### Import

``` python
from cjm_media_plugin_ffmpeg.utils.codec import (
    get_audio_codec
)
```

#### Functions

``` python
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

``` python
from cjm_media_plugin_ffmpeg.meta import (
    get_plugin_metadata
)
```

#### Functions

``` python
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

``` python
from cjm_media_plugin_ffmpeg.plugin import (
    FFmpegPluginConfig,
    FFmpegProcessingPlugin
)
```

#### Classes

``` python
@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(...)
```

``` python
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

``` python
from cjm_media_plugin_ffmpeg.utils.probe import (
    get_media_duration
)
```

#### Functions

``` python
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

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

#### Functions

``` python
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."
```

``` python
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

``` python
from cjm_media_plugin_ffmpeg.utils.segments import (
    extract_audio_segment
)
```

#### Functions

``` python
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."
```
