Bound local audio transcription requests

python/stt/whisperService/README.md

@@ -7,16 +7,16 @@ You can launch the service by running e.g. `python faster-whisper.py`.
 
 ### Prerequisites
 
-- [Python 3](https://wiki.python.org/moin/BeginnersGuide/Download)
+- [Python 3.9 or newer](https://wiki.python.org/moin/BeginnersGuide/Download)
 - [Pip](https://pip.pypa.io/en/stable/installation/)
 - **Windows**
   - [FFMpeg](https://www.gyan.dev/ffmpeg/builds/) - you can run `winget install ffmpeg` to install the package
 - **Linux**
   - `apt-get -y install python3-pyaudio portaudio19-dev` - needed to install `pyaudio`
-  - `apt-get -y install ffmpeg libavcodec-extra` - needed for pydub
+  - `apt-get -y install ffmpeg libavcodec-extra` - needed for audio decoding
   - `apt-get -y install cudnn9-cuda-12`- Follow instruction in [NVIDIA cuDNN on Linux](https://docs.nvidia.com/deeplearning/cudnn/latest/installation/linux.html) to enable NVIDIA network repository.
 - **MacOS**
-  - `brew install portaudio` - needed to install `pyaudio`
+  - `brew install ffmpeg portaudio` - needed for audio decoding and `pyaudio`
 
 ### Install
 
@@ -46,3 +46,20 @@ You can connect to the whisper service using the example "whisperClient" project
 - Start the web UI using `pnpm run start`
 
 This web client will capture audio from microphone, send to the local service for transcription and show the result.
+
+## Input and processing limits
+
+The service accepts audio files up to **30 MiB** and recordings up to **five minutes**. The complete multipart request has a separate 30 MiB + 64 KiB limit for form headers. Both declared and streamed request sizes are checked before multipart parsing. Once a request reaches the front of the queue, its complete body must arrive within **30 seconds**, including continuously streamed uploads. An upload timeout returns HTTP 408 with an `error` message and releases capacity for the next request; queueing and transcription do not count toward this deadline.
+
+FFmpeg decodes only the first audio stream into 16 kHz mono 16-bit PCM, with bounded output and a 30-second decoding timeout. Oversized files or recordings return HTTP 413, invalid audio returns HTTP 400, and a decoding timeout returns HTTP 422. These responses contain an `error` message instead of transcription text.
+
+One request is processed at a time, and one additional request can wait. Further requests receive HTTP 503 with `Retry-After: 1`. Processing remains reserved until the worker finishes, including when its client cancels. The Shell and example client display failures and stop the failed recording; start a new recording to retry. Delayed results or errors from an earlier recording do not affect a new recording.
+
+## Offline tests
+
+The tests use the installed FFmpeg executable and a small model stand-in, without downloading a Whisper model or requiring a GPU:
+
+```shell
+pip install -r requirements-test.txt
+python -m unittest -v test_audio_limits
+```

python/stt/whisperService/faster-whisper.py

@@ -1,25 +1,122 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
-from io import BytesIO
-from pydub import AudioSegment
+import asyncio
+from pathlib import Path
+import subprocess
+import tempfile
 import uvicorn
 import numpy as np
 import torch
-from fastapi import FastAPI, UploadFile, File
+from fastapi import FastAPI, UploadFile, File, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
+from starlette.datastructures import Headers, State
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
 import logging
 from faster_whisper import WhisperModel
 
 import os
-os.environ['KMP_DUPLICATE_LIB_OK']='True'
+os.environ["KMP_DUPLICATE_LIB_OK"] = "True"
 
 # Configure logging
-logging.basicConfig(level=logging.ERROR, format='%(asctime)s - %(levelname)s - %(message)s')
+logging.basicConfig(
+    level=logging.ERROR,
+    format="%(asctime)s - %(levelname)s - %(message)s",
+)
+
+MAX_UPLOAD_BYTES = 30 * 1024 * 1024
+MAX_REQUEST_BYTES = MAX_UPLOAD_BYTES + 64 * 1024
+MAX_AUDIO_SECONDS = 300
+SAMPLE_RATE = 16000
+UPLOAD_TIMEOUT_SECONDS = 30
+DECODE_TIMEOUT_SECONDS = 30
+READ_CHUNK_BYTES = 64 * 1024
+
+
+class TranscriptionInputMiddleware:
+    def __init__(self, app: ASGIApp, state: State) -> None:
+        self.app = app
+        self.state = state
+        self.lock = asyncio.Lock()
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if (
+            scope["type"] != "http"
+            or scope["method"] != "POST"
+            or scope["path"].rstrip("/") != "/transcribe"
+        ):
+            await self.app(scope, receive, send)
+            return
+
+        async def reject(status: int, message: str) -> None:
+            response = JSONResponse(
+                {"error": message},
+                status_code=status,
+                headers={"Retry-After": "1"} if status == 503 else None,
+            )
+            await response(scope, receive, send)
+
+        content_length = Headers(scope=scope).get("content-length")
+        if content_length is not None:
+            if not content_length.isdecimal() or len(content_length) > 20:
+                await reject(400, "Invalid request size.")
+                return
+            if int(content_length) > MAX_REQUEST_BYTES:
+                await reject(
+                    413, "The audio upload exceeds the request size limit."
+                )
+                return
+        if self.state.pending_transcriptions >= 2:
+            await reject(503, "The transcription service is busy. Try again shortly.")
+            return
+
+        self.state.pending_transcriptions += 1
+        try:
+            async with self.lock:
+                with tempfile.SpooledTemporaryFile(max_size=1024 * 1024) as body:
+                    size = 0
+                    loop = asyncio.get_running_loop()
+                    upload_deadline = loop.time() + UPLOAD_TIMEOUT_SECONDS
+                    while True:
+                        try:
+                            message = await asyncio.wait_for(
+                                receive(), timeout=upload_deadline - loop.time()
+                            )
+                        except asyncio.TimeoutError:
+                            await reject(408, "The audio upload exceeded the time limit.")
+                            return
+                        if message["type"] == "http.disconnect":
+                            return
+                        if message["type"] != "http.request":
+                            raise RuntimeError("Unexpected transcription request message.")
+                        chunk = message.get("body", b"")
+                        size += len(chunk)
+                        if size > MAX_REQUEST_BYTES:
+                            await reject(413, "The audio upload exceeds the request size limit.")
+                            return
+                        body.write(chunk)
+                        if not message.get("more_body", False):
+                            break
+                    body.seek(0)
+
+                    async def replay() -> Message:
+                        chunk = body.read(READ_CHUNK_BYTES)
+                        return {
+                            "type": "http.request",
+                            "body": chunk,
+                            "more_body": body.tell() < size,
+                        }
+
+                    await self.app(scope, replay, send)
+        finally:
+            self.state.pending_transcriptions -= 1
+
 
 # Initialize the app
 app = FastAPI()
+app.state.pending_transcriptions = 0
+app.add_middleware(TranscriptionInputMiddleware, state=app.state)
 
 # allow all cors
 app.add_middleware(
@@ -38,30 +135,98 @@
 print("Model loaded!")
 
 
-@app.post("/transcribe/")
-async def transcription(file: UploadFile = File(...)):
-    try:
-        audio_bytes = await file.read()
+def decode_audio(audio_bytes: bytes) -> np.ndarray:
+    max_samples = int(MAX_AUDIO_SECONDS * SAMPLE_RATE)
+    with tempfile.TemporaryDirectory(prefix="typeagent-whisper-") as directory:
+        source = Path(directory) / "audio"
+        source.write_bytes(audio_bytes)
+        try:
+            result = subprocess.run(
+                [
+                    "ffmpeg",
+                    "-hide_banner",
+                    "-nostdin",
+                    "-loglevel",
+                    "fatal",
+                    "-max_alloc",
+                    str(64 * 1024 * 1024),
+                    "-threads",
+                    "1",
+                    "-protocol_whitelist",
+                    "file",
+                    "-i",
+                    str(source),
+                    "-map",
+                    "0:a:0",
+                    "-vn",
+                    "-sn",
+                    "-dn",
+                    "-ac",
+                    "1",
+                    "-ar",
+                    str(SAMPLE_RATE),
+                    "-t",
+                    str((max_samples + 1) / SAMPLE_RATE),
+                    "-fs",
+                    str((max_samples + 1) * 2),
+                    "-f",
+                    "s16le",
+                    "pipe:1",
+                ],
+                stdout=subprocess.PIPE,
+                stderr=subprocess.DEVNULL,
+                timeout=DECODE_TIMEOUT_SECONDS,
+                check=False,
+            )
+        except subprocess.TimeoutExpired as error:
+            raise HTTPException(
+                422, "Audio decoding exceeded the processing time limit."
+            ) from error
+    if result.returncode != 0 or not result.stdout or len(result.stdout) % 2:
+        raise HTTPException(400, "Audio could not be decoded.")
+    if len(result.stdout) > max_samples * 2:
+        raise HTTPException(413, "The audio recording exceeds the duration limit.")
+    audio = np.frombuffer(result.stdout, dtype="<i2").astype(np.float32)
+    audio /= 32768.0
+    return audio
 
-        # Use pydub to handle different audio formats and convert audio
-        audio = AudioSegment.from_file(BytesIO(audio_bytes))
-        audio = audio.set_channels(1).set_frame_rate(16000).set_sample_width(2)
 
-        # Convert data from 16 bit wide integers to floating point with a width of 32 bits.
-        # Clamp the audio stream frequency to a PCM wavelength compatible default of 32768hz max.
-        audio_np = (
-            np.frombuffer(audio.raw_data, dtype=np.int16).astype(np.float32) / 32768.0
-        )
+def transcribe_audio(audio_bytes: bytes) -> str:
+    segments, _ = model.transcribe(decode_audio(audio_bytes))
+    return "\n".join(segment.text for segment in segments)
 
-        # faster_whisper returns a generator. Iterate though it to get the transcription
-        segments, _ = model.transcribe(audio_np)
-        segments = list(segments)
-        transcription = "\n".join(i.text for i in segments)
 
-        return JSONResponse(content={"transcription": transcription}, status_code=200)
-    except Exception as e:
+@app.post("/transcribe/")
+async def transcription(file: UploadFile = File(...)) -> JSONResponse:
+    try:
+        audio_bytes = await file.read(MAX_UPLOAD_BYTES + 1)
+        if len(audio_bytes) > MAX_UPLOAD_BYTES:
+            raise HTTPException(413, "The audio file exceeds the upload size limit.")
+        worker = asyncio.create_task(asyncio.to_thread(transcribe_audio, audio_bytes))
+        try:
+            text = await asyncio.shield(worker)
+        except asyncio.CancelledError:
+            cleanup = asyncio.gather(worker, return_exceptions=True)
+            while not cleanup.done():
+                try:
+                    await asyncio.shield(cleanup)
+                except asyncio.CancelledError:
+                    continue
+            results = cleanup.result()
+            if isinstance(results[0], Exception):
+                logging.error("Cancelled transcription failed", exc_info=results[0])
+            raise
+        return JSONResponse(content={"transcription": text}, status_code=200)
+    except HTTPException as error:
+        logging.warning("Audio input rejected: %s", error.detail)
+        return JSONResponse(
+            content={"error": error.detail}, status_code=error.status_code
+        )
+    except Exception:
         logging.error("An error occurred during transcription", exc_info=True)
-        return JSONResponse(content={"error": "An internal error has occurred!"}, status_code=500)
+        return JSONResponse(
+            content={"error": "An internal error has occurred!"}, status_code=500
+        )
 
 
 if __name__ == "__main__":

python/stt/whisperService/requirements-test.txt

@@ -0,0 +1,5 @@
+fastapi
+httpx
+numpy
+python-multipart
+uvicorn

python/stt/whisperService/requirements.txt

@@ -1,7 +1,6 @@
 pyaudio
 torch 
 numpy
-pydub
 fastapi
 uvicorn
 python-multipart