ddidderr
138c796399
Short chunk overlaps often contain only one participant, so returning speakers receive new transcript IDs. MAI can also split one voice into several IDs within a single request, which overlap stitching cannot fix. Add an opt-in audio comparison pass: --speakers N selects Gemini 3.8 Flash with a known count, and --speaker-model allows another compatible model. Keep request-local identities separate, select up to three clean excerpts per ID, and compare all excerpts in one OpenRouter request. Gemini uses Google AI Studio because the Vertex backend limits prompts to one audio file. No local model or extra Python dependency is required. Only reconcile identities whose sample assignments agree. Preserve uncertain identities separately, reject incomplete or malformed replies, and retain the original words and timestamps. Bound sample count and request size, clip sampling to the real audio duration, and extend worker cleanup and buffered output to the comparison step. Document the sampled method's limitations and additional provider request. Test Plan: - python3 -B -m unittest -v: all 82 tests passed, including real MP3 encoding with fake transcription/comparison responses and cancellation. - bash -n transcribe: passed. - git diff --cached --check: passed. - Live API and voice accuracy unverified: this workspace could not access the configured 1Password credential. Mocked responses prove pipeline behavior, not model accuracy on the supplied recording.
210 lines
10 KiB
Python
210 lines
10 KiB
Python
import contextlib
|
|
import copy
|
|
import io
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
import wave
|
|
|
|
from chunks import CONTEXT, ENCODING_MARGIN, merge, plan, speaker_mapping, stitch
|
|
from subtitles import convert
|
|
|
|
|
|
def word(text, start, end=None):
|
|
return {"word": text, "start": start, "end": start + 0.4 if end is None else end}
|
|
|
|
|
|
def make_audio(path, duration, pauses=()):
|
|
rate = 16000
|
|
samples = bytearray(b"\x10\x27" * round(rate * duration))
|
|
for start, end in pauses:
|
|
first, last = round(start * rate) * 2, round(end * rate) * 2
|
|
samples[first:last] = bytes(last - first)
|
|
with wave.open(str(path), "wb") as audio:
|
|
audio.setparams((1, 2, rate, 0, "NONE", "not compressed"))
|
|
audio.writeframes(samples)
|
|
|
|
|
|
class ChunkPlanTests(unittest.TestCase):
|
|
def planned(self, duration, pauses=(), limit=60):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "audio.wav"
|
|
make_audio(path, duration, pauses)
|
|
return plan(path, limit)
|
|
|
|
def test_short_audio_is_one_upload(self):
|
|
self.assertEqual(self.planned(20), [{"start": 0, "end": 20, "seam": 0}])
|
|
|
|
def test_continuous_speech_is_bounded_and_has_no_gaps(self):
|
|
for duration in (59.9, 60, 70, 120, 140):
|
|
with self.subTest(duration=duration):
|
|
chunks = self.planned(duration)
|
|
self.assertEqual(chunks[0]["start"], 0)
|
|
self.assertEqual(chunks[-1]["end"], duration)
|
|
for chunk in chunks:
|
|
self.assertGreater(chunk["end"], chunk["start"])
|
|
self.assertLessEqual(chunk["end"] - chunk["start"], 60 - ENCODING_MARGIN + 1e-8)
|
|
for previous, current in zip(chunks, chunks[1:]):
|
|
self.assertAlmostEqual(previous["end"] - current["start"], CONTEXT * 2)
|
|
self.assertAlmostEqual(current["seam"] - current["start"], CONTEXT)
|
|
|
|
def test_uses_last_pause_before_limit(self):
|
|
chunks = self.planned(130, [(30, 31), (53, 54)])
|
|
self.assertAlmostEqual(chunks[1]["seam"], 53.5)
|
|
self.assertAlmostEqual(chunks[0]["end"], 56.5)
|
|
self.assertAlmostEqual(chunks[1]["start"], 50.5)
|
|
|
|
def test_short_quiet_sound_does_not_count_as_pause(self):
|
|
chunks = self.planned(70, [(53, 53.1)])
|
|
self.assertAlmostEqual(chunks[1]["seam"], 56.9)
|
|
|
|
def test_all_silence_still_makes_progress(self):
|
|
chunks = self.planned(140, [(0, 140)])
|
|
self.assertEqual(chunks[-1]["end"], 140)
|
|
self.assertLess(len(chunks), 10)
|
|
self.assertTrue(all(0 < c["end"] - c["start"] <= 60 for c in chunks))
|
|
|
|
def test_invalid_limit_and_empty_audio(self):
|
|
for limit in (0, 59, 1801, float("inf"), float("nan")):
|
|
with self.subTest(limit=limit), self.assertRaises(ValueError):
|
|
self.planned(1, limit=limit)
|
|
with self.assertRaisesRegex(ValueError, "no samples"):
|
|
self.planned(0)
|
|
|
|
|
|
class ChunkMergeTests(unittest.TestCase):
|
|
def test_speaker_mapping_uses_shared_words_even_when_provider_ids_swap(self):
|
|
left = [dict(word(text, start), speaker=speaker) for text, start, speaker in (
|
|
("one", 58, 0), ("two", 58.5, 0), ("three", 60, 1), ("four", 60.5, 1),
|
|
)]
|
|
right = [dict(w, speaker=1 - w["speaker"]) for w in left]
|
|
self.assertEqual(speaker_mapping(left, right, 60), {1: 0, 0: 1})
|
|
|
|
def test_speaker_mapping_rejects_missing_sparse_and_conflicting_evidence(self):
|
|
left = [dict(word(text, start), speaker=0) for text, start in (
|
|
("one", 58), ("two", 58.5), ("three", 60), ("four", 60.5),
|
|
)]
|
|
self.assertEqual(speaker_mapping(left, [dict(left[0], speaker=5)], 60), {})
|
|
self.assertEqual(speaker_mapping(left, [dict(word("elsewhere", 90), speaker=0)], 60), {})
|
|
conflicting = [dict(w, speaker=i // 2) for i, w in enumerate(left)]
|
|
self.assertEqual(speaker_mapping(left, conflicting, 60), {})
|
|
self.assertEqual(speaker_mapping(conflicting, left, 60), {})
|
|
repeated = [dict(word("yes", 60), speaker=0)]
|
|
self.assertEqual(speaker_mapping(repeated, repeated * 2, 60), {})
|
|
|
|
def test_overlap_keeps_speaker_identity_through_three_requests(self):
|
|
chunks = [{"start": 0, "end": 63, "seam": 0},
|
|
{"start": 57, "end": 120, "seam": 60},
|
|
{"start": 114, "end": 140, "seam": 117}]
|
|
responses = [
|
|
{"words": [dict(word(text, start), speaker=7) for text, start in
|
|
(("First", 1), ("shared", 59), ("words", 60))]},
|
|
{"words": [dict(word(text, start), speaker=2) for text, start in
|
|
(("shared", 2), ("words", 3), ("more", 59), ("speech", 60))]},
|
|
{"words": [dict(word(text, start), speaker=9) for text, start in
|
|
(("more", 2), ("speech", 3), ("Last", 20))]},
|
|
]
|
|
original = copy.deepcopy(responses)
|
|
with contextlib.redirect_stderr(io.StringIO()) as errors:
|
|
result = merge(chunks, responses)
|
|
self.assertEqual(result["text"], "First shared words more speech Last")
|
|
self.assertEqual({w["speaker"] for w in result["words"]}, {0})
|
|
self.assertEqual(errors.getvalue(), "")
|
|
self.assertEqual(responses, original)
|
|
|
|
def test_boundary_word_survives_once_despite_timestamp_drift(self):
|
|
left = [word("before", 8), word("boundary,", 9.8, 10.4), word("after", 10.5)]
|
|
right = [word("Boundary", 10.0, 10.6), word("after", 10.6), word("later", 14)]
|
|
result = stitch(left, right, 10)
|
|
self.assertEqual([w["word"] for w in result], ["before", "boundary,", "after", "later"])
|
|
|
|
def test_scoped_speakers_keep_shared_words_but_defer_identity_to_audio(self):
|
|
chunks = [{"start": 0, "end": 63, "seam": 0},
|
|
{"start": 57, "end": 100, "seam": 60}]
|
|
responses = [
|
|
{"words": [dict(word(text, start), speaker=0) for text, start in
|
|
(("First", 1), ("shared", 59), ("words", 60))]},
|
|
{"words": [dict(word(text, start), speaker=0) for text, start in
|
|
(("shared", 2), ("words", 3), ("Last", 20))]},
|
|
]
|
|
with contextlib.redirect_stderr(io.StringIO()) as errors:
|
|
result = merge(chunks, responses, scope_speakers=True)
|
|
self.assertEqual(result["text"], "First shared words Last")
|
|
self.assertEqual({w["speaker"] for w in result["words"]}, {0, 1})
|
|
self.assertNotEqual(result["words"][0]["speaker"], result["words"][-1]["speaker"])
|
|
self.assertEqual(errors.getvalue(), "")
|
|
|
|
def test_words_cut_at_upload_edges_are_discarded(self):
|
|
left = [word("Before", 5), word("shared", 9), word("words", 10), word("incom-", 12.8)]
|
|
right = [word("-fore", 7), word("shared", 9.1), word("words", 10.1), word("incomplete", 12.6)]
|
|
result = stitch(left, right, 10)
|
|
self.assertEqual([w["word"] for w in result], ["Before", "shared", "words", "incomplete"])
|
|
|
|
def test_intentional_repeated_words_are_not_removed(self):
|
|
left = [word("yes", start, start + 0.1) for start in (9.4, 9.8, 10.2, 10.6)]
|
|
right = copy.deepcopy(left) + [word("done", 14)]
|
|
result = stitch(left, right, 10)
|
|
self.assertEqual([w["word"] for w in result], ["yes"] * 4 + ["done"])
|
|
|
|
def test_identical_text_outside_overlap_is_not_deduplicated(self):
|
|
left = [word("again", 1)]
|
|
right = [word("again", 20)]
|
|
self.assertEqual(stitch(left, right, 10), left + right)
|
|
|
|
def test_no_match_uses_whole_words_and_warns(self):
|
|
left = [word("before", 8), word("left", 9.5, 10.1), word("discard", 11)]
|
|
right = [word("discard", 8), word("right", 10.1), word("after", 14)]
|
|
with contextlib.redirect_stderr(io.StringIO()) as errors:
|
|
result = stitch(left, right, 10)
|
|
self.assertEqual([w["word"] for w in result], ["before", "left", "right", "after"])
|
|
self.assertIn("joining by timestamps", errors.getvalue())
|
|
|
|
def test_silent_overlap_does_not_warn(self):
|
|
with contextlib.redirect_stderr(io.StringIO()) as errors:
|
|
self.assertEqual(stitch([word("before", 1)], [word("after", 20)], 10),
|
|
[word("before", 1), word("after", 20)])
|
|
self.assertEqual(errors.getvalue(), "")
|
|
|
|
def test_offsets_three_chunks_and_keeps_srt_numbering(self):
|
|
chunks = [{"start": 0, "end": 63, "seam": 0},
|
|
{"start": 57, "end": 120, "seam": 60},
|
|
{"start": 114, "end": 140, "seam": 117}]
|
|
responses = [
|
|
{"words": [word("First.", 1), word("Join.", 59.8, 60.2)]},
|
|
{"words": [word("Join.", 2.9, 3.3), word("Second.", 20), word("Next.", 59.8, 60.2)]},
|
|
{"words": [word("Next.", 2.9, 3.3), word("Last.", 20)]},
|
|
]
|
|
original = copy.deepcopy(responses)
|
|
result = merge(chunks, responses)
|
|
self.assertEqual(result["text"], "First. Join. Second. Next. Last.")
|
|
self.assertEqual([w["start"] for w in result["words"]], [1, 59.8, 77, 116.8, 134])
|
|
self.assertEqual(responses, original)
|
|
srt = convert(result)
|
|
self.assertIn("5\n00:02:14,000", srt)
|
|
|
|
def test_regression_preserves_order_and_scopes_speakers_per_chunk(self):
|
|
chunks = [{"start": 0, "end": 63, "seam": 0}, {"start": 57, "end": 100, "seam": 60}]
|
|
first = dict(word("First", 10), speaker=0)
|
|
second = dict(word("Second", 9), speaker=0)
|
|
last = dict(word("Last", 20), speaker=0)
|
|
result = merge(chunks, [{"words": [first, second]}, {"words": [last]}])
|
|
self.assertEqual([w["start"] for w in result["words"]], [10, 10, 77])
|
|
self.assertNotEqual(result["words"][0]["speaker"], result["words"][-1]["speaker"])
|
|
|
|
def test_missing_or_malformed_words_fail(self):
|
|
chunks = [{"start": 0, "end": 30, "seam": 0}]
|
|
for response in ({"text": "missing"}, {"text": "missing", "words": []},
|
|
{"text": "missing", "words": [word(" ", 1)]},
|
|
{"text": None, "words": []},
|
|
{"words": [word("invalid", 100)]}, {"words": [word("invalid", -1)]},
|
|
{"error": "bad response"}, []):
|
|
with self.subTest(response=response), self.assertRaisesRegex(ValueError, "Chunk 1"):
|
|
merge(chunks, [response])
|
|
with self.assertRaises(ValueError):
|
|
merge(chunks, [])
|
|
self.assertEqual(merge(chunks, [{"text": "", "words": []}])["text"], "")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|