d3ro-voice/apps/desktop/sidecar/main.py

"""
D3RO-VOICE STT Sidecar (FastAPI HTTP 서버)
faster-whisper + CTranslate2만 사용. torch/pyannote 비의존으로 슬림 배포.

사용법:
    python main.py --port 18765

엔드포인트:
    GET  /health     - 헬스체크
    POST /load       - Whisper 모델 로딩
    POST /transcribe - 오디오 전사 (multipart)
    POST /shutdown   - 서버 종료

주: 화자 구분(diarization)은 Phase 15.5에서 LLM 추정 경로가 primary이며,
pyannote 기반 고정밀 화자 구분은 추후 서버 사이드 API로 제공될 예정.
"""

from __future__ import annotations

import argparse
import logging
import os
import signal
import sys
import time
from contextlib import asynccontextmanager
from typing import AsyncGenerator

import numpy as np
import uvicorn
from fastapi import FastAPI, File, Form, UploadFile
from fastapi.responses import JSONResponse

# ── 로깅 설정 ──────────────────────────────────────────────

logging.basicConfig(
    level=logging.INFO,
    format="[%(asctime)s] [%(levelname)s] [%(name)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
    stream=sys.stdout,
)
logger = logging.getLogger("sidecar")

# ── 전역 상태 ──────────────────────────────────────────────

_model: "WhisperModel | None" = None
_model_id: str | None = None
_gpu_available: bool = False
_server: uvicorn.Server | None = None

# ── FastAPI 앱 ─────────────────────────────────────────────


@asynccontextmanager
async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
    """서버 시작/종료 생명주기."""
    logger.info("Sidecar 서버 시작")
    _detect_gpu()
    yield
    logger.info("Sidecar 서버 종료")


app = FastAPI(title="D3RO-VOICE STT Sidecar", lifespan=lifespan)


def _detect_gpu() -> None:
    """GPU(CUDA) 사용 가능 여부를 ctranslate2로 감지한다.

    torch 의존 제거를 위해 ctranslate2의 네이티브 CUDA 감지를 사용한다.
    ctranslate2는 faster-whisper의 백엔드이므로 항상 함께 설치된다.
    """
    global _gpu_available
    try:
        import ctranslate2

        cuda_count = ctranslate2.get_cuda_device_count()
        _gpu_available = cuda_count > 0
        if _gpu_available:
            logger.info("GPU 감지: CUDA 디바이스 %d개", cuda_count)
        else:
            logger.info("GPU 미감지, CPU 모드로 동작")
    except Exception as exc:
        _gpu_available = False
        logger.info("GPU 감지 실패, CPU 모드로 동작: %s", exc)


# ── 엔드포인트 ─────────────────────────────────────────────


@app.get("/health")
async def health() -> JSONResponse:
    """헬스체크. sidecar가 준비되었는지 확인한다."""
    return JSONResponse(
        content={
            "status": "ready" if _model is not None else "no_model",
            "model": _model_id,
            "gpu": _gpu_available,
        }
    )


@app.post("/load")
async def load_model(body: dict) -> JSONResponse:  # noqa: ANN001
    """Whisper 모델을 로딩한다.

    Request body:
        { "model_id": "large-v3" }  -- tiny, base, small, medium, large-v3

    Returns:
        { "status": "loaded", "model_id": "large-v3", "load_time_ms": 1234 }
    """
    global _model, _model_id

    model_id: str = body.get("model_id", "large-v3")
    logger.info("모델 로딩 시작: %s", model_id)

    start_time = time.monotonic()

    try:
        from faster_whisper import WhisperModel

        device = "cuda" if _gpu_available else "cpu"
        compute_type = "float16" if _gpu_available else "int8"

        _model = WhisperModel(
            model_id,
            device=device,
            compute_type=compute_type,
        )
        _model_id = model_id

        load_time_ms = int((time.monotonic() - start_time) * 1000)
        logger.info(
            "모델 로딩 완료: %s (device=%s, compute=%s, %dms)",
            model_id,
            device,
            compute_type,
            load_time_ms,
        )

        return JSONResponse(
            content={
                "status": "loaded",
                "model_id": model_id,
                "load_time_ms": load_time_ms,
            }
        )

    except Exception as exc:
        logger.error("모델 로딩 실패: %s", exc)
        return JSONResponse(
            status_code=500,
            content={"status": "error", "message": str(exc)},
        )


