diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..2d9237e --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,70 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + workflow_dispatch: + +concurrency: + group: ci-${{ github.ref }} + cancel-in-progress: true + +jobs: + test-windows: + name: Tests (Windows) + runs-on: windows-latest + + steps: + - name: Checkout repository + uses: actions/checkout@v7 + + - name: Set up Python 3.12 + uses: actions/setup-python@v7 + with: + python-version: "3.12" + cache: pip + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + + - name: Run test suite + run: python run_tests.py + + test-linux: + name: Tests (Linux) + runs-on: ubuntu-latest + + steps: + - name: Checkout repository + uses: actions/checkout@v7 + + - name: Install PortAudio, Tkinter and a virtual display + run: | + sudo apt-get update + sudo apt-get install -y portaudio19-dev python3-tk xvfb + + - name: Set up Python 3.12 + uses: actions/setup-python@v7 + with: + python-version: "3.12" + cache: pip + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + + # Guards the exact regression that shipped two broken release assets: + # ui/app.py used to import pyaudiowpatch unconditionally, which only + # exists on Windows, and Pillow was missing from requirements.txt. Both + # blow up at import time, long before any test asserts anything - and no + # test imports this module unless a display is available. + - name: Import check (non-Windows startup path) + run: python -c "import audio_transcriber.ui.app; print('ui.app imports cleanly')" + + # xvfb so the GUI smoke test actually runs instead of skipping itself. + - name: Run test suite + run: xvfb-run -a python run_tests.py diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 1cfe9a0..570b6dd 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -10,16 +10,58 @@ permissions: contents: write jobs: + # Resolve the release version once so every job and every archive name agrees. + # A tag push uses the tag; a manual dispatch produces a dev version. + version: + name: Resolve Release Version + runs-on: ubuntu-latest + outputs: + version: ${{ steps.resolve.outputs.version }} + steps: + - name: Resolve version from tag or dispatch + id: resolve + run: | + if [ "${{ github.ref_type }}" = "tag" ]; then + VERSION="${{ github.ref_name }}" + else + VERSION="v0.0.0-dev+${GITHUB_SHA::7}" + fi + echo "version=$VERSION" >> "$GITHUB_OUTPUT" + echo "Release version: $VERSION" + + test: + name: Run Test Suite + runs-on: windows-latest + steps: + - name: Checkout repository + uses: actions/checkout@v7 + + - name: Set up Python 3.12 + uses: actions/setup-python@v7 + with: + python-version: "3.12" + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + + - name: Run tests + run: python run_tests.py + build-windows: name: Build Windows x64 Standalone Package runs-on: windows-latest + needs: [version, test] + env: + RELEASE_VERSION: ${{ needs.version.outputs.version }} steps: - name: Checkout repository - uses: actions/checkout@v4 + uses: actions/checkout@v7 - name: Set up Python 3.12 - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: "3.12" @@ -30,11 +72,10 @@ jobs: pip install pyinstaller - name: Build Windows executable & zip package - run: | - python build_release.py + run: python build_release.py - name: Upload Windows Build Artifact - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@v7 with: name: windows-release-package path: dist/AudioTranscriber-*-windows-x64.zip @@ -42,10 +83,13 @@ jobs: build-linux: name: Build Linux x64 Standalone Package runs-on: ubuntu-latest + needs: [version, test] + env: + RELEASE_VERSION: ${{ needs.version.outputs.version }} steps: - name: Checkout repository - uses: actions/checkout@v4 + uses: actions/checkout@v7 - name: Install Linux system audio dependencies & Tkinter run: | @@ -53,34 +97,21 @@ jobs: sudo apt-get install -y portaudio19-dev python3-tk libasound2-dev libjack-jackd2-dev - name: Set up Python 3.12 - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: "3.12" - name: Install Python dependencies run: | python -m pip install --upgrade pip - pip install PyAudio soundfile scipy numpy pyinstaller + pip install -r requirements.txt + pip install pyinstaller - name: Build Linux executable & tar.gz package - run: | - pyinstaller --name AudioTranscriber --windowed --onedir --noconfirm --clean \ - --collect-all soundfile \ - --hidden-import scipy.signal \ - --exclude-module torch \ - --exclude-module torchvision \ - --exclude-module torchaudio \ - --exclude-module pandas \ - --exclude-module sklearn \ - --exclude-module matplotlib \ - --exclude-module pyarrow \ - main.py - - cd dist - tar -czvf AudioTranscriber-v1.0.0-linux-x64.tar.gz AudioTranscriber + run: python build_release.py - name: Upload Linux Build Artifact - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@v7 with: name: linux-release-package path: dist/AudioTranscriber-*-linux-x64.tar.gz @@ -88,93 +119,76 @@ jobs: build-macos: name: Build macOS Standalone Package runs-on: macos-latest + needs: [version, test] + env: + RELEASE_VERSION: ${{ needs.version.outputs.version }} steps: - name: Checkout repository - uses: actions/checkout@v4 + uses: actions/checkout@v7 - name: Install PortAudio via Homebrew - run: | - brew install portaudio + run: brew install portaudio - name: Set up Python 3.12 - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: "3.12" - name: Install Python dependencies run: | python -m pip install --upgrade pip - pip install PyAudio soundfile scipy numpy pyinstaller + pip install -r requirements.txt + pip install pyinstaller - name: Build macOS app bundle & zip package - run: | - pyinstaller --name AudioTranscriber --windowed --onedir --noconfirm --clean \ - --collect-all soundfile \ - --hidden-import scipy.signal \ - --exclude-module torch \ - --exclude-module torchvision \ - --exclude-module torchaudio \ - --exclude-module pandas \ - --exclude-module sklearn \ - --exclude-module matplotlib \ - --exclude-module pyarrow \ - main.py - - cd dist - zip -r AudioTranscriber-v1.0.0-macos-universal.zip AudioTranscriber.app || zip -r AudioTranscriber-v1.0.0-macos-universal.zip AudioTranscriber + run: python build_release.py - name: Upload macOS Build Artifact - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@v7 with: name: macos-release-package - path: dist/AudioTranscriber-*-macos-universal.zip + path: dist/AudioTranscriber-*-macos-*.zip publish-release: name: Publish GitHub Release with Multi-Platform Assets - needs: [build-windows, build-linux, build-macos] + needs: [version, build-windows, build-linux, build-macos] + if: github.ref_type == 'tag' runs-on: ubuntu-latest + env: + RELEASE_VERSION: ${{ needs.version.outputs.version }} steps: - - name: Checkout repository - uses: actions/checkout@v4 - - - name: Download Windows Release Asset - uses: actions/download-artifact@v4 + - name: Download all release assets + uses: actions/download-artifact@v8 with: - name: windows-release-package + pattern: "*-release-package" + merge-multiple: true path: release-assets/ - - name: Download Linux Release Asset - uses: actions/download-artifact@v4 - with: - name: linux-release-package - path: release-assets/ - - - name: Download macOS Release Asset - uses: actions/download-artifact@v4 - with: - name: macos-release-package - path: release-assets/ + - name: List collected assets + run: ls -lh release-assets/ - name: Create or Update GitHub Release - uses: softprops/action-gh-release@v2 + uses: softprops/action-gh-release@v3 with: files: release-assets/* - name: Audio AI Recorder & Transcriber ${{ github.ref_name }} + name: Audio AI Recorder & Transcriber ${{ env.RELEASE_VERSION }} body: | - ## πŸš€ Multi-Platform Standalone Release (${{ github.ref_name }}) + ## πŸš€ Multi-Platform Standalone Release (${{ env.RELEASE_VERSION }}) ### πŸ’» Downloads - - **Windows (x64)**: `AudioTranscriber-${{ github.ref_name }}-windows-x64.zip` - - **Linux (x64)**: `AudioTranscriber-${{ github.ref_name }}-linux-x64.tar.gz` - - **macOS (Intel & Apple Silicon)**: `AudioTranscriber-${{ github.ref_name }}-macos-universal.zip` + - **Windows (x64)**: `AudioTranscriber-${{ env.RELEASE_VERSION }}-windows-x64.zip` + - **Linux (x64)**: `AudioTranscriber-${{ env.RELEASE_VERSION }}-linux-x64.tar.gz` + - **macOS (Apple Silicon)**: `AudioTranscriber-${{ env.RELEASE_VERSION }}-macos-arm64.zip` ### πŸ“– Quick Start 1. Download the appropriate package for your operating system. 2. Extract the archive to any directory. 3. Run `AudioTranscriber.exe` (Windows), `./AudioTranscriber` (Linux), or open `AudioTranscriber.app` (macOS). + > 🍎 **Intel Macs**: this build is Apple Silicon only. Run from source or build locally with `python build_release.py` to get a `macos-x64` archive. + > 🎧 **macOS Note**: System audio recording on macOS requires a virtual loopback device like [BlackHole](https://github.com/ExistentialAudio/BlackHole) (Free & Open Source). env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} diff --git a/.gitignore b/.gitignore index c355576..b79e3ac 100644 --- a/.gitignore +++ b/.gitignore @@ -28,5 +28,3 @@ venv/ .pytest_cache/ .coverage htmlcov/ -tests/ -run_tests.py diff --git a/README.md b/README.md index f0b3a67..3106e06 100644 --- a/README.md +++ b/README.md @@ -34,12 +34,16 @@ An advanced, privacy-focused audio recording and AI transcription suite. It capt No Python installation or terminal setup required! 1. Download the latest release package for your operating system from the **[Releases Page](https://github.com/SecretLUL/Audio-Transcriber/releases)**: - - **Windows**: `AudioTranscriber-v1.0.2-windows-x64.zip` - - **Linux**: `AudioTranscriber-v1.0.2-linux-x64.tar.gz` - - **macOS**: `AudioTranscriber-v1.0.2-macos-universal.zip` + - **Windows (x64)**: `AudioTranscriber--windows-x64.zip` + - **Linux (x64)**: `AudioTranscriber--linux-x64.tar.gz` + - **macOS (Apple Silicon)**: `AudioTranscriber--macos-arm64.zip` 2. Extract the archive to any folder. 3. Launch `AudioTranscriber.exe` (Windows), `./AudioTranscriber` (Linux), or `AudioTranscriber.app` (macOS). +> 🍎 **Intel Macs**: the released macOS build is Apple Silicon only. On an Intel +> Mac, run from source (Option B) or build locally with `python build_release.py`, +> which produces a `macos-x64` archive. + --- ### Option B: Running from Source (For Developers πŸ› οΈ) @@ -94,7 +98,7 @@ Audio-Transcriber/ β”œβ”€β”€ main.py Entry point for python / pythonw launch β”œβ”€β”€ Start-Recorder.vbs Windows double-click launcher β”œβ”€β”€ build_release.py Automated PyInstaller standalone build & zip packaging - β”œβ”€β”€ requirements.txt Core dependencies (pyaudiowpatch, soundfile, scipy, numpy) + β”œβ”€β”€ requirements.txt Core dependencies (pyaudiowpatch, soundfile, scipy, numpy, Pillow) β”œβ”€β”€ LICENSE MIT License (100% FOSS) β”œβ”€β”€ README.md Project documentation β”œβ”€β”€ audio_transcriber/ Main application package @@ -180,13 +184,26 @@ python run_tests.py ## πŸ“¦ Automated Release Build -To build a standalone executable and zip archive locally using PyInstaller: +To build a standalone executable and release archive locally using PyInstaller: ```shell python build_release.py ``` -The output executable and `.zip` package will be saved in the `dist/` directory. +The build runs on Windows, Linux and macOS and names the archive after the +current platform and version, e.g. `AudioTranscriber-v1.2.0-windows-x64.zip`. + +The version is resolved in this order: + +1. An explicit argument β€” `python build_release.py v1.2.0` +2. The `RELEASE_VERSION` environment variable (the CI sets this from the tag) +3. The git tag pointing at `HEAD` +4. `v0.0.0-dev` as a fallback + +Before archiving, the script verifies that every module the app imports at +startup actually made it into the bundle and fails the build otherwise. + +The output executable and archive are saved in the `dist/` directory. --- diff --git a/audio_transcriber/audio/capture.py b/audio_transcriber/audio/capture.py index 24b7c7f..7da4483 100644 --- a/audio_transcriber/audio/capture.py +++ b/audio_transcriber/audio/capture.py @@ -123,8 +123,22 @@ def __init__(self, pa, tmp_dir): self._tracks = {} self._recording = False self._gen_lock = threading.Lock() + # Serialises whole reconfigurations. _gen_lock alone is not enough: + # configure() releases it between stop_streams() and opening the new + # streams, so two rapid device changes could interleave as + # A.stop / B.stop / A.assign / B.assign and leak A's streams. + self._configure_lock = threading.Lock() self.last_error = None + @property + def is_configuring(self): + """True while a reconfiguration is in flight. + + During that window the engine may legitimately have no active track, + which is not the same thing as having no working device. + """ + return self._configure_lock.locked() + # ------------------------------------------------------------------ # Levels for the meters (raw, before gain - see H2) # ------------------------------------------------------------------ @@ -153,6 +167,10 @@ def configure(self, mic_device, sys_device): Must run off the GUI thread (join blocks for up to 2 s). """ + with self._configure_lock: + return self._configure(mic_device, sys_device) + + def _configure(self, mic_device, sys_device): warnings = [] self.stop_streams() diff --git a/audio_transcriber/audio/loader.py b/audio_transcriber/audio/loader.py index 1fe7491..70a7a1d 100644 --- a/audio_transcriber/audio/loader.py +++ b/audio_transcriber/audio/loader.py @@ -29,15 +29,19 @@ def load_audio_file(file_path: str, target_rate: int = dsp.TARGET_RATE) -> np.nd # 1. Try reading directly with soundfile (fast, native) try: data, rate = sf.read(file_path, dtype="float32", always_2d=False) + except Exception: + # soundfile failed or format unsupported (e.g. m4a, aac, wma) -> try FFmpeg fallback + pass + else: + # A readable-but-empty file is a decoding result, not a decoding + # failure. Raising inside the try above sent it down the FFmpeg path + # and reported a misleading "FFmpeg not found" instead. if data.ndim > 1: data = data.mean(axis=1) resampled = dsp.resample(data, rate, target_rate) if len(resampled) == 0: raise TranscriptionError(f"Audio file contains no sample data: {file_path}") return resampled - except Exception: - # soundfile failed or format unsupported (e.g. m4a, aac, wma) -> try FFmpeg fallback - pass # 2. Fallback: FFmpeg conversion to temp WAV return _load_via_ffmpeg(file_path, target_rate) diff --git a/audio_transcriber/paths.py b/audio_transcriber/paths.py index 87d20e2..a7d223a 100644 --- a/audio_transcriber/paths.py +++ b/audio_transcriber/paths.py @@ -35,9 +35,15 @@ def safe_output_name(user_input, default="my_meeting"): """Turn user input into a safe base name without any path component. Blocks path traversal ('..\\..\\windows\\x') and empty names. + + Both separators are stripped on every platform. os.path.basename() follows + the host rules, so on Linux and macOS a backslash is an ordinary character + and a name written on Windows sanitised to something else entirely - the + same settings.json produced a different file name depending on where it + ran. The result is still safe either way, just not the same. """ name = (user_input or "").strip() - name = os.path.basename(name) + name = name.replace("\\", "/").rsplit("/", 1)[-1] name = os.path.splitext(name)[0] # Strip characters Windows does not allow in file names for char in '<>:"/\\|?*': diff --git a/audio_transcriber/pipeline.py b/audio_transcriber/pipeline.py index c149f16..3b1b06b 100644 --- a/audio_transcriber/pipeline.py +++ b/audio_transcriber/pipeline.py @@ -16,10 +16,10 @@ import numpy as np import soundfile as sf -from . import diarize +from . import diarize, paths from .audio import capture, dsp from .events import Failed, Finished, Log, Progress, Status -from .paths import OUT_DIR, TMP_DIR +from .paths import TMP_DIR from .transcribe.base import TranscriptionError from .transcribe.elevenlabs import ElevenLabsBackend from .transcribe.whispercpp import WhisperCppBackend @@ -164,9 +164,6 @@ def _process(self, recording, base_name): with open(txt_path, "w", encoding="utf-8") as handle: handle.write(text + "\n") - with open(txt_path, "w", encoding="utf-8") as handle: - handle.write(text + "\n") - # --- 6. Clean up ------------------------------------------------- if not settings.keep_raw_tracks: for track in (recording.mic, recording.sys): diff --git a/audio_transcriber/transcribe/elevenlabs.py b/audio_transcriber/transcribe/elevenlabs.py index 6184b73..145d0b2 100644 --- a/audio_transcriber/transcribe/elevenlabs.py +++ b/audio_transcriber/transcribe/elevenlabs.py @@ -82,6 +82,7 @@ def __init__(self, api_key, model_id="scribe_v2", diarize=True, self.diarize = diarize self.tag_audio_events = tag_audio_events self._cancelled = False + self._response = None # live response, so cancel() can close it # ------------------------------------------------------------------ def transcribe(self, wav_path, language="de", log=None, track="", @@ -153,6 +154,9 @@ def _post(self, wav_path, language, model_id): try: with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT) as response: + self._response = response + if self._cancelled: + raise TranscriptionError("Transcription was cancelled.") raw = response.read().decode("utf-8", errors="replace") except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace")[:1500] @@ -168,8 +172,11 @@ def _post(self, wav_path, language, model_id): raise TranscriptionError( f"Could not reach ElevenLabs: {exc.reason}") from exc except OSError as exc: + if self._cancelled: + raise TranscriptionError("Transcription was cancelled.") from exc raise TranscriptionError(f"Upload failed: {exc}") from exc finally: + self._response = None body.close() try: @@ -220,7 +227,8 @@ def flush(): buffer = [] word_count = 0 seg_start = None - speaker = "" + seg_end = 0.0 # must reset too, else the next segment + speaker = "" # inherits this one's end time for token in words: kind = token.get("type", "word") @@ -252,4 +260,17 @@ def flush(): return segments def cancel(self): + """Abort a running request. + + There is no clean interrupt for urlopen, so the socket is closed from + underneath the reader: the blocked read raises OSError, which _post + turns back into a cancellation. Without this the worker sat on the + 900 s timeout after the window had already closed. + """ self._cancelled = True + response = self._response + if response is not None: + try: + response.close() + except Exception: + pass diff --git a/audio_transcriber/ui/app.py b/audio_transcriber/ui/app.py index 83dcf4c..c88bb00 100644 --- a/audio_transcriber/ui/app.py +++ b/audio_transcriber/ui/app.py @@ -14,7 +14,13 @@ import tkinter as tk from tkinter import filedialog, messagebox, ttk -import pyaudiowpatch as pyaudio +try: + # Windows: the WASAPI loopback fork. Everywhere else plain PyAudio - the + # app then has no loopback capture, but recording and file upload work. + # capture.py does the same dance when it opens a stream. + import pyaudiowpatch as pyaudio +except ImportError: + import pyaudio from .. import config, paths, pipeline from ..audio import devices as devmod @@ -27,6 +33,11 @@ METER_INTERVAL_MS = 40 +# How long Start waits for an in-flight device reconfiguration before giving up +# and letting the engine report whatever is actually wrong. configure() joins +# its reader threads with a 2 s timeout each, so this leaves room for both. +DEVICE_READY_TIMEOUT_S = 6.0 + class RecorderApp: def __init__(self, root): @@ -46,6 +57,8 @@ def __init__(self, root): self.live_preview = None self.recording_base_name = None self.recording_started_at = None + self._monitor_thread = None + self._start_deadline = 0.0 self._shutting_down = False self._meter_after_id = None self._icon_refs = {} @@ -298,10 +311,10 @@ def _build_options_card(self, parent): W.Switch(options, "Keep raw tracks", self.keep_raw_var).grid( row=0, column=3, sticky="w", padx=(T.MD, 0)) - self.save_btn = W.Button(options, text="Save settings", - kind="ghost", width=150, height=32, - command=self.save_settings) - self.save_btn.grid(row=0, column=4, sticky="e", padx=(T.MD, 0)) + self.save_settings_btn = W.Button(options, text="Save settings", + kind="ghost", width=150, height=32, + command=self.save_settings) + self.save_settings_btn.grid(row=0, column=4, sticky="e", padx=(T.MD, 0)) options.columnconfigure(4, weight=1) @@ -352,9 +365,10 @@ def _build_transcript(self, parent): width=80, height=28, command=lambda: self.transcript.clear()) self.clear_btn.pack(side=tk.RIGHT) - self.save_btn = W.Button(toolbar, text="Save", icon_name="save", kind="quiet", - width=80, height=28, command=self._save_transcript) - self.save_btn.pack(side=tk.RIGHT, padx=(0, T.XS)) + self.save_transcript_btn = W.Button(toolbar, text="Save", icon_name="save", + kind="quiet", width=80, height=28, + command=self._save_transcript) + self.save_transcript_btn.pack(side=tk.RIGHT, padx=(0, T.XS)) self.copy_btn = W.Button(toolbar, text="Copy", icon_name="copy", kind="quiet", width=80, height=28, command=self._copy_transcript) @@ -462,7 +476,10 @@ def worker(): for warning in self.engine.configure(mic, loopback): self.bridge.post(Log(f"⚠ {warning}\n")) - threading.Thread(target=worker, name="restart-monitor", daemon=True).start() + thread = threading.Thread(target=worker, name="restart-monitor", + daemon=True) + self._monitor_thread = thread + thread.start() # ================================================================== # Recording @@ -480,6 +497,38 @@ def start_recording(self): "Please select a microphone and a playback device.") return + # Disable straight away so F5 or a second click cannot start twice + # while we are still waiting for the devices. + self.start_btn.config(state="disabled") + self._start_deadline = time.monotonic() + DEVICE_READY_TIMEOUT_S + self._start_when_devices_ready() + + def _start_when_devices_ready(self): + """Begin recording once no device reconfiguration is in flight. + + engine.configure() runs off the GUI thread, and between closing the old + streams and assigning the new ones the engine has no active track. + Hitting Start in that window failed with "Neither audio source is + active" immediately after a perfectly valid device change. We poll + instead of joining so the interface stays responsive. + + On timeout we fall through deliberately: start_recording then reports + the engine's real error rather than hiding it behind a spinner. + """ + if self._shutting_down: + return + + thread = self._monitor_thread + busy = ((thread is not None and thread.is_alive()) + or self.engine.is_configuring) + if busy and time.monotonic() < self._start_deadline: + self.status.set("preparing devices…", T.WARN) + self.root.after(80, self._start_when_devices_ready) + return + + self._begin_recording() + + def _begin_recording(self): self._sync_settings_from_ui() base_name = paths.safe_output_name(self.filename_entry.get()) self.filename_entry.delete(0, tk.END) @@ -488,6 +537,8 @@ def start_recording(self): try: self.engine.start_recording(base_name) except RuntimeError as exc: + self.start_btn.config(state="normal") + self.status.set("ready", T.TEXT_MUTE) messagebox.showerror("Cannot record", str(exc)) return diff --git a/audio_transcriber/ui/theme.py b/audio_transcriber/ui/theme.py index bcae317..a254102 100644 --- a/audio_transcriber/ui/theme.py +++ b/audio_transcriber/ui/theme.py @@ -9,7 +9,6 @@ anywhere else in the UI code. """ -import tkinter as tk import tkinter.font as tkfont from tkinter import ttk diff --git a/audio_transcriber/ui/widgets.py b/audio_transcriber/ui/widgets.py index 9be7433..42d540c 100644 --- a/audio_transcriber/ui/widgets.py +++ b/audio_transcriber/ui/widgets.py @@ -321,6 +321,7 @@ def __init__(self, parent, from_=-20.0, to=20.0, value=0.0, command=None, self._centered = centered # fill from the centre instead of the left self._hover = False self._dragging = False + self._enabled = True self.bind("", self._on_click) self.bind("", self._on_drag) @@ -361,17 +362,21 @@ def _x_to_value(self, x): return self.from_ + max(0.0, min(1.0, share)) * (self.to - self.from_) def _on_click(self, event): + if not self._enabled: + return self._dragging = True self.set(self._x_to_value(event.x), notify=True) def _on_drag(self, event): - if self._dragging: + if self._dragging and self._enabled: self.set(self._x_to_value(event.x), notify=True) def _on_release(self, _event): self._dragging = False def _on_wheel(self, event): + if not self._enabled: + return step = (self.to - self.from_) / 80.0 self.set(self._value + (step if event.delta > 0 else -step), notify=True) diff --git a/build_release.py b/build_release.py index 2eed36e..3775037 100644 --- a/build_release.py +++ b/build_release.py @@ -1,27 +1,39 @@ """Automated build script for Audio AI Recorder & Transcriber. -Builds a lightweight standalone Windows executable using PyInstaller, prunes heavy -unused packages from the local environment, packages it into a zip archive, and -prepares the release asset for GitHub Releases. +Builds a standalone executable using PyInstaller, prunes heavy unused packages +from the local environment, and packages the result into a release archive. + +Runs on Windows, Linux and macOS - the CI calls this same script on all three +so the archive naming lives in exactly one place. + +Version resolution, highest priority first: + 1. python build_release.py v1.2.3 + 2. RELEASE_VERSION=v1.2.3 python build_release.py (set by the CI from the tag) + 3. the git tag pointing at HEAD + 4. v0.0.0-dev """ import os +import platform import shutil import subprocess import sys +import tarfile import zipfile APP_NAME = "AudioTranscriber" -VERSION = "v1.0.0" -ZIP_NAME = f"{APP_NAME}-{VERSION}-windows-x64.zip" +FALLBACK_VERSION = "v0.0.0-dev" ROOT_DIR = os.path.dirname(os.path.abspath(__file__)) DIST_DIR = os.path.join(ROOT_DIR, "dist") BUILD_DIR = os.path.join(ROOT_DIR, "build") APP_DIST_DIR = os.path.join(DIST_DIR, APP_NAME) -RELEASE_ZIP_PATH = os.path.join(DIST_DIR, ZIP_NAME) +MAC_APP_DIR = os.path.join(DIST_DIR, f"{APP_NAME}.app") -# Heavy packages from local Python environment that are not used by the app +# Heavy packages from the local Python environment that the app does not use. +# NOTE: Pillow (PIL) must NOT be listed here - ui/icons.py imports it at module +# level to render the vector icons, so excluding it produces a build that dies +# with "No module named 'PIL'" on the very first import. UNNEEDED_PACKAGES = [ "torch", "torchvision", @@ -33,7 +45,6 @@ "pyarrow", "pyarrow.libs", "jedi", - "PIL", "grpc", "IPython", "notebook", @@ -44,6 +55,51 @@ ] +# ---------------------------------------------------------------------- +def resolve_version(): + """Version for the archive name - see the module docstring for the order.""" + args = [arg for arg in sys.argv[1:] if not arg.startswith("-")] + if args: + return _normalize_version(args[0]) + + from_env = os.environ.get("RELEASE_VERSION", "").strip() + if from_env: + return _normalize_version(from_env) + + try: + tag = subprocess.run( + ["git", "describe", "--tags", "--exact-match"], + cwd=ROOT_DIR, capture_output=True, text=True, timeout=10) + if tag.returncode == 0 and tag.stdout.strip(): + return _normalize_version(tag.stdout.strip()) + except (OSError, subprocess.SubprocessError): + pass + + print(f"No version tag found - falling back to {FALLBACK_VERSION}") + return FALLBACK_VERSION + + +def _normalize_version(value): + """'1.2.3' and 'v1.2.3' both become 'v1.2.3'.""" + value = value.strip() + return value if value.startswith("v") else f"v{value}" + + +def platform_tag(): + """(archive suffix, platform slug) for the current OS.""" + if sys.platform == "win32": + return "zip", "windows-x64" + if sys.platform == "darwin": + # Not "universal": PyInstaller builds for the host architecture only + # unless target_arch=universal2 is set explicitly, so an arm64 runner + # produces an arm64-only binary. Naming it universal promised Intel + # users a build that would not run for them. + machine = platform.machine().lower() + return "zip", "macos-arm64" if machine in ("arm64", "aarch64") else "macos-x64" + return "tar.gz", "linux-x64" + + +# ---------------------------------------------------------------------- def clean(): """Clean build artifacts.""" print("Cleaning build directories...") @@ -68,15 +124,20 @@ def build_exe(): "--onedir", # Directory mode for fast launch & robust DLL loading "--noconfirm", "--clean", - "--collect-all", "pyaudiowpatch", "--collect-all", "soundfile", "--hidden-import", "scipy.signal", - "main.py", ] + # Only Windows has the WASAPI loopback fork; elsewhere plain PyAudio is + # used and ui/app.py falls back to it at import time. + if sys.platform == "win32": + cmd += ["--collect-all", "pyaudiowpatch"] + for mod in UNNEEDED_PACKAGES: cmd.extend(["--exclude-module", mod]) + cmd.append("main.py") + result = subprocess.run(cmd, cwd=ROOT_DIR) if result.returncode != 0: print("ERROR: PyInstaller build failed!") @@ -86,7 +147,7 @@ def build_exe(): def prune_unneeded(): - """Prune any leftover heavy unneeded directories collected by PyInstaller hooks.""" + """Prune leftover heavy directories collected by PyInstaller hooks.""" internal_dir = os.path.join(APP_DIST_DIR, "_internal") if not os.path.exists(internal_dir): return @@ -103,28 +164,73 @@ def prune_unneeded(): print(f" - Pruned file: {item}") -def create_zip(): - """Package the built executable directory into a zip archive.""" - print(f"Creating zip archive: {RELEASE_ZIP_PATH}...") +def verify_bundle(): + """Fail loudly if a module the app imports at startup is missing. + + An excluded dependency only shows up when a user double-clicks the binary, + which is far too late - Pillow was silently pruned this way and every + released Windows build died with "No module named 'PIL'". + """ + internal_dir = os.path.join(APP_DIST_DIR, "_internal") + if not os.path.exists(internal_dir): + return + + entries = os.listdir(internal_dir) + required = { + "PIL": "Pillow (ui/icons.py renders the vector icons with it)", + "soundfile": "soundfile (WAV reading/writing)", + } + + missing = [f"{name} - {why}" for name, why in required.items() + if not any(entry.split(".")[0] == name or entry.startswith(name + "-") + for entry in entries)] + if missing: + print("ERROR: the bundle is missing modules the app imports at startup:") + for item in missing: + print(f" - {item}") + sys.exit(1) + + print("Bundle verified: all startup imports are present.") + - with zipfile.ZipFile(RELEASE_ZIP_PATH, "w", zipfile.ZIP_DEFLATED) as zip_file: - for root, dirs, files in os.walk(APP_DIST_DIR): - for file in files: - full_path = os.path.join(root, file) - rel_path = os.path.relpath(full_path, DIST_DIR) - zip_file.write(full_path, rel_path) +def create_archive(archive_path, suffix): + """Package the built application directory into the release archive.""" + print(f"Creating archive: {archive_path}...") - zip_size_mb = os.path.getsize(RELEASE_ZIP_PATH) / (1024 * 1024) - print(f"Release package created: {ZIP_NAME} ({zip_size_mb:.1f} MB)") + # macOS --windowed produces a .app bundle next to the plain directory. + source_dir = MAC_APP_DIR if os.path.isdir(MAC_APP_DIR) else APP_DIST_DIR + arc_root = os.path.basename(source_dir) + + if suffix == "tar.gz": + with tarfile.open(archive_path, "w:gz") as tar: + tar.add(source_dir, arcname=arc_root) + else: + with zipfile.ZipFile(archive_path, "w", zipfile.ZIP_DEFLATED) as zip_file: + for root, _dirs, files in os.walk(source_dir): + for file in files: + full_path = os.path.join(root, file) + rel_path = os.path.join( + arc_root, os.path.relpath(full_path, source_dir)) + zip_file.write(full_path, rel_path) + + size_mb = os.path.getsize(archive_path) / (1024 * 1024) + print(f"Release package created: {os.path.basename(archive_path)} ({size_mb:.1f} MB)") def main(): + version = resolve_version() + suffix, slug = platform_tag() + archive_name = f"{APP_NAME}-{version}-{slug}.{suffix}" + archive_path = os.path.join(DIST_DIR, archive_name) + + print(f"--- Building {APP_NAME} {version} for {slug} ---") clean() build_exe() prune_unneeded() - create_zip() + verify_bundle() + create_archive(archive_path, suffix) print("\n--- RELEASE BUILD COMPLETE ---") - print(f"Zip archive ready for GitHub Release: {RELEASE_ZIP_PATH}") + print(f"Archive ready for GitHub Release: {archive_path}") if __name__ == "__main__": diff --git a/legacy/settings.v1.json b/legacy/settings.v1.json index a32d0bc..cda8c1c 100644 --- a/legacy/settings.v1.json +++ b/legacy/settings.v1.json @@ -4,7 +4,7 @@ "mic_gain_db": -8.0, "loop_gain_db": 10.0, "model_index": 0, - "elevenlabs_api_key": "sk_d3f9a431e0c35f5958c2b19b3ffe0fcf5da09b594a2103eb", + "elevenlabs_api_key": "sk_REDACTED_EXAMPLE_KEY_DO_NOT_USE", "live_transcribe": true, "filename": "test.wav" } \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 0021095..862ac7b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,8 +1,10 @@ # Runtime dependencies -PyAudioWPatch>=0.2.12 # WASAPI loopback (Windows) +PyAudioWPatch>=0.2.12; sys_platform == "win32" # WASAPI loopback (Windows) +PyAudio>=0.2.13; sys_platform != "win32" # portaudio elsewhere soundfile>=0.12.1 # WAV reading/writing numpy>=1.24 scipy>=1.10 # polyphase resampling +Pillow>=10.0 # vector icon rendering in ui/icons.py # Key storage uses the Windows DPAPI through ctypes and needs no extra # library. Tkinter ships with the standard Python installation on Windows. diff --git a/run_tests.py b/run_tests.py new file mode 100644 index 0000000..4e7c18c --- /dev/null +++ b/run_tests.py @@ -0,0 +1,28 @@ +"""Run the full test suite. + + python run_tests.py all tests + python run_tests.py -v with per-test output + python run_tests.py dsp only tests/test_dsp.py +""" + +import sys +import unittest + + +def main(): + names = [arg for arg in sys.argv[1:] if not arg.startswith("-")] + verbosity = 2 if "-v" in sys.argv else 1 + + loader = unittest.TestLoader() + if names: + suite = unittest.TestSuite( + loader.loadTestsFromName(f"tests.test_{name}") for name in names) + else: + suite = loader.discover("tests", top_level_dir=".") + + result = unittest.TextTestRunner(verbosity=verbosity).run(suite) + return 0 if result.wasSuccessful() else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_binaries.py b/tests/test_binaries.py new file mode 100644 index 0000000..2baa98d --- /dev/null +++ b/tests/test_binaries.py @@ -0,0 +1,155 @@ +"""Tests for the download and extraction layer (audit findings H8 and M2).""" + +import http.server +import os +import shutil +import tempfile +import threading +import unittest +import zipfile + +from audio_transcriber.transcribe import binaries + +PAYLOAD = b"x" * (256 * 1024) + + +class _Handler(http.server.BaseHTTPRequestHandler): + """A server with deliberately broken responses.""" + + def log_message(self, *args): + pass + + def _send(self, body, length=None, status=200): + self.send_response(status) + self.send_header("Content-Length", + str(length if length is not None else len(body))) + self.end_headers() + if body: + self.wfile.write(body) + + def do_GET(self): + if self.path == "/ok": + self._send(PAYLOAD) + elif self.path == "/truncated": + # Announces the full length but delivers only half of it + self._send(PAYLOAD[:len(PAYLOAD) // 2], length=len(PAYLOAD)) + elif self.path == "/missing": + self._send(b"not found", status=404) + else: + self._send(b"", status=400) + + +class _Server: + def __enter__(self): + self.httpd = http.server.HTTPServer(("127.0.0.1", 0), _Handler) + self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True) + self.thread.start() + return f"http://127.0.0.1:{self.httpd.server_port}" + + def __exit__(self, *exc): + self.httpd.shutdown() + self.httpd.server_close() + + +class TestDownload(unittest.TestCase): + def setUp(self): + self.dir = tempfile.mkdtemp() + + def tearDown(self): + shutil.rmtree(self.dir, ignore_errors=True) + + def test_successful_download(self): + dest = os.path.join(self.dir, "model.bin") + seen = [] + with _Server() as base: + binaries.download(f"{base}/ok", dest, "Test file", progress=seen.append) + self.assertEqual(os.path.getsize(dest), len(PAYLOAD)) + self.assertFalse(os.path.exists(dest + ".part")) + + def test_truncated_download_leaves_no_file(self): + """Regression H8: the previous version wrote straight to the target + file. An abort left a partial file behind that passed as a valid model + on the next start.""" + dest = os.path.join(self.dir, "model.bin") + with _Server() as base: + with self.assertRaises(binaries.DownloadError) as ctx: + binaries.download(f"{base}/truncated", dest, "Test file") + self.assertIn("incompletely", str(ctx.exception)) + self.assertFalse(os.path.exists(dest)) + self.assertFalse(os.path.exists(dest + ".part")) + + def test_http_error_is_reported(self): + dest = os.path.join(self.dir, "model.bin") + with _Server() as base: + with self.assertRaises(binaries.DownloadError) as ctx: + binaries.download(f"{base}/missing", dest, "Test file") + self.assertIn("404", str(ctx.exception)) + self.assertFalse(os.path.exists(dest)) + + def test_unreachable_host_is_reported(self): + dest = os.path.join(self.dir, "model.bin") + with self.assertRaises(binaries.DownloadError): + binaries.download("http://127.0.0.1:1/nothing", dest, "Test file", + timeout=2) + + def test_stale_part_file_is_replaced(self): + dest = os.path.join(self.dir, "model.bin") + with open(dest + ".part", "wb") as handle: + handle.write(b"garbage") + with _Server() as base: + binaries.download(f"{base}/ok", dest, "Test file") + self.assertEqual(os.path.getsize(dest), len(PAYLOAD)) + + +class TestSafeExtract(unittest.TestCase): + def setUp(self): + self.dir = tempfile.mkdtemp() + + def tearDown(self): + shutil.rmtree(self.dir, ignore_errors=True) + + def test_normal_archive(self): + zip_path = os.path.join(self.dir, "good.zip") + with zipfile.ZipFile(zip_path, "w") as archive: + archive.writestr("whisper-cli.exe", b"MZ") + archive.writestr("subfolder/whisper.dll", b"MZ") + target = os.path.join(self.dir, "bin") + os.makedirs(target) + binaries.safe_extract(zip_path, target) + self.assertTrue(os.path.exists(os.path.join(target, "whisper-cli.exe"))) + + def test_path_traversal_is_rejected(self): + """Regression M2 (zip slip): extractall() of the previous version + would have placed this file outside the target directory.""" + zip_path = os.path.join(self.dir, "evil.zip") + with zipfile.ZipFile(zip_path, "w") as archive: + archive.writestr("harmless.txt", b"ok") + archive.writestr("../../escaped.txt", b"pwned") + target = os.path.join(self.dir, "bin") + os.makedirs(target) + + with self.assertRaises(binaries.DownloadError) as ctx: + binaries.safe_extract(zip_path, target) + self.assertIn("outside", str(ctx.exception)) + self.assertFalse(os.path.exists(os.path.join(self.dir, "..", "escaped.txt"))) + + def test_absolute_path_is_rejected(self): + zip_path = os.path.join(self.dir, "absolute.zip") + with zipfile.ZipFile(zip_path, "w") as archive: + info = zipfile.ZipInfo("C:/Windows/Temp/evil.txt") + archive.writestr(info, b"pwned") + target = os.path.join(self.dir, "bin") + os.makedirs(target) + # Depending on normalisation either rejected or kept inside the target + try: + binaries.safe_extract(zip_path, target) + except binaries.DownloadError: + return + for root, _dirs, files in os.walk(target): + for name in files: + self.assertTrue(os.path.realpath(os.path.join(root, name)) + .startswith(os.path.realpath(target))) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_capture.py b/tests/test_capture.py new file mode 100644 index 0000000..067167a --- /dev/null +++ b/tests/test_capture.py @@ -0,0 +1,197 @@ +"""Tests for the recording timeline (audit finding H3).""" + +import os +import shutil +import tempfile +import threading +import time +import unittest + +import numpy as np +import soundfile as sf + +from audio_transcriber.audio import capture, dsp + +RATE = 48000 +BLOCK = capture.BLOCK_FRAMES + + +class TestDriftCorrection(unittest.TestCase): + """Both track timelines must stay tied to the system clock. + + In the previous version lost blocks (exception_on_overflow=False) were + swallowed silently. Since both tracks were read independently they drifted + apart - and the whole speaker attribution relied on the assumption that + sample index equals time. + """ + + def test_no_correction_when_in_sync(self): + elapsed = 100 * BLOCK / RATE + written = 99 * BLOCK + self.assertEqual(capture.drift_deficit(elapsed, RATE, written, BLOCK), 0) + + def test_dropout_produces_positive_deficit(self): + # 500 ms have passed but only 100 ms were written + deficit = capture.drift_deficit(0.5, RATE, int(0.1 * RATE), 0) + self.assertGreater(deficit, 0) + self.assertAlmostEqual(deficit / RATE, 0.4, places=2) + + def test_small_jitter_is_ignored(self): + # 20 ms of deviation is below the 50 ms threshold + deficit = capture.drift_deficit(0.52, RATE, int(0.5 * RATE), 0) + self.assertEqual(deficit, 0) + + def test_device_clock_ahead_is_reported_negative(self): + deficit = capture.drift_deficit(0.5, RATE, int(0.7 * RATE), 0) + self.assertLess(deficit, 0) + + def test_correction_keeps_tracks_aligned(self): + """Simulation: track A loses 300 ms halfway through, track B does not. + Without correction A ends 300 ms early - with correction they match.""" + total_blocks = 200 + block_s = BLOCK / RATE + dropout_s = 0.3 + + for corrected in (False, True): + frames = 0 + elapsed = 0.0 + for index in range(total_blocks): + # The wall clock keeps running; during a dropout time passes + # without samples arriving. + elapsed += block_s + if index == 100: + elapsed += dropout_s + if corrected: + deficit = capture.drift_deficit(elapsed, RATE, frames, BLOCK) + if deficit > 0: + frames += deficit + frames += BLOCK + + # On the common timeline the track should be as long as the wall + # clock says. + drift_s = abs(frames / RATE - elapsed) + if corrected: + self.assertLess(drift_s, 0.02, + "the correction does not compensate the dropout") + else: + self.assertAlmostEqual(drift_s, dropout_s, places=2) + + +class TestTrackLoading(unittest.TestCase): + def setUp(self): + self.dir = tempfile.mkdtemp() + + def tearDown(self): + shutil.rmtree(self.dir, ignore_errors=True) + + def _write(self, name, seconds, rate=RATE): + path = os.path.join(self.dir, name) + data = np.sin(2 * np.pi * 440 * np.arange(int(seconds * rate)) / rate) + sf.write(path, data.astype(np.float32), rate, subtype="PCM_16") + return path + + def test_resamples_to_16k(self): + path = self._write("track.wav", 2.0) + result = capture.TrackResult(path=path, rate=RATE, frames=2 * RATE) + audio = capture.load_track(result) + self.assertAlmostEqual(len(audio) / dsp.TARGET_RATE, 2.0, places=2) + + def test_start_offset_is_padded_at_the_front(self): + """The track whose stream started later missed the beginning and has + to move back on the common timeline.""" + path = self._write("track.wav", 1.0) + result = capture.TrackResult(path=path, rate=RATE, frames=RATE, + start_offset_s=0.25) + audio = capture.load_track(result) + pad = int(0.25 * dsp.TARGET_RATE) + np.testing.assert_allclose(audio[:pad], 0.0, atol=1e-7) + self.assertGreater(float(np.max(np.abs(audio[pad:]))), 0.5) + self.assertAlmostEqual(len(audio) / dsp.TARGET_RATE, 1.25, places=2) + + def test_missing_file_returns_empty(self): + result = capture.TrackResult(path=os.path.join(self.dir, "gone.wav"), + rate=RATE, frames=0) + self.assertEqual(len(capture.load_track(result)), 0) + self.assertEqual(len(capture.load_track(None)), 0) + + def test_duration_property(self): + result = capture.TrackResult(path="x", rate=16000, frames=32000) + self.assertAlmostEqual(result.duration_s, 2.0) + + +class TestConfigureIsAtomic(unittest.TestCase): + """A reconfiguration must not be observable half-done. + + configure() closes the old streams and only then opens the new ones. It + briefly holds no active track, and start_recording() reads exactly that + state - so hitting Start right after a device change reported "Neither + audio source is active" for a perfectly healthy device. + """ + + def setUp(self): + self.dir = tempfile.mkdtemp() + self.engine = capture.AudioEngine(pa=None, tmp_dir=self.dir) + + def tearDown(self): + shutil.rmtree(self.dir, ignore_errors=True) + + def test_is_configuring_reports_the_window(self): + self.assertFalse(self.engine.is_configuring) + + seen = [] + barrier = threading.Event() + released = threading.Event() + + def slow_configure(_mic, _sys): + barrier.set() + released.wait(timeout=5.0) + return [] + + # Stand in for the real device work; only the locking is under test. + self.engine._configure = slow_configure + + worker = threading.Thread( + target=lambda: self.engine.configure(None, None), daemon=True) + worker.start() + + self.assertTrue(barrier.wait(timeout=5.0)) + seen.append(self.engine.is_configuring) + released.set() + worker.join(timeout=5.0) + + self.assertEqual(seen, [True], "is_configuring must be True mid-flight") + self.assertFalse(self.engine.is_configuring) + + def test_two_reconfigurations_do_not_interleave(self): + """Without the lock the calls interleave as A-enter, B-enter, ...""" + events = [] + lock = threading.Lock() + + def tracked_configure(_mic, name): + with lock: + events.append(f"enter-{name}") + time.sleep(0.05) + with lock: + events.append(f"leave-{name}") + return [] + + self.engine._configure = tracked_configure + + threads = [threading.Thread(target=self.engine.configure, + args=(None, name), daemon=True) + for name in ("A", "B")] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=5.0) + + self.assertEqual(len(events), 4) + # Every enter must be followed by its own leave. + for index in range(0, 4, 2): + self.assertTrue(events[index].startswith("enter-"), events) + self.assertEqual(events[index + 1], + events[index].replace("enter-", "leave-"), events) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_capture_hardware.py b/tests/test_capture_hardware.py new file mode 100644 index 0000000..aa9fc36 --- /dev/null +++ b/tests/test_capture_hardware.py @@ -0,0 +1,141 @@ +"""Recording test against real audio hardware. + +Skipped by default because it opens the microphone and the loopback. Enable +with: + + set AUDIO_TRANSCRIBER_HW_TEST=1 + python run_tests.py capture_hardware + +Checks the parts of audit finding H3 that cannot show up without hardware: +opening the streams, the timeline against the system clock, alignment of both +tracks, constant memory use and clean thread shutdown. +""" + +import os +import shutil +import tempfile +import threading +import time +import unittest + +import soundfile as sf + +ENABLED = os.environ.get("AUDIO_TRANSCRIBER_HW_TEST") == "1" +DURATION_S = 6.0 + + +@unittest.skipUnless(ENABLED, "set AUDIO_TRANSCRIBER_HW_TEST=1 to enable") +class TestRealCapture(unittest.TestCase): + def setUp(self): + import pyaudiowpatch as pyaudio + from audio_transcriber.audio import devices as devmod + from audio_transcriber.audio.capture import AudioEngine + + self.dir = tempfile.mkdtemp() + self.pa = pyaudio.PyAudio() + self.devices = devmod.enumerate_devices(self.pa) + + mics = devmod.microphone_candidates(self.devices) + outputs = devmod.playback_candidates(self.devices) + if not mics or not outputs: + self.skipTest("no suitable devices found") + + self.mic = mics[0] + self.loopback, _reason = devmod.find_loopback_for(self.devices, outputs[0]) + self.engine = AudioEngine(self.pa, self.dir) + + def tearDown(self): + try: + self.engine.stop_streams() + finally: + self.pa.terminate() + shutil.rmtree(self.dir, ignore_errors=True) + + def test_records_both_tracks_with_correct_timeline(self): + for warning in self.engine.configure(self.mic, self.loopback): + print(" note:", warning) + + threads_before = threading.active_count() + self.engine.start_recording("hwtest") + started = time.perf_counter() + time.sleep(DURATION_S) + result = self.engine.stop_recording() + wall_clock = time.perf_counter() - started + + self.assertTrue(result.has_audio, "neither track delivered data") + + for name, track in (("mic", result.mic), ("sys", result.sys)): + with self.subTest(track=name): + if track is None: + continue + self.assertTrue(os.path.exists(track.path)) + info = sf.info(track.path) + self.assertEqual(info.channels, 1, "the track must be mono") + + print(f" {name}: {track.device_name[:34]:36s} " + f"{info.frames / info.samplerate:6.2f} s @ {info.samplerate} Hz, " + f"dropouts {track.inserted_silence_s * 1000:5.1f} ms, " + f"start offset {track.start_offset_s * 1000:5.1f} ms") + + # What matters is not the raw length but the length on the + # COMMON timeline: raw length plus start offset. A loopback + # device occasionally needs a few hundred milliseconds before + # the first buffer arrives - that is what start_offset_s is for. + on_timeline = info.frames / info.samplerate + track.start_offset_s + self.assertAlmostEqual(on_timeline, wall_clock, delta=0.6) + # No dropout correction during a quiet six second recording + self.assertLess(track.inserted_silence_s, 0.5) + + if result.mic and result.sys: + offset_ms = abs(result.mic.start_offset_s - result.sys.start_offset_s) * 1000 + print(f" offset between the tracks: {offset_ms:.1f} ms") + self.assertLess(offset_ms, 250.0, + "the two streams started too far apart") + + # Both tracks must be equally long on the common timeline + from audio_transcriber.audio import capture + mic_audio = capture.load_track(result.mic) + sys_audio = capture.load_track(result.sys) + drift_ms = abs(len(mic_audio) - len(sys_audio)) / 16000 * 1000 + print(f" length difference after alignment: {drift_ms:.1f} ms") + self.assertLess(drift_ms, 300.0) + + # The capture threads of this generation must terminate + self.engine.stop_streams() + time.sleep(0.3) + self.assertLessEqual(threading.active_count(), threads_before, + "capture threads did not terminate") + + def test_memory_stays_flat(self): + """Regression K3: memory use must not grow with recording length. What + is measured is the size of the live window, the only thing that stays + in RAM.""" + self.engine.configure(self.mic, self.loopback) + self.engine.start_recording("hwmem") + try: + sizes = [] + for _ in range(3): + time.sleep(2.0) + mic, system = self.engine.live_window() + sizes.append(sum(len(x) for x in (mic, system) if x is not None)) + print(f" live window: {sizes[-1]} samples") + # The window is capped (30 s at 16 kHz per track) + self.assertLessEqual(max(sizes), 2 * 30 * 16000 * 1.1) + finally: + self.engine.stop_recording() + + def test_restarting_monitoring_does_not_leak_threads(self): + """Regression H1: in the previous version every device change could + leave another generation of capture threads behind.""" + baseline = threading.active_count() + for _ in range(4): + self.engine.configure(self.mic, self.loopback) + time.sleep(0.3) + self.engine.stop_streams() + time.sleep(0.5) + self.assertLessEqual(threading.active_count(), baseline + 1, + "threads of an old generation are still alive") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..f105a8c --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,146 @@ +"""Tests for settings and key storage (audit finding H6).""" + +import json +import os +import tempfile +import unittest + +from audio_transcriber import config, secretstore + +SAMPLE_KEY = "sk_testkey_0123456789abcdef" + + +class TestSecretStore(unittest.TestCase): + @unittest.skipUnless(secretstore.is_available(), "DPAPI is Windows only") + def test_roundtrip(self): + token = secretstore.encrypt(SAMPLE_KEY) + self.assertNotIn(SAMPLE_KEY, token) + self.assertEqual(secretstore.decrypt(token), SAMPLE_KEY) + + def test_empty_values(self): + self.assertEqual(secretstore.decrypt(""), "") + self.assertEqual(secretstore.decrypt("not-base64!!"), "") + self.assertEqual(secretstore.decrypt("AAAAAAAAAA"), "") + + +class TestSettingsFile(unittest.TestCase): + def setUp(self): + self.dir = tempfile.mkdtemp() + self.path = os.path.join(self.dir, "settings.json") + + def tearDown(self): + for name in os.listdir(self.dir): + os.remove(os.path.join(self.dir, name)) + os.rmdir(self.dir) + + def test_roundtrip(self): + settings = config.Settings(mic_device="20: Microphone", + loop_device="22: Loopback", + mic_gain_db=-8.0, loop_gain_db=10.0, + model_index=5, language="en", + filename="meeting", + output_dir="/custom/output/folder") + settings.api_key = SAMPLE_KEY + config.save(settings, self.path) + + loaded, warnings = config.load(self.path) + self.assertEqual(warnings, []) + self.assertEqual(loaded.mic_device, "20: Microphone") + self.assertEqual(loaded.mic_gain_db, -8.0) + self.assertEqual(loaded.model_index, 5) + self.assertEqual(loaded.language, "en") + self.assertEqual(loaded.output_dir, "/custom/output/folder") + self.assertEqual(loaded.get_output_dir(), os.path.abspath("/custom/output/folder")) + if secretstore.is_available(): + self.assertEqual(loaded.api_key, SAMPLE_KEY) + + + def test_key_is_never_written_in_clear_text(self): + """The central requirement of H6.""" + settings = config.Settings() + settings.api_key = SAMPLE_KEY + config.save(settings, self.path) + + with open(self.path, "rb") as handle: + raw = handle.read() + self.assertNotIn(SAMPLE_KEY.encode(), raw) + + data = json.loads(raw.decode("utf-8")) + self.assertNotIn("elevenlabs_api_key", data) + if secretstore.is_available(): + self.assertTrue(data["elevenlabs_api_key_enc"]) + + def test_migration_from_plaintext_schema_v1(self): + """An existing file in the old format is adopted - with a warning.""" + legacy = { + "mic_device": "1: Microphone (Yeti X)", + "loop_device": "4: Headphones (PRO X 2 LIGHTSPEED)", + "mic_gain_db": -8.0, + "loop_gain_db": 10.0, + "model_index": 0, + "elevenlabs_api_key": SAMPLE_KEY, + "live_transcribe": True, + "filename": "test.wav", + } + with open(self.path, "w", encoding="utf-8") as handle: + json.dump(legacy, handle) + + loaded, warnings = config.load(self.path) + self.assertEqual(loaded.api_key, SAMPLE_KEY) + self.assertTrue(loaded.migrated_plaintext_key) + self.assertTrue(any("revoke" in warning for warning in warnings)) + self.assertEqual(loaded.mic_gain_db, -8.0) + + # After saving, the key is no longer in the file as clear text + config.save(loaded, self.path) + with open(self.path, "rb") as handle: + self.assertNotIn(SAMPLE_KEY.encode(), handle.read()) + + def test_corrupt_file_falls_back_to_defaults(self): + with open(self.path, "w", encoding="utf-8") as handle: + handle.write("{ this is not json") + loaded, warnings = config.load(self.path) + self.assertTrue(warnings) + self.assertEqual(loaded.model_index, config.Settings().model_index) + + def test_invalid_single_value_is_ignored(self): + with open(self.path, "w", encoding="utf-8") as handle: + json.dump({"mic_gain_db": "very loud", "filename": "ok"}, handle) + loaded, warnings = config.load(self.path) + self.assertEqual(loaded.filename, "ok") + self.assertEqual(loaded.mic_gain_db, 0.0) + self.assertTrue(warnings) + + def test_model_index_is_clamped(self): + with open(self.path, "w", encoding="utf-8") as handle: + json.dump({"model_index": 99}, handle) + loaded, _warnings = config.load(self.path) + self.assertEqual(loaded.model_index, len(config.MODEL_CHOICES) - 1) + + def test_save_is_atomic(self): + config.save(config.Settings(), self.path) + self.assertFalse(os.path.exists(self.path + ".tmp")) + + +class TestModelSelection(unittest.TestCase): + def test_cloud_entry(self): + settings = config.Settings(model_index=0) + self.assertTrue(settings.uses_cloud()) + self.assertIsNone(settings.model_name()) + + def test_live_model_never_exceeds_small(self): + """Regression H9: the live preview must not start large-v3 on the CPU + and block every core.""" + for index in range(len(config.MODEL_CHOICES)): + settings = config.Settings(model_index=index) + self.assertIn(settings.live_model_name(), ("tiny", "base", "small")) + + def test_thread_count_leaves_headroom(self): + settings = config.Settings(whisper_threads=0) + self.assertGreaterEqual(settings.threads(), 1) + self.assertLessEqual(settings.threads(), max(1, (os.cpu_count() or 4) - 2)) + self.assertEqual(config.Settings(whisper_threads=6).threads(), 6) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_diarize.py b/tests/test_diarize.py new file mode 100644 index 0000000..74e3c15 --- /dev/null +++ b/tests/test_diarize.py @@ -0,0 +1,171 @@ +"""Tests for speaker attribution (audit finding H2 and crosstalk).""" + +import unittest + +import numpy as np + +from audio_transcriber import diarize +from audio_transcriber.audio import dsp +from audio_transcriber.transcribe.base import Segment + +RATE = dsp.TARGET_RATE + + +def track(bursts, seconds=20.0, rate=RATE, seed=0): + """Build a track of noise bursts: bursts = [(start_s, end_s, amplitude)].""" + rng = np.random.default_rng(seed) + signal = np.zeros(int(seconds * rate), dtype=np.float32) + for start, end, amplitude in bursts: + i_start, i_end = int(start * rate), int(end * rate) + signal[i_start:i_end] = rng.normal(0, amplitude, i_end - i_start) + return signal + + +class TestGainInvariance(unittest.TestCase): + """The core of audit finding H2. + + In the previous version determine_speaker() compared the absolute levels + of both tracks AFTER the gain sliders had been applied. With the settings + actually found on disk (mic -8 dB, system +10 dB) that produced an 18 dB + bias in favour of the other party. + """ + + def setUp(self): + # You speak from 1-3 s, the other party from 5-7 s. + self.mic = track([(1.0, 3.0, 0.20)], seed=1) + self.sys = track([(5.0, 7.0, 0.20)], seed=2) + self.mic_segments = [Segment(1.0, 3.0, "Hello, how are you?", "mic")] + self.sys_segments = [Segment(5.0, 7.0, "Thanks, doing well.", "sys")] + + def _labels(self, mic_factor=1.0, sys_factor=1.0): + report = diarize.merge(self.mic_segments, self.sys_segments, + mic_audio=self.mic * mic_factor, + sys_audio=self.sys * sys_factor) + return [(line.speaker, line.text) for line in report.lines] + + def test_baseline(self): + self.assertEqual(self._labels(), [ + (diarize.LABEL_SELF, "Hello, how are you?"), + (diarize.LABEL_OTHER, "Thanks, doing well."), + ]) + + def test_result_is_independent_of_gain_settings(self): + """Exactly the configuration found in the original settings.json: + mic_gain_db = -8.0, loop_gain_db = +10.0.""" + expected = self._labels() + mic_factor = 10 ** (-8.0 / 20) + sys_factor = 10 ** (10.0 / 20) + self.assertEqual(self._labels(mic_factor, sys_factor), expected) + + def test_extreme_gain_difference_still_correct(self): + expected = self._labels() + for mic_factor, sys_factor in [(0.01, 10.0), (10.0, 0.01), + (0.1, 0.1), (5.0, 5.0)]: + with self.subTest(mic=mic_factor, sys=sys_factor): + self.assertEqual(self._labels(mic_factor, sys_factor), expected) + + +class TestBleedSuppression(unittest.TestCase): + def test_speaker_bleed_into_microphone_is_dropped(self): + """Speaker playback: the other party's voice reaches the microphone + quietly and is recognised there a second time. + + This is the only physically possible crosstalk direction - the + loopback only captures what the computer plays back, so your own voice + cannot show up there. + """ + # Both tracks also contain genuine speech so that the per-track + # reference level reflects the actual speech level. + mic = track([(1.0, 3.0, 0.25), (6.0, 8.0, 0.02)], seed=3) + sys = track([(6.0, 8.0, 0.22)], seed=4) + report = diarize.merge( + [Segment(1.0, 3.0, "This is my own statement.", "mic"), + Segment(6.0, 8.0, "This is what the other person says.", "mic")], + [Segment(6.0, 8.0, "This is what the other person says.", "sys")], + mic_audio=mic, sys_audio=sys) + + self.assertEqual(len(report.lines), 2) + self.assertEqual(report.lines[0].speaker, diarize.LABEL_SELF) + self.assertEqual(report.lines[1].speaker, diarize.LABEL_OTHER) + self.assertGreaterEqual(report.dropped_bleed + report.dropped_duplicate, 1) + + def test_tie_break_favours_system_track(self): + """If a track contains nothing but crosstalk, its own reference level + sits at crosstalk level and the normalisation says nothing. The + physical crosstalk direction then decides.""" + mic = track([(1.0, 3.0, 0.02)], seed=5) # crosstalk only + sys = track([(1.0, 3.0, 0.25)], seed=6) + report = diarize.merge( + [Segment(1.0, 3.0, "Identical sentence.", "mic")], + [Segment(1.0, 3.0, "Identical sentence.", "sys")], + mic_audio=mic, sys_audio=sys) + + self.assertEqual(len(report.lines), 1) + self.assertEqual(report.lines[0].speaker, diarize.LABEL_OTHER) + + def test_simultaneous_speech_keeps_both(self): + """If both speak at once at comparable level with different text, both + must survive.""" + mic = track([(2.0, 4.0, 0.20)], seed=5) + sys = track([(2.0, 4.0, 0.18)], seed=6) + report = diarize.merge( + [Segment(2.0, 4.0, "I do not think that is right.", "mic")], + [Segment(2.0, 4.0, "Hold on, I have to disagree there.", "sys")], + mic_audio=mic, sys_audio=sys) + self.assertEqual(len(report.lines), 2) + self.assertEqual({line.speaker for line in report.lines}, + {diarize.LABEL_SELF, diarize.LABEL_OTHER}) + + +class TestHallucinationFilter(unittest.TestCase): + def test_segment_without_signal_is_dropped(self): + """whisper likes to invent text during pauses ('Subtitles by ...'). + Without energy on its own track the segment is dropped.""" + mic = track([(1.0, 3.0, 0.25)], seed=7) + sys = np.zeros(int(20 * RATE), dtype=np.float32) + report = diarize.merge( + [Segment(1.0, 3.0, "A real sentence.", "mic"), + Segment(12.0, 14.0, "Subtitles by the community", "mic")], + [], mic_audio=mic, sys_audio=sys) + + self.assertEqual([line.text for line in report.lines], ["A real sentence."]) + self.assertEqual(report.dropped_silence, 1) + + def test_without_audio_nothing_is_filtered(self): + """Without a reference signal the filter must not fire blindly.""" + report = diarize.merge( + [Segment(0.0, 1.0, "A", "mic")], [Segment(2.0, 3.0, "B", "sys")]) + self.assertEqual(len(report.lines), 2) + self.assertEqual(report.dropped_silence, 0) + + +class TestOrderingAndRendering(unittest.TestCase): + def test_chronological_order(self): + mic = track([(1.0, 2.0, 0.2), (7.0, 8.0, 0.2)], seed=8) + sys = track([(4.0, 5.0, 0.2)], seed=9) + report = diarize.merge( + [Segment(1.0, 2.0, "One", "mic"), Segment(7.0, 8.0, "Three", "mic")], + [Segment(4.0, 5.0, "Two", "sys")], + mic_audio=mic, sys_audio=sys) + self.assertEqual([line.text for line in report.lines], ["One", "Two", "Three"]) + + def test_render_format(self): + mic = track([(65.0, 67.0, 0.2)], seconds=80.0, seed=10) + report = diarize.merge([Segment(65.0, 67.0, "After one minute", "mic")], + [], mic_audio=mic, + sys_audio=np.zeros(int(80 * RATE), dtype=np.float32)) + self.assertEqual(diarize.render(report), "[01:05] [You]: After one minute") + + def test_multiple_participants_on_system_track(self): + """When ElevenLabs supplies speaker ids on the system track the + participants are told apart instead of collapsing into one speaker.""" + sys = track([(1.0, 2.0, 0.2), (3.0, 4.0, 0.2)], seed=11) + first = Segment(1.0, 2.0, "My name is Anna.", "sys", speaker_hint="speaker_0") + second = Segment(3.0, 4.0, "And I am Ben.", "sys", speaker_hint="speaker_1") + report = diarize.merge([], [first, second], mic_audio=None, sys_audio=sys) + self.assertEqual([line.speaker for line in report.lines], + ["[Participant A]", "[Participant B]"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_dsp.py b/tests/test_dsp.py new file mode 100644 index 0000000..805edcc --- /dev/null +++ b/tests/test_dsp.py @@ -0,0 +1,156 @@ +"""Tests for the signal processing layer.""" + +import math +import unittest + +import numpy as np + +from audio_transcriber.audio import dsp + + +class TestDownmix(unittest.TestCase): + def test_ignores_silent_surround_channels(self): + """An 8-channel loopback carrying stereo must not lose 6 dB. + + Regression for the audit finding about the naive mean(axis=1): a real + loopback device reports 8 channels, six of which are digitally silent + during stereo playback. + """ + frames = 4800 + rng = np.random.default_rng(0) + signal = rng.normal(0, 0.2, frames).astype(np.float32) + + multi = np.zeros((frames, 8), dtype=np.float32) + multi[:, 0] = signal + multi[:, 1] = signal + interleaved = multi.reshape(-1) + + mono = dsp.downmix_active(interleaved, 8) + naive = multi.mean(axis=1) + + self.assertAlmostEqual(dsp.rms(mono), dsp.rms(signal), places=5) + # The naive route loses 20*log10(8/2) = 12 dB + loss_db = 20 * math.log10(dsp.rms(mono) / dsp.rms(naive)) + self.assertAlmostEqual(loss_db, 12.0, places=1) + + def test_channel_activity_is_sticky(self): + """A channel that once carried signal stays in the mix - otherwise the + level jumps on every pause in speech.""" + mixer = dsp.ActiveChannelDownmixer(2) + loud = np.zeros((512, 2), dtype=np.float32) + loud[:, 0] = 0.5 + loud[:, 1] = 0.5 + mixer.process(loud.reshape(-1)) + self.assertEqual(mixer.active_channels, [0, 1]) + + # A block where only channel 0 carries signal + partial = np.zeros((512, 2), dtype=np.float32) + partial[:, 0] = 0.5 + mixer.process(partial.reshape(-1)) + self.assertEqual(mixer.active_channels, [0, 1]) + + def test_mono_passthrough(self): + data = np.array([0.1, -0.2, 0.3], dtype=np.float32) + np.testing.assert_allclose(dsp.downmix_active(data, 1), data) + + def test_incomplete_frame_is_discarded(self): + # 7 samples across 2 channels -> only 3 complete frames + data = np.arange(7, dtype=np.float32) + self.assertEqual(len(dsp.downmix_active(data, 2)), 3) + + +class TestResample(unittest.TestCase): + def test_length_and_frequency_preserved(self): + rate, target, freq, seconds = 48000, 16000, 1000.0, 1.0 + t = np.arange(int(rate * seconds)) / rate + sine = np.sin(2 * np.pi * freq * t).astype(np.float32) + + out = dsp.resample(sine, rate, target) + self.assertAlmostEqual(len(out) / target, seconds, places=2) + + spectrum = np.abs(np.fft.rfft(out)) + peak_hz = np.fft.rfftfreq(len(out), 1 / target)[int(np.argmax(spectrum))] + self.assertAlmostEqual(peak_hz, freq, delta=5.0) + + def test_44100_to_16000(self): + out = dsp.resample(np.zeros(44100, dtype=np.float32), 44100, 16000) + self.assertAlmostEqual(len(out), 16000, delta=2) + + def test_identity(self): + data = np.arange(10, dtype=np.float32) + np.testing.assert_array_equal(dsp.resample(data, 16000, 16000), data) + + +class TestLevels(unittest.TestCase): + def _speech_like(self, rate=16000, seconds=10.0, level=0.2): + """Speech-like signal: loud passages with silence in between.""" + count = int(rate * seconds) + signal = np.zeros(count, dtype=np.float32) + rng = np.random.default_rng(1) + for start in range(0, count, rate * 2): + end = min(count, start + rate) + signal[start:end] = rng.normal(0, level, end - start) + return signal + + def test_reference_level_ignores_silence(self): + """The reference measures the speech level, not the mean including + pauses - the basis of the gain-neutral attribution.""" + signal = self._speech_like(level=0.2) + reference = dsp.reference_level(signal) + self.assertGreater(reference, 0.15) + self.assertLess(reference, 0.30) + # The naive overall RMS would sit much lower because of the pauses + self.assertLess(dsp.rms(signal), reference) + + def test_reference_level_scales_linearly(self): + """Core assumption of the diarization: a constant factor (the gain + slider) shifts segment level and reference level equally, so it + cancels out in the ratio.""" + signal = self._speech_like() + for factor in (0.1, 0.5, 2.0, 8.0): + self.assertAlmostEqual( + dsp.reference_level(signal * factor) / factor, + dsp.reference_level(signal), + places=5) + + def test_segment_rms_window(self): + rate = 16000 + signal = np.zeros(rate * 4, dtype=np.float32) + signal[rate:rate * 2] = 0.5 + self.assertAlmostEqual(dsp.segment_rms(signal, 1.0, 2.0, rate), 0.5, places=3) + self.assertAlmostEqual(dsp.segment_rms(signal, 2.5, 3.5, rate), 0.0, places=6) + + def test_segment_rms_out_of_range(self): + signal = np.ones(1000, dtype=np.float32) + self.assertEqual(dsp.segment_rms(signal, 10.0, 11.0, 16000), 0.0) + self.assertEqual(dsp.segment_rms(None, 0.0, 1.0), 0.0) + + +class TestGain(unittest.TestCase): + def test_limit_peak_never_amplifies(self): + quiet = np.full(100, 0.1, dtype=np.float32) + np.testing.assert_allclose(dsp.limit_peak(quiet), quiet) + + def test_limit_peak_reduces_clipping(self): + loud = np.full(100, 2.0, dtype=np.float32) + self.assertAlmostEqual(float(np.max(dsp.limit_peak(loud))), 0.95, places=5) + + def test_apply_gain_db(self): + data = np.ones(10, dtype=np.float32) + np.testing.assert_allclose(dsp.apply_gain(data, 6.0), 1.995, rtol=1e-3) + np.testing.assert_allclose(dsp.apply_gain(data, -6.0), 0.5012, rtol=1e-3) + + def test_normalize_for_asr_raises_quiet_track(self): + rng = np.random.default_rng(2) + quiet = rng.normal(0, 0.004, 16000 * 3).astype(np.float32) + out = dsp.normalize_for_asr(quiet, target_rms=0.06) + self.assertGreater(dsp.reference_level(out), 0.04) + self.assertLessEqual(float(np.max(np.abs(out))), 0.95 + 1e-6) + + def test_normalize_leaves_digital_silence_alone(self): + silence = np.zeros(16000, dtype=np.float32) + np.testing.assert_array_equal(dsp.normalize_for_asr(silence), silence) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_elevenlabs.py b/tests/test_elevenlabs.py new file mode 100644 index 0000000..359c0f4 --- /dev/null +++ b/tests/test_elevenlabs.py @@ -0,0 +1,208 @@ +"""Tests for the ElevenLabs backend (audit findings M8, H8, K2).""" + +import http.server +import json +import os +import shutil +import tempfile +import threading +import unittest + +import numpy as np +import soundfile as sf + +from audio_transcriber.transcribe import elevenlabs +from audio_transcriber.transcribe.base import TranscriptionError + +CAPTURED = {} +BEHAVIOUR = {"mode": "ok"} + +WORDS_RESPONSE = { + "text": "Hello world. How are you?", + "words": [ + {"text": "Hello", "start": 0.0, "end": 0.4, "type": "word", "speaker_id": "speaker_0"}, + {"text": " ", "start": 0.4, "end": 0.45, "type": "spacing"}, + {"text": "world.", "start": 0.45, "end": 0.9, "type": "word", "speaker_id": "speaker_0"}, + {"text": " ", "start": 0.9, "end": 1.0, "type": "spacing"}, + {"text": "How", "start": 1.0, "end": 1.2, "type": "word", "speaker_id": "speaker_1"}, + {"text": " ", "start": 1.2, "end": 1.25, "type": "spacing"}, + {"text": "are", "start": 1.25, "end": 1.5, "type": "word", "speaker_id": "speaker_1"}, + {"text": " ", "start": 1.5, "end": 1.55, "type": "spacing"}, + {"text": "you?", "start": 1.55, "end": 2.1, "type": "word", "speaker_id": "speaker_1"}, + ], +} + + +class _Handler(http.server.BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_POST(self): + length = int(self.headers.get("Content-Length", 0)) + body = self.rfile.read(length) + CAPTURED["content_length"] = length + CAPTURED["actual_length"] = len(body) + CAPTURED["content_type"] = self.headers.get("Content-Type", "") + CAPTURED["api_key"] = self.headers.get("xi-api-key", "") + CAPTURED["body"] = body + + if BEHAVIOUR["mode"] == "model_error" and b"scribe_v2" in body: + payload = json.dumps({"detail": {"message": "model_id not found"}}).encode() + self.send_response(422) + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + return + if BEHAVIOUR["mode"] == "unauthorized": + payload = b'{"detail":"invalid api key"}' + self.send_response(401) + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + return + + payload = json.dumps(WORDS_RESPONSE).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + +class ElevenLabsTestCase(unittest.TestCase): + def setUp(self): + CAPTURED.clear() + BEHAVIOUR["mode"] = "ok" + self.dir = tempfile.mkdtemp() + self.wav = os.path.join(self.dir, "recording.wav") + sf.write(self.wav, np.zeros(16000 * 3, dtype=np.float32), 16000, + subtype="PCM_16") + + self.httpd = http.server.HTTPServer(("127.0.0.1", 0), _Handler) + threading.Thread(target=self.httpd.serve_forever, daemon=True).start() + self._original_url = elevenlabs.API_URL + elevenlabs.API_URL = f"http://127.0.0.1:{self.httpd.server_port}/v1/speech-to-text" + + def tearDown(self): + elevenlabs.API_URL = self._original_url + self.httpd.shutdown() + self.httpd.server_close() + shutil.rmtree(self.dir, ignore_errors=True) + + +class TestChainedBody(unittest.TestCase): + def test_streams_prefix_file_suffix_exactly(self): + """Regression M8: the previous version assembled the whole body as a + bytearray in RAM (roughly 700 MB peak for an hour of audio).""" + directory = tempfile.mkdtemp() + try: + path = os.path.join(directory, "data.bin") + payload = bytes(range(256)) * 400 + with open(path, "wb") as handle: + handle.write(payload) + + handle = open(path, "rb") + body = elevenlabs._ChainedBody([b"PREFIX", handle, b"SUFFIX"]) + chunks = [] + while True: + chunk = body.read(8192) + if not chunk: + break + chunks.append(chunk) + self.assertEqual(b"".join(chunks), b"PREFIX" + payload + b"SUFFIX") + finally: + shutil.rmtree(directory, ignore_errors=True) + + def test_small_reads(self): + body = elevenlabs._ChainedBody([b"abc", b"de"]) + self.assertEqual(body.read(2), b"ab") + self.assertEqual(body.read(2), b"c") + self.assertEqual(body.read(10), b"de") + self.assertEqual(body.read(10), b"") + + +class TestRequest(ElevenLabsTestCase): + def test_multipart_is_well_formed(self): + backend = elevenlabs.ElevenLabsBackend(api_key="sk_test", model_id="scribe_v2") + backend.transcribe(self.wav, language="de") + + self.assertEqual(CAPTURED["api_key"], "sk_test") + self.assertEqual(CAPTURED["content_length"], CAPTURED["actual_length"]) + self.assertIn("multipart/form-data; boundary=", CAPTURED["content_type"]) + + body = CAPTURED["body"] + self.assertIn(b'name="model_id"', body) + self.assertIn(b"scribe_v2", body) + self.assertIn(b'name="diarize"', body) + self.assertIn(b'name="language_code"', body) + self.assertIn(b"\r\nde\r\n", body) + self.assertIn(b'filename="recording.wav"', body) + self.assertTrue(body.rstrip().endswith(b"--")) + + def test_boundary_is_random(self): + backend = elevenlabs.ElevenLabsBackend(api_key="sk_test") + backend.transcribe(self.wav) + first = CAPTURED["content_type"] + backend.transcribe(self.wav) + self.assertNotEqual(first, CAPTURED["content_type"]) + + def test_language_auto_is_omitted(self): + elevenlabs.ElevenLabsBackend(api_key="sk_test").transcribe(self.wav, + language="auto") + self.assertNotIn(b'name="language_code"', CAPTURED["body"]) + + +class TestResponseParsing(ElevenLabsTestCase): + def test_words_are_joined_without_double_spaces(self): + """Regression M8: ' '.join() across all tokens produced double spaces + and detached punctuation.""" + segments = elevenlabs.ElevenLabsBackend(api_key="sk_test").transcribe(self.wav) + texts = [segment.text for segment in segments] + self.assertEqual(texts, ["Hello world.", "How are you?"]) + for text in texts: + self.assertNotIn(" ", text) + self.assertNotIn(" .", text) + self.assertNotIn(" ?", text) + + def test_timestamps_and_speaker_hints(self): + segments = elevenlabs.ElevenLabsBackend(api_key="sk_test").transcribe(self.wav) + self.assertAlmostEqual(segments[0].start, 0.0) + self.assertAlmostEqual(segments[0].end, 0.9) + self.assertAlmostEqual(segments[1].start, 1.0) + self.assertEqual(segments[0].speaker_hint, "speaker_0") + self.assertEqual(segments[1].speaker_hint, "speaker_1") + + +class TestErrorHandling(ElevenLabsTestCase): + def test_missing_key_is_reported_clearly(self): + with self.assertRaises(TranscriptionError) as ctx: + elevenlabs.ElevenLabsBackend(api_key="").transcribe(self.wav) + self.assertIn("API key", str(ctx.exception)) + + def test_unauthorized_gives_actionable_message(self): + """Regression K2: in the previous version every error message vanished + into a NameError and the user saw nothing at all.""" + BEHAVIOUR["mode"] = "unauthorized" + with self.assertRaises(TranscriptionError) as ctx: + elevenlabs.ElevenLabsBackend(api_key="sk_wrong").transcribe(self.wav) + message = str(ctx.exception) + self.assertIn("401", message) + self.assertIn("rejected", message) + + def test_model_fallback_to_scribe_v1(self): + """If the API rejects scribe_v2, scribe_v1 is tried automatically.""" + BEHAVIOUR["mode"] = "model_error" + backend = elevenlabs.ElevenLabsBackend(api_key="sk_test", model_id="scribe_v2") + segments = backend.transcribe(self.wav) + self.assertTrue(segments) + self.assertEqual(backend.model_id, "scribe_v1") + self.assertIn(b"scribe_v1", CAPTURED["body"]) + + def test_missing_file(self): + with self.assertRaises(TranscriptionError): + elevenlabs.ElevenLabsBackend(api_key="sk_test").transcribe( + os.path.join(self.dir, "does-not-exist.wav")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_events.py b/tests/test_events.py new file mode 100644 index 0000000..cbc7a8d --- /dev/null +++ b/tests/test_events.py @@ -0,0 +1,149 @@ +"""Tests for the worker-to-GUI bridge - a direct regression for K2.""" + +import queue +import threading +import unittest + +from audio_transcriber.events import Failed, Log, Status, UiBridge + + +class FakeRoot: + """Minimal Tk stand-in: after() only remembers the callback.""" + + def __init__(self): + self.scheduled = [] + self.cancelled = [] + + def after(self, _ms, func=None): + self.scheduled.append(func) + return len(self.scheduled) + + def after_cancel(self, ident): + self.cancelled.append(ident) + + +class TestLateBoundExceptionBug(unittest.TestCase): + """The actual core of K2. + + The previous version wrote this inside worker threads: + except Exception as e: + self.root.after(0, lambda: self.transcription_failed(str(e))) + + Python deletes 'e' at the end of the except block. The lambda only ran + later on the Tk event loop - by then the closure cell was empty and it + raised NameError instead of the error message. Result: no dialog, no + reset, a locked user interface. + """ + + def test_old_pattern_raises_nameerror(self): + deferred = queue.Queue() + + def worker(): + try: + raise ValueError("model missing") + except Exception as e: # noqa: F841 + deferred.put(lambda: f"Error: {e}") # the old spelling + + thread = threading.Thread(target=worker) + thread.start() + thread.join() + + with self.assertRaises(NameError): + deferred.get()() + + def test_new_pattern_transports_the_message(self): + bridge = UiBridge(FakeRoot()) + received = [] + bridge.on(Failed, lambda event: received.append(event.message)) + + def worker(): + try: + raise ValueError("model missing") + except Exception as exc: + bridge.post_exception("Transcription failed", exc) + + thread = threading.Thread(target=worker) + thread.start() + thread.join() + + bridge.drain_now() + self.assertEqual(len(received), 1) + self.assertIn("model missing", received[0]) + self.assertIn("Transcription failed", received[0]) + + def test_failed_event_is_a_plain_string(self): + """Events carry data only - no callable with closure state crosses the + thread boundary.""" + event = Failed(message="something went wrong") + self.assertIsInstance(event.message, str) + with self.assertRaises(Exception): + event.message = "immutable" # frozen dataclass + + +class TestBridge(unittest.TestCase): + def setUp(self): + self.root = FakeRoot() + self.bridge = UiBridge(self.root) + + def test_events_are_dispatched_in_order(self): + seen = [] + self.bridge.on(Log, lambda event: seen.append(event.text)) + for index in range(5): + self.bridge.post(Log(text=str(index))) + self.bridge.drain_now() + self.assertEqual(seen, ["0", "1", "2", "3", "4"]) + + def test_events_from_many_threads_arrive(self): + seen = [] + self.bridge.on(Log, lambda event: seen.append(event.text)) + threads = [threading.Thread(target=lambda i=i: self.bridge.post(Log(str(i)))) + for i in range(20)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + self.bridge.drain_now() + self.assertEqual(sorted(seen, key=int), [str(i) for i in range(20)]) + + def test_broken_handler_does_not_stop_the_pump(self): + import contextlib + import io + + def explode(_event): + raise RuntimeError("broken") + + seen = [] + self.bridge.on(Log, explode) + self.bridge.on(Status, lambda event: seen.append(event.text)) + self.bridge.post(Log("whatever")) + self.bridge.post(Status("carry on")) + + # The handler deliberately prints a traceback - keep it quiet here. + with contextlib.redirect_stderr(io.StringIO()): + self.bridge.drain_now() + self.assertEqual(seen, ["carry on"]) + + def test_unknown_event_types_are_ignored(self): + self.bridge.post(Status("no handler")) + self.bridge.drain_now() # must not raise + + def test_pump_reschedules_itself(self): + self.bridge.start() + self.assertTrue(self.root.scheduled) + self.bridge.stop() + self.assertTrue(self.root.cancelled) + + def test_batch_limit_protects_the_gui(self): + """A download must not starve the interface with events.""" + seen = [] + self.bridge.on(Log, lambda event: seen.append(event.text)) + for index in range(500): + self.bridge.post(Log(str(index))) + self.bridge.drain_now() + self.assertEqual(len(seen), 200) + self.bridge.drain_now() + self.assertEqual(len(seen), 400) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_file_upload.py b/tests/test_file_upload.py new file mode 100644 index 0000000..fda0115 --- /dev/null +++ b/tests/test_file_upload.py @@ -0,0 +1,256 @@ +"""Tests for the Audio File Upload and Transcription feature.""" + +import os +import shutil +import tempfile +import unittest +from unittest.mock import MagicMock, patch + +import numpy as np +import soundfile as sf + +from audio_transcriber.audio import dsp +from audio_transcriber.audio.loader import load_audio_file +from audio_transcriber.config import Settings +from audio_transcriber.events import Failed, Finished, Log, Progress, Status +from audio_transcriber.pipeline import FileFinalizer +from audio_transcriber.transcribe.base import Segment, TranscriptionError + + +class TestAudioLoader(unittest.TestCase): + + def setUp(self): + self.temp_dir = tempfile.mkdtemp() + + def tearDown(self): + shutil.rmtree(self.temp_dir, ignore_errors=True) + + def test_nonexistent_file(self): + fake_path = os.path.join(self.temp_dir, "nonexistent.wav") + with self.assertRaises(TranscriptionError) as ctx: + load_audio_file(fake_path) + self.assertIn("not found", str(ctx.exception)) + + def test_empty_file(self): + empty_path = os.path.join(self.temp_dir, "empty.wav") + with open(empty_path, "wb") as f: + f.write(b"") + with self.assertRaises(TranscriptionError) as ctx: + load_audio_file(empty_path) + self.assertIn("empty", str(ctx.exception)) + + def test_load_valid_mono_wav(self): + wav_path = os.path.join(self.temp_dir, "sample.wav") + sample_rate = 44100 + t = np.linspace(0, 1, sample_rate, dtype=np.float32) + sine = 0.5 * np.sin(2 * np.pi * 440 * t) + sf.write(wav_path, sine, sample_rate) + + loaded = load_audio_file(wav_path, target_rate=16000) + self.assertIsInstance(loaded, np.ndarray) + self.assertEqual(loaded.ndim, 1) + self.assertAlmostEqual(len(loaded), 16000, delta=100) + + def test_load_valid_stereo_wav(self): + wav_path = os.path.join(self.temp_dir, "stereo.wav") + sample_rate = 16000 + t = np.linspace(0, 1, sample_rate, dtype=np.float32) + left = 0.5 * np.sin(2 * np.pi * 440 * t) + right = 0.5 * np.sin(2 * np.pi * 880 * t) + stereo = np.column_stack([left, right]) + sf.write(wav_path, stereo, sample_rate) + + loaded = load_audio_file(wav_path, target_rate=16000) + self.assertEqual(loaded.ndim, 1) + self.assertEqual(len(loaded), 16000) + + @patch("audio_transcriber.audio.loader._load_via_ffmpeg") + def test_ffmpeg_fallback_trigger(self, mock_ffmpeg): + mock_ffmpeg.return_value = np.zeros(16000, dtype=np.float32) + m4a_path = os.path.join(self.temp_dir, "dummy.m4a") + with open(m4a_path, "wb") as f: + f.write(b"fake m4a header data") + + loaded = load_audio_file(m4a_path, target_rate=16000) + mock_ffmpeg.assert_called_once() + self.assertEqual(len(loaded), 16000) + + +class TestMockBridge: + def __init__(self): + self.events = [] + + def post(self, event): + self.events.append(event) + + def post_exception(self, msg, exc): + self.events.append(Failed(message=f"{msg}: {exc}")) + + +class TestMockBackend: + def __init__(self, segments=None): + self.segments = segments or [ + Segment(start=0.0, end=2.5, text="Hello world from file.", track="file") + ] + self.cancelled = False + + def transcribe(self, path, language="de", log=None, track="", progress=None): + if log: + log("Mock transcribing...\n") + return self.segments + + def cancel(self): + self.cancelled = True + + +class TestFileFinalizer(unittest.TestCase): + + def setUp(self): + self.temp_dir = tempfile.mkdtemp() + self.out_dir = os.path.join(self.temp_dir, "out") + self.tmp_dir = os.path.join(self.temp_dir, "tmp") + os.makedirs(self.out_dir, exist_ok=True) + os.makedirs(self.tmp_dir, exist_ok=True) + + # Only TMP_DIR is read from the pipeline module namespace. The output + # directory comes from Settings.get_output_dir(), so patching an + # OUT_DIR here would do nothing and the tests would quietly write into + # the real output/ folder - every Settings() below sets output_dir. + self.paths_patcher = patch.multiple( + "audio_transcriber.pipeline", + TMP_DIR=self.tmp_dir + ) + self.paths_patcher.start() + + def tearDown(self): + self.paths_patcher.stop() + shutil.rmtree(self.temp_dir, ignore_errors=True) + + def test_file_finalizer_success(self): + # Create a test audio file + wav_path = os.path.join(self.temp_dir, "test_input.wav") + sample_rate = 16000 + t = np.linspace(0, 2, sample_rate * 2, dtype=np.float32) + audio_data = 0.5 * np.sin(2 * np.pi * 440 * t) + sf.write(wav_path, audio_data, sample_rate) + + bridge = TestMockBridge() + settings = Settings(output_dir=self.out_dir) + backend = TestMockBackend() + + finalizer = FileFinalizer(bridge, settings, backend_factory=lambda s: backend) + thread = finalizer.run_async(wav_path, base_name="test_file") + thread.join(timeout=5.0) + + # Check posted events + finished_events = [e for e in bridge.events if isinstance(e, Finished)] + self.assertEqual(len(finished_events), 1) + finished = finished_events[0] + self.assertIn("Hello world from file.", finished.text) + self.assertTrue(os.path.exists(finished.txt_path)) + self.assertTrue(os.path.exists(finished.audio_path)) + + def test_file_finalizer_custom_output_dir(self): + wav_path = os.path.join(self.temp_dir, "test_custom.wav") + custom_out = os.path.join(self.temp_dir, "custom_target_folder") + sample_rate = 16000 + t = np.linspace(0, 2, sample_rate * 2, dtype=np.float32) + audio_data = 0.5 * np.sin(2 * np.pi * 440 * t) + sf.write(wav_path, audio_data, sample_rate) + + bridge = TestMockBridge() + settings = Settings(output_dir=custom_out) + backend = TestMockBackend() + + finalizer = FileFinalizer(bridge, settings, backend_factory=lambda s: backend) + thread = finalizer.run_async(wav_path, base_name="custom_file") + thread.join(timeout=5.0) + + finished_events = [e for e in bridge.events if isinstance(e, Finished)] + self.assertEqual(len(finished_events), 1) + finished = finished_events[0] + self.assertEqual(os.path.dirname(finished.txt_path), os.path.abspath(custom_out)) + self.assertEqual(os.path.dirname(finished.audio_path), os.path.abspath(custom_out)) + self.assertTrue(os.path.exists(finished.txt_path)) + self.assertTrue(os.path.exists(finished.audio_path)) + + def test_file_finalizer_derives_base_name_from_file(self): + """Without an explicit base_name the name comes from the file itself. + + Regression: run_async called paths.safe_output_name() while pipeline.py + only imported OUT_DIR/TMP_DIR from .paths, so this branch raised + NameError. Every other test passed base_name explicitly and never + reached it. + """ + wav_path = os.path.join(self.temp_dir, "Team Meeting 2026.wav") + sample_rate = 16000 + t = np.linspace(0, 2, sample_rate * 2, dtype=np.float32) + sf.write(wav_path, 0.5 * np.sin(2 * np.pi * 440 * t), sample_rate) + + bridge = TestMockBridge() + settings = Settings(output_dir=self.out_dir) + backend = TestMockBackend() + + finalizer = FileFinalizer(bridge, settings, backend_factory=lambda s: backend) + thread = finalizer.run_async(wav_path) # no base_name + thread.join(timeout=5.0) + + failed = [e for e in bridge.events if isinstance(e, Failed)] + self.assertEqual(failed, [], f"unexpected failure: {failed}") + + finished = [e for e in bridge.events if isinstance(e, Finished)] + self.assertEqual(len(finished), 1) + self.assertEqual(os.path.basename(finished[0].txt_path), + "Team Meeting 2026.txt") + + def test_file_finalizer_silent_audio(self): + + silent_path = os.path.join(self.temp_dir, "silent.wav") + sf.write(silent_path, np.zeros(16000, dtype=np.float32), 16000) + + bridge = TestMockBridge() + settings = Settings(output_dir=self.out_dir) + backend = TestMockBackend() + + finalizer = FileFinalizer(bridge, settings, backend_factory=lambda s: backend) + thread = finalizer.run_async(silent_path, base_name="silent_test") + thread.join(timeout=5.0) + + failed_events = [e for e in bridge.events if isinstance(e, Failed)] + self.assertEqual(len(failed_events), 1) + self.assertIn("silent", failed_events[0].message.lower()) + + +class TestSaveTranscript(unittest.TestCase): + + def setUp(self): + self.temp_dir = tempfile.mkdtemp() + + def tearDown(self): + shutil.rmtree(self.temp_dir, ignore_errors=True) + + @patch("tkinter.filedialog.asksaveasfilename") + @patch("tkinter.messagebox.showinfo") + def test_save_transcript_success(self, mock_info, mock_dialog): + save_path = os.path.join(self.temp_dir, "custom_transcript.txt") + mock_dialog.return_value = save_path + + mock_app = MagicMock() + mock_app.transcript.text.get.return_value = "[00:01] Hello test transcript" + mock_app.recording_base_name = "test_rec" + mock_app.filename_entry.get.return_value = "default_name" + + from audio_transcriber.ui.app import RecorderApp + RecorderApp._save_transcript(mock_app) + + mock_dialog.assert_called_once() + self.assertTrue(os.path.exists(save_path)) + with open(save_path, "r", encoding="utf-8") as f: + content = f.read() + self.assertIn("[00:01] Hello test transcript", content) + mock_app.status.set.assert_called_once() + + +if __name__ == "__main__": + unittest.main() + diff --git a/tests/test_gui_smoke.py b/tests/test_gui_smoke.py new file mode 100644 index 0000000..29cf7b3 --- /dev/null +++ b/tests/test_gui_smoke.py @@ -0,0 +1,166 @@ +"""Smoke test for the user interface. + +Builds the complete window invisibly, checks that no exception is raised and +shuts it down cleanly. Covers audit finding M7 (the previous version had no +WM_DELETE_WINDOW handler and never terminated PyAudio). +""" + +import os +import tempfile +import unittest + +try: + import tkinter as tk + _HAS_TK = True +except Exception: # pragma: no cover + _HAS_TK = False + + +def _can_open_window(): + if not _HAS_TK: + return False + try: + root = tk.Tk() + root.destroy() + return True + except Exception: + return False + + +@unittest.skipUnless(_can_open_window(), "no graphical display available") +class TestAppLifecycle(unittest.TestCase): + def setUp(self): + from audio_transcriber import config, paths + from audio_transcriber.ui import icons + icons._ICON_CACHE.clear() + self.tmpdir = tempfile.mkdtemp() + self._orig_cfg = config.CFG_PATH + # Never touch the user's real settings + config.CFG_PATH = os.path.join(self.tmpdir, "settings.json") + self.paths = paths + + def tearDown(self): + import shutil + from audio_transcriber import config + from audio_transcriber.ui import icons + icons._ICON_CACHE.clear() + config.CFG_PATH = self._orig_cfg + shutil.rmtree(self.tmpdir, ignore_errors=True) + + + def test_build_and_close(self): + from audio_transcriber import config + from audio_transcriber.ui.app import RecorderApp + + root = tk.Tk() + root.withdraw() # build it invisibly + app = None + try: + app = RecorderApp(root) + root.update() # run one event cycle + + self.assertTrue(app.model_combo["values"]) + self.assertTrue(app.lang_combo["values"]) + self.assertEqual(str(app.start_btn["state"]), "normal") + self.assertEqual(str(app.stop_btn["state"]), "disabled") + + # Collect the settings from the interface + app._sync_settings_from_ui() + self.assertIsInstance(app.settings.mic_gain_db, float) + self.assertIn(app.settings.language, + [code for _label, code in config.LANGUAGE_CHOICES]) + + # Tick the level meters once + app._tick() + root.update() + finally: + if app is not None: + app.on_close() + else: # pragma: no cover + root.destroy() + + def test_gain_slider_updates_meters(self): + """Regression test: gain sliders must dynamically adjust the VU meter levels.""" + from audio_transcriber.ui.app import RecorderApp + from unittest.mock import PropertyMock, patch + + root = tk.Tk() + root.withdraw() + app = None + try: + app = RecorderApp(root) + with patch.object(type(app.engine), 'mic_level', new_callable=PropertyMock) as mock_mic: + mock_mic.return_value = 0.1 # ~ -20 dB + + # 0 dB Gain -> ~ -20 dB + app._on_mic_gain(0.0) + app._tick() + db_0 = app.mic_meter.db + self.assertAlmostEqual(db_0, -20.0, delta=1.0) + + # +10 dB Gain -> ~ -10 dB (+10 dB shift) + app._on_mic_gain(10.0) + app._tick() + db_plus_10 = app.mic_meter.db + self.assertAlmostEqual(db_plus_10, -10.0, delta=1.0) + + # -10 dB Gain -> ~ -30 dB (-10 dB shift) + app._on_mic_gain(-10.0) + app._tick() + db_minus_10 = app.mic_meter.db + self.assertAlmostEqual(db_minus_10, -30.0, delta=1.0) + + self.assertGreater(db_plus_10, db_0) + self.assertLess(db_minus_10, db_0) + finally: + if app is not None: + app.on_close() + else: + root.destroy() + + + def test_output_name_sanitising(self): + """Path traversal through the file name field must be impossible. + + The results are identical on every platform. os.path.basename() used to + decide this by host rules, so a backslash path sanitised one way on + Windows and another on Linux - caught by CI on the first Linux run. + """ + cases = { + "..\\..\\windows\\system32\\evil": "evil", + "my_meeting.wav": "my_meeting", + " ": "my_meeting", + "": "my_meeting", + "C:/temp/report.wav": "report", + 'inlid:"|?*': "in_va_lid_____", + # POSIX separators must be blocked on Windows just as well + "../../etc/passwd": "passwd", + "/etc/shadow": "shadow", + "..\\../mix/ed\\name.wav": "name", + } + for raw, expected in cases.items(): + with self.subTest(raw=raw): + self.assertEqual(self.paths.safe_output_name(raw), expected) + + def test_output_name_never_escapes_its_directory(self): + """The property behind the table above, stated directly.""" + hostile = [ + "../" * 8 + "etc/passwd", + "..\\" * 8 + "windows\\system32\\cmd.exe", + "/absolute/path", "C:\\Windows\\System32", "...", "..", ".", + "con.txt", " .. ", "a/b\\c/d", + ] + for raw in hostile: + with self.subTest(raw=raw): + name = self.paths.safe_output_name(raw) + self.assertNotIn("/", name) + self.assertNotIn("\\", name) + self.assertNotIn("..", name) + self.assertTrue(name) + # Must stay inside the directory it is joined onto + joined = os.path.normpath(os.path.join("/base", name)) + self.assertTrue(joined.replace("\\", "/").startswith("/base/")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py new file mode 100644 index 0000000..e3147e5 --- /dev/null +++ b/tests/test_pipeline.py @@ -0,0 +1,253 @@ +"""End-to-end tests of the post-processing (audit findings K1, K4, H5).""" + +import os +import queue +import shutil +import tempfile +import unittest + +import numpy as np +import soundfile as sf + +from audio_transcriber import config, pipeline +from audio_transcriber.audio import capture, dsp +from audio_transcriber.events import Failed, Finished, Log, Progress, Status +from audio_transcriber.transcribe.base import Backend, Segment + +RATE = 48000 + + +class FakeBridge: + """Collects events instead of forwarding them to Tkinter.""" + + def __init__(self): + self.events = queue.Queue() + + def post(self, event): + self.events.put(event) + + def post_exception(self, prefix, exc): + self.events.put(Failed(message=f"{prefix}: {exc}")) + + def all(self): + items = [] + while not self.events.empty(): + items.append(self.events.get()) + return items + + def first(self, event_type, events=None): + for event in (events if events is not None else self.all()): + if isinstance(event, event_type): + return event + return None + + +class FakeBackend(Backend): + """Returns predefined segments per track and counts the calls.""" + + calls = [] + + def __init__(self, per_track): + self.per_track = per_track + + def transcribe(self, wav_path, language="de", log=None, track="", progress=None): + FakeBackend.calls.append((track, wav_path, language)) + return [Segment(start, end, text, track) + for start, end, text in self.per_track.get(track, [])] + + def cancel(self): + pass + + +class TestFinalizer(unittest.TestCase): + def setUp(self): + FakeBackend.calls = [] + self.dir = tempfile.mkdtemp() + self.out = os.path.join(self.dir, "output") + self.tmp = os.path.join(self.out, ".tmp") + os.makedirs(self.tmp) + + # Only TMP_DIR is read from the pipeline module namespace. The output + # directory comes from Settings.get_output_dir(), so it has to be + # steered via output_dir below - patching a pipeline.OUT_DIR would do + # nothing and the run would land in the real output/ folder. + self._orig_tmp = pipeline.TMP_DIR + pipeline.TMP_DIR = self.tmp + + self.settings = config.Settings(model_index=3, mic_gain_db=-8.0, + loop_gain_db=10.0, language="en", + keep_raw_tracks=False, + output_dir=self.out) + self.bridge = FakeBridge() + + def tearDown(self): + pipeline.TMP_DIR = self._orig_tmp + shutil.rmtree(self.dir, ignore_errors=True) + + # ------------------------------------------------------------------ + def _make_recording(self, seconds=10.0, bursts=None): + """Two raw tracks made of noise bursts. + + Default: you speak from 1-3 s, the other party from 5-7 s. Segments a + test claims later must also exist acoustically here - otherwise the + hallucination filter drops them, and rightly so. + """ + bursts = bursts or {"mic": [(1.0, 3.0)], "sys": [(5.0, 7.0)]} + rng = np.random.default_rng(4) + result = capture.RecordingResult() + for kind in ("mic", "sys"): + data = np.zeros(int(seconds * RATE), dtype=np.float32) + for start, end in bursts.get(kind, []): + i_start, i_end = int(start * RATE), int(end * RATE) + data[i_start:i_end] = rng.normal(0, 0.2, i_end - i_start) + path = os.path.join(self.tmp, f"session.{kind}.raw.wav") + sf.write(path, data, RATE, subtype="PCM_16") + setattr(result, kind, capture.TrackResult( + path=path, rate=RATE, frames=len(data), device_name=kind)) + return result + + def _run(self, recording, per_track): + finalizer = pipeline.Finalizer( + self.bridge, self.settings, + backend_factory=lambda s: FakeBackend(per_track)) + finalizer._process(recording, "session") + return self.bridge.all() + + # ------------------------------------------------------------------ + def test_full_run_produces_mix_and_transcript(self): + recording = self._make_recording() + events = self._run(recording, { + "mic": [(1.0, 3.0, "Hello, can you hear me?")], + "sys": [(5.0, 7.0, "Yes, loud and clear.")], + }) + + finished = self.bridge.first(Finished, events) + self.assertIsNotNone(finished, "no Finished event was posted") + + # Audible mixdown: stereo, 16 kHz + info = sf.info(finished.audio_path) + self.assertEqual(info.channels, 2) + self.assertEqual(info.samplerate, dsp.TARGET_RATE) + self.assertAlmostEqual(info.frames / info.samplerate, 10.0, places=1) + + with open(finished.txt_path, encoding="utf-8") as handle: + text = handle.read() + self.assertIn("[00:01] [You]: Hello, can you hear me?", text) + self.assertIn("[00:05] [Participant]: Yes, loud and clear.", text) + + def test_both_tracks_are_transcribed_separately(self): + """The core decision: separate tracks instead of guessed attribution.""" + self.settings.separate_tracks = True + self._run(self._make_recording(), {"mic": [(1.0, 3.0, "A")], + "sys": [(5.0, 7.0, "B")]}) + tracks = sorted(call[0] for call in FakeBackend.calls) + self.assertEqual(tracks, ["mic", "sys"]) + + def test_transcript_covers_the_end_of_the_recording(self): + """Regression K4: in the previous version the saved transcript was + missing the end of the recording (16.6 % measured on a real file) + because the last live pass was written as the final result.""" + recording = self._make_recording( + seconds=10.0, + bursts={"mic": [(1.0, 3.0)], "sys": [(5.0, 7.0), (8.5, 9.8)]}) + events = self._run(recording, { + "mic": [(1.0, 3.0, "Beginning")], + "sys": [(5.0, 7.0, "Middle"), (8.5, 9.8, "And this is the end.")], + }) + finished = self.bridge.first(Finished, events) + self.assertIn("And this is the end.", finished.text) + # The last segment sits in the final 15 % of the recording - exactly + # the part the previous version lost systematically. + self.assertTrue(finished.text.rstrip().endswith("And this is the end.")) + + def test_cloud_backend_is_reached_when_selected(self): + """Regression K1: with the live preview enabled the cloud path was + unreachable in the previous version.""" + self.settings.model_index = 0 # ElevenLabs + self.settings.live_transcribe = True + used = [] + + def factory(settings): + used.append(settings.uses_cloud()) + return FakeBackend({"mic": [(1.0, 3.0, "Cloud")], "sys": []}) + + finalizer = pipeline.Finalizer(self.bridge, self.settings, + backend_factory=factory) + finalizer._process(self._make_recording(), "session") + self.assertTrue(used and all(used), "the cloud path was not used") + + def test_gain_settings_do_not_change_speaker_assignment(self): + """Regression H2 at pipeline level.""" + segments = {"mic": [(1.0, 3.0, "I am talking.")], + "sys": [(5.0, 7.0, "And so am I.")]} + + results = [] + for mic_gain, sys_gain in ((0.0, 0.0), (-8.0, 10.0), (12.0, -15.0)): + self.setUp() + self.settings.mic_gain_db = mic_gain + self.settings.loop_gain_db = sys_gain + events = self._run(self._make_recording(), segments) + results.append(self.bridge.first(Finished, events).text) + self.assertEqual(len(set(results)), 1, + "the gain sliders influence speaker attribution") + + def test_temporary_tracks_are_removed(self): + recording = self._make_recording() + raw_paths = [recording.mic.path, recording.sys.path] + self._run(recording, {"mic": [(1.0, 3.0, "A")], "sys": []}) + for path in raw_paths: + self.assertFalse(os.path.exists(path), f"{path} was left behind") + leftovers = [name for name in os.listdir(self.tmp) + if name.endswith(".asr.wav")] + self.assertEqual(leftovers, []) + + def test_keep_raw_tracks_option(self): + self.settings.keep_raw_tracks = True + recording = self._make_recording() + self._run(recording, {"mic": [(1.0, 3.0, "A")], "sys": []}) + self.assertTrue(os.path.exists(recording.mic.path)) + + def test_silent_recording_reports_clear_error(self): + """Regression M3: the previous version only said 'no data'.""" + recording = capture.RecordingResult() + for kind in ("mic", "sys"): + path = os.path.join(self.tmp, f"silent.{kind}.raw.wav") + sf.write(path, np.zeros(RATE * 3, dtype=np.float32), RATE, + subtype="PCM_16") + setattr(recording, kind, capture.TrackResult( + path=path, rate=RATE, frames=RATE * 3, device_name=kind)) + + finalizer = pipeline.Finalizer( + self.bridge, self.settings, + backend_factory=lambda s: FakeBackend({})) + finalizer._run(recording, "silent") + + failed = self.bridge.first(Failed) + self.assertIsNotNone(failed) + self.assertIn("silent", failed.message.lower()) + + def test_backend_error_becomes_failed_event(self): + """Regression K2: errors must not disappear into a NameError.""" + class ExplodingBackend(FakeBackend): + def transcribe(self, *args, **kwargs): + raise RuntimeError("model could not be loaded") + + finalizer = pipeline.Finalizer( + self.bridge, self.settings, + backend_factory=lambda s: ExplodingBackend({})) + finalizer._run(self._make_recording(), "session") + + failed = self.bridge.first(Failed) + self.assertIsNotNone(failed, "the error was swallowed") + self.assertIn("model could not be loaded", failed.message) + + def test_start_offset_is_applied(self): + recording = self._make_recording() + recording.sys.start_offset_s = 0.5 + events = self._run(recording, {"mic": [], "sys": [(5.5, 7.5, "Offset")]}) + finished = self.bridge.first(Finished, events) + self.assertIn("Offset", finished.text) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_timestamps.py b/tests/test_timestamps.py new file mode 100644 index 0000000..a744e23 --- /dev/null +++ b/tests/test_timestamps.py @@ -0,0 +1,74 @@ +"""Tests for the timestamp parser (audit findings H7 and N4).""" + +import re +import unittest + +from audio_transcriber.transcribe.base import (format_timestamp, parse_line, + parse_timestamp) + +# Regex of the previous version - kept here as the regression reference. +OLD_RE = re.compile( + r'^\s*\[(\d{1,2}):(\d{2})(?::(\d{2}))?(?:\.\d+)?\s*-->\s*' + r'(\d{1,2}):(\d{2})(?::(\d{2}))?(?:\.\d+)?\]\s*(.*)$') + + +class TestParsing(unittest.TestCase): + def test_whisper_default_format(self): + line = "[00:00:03.480 --> 00:00:06.120] Good morning everyone." + start, end, text = parse_line(line) + self.assertAlmostEqual(start, 3.480, places=3) + self.assertAlmostEqual(end, 6.120, places=3) + self.assertEqual(text, "Good morning everyone.") + + def test_milliseconds_are_kept(self): + """Regression H7: the previous version discarded the fractional part, + so the analysis window for speaker attribution could be off by up to a + full second.""" + line = "[00:00:03.900 --> 00:00:04.100] Yes." + start, end, _ = parse_line(line) + self.assertAlmostEqual(start, 3.9, places=3) + self.assertAlmostEqual(end, 4.1, places=3) + + old = OLD_RE.match(line.strip()) + old_start = int(old.group(1)) * 3600 + int(old.group(2)) * 60 + int(old.group(3)) + self.assertEqual(old_start, 3) # truncated + self.assertGreater(start - old_start, 0.8) # almost a second of error + + def test_hours(self): + start, end, _ = parse_line("[01:23:45.000 --> 01:23:47.500] Text") + self.assertAlmostEqual(start, 5025.0, places=3) + self.assertAlmostEqual(end, 5027.5, places=3) + + def test_short_mm_ss_format(self): + start, end, _ = parse_line("[02:15 --> 02:19] Short form") + self.assertAlmostEqual(start, 135.0) + self.assertAlmostEqual(end, 139.0) + + def test_non_matching_lines(self): + for line in ("whisper_model_load: loading model", "", " ", + "system_info: n_threads = 4"): + self.assertIsNone(parse_line(line)) + + def test_parse_timestamp_units(self): + self.assertEqual(parse_timestamp("0", "30", None, None), 30.0) + self.assertEqual(parse_timestamp("1", "30", None, None), 90.0) + self.assertEqual(parse_timestamp("1", "00", "00", None), 3600.0) + + +class TestFormatting(unittest.TestCase): + def test_minutes(self): + self.assertEqual(format_timestamp(0), "[00:00]") + self.assertEqual(format_timestamp(75.9), "[01:15]") + self.assertEqual(format_timestamp(3599), "[59:59]") + + def test_hours_regression(self): + """Regression N4: the previous version formatted 90 minutes as [90:12].""" + self.assertEqual(format_timestamp(5412), "[01:30:12]") + self.assertEqual(format_timestamp(7200), "[02:00:00]") + + def test_negative_is_clamped(self): + self.assertEqual(format_timestamp(-5), "[00:00]") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_ui_helpers.py b/tests/test_ui_helpers.py new file mode 100644 index 0000000..f9ae36b --- /dev/null +++ b/tests/test_ui_helpers.py @@ -0,0 +1,73 @@ +"""Tests for the pure UI helper functions (no window required).""" + +import unittest + +from audio_transcriber.ui import theme, widgets + + +class TestColourMix(unittest.TestCase): + def test_endpoints(self): + self.assertEqual(theme.mix("#000000", "#ffffff", 0.0), "#000000") + self.assertEqual(theme.mix("#000000", "#ffffff", 1.0), "#ffffff") + + def test_midpoint(self): + self.assertEqual(theme.mix("#000000", "#ffffff", 0.5), "#808080") + + def test_channelwise(self): + self.assertEqual(theme.mix("#ff0000", "#0000ff", 0.5), "#800080") + + +class TestTimestampDetection(unittest.TestCase): + def test_accepts_timestamps(self): + for token in ("[00:00]", "[01:15]", "[01:30:12]", "[99:59]"): + self.assertTrue(widgets._looks_like_timestamp(token), token) + + def test_rejects_speakers_and_noise(self): + for token in ("[You]", "[Participant]", "[Participant A]", + "[ERROR]", "[]", "[a:b]"): + self.assertFalse(widgets._looks_like_timestamp(token), token) + + +class TestSpeakerPattern(unittest.TestCase): + def test_matches_speaker_prefix(self): + match = widgets._SPEAKER_RE.match(" [You]: Hello everyone.") + self.assertIsNotNone(match) + self.assertEqual(match.group(1).strip(), "[You]:") + self.assertEqual(match.group(2).strip(), "Hello everyone.") + + def test_matches_other_speaker(self): + match = widgets._SPEAKER_RE.match("[Participant]: Sure, happy to.") + self.assertIsNotNone(match) + self.assertIn("Participant", match.group(1)) + + def test_ignores_plain_text(self): + for line in ("Recording started.", "whisper_model_load: loading", + "Quality filter: 2 crosstalk segment(s) dropped."): + self.assertIsNone(widgets._SPEAKER_RE.match(line), line) + + def test_keeps_text_with_colons(self): + match = widgets._SPEAKER_RE.match("[You]: It is 12:30 now.") + self.assertEqual(match.group(2).strip(), "It is 12:30 now.") + + +class TestRoundRectGeometry(unittest.TestCase): + def test_radius_is_clamped_to_the_shape(self): + """An oversized radius must not distort the polygon.""" + captured = {} + + class FakeCanvas: + def create_polygon(self, points, **kwargs): + captured["points"] = points + return 1 + + theme.round_rect(FakeCanvas(), 0, 0, 10, 4, 50) + xs = captured["points"][0::2] + ys = captured["points"][1::2] + self.assertGreaterEqual(min(xs), 0) + self.assertLessEqual(max(xs), 10) + self.assertGreaterEqual(min(ys), 0) + self.assertLessEqual(max(ys), 4) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_whisper_integration.py b/tests/test_whisper_integration.py new file mode 100644 index 0000000..5670ca0 --- /dev/null +++ b/tests/test_whisper_integration.py @@ -0,0 +1,117 @@ +"""Integration tests against the real whisper-cli.exe. + +Skipped when the binary or the model is missing. +""" + +import os +import unittest + +import numpy as np +import soundfile as sf + +from audio_transcriber import paths +from audio_transcriber.transcribe.base import TranscriptionError +from audio_transcriber.transcribe.whispercpp import WhisperCppBackend + +TINY = paths.model_path("tiny") +SAMPLE = os.path.join(paths.OUT_DIR, ".tmp", "integration_sample.wav") + + +def _have_whisper(): + return os.path.exists(paths.WHISPER_EXE) and os.path.exists(TINY) + + +def _find_sample_audio(): + """Any existing recording in output/ works as sample material.""" + if not os.path.isdir(paths.OUT_DIR): + return None + for name in sorted(os.listdir(paths.OUT_DIR)): + if name.endswith(".wav"): + return os.path.join(paths.OUT_DIR, name) + return None + + +class TestCommandBuilder(unittest.TestCase): + def test_default_flags(self): + backend = WhisperCppBackend("small", threads=10, use_vad=True) + command = backend.build_command("whisper.exe", "model.bin", "a.wav", "de", + vad_model="vad.bin") + self.assertIn("-t", command) + self.assertEqual(command[command.index("-t") + 1], "10") + self.assertIn("-ng", command) # Vulkan crashes on this build + self.assertIn("-np", command) # no diagnostics in the transcript + self.assertIn("-sns", command) # suppress non-speech tokens + self.assertIn("--vad", command) + self.assertEqual(command[command.index("-l") + 1], "de") + + def test_greedy_mode_for_live_preview(self): + command = WhisperCppBackend("tiny", greedy=True).build_command( + "w.exe", "m.bin", "a.wav", "de") + self.assertEqual(command[command.index("-bs") + 1], "1") + self.assertEqual(command[command.index("-bo") + 1], "1") + + def test_vad_omitted_when_model_missing(self): + command = WhisperCppBackend("tiny", use_vad=True).build_command( + "w.exe", "m.bin", "a.wav", "de", vad_model=None) + self.assertNotIn("--vad", command) + + def test_vad_is_off_by_default(self): + """It merges distant speech regions on this build - see whispercpp.py.""" + self.assertFalse(WhisperCppBackend("tiny").use_vad) + + def test_missing_audio_file_raises(self): + with self.assertRaises(TranscriptionError): + WhisperCppBackend("tiny").transcribe("does-not-exist.wav") + + +@unittest.skipUnless(_have_whisper(), "whisper-cli.exe or ggml-tiny.bin missing") +class TestRealTranscription(unittest.TestCase): + @classmethod + def setUpClass(cls): + os.makedirs(os.path.dirname(SAMPLE), exist_ok=True) + source = _find_sample_audio() + if source: + data, rate = sf.read(source, dtype="float32", always_2d=True) + mono = data.mean(axis=1)[:rate * 25] + sf.write(SAMPLE, mono, rate, subtype="PCM_16") + else: + sf.write(SAMPLE, np.zeros(16000 * 3, dtype=np.float32), 16000, + subtype="PCM_16") + + def test_produces_segments_with_subsecond_timestamps(self): + backend = WhisperCppBackend("tiny", threads=8, use_vad=False) + segments = backend.transcribe(SAMPLE, language="de", track="mic") + + if not segments: + self.skipTest("the sample file contains no recognisable speech") + + self.assertTrue(all(segment.track == "mic" for segment in segments)) + self.assertTrue(all(segment.end >= segment.start for segment in segments)) + starts = [segment.start for segment in segments] + self.assertEqual(starts, sorted(starts)) + # Regression H7: at least one timestamp carries a fractional part + self.assertTrue(any(abs(segment.start - round(segment.start)) > 1e-6 + for segment in segments), + "timestamps were rounded to whole seconds") + + def test_stderr_is_drained_without_deadlock(self): + """Regression H10: stderr was an unread pipe. If this call completes, + draining is proven to work.""" + backend = WhisperCppBackend("tiny", threads=8, use_vad=False) + backend.transcribe(SAMPLE, language="de") + + def test_cancel_terminates_process(self): + import threading + import time + backend = WhisperCppBackend("tiny", threads=1, use_vad=False) + threading.Timer(1.0, backend.cancel).start() + started = time.monotonic() + try: + backend.transcribe(SAMPLE, language="de") + except TranscriptionError: + pass + self.assertLess(time.monotonic() - started, 30.0) + + +if __name__ == "__main__": + unittest.main()