cjm-transcription-audio-segment 0.0.4

FastHTML audio segmentation component for transcription workflows — VAD analysis, configurable boundary-midpoint segmentation, audio splitting at speech boundaries, and spot-check verification.

cjm-transcription-audio-segment

Install

pip install cjm_transcription_audio_segment

Project Structure

nbs/
├── components/ (7)
│   ├── config_panel.ipynb     # Configuration panel — segment-duration preset buttons (5 min / 10 min) + custom-duration numeric input. Composes V1 button roles into a V10 content_panel.
│   ├── helpers.ipynb          # Shared rendering helpers — section headers + row-action icon buttons. Mirrors the local-helper pattern in `cjm-transcript-workflow-management` and `cjm-fasthtml-workflow-session-management` pending closure of G23 (row-action-icon role catalog gap).
│   ├── segment_results.ipynb  # Segmentation results panel — aggregate stats + collapsible per-source segment list. Composes V10 panels + V13 text tiers.
│   ├── source_overview.ipynb  # Source overview panel — table of input audio sources with per-source operational status (Not analyzed → N chunks → N segments).
│   ├── spot_check.ipynb       # Spot-check panel — source dropdown, random + jump-to-index controls, HTML5 audio player. Verifies segment audio quality after segmentation. Optional / advisory step (does not gate StepFlow Next).
│   ├── step_renderer.ipynb    # Step renderer factory — builds the `render_step(ctx)` closure that assembles the V2 step header band, all panels, JobMonitor mount slots, and `check_inflight_job` resume-on-reload UX. Called by `init_audio_segment_routers()`.
│   └── vad_results.ipynb      # VAD results panel — per-source + aggregate VAD statistics, plus chunk-duration distribution. Composes V10 panels + V13 text tiers.
├── routes/ (5)
│   ├── analyze.ipynb     # VAD analysis route — wires `cjm-fasthtml-job-monitor` to run Silero VAD per source via `JobMonitorService`. Configures the monitor with `state_key=vad_job_seq` + `id_prefix=vad-jm` per the convention in `html_ids.py`.
│   ├── core.ipynb        # Shared route helpers and state I/O for the audio-segment step. Owns the `STEP_ID` constant, state-access helpers, the audio file serving route, and the `set_duration` config route (handles re-configure invalidation).
│   ├── init.ipynb        # `init_audio_segment_routers()` — the integration surface for hosts. Constructs services, wires all sub-routers (core, analyze, segment, spot-check), builds the V2-step-header-band step renderer, and returns an `AudioSegmentResult` dataclass with `routers`, `urls`, `render_step`, `sse_headers`, `audio_segment_available`, and `validate`.
│   ├── segment.ipynb     # Audio segmentation route — wires `cjm-fasthtml-job-monitor` to run ffmpeg `segment_audio` per source via `JobMonitorService`. Configures the monitor with `state_key=segment_job_seq` + `id_prefix=seg-jm` per the convention in `html_ids.py`.
│   └── spot_check.ipynb  # Spot-check route handlers — `random_segment` (GET) and `jump_to_segment` (GET, ?source=...&index=...). Both return the re-rendered spot-check panel for HTMX outerHTML swap.
├── services/ (1)
│   └── audio_segment.ipynb  # AudioSegmentService — VAD analysis, silence-gap-midpoint boundary computation, and audio segmentation via JobQueue-submitted plugin calls.
├── html_ids.ipynb  # Stable HTML ID constants for the audio-segment step
├── models.ipynb    # Data types, URL bundles, and result containers for the audio segmentation step
└── utils.ipynb     # Pure utility helpers: duration formatting, chunk-distribution summaries, segment-length preset table

Total: 16 notebooks across 3 directories

Module Dependencies

graph LR
    components_config_panel[components.config_panel<br/>components.config_panel]
    components_helpers[components.helpers<br/>components.helpers]
    components_segment_results[components.segment_results<br/>components.segment_results]
    components_source_overview[components.source_overview<br/>components.source_overview]
    components_spot_check[components.spot_check<br/>components.spot_check]
    components_step_renderer[components.step_renderer<br/>components.step_renderer]
    components_vad_results[components.vad_results<br/>components.vad_results]
    html_ids[html_ids<br/>html_ids]
    models[models<br/>models]
    routes_analyze[routes.analyze<br/>routes.analyze]
    routes_core[routes.core<br/>routes.core]
    routes_init[routes.init<br/>routes.init]
    routes_segment[routes.segment<br/>routes.segment]
    routes_spot_check[routes.spot_check<br/>routes.spot_check]
    services_audio_segment[services.audio_segment<br/>services.audio_segment]
    utils[utils<br/>utils]

    components_config_panel --> html_ids
    components_config_panel --> utils
    components_config_panel --> components_helpers
    components_segment_results --> html_ids
    components_segment_results --> utils
    components_segment_results --> models
    components_segment_results --> components_helpers
    components_source_overview --> html_ids
    components_source_overview --> utils
    components_source_overview --> models
    components_source_overview --> components_helpers
    components_spot_check --> html_ids
    components_spot_check --> utils
    components_spot_check --> models
    components_spot_check --> components_helpers
    components_step_renderer --> html_ids
    components_step_renderer --> models
    components_step_renderer --> components_vad_results
    components_step_renderer --> components_source_overview
    components_step_renderer --> routes_core
    components_step_renderer --> services_audio_segment
    components_step_renderer --> components_segment_results
    components_step_renderer --> components_config_panel
    components_step_renderer --> components_spot_check
    components_step_renderer --> utils
    components_vad_results --> html_ids
    components_vad_results --> utils
    components_vad_results --> models
    components_vad_results --> components_helpers
    routes_analyze --> html_ids
    routes_analyze --> components_vad_results
    routes_analyze --> routes_core
    routes_analyze --> components_source_overview
    routes_analyze --> services_audio_segment
    routes_core --> html_ids
    routes_core --> utils
    routes_init --> models
    routes_init --> routes_core
    routes_init --> routes_segment
    routes_init --> routes_analyze
    routes_init --> services_audio_segment
    routes_init --> components_step_renderer
    routes_init --> routes_spot_check
    routes_init --> utils
    routes_segment --> html_ids
    routes_segment --> routes_core
    routes_segment --> components_source_overview
    routes_segment --> services_audio_segment
    routes_segment --> components_segment_results
    routes_segment --> components_spot_check
    routes_segment --> utils
    routes_spot_check --> routes_core
    routes_spot_check --> components_spot_check
    routes_spot_check --> services_audio_segment
    services_audio_segment --> models
    services_audio_segment --> services_audio_segment

56 cross-module dependencies detected

CLI Reference

No CLI commands found in this project.

Module Overview

Detailed documentation for each module in the project:

routes.analyze (analyze.ipynb)

VAD analysis route — wires cjm-fasthtml-job-monitor to run Silero VAD per source via JobMonitorService. Configures the monitor with state_key=vad_job_seq + id_prefix=vad-jm per the convention in html_ids.py.

Import

from cjm_transcription_audio_segment.routes.analyze import (
    init_analyze_routes
)

Functions

def init_analyze_routes(
    """
    Build the VAD-analysis JobMonitor router for the audio-segment step.
    
    On completion, OOB-swaps:
      - source overview (statuses flip from 'Not analyzed' to 'N chunks')
      - VAD results panel (becomes visible with stats)
      - Segment trigger (if segment artifacts provided): re-renders with disabled=False
        so the user can click Segment without needing to refresh the page.
    """

services.audio_segment (audio_segment.ipynb)

AudioSegmentService — VAD analysis, silence-gap-midpoint boundary computation, and audio segmentation via JobQueue-submitted plugin calls.

Import

from cjm_transcription_audio_segment.services.audio_segment import (
    AudioSegmentService
)

Classes

class AudioSegmentService:
    def __init__(
        self,
        plugin_manager: PluginManager,                       # Loaded plugin registry
        queue: JobQueue,                                      # Shared job queue
        vad_plugin_name: str = "cjm-media-plugin-silero-vad",
        ffmpeg_plugin_name: str = "cjm-media-plugin-ffmpeg",
    )
    "Service for VAD analysis, boundary computation, and audio segmentation."
    
    def __init__(
            self,
            plugin_manager: PluginManager,                       # Loaded plugin registry
            queue: JobQueue,                                      # Shared job queue
            vad_plugin_name: str = "cjm-media-plugin-silero-vad",
            ffmpeg_plugin_name: str = "cjm-media-plugin-ffmpeg",
        )
    
    def is_available(self) -> bool
        "True iff both required plugins are loaded in the manager."
    
    async def analyze_audio(
        "Submit a VAD job, wait, and parse the result.

Silero VAD caches by `(file_path, config_hash)` — re-running with the same
config returns instantly from the plugin's cache. Pass `force=True` to bypass."
    
    def vad_result_to_dict(
            self,
            result: Any,        # MediaAnalysisResult OR a dict in compatible shape
            audio_path: str,    # The audio_path to embed in the VADResult
        ) -> VADResult
        "Parse a MediaAnalysisResult (or compatible dict) into VADResult shape.

Accepts both object form (`result.ranges`, `result.metadata`) and dict form
(`result['ranges']`, `result['metadata']`) so it survives direct plugin returns
AND JSON-deserialized cache hits without branching at call sites."
    
    def compute_segment_boundaries(
            self,
            vad_chunks: List[Dict[str, float]],   # [{start, end, ...}, ...] sorted by start
            max_segment_duration: float,           # Target max wall-clock segment length in seconds
            audio_duration: float,                 # Full audio duration in seconds
        ) -> List[Dict[str, float]]
        "Group VAD chunks into segments cut at silence-gap midpoints.

**Wall-clock-aware, pre-emptive cuts.** `max_segment_duration` caps the
wall-clock duration of each output segment (not the speech-only duration
within it) — this matches the downstream Qwen3 forced-alignment model's
constraint, which operates on the resulting audio file's length.

Algorithm:
  1. If audio_duration <= max_segment_duration or no chunks: single segment
     covering [0, audio_duration].
  2. Walk chunks sequentially. For each chunk after the first, check
     whether accepting it would push the in-progress segment's wall-clock
     duration over max. If so AND we already have content: cut **before**
     this chunk at the silence-gap midpoint between the previous chunk's
     end and this chunk's start. (If chunks abut with no gap, cut exactly
     at the previous chunk's end.)
  3. After the (possibly skipped) cut, accept the current chunk into the
     new in-progress segment.
  4. The final segment extends to audio_duration.

**Wall-clock invariant.** Every NON-FINAL segment's wall-clock duration is
<= max_segment_duration. The final segment may exceed max only because it
extends to audio_duration to cover any trailing silence after the last VAD
chunk. In practice this is small (real audio rarely has multi-minute trailing
silence). For very-trailing-silence inputs the final segment can be reduced
by reducing audio_duration to the last chunk's end + a small pad — but the
current implementation preserves the "cover all audio" guarantee per the spec.

Additionally: when a single VAD chunk's own duration exceeds max, that chunk
forms a segment of its native length — speech is never split mid-chunk."
    
    async def segment_audio(
            self,
            audio_path: str,                          # Absolute path to source audio
            boundaries: List[Dict[str, float]],        # [{start, end}, ...] from compute_segment_boundaries
        ) -> SegmentResult
        "Submit an ffmpeg `segment_audio` job, wait, and parse the result."
    
    def segment_result_to_dict(
        "Parse the ffmpeg `segment_audio` result into SegmentResult shape."
    
    def get_segment_info(
            self,
            segment_results: Dict[str, SegmentResult],   # audio_path -> SegmentResult
            source_path: Optional[str] = None,            # Defaults to first source with results
            index: Optional[int] = None,                  # Defaults to 0; clamped to valid range
        ) -> Optional[Dict[str, Any]]
        "Get display-ready info for a specific segment. Returns None if no match."
    
    def get_random_segment(
            self,
            segment_results: Dict[str, SegmentResult],  # audio_path -> SegmentResult
        ) -> Optional[Dict[str, Any]]
        "Pick a random segment from a random source. Returns None if no segments exist."

components.config_panel (config_panel.ipynb)

Configuration panel — segment-duration preset buttons (5 min / 10 min) + custom-duration numeric input. Composes V1 button roles into a V10 content_panel.

Import

from cjm_transcription_audio_segment.components.config_panel import (
    render_config_panel
)

Functions

def _preset_button(label: str, value: float, current: float, set_duration_url: str) -> FT
    "Render a single preset button. Active leg uses V1 primary_action; inactive uses secondary_action."

def render_config_panel(
    current_duration: float,         # Currently selected max segment duration (seconds)
    set_duration_url: str,           # POST URL for setting a new duration
) -> FT
    """
    Render the configuration panel — preset buttons + custom-duration input.
    
    Preset buttons cover the common values (5 min / 10 min). The custom-duration
    input lets the user set any positive value in minutes; it posts to the same
    `set_duration_url` (which expects `duration` in seconds — we multiply on submit
    via an HX-Vals JSON expression that wraps the input value).
    """

routes.core (core.ipynb)

Shared route helpers and state I/O for the audio-segment step. Owns the STEP_ID constant, state-access helpers, the audio file serving route, and the set_duration config route (handles re-configure invalidation).

Import

from cjm_transcription_audio_segment.routes.core import (
    STEP_ID,
    get_audio_segment_state,
    update_audio_segment_state,
    init_core_routes
)

Functions

def get_audio_segment_state(
    state_store: SQLiteWorkflowStateStore,   # State store instance
    workflow_id: str,                          # Workflow ID for state access
    session_id: str,                           # Session ID for state access
) -> Tuple[Dict[str, Any], Dict[str, Any]]
    "Return (step_state, full_state) for the audio-segment step."

def update_audio_segment_state(
    state_store: SQLiteWorkflowStateStore,
    workflow_id: str,
    session_id: str,
    full_state: Dict[str, Any],                # Full workflow state (will be mutated and persisted)
    step_state_updates: Dict[str, Any],         # Keys to merge into step_states[STEP_ID]
) -> None
    "Merge step_state_updates into step_states[STEP_ID] and persist."

def init_core_routes(
    state_store: SQLiteWorkflowStateStore,
    workflow_id: str,
    prefix: str = "/core",
) -> Tuple[APIRouter, str, str]
    """
    Build the core router and return `(router, audio_src_url, set_duration_url)`.
    
    Exposing the URLs lets `init_audio_segment_routers()` populate the
    `AudioSegmentUrls` bundle without having to grep the router's internal
    handler functions.
    """

Variables

STEP_ID: str = 'audio_segment'

components.helpers (helpers.ipynb)

Shared rendering helpers — section headers + row-action icon buttons. Mirrors the local-helper pattern in cjm-transcript-workflow-management and cjm-fasthtml-workflow-session-management pending closure of G23 (row-action-icon role catalog gap).

Import

from cjm_transcription_audio_segment.components.helpers import (
    render_section_header,
    render_icon_button
)

Functions

def render_section_header(
    title: str,         # Section title text
    icon_name: str,     # Lucide icon name (kebab-case)
) -> FT
    "Render a section header with a leading icon at V11 `icons.section_header` size."

def render_icon_button(
    icon_name: str,                       # Lucide icon name
    label: str,                           # Accessible (title) label
    color: Optional[str] = None,           # Optional daisyui btn color class
    size: Optional[str] = None,            # Optional daisyui btn size class (default sm)
    **kwargs,                              # HTMX or HTML attributes
) -> FT
    "Render a ghost-styled icon-only button (toolbar / row-action pattern)."

html_ids (html_ids.ipynb)

Stable HTML ID constants for the audio-segment step

Import

from cjm_transcription_audio_segment.html_ids import (
    AudioSegmentHtmlIds
)

Classes

class AudioSegmentHtmlIds:
    "Stable HTML IDs for the audio-segment step's panels and mount slots."

routes.init (init.ipynb)

init_audio_segment_routers() — the integration surface for hosts. Constructs services, wires all sub-routers (core, analyze, segment, spot-check), builds the V2-step-header-band step renderer, and returns an AudioSegmentResult dataclass with routers, urls, render_step, sse_headers, audio_segment_available, and validate.

Import

from cjm_transcription_audio_segment.routes.init import (
    init_audio_segment_routers
)

Functions

def _make_validate(state_store: SQLiteWorkflowStateStore, workflow_id: str):
    """Build the validate(state) closure for StepFlow Next-gating."""
    def validate(state: Dict[str, Any]) -> bool
    "Build the validate(state) closure for StepFlow Next-gating."

def _make_on_enter(state_store: SQLiteWorkflowStateStore, workflow_id: str):
    """Build the on_enter(state, request, sess) closure. Pulls Step 1's committed_audio_paths."""
    PREV_STEP_ID = "source_select"

    def on_enter(state: Dict[str, Any], request, sess)
    "Build the on_enter(state, request, sess) closure. Pulls Step 1's committed_audio_paths."

def init_audio_segment_routers(
    state_store: SQLiteWorkflowStateStore,        # Workflow state store
    workflow_id: str,                              # Workflow identifier
    plugin_manager: PluginManager,                 # Plugin registry
    job_queue: JobQueue,                           # Shared job queue
    prefix: str = "",                              # URL prefix (e.g., "/audio_segment" for orchestration)
    vad_plugin_name: str = "cjm-media-plugin-silero-vad",
    ffmpeg_plugin_name: str = "cjm-media-plugin-ffmpeg",
    sysmon_plugin_name: Optional[str] = None,      # Optional GPU/CPU sysmon plugin for the Resources tab
    kb_system_id: Optional[str] = None,            # Optional keyboard-system ID for pause/resume
) -> AudioSegmentResult
    """
    Initialize the audio-segment step's services + routes + step renderer.
    
    Returns an `AudioSegmentResult` carrying everything the host needs:
      - `routers`: list of APIRouters to register on the FastHTML app
      - `urls`: AudioSegmentUrls bundle (consumer URLs; JM URLs stay internal)
      - `render_step(ctx) -> FT`: render this step's page
      - `sse_headers`: htmx-ext-sse + cleanup scripts (add once to `fast_app(hdrs=...)`)
      - `audio_segment_available`: True iff both required plugins are loaded
      - `validate(state) -> bool`: StepFlow Next-gate predicate
    """

models (models.ipynb)

Data types, URL bundles, and result containers for the audio segmentation step

Import

from cjm_transcription_audio_segment.models import (
    VADResult,
    SegmentResult,
    AudioSegmentState,
    AudioSegmentUrls,
    AudioSegmentResult
)

Classes

class VADResult(TypedDict):
    "Voice activity detection result for a single audio file."

class SegmentResult(TypedDict):
    "Segmentation result (set of audio segments) for a single source file."

class AudioSegmentState(TypedDict):
    "State for the audio segmentation workflow step."

@dataclass
class AudioSegmentUrls:
    "URL bundle for audio-segment route handlers."
    
    set_duration: str = ''  # POST: set max segment duration
    random_segment: str = ''  # GET: pick a random segment for playback
    jump_to_segment: str = ''  # GET: load a specific segment by source + index
    audio_src: str = ''  # GET: serve a segment audio file by path

@dataclass
class AudioSegmentResult:
    "Everything the host needs from init_audio_segment_routers()."
    
    urls: AudioSegmentUrls  # Consumer-side URL bundle (excludes JM URLs)
    render_step: Callable  # fn(ctx: InteractionContext) -> FT
    routers: List[APIRouter] = field(...)  # All routers to register on the host app
    sse_headers: list = field(...)  # Headers for app (SSE extension; covers both monitors)
    audio_segment_available: bool = False  # True iff both VAD + ffmpeg plugins are loaded
    validate: Optional[Callable]  # fn(state: dict) -> bool — StepFlow Next-gate

routes.segment (segment.ipynb)

Audio segmentation route — wires cjm-fasthtml-job-monitor to run ffmpeg segment_audio per source via JobMonitorService. Configures the monitor with state_key=segment_job_seq + id_prefix=seg-jm per the convention in html_ids.py.

Import

from cjm_transcription_audio_segment.routes.segment import (
    init_segment_routes
)

Functions

def init_segment_routes(
    monitor_service: JobMonitorService,        # Shared job-monitor service
    state_store: SQLiteWorkflowStateStore,     # Workflow state store
    workflow_id: str,                           # Workflow identifier
    service: AudioSegmentService,               # Audio-segment service (for boundary computation + parsing)
    prefix: str = "/segment",                  # URL prefix for segmentation routes
    overlay_target_id: Optional[str] = None,    # DOM element to darken under the modal
    kb_system_id: Optional[str] = None,         # Optional keyboard-system ID to pause/resume
    # URLs needed for OOB-rendering the spot-check panel on segmentation completion.
    # Empty defaults are fine when no spot-check route is registered — the panel
    # renders without a working audio player; pass real URLs from init_spot_check_routes.
    random_segment_url: str = "",
    jump_to_segment_url: str = "",
    audio_src_url: str = "",
) -> Tuple[APIRouter, JobMonitorUrls, JobMonitorHtmlIds, JobMonitorConfig]
    """
    Build the audio-segmentation JobMonitor router for the audio-segment step.
    
    Boundaries are computed from cached VAD results inside `job_args_builder` —
    so the segmentation route only requires VAD to have completed before invocation;
    no separate pre-compute step is needed.
    
    On completion, OOB-swaps three panels at once: the source list (statuses flip
    to "N segments"), the segment-results panel (becomes visible with collapsibles),
    and the spot-check panel (becomes visible with the first segment loaded).
    """

components.segment_results (segment_results.ipynb)

Segmentation results panel — aggregate stats + collapsible per-source segment list. Composes V10 panels + V13 text tiers.

Import

from cjm_transcription_audio_segment.components.segment_results import (
    render_segment_results
)

Functions

def _render_per_source_details(audio_path: str, sr: SegmentResult) -> FT:
    """Render a single Details/Summary block listing all segments for a source."""
    segments = sr.get("segments", [])
    fname = Path(audio_path).name

    segment_lines = []
    for seg in segments
    "Render a single Details/Summary block listing all segments for a source."

def render_segment_results(
    segment_results: Dict[str, SegmentResult],   # audio_path -> SegmentResult (or empty)
) -> FT
    "Render the segmentation-results panel. Empty -> empty placeholder Div with stable id."

components.source_overview (source_overview.ipynb)

Source overview panel — table of input audio sources with per-source operational status (Not analyzed → N chunks → N segments).

Import

from cjm_transcription_audio_segment.components.source_overview import (
    render_source_row,
    render_source_overview
)

Functions

def _source_status(
    audio_path: str,
    vad_results: Dict[str, VADResult],
    segment_results: Dict[str, SegmentResult],
) -> Tuple[str, str]
    "Return (status_text, badge_color_class) for an audio_path's current stage."

def render_source_row(
    idx: int,                                                    # Position in the source list
    audio_path: str,                                              # Absolute audio path
    vad_results: Dict[str, VADResult],                            # All VAD results by path
    segment_results: Dict[str, SegmentResult],                    # All segmentation results by path
) -> FT
    "Render a single Tr for the source-overview table. Stable id per row supports OOB swaps."

def render_source_overview(
    audio_paths: List[str],                                       # Input audio paths
    vad_results: Optional[Dict[str, VADResult]] = None,            # Per-source VAD result
    segment_results: Optional[Dict[str, SegmentResult]] = None,    # Per-source segmentation result
) -> FT
    """
    Render the source-overview panel as a V10 content_panel.
    
    Empty state (no audio paths) uses V8 `render_empty_state` — covers the case where
    Step 1's output hasn't been wired up yet.
    """

components.spot_check (spot_check.ipynb)

Spot-check panel — source dropdown, random + jump-to-index controls, HTML5 audio player. Verifies segment audio quality after segmentation. Optional / advisory step (does not gate StepFlow Next).

Import

from cjm_transcription_audio_segment.components.spot_check import (
    render_spot_check_panel
)

Functions

def _default_current_segment(
    segment_results: Dict[str, SegmentResult],
) -> Optional[Dict[str, Any]]
    "Return the first segment of the first source with segments, augmented per get_segment_info."

def render_spot_check_panel(
    segment_results: Dict[str, SegmentResult],     # All segmentation results
    random_segment_url: str,                        # GET URL for /spot_check/random_segment
    jump_to_segment_url: str,                       # GET URL for /spot_check/jump_to_segment
    audio_src_url: str,                             # GET URL for /core/audio_src (path query param)
    current_segment: Optional[Dict[str, Any]] = None,  # Currently displayed segment; defaults to first available
) -> FT
    "Render the spot-check panel. Empty -> empty placeholder Div with stable id."

routes.spot_check (spot_check.ipynb)

Spot-check route handlers — random_segment (GET) and jump_to_segment (GET, ?source=…&index=…). Both return the re-rendered spot-check panel for HTMX outerHTML swap.

Import

from cjm_transcription_audio_segment.routes.spot_check import (
    init_spot_check_routes
)

Functions

def init_spot_check_routes(
    state_store: SQLiteWorkflowStateStore,
    workflow_id: str,
    service: AudioSegmentService,
    audio_src_url: str,                      # GET URL for /core/audio_src — used by audio player
    prefix: str = "/spot_check",
) -> Tuple[APIRouter, str, str]
    """
    Build the spot-check router and return `(router, random_segment_url, jump_to_segment_url)`.
    
    Both handlers return the re-rendered spot-check panel; HTMX outerHTML-swaps it
    into the stable `tas-spot-check` id, replacing dropdown/buttons/audio player.
    """

components.step_renderer (step_renderer.ipynb)

Step renderer factory — builds the render_step(ctx) closure that assembles the V2 step header band, all panels, JobMonitor mount slots, and check_inflight_job resume-on-reload UX. Called by init_audio_segment_routers().

Import

from cjm_transcription_audio_segment.components.step_renderer import (
    build_render_step
)

Functions

def build_render_step(
    state_store: SQLiteWorkflowStateStore,
    workflow_id: str,
    service: AudioSegmentService,
    monitor_service: JobMonitorService,
    urls: AudioSegmentUrls,                             # Consumer-side URL bundle
    set_duration_url: str,                              # /core/set_duration URL
    analyze_urls: JobMonitorUrls,
    analyze_ids: JobMonitorHtmlIds,
    analyze_config: JobMonitorConfig,
    segment_urls: JobMonitorUrls,
    segment_ids: JobMonitorHtmlIds,
    segment_config: JobMonitorConfig,
    vad_plugin_name: str,
    ffmpeg_plugin_name: str,
) -> Callable
    "Build and return the `render_step(ctx)` callable for this step."

utils (utils.ipynb)

Pure utility helpers: duration formatting, chunk-distribution summaries, segment-length preset table

Import

from cjm_transcription_audio_segment.utils import (
    DEFAULT_MAX_SEGMENT_DURATION,
    SEGMENT_DURATION_PRESETS,
    format_duration,
    summarize_chunk_distribution
)

Functions

def format_duration(
    seconds: float,  # Duration in seconds; negative values are clamped to 0
) -> str
    "Format a duration as 'H:MM:SS' or 'M:SS' depending on length."

def summarize_chunk_distribution(
    chunks: List[Dict[str, float]],  # List of VAD chunks; each must have 'start' and 'end'
) -> Dict[str, float]
    "Compute min/avg/max of chunk durations. Returns {} for empty input."

Variables

DEFAULT_MAX_SEGMENT_DURATION: float = 300.0
SEGMENT_DURATION_PRESETS: List[Dict[str, Any]]

components.vad_results (vad_results.ipynb)

VAD results panel — per-source + aggregate VAD statistics, plus chunk-duration distribution. Composes V10 panels + V13 text tiers.

Import

from cjm_transcription_audio_segment.components.vad_results import (
    render_vad_results
)

Functions

def _aggregate_tile(label: str, value: str) -> FT
    "Render a single dashboard tile (V10 P1)."

def render_vad_results(
    vad_results: Dict[str, VADResult],   # audio_path -> VADResult (or empty dict)
) -> FT
    "Render the VAD results panel. Empty dict produces an empty placeholder Div with the stable id (for OOB swap)."