Keep the meeting a quiet microphone gave us, and read both captures at once

A recording is never deleted for being disappointing. A microphone that
handed over nothing still leaves the right channel, which is everyone
else, and an hour of them is worth more than the empty channel costs; the
one thing the user cannot get back is the half that was there. So the
exact-zero check stays and stops throwing the file away: it says what the
microphone did, in a tray warning next to the recording being written up,
and the minutes are produced from what there is.

Reading the two capture pipes in turn from one thread put the failure it
was meant to fix back in a worse place. A microphone that stops delivering
leaves that read waiting forever, and the far side is not read either
until its pipe fills and its ffmpeg stops writing into it: the meeting
freezes, the levels sit still, and nothing is said for as long as nobody
looks. Each stream now has a reader of its own and a queue, so neither can
hold the other up, and a side that has said nothing for STALL_SECONDS ends
the recording the way a dead ffmpeg already did, out loud and keeping what
was captured.

Which system needs how many processes belongs in the table with everything
else that differs, so meeting() returns the list of commands it takes:
one on PulseAudio, one per device on a Mac. meeting_commands() is the
chooser again rather than a function with a Mac inside it, and the empty
entry in COREAUDIO is gone. The two AVFoundation targets are resolved
against a single device listing, which costs one ffmpeg run instead of two
and cannot see the indexes renumber between the microphone and the far
side.

audio.py

@@ -2,15 +2,16 @@
 
 Dictation records one source. A meeting records two of them at once, the
 microphone and what comes out of the speakers, and for that it goes through
-ffmpeg: one process reading both devices and merging them into the two channels
-of a single stream, which is the only way the two stay aligned with each other
-over an hour.
+ffmpeg. PulseAudio hands both devices to a single process, which merges them
+into the two channels of one stream and keeps them aligned itself. AVFoundation
+cannot be asked the same: two of its sessions inside one process starve each
+other, so a Mac captures each device on its own and the two mono streams are
+interleaved here as they arrive.
 
 Which programs do the capturing is a property of the machine, not of the code
 above: PulseAudio or PipeWire on Linux, AVFoundation through ffmpeg on macOS.
 They are gathered into one group each near the bottom of this file, and a
-chooser picks between them. macOS uses one ffmpeg process per AVFoundation
-device: two AVFoundation sessions in one process silently starve one another.
+chooser picks between them.
 """
 
 import array
@@ -18,6 +19,7 @@ import collections
 import json
 import math
 import os
+import queue
 import re
 import shutil
 import signal
@@ -39,6 +41,21 @@ CHUNK_BYTES = CHUNK_FRAMES * SAMPLE_WIDTH * CHANNELS
 CHUNK_LATENCY_MS = round(CHUNK_FRAMES / RATE * 1000)
 MIN_FRAMES = int(RATE * 0.25)
 
+# A capture process hands over a block every CHUNK_LATENCY_MS. One that has said
+# nothing for this long has stopped rather than fallen behind, and the meeting
+# ends and says so instead of sitting on a read that will never return.
+STALL_SECONDS = 5.0
+# Room for a whole stall of the other stream, so the side still delivering is
+# never the one left waiting.
+QUEUE_BLOCKS = int(STALL_SECONDS * RATE / CHUNK_FRAMES) + 8
+
+# Exact zeroes are not quiet, they are nothing: a microphone that is really in
+# the room has a noise floor. This much of a recording that long means it handed
+# nothing over, which is worth saying once the meeting is over and nothing can
+# be done about it any more.
+QUIET_MIC_SECONDS = 10
+QUIET_MIC_SHARE = 0.5
+
 
 class Recorder(QObject):
     """Runs the available sound-server recorder and reads raw PCM from stdout."""
@@ -193,17 +210,12 @@ def recording_command(target=""):
 def meeting_commands(mic_target, system_target):
     """The capture processes that produce one stereo meeting stream.
 
