cjm-source-separation-adapter-interface

Install

pip install cjm_source_separation_adapter_interface

Project Structure

nbs/
├── adapter.ipynb # The typed source-separation (audio-preprocessing) task contract — the `SourceSeparationAdapter` ABC + the `SourceSeparationToolProtocol` structural contract (capability-unit Option C, pass-2 Thread 3).
├── generic.ipynb # The generic (tool-agnostic) source-separation adapter — cache-check, invoke the bound tool's pure-compute `separate_vocals` (writing the artifact to an adapter-chosen dir), persist the artifact pointer. Reused across every tool capability satisfying `SourceSeparationToolProtocol`, exactly as `GenericVADAdapter` is reused across VAD tools.
└── storage.ipynb # Standardized SQLite storage for source-separation results (the produced-artifact pointer) with content hashing.

Total: 3 notebooks

Module Dependencies

graph LR
    adapter["adapter<br/>Source Separation Adapter"]
    generic["generic<br/>Generic Source Separation Adapter"]
    storage["storage<br/>Source Separation Storage"]

    generic --> adapter
    generic --> storage

2 cross-module dependencies detected

CLI Reference

No CLI commands found in this project.

Module Overview

Detailed documentation for each module in the project:

Source Separation Adapter (adapter.ipynb)

The typed source-separation (audio-preprocessing) task contract — the SourceSeparationAdapter ABC + the SourceSeparationToolProtocol structural contract (capability-unit Option C, pass-2 Thread 3).

Import

from cjm_source_separation_adapter_interface.adapter import (
    SourceSeparationToolProtocol,
    SourceSeparationAdapter
)

Classes

@runtime_checkable
class SourceSeparationToolProtocol(Protocol):
    """
    Structural contract for source-separation tool capabilities
    (born-final at stage 8 — derived from the native tool surface).
    
    Pure compute: `separate_vocals` reads the input audio, runs separation, and
    WRITES the isolated-audio artifact into `output_dir` (the adapter-chosen,
    content+config-addressed location), returning a `SourceSeparationResult`
    whose `output_path` points at it. `get_current_config` supplies the
    effective config the generic adapter hashes for its cache key.
    
    The `output_dir` parameter is the artifact-producer's twist on the
    native-surface seam: persistence LOCATION is the adapter's concern (the
    `db_path`-off-the-tool rule), so the adapter passes it in rather than the
    tool inventing a path. Encoding the audio stays tool-side compute.
    """
    
    def separate_vocals(self, audio: Union[str, Path], output_dir: str, **kwargs) -> SourceSeparationResult: ...
        def get_current_config(self) -> Dict[str, Any]: ...
    
    def get_current_config(self) -> Dict[str, Any]: ...

class SourceSeparationAdapter:
    def __init__(
        self,
        tool: SourceSeparationToolProtocol,  # The bound tool capability instance (worker-side binding)
    )
    """
    Typed source-separation task adapter: input audio in, a vocals-isolated
    `SourceSeparationResult` (output_path + metadata) out.
    
    Input contract (DELIBERATELY different from VAD/transcription): the tool
    receives FULL-BAND segment audio — source separation works best on
    full-fidelity input, and the model-ready (mono 8k/16k) convert runs
    DOWNSTREAM of separation in the transcription pipeline (vocals -> convert ->
    transcribe / VAD / FA). So this adapter does NOT assume model-ready input.
    
    Native-surface model (stage 8 / PILLAR 1c): the TOOL is pure compute; the
    ADAPTER owns the cache + persistence bookends (see
    `GenericSourceSeparationAdapter`) + the per-call `force` control. The
    produced audio ARTIFACT lives under the substrate-injected
    `CAPABILITY_DATA_DIR`; the cache row maps (input, config) -> output_path. The
    adapter chooses the output location and passes it to the tool; `db_path` is
    not on the tool protocol.
    
    Implementations run in-worker beside their tool capability and are
    constructed with the bound tool instance: `AdapterClass(tool)` (mirrors
    `GraphStorageAdapter`). The result DTO is wire-registered
    ("source_separation.result"): returned values cross the worker boundary
    typed.
    """
    
    def __init__(
            self,
            tool: SourceSeparationToolProtocol,  # The bound tool capability instance (worker-side binding)
        )
    
    def separate_vocals(
            self,
            audio: Union[str, Path],  # Path to the input audio to separate (full-band; NOT model-ready)
            **kwargs,                 # Provenance + tool options
        ) -> SourceSeparationResult:  # Vocals-isolated artifact (output_path + metadata)
        "Separate vocals from the input audio, returning the produced artifact."

Generic Source Separation Adapter (generic.ipynb)

The generic (tool-agnostic) source-separation adapter — cache-check, invoke the bound tool’s pure-compute separate_vocals (writing the artifact to an adapter-chosen dir), persist the artifact pointer. Reused across every tool capability satisfying SourceSeparationToolProtocol, exactly as GenericVADAdapter is reused across VAD tools.

Import

from cjm_source_separation_adapter_interface.generic import (
    GenericSourceSeparationAdapter
)

Classes

