cjm-transcription-plugin-qwen3-forced-aligner 0.0.13

A Qwen3-based forced alignment plugin for the cjm-transcription-plugin-system that provides word-level audio-text alignment using Qwen/Qwen3-ForcedAligner-0.6B with progress reporting.

cjm-transcription-plugin-qwen3-forced-aligner

Install

pip install cjm_transcription_plugin_qwen3_forced_aligner

Project Structure

nbs/
├── meta.ipynb   # Metadata introspection for the Qwen3 Forced Aligner plugin used by cjm-ctl to generate the registration manifest.
└── plugin.ipynb # Plugin implementation for Qwen3 word-level forced alignment

Total: 2 notebooks

Module Dependencies

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

    plugin --> meta

1 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 Qwen3 Forced Aligner plugin used by cjm-ctl to generate the registration manifest.

Import

from cjm_transcription_plugin_qwen3_forced_aligner.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-transcription-plugin-qwen3-forced-aligner"
    if cjm_data_dir
    "Return metadata required to register this plugin with the PluginManager."

Qwen3 Forced Aligner Plugin (plugin.ipynb)

Plugin implementation for Qwen3 word-level forced alignment

Import

from cjm_transcription_plugin_qwen3_forced_aligner.plugin import (
    Qwen3ForcedAlignerConfig,
    Qwen3ForcedAlignerPlugin
)

Functions

@patch
def is_available(self:Qwen3ForcedAlignerPlugin) -> bool:  # True if qwen_asr is importable
    "Check if the Qwen3 forced aligner is available."

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

@patch
def _load_model(self:Qwen3ForcedAlignerPlugin):
    """Load model on first use. Heartbeat-wrapped HF Hub download + build."""
    if self._model is not None
    "Load model on first use. Heartbeat-wrapped HF Hub download + build."

@patch
def _release_model(self:Qwen3ForcedAlignerPlugin)
    """
    CR-4: release the model + free CUDA cache (cjm-torch-plugin-utils).
    RELOAD_TRIGGER target for model_id/device/dtype/attn_implementation; on_disable /
    cleanup delegate here. Idempotent (release_model no-ops when already None).
    """

@patch
def prefetch(self:Qwen3ForcedAlignerPlugin) -> 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:Qwen3ForcedAlignerPlugin) -> 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:Qwen3ForcedAlignerPlugin) -> None
    "Clean up resources."

Classes

@dataclass
class Qwen3ForcedAlignerConfig(HFCacheConfig):
    """
    Configuration for the Qwen3 Forced Aligner plugin.
    
    Composes HFCacheConfig (cache_dir / revision / local_files_only / trust_remote_code,
    each RELOAD_TRIGGER-tagged) so HF Hub download behaviour is operator-controllable.
    """
    
    model_id: str = field(...)
    device: str = field(...)
    dtype: str = field(...)
    attn_implementation: str = field(...)
    language: str = field(...)

class Qwen3ForcedAlignerPlugin:
    def __init__(self):
        """Initialize the plugin with default state."""
        self.logger = logging.getLogger(f"{__name__}.{type(self).__name__}")
        self.config: Qwen3ForcedAlignerConfig = None
    "Qwen3 Forced Alignment plugin for word-level audio-text alignment."
    
    def __init__(self):
            """Initialize the plugin with default state."""
            self.logger = logging.getLogger(f"{__name__}.{type(self).__name__}")
            self.config: Qwen3ForcedAlignerConfig = None
        "Initialize the plugin with default state."
    
    def name(self) -> str:  # Plugin name identifier
            return "cjm-transcription-plugin-qwen3-forced-aligner"
    
        @property
        def version(self) -> str:  # Plugin version string
    
    def version(self) -> str:  # Plugin version string
            from cjm_transcription_plugin_qwen3_forced_aligner import __version__
            return __version__
    
        @property
        def supported_formats(self) -> List[str]:  # Supported audio file formats
    
    def supported_formats(self) -> List[str]:  # Supported audio file formats
            return ["wav", "mp3", "flac", "ogg", "m4a"]
    
    
        def get_config_schema(self) -> Dict[str, Any]:  # JSON Schema for configuration
    
    def get_config_schema(self) -> Dict[str, Any]:  # JSON Schema for configuration
            """Return JSON Schema for UI generation."""
            return dataclass_to_jsonschema(Qwen3ForcedAlignerConfig)
    
        def get_current_config(self) -> Dict[str, Any]:  # Current configuration as dictionary
        "Return JSON Schema for UI generation."
    
    def get_current_config(self) -> Dict[str, Any]:  # Current configuration as dictionary
            """Return current configuration state."""
            if not self.config
        "Return current configuration state."
    
    def initialize(
            self,
            config: Optional[Any] = None  # Configuration dataclass, dict, or None
        ) -> None
        "First-time setup. CR-4: the manual diff-and-reload is replaced by declarative
RELOAD_TRIGGER metadata; the substrate's reconfigure path fires _release_model
then re-applies config via _apply_config."
    
    def execute(
            self,
            audio: Union[str, Path],  # Audio data or file path
            text: str,                            # Transcript text to align against
            **kwargs
        ) -> ForcedAlignResult:  # Word-level alignment result
        "Perform forced alignment of text against audio."