diff --git a/audio.py b/audio.py index 55beb50..bf15ff6 100644 --- a/audio.py +++ b/audio.py @@ -9,7 +9,8 @@ over an hour. 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. +chooser picks between them. macOS uses one ffmpeg process per AVFoundation +device: two AVFoundation sessions in one process silently starve one another. """ import array @@ -63,7 +64,11 @@ class Recorder(QObject): def start(self, target="", max_seconds=300): if self.active: return - cmd = recording_command(target) + try: + cmd = recording_command(target) + except AudioDeviceError as exc: + self.failed.emit(str(exc)) + return if not cmd: self.failed.emit(t(sound().missing)) return @@ -185,9 +190,24 @@ def recording_command(target=""): return sound().record(target) -def meeting_command(mic_target, system_target): - """One ffmpeg reading both devices and merging them into two channels.""" - return sound().meeting(mic_target, system_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. + """ + if sound() is COREAUDIO: + return [ + _avfoundation_meeting_capture(mic_target), + _avfoundation_meeting_capture(system_target), + ] + return [sound().meeting(mic_target, system_target)] + + +class AudioDeviceError(RuntimeError): + """A saved capture device can no longer be selected safely.""" class MeetingRecorder(QObject): @@ -208,11 +228,14 @@ class MeetingRecorder(QObject): def __init__(self, parent=None): super().__init__(parent) self._proc = None + self._procs = [] self._thread = None self._wav = None - self._log = None + self._logs = [] self._path = "" self._frames = 0 + self._mic_zero_frames = 0 + self._split_inputs = False self._cancelled = False self._stopping = False self._lock = threading.Lock() @@ -234,7 +257,11 @@ class MeetingRecorder(QObject): "Pick one in Settings → Meeting.")) return - cmd = meeting_command(mic_target, system_target) + try: + commands = meeting_commands(mic_target, system_target) + except AudioDeviceError as exc: + self.failed.emit(str(exc)) + return try: os.makedirs(os.path.dirname(path), exist_ok=True) @@ -244,11 +271,17 @@ class MeetingRecorder(QObject): self._wav.setframerate(RATE) # ffmpeg keeps talking to stderr for as long as it runs; a pipe # nobody drains would eventually block it, so it writes to a file. - self._log = tempfile.TemporaryFile() - self._proc = subprocess.Popen( - cmd, stdout=subprocess.PIPE, stderr=self._log, bufsize=0 - ) + self._logs = [tempfile.TemporaryFile() for _ in commands] + self._procs = [] + for command, log in zip(commands, self._logs): + self._procs.append(subprocess.Popen( + command, stdout=subprocess.PIPE, stderr=log, bufsize=0 + )) + self._proc = self._procs[0] except (OSError, wave.Error) as exc: + self._terminate_processes() + self._proc = None + self._procs = [] self._close_file() self._drop_log() try: @@ -260,6 +293,8 @@ class MeetingRecorder(QObject): self._path = path self._frames = 0 + self._mic_zero_frames = 0 + self._split_inputs = len(self._procs) == 2 self._cancelled = False self._stopping = False self._max_frames = int(max_seconds * RATE) @@ -267,38 +302,75 @@ class MeetingRecorder(QObject): self._thread.start() def _pump(self): - stdout = self._proc.stdout - block = CHUNK_FRAMES * SAMPLE_WIDTH * 2 - try: - while True: - chunk = stdout.read(block) - if not chunk: - break - mine, theirs = stereo_levels(chunk) - with self._lock: - if self._wav is None: - break - self._wav.writeframes(chunk) - self._frames += len(chunk) // (SAMPLE_WIDTH * 2) - too_long = self._frames >= self._max_frames - self.levels.emit(mine, theirs) - if too_long: - self._terminate() - break - except (OSError, ValueError, wave.Error): - pass + if self._split_inputs: + self._pump_split() + else: + self._pump_merged() # Nobody asked it to end: the sound device went away, or ffmpeg fell # over. An hour into a meeting that has to be said out loud rather than # discovered afterwards. if not self._stopping: self.died.emit() + def _pump_merged(self): + stdout = self._procs[0].stdout + block = CHUNK_FRAMES * SAMPLE_WIDTH * 2 + try: + while True: + chunk = stdout.read(block) + if not chunk: + break + if not self._write_chunk(chunk): + break + except (OSError, ValueError, wave.Error): + pass + + def _pump_split(self): + left = self._procs[0].stdout + right = self._procs[1].stdout + block = CHUNK_FRAMES * SAMPLE_WIDTH + try: + while True: + mine = _read_exact(left, block) + theirs = _read_exact(right, block) + if not mine or not theirs: + break + frames = min(len(mine), len(theirs)) // SAMPLE_WIDTH + mine = mine[:frames * SAMPLE_WIDTH] + theirs = theirs[:frames * SAMPLE_WIDTH] + self._mic_zero_frames += _zero_samples(mine) + if not self._write_chunk(interleave_mono(mine, theirs)): + break + except (OSError, ValueError, wave.Error): + pass + + def _write_chunk(self, chunk): + mine, theirs = stereo_levels(chunk) + with self._lock: + if self._wav is None: + return False + self._wav.writeframes(chunk) + self._frames += len(chunk) // (SAMPLE_WIDTH * 2) + too_long = self._frames >= self._max_frames + self.levels.emit(mine, theirs) + if too_long: + self._terminate() + return False + return True + def _terminate(self): self._stopping = True - proc = self._proc - if proc and proc.poll() is None: + self._terminate_processes() + + def _terminate_processes(self): + running = [proc for proc in self._procs if proc.poll() is None] + for proc in running: try: proc.send_signal(signal.SIGINT) + except OSError: + pass + for proc in running: + try: proc.wait(timeout=2) except (subprocess.TimeoutExpired, OSError): try: @@ -316,23 +388,27 @@ class MeetingRecorder(QObject): pass def _error_tail(self): - if self._log is None: - return "" - try: - self._log.seek(0) - text = self._log.read().decode("utf-8", "replace").strip() - except OSError: - return "" - lines = [line for line in text.splitlines() if line.strip()] - return lines[-1] if lines else "" + tails = [] + for log in self._logs: + try: + log.seek(0) + text = log.read().decode("utf-8", "replace").strip() + except OSError: + continue + lines = [line for line in text.splitlines() if line.strip()] + if lines: + tails.append(lines[-1]) + return " | ".join(tails) def _finish_process(self): self._terminate() if self._thread: self._thread.join(timeout=3) self._thread = None - code = self._proc.poll() if self._proc else 0 + 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 @@ -370,16 +446,30 @@ 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() self.stopped.emit(self._path, frames / RATE) def _drop_log(self): - if self._log is not None: + for log in self._logs: try: - self._log.close() + log.close() except OSError: pass - self._log = None + self._logs = [] def chunk_levels(chunk): @@ -405,6 +495,38 @@ def stereo_levels(chunk): return _peak(left), _peak(right) +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]) + 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 + ) + return stereo_samples.tobytes() + + +def _read_exact(stream, size): + """Read one meter-sized block, tolerating short unbuffered pipe reads.""" + out = bytearray() + while len(out) < size: + chunk = stream.read(size - len(out)) + if not chunk: + break + out.extend(chunk) + return bytes(out) + + +def _zero_samples(chunk): + samples = array.array("h") + samples.frombytes(chunk[:len(chunk) - len(chunk) % SAMPLE_WIDTH]) + return sum(sample == 0 for sample in samples) + + def _peak(samples): if not samples: return 0.0 @@ -542,6 +664,7 @@ LOOPBACK_DEVICES = ("blackhole", "loopback", "soundflower") def _avfoundation_record(target): if not shutil.which("ffmpeg"): return [] + target = _resolve_avfoundation_target(target) return [ "ffmpeg", "-hide_banner", "-nostdin", "-loglevel", "error", # AVFoundation names an input "video:audio", so the empty half in front @@ -551,15 +674,15 @@ def _avfoundation_record(target): ] -def _avfoundation_meeting(mic_target, system_target): +def _avfoundation_meeting_capture(target): + target = _resolve_avfoundation_target(target) return [ "ffmpeg", "-hide_banner", "-nostdin", "-loglevel", "error", "-thread_queue_size", "4096", - "-f", "avfoundation", "-i", f":{mic_target or 'default'}", - "-thread_queue_size", "4096", - "-f", "avfoundation", "-i", f":{system_target}", - "-filter_complex", MERGE_FILTER, "-map", "[out]", - "-f", "s16le", "-ar", str(RATE), "-", + "-f", "avfoundation", "-i", f":{target or 'default'}", + "-af", (f"aresample={RATE}:async=1:first_pts=0," + "aformat=sample_fmts=s16:channel_layouts=mono"), + "-f", "s16le", "-ar", str(RATE), "-ac", "1", "-", ] @@ -596,10 +719,40 @@ def _avfoundation_inputs(): return devices +def _avfoundation_named_inputs(): + """Stable settings values: the name is saved, never the moving index.""" + return [(description, description) + for _index, description in _avfoundation_inputs()] + + +def _resolve_avfoundation_target(target): + """Resolve a stored device name to its current, positional ffmpeg index.""" + if not target or target == "default": + return "default" + if str(target).isdigit(): + raise AudioDeviceError(t( + "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 description == target] + if not matches: + raise AudioDeviceError(t( + "The saved macOS audio device is no longer connected: {device}. " + "Open Settings and select another device.", device=target, + )) + if len(matches) > 1: + raise AudioDeviceError(t( + "More than one macOS audio device is named {device}. Disconnect the " + "duplicate or choose a different device.", device=target, + )) + return matches[0] + + def _avfoundation_default_output(): - for name, description in _avfoundation_inputs(): + for _index, description in _avfoundation_inputs(): if any(word in description.lower() for word in LOOPBACK_DEVICES): - return name + return description return "" @@ -622,12 +775,12 @@ PULSE = Sound( COREAUDIO = Sound( record=_avfoundation_record, - meeting=_avfoundation_meeting, - inputs=_avfoundation_inputs, + meeting=None, # two separate capture processes; see meeting_commands() + 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 # empty list would leave nothing to pick. - outputs=_avfoundation_inputs, + outputs=_avfoundation_named_inputs, default_output=_avfoundation_default_output, missing="ffmpeg not found. Install it with: brew install ffmpeg", ) diff --git a/i18n.py b/i18n.py index 436f917..52ab91c 100644 --- a/i18n.py +++ b/i18n.py @@ -554,6 +554,22 @@ TR = { "Hangi ses çıkışının kaydedileceği anlaşılamadı. Ayarlar → Toplantı " "sekmesinden seç.", "Nothing was recorded: {error}": "Hiçbir şey kaydedilmedi: {error}", + "The saved macOS audio device uses an old numeric index. Open Settings and " + "select the device again before recording.": + "Kayıtlı macOS ses aygıtı eski bir sayısal indeks kullanıyor. Kayıttan " + "önce Ayarlar'ı açıp aygıtı yeniden seç.", + "The saved macOS audio device is no longer connected: {device}. Open " + "Settings and select another device.": + "Kayıtlı macOS ses aygıtı artık bağlı değil: {device}. Ayarlar'ı açıp " + "başka bir aygıt seç.", + "More than one macOS audio device is named {device}. Disconnect the " + "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.", "Transcribing {side}: {index}/{count}…": "{side} yazıya çevriliyor: {index}/{count}…", "you": "sen", diff --git a/tests/test_audio.py b/tests/test_audio.py index 22d4f2a..7a6a09e 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -100,6 +100,22 @@ class StereoLevels(unittest.TestCase): self.assertAlmostEqual(left, 0.25, places=3) self.assertAlmostEqual(right, 0.5, places=3) + def test_two_mono_streams_are_interleaved_left_then_right(self): + self.assertEqual( + list(array.array("h", audio.interleave_mono( + pcm([100, 200, 300]), pcm([-100, -200, -300]) + ))), + [100, -100, 200, -200, 300, -300], + ) + + def test_interleaving_stops_at_the_shorter_stream(self): + self.assertEqual( + list(array.array("h", audio.interleave_mono( + pcm([100, 200]), pcm([-100]) + ))), + [100, -100], + ) + class WriteWav(DikteTest): def test_the_header_says_what_the_recorder_captured(self): @@ -465,42 +481,131 @@ class RecorderChain(OnLinux, DikteTest): self.assertFalse(recorder.active) -class MeetingCommand(unittest.TestCase): - """One process reading both devices, because two would drift apart.""" +class MeetingCommands(unittest.TestCase): + """Pulse can share a process; AVFoundation sessions cannot.""" - def command(self, platform, mic="", system="them"): - with mock.patch.object(sys, "platform", platform): - return audio.meeting_command(mic, system) + 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"): + return audio.meeting_commands(mic, system) def test_linux_reads_both_through_pulse(self): - cmd = self.command("linux", mic="mine") + commands = self.commands("linux", mic="mine") + self.assertEqual(len(commands), 1) + cmd = commands[0] self.assertEqual(cmd.count("pulse"), 2) self.assertEqual(cmd[cmd.index("mine") - 1], "-i") self.assertEqual(cmd[cmd.index("them") - 1], "-i") - def test_a_mac_reads_both_through_avfoundation(self): - cmd = self.command("darwin", mic="1") - self.assertEqual(cmd.count("avfoundation"), 2) - self.assertIn(":1", cmd) - self.assertIn(":them", cmd) + def test_a_mac_gives_each_avfoundation_device_its_own_process(self): + commands = self.commands("darwin", mic="mine") + self.assertEqual(len(commands), 2) + self.assertTrue(all(command.count("avfoundation") == 1 + for command in commands)) + self.assertIn(":mine", commands[0]) + self.assertIn(":them", commands[1]) def test_no_microphone_named_means_the_default_one(self): - self.assertIn("default", self.command("linux")) - self.assertIn(":default", self.command("darwin")) + self.assertIn("default", self.commands("linux")[0]) + self.assertIn(":default", self.commands("darwin")[0]) - def test_both_merge_the_two_into_one_stereo_stream(self): - for platform in ("linux", "darwin"): - with self.subTest(platform=platform): - cmd = self.command(platform) - self.assertIn(audio.MERGE_FILTER, cmd) - self.assertEqual(cmd[cmd.index("-map") + 1], "[out]") - self.assertEqual(cmd[cmd.index("-f", cmd.index("-map")) + 1], "s16le") + def test_pulse_merges_the_two_into_one_stereo_stream(self): + cmd = self.commands("linux")[0] + self.assertIn(audio.MERGE_FILTER, cmd) + self.assertEqual(cmd[cmd.index("-map") + 1], "[out]") + self.assertEqual(cmd[cmd.index("-f", cmd.index("-map")) + 1], "s16le") + + def test_each_mac_process_produces_clock_corrected_mono_pcm(self): + for cmd in self.commands("darwin"): + self.assertIn("first_pts=0", cmd[cmd.index("-af") + 1]) + self.assertEqual(cmd[cmd.index("-ac") + 1], "1") + self.assertEqual(cmd[-2:], ["1", "-"]) def test_neither_lets_ffmpeg_read_the_terminal(self): """It shares stdin with Dikte, and would eat a keypress meant for it.""" for platform in ("linux", "darwin"): with self.subTest(platform=platform): - self.assertIn("-nostdin", self.command(platform)) + for command in self.commands(platform): + self.assertIn("-nostdin", command) + + +class MacMeetingRecorder(OnMacOS, DikteTest): + def record(self, mine, theirs): + path = str(self.path("meeting.wav")) + recorder = audio.MeetingRecorder() + stopped, failed = [], [] + recorder.stopped.connect(lambda *args: stopped.append(args)) + recorder.failed.connect(failed.append) + processes = [FakeProcess(mine), FakeProcess(theirs)] + with only_these_tools("ffmpeg"), \ + mock.patch.object(audio, "_resolve_avfoundation_target", + side_effect=("2", "1")), \ + 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 + + def test_the_two_capture_processes_become_one_stereo_file(self): + path, _, stopped, failed, _, _ = self.record( + tone(1.0, freq=440), tone(1.0, freq=880) + ) + self.assertEqual(failed, []) + self.assertEqual(len(stopped), 1) + with contextlib.closing(wave.open(path, "rb")) as wav: + self.assertEqual(wav.getnchannels(), 2) + self.assertEqual(wav.getframerate(), audio.RATE) + self.assertEqual(wav.getnframes(), audio.RATE) + + def test_each_avfoundation_device_is_opened_by_a_different_process(self): + _, _, _, _, _, popen = self.record(tone(0.5), tone(0.5)) + commands = [call.args[0] for call in popen.call_args_list] + self.assertEqual(len(commands), 2) + self.assertTrue(all(command.count("avfoundation") == 1 + for command in commands)) + 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( + 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)) + + def test_stopping_ends_both_capture_processes(self): + _, _, _, _, processes, _ = self.record(tone(0.5), tone(0.5)) + self.assertTrue(all(process.signals for process in processes)) + + def test_a_legacy_numeric_target_fails_before_recording(self): + recorder = audio.MeetingRecorder() + failed = [] + recorder.failed.connect(failed.append) + with only_these_tools("ffmpeg"), \ + mock.patch.object(audio, "_avfoundation_inputs", return_value=[]), \ + 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): + 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")), \ + mock.patch.object(subprocess, "Popen", + side_effect=(first, OSError("refused"))): + recorder.start(path, "MacBook Pro Microphone", "BlackHole 2ch") + self.assertTrue(first.signals) + self.assertIn("refused", failed[0]) + self.assertFalse(os.path.exists(path)) class MacDevices(OnMacOS, DikteTest): @@ -527,12 +632,13 @@ class MacDevices(OnMacOS, DikteTest): def test_the_audio_half_of_the_listing_is_the_only_half_read(self): with self.listing(): self.assertEqual(audio.list_sources(), - [("0", "MacBook Pro Microphone"), ("1", "BlackHole 2ch")]) + [("MacBook Pro Microphone", "MacBook Pro Microphone"), + ("BlackHole 2ch", "BlackHole 2ch")]) - def test_the_index_is_what_ffmpeg_is_given_and_the_name_what_is_shown(self): + def test_the_name_is_both_saved_and_shown(self): with self.listing(): name, description = audio.list_sources()[1] - self.assertEqual(name, "1") + self.assertEqual(name, "BlackHole 2ch") self.assertIn("BlackHole", description) def test_no_ffmpeg_installed(self): @@ -557,7 +663,7 @@ class MacDevices(OnMacOS, DikteTest): def test_the_loopback_driver_is_picked_out_by_name(self): with self.listing(): - self.assertEqual(audio.default_monitor(), "1") + self.assertEqual(audio.default_monitor(), "BlackHole 2ch") def test_the_other_two_drivers_people_install(self): for name in ("Loopback Audio", "Soundflower (2ch)"): @@ -565,13 +671,43 @@ class MacDevices(OnMacOS, DikteTest): listing = ("AVFoundation audio devices:\n" f"[0] Built-in Microphone\n[1] {name}\n") with self.listing(stderr=listing): - self.assertEqual(audio.default_monitor(), "1") + self.assertEqual(audio.default_monitor(), name) def test_a_mac_with_nothing_to_record_the_far_side_from(self): listing = "AVFoundation audio devices:\n[0] MacBook Pro Microphone\n" with self.listing(stderr=listing): self.assertEqual(audio.default_monitor(), "") + def test_a_saved_name_is_resolved_against_the_current_index(self): + with self.listing(): + self.assertEqual(audio._resolve_avfoundation_target("BlackHole 2ch"), "1") + + def test_a_saved_name_follows_the_device_when_an_earlier_one_disappears(self): + listing = ("AVFoundation audio devices:\n" + "[0] BlackHole 2ch\n[1] MacBook Pro Microphone\n") + with self.listing(stderr=listing): + self.assertEqual( + audio._resolve_avfoundation_target("MacBook Pro Microphone"), "1" + ) + + def test_an_old_numeric_setting_is_not_silently_reused(self): + with self.assertRaises(audio.AudioDeviceError) as caught: + audio._resolve_avfoundation_target("1") + self.assertIn("old numeric index", str(caught.exception)) + + def test_a_device_that_went_away_is_said_out_loud(self): + with self.listing(), self.assertRaises(audio.AudioDeviceError) as caught: + audio._resolve_avfoundation_target("USB Microphone") + self.assertIn("no longer connected", str(caught.exception)) + + def test_duplicate_names_are_not_guessed_between(self): + listing = ("AVFoundation audio devices:\n" + "[0] USB Microphone\n[1] USB Microphone\n") + with self.listing(stderr=listing), \ + self.assertRaises(audio.AudioDeviceError) as caught: + audio._resolve_avfoundation_target("USB Microphone") + self.assertIn("More than one", str(caught.exception)) + class MacRecordingCommand(OnMacOS, DikteTest): def test_the_microphone_is_read_through_avfoundation(self): @@ -583,7 +719,11 @@ class MacRecordingCommand(OnMacOS, DikteTest): def test_the_empty_half_in_front_of_the_colon_is_the_missing_picture(self): with only_these_tools("ffmpeg"): self.assertIn(":default", audio.recording_command()) - self.assertIn(":2", audio.recording_command("2")) + listing = "AVFoundation audio devices:\n[2] USB Microphone\n" + completed = FakeCompleted(returncode=1, stderr=listing) + with only_these_tools("ffmpeg"), \ + mock.patch.object(subprocess, "run", return_value=completed): + self.assertIn(":2", audio.recording_command("USB Microphone")) def test_it_captures_the_format_the_rest_of_the_code_expects(self): with only_these_tools("ffmpeg"):