cjm-transcript-graph-schema 0.0.5

Audio-transcript layer schema for context graphs: Source, AudioSegment, Transcript, and Segment nodes with deterministic identity tuples and graph-node mapping, shared by the transcription, decomposition, and correction workflow cores.

cjm-transcript-graph-schema

Install

pip install cjm_transcript_graph_schema

Project Structure

nbs/
└── schema.ipynb # The audio-transcript layer schema (where-graph-begins locked layer schema): `Source -> AudioSegment(coarse, owns the model-input WAV) -> Transcript(per-transcriber variants)` emitted by transcription, extended with the fine `Segment` spine by decomposition. Deterministic identity tuples per the stage-5 ratified rule; `Document` from the pre-CR-18 era dissolves into `Source`.

Total: 1 notebook

Module Dependencies

graph LR
    schema["schema<br/>schema"]

No cross-module dependencies detected.

CLI Reference

No CLI commands found in this project.

Module Overview

Detailed documentation for each module in the project:

schema (schema.ipynb)

The audio-transcript layer schema (where-graph-begins locked layer schema): Source -> AudioSegment(coarse, owns the model-input WAV) -> Transcript(per-transcriber variants) emitted by transcription, extended with the fine Segment spine by decomposition. Deterministic identity tuples per the stage-5 ratified rule; Document from the pre-CR-18 era dissolves into Source.

Import

from cjm_transcript_graph_schema.schema import (
    TranscriptGraphLabels,
    source_node_id,
    audio_segment_node_id,
    transcript_node_id,
    segment_node_id,
    SourceNode,
    AudioSegmentNode,
    TranscriptNode,
    TranscriptSliceRef,
    SegmentNode
)

Functions

def source_node_id(
    content_hash: str,  # Content hash of the source media file ("algo:hexdigest")
) -> str:  # Deterministic Source node id
    "Source identity = the ingested file's content hash (Thread-1 ingested-root identity)."

def audio_segment_node_id(
    source_id: str,  # Owning Source node id
    start: float,    # Boundary start (source-coordinate seconds)
    end: float,      # Boundary end (source-coordinate seconds)
) -> str:  # Deterministic AudioSegment node id
    """
    AudioSegment identity = (source, boundary range).
    
    Conversion-config-independent: the model-input WAV is payload/provenance,
    not identity — the boundary computation is pure and deterministic, so
    re-derivation reproduces the id.
    """

def transcript_node_id(
    audio_segment_id: str,  # Owning AudioSegment node id
    transcriber: str,       # Transcriber capability name (e.g. "cjm-transcription-plugin-whisper")
    config_hash: str,       # Transcriber config hash
) -> str:  # Deterministic Transcript node id
    """
    Transcript identity = (audio segment, transcriber, config) — MIRRORS the
    capability cache key UNIQUE(audio_path, config_hash), so the graph node is
    the durable face of the cached row (the structural E13 fix).
    """

def segment_node_id(
    audio_segment_id: str,  # Owning AudioSegment node id
    vad_config_hash: str,   # VAD capability config hash (skeleton identity input)
    chunk_start: float,     # VAD chunk start (chunk-local seconds within the AudioSegment)
    chunk_end: float,       # VAD chunk end (chunk-local seconds)
) -> str:  # Deterministic Segment node id
    """
    Fine Segment identity = audio-side only (audio segment, VAD config, chunk
    range) — so the skeleton's ids are SHARED across transcribers by
    construction (C4 "store agreement once" falls out of identity design).
    """

Classes

class TranscriptGraphLabels:
    "Node labels of the audio-transcript layer schema."
    
    def all(cls) -> list:  # All schema labels
        "All schema labels."