@app.post("/transcribe")
async def transcribe(
    audio: UploadFile = File(...),
    language: str = Form("auto"),
    vad_filter: str = Form("true"),
    initial_prompt: str = Form(""),
) -> JSONResponse:
    """오디오 파일을 전사한다.

    Multipart form:
        audio         - PCM16 16kHz mono 바이너리 파일
        language      - 언어 코드 ('auto', 'ko', 'en', ...)
        vad_filter    - VAD 필터 활성화 ('true' / 'false')
        initial_prompt - 초기 프롬프트 (컨텍스트 힌트)
    """
    if _model is None:
        return JSONResponse(
            status_code=503,
            content={"status": "error", "message": "모델이 로딩되지 않았습니다"},
        )

    start_time = time.monotonic()

    try:
        pcm_bytes = await audio.read()

        if len(pcm_bytes) == 0:
            return JSONResponse(
                status_code=400,
                content={"status": "error", "message": "오디오 데이터가 비어있습니다"},
            )

        audio_array = (
            np.frombuffer(pcm_bytes, dtype=np.int16).astype(np.float32) / 32768.0
        )

        sample_rate = 16000
        audio_duration = len(audio_array) / sample_rate

        logger.info(
            "전사 시작: %.1f초 오디오, language=%s, vad=%s",
            audio_duration,
            language,
            vad_filter,
        )

        transcribe_kwargs: dict = {
            "vad_filter": vad_filter.lower() == "true",
            "beam_size": 5,
        }

        if language != "auto":
            transcribe_kwargs["language"] = language

        if initial_prompt:
            transcribe_kwargs["initial_prompt"] = initial_prompt

        # VAD가 전체 오디오를 제거하면 max() 에러 발생 → VAD 없이 재시도
        try:
            segments_iter, info = _model.transcribe(audio_array, **transcribe_kwargs)
        except ValueError as ve:
            if "empty sequence" in str(ve) and transcribe_kwargs.get("vad_filter"):
                logger.warning("VAD가 전체 오디오를 제거함 → VAD 없이 재시도")
                transcribe_kwargs["vad_filter"] = False
                segments_iter, info = _model.transcribe(audio_array, **transcribe_kwargs)
            else:
                raise

        segments_list: list[dict] = []
        full_text_parts: list[str] = []

        for segment in segments_iter:
            seg_dict = {
                "text": segment.text.strip(),
                "start": round(segment.start, 3),
                "end": round(segment.end, 3),
                "avg_logprob": round(segment.avg_logprob, 4),
            }
            segments_list.append(seg_dict)
            full_text_parts.append(segment.text.strip())

        full_text = " ".join(full_text_parts).strip()
        processing_time = int((time.monotonic() - start_time) * 1000)

        detected_language = info.language if info.language else "unknown"

        logger.info(
            "전사 완료: '%s' (lang=%s, %.1f초, %dms)",
            full_text[:80],
            detected_language,
            audio_duration,
            processing_time,
        )

        return JSONResponse(
            content={
                "text": full_text,
                "segments": segments_list,
                "language": detected_language,
                "duration": round(audio_duration, 3),
                "processing_time": processing_time,
            }
        )

    except Exception as exc:
        logger.error("전사 실패: %s", exc, exc_info=True)
        return JSONResponse(
            status_code=500,
            content={"status": "error", "message": str(exc)},
        )


@app.post("/shutdown")
async def shutdown() -> JSONResponse:
    """서버를 graceful하게 종료한다."""
    logger.info("종료 요청 수신")

    if _server is not None:
        _server.should_exit = True

    return JSONResponse(content={"status": "shutting_down"})


# ── 메인 ───────────────────────────────────────────────────


def main() -> None:
    """CLI 진입점."""
    global _server

    parser = argparse.ArgumentParser(description="D3RO-VOICE STT Sidecar")
    parser.add_argument(
        "--port",
        type=int,
        default=18765,
        help="HTTP 서버 포트 (기본: 18765)",
    )
    parser.add_argument(
        "--host",
        type=str,
        default="127.0.0.1",
        help="HTTP 서버 호스트 (기본: 127.0.0.1)",
    )
    args = parser.parse_args()

    logger.info("D3RO-VOICE STT Sidecar 시작 (port=%d)", args.port)

    def signal_handler(signum: int, _frame: object) -> None:
        sig_name = signal.Signals(signum).name
        logger.info("시그널 수신: %s, 종료 시작", sig_name)
        if _server is not None:
            _server.should_exit = True

    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)

    config = uvicorn.Config(
        app=app,
        host=args.host,
        port=args.port,
        log_level="warning",
        access_log=False,
    )
    _server = uvicorn.Server(config)
    _server.run()

    logger.info("Sidecar 서버 종료 완료")


if __name__ == "__main__":
    main()