diff --git a/dikte/api.py b/dikte/api.py index 0c76b80..7a0e78f 100644 --- a/dikte/api.py +++ b/dikte/api.py @@ -58,10 +58,20 @@ def timestamp_model(provider, selected=""): return "openai/whisper-1" if provider == "openrouter" else "whisper-1" +# What a gateway in front of the model answers of its own accord: the request +# never reached the model, or the model was still working when the connection +# was given up on. Trying again is the only thing that fixes any of them, and +# with a long file it is worth the second try rather than losing the run. +RETRY_STATUS = frozenset({408, 429, 500, 502, 503, 504}) + + class ApiError(Exception): - def __init__(self, message, status=None): + def __init__(self, message, status=None, retryable=None): super().__init__(message) self.status = status + # Anything not on that list is the request itself being wrong, and it + # will be just as wrong the second time. + self.retryable = status in RETRY_STATUS if retryable is None else retryable class Aborted(Exception): @@ -218,7 +228,7 @@ def explain(exc, service): if exc.status == 429: return ApiError(t("{service} is rate limiting you (HTTP 429). Try again in " "a moment.", service=service), exc.status) - return ApiError(f"{service}: {exc}", exc.status) + return ApiError(f"{service}: {exc}", exc.status, retryable=exc.retryable) def _request(url, data, headers, timeout=120, aborter=None): @@ -234,8 +244,11 @@ def _request(url, data, headers, timeout=120, aborter=None): # not the network failing. URLError is an OSError, so both land here. if aborter is not None and aborter.aborted: raise Aborted from None + # A connection that dropped or timed out is the same bad minute as a + # 502, so it is worth the same second try. raise ApiError(t("Could not connect: {reason}", - reason=getattr(exc, "reason", exc))) from exc + reason=getattr(exc, "reason", exc)), + retryable=True) from exc except json.JSONDecodeError as exc: raise ApiError(t("Could not parse the response: {error}", error=exc)) from exc @@ -318,7 +331,7 @@ def local_failure(service, server, exc): """ detail = server.error() return ApiError(f"{service}: {exc}" + (f" ({detail})" if detail else ""), - exc.status) + exc.status, retryable=exc.retryable) def _transcribe_request(target, audio_path, language, prompt, response_format, diff --git a/dikte/filetranscribe.py b/dikte/filetranscribe.py index 8415752..084bb8f 100644 --- a/dikte/filetranscribe.py +++ b/dikte/filetranscribe.py @@ -1,15 +1,19 @@ """Transcribe an existing audio/video file with the same models. ffmpeg converts whatever comes in to 16 kHz mono WAV, and for a hosted API to -mp3 on top of that. The upload limit is the only reason a file is ever cut up, -and uncompressed audio reaches it after ten minutes where mp3 takes an hour. +mp3 on top of that. Two things decide where a file is cut up: the upload limit, +which uncompressed audio reaches after ten minutes where mp3 takes an hour, and +the clock. An hour of audio in one request is minutes of work at the other end, +and the gateway in front of the model hangs up long before the answer comes +back, which arrives here as a 502 with the whole chunk lost. So a chunk is also +capped at MAX_CHUNK_SECONDS however small it is on disk. -That is worth the encoder, because a cut is not free. Whisper hears in thirty -second windows and decides for itself where one cue ends and the next begins; a -chunk that starts in the middle of a sentence can come back as one cue per -window, twenty seconds of text at a time, for the whole rest of the chunk. So -the file is cut as rarely as the limit allows, what is cut overlaps, and -stitch() drops the half that was heard twice. +A cut is not free, which is what the encoder buys and why nothing is cut more +finely than that. Whisper hears in thirty second windows and decides for itself +where one cue ends and the next begins; a chunk that starts in the middle of a +sentence can come back as one cue per window, twenty seconds of text at a time, +for the whole rest of the chunk. So what is cut overlaps, and stitch() drops +the half that was heard twice. """ import contextlib @@ -19,6 +23,7 @@ import shutil import subprocess import tempfile import threading +import time import wave from PyQt6.QtCore import QObject, pyqtSignal @@ -29,10 +34,14 @@ from . import ggml from .i18n import t UPLOAD_LIMIT = 24 * 1024 * 1024 # the APIs take 25 MB; leave the form its room +MAX_CHUNK_SECONDS = 900 # as much audio as a hosted request can outlive MP3_BITRATE = "48k" # mono speech at 16 kHz: whisper hears nothing less OVERLAP_SECONDS = 30 # a whisper window: how far back a chunk starts WAV_CHUNK_SECONDS = 600 # 19 MB, for the caller that uploads the WAV itself CLEANUP_CHUNK_CHARS = 12000 # keep each cleanup call comfortably small +HOSTED_TIMEOUT = 600 # a quarter hour of audio, with room for the upload +RETRIES = 3 # how many times one chunk is asked for in all +RETRY_WAIT = 5 # seconds before the second try, doubled after that RATE = 16000 MIN_SUBTITLE_SECONDS = 1.5 # how long a cue with no end time of its own stays up @@ -86,9 +95,40 @@ class FileTranscriber(QObject): def _check(self): self._abort.check() + def _wait(self, seconds): + """Sleep on it, with the Stop button still able to get through.""" + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + self._check() + time.sleep(0.25) + self._check() + + def _attempt(self, call, stage): + """`call`, asked again when what failed was the network rather than us. + + One chunk is a quarter hour of audio that took a minute to encode and a + minute to upload, so a gateway having a bad moment is worth waiting out + rather than throwing the run away over. `stage` is what the status line + said before the failure, put back once the wait is over. + """ + for attempt in range(1, RETRIES + 1): + self._check() + try: + return call() + except api.ApiError as exc: + if attempt == RETRIES or not exc.retryable: + raise + self.progress.emit(t( + "{error} Trying again ({attempt}/{total})…", + error=exc, attempt=attempt + 1, total=RETRIES)) + self._wait(RETRY_WAIT * 2 ** (attempt - 1)) + self.progress.emit(stage) + def _work(self, path, timestamps, do_cleanup): conf = self.conf workdir = None + pieces = [] + segments = [] try: if not shutil.which("ffmpeg"): raise api.ApiError(t("ffmpeg not found. Install it to transcribe files.")) @@ -104,39 +144,36 @@ class FileTranscriber(QObject): if len(chunks) > 1: self.progress.emit(t("Splitting into {count} chunks…", count=len(chunks))) - pieces = [] - segments = [] for index, (chunk_path, offset) in enumerate(chunks, start=1): self._check() - self.progress.emit( - t("Transcribing chunk {index}/{count}…", - index=index, count=len(chunks)) - if len(chunks) > 1 else t("Transcribing…") - ) + stage = (t("Transcribing chunk {index}/{count}…", + index=index, count=len(chunks)) + if len(chunks) > 1 else t("Transcribing…")) + self.progress.emit(stage) if timestamps: - segments = stitch(segments, [ - (start + offset, end + offset, line) - for start, end, line in api.transcribe_segments( - target, - chunk_path, - language=conf["language"], - prompt=conf["transcribe_prompt"], - aborter=self._abort, - ) - ]) - else: - pieces.append(api.transcribe( + heard = self._attempt(lambda: api.transcribe_segments( target, chunk_path, language=conf["language"], prompt=conf["transcribe_prompt"], + timeout=HOSTED_TIMEOUT, aborter=self._abort, - )) + ), stage) + segments = stitch(segments, [ + (start + offset, end + offset, line) + for start, end, line in heard + ]) + else: + pieces.append(self._attempt(lambda: api.transcribe( + target, + chunk_path, + language=conf["language"], + prompt=conf["transcribe_prompt"], + timeout=HOSTED_TIMEOUT, + aborter=self._abort, + ), stage)) - if timestamps: - pieces = [f"[{format_timestamp(start)}] {line}" - for start, _, line in segments] - text = "\n".join(pieces) if timestamps else " ".join(pieces) + text = _joined(pieces, segments, timestamps) if do_cleanup and text: self._check() @@ -148,7 +185,16 @@ class FileTranscriber(QObject): except Cancelled: self.progress.emit(t("Stopped.")) except (api.ApiError, OSError, subprocess.SubprocessError, wave.Error) as exc: - self.failed.emit(str(exc)) + # An hour of a long file already heard is not worth throwing away + # because the chunk after it failed, or because cleanup did. Hand + # over what there is, and say in the same breath where it stops. + partial = _joined(pieces, segments, timestamps) + if partial: + self.finished.emit(partial, segments) + self.failed.emit(t("{error} The transcript up to there is below.", + error=exc)) + else: + self.failed.emit(str(exc)) finally: self._local = None if workdir: @@ -181,12 +227,21 @@ class FileTranscriber(QObject): self._local = ggml.llm if cleanup.provider(conf) == "local" else None prompt = conf.cleanup_prompt(with_timestamps=timestamps, subtitles=True) out = [] + stage = t("Cleaning up…") for block in split_text(text, timestamps): self._check() - out.append(cleanup.run(block, conf, prompt, aborter=self._abort)) + out.append(self._attempt( + lambda: cleanup.run(block, conf, prompt, aborter=self._abort), stage)) return ("\n" if timestamps else "\n\n").join(out) +def _joined(pieces, segments, timestamps): + """The transcript as one string, out of whichever of the two is holding it.""" + if timestamps: + pieces = [f"[{format_timestamp(start)}] {line}" for start, _, line in segments] + return "\n".join(pieces) if timestamps else " ".join(pieces) + + def format_timestamp(seconds): seconds = int(seconds) hours, rest = divmod(seconds, 3600) @@ -309,13 +364,20 @@ def wav_seconds(wav_path): def chunk_seconds(path, duration): """How many seconds of this audio fit in one request, or 0 when all of it does. - Measured rather than worked out: what an encoder makes of an hour of speech - depends on the speech, and the file on disk is the only honest answer. + Whichever of the two limits bites first. How much fits under the upload + limit is measured rather than worked out: what an encoder makes of an hour + of speech depends on the speech, and the file on disk is the only honest + answer. The other limit is MAX_CHUNK_SECONDS, and it is the one that catches + a long file at this bitrate: an hour and a half of mp3 is two chunks by size + and one of them is an hour of audio in a single request, which no hosted + gateway stays on the line for. """ - size = os.path.getsize(path) - if size <= UPLOAD_LIMIT or duration <= 0: + if duration <= 0: return 0.0 - return max(60.0, duration * UPLOAD_LIMIT / size * 0.95) + size = os.path.getsize(path) + fits = duration * UPLOAD_LIMIT / size * 0.95 if size > UPLOAD_LIMIT else duration + seconds = max(60.0, min(fits, MAX_CHUNK_SECONDS)) + return 0.0 if seconds >= duration else seconds def split_wav(wav_path, workdir, seconds=WAV_CHUNK_SECONDS, overlap=OVERLAP_SECONDS): diff --git a/dikte/i18n.py b/dikte/i18n.py index 9b4826c..398f740 100644 --- a/dikte/i18n.py +++ b/dikte/i18n.py @@ -297,6 +297,10 @@ TR = { "Converting audio…": "Ses dönüştürülüyor…", "Splitting into {count} chunks…": "{count} parçaya bölünüyor…", "Transcribing chunk {index}/{count}…": "{index}/{count} parça yazıya çevriliyor…", + "{error} Trying again ({attempt}/{total})…": + "{error} Yeniden deneniyor ({attempt}/{total})…", + "{error} The transcript up to there is below.": + "{error} Oraya kadar çevrilen metin aşağıda.", "Done: {chars} characters.": "Bitti: {chars} karakter.", "Stopped.": "Durduruldu.", "Failed: {error}": "Başarısız: {error}", diff --git a/tests/test_api.py b/tests/test_api.py index 58ab960..6bb28f8 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -79,6 +79,33 @@ class Explain(DikteTest): def test_the_status_is_carried_through(self): self.assertEqual(self.error(429).status, 429) + def test_so_is_whether_it_is_worth_asking_again(self): + self.assertTrue(self.error(502).retryable) + self.assertFalse(self.error(401).retryable) + + +class Retryable(unittest.TestCase): + """Which failures a second try can fix, and which will fail the same way.""" + + def test_a_gateway_that_gave_up_waiting(self): + for status in (408, 429, 500, 502, 503, 504): + with self.subTest(status=status): + self.assertTrue(api.ApiError("x", status).retryable) + + def test_a_request_that_was_wrong(self): + for status in (400, 401, 402, 403, 404, 413, 422): + with self.subTest(status=status): + self.assertFalse(api.ApiError("x", status).retryable) + + def test_an_error_of_our_own_is_not_the_network(self): + self.assertFalse(api.ApiError("Transcript came back empty.").retryable) + + def test_a_connection_that_dropped_is_worth_a_second_try(self): + with fake_urlopen(url_error("connection reset")): + with self.assertRaises(api.ApiError) as caught: + api._request("https://example.test", b"{}", {}) + self.assertTrue(caught.exception.retryable) + class ExtractError(unittest.TestCase): def test_the_usual_shape(self): diff --git a/tests/test_filetranscribe.py b/tests/test_filetranscribe.py index 2adc5ee..5bb2f75 100644 --- a/tests/test_filetranscribe.py +++ b/tests/test_filetranscribe.py @@ -214,10 +214,22 @@ class ChunkSeconds(DikteTest): self.assertEqual(ft.chunk_seconds(self.file(1024), 600), 0.0) def test_a_file_over_the_limit_is_cut_by_what_it_measured(self): - # Twice the limit over an hour, so a little under half an hour fits. - seconds = ft.chunk_seconds(self.file(ft.UPLOAD_LIMIT * 2), 3600) - self.assertGreater(seconds, 1500) - self.assertLess(seconds, 1800) + # Twice the limit over twenty minutes, so a little under ten fits. + seconds = ft.chunk_seconds(self.file(ft.UPLOAD_LIMIT * 2), 1200) + self.assertGreater(seconds, 500) + self.assertLess(seconds, 600) + + def test_a_chunk_is_never_more_audio_than_a_request_can_outlive(self): + """An hour in one request is a 502 from the gateway, whatever it weighs.""" + self.assertEqual(ft.chunk_seconds(self.file(ft.UPLOAD_LIMIT * 2), 3600), + ft.MAX_CHUNK_SECONDS) + + def test_a_small_file_that_is_still_hours_long_is_cut_on_the_clock(self): + self.assertEqual(ft.chunk_seconds(self.file(1024), 7200), + ft.MAX_CHUNK_SECONDS) + + def test_a_file_short_enough_on_both_counts_is_not_cut(self): + self.assertEqual(ft.chunk_seconds(self.file(1024), ft.MAX_CHUNK_SECONDS), 0.0) def test_a_file_with_no_length_is_left_whole(self): self.assertEqual(ft.chunk_seconds(self.file(ft.UPLOAD_LIMIT * 2), 0), 0.0) @@ -378,6 +390,58 @@ class Transcriber(DikteTest): worker.stop() self.assertTrue(worker._abort.aborted) + def test_a_chunk_is_given_longer_to_answer_than_a_dictation(self): + """A quarter hour of audio is not a sentence: the default would cut it off.""" + worker = ft.FileTranscriber(self.conf) + with mock.patch.object(ft, "_to_wav", side_effect=lambda *a: self.source), \ + mock.patch.object(ft, "_to_mp3", + side_effect=lambda path, *a, **k: path), \ + mock.patch.object(ft.shutil, "which", return_value="/usr/bin/ffmpeg"), \ + mock.patch.object(api, "transcribe", return_value="text") as call: + worker._work(self.source, False, False) + self.assertEqual(call.call_args.kwargs["timeout"], ft.HOSTED_TIMEOUT) + + def test_a_gateway_having_a_bad_moment_is_asked_again(self): + with mock.patch.object(ft.FileTranscriber, "_wait"): + done, failures, progress, _ = self.run_chain( + fail=[api.ApiError("HTTP 502: timeout", 502), "raw text"]) + self.assertEqual(failures, []) + self.assertEqual(done[0][0], "raw text") + self.assertTrue(any("Trying again" in message for message in progress)) + + def test_a_rejected_key_is_not_asked_again(self): + """Trying again with the same key is only a slower way to fail.""" + call = mock.Mock(side_effect=api.ApiError("rejected the API key", 401)) + with mock.patch.object(ft.FileTranscriber, "_wait"): + _, failures, _, _ = self.run_chain(fail=call) + self.assertEqual(call.call_count, 1) + self.assertIn("rejected", failures[0]) + + def test_a_chunk_is_given_up_on_after_the_last_try(self): + call = mock.Mock(side_effect=api.ApiError("HTTP 502: timeout", 502)) + with mock.patch.object(ft.FileTranscriber, "_wait"): + _, failures, _, _ = self.run_chain(fail=call) + self.assertEqual(call.call_count, ft.RETRIES) + self.assertIn("502", failures[0]) + + def test_what_was_heard_before_the_failure_is_still_handed_over(self): + """An hour already transcribed is not thrown away over the chunk after it.""" + boom = api.ApiError("HTTP 502: timeout", 502) + with mock.patch.object(ft.FileTranscriber, "_wait"), \ + mock.patch.object(ft.FileTranscriber, "_chunks", + side_effect=lambda wav, *a: [(wav, 0.0), (wav, 10.0)]): + done, failures, _, _ = self.run_chain( + fail=["first half"] + [boom] * ft.RETRIES) + self.assertEqual(done[0][0], "first half") + self.assertIn("502", failures[0]) + + def test_nothing_heard_at_all_is_a_plain_failure(self): + call = mock.Mock(side_effect=api.ApiError("rejected the API key", 401)) + with mock.patch.object(ft.FileTranscriber, "_wait"): + done, failures, _, _ = self.run_chain(fail=call) + self.assertEqual(done, []) + self.assertEqual(failures[0], "rejected the API key") + def test_a_second_start_while_one_is_running_is_ignored(self): worker = ft.FileTranscriber(self.conf) worker._thread = mock.Mock(is_alive=lambda: True)