-    PulseAudio can keep both inputs in one ffmpeg process. AVFoundation cannot:
-    on a real Mac its two sessions silently starve the microphone, so each Mac
-    device is captured and clock-corrected by its own process. MeetingRecorder
-    interleaves those two mono streams after that.
+    One of them on PulseAudio, which merges both inputs itself; one per device
+    on a Mac, because two AVFoundation sessions in a process starve each other.
+    Which of the two it is stays in the table with everything else the sound
+    system decides, and MeetingRecorder reads the count rather than the machine.
     """
-    if sound() is COREAUDIO:
-        return [
-            _avfoundation_meeting_capture(mic_target),
-            _avfoundation_meeting_capture(system_target),
-        ]
-    return [sound().meeting(mic_target, system_target)]
+    return sound().meeting(mic_target, system_target)
 
 
 class AudioDeviceError(RuntimeError):
@@ -223,11 +235,11 @@ class MeetingRecorder(QObject):
     levels = pyqtSignal(float, float)      # mine, theirs
     stopped = pyqtSignal(str, float)       # wav path, duration (s)
     died = pyqtSignal()                    # ffmpeg quit on its own
+    warned = pyqtSignal(str)               # recorded, but something was wrong
     failed = pyqtSignal(str)
 
     def __init__(self, parent=None):
         super().__init__(parent)
-        self._proc = None
         self._procs = []
         self._thread = None
         self._wav = None
@@ -277,10 +289,10 @@ class MeetingRecorder(QObject):
                 self._procs.append(subprocess.Popen(
                     command, stdout=subprocess.PIPE, stderr=log, bufsize=0
                 ))
-            self._proc = self._procs[0]
         except (OSError, wave.Error) as exc:
+            # One of two capture processes may already be running, and a Mac
+            # left holding an open AVFoundation session records nothing else.
             self._terminate_processes()
-            self._proc = None
             self._procs = []
             self._close_file()
             self._drop_log()
@@ -326,13 +338,21 @@ class MeetingRecorder(QObject):
             pass
 
     def _pump_split(self):
-        left = self._procs[0].stdout
-        right = self._procs[1].stdout
+        # A reader thread per process. Taking turns on the two pipes from one
+        # thread would let a starved microphone hold up the far side too: its
+        # blocks would sit unread until the pipe filled and its ffmpeg stopped
+        # writing, and an hour of meeting would freeze with nothing said. Each
+        # stream is read as fast as it arrives, and a side that goes quiet for
+        # STALL_SECONDS ends the recording rather than hanging it.
         block = CHUNK_FRAMES * SAMPLE_WIDTH
+        streams = [queue.Queue(maxsize=QUEUE_BLOCKS) for _ in self._procs]
+        for proc, blocks in zip(self._procs, streams):
+            threading.Thread(target=_read_blocks, daemon=True,
+                             args=(proc.stdout, blocks, block)).start()
         try:
             while True:
-                mine = _read_exact(left, block)
-                theirs = _read_exact(right, block)
+                mine = _next_block(streams[0])
+                theirs = _next_block(streams[1])
                 if not mine or not theirs:
                     break
                 frames = min(len(mine), len(theirs)) // SAMPLE_WIDTH
@@ -407,7 +427,6 @@ class MeetingRecorder(QObject):
         self._thread = None
         codes = [proc.poll() for proc in self._procs]
         code = next((value for value in codes if value), 0)
-        self._proc = None
         self._procs = []
         self._close_file()
         return code
@@ -422,7 +441,7 @@ class MeetingRecorder(QObject):
             pass
 
     def stop(self):
-        if not self._proc:
+        if not self._procs:
             return
         # The count is read after the join: the pump thread is still appending
         # the last blocks up to the moment it ends.
@@ -446,21 +465,19 @@ class MeetingRecorder(QObject):
                 if tail or code else t("Recording too short, speak for at least 0.3 s")
             )
             return
-        if (self._split_inputs and frames >= RATE * 10
-                and self._mic_zero_frames / frames > 0.5):
-            empty = round(self._mic_zero_frames / frames * 100)
-            self._drop_log()
-            try:
-                os.unlink(self._path)
-            except OSError:
-                pass
-            self.failed.emit(t(
-                "The macOS microphone stopped delivering audio ({percent}% was "
-                "empty). The unusable recording was discarded; reconnect the "
-                "device and try again.", percent=empty,
-            ))
-            return
         self._drop_log()
+        # A microphone that handed nothing over costs the left channel, and the
+        # recording is kept anyway: the right one is everyone else, and an hour
+        # of them is worth more than an empty channel costs. Only the split
+        # capture can starve a device this way; one ffmpeg reading both cannot.
+        if (self._split_inputs and frames >= RATE * QUIET_MIC_SECONDS
+                and self._mic_zero_frames / frames > QUIET_MIC_SHARE):
+            self.warned.emit(t(
+                "The microphone handed over almost nothing ({percent}% of the "
+                "recording was empty), so your own side of the meeting will be "
+                "mostly missing. Check the device before the next one.",
+                percent=round(self._mic_zero_frames / frames * 100),
+            ))
         self.stopped.emit(self._path, frames / RATE)
 
     def _drop_log(self):
@@ -496,20 +513,40 @@ def stereo_levels(chunk):
 
 
 def interleave_mono(left, right):
-    """Two equally long mono-s16 buffers into one stereo-s16 buffer."""
-    left_samples = array.array("h")
-    right_samples = array.array("h")
-    left_samples.frombytes(left[:len(left) - len(left) % SAMPLE_WIDTH])
-    right_samples.frombytes(right[:len(right) - len(right) % SAMPLE_WIDTH])
+    """Two mono-s16 buffers into one stereo-s16 buffer, the shorter one setting
+    the length."""
+    left_samples, right_samples = _samples(left), _samples(right)
     frames = min(len(left_samples), len(right_samples))
-    stereo_samples = array.array("h")
-    stereo_samples.extend(
-        sample for pair in zip(left_samples[:frames], right_samples[:frames])
-        for sample in pair
-    )
+    stereo_samples = array.array("h", bytes(frames * 2 * SAMPLE_WIDTH))
+    stereo_samples[0::2] = left_samples[:frames]
+    stereo_samples[1::2] = right_samples[:frames]
     return stereo_samples.tobytes()
 
 
+def _read_blocks(stream, blocks, size):
+    """One stream's blocks onto its queue, ending with the empty one.
+
+    A queue that stays full is the pump having given up on this recording, and
+    then there is nobody left to hand anything to.
+    """
+    try:
+        while True:
+            block = _read_exact(stream, size)
+            blocks.put(block, timeout=STALL_SECONDS)
+            if not block:
+                return
+    except (OSError, ValueError, queue.Full):
+        pass
+
+
+def _next_block(blocks):
+    """The next block of a stream, empty once it ends or falls silent."""
+    try:
+        return blocks.get(timeout=STALL_SECONDS)
+    except queue.Empty:
+        return b""
+
+
 def _read_exact(stream, size):
     """Read one meter-sized block, tolerating short unbuffered pipe reads."""
     out = bytearray()
@@ -522,9 +559,13 @@ def _read_exact(stream, size):
 
 
 def _zero_samples(chunk):
+    return _samples(chunk).count(0)
+
+
+def _samples(chunk):
     samples = array.array("h")
     samples.frombytes(chunk[:len(chunk) - len(chunk) % SAMPLE_WIDTH])
-    return sum(sample == 0 for sample in samples)
+    return samples
 
 
 def _peak(samples):
@@ -535,8 +576,9 @@ def _peak(samples):
 
 # --- the sound system, one group per machine -------------------------------
 
-# Both meeting commands merge the same way: each input down to mono at our own
-# rate, then the two of them into the left and right of one stream.
+# How the one PulseAudio process merges: each input down to mono at our own
+# rate, then the two of them into the left and right of one stream. A Mac does
+# the first half per process and the second half itself, in interleave_mono().
 MERGE_FILTER = (
     f"[0:a]aresample={RATE}:async=1,aformat=sample_fmts=s16:channel_layouts=mono[m];"
     f"[1:a]aresample={RATE}:async=1,aformat=sample_fmts=s16:channel_layouts=mono[s];"
@@ -600,13 +642,14 @@ def _pw_record_raw_option():
 
 
 def _pulse_meeting(mic_target, system_target):
-    return [
+    """One process for both devices: PulseAudio keeps them aligned itself."""
+    return [[
         "ffmpeg", "-hide_banner", "-nostdin", "-loglevel", "error",
         "-f", "pulse", "-thread_queue_size", "4096", "-i", mic_target or "default",
         "-f", "pulse", "-thread_queue_size", "4096", "-i", system_target,
         "-filter_complex", MERGE_FILTER, "-map", "[out]",
         "-f", "s16le", "-ar", str(RATE), "-",
-    ]
+    ]]
 
 
 def _pactl_sources():
@@ -674,8 +717,20 @@ def _avfoundation_record(target):
     ]
 
 
-def _avfoundation_meeting_capture(target):
-    target = _resolve_avfoundation_target(target)
+def _avfoundation_meeting(mic_target, system_target):
+    """A process per device, both names read off the same device listing.
+
+    Asking ffmpeg what is plugged in costs a process of its own, and a listing
+    taken twice could renumber in between: the two targets have to be resolved
+    against the same one to name the same machine the user picked from.
+    """
+    inputs = _avfoundation_inputs()
+    return [_avfoundation_meeting_capture(mic_target, inputs),
+            _avfoundation_meeting_capture(system_target, inputs)]
+
+
+def _avfoundation_meeting_capture(target, inputs=None):
+    target = _resolve_avfoundation_target(target, inputs)
     return [
         "ffmpeg", "-hide_banner", "-nostdin", "-loglevel", "error",
         "-thread_queue_size", "4096",
@@ -725,7 +780,7 @@ def _avfoundation_named_inputs():
             for _index, description in _avfoundation_inputs()]
 
 
-def _resolve_avfoundation_target(target):
+def _resolve_avfoundation_target(target, inputs=None):
     """Resolve a stored device name to its current, positional ffmpeg index."""
     if not target or target == "default":
         return "default"
@@ -734,7 +789,9 @@ def _resolve_avfoundation_target(target):
             "The saved macOS audio device uses an old numeric index. Open "
             "Settings and select the device again before recording."
         ))
-    matches = [index for index, description in _avfoundation_inputs()
+    if inputs is None:
+        inputs = _avfoundation_inputs()
+    matches = [index for index, description in inputs
                if description == target]
     if not matches:
         raise AudioDeviceError(t(
@@ -758,9 +815,10 @@ def _avfoundation_default_output():
 
 Sound = collections.namedtuple(
     "Sound",
-    # How to capture one source and two at once, the two device lists, which
-    # device a meeting records the far side from, and what to say when the
-    # programs for any of it are not installed.
+    # How to capture one source and how to capture two at once, that one as the
+    # list of processes it takes, the two device lists, which device a meeting
+    # records the far side from, and what to say when the programs for any of
+    # it are not installed.
     "record meeting inputs outputs default_output missing",
 )
 
@@ -775,7 +833,7 @@ PULSE = Sound(
 
 COREAUDIO = Sound(
     record=_avfoundation_record,
-    meeting=None,  # two separate capture processes; see meeting_commands()
+    meeting=_avfoundation_meeting,
     inputs=_avfoundation_named_inputs,
     # Every macOS capture device is offered as the far side of a meeting, the
     # loopback driver among them: there is no way to tell them apart, and an

dikte.py

@@ -123,6 +123,7 @@ class Dikte:
         self.meeting_recorder.levels.connect(self._on_meeting_levels)
         self.meeting_recorder.stopped.connect(self._on_meeting_recorded)
         self.meeting_recorder.died.connect(self._on_meeting_died)
+        self.meeting_recorder.warned.connect(self._on_meeting_warning)
         self.meeting_recorder.failed.connect(self._on_meeting_error)
         self.meetings.progress.connect(self._on_meeting_progress)
         self.meetings.finished.connect(self._on_meeting_finished)
@@ -721,6 +722,11 @@ class Dikte:
         self._settle(MEETING, {"ok": False, "error": message})
         self._on_error(message)
 
+    def _on_meeting_warning(self, message):
+        """It was recorded and it is being written up, but read this first."""
+        self.tray.showMessage("Dikte", message,
+                              QSystemTrayIcon.MessageIcon.Warning, 12000)
+
     def _on_meeting_died(self):
         if self.meeting_state != M_RECORDING:
             return

i18n.py

@@ -566,10 +566,12 @@ TR = {
     "duplicate or choose a different device.":
         "Birden fazla macOS ses aygıtının adı {device}. Aynı adlı aygıtlardan "
         "birini çıkar ya da başka bir aygıt seç.",
-    "The macOS microphone stopped delivering audio ({percent}% was empty). The "
-    "unusable recording was discarded; reconnect the device and try again.":
-        "macOS mikrofonu ses iletmeyi durdurdu (kaydın %{percent} kadarı boştu). "
-        "Kullanılamaz kayıt silindi; aygıtı yeniden bağlayıp tekrar dene.",
+    "The microphone handed over almost nothing ({percent}% of the recording was "
+    "empty), so your own side of the meeting will be mostly missing. Check the "
+    "device before the next one.":
+        "Mikrofon neredeyse hiçbir şey iletmedi (kaydın %{percent} kadarı boştu), "
+        "toplantının senin tarafın büyük ölçüde eksik olacak. Bir sonrakinden "
+        "önce aygıtı kontrol et.",
     "Transcribing {side}: {index}/{count}…":
         "{side} yazıya çevriliyor: {index}/{count}…",
     "you": "sen",

tests/test_audio.py

@@ -16,6 +16,7 @@ import json
 import os
 import subprocess
 import sys
+import threading
 import unittest
 import wave
 from unittest import mock
@@ -251,6 +252,34 @@ class FakeProcess:
         self._alive = False
 
 
+class StalledProcess(FakeProcess):
+    """A capture that hands over a buffer and then stops answering at all.
+
+    Not the same thing as one that ends: the device is still there and the pipe
+    is still open, and a read of it never comes back.
+    """
+
+    def __init__(self, data):
+        super().__init__(data)
+        self.stdout = _StalledStream(data)
+
+
+class _StalledStream:
+    def __init__(self, data):
+        self._data = io.BytesIO(data)
+        self._released = threading.Event()
+
+    def read(self, size):
+        chunk = self._data.read(size)
+        if chunk:
+            return chunk
+        self._released.wait()
+        return b""
+
+    def release(self):
+        self._released.set()
+
+
 class RecordingCommand(OnLinux, DikteTest):
     """Which program captures the microphone, and how it is asked to."""
 
@@ -486,8 +515,10 @@ class MeetingCommands(unittest.TestCase):
 
     def commands(self, platform, mic="", system="them"):
         with mock.patch.object(sys, "platform", platform), \
-                mock.patch.object(audio, "_resolve_avfoundation_target",
-                                  side_effect=lambda target: target or "default"):
+                mock.patch.object(audio, "_avfoundation_inputs", return_value=[]), \
+                mock.patch.object(
+                    audio, "_resolve_avfoundation_target",
+                    side_effect=lambda target, inputs=None: target or "default"):
             return audio.meeting_commands(mic, system)
 
     def test_linux_reads_both_through_pulse(self):
@@ -529,23 +560,41 @@ class MeetingCommands(unittest.TestCase):
                 for command in self.commands(platform):
                     self.assertIn("-nostdin", command)
 
+    def test_both_mac_devices_are_read_off_one_listing(self):
+        """Asking twice costs an ffmpeg run, and the second answer could have
+        renumbered between the two."""
+        with mock.patch.object(sys, "platform", "darwin"), \
+                mock.patch.object(audio, "_avfoundation_inputs",
+                                  return_value=[("0", "mine"),
+                                                ("1", "them")]) as inputs:
+            audio.meeting_commands("mine", "them")
+        inputs.assert_called_once_with()
+
 
 class MacMeetingRecorder(OnMacOS, DikteTest):
+    # Whichever way the machine has them ordered, a name is what is saved and
+    # the index it happens to hold now is what ffmpeg is given.
+    DEVICES = [("0", "External Headset"), ("1", "BlackHole 2ch"),
+               ("2", "MacBook Pro Microphone")]
+
+    def devices(self):
+        return mock.patch.object(audio, "_avfoundation_inputs",
+                                 return_value=self.DEVICES)
+
     def record(self, mine, theirs):
         path = str(self.path("meeting.wav"))
         recorder = audio.MeetingRecorder()
-        stopped, failed = [], []
+        stopped, failed, warnings = [], [], []
         recorder.stopped.connect(lambda *args: stopped.append(args))
         recorder.failed.connect(failed.append)
+        recorder.warned.connect(warnings.append)
         processes = [FakeProcess(mine), FakeProcess(theirs)]
-        with only_these_tools("ffmpeg"), \
-                mock.patch.object(audio, "_resolve_avfoundation_target",
-                                  side_effect=("2", "1")), \
+        with only_these_tools("ffmpeg"), self.devices(), \
                 mock.patch.object(subprocess, "Popen", side_effect=processes) as popen:
             recorder.start(path, "MacBook Pro Microphone", "BlackHole 2ch")
             recorder._thread.join(timeout=5)
             recorder.stop()
-        return path, recorder, stopped, failed, processes, popen
+        return path, warnings, stopped, failed, processes, popen
 
     def test_the_two_capture_processes_become_one_stereo_file(self):
         path, _, stopped, failed, _, _ = self.record(
@@ -567,14 +616,42 @@ class MacMeetingRecorder(OnMacOS, DikteTest):
         self.assertIn(":2", commands[0])
         self.assertIn(":1", commands[1])
 
-    def test_an_unusable_mostly_empty_microphone_is_not_transcribed(self):
-        path, _, stopped, failed, _, _ = self.record(
+    def test_a_mostly_empty_microphone_is_said_out_loud_and_still_kept(self):
+        """Half the file is everyone else, and an hour of them is worth more
+        than the empty channel costs."""
+        path, warnings, stopped, failed, _, _ = self.record(
             silence(11.0), tone(11.0)
         )
-        self.assertEqual(stopped, [])
-        self.assertEqual(len(failed), 1)
-        self.assertIn("empty", failed[0])
-        self.assertFalse(os.path.exists(path))
+        self.assertEqual(failed, [])
+        self.assertEqual(len(stopped), 1)
+        self.assertTrue(os.path.exists(path))
+        self.assertIn("empty", warnings[0])
+
+    def test_a_microphone_that_was_merely_quiet_is_not_complained_about(self):
+        _, warnings, stopped, _, _, _ = self.record(tone(11.0), tone(11.0))
+        self.assertEqual(warnings, [])
+        self.assertEqual(len(stopped), 1)
+
+    def test_a_capture_that_falls_silent_ends_the_meeting_rather_than_hanging(self):
+        """One thread taking turns on both pipes would sit on the dead read
+        until somebody noticed, an hour later."""
+        path = str(self.path("meeting.wav"))
+        recorder = audio.MeetingRecorder()
+        stopped = []
+        recorder.stopped.connect(lambda *args: stopped.append(args))
+        mine, theirs = StalledProcess(tone(0.512)), FakeProcess(tone(30.0))
+        with only_these_tools("ffmpeg"), self.devices(), \
+                mock.patch.object(audio, "STALL_SECONDS", 0.2), \
+                mock.patch.object(subprocess, "Popen", side_effect=(mine, theirs)):
+            try:
+                recorder.start(path, "MacBook Pro Microphone", "BlackHole 2ch")
+                recorder._thread.join(timeout=2)
+                self.assertFalse(recorder.active)
+                recorder.stop()
+            finally:
+                mine.stdout.release()
+        self.assertAlmostEqual(stopped[0][1], 0.512, places=3)
+        self.assertTrue(os.path.exists(path))
 
     def test_stopping_ends_both_capture_processes(self):
         _, _, _, _, processes, _ = self.record(tone(0.5), tone(0.5))
@@ -584,22 +661,21 @@ class MacMeetingRecorder(OnMacOS, DikteTest):
         recorder = audio.MeetingRecorder()
         failed = []
         recorder.failed.connect(failed.append)
-        with only_these_tools("ffmpeg"), \
-                mock.patch.object(audio, "_avfoundation_inputs", return_value=[]), \
+        with only_these_tools("ffmpeg"), self.devices(), \
                 mock.patch.object(subprocess, "Popen") as popen:
             recorder.start(str(self.path("meeting.wav")), "2", "1")
         popen.assert_not_called()
         self.assertIn("old numeric index", failed[0])
 
     def test_a_second_capture_process_that_cannot_start_cleans_up_the_first(self):
+        """A Mac left holding an open AVFoundation session records nothing
+        else until it is let go."""
         path = str(self.path("meeting.wav"))
         recorder = audio.MeetingRecorder()
         failed = []
         recorder.failed.connect(failed.append)
         first = FakeProcess(tone(1.0))
-        with only_these_tools("ffmpeg"), \
-                mock.patch.object(audio, "_resolve_avfoundation_target",
-                                  side_effect=("2", "1")), \
+        with only_these_tools("ffmpeg"), self.devices(), \
                 mock.patch.object(subprocess, "Popen",
                                   side_effect=(first, OSError("refused"))):
             recorder.start(path, "MacBook Pro Microphone", "BlackHole 2ch")