version: "3"

 tasks:

   test:

     cmds:

       - pytest -o log_cli=true -o log_cli_level=DEBUG {{.CLI_ARGS}}

     "ISC001", # recommended to disable for formatting

     "INP001",

     "PT018",

 ]

 [tool.ruff.lint.isort]

 import asyncio

 import time

 from faster_whisper import transcribe

 from faster_whisper_server.audio import Audio

 from faster_whisper_server.core import Segment, Transcription, Word

 class FasterWhisperASR:

 from __future__ import annotations

 import asyncio

 from typing import TYPE_CHECKING, BinaryIO

 import numpy as np

 import soundfile as sf

 from faster_whisper_server.config import SAMPLES_PER_SECOND

 if TYPE_CHECKING:

     from collections.abc import AsyncGenerator

     from numpy.typing import NDArray

 def audio_samples_from_file(file: BinaryIO) -> NDArray[np.float32]:

     audio_and_sample_rate = sf.read(

         file,

                 f"Number of preloaded models ({len(self.preload_models)}) is greater than max_models ({self.max_models})"  # noqa: E501

             )

         return self

 from pydantic import BaseModel

 if TYPE_CHECKING:

     from collections.abc import Iterable

         self.words.extend(words)

     def _ensure_no_word_overlap(self, words: list[Word]) -> None:

         if len(self.words) > 0 and len(words) > 0:

             if words[0].start + config.word_timestamp_error_margin <= self.words[-1].end:

                 raise ValueError(

 from collections.abc import Generator

 from pathlib import Path

 import typing

 import huggingface_hub

 LIBRARY_NAME = "ctranslate2"

 TASK_NAME = "automatic-speech-recognition"

 import logging

 from __future__ import annotations

 from contextlib import asynccontextmanager

 from typing import TYPE_CHECKING

 from fastapi import (

 )

 from fastapi.middleware.cors import CORSMiddleware

 from faster_whisper_server.routers.list_models import (

     router as list_models_router,

 )

     from collections.abc import AsyncGenerator

 from collections import OrderedDict

 import gc

 import time

 from faster_whisper import WhisperModel

 class ModelManager:

         self.loaded_models: OrderedDict[str, WhisperModel] = OrderedDict()

     def load_model(self, model_name: str) -> WhisperModel:

         if model_name in self.loaded_models:

             logger.debug(f"{model_name} model already loaded")

             return self.loaded_models[model_name]

             oldest_model_name = next(iter(self.loaded_models))

             del self.loaded_models[oldest_model_name]

             gc.collect()

         logger.debug(f"Loading {model_name}...")

         # NOTE: will raise an exception if the model name isn't valid. Should I do an explicit check?

         whisper = WhisperModel(

             model_name,

         )

         logger.info(

         )

         self.loaded_models[model_name] = whisper

         return whisper

     APIRouter,

     Response,

 )

 import huggingface_hub

 from huggingface_hub.hf_api import RepositoryNotFoundError

 router = APIRouter()

 @router.get("/api/ps", tags=["experimental"], summary="Get a list of loaded models.")

     return {"models": list(model_manager.loaded_models.keys())}

 @router.post("/api/ps/{model_name:path}", tags=["experimental"], summary="Load a model into memory.")

     if model_name in model_manager.loaded_models:

         return Response(status_code=409, content="Model already loaded")

     model_manager.load_model(model_name)

 @router.delete("/api/ps/{model_name:path}", tags=["experimental"], summary="Unload a model from memory.")

     model = model_manager.loaded_models.get(model_name)

     if model is not None:

         del model_manager.loaded_models[model_name]

 import asyncio

 from io import BytesIO

 from typing import TYPE_CHECKING, Annotated, Literal

 from fastapi import (

 from fastapi.responses import StreamingResponse

 from fastapi.websockets import WebSocketState

 from faster_whisper.vad import VadOptions, get_speech_timestamps

 from faster_whisper_server.asr import FasterWhisperASR

 from faster_whisper_server.audio import AudioStream, audio_samples_from_file

 from faster_whisper_server.config import (

     Language,

     ResponseFormat,

     Task,

 )

 from faster_whisper_server.core import Segment, segments_to_srt, segments_to_text, segments_to_vtt

 from faster_whisper_server.server_models import (

     TranscriptionJsonResponse,

     TranscriptionVerboseJsonResponse,

 )

 from faster_whisper_server.transcriber import audio_transcriber

 if TYPE_CHECKING:

     from collections.abc import Generator, Iterable

     from faster_whisper.transcribe import TranscriptionInfo

 router = APIRouter()

     For example, https://github.com/open-webui/open-webui/issues/2248#issuecomment-2162997623.

     """

     if model_name == "whisper-1":

         logger.info(f"{model_name} is not a valid model name. Using {config.whisper.model} instead.")

         return config.whisper.model

     response_model=str | TranscriptionJsonResponse | TranscriptionVerboseJsonResponse,

 )

 def translate_file(

     file: Annotated[UploadFile, Form()],

     prompt: Annotated[str | None, Form()] = None,

     temperature: Annotated[float, Form()] = 0.0,

     stream: Annotated[bool, Form()] = False,

 ) -> Response | StreamingResponse:

     whisper = model_manager.load_model(model)

     segments, transcription_info = whisper.transcribe(

         file.file,

     response_model=str | TranscriptionJsonResponse | TranscriptionVerboseJsonResponse,

 )

 def transcribe_file(

     file: Annotated[UploadFile, Form()],

     prompt: Annotated[str | None, Form()] = None,

     temperature: Annotated[float, Form()] = 0.0,

     timestamp_granularities: Annotated[

         list[Literal["segment", "word"]],

     stream: Annotated[bool, Form()] = False,

     hotwords: Annotated[str | None, Form()] = None,

 ) -> Response | StreamingResponse:

     whisper = model_manager.load_model(model)

     segments, transcription_info = whisper.transcribe(

         file.file,

 async def audio_receiver(ws: WebSocket, audio_stream: AudioStream) -> None:

     try:

         while True:

             bytes_ = await asyncio.wait_for(ws.receive_bytes(), timeout=config.max_no_data_seconds)

 @router.websocket("/v1/audio/transcriptions")

 async def transcribe_stream(

     ws: WebSocket,

     temperature: Annotated[float, Query()] = 0.0,

 ) -> None:

     await ws.accept()

     transcribe_opts = {

         "language": language,

     audio_stream = AudioStream()

     async with asyncio.TaskGroup() as tg:

         tg.create_task(audio_receiver(ws, audio_stream))

             logger.debug(f"Sending transcription: {transcription.text}")

             if ws.client_state == WebSocketState.DISCONNECTED:

                 break

 from __future__ import annotations

 from typing import TYPE_CHECKING

 from faster_whisper_server.audio import Audio, AudioStream

 from faster_whisper_server.core import Transcription, Word, common_prefix, to_full_sentences, word_to_text

 if TYPE_CHECKING:

     from collections.abc import AsyncGenerator

     from faster_whisper_server.asr import FasterWhisperASR

 class LocalAgreement:

 async def audio_transcriber(

     asr: FasterWhisperASR,

     audio_stream: AudioStream,

 ) -> AsyncGenerator[Transcription, None]:

     local_agreement = LocalAgreement()

     full_audio = Audio()

     confirmed = Transcription()

         full_audio.extend(chunk)

         audio = full_audio.after(needs_audio_after(confirmed))

         transcription, _ = await asr.transcribe(audio, prompt(confirmed))

 from collections.abc import AsyncGenerator, Generator

 import logging

 from fastapi.testclient import TestClient

 from httpx import ASGITransport, AsyncClient

 from openai import OpenAI

 import pytest

 @pytest.fixture()

 def client() -> Generator[TestClient, None, None]:

         yield client

 @pytest_asyncio.fixture()

 async def aclient() -> AsyncGenerator[AsyncClient, None]:

         yield aclient