@dataclass
class SourceNode:
    "The provenance root: one ingested media file."
    
    content_hash: str  # Content hash over the file ("algo:hexdigest"; the identity input)
    path: str  # Original media path (location, may dangle; identity is the hash)
    title: Optional[str]  # Display title; None = path stem
    media_type: str = 'audio'  # Media type
    
    def id(self) -> str:  # Deterministic node id
            """Deterministic node id."""
            return source_node_id(self.content_hash)
    
        def to_graph_node(self) -> Dict[str, Any]:  # Node wire dict
        "Deterministic node id."
    
    def to_graph_node(self) -> Dict[str, Any]:  # Node wire dict
            """Build the Source node wire dict (root_kind=ingested; FileRef provenance)."""
            return {
                "id": self.id,
        "Build the Source node wire dict (root_kind=ingested; FileRef provenance)."

@dataclass
class AudioSegmentNode:
    """
    Coarse ~5-min spine member. Owns the model-input WAV (E14: the audio of
    record) as payload/provenance — the WAV is NOT its own node.
    
    Provenance note (deliberate): the SourceRef anchors the owned model-input
    WAV artifact (hash_file-verifiable). A hash over the sliced ORIGINAL source
    bytes is not materializable without extracting per-range artifacts (decoded
    ranges are not byte ranges); the structural chain to the Source rides the
    PART_OF edge + the start/end properties.
    """
    
    source: str  # Owning Source node id
    index: int  # Position in the coarse spine (0-based)
    start: float  # Boundary start (source-coordinate seconds)
    end: float  # Boundary end (source-coordinate seconds)
    model_input_path: str  # The 16kHz mono WAV consumed by transcription/VAD/FA
    model_input_hash: str  # Content hash over that WAV
    segment_path: Optional[str]  # Source-codec cut file (archival pointer)
    
    def id(self) -> str:  # Deterministic node id
            """Deterministic node id."""
            return audio_segment_node_id(self.source, self.start, self.end)
    
        def to_graph_node(self) -> Dict[str, Any]:  # Node wire dict
        "Deterministic node id."
    
    def to_graph_node(self) -> Dict[str, Any]:  # Node wire dict
            """Build the AudioSegment node wire dict."""
            props: Dict[str, Any] = {
        "Build the AudioSegment node wire dict."

@dataclass
class TranscriptNode:
    """
    One transcriber's text for one AudioSegment (per-transcriber variants at
    the coarse layer — cross-transcriber divergence lives here, C4/C14).
    """
    
    audio_segment: str  # Owning AudioSegment node id
    transcriber: str  # Transcriber capability name
    config_hash: str  # Transcriber config hash
    text: str  # The transcript text (stored ONCE here; fine Segments slice into it)
    audio_hash: str  # Content hash of the consumed model-input WAV
    metadata: Dict[str, Any] = field(...)  # Transcriber-reported metadata
    asserted_at: Optional[float]  # Derivation timestamp; None = now
    
    def id(self) -> str:  # Deterministic node id
            """Deterministic node id."""
            return transcript_node_id(self.audio_segment, self.transcriber, self.config_hash)
    
        def to_graph_node(self) -> Dict[str, Any]:  # Node wire dict
        "Deterministic node id."
    
    def to_graph_node(self) -> Dict[str, Any]:  # Node wire dict
            """Build the Transcript node wire dict (capability attribution included)."""
            props: Dict[str, Any] = {
        "Build the Transcript node wire dict (capability attribution included)."
    
    def derived_edge(self) -> Dict[str, Any]:  # Edge wire dict
        "DERIVED_FROM edge: this Transcript derives from its AudioSegment."

@dataclass
class TranscriptSliceRef:
    """
    One per-transcriber char-range reference for a fine Segment: where this
    segment's text lives inside a Transcript node's text (Thread-1
    slices-until-promoted — variants are slices, never duplicated fine text).
    """
    
    transcript: str  # Transcript node id
    start_char: int  # Slice start into the Transcript's text
    end_char: int  # Slice end
    text: str  # The sliced text (content-hashed for the ref)
    
    def to_source_ref(self) -> Dict[str, Any]:  # SourceRef wire dict
        "Build the slice-shaped provenance ref."

@dataclass
class SegmentNode:
    """
    Fine spine member: one VAD chunk — IMMUTABLE audio range + CORRECTABLE
    text (the immutable-audio/mutable-text spine seam).
    
    Layer-0 `text` is the ACCURACY transcriber's alignment; the designation is
    per-segment provenance, not global config (`text_from` names the
    authoritative Transcript; every transcriber's char range rides
    `text_slices`, the authoritative one included).
    """
    
    audio_segment: str  # Owning AudioSegment node id
    vad_config_hash: str  # VAD config hash (skeleton identity input)
    chunk_start: float  # VAD chunk start (chunk-local seconds within the AudioSegment WAV)
    chunk_end: float  # VAD chunk end (chunk-local seconds)
    index: int  # Source-wide fine-spine index (0-based)
    start_time: float  # Source-coordinate start (navigation)
    end_time: float  # Source-coordinate end
    text: str = ''  # Layer-0 text (accuracy alignment; "" = no aligned words, D14 class)
    audio_hash: str = ''  # Content hash of the owning AudioSegment's model-input WAV
    source: Optional[str]  # Source node id (convenience for direct filters)
    text_from: Optional[str]  # Authoritative Transcript node id (None when text is empty)
    text_slices: List[TranscriptSliceRef] = field(...)  # All per-transcriber slice refs
    
    def id(self) -> str:  # Deterministic node id
            """Deterministic node id (audio-side identity; shared across transcribers)."""
            return segment_node_id(self.audio_segment, self.vad_config_hash,
                                   self.chunk_start, self.chunk_end)
    
        def to_graph_node(self) -> Dict[str, Any]:  # Node wire dict
        "Deterministic node id (audio-side identity; shared across transcribers)."
    
    def to_graph_node(self) -> Dict[str, Any]:  # Node wire dict
            """Build the Segment node wire dict (audio ref + per-transcriber text slice refs)."""
            props: Dict[str, Any] = {
        "Build the Segment node wire dict (audio ref + per-transcriber text slice refs)."