cjm-capability-primitives

Install

pip install cjm_capability_primitives

Project Structure

nbs/
├── forced_alignment.ipynb  # Standardized word-level forced-alignment DTOs — the data noun forced-alignment tool capabilities emit and task adapters / workflow cores consume, wire-registered so results cross the worker boundary typed.
├── media_processing.ipynb  # Standardized result DTOs for the media-processing task — the data nouns media-processing tool capabilities (ffmpeg today) emit and the multi-method task adapter / workflow cores consume, wire-registered so results cross the worker boundary typed.
├── source_separation.ipynb # Standardized result DTO for the source-separation (audio-preprocessing) task — the data noun source-separation tool capabilities emit and task adapters / workflow cores consume, wire-registered so results cross the worker boundary typed.
├── transcription.ipynb     # Standardized result DTO for the transcription task — the data noun tool capabilities emit and task adapters / workflow cores consume, wire-registered so results cross the worker boundary typed.
└── vad.ipynb               # Standardized result DTO for the voice-activity-detection task — the data noun VAD tool capabilities emit and task adapters / workflow cores consume, wire-registered so results cross the worker boundary typed.

Total: 5 notebooks

Module Dependencies

graph LR
    forced_alignment["forced_alignment<br/>Forced Alignment Result"]
    media_processing["media_processing<br/>Media Processing Results"]
    source_separation["source_separation<br/>Source Separation Result"]
    transcription["transcription<br/>Transcription Result"]
    vad["vad<br/>VAD Result"]

No cross-module dependencies detected.

CLI Reference

No CLI commands found in this project.

Module Overview

Detailed documentation for each module in the project:

Forced Alignment Result (forced_alignment.ipynb)

Standardized word-level forced-alignment DTOs — the data noun forced-alignment tool capabilities emit and task adapters / workflow cores consume, wire-registered so results cross the worker boundary typed.

Import

from cjm_capability_primitives.forced_alignment import (
    ForcedAlignItem,
    ForcedAlignResult
)

Classes

@dataclass
class ForcedAlignItem:
    "A single word-level alignment result."
    
    text: str  # The aligned word (punctuation typically stripped by model)
    start_time: float  # Start time in seconds
    end_time: float  # End time in seconds