class GenericSourceSeparationAdapter(SourceSeparationAdapter):
    """
    Generic source-separation adapter: cache-check -> pure-compute tool
    (writing the artifact to an adapter-chosen dir) -> persist the artifact pointer.
    
    Works against ANY tool satisfying `SourceSeparationToolProtocol`. The bookends:
    
      1. cache check (input_path + input_hash + config_hash) BEFORE invoking the
         tool, AND confirm the produced artifact still exists on disk, so a hit
         never loads the model nor re-separates;
      2. on a miss / forced call / vanished artifact, compute the
         content+config-addressed `output_dir` under `CAPABILITY_DATA_DIR` and pass
         it to the tool's pure-compute `separate_vocals`, which writes the vocals
         stem there;
      3. `save_with_logging` the (input, config) -> output_path mapping.
    
    `config_hash` reuses `hash_dict_canonical(get_current_config())` (the SAME
    canonical hash the fused-era plugin used). `force` rides
    `CallEnvelope.control` (not a task kwarg, keeping `separate_vocals(audio)`
    pure). Storage lives at `<CAPABILITY_DATA_DIR>/source_separations.db` and the
    produced artifacts live under the same `CAPABILITY_DATA_DIR` (the adapter picks
    the dir via the substrate's `cache_dir_for_config`), so the tool neither
    hard-codes a path nor invents the cache layout. The artifact-existence
    re-check on a hit is the file analog of a cache row — the DB could outlive a
    cleaned-up file.
    """
    
    def separate_vocals(
            self,
            audio: Union[str, Path],  # Path to the input audio to separate (full-band; NOT model-ready)
            **kwargs,                 # Provenance + tool options
        ) -> SourceSeparationResult:  # Vocals-isolated artifact (output_path + metadata)
        "Cache-check (+ artifact existence), invoke the bound tool's pure-compute
`separate_vocals` into an adapter-chosen dir, persist the artifact pointer."

Source Separation Storage (storage.ipynb)

Standardized SQLite storage for source-separation results (the produced-artifact pointer) with content hashing.

Import

from cjm_source_separation_adapter_interface.storage import (
    SourceSeparationRow,
    SourceSeparationStorage
)

Classes

@dataclass
class SourceSeparationRow:
    "A single row from the source_separation_results table."
    
    input_path: str  # Path to the input audio that was separated
    input_hash: str  # Hash of the input file in "algo:hexdigest" format
    config_hash: str  # Hash of the separation config used
    output_path: str  # Path to the produced isolated-audio artifact (e.g. vocals stem)
    output_hash: Optional[str]  # Hash of the produced artifact
    metadata: Optional[Dict[str, Any]]  # Separation metadata (duration, sample_rate, model, stems, ...)
    created_at: Optional[float]  # Unix timestamp

class SourceSeparationStorage:
    def __init__(
        self,
        db_path: str  # Absolute path to the SQLite database file
    )
    "Standardized SQLite storage for source-separation results (artifact pointers)."
    
    def __init__(
            self,
            db_path: str  # Absolute path to the SQLite database file
        )
        "Initialize storage and create table if needed."
    
    def save(
            self,
            input_path: str,    # Path to the input audio that was separated
            input_hash: str,    # Hash of the input file in "algo:hexdigest" format
            config_hash: str,   # Hash of the separation config
            output_path: str,   # Path to the produced isolated-audio artifact
            output_hash: Optional[str] = None,        # Hash of the produced artifact
            metadata: Optional[Dict[str, Any]] = None  # Separation metadata
        ) -> None
        "Save or replace a source-separation result (upsert by input_path + config_hash)."
    
    def save_with_logging(
            self,
            *,
            input_path: str,    # Path to the input audio that was separated
            input_hash: str,    # Hash of the input file in "algo:hexdigest" format
            config_hash: str,   # Hash of the separation config
            output_path: str,   # Path to the produced isolated-audio artifact
            output_hash: Optional[str] = None,         # Hash of the produced artifact
            metadata: Optional[Dict[str, Any]] = None,  # Separation metadata
            logger: Optional[logging.Logger] = None      # Optional logger for success/failure messages
        ) -> bool:  # True if saved; False if the save failed (error logged, not raised)
        "Save a result, logging success/failure. Failures are logged and swallowed (returns False).

CR-14 follow-up: records a RESULT_SAVED account either way (ok flag +
input/output/config references — the journal never carries content) so
saves AND swallowed save-failures become auditable journal rows."
    
    def get_cached(
            self,
            input_path: str,   # Path to the input audio
            input_hash: str,   # Content hash of the input (cache miss if the input changed)
            config_hash: str   # Config hash to match
        ) -> Optional[SourceSeparationRow]:  # Cached row or None
        "Retrieve a content-correct cached source-separation result.

Matches on input_path + input_hash + config_hash, so a changed input
(new input_hash) misses the cache even though a stale row may still exist
at the same (input_path, config_hash) — the next save() replaces it. The
CALLER must still confirm the artifact at `output_path` exists before
serving it (the file lives outside the DB).

CR-14 follow-up: a hit records a CACHE_HIT account (the cache-serving
decision is an account-of-action)."
    
    def list_jobs(
            self,
            limit: int = 100  # Maximum number of rows to return
        ) -> List[SourceSeparationRow]:  # List of source-separation rows
        "List source-separation results ordered by creation time (newest first)."
    
    def verify_input(
            self,
            input_path: str,  # Path to the input audio
            config_hash: str  # Config hash to look up
        ) -> Optional[bool]:  # True if input matches, False if changed, None if not found
        "Verify the input file still matches the hash stored for (input_path, config_hash)."