Dikte 1.0.1

Cut a long file into chunks a hosted request can outlive

dikte/__init__.py

@@ -10,4 +10,4 @@ business loading Qt to answer one question.
 # both the .dmg's Info.plist and the AppImage's file name are built from it. A
 # build off master rather than off a tag appends the commit to it, so that a
 # bug report from someone running "latest" names a commit.
-__version__ = "1.0.0"
+__version__ = "1.0.1"

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,

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):

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}",

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):

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)