@dataclass
class ForcedAlignResult:
    "Standardized output for all forced alignment capabilities."
    
    items: List[ForcedAlignItem]  # Word-level alignments
    metadata: Dict[str, Any] = field(...)  # Capability-specific metadata
    
    def from_dict(
        "Reconstruct from a wire payload, re-typing nested items.

`items` holds typed `ForcedAlignItem` objects, so the substrate's typed
wire envelope (stage 2) reconstructs them host-side here rather than
leaving bare dicts (which would break attribute access like `it.text`)."

Media Processing Results (media_processing.ipynb)

Standardized result DTOs for the media-processing task — the data nouns media-processing tool capabilities (ffmpeg today) emit and the multi-method task adapter / workflow cores consume, wire-registered so results cross the worker boundary typed.

Import

from cjm_capability_primitives.media_processing import (
    MediaSegment,
    MediaArtifactResult,
    MediaSegmentationResult,
    MediaMetadata
)

Classes

@dataclass
class MediaSegment:
    """
    One produced segment file from a `segment_audio` batch cut.
    
    The per-segment entry the fused-era ffmpeg `segment_audio` returned as a
    dict, now a typed noun (the dead `job_id` dropped — born-final; the adapter
    owns persistence). Workflow cores read `index`/`output_path`/`start`/`end`
    to build the per-segment composition.
    """
    
    index: int  # 0-based position of this segment within the batch
    output_path: str  # Path to the produced segment file the tool wrote
    start: float  # Segment start time in the source (seconds)
    end: float  # Segment end time in the source (seconds)
    duration: float  # end - start (seconds)
    
    def to_dict(self) -> Dict[str, Any]:  # Serialized representation
        "Convert to dictionary for JSON serialization."

@dataclass
class MediaArtifactResult:
    """
    A single produced audio artifact (the `convert` / `extract_audio` output).
    
    The artifact-producing shape (cf. `SourceSeparationResult`): `output_path`
    is the file the tool wrote to the adapter-chosen location; `metadata`
    carries the stats the fused-era return dict / row held (codec, duration,
    stream_copy, the effective convert parameters, ...). Flat fields (str +
    dict), so the default wire reconstruction suffices — no custom from_dict.
    """
    
    output_path: str  # Path to the produced audio file
    metadata: Dict[str, Any] = field(...)  # Stats (codec, duration, parameters, ...)

@dataclass
class MediaSegmentationResult:
    """
    A BATCH of produced segment files (the `segment_audio` output).
    
    Holds typed `MediaSegment`s plus the batch metadata the fused-era return
    dict carried (`input_path`, `segment_count`, `total_duration`, `batch_key`
    — the label linking the cut files in the run manifest). Because `segments`
    holds typed objects, a custom `from_dict` re-types them on wire-decode (the
    auto flat reconstruct would leave bare dicts, breaking `seg.output_path`
    access) — the `VADResult` precedent.
    """
    
    segments: List[MediaSegment]  # The produced segment files, ordered by index
    input_path: str = ''  # The source audio that was cut
    segment_count: int = 0  # Number of segments produced
    total_duration: float = 0.0  # Sum of segment durations (seconds)
    batch_key: str = ''  # Label linking this batch's cut files (run-manifest field)
    
    def from_dict(
        "Reconstruct from a wire payload, re-typing nested MediaSegments."

@dataclass
class MediaMetadata:
    """
    Probed metadata for a media file (the `get_info` result) — inline data, no artifact.
    
    Relocated to `cjm-capability-primitives` from the dissolving
    `cjm-media-plugin-system.core` (the `TranscriptionResult`/`ForcedAlignResult`
    relocation precedent). `get_info` is the media-processing task's UNCACHED
    probe op, so this is a read result, not a produced-artifact pointer. The
    stream lists are plain dicts, so the default wire reconstruction suffices.
    """
    
    path: str  # File path probed
    duration: float  # Duration in seconds
    format: str  # Container format (e.g. 'mp4', 'mkv')
    size_bytes: int  # File size in bytes
    video_streams: List[Dict[str, Any]] = field(...)  # Per-video-stream info (codec, width, height, fps)
    audio_streams: List[Dict[str, Any]] = field(...)  # Per-audio-stream info (codec, sample_rate, channels, duration)
    
    def to_dict(self) -> Dict[str, Any]:  # Serialized representation
        "Convert to dictionary for JSON serialization."

Source Separation Result (source_separation.ipynb)

Standardized result DTO for the source-separation (audio-preprocessing) task — the data noun source-separation tool capabilities emit and task adapters / workflow cores consume, wire-registered so results cross the worker boundary typed.

Import

from cjm_capability_primitives.source_separation import (
    SourceSeparationResult
)

Classes

@dataclass
class SourceSeparationResult:
    """
    Standardized output for source-separation (audio-preprocessing) capabilities.
    
    The payload is an AUDIO ARTIFACT, not inline data: `output_path` is the
    produced isolated-audio file (e.g. the vocals stem) the tool wrote to the
    location the adapter chose. `metadata` carries the stats the fused-era
    return dict held (duration, sample_rate, model, stems_available, and any
    extra-stem paths when the tool was asked to keep them).
    """
    
    output_path: str  # Path to the produced isolated-audio artifact (e.g. vocals stem)
    metadata: Dict[str, Any] = field(...)  # Stats (duration, sample_rate, model, stems_available, other_stems, ...)

Transcription Result (transcription.ipynb)

Standardized result DTO for the transcription task — the data noun tool capabilities emit and task adapters / workflow cores consume, wire-registered so results cross the worker boundary typed.

Import

from cjm_capability_primitives.transcription import (
    TranscriptionResult
)

Classes

@dataclass
class TranscriptionResult:
    "Standardized output for all transcription plugins."
    
    text: str  # The transcribed text
    confidence: Optional[float]  # Overall confidence (0.0 to 1.0)
    segments: Optional[List[Dict[str, Any]]]  # Timestamped segments
    metadata: Dict[str, Any] = field(...)  # Additional metadata

VAD Result (vad.ipynb)

Standardized result DTO for the voice-activity-detection task — the data noun VAD tool capabilities emit and task adapters / workflow cores consume, wire-registered so results cross the worker boundary typed.

Import

from cjm_capability_primitives.vad import (
    TimeRange,
    VADResult
)

Classes

@dataclass
class TimeRange:
    "A temporal segment within an audio source (the VAD speech/silence span)."
    
    start: float  # Start time in seconds
    end: float  # End time in seconds
    label: str = 'speech'  # Segment type (e.g. 'speech')
    confidence: Optional[float]  # Detection confidence (0.0 to 1.0)
    payload: Dict[str, Any] = field(...)  # Extra data (reserved)
    
    def to_dict(self) -> Dict[str, Any]:  # Serialized representation
        "Convert to dictionary for JSON serialization."

@dataclass
class VADResult:
    "Standardized output for voice-activity-detection capabilities."
    
    ranges: List[TimeRange]  # Detected speech segments, sorted by start
    metadata: Dict[str, Any] = field(...)  # Global VAD stats (duration, sample_rate, total_speech, ...)
    
    def from_dict(
        "Reconstruct from a wire payload, re-typing nested TimeRanges.

`ranges` holds typed `TimeRange` objects, so the substrate's typed wire
envelope (stage 2) reconstructs them host-side here rather than leaving
bare dicts (which would break attribute access like `r.start`)."