cjm-media-plugin-demucs 0.0.15

Demucs v4 audio source separation plugin for the cjm-plugin-system that provides vocals extraction for removing background noise and music from speech audio.

cjm-media-plugin-demucs

Install

pip install cjm_media_plugin_demucs

Project Structure

nbs/
├── meta.ipynb   # Metadata introspection for the Demucs plugin used by cjm-ctl to generate the registration manifest.
└── plugin.ipynb # Demucs v4 audio source separation plugin — provides vocals extraction for removing background noise and music from speech audio.

Total: 2 notebooks

Module Dependencies

graph LR
    meta["meta<br/>Metadata"]
    plugin["plugin<br/>Plugin"]

No cross-module dependencies detected.

CLI Reference

No CLI commands found in this project.

Module Overview

Detailed documentation for each module in the project:

Metadata (meta.ipynb)

Metadata introspection for the Demucs plugin used by cjm-ctl to generate the registration manifest.

Import

from cjm_media_plugin_demucs.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."""
    # Fallback base path (current behavior for backward compatibility)
    base_path = os.path.dirname(os.path.dirname(sys.executable))
    
    # Use CJM config if available, else fallback to env-relative paths
    cjm_data_dir = os.environ.get("CJM_DATA_DIR")
    
    # Plugin data directory
    plugin_name = "cjm-media-plugin-demucs"
    if cjm_data_dir
    "Return metadata required to register this plugin with the PluginManager."

Plugin (plugin.ipynb)

Demucs v4 audio source separation plugin — provides vocals extraction for removing background noise and music from speech audio.

Import

from cjm_media_plugin_demucs.plugin import (
    DemucsPluginConfig,
    DemucsProcessingPlugin
)

Functions

@patch
def _apply_config(self:DemucsProcessingPlugin,
                  config: Optional[Any] = None,  # Configuration dict or None for defaults
                 ) -> None
    """
    CR-4: apply config values only. Called by initialize (first-time) and the
    substrate's reconfigure delta path. Model release on a model/device change is
    handled declaratively via RELOAD_TRIGGER -> _release_model (device resolved
    lazily in _load_model).
    """

@patch
def prefetch(self:DemucsProcessingPlugin) -> None
    """
    CR-4 (SG-19): eagerly load the model so the first execute() doesn't pay
    the download/load cost. Idempotent via _load_model's None-guard.
    """

@patch
def on_disable(self:DemucsProcessingPlugin) -> None
    """
    CR-2: release the GPU model when the operator disables the plugin (the
    worker stays alive); lazy reload on the next execute after re-enable.
    """

@patch
def cleanup(self:DemucsProcessingPlugin) -> None
    "Clean up plugin resources."

@patch
def is_available(self:DemucsProcessingPlugin) -> bool:  # Whether the plugin can run
    """Check if the plugin is available on this system."""
    try
    "Check if the plugin is available on this system."

@patch
def _load_model(self:DemucsProcessingPlugin) -> None:
    """Load the Demucs Separator (lazy, cached).

    CR-4: a model/device change releases the separator declaratively via
    RELOAD_TRIGGER -> _release_model, so no manual change-detection is needed here —
    a None separator means a (re)load is required. The heartbeat wraps the WHOLE
    load: Separator() downloads weights via torch.hub on a cold cache (silent to the
    substrate's stall detector), so the heartbeat keeps the (progress, message)
    tuple advancing to avoid a false-positive stall."""
    if self._separator is not None
    """
    Load the Demucs Separator (lazy, cached).
    
    CR-4: a model/device change releases the separator declaratively via
    RELOAD_TRIGGER -> _release_model, so no manual change-detection is needed here —
    a None separator means a (re)load is required. The heartbeat wraps the WHOLE
    load: Separator() downloads weights via torch.hub on a cold cache (silent to the
    substrate's stall detector), so the heartbeat keeps the (progress, message)
    tuple advancing to avoid a false-positive stall.
    """

@patch
def _release_model(self:DemucsProcessingPlugin) -> None
    """
    CR-4: release the Demucs Separator + free CUDA cache. RELOAD_TRIGGER target
    for model/device; on_disable / cleanup delegate here. Idempotent via
    cjm-torch-plugin-utils' release_model (no-op when already released).
    """

@patch
def _store_job(self:DemucsProcessingPlugin,
    """
    Hash input/output files and store a processing job record (upsert by
    action + input_path + config_hash; logs + swallows save failures).
    """

@patch
def _action_get_info(self:DemucsProcessingPlugin, **kwargs) -> Dict[str, Any]
    "Action wrapper -> get_info()."

@patch
def _action_separate_vocals(self:DemucsProcessingPlugin, **kwargs) -> Dict[str, Any]
    "Action wrapper -> _separate_vocals()."

@patch
def _separate_vocals(self:DemucsProcessingPlugin,
                     input_path: str,  # Path to audio file
                     output_dir: Optional[str] = None,  # Output directory (default: content+config cache dir)
                     output_format: Optional[str] = None,  # Output format override
                    ) -> Dict[str, Any]:  # Separation result
    "Extract vocals stem from an audio file."

Classes

@dataclass
class DemucsPluginConfig:
    "Configuration for the Demucs processing plugin."
    
    model: str = field(...)
    device: str = field(...)
    shifts: int = field(...)
    overlap: float = field(...)
    segment: Optional[int] = field(...)
    save_other_stems: bool = field(...)
    output_format: str = field(...)

class DemucsProcessingPlugin:
    def __init__(self):
        """Initialize the plugin."""
        self.logger = logging.getLogger(f"{__name__}.{type(self).__name__}")
        self.config: Optional[DemucsPluginConfig] = None
    "Demucs v4 source separation plugin for vocals extraction."
    
    def __init__(self):
            """Initialize the plugin."""
            self.logger = logging.getLogger(f"{__name__}.{type(self).__name__}")
            self.config: Optional[DemucsPluginConfig] = None
        "Initialize the plugin."
    
    def name(self) -> str:  # Plugin name identifier
            """Get the plugin name."""
            return "cjm-media-plugin-demucs"
        
        @property
        def version(self) -> str:  # Plugin version string
        "Get the plugin name."
    
    def version(self) -> str:  # Plugin version string
            """Get the plugin version."""
            from cjm_media_plugin_demucs import __version__
            return __version__
        
        @property
        def supported_media_types(self) -> List[str]:  # Supported media types
        "Get the plugin version."
    
    def supported_media_types(self) -> List[str]:  # Supported media types
            """Get supported media types."""
            return ["audio"]
        
        # ── Lifecycle ────────────────────────────────────────────────────
        
    
        def initialize(self,
                       config: Optional[Any] = None,  # Configuration dict or None for defaults
                      ) -> None
        "Get supported media types."
    
    def initialize(self,
                       config: Optional[Any] = None,  # Configuration dict or None for defaults
                      ) -> None
        "First-time setup. CR-4: config application factored into _apply_config; the
substrate's reconfigure path fires _release_model on a model/device change then
re-applies config."
    
    def get_config_schema(self) -> Dict[str, Any]:  # JSON Schema for UI forms
            """Return JSON Schema for the plugin configuration."""
            return dataclass_to_jsonschema(DemucsPluginConfig)
        
        def get_current_config(self) -> Dict[str, Any]:  # Current config as dict
        "Return JSON Schema for the plugin configuration."
    
    def get_current_config(self) -> Dict[str, Any]:  # Current config as dict
            """Return the current configuration."""
            return config_to_dict(self.config) if self.config else {}
        
        # ── Model Management ────────────────────────────────────────────
        
        
        
        # ── Job Storage ──────────────────────────────────────────────────
        
        
        # ── Action Dispatch ──────────────────────────────────────────────
        
        def execute(self,
                    action: str = "separate_vocals",  # Action to perform
                    **kwargs
                   ) -> Dict[str, Any]:  # Action result
        "Return the current configuration."
    
    def execute(self,
                    action: str = "separate_vocals",  # 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: str,  # Path to audio file
                    ) -> MediaMetadata:  # File metadata
        "Get basic audio file metadata via ffprobe."