lct-hack/backend/app/api/ws/call.py

"""Сокет курсанта-оператора 112.

Здесь только JSON-часть: приём вызова, правки карточки, подсказки, передача
в ДДС, завершение. Бинарные аудиокадры и голосовой контур — карточка lct-06.
"""

import asyncio
import json
import logging
from uuid import UUID

from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from pydantic import TypeAdapter, ValidationError

from app.domain.events import (
    CallEnded,
    CallEndReason,
    CallStarted,
    ErrorEvent,
    ErrorKind,
    HintShown,
    KioState,
    SessionEnded,
    SessionMode,
    TimerTick,
    TraineeToServer,
)
from app.scenarios import store
from app.session.hub import hub
from app.session.state import now_utc

log = logging.getLogger(__name__)
router = APIRouter()

#: Кадр контракта: 20 мс PCM16 моно 16 кГц = 320 сэмплов = 640 байт.
FRAME_BYTES = 640
FRAMES_PER_LOG = 250  # раз в пять секунд звука

_adapter = TypeAdapter(TraineeToServer)


def _on_audio(session_id: UUID, state, frame: bytes) -> None:
    """Приём аудиокадра. Голосовой контур (lct-06) заменит счёт на VAD → STT."""
    if len(frame) != FRAME_BYTES:
        state.bad_frames += 1
        if state.bad_frames == 1:
            # Один раз, а не на каждый кадр: неверный формат повторяется 50 раз в секунду.
            log.warning("сессия %s: кадр %d байт вместо %d — проверь ресемплинг на фронте",
                        session_id, len(frame), FRAME_BYTES)
        return
    state.audio_frames += 1
    if state.audio_frames % FRAMES_PER_LOG == 0:
        log.info("сессия %s: получено %d кадров (%.0f с звука)",
                 session_id, state.audio_frames, state.audio_frames * 0.02)


def _next_hint(state) -> tuple[str, str] | None:
    """Следующий неотработанный пункт чек-листа, который ещё не подсказывали.

    «Неотработанный» знает слот-автомат: пункт, о котором оператор уже спросил
    своими словами, подсказывать бессмысленно. Без модели эмбеддингов автомата
    нет — тогда подсказка идёт по порядку чек-листа.
    """
    if state.slots is not None:
        candidates = state.slots.unasked()
    else:
        scenario = store.get(state.scenario_id)
        candidates = scenario.checklist if scenario else []
    for item in candidates:
        if item.id not in state.hints_shown and item.question:
            return item.id, item.question
    return None


async def _handle(session_id: UUID, state, event) -> None:
    match event.type:
        case "call.answer":
            state.on_event("call.answer")
            state.started_at = now_utc()
            hub.to_trainee(session_id, CallStarted(started_at=state.started_at))
            hub.to_observers(session_id, state.snapshot())
            if hub.journal:
                await hub.journal.session_started(session_id, state.started_at)

        case "kio.patch":
            state.patch_kio(event.fields)
            # Наблюдателю уходит карточка целиком: рассинхрон на внешнем мониторе
            # посреди занятия дороже лишних килобайт.
            hub.to_observers(session_id, KioState(kio=state.kio))

        case "hint.request":
            if state.mode is SessionMode.EXAM:
                # На экзамене опоры нет — это часть нормы контроля.
                hub.to_trainee(
                    session_id,
                    ErrorEvent(
                        code=ErrorKind.HINT_DENIED_IN_EXAM,
                        message="В контрольном режиме подсказки недоступны",
                    ),
                )
                return
            nxt = _next_hint(state)
            if nxt is None:
                return
            checklist_id, question = nxt
            state.hints_shown.append(checklist_id)
            shown = HintShown(checklist_id=checklist_id, question=question)
            hub.broadcast(session_id, shown)
            if hub.journal:
                await hub.journal.hint(session_id, checklist_id, question, now_utc())

        case "dds.dispatch":
            state.on_event("dds.dispatch")
            state.dispatch(event.service.value)
            hub.to_observers(session_id, KioState(kio=state.kio))
            hub.to_observers(session_id, TimerTick(timers=state.timers.snapshot()))

        case "callback.dial":
            state.on_event("callback.dial")

        case "self_assessment.submit":
            if hub.journal:
                await hub.journal.self_assessment(
                    session_id, event.missed, event.comment, now_utc()
                )

        case "call.hangup":
            state.ended_at = now_utc()
            state.end_reason = CallEndReason.HANGUP
            hub.stop_ticker(session_id)
            hub.to_trainee(session_id, CallEnded(reason=CallEndReason.HANGUP))
            hub.to_observers(session_id, SessionEnded(reason=CallEndReason.HANGUP))
            if hub.journal:
                await hub.journal.session_ended(
                    session_id, state.ended_at, CallEndReason.HANGUP.value
                )


async def _pump(ws: WebSocket, queue: asyncio.Queue) -> None:
    while True:
        event = await queue.get()
        await ws.send_text(event.model_dump_json())


@router.websocket("/ws/call/{session_id}")
async def call(ws: WebSocket, session_id: UUID) -> None:
    await ws.accept()

    state = hub.get(session_id)
    if state is None:
        await ws.send_text(
            ErrorEvent(
                code=ErrorKind.SESSION_NOT_FOUND, message="Занятие ещё не запущено преподавателем"
            ).model_dump_json()
        )
        await ws.close()
        return

    with hub.trainee(session_id) as queue:
        writer = asyncio.create_task(_pump(ws, queue))
        try:
            while True:
                message = await ws.receive()
                if message["type"] == "websocket.disconnect":
                    return
                # Бинарные кадры — аудио, текстовые — события. Направление определяется
                # каналом, обёртки JSON вокруг звука нет (docs/arch/CONTRACT.md).
                if message.get("bytes") is not None:
                    _on_audio(session_id, state, message["bytes"])
                    continue
                try:
                    payload = json.loads(message.get("text") or "")
                except json.JSONDecodeError:
                    hub.to_trainee(
                        session_id,
                        ErrorEvent(code=ErrorKind.UNSUPPORTED_EVENT, message="не JSON"),
                    )
                    continue
                try:
                    event = _adapter.validate_python(payload)
                except ValidationError:
                    hub.to_trainee(
                        session_id,
                        ErrorEvent(code=ErrorKind.UNSUPPORTED_EVENT, message=str(payload)[:200]),
                    )
                    continue
                await _handle(session_id, state, event)
        except WebSocketDisconnect:
            return
        finally:
            writer.cancel()