diff --git a/chapters.py b/chapters.py new file mode 100644 index 0000000..d1646ee --- /dev/null +++ b/chapters.py @@ -0,0 +1,237 @@ +"""Chapter detection from silence structure. + +Long recordings (lectures, meetings, podcasts) usually contain natural +sections separated by long stretches of silence. This module turns a list +of detected silence intervals plus the total media duration into proposed +chapter markers that media players can use for navigation. + +The module is deliberately independent of ``media_shrinker``: it accepts any +object exposing ``start_seconds`` and ``end_seconds`` float attributes (the +same duck-type shape as ``media_shrinker.SilenceInterval``), so callers can +feed it silencedetect output without a hard import dependency. + +Outputs are exportable as: + +* ffmpeg FFMETADATA chapters (``;FFMETADATA1`` format) via + :func:`to_ffmetadata`, suitable for + ``ffmpeg -i input -i chapters.txt -map_metadata 1 -codec copy output``. +* A simple JSON listing via :func:`to_json`. + +Only the Python standard library is used. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import Protocol + + +class SilenceLike(Protocol): + """Duck-type contract for a silence interval. + + Any object with ``start_seconds`` and ``end_seconds`` float attributes + satisfies this protocol, including ``media_shrinker.SilenceInterval``. + """ + + start_seconds: float + end_seconds: float + + +@dataclass(frozen=True) +class Chapter: + """A proposed chapter marker covering ``[start, end)`` in seconds.""" + + index: int + start: float + end: float + title: str + + @property + def duration(self) -> float: + """Return the chapter length in seconds.""" + + return self.end - self.start + + +def _clamped_silences( + silences: list[SilenceLike] | tuple[SilenceLike, ...], + total_duration: float, +) -> list[tuple[float, float]]: + """Sort, clamp, and merge raw silence intervals into clean spans. + + Intervals are clamped to ``[0, total_duration]``, empty or inverted + spans are discarded, and overlapping or touching spans are merged so + downstream boundary logic sees each silent region exactly once. + """ + + spans: list[tuple[float, float]] = [] + for silence in silences: + start = max(0.0, min(float(silence.start_seconds), total_duration)) + end = max(0.0, min(float(silence.end_seconds), total_duration)) + if end > start: + spans.append((start, end)) + spans.sort() + merged: list[tuple[float, float]] = [] + for start, end in spans: + if merged and start <= merged[-1][1]: + merged[-1] = (merged[-1][0], max(merged[-1][1], end)) + else: + merged.append((start, end)) + return merged + + +def _boundaries_from_silences( + spans: list[tuple[float, float]], + total_duration: float, + min_gap_seconds: float, +) -> list[float]: + """Return chapter split points at the midpoints of long silences. + + Only silences lasting at least ``min_gap_seconds`` produce a boundary, + and boundaries falling on the extremes of the timeline are dropped + because they would create zero-length chapters. + """ + + boundaries: list[float] = [] + for start, end in spans: + if end - start >= min_gap_seconds: + midpoint = (start + end) / 2.0 + if 0.0 < midpoint < total_duration: + boundaries.append(midpoint) + return boundaries + + +def _merge_short_chapters( + edges: list[float], + min_chapter_seconds: float, +) -> list[float]: + """Drop boundaries that would create chapters shorter than the minimum. + + ``edges`` is the full edge list ``[0, b1, ..., bn, total_duration]``. + A too-short chapter is merged into the previous chapter by removing the + boundary that starts it; a too-short first chapter is merged into the + following chapter instead, since it has no predecessor. + """ + + kept = list(edges) + changed = True + while changed and len(kept) > 2: + changed = False + for i in range(1, len(kept)): + if kept[i] - kept[i - 1] < min_chapter_seconds: + # Merge into the previous chapter by removing this + # chapter's starting boundary; the first chapter merges + # forward by removing its ending boundary instead. + del kept[i - 1 if i - 1 > 0 else i] + changed = True + break + return kept + + +def detect_chapters( + silences: list[SilenceLike] | tuple[SilenceLike, ...], + total_duration: float, + min_chapter_seconds: float = 60.0, + min_gap_seconds: float = 3.0, +) -> list[Chapter]: + """Propose chapter markers from silence structure. + + Rules: + + * A silence lasting at least ``min_gap_seconds`` ends a chapter at the + silence midpoint. + * Chapters shorter than ``min_chapter_seconds`` are merged into the + previous chapter (the first chapter merges into the next one). + * The resulting chapters always cover ``[0, total_duration]`` exactly. + * With no qualifying silences the whole recording is a single chapter. + + Silences are sorted first, so unsorted or overlapping input is handled; + intervals extending beyond ``total_duration`` (or before zero) are + clamped. + + Args: + silences: Objects with ``start_seconds``/``end_seconds`` attributes, + e.g. ``media_shrinker.SilenceInterval`` instances. + total_duration: Total media duration in seconds; must be positive. + min_chapter_seconds: Minimum chapter length before merging applies. + min_gap_seconds: Minimum silence length that splits chapters. + + Returns: + Chapters with 1-based ``index`` and titles like ``"Chapter 1"``. + + Raises: + ValueError: If ``total_duration`` is zero or negative. + """ + + if total_duration <= 0: + raise ValueError( + f"total_duration must be positive, got {total_duration!r}" + ) + spans = _clamped_silences(silences, total_duration) + boundaries = _boundaries_from_silences( + spans, total_duration, min_gap_seconds + ) + edges = [0.0, *boundaries, total_duration] + edges = _merge_short_chapters(edges, min_chapter_seconds) + chapters: list[Chapter] = [] + for position in range(len(edges) - 1): + chapters.append( + Chapter( + index=position + 1, + start=edges[position], + end=edges[position + 1], + title=f"Chapter {position + 1}", + ) + ) + return chapters + + +def _escape_ffmetadata(value: str) -> str: + """Escape characters that are special in FFMETADATA values.""" + + escaped = value + for character in ("\\", "=", ";", "#", "\n"): + escaped = escaped.replace(character, "\\" + character) + return escaped + + +def to_ffmetadata(chapters: list[Chapter]) -> str: + """Serialize chapters to ffmpeg's FFMETADATA1 chapter format. + + The output starts with the ``;FFMETADATA1`` header followed by one + ``[CHAPTER]`` block per chapter using ``TIMEBASE=1/1000`` with + ``START``/``END`` expressed in integer milliseconds. Feed the result to + ffmpeg via ``-map_metadata`` to embed chapter navigation in the output. + """ + + lines = [";FFMETADATA1"] + for chapter in chapters: + lines.append("") + lines.append("[CHAPTER]") + lines.append("TIMEBASE=1/1000") + lines.append(f"START={round(chapter.start * 1000)}") + lines.append(f"END={round(chapter.end * 1000)}") + lines.append(f"title={_escape_ffmetadata(chapter.title)}") + return "\n".join(lines) + "\n" + + +def to_json(chapters: list[Chapter]) -> str: + """Serialize chapters to a simple JSON listing. + + Each chapter becomes an object with ``index``, ``start``, ``end`` (both + in seconds), and ``title`` keys, preserving chapter order. + """ + + return json.dumps( + [ + { + "index": chapter.index, + "start": chapter.start, + "end": chapter.end, + "title": chapter.title, + } + for chapter in chapters + ], + indent=2, + ) diff --git a/tests/test_chapters.py b/tests/test_chapters.py new file mode 100644 index 0000000..a510ab3 --- /dev/null +++ b/tests/test_chapters.py @@ -0,0 +1,217 @@ +"""Tests for silence-based chapter detection and export.""" + +from __future__ import annotations + +import json +import sys +import unittest +from dataclasses import dataclass +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from chapters import Chapter, detect_chapters, to_ffmetadata, to_json + + +@dataclass(frozen=True) +class FakeSilence: + """Minimal duck-typed silence interval for the chapters contract.""" + + start_seconds: float + end_seconds: float + + +class DetectChaptersTest(unittest.TestCase): + """Behavioural tests for detect_chapters.""" + + def test_single_chapter_when_no_silences(self) -> None: + """No silences yields one chapter covering the full duration.""" + + chapters = detect_chapters([], total_duration=600.0) + self.assertEqual(len(chapters), 1) + self.assertEqual(chapters[0].index, 1) + self.assertEqual(chapters[0].start, 0.0) + self.assertEqual(chapters[0].end, 600.0) + self.assertEqual(chapters[0].title, "Chapter 1") + + def test_boundary_at_long_silence_midpoint(self) -> None: + """A qualifying silence splits chapters at its midpoint.""" + + silences = [FakeSilence(300.0, 310.0)] + chapters = detect_chapters(silences, total_duration=600.0) + self.assertEqual(len(chapters), 2) + self.assertEqual(chapters[0].start, 0.0) + self.assertEqual(chapters[0].end, 305.0) + self.assertEqual(chapters[1].start, 305.0) + self.assertEqual(chapters[1].end, 600.0) + self.assertEqual( + [c.title for c in chapters], ["Chapter 1", "Chapter 2"] + ) + + def test_short_silence_does_not_split(self) -> None: + """Silences shorter than min_gap_seconds produce no boundary.""" + + silences = [FakeSilence(300.0, 301.0)] + chapters = detect_chapters( + silences, total_duration=600.0, min_gap_seconds=3.0 + ) + self.assertEqual(len(chapters), 1) + + def test_short_chapter_merges_into_previous(self) -> None: + """A too-short trailing chapter merges into its predecessor.""" + + silences = [FakeSilence(300.0, 306.0), FakeSilence(580.0, 590.0)] + chapters = detect_chapters( + silences, total_duration=600.0, min_chapter_seconds=60.0 + ) + # The 585..600 chapter (15s) merges back into the previous one. + self.assertEqual(len(chapters), 2) + self.assertEqual(chapters[0].end, 303.0) + self.assertEqual(chapters[1].start, 303.0) + self.assertEqual(chapters[1].end, 600.0) + + def test_short_first_chapter_merges_forward(self) -> None: + """A too-short first chapter merges into the following chapter.""" + + silences = [FakeSilence(10.0, 20.0), FakeSilence(300.0, 310.0)] + chapters = detect_chapters( + silences, total_duration=600.0, min_chapter_seconds=60.0 + ) + self.assertEqual(len(chapters), 2) + self.assertEqual(chapters[0].start, 0.0) + self.assertEqual(chapters[0].end, 305.0) + self.assertEqual(chapters[1].end, 600.0) + + def test_chapters_always_cover_full_duration(self) -> None: + """First chapter starts at 0 and last ends at total_duration.""" + + silences = [ + FakeSilence(100.0, 110.0), + FakeSilence(250.0, 260.0), + FakeSilence(400.0, 410.0), + ] + chapters = detect_chapters(silences, total_duration=500.0) + self.assertEqual(chapters[0].start, 0.0) + self.assertEqual(chapters[-1].end, 500.0) + for previous, current in zip(chapters, chapters[1:]): + self.assertEqual(previous.end, current.start) + self.assertEqual([c.index for c in chapters], [1, 2, 3, 4]) + + def test_silences_beyond_duration_are_clamped(self) -> None: + """Silence spans past total_duration are clamped, not boundaries.""" + + silences = [FakeSilence(590.0, 700.0), FakeSilence(650.0, 800.0)] + chapters = detect_chapters(silences, total_duration=600.0) + # Clamped span 590..600 has midpoint 595; the resulting 5s tail + # chapter merges into the previous one, leaving a single chapter. + self.assertEqual(len(chapters), 1) + self.assertEqual(chapters[0].end, 600.0) + + def test_unsorted_and_overlapping_silences(self) -> None: + """Unsorted, overlapping input is sorted and merged first.""" + + silences = [ + FakeSilence(400.0, 410.0), + FakeSilence(100.0, 106.0), + FakeSilence(103.0, 108.0), + ] + chapters = detect_chapters(silences, total_duration=600.0) + self.assertEqual(len(chapters), 3) + # 100..108 merged span has midpoint 104. + self.assertEqual(chapters[0].end, 104.0) + self.assertEqual(chapters[1].end, 405.0) + + def test_zero_duration_raises(self) -> None: + """A zero total_duration is rejected.""" + + with self.assertRaises(ValueError): + detect_chapters([], total_duration=0.0) + + def test_negative_duration_raises(self) -> None: + """A negative total_duration is rejected.""" + + with self.assertRaises(ValueError): + detect_chapters([], total_duration=-5.0) + + def test_short_recording_is_single_chapter(self) -> None: + """A recording shorter than min_chapter_seconds stays one chapter.""" + + silences = [FakeSilence(10.0, 15.0)] + chapters = detect_chapters( + silences, total_duration=30.0, min_chapter_seconds=60.0 + ) + self.assertEqual(len(chapters), 1) + self.assertEqual(chapters[0].end, 30.0) + + +class ExportTest(unittest.TestCase): + """Tests for FFMETADATA and JSON serialization.""" + + def test_ffmetadata_exact_format(self) -> None: + """FFMETADATA output matches ffmpeg's expected layout exactly.""" + + chapters = [ + Chapter(index=1, start=0.0, end=305.0, title="Chapter 1"), + Chapter(index=2, start=305.0, end=600.5, title="Chapter 2"), + ] + expected = ( + ";FFMETADATA1\n" + "\n" + "[CHAPTER]\n" + "TIMEBASE=1/1000\n" + "START=0\n" + "END=305000\n" + "title=Chapter 1\n" + "\n" + "[CHAPTER]\n" + "TIMEBASE=1/1000\n" + "START=305000\n" + "END=600500\n" + "title=Chapter 2\n" + ) + self.assertEqual(to_ffmetadata(chapters), expected) + + def test_ffmetadata_header_only_for_empty_list(self) -> None: + """An empty chapter list still yields the FFMETADATA header.""" + + self.assertEqual(to_ffmetadata([]), ";FFMETADATA1\n") + + def test_json_round_trip(self) -> None: + """JSON output parses back into the same chapter fields.""" + + chapters = detect_chapters( + [FakeSilence(300.0, 310.0)], total_duration=600.0 + ) + payload = json.loads(to_json(chapters)) + self.assertEqual( + payload, + [ + { + "index": 1, + "start": 0.0, + "end": 305.0, + "title": "Chapter 1", + }, + { + "index": 2, + "start": 305.0, + "end": 600.0, + "title": "Chapter 2", + }, + ], + ) + + def test_end_to_end_pipeline(self) -> None: + """detect_chapters output feeds directly into to_ffmetadata.""" + + silences = [FakeSilence(120.0, 126.0), FakeSilence(360.0, 366.0)] + chapters = detect_chapters(silences, total_duration=600.0) + metadata = to_ffmetadata(chapters) + self.assertTrue(metadata.startswith(";FFMETADATA1\n")) + self.assertEqual(metadata.count("[CHAPTER]"), 3) + self.assertIn("START=123000", metadata) + self.assertIn("END=363000", metadata) + + +if __name__ == "__main__": + unittest.main()