483 lines
22 KiB
Python
483 lines
22 KiB
Python
"""Канал преподавателя: только передача.
|
||
|
||
**Ни одной команды, меняющей карточку курсанта.** Преподаватель управляет
|
||
ситуацией, а не работой обучаемого, иначе оценка перестаёт быть оценкой
|
||
курсанта (docs/arch/CONTRACT.md).
|
||
|
||
Ответы сюда не идут — канал односторонний. Всё, что сервер хочет сказать
|
||
преподавателю, уходит на его же сокет `observe`.
|
||
"""
|
||
|
||
import asyncio
|
||
import logging
|
||
import math
|
||
import secrets
|
||
from uuid import UUID
|
||
|
||
from fastapi import APIRouter, WebSocket, WebSocketDisconnect
|
||
from pydantic import TypeAdapter, ValidationError
|
||
|
||
from app.api.ws.session import close_fenced, run_command, session_socket
|
||
from app.config import get_settings
|
||
from app.db.base import get_sessionmaker
|
||
from app.db.repo import SessionNodeConflict
|
||
from app.dialog.director import apply as apply_directive
|
||
from app.dialog.director import mood_of
|
||
from app.dialog.factory import build_caller
|
||
from app.dialog.persona import PersonaState
|
||
from app.dialog.runtime import get_embedder
|
||
from app.dialog.slots import SlotMachine
|
||
from app.domain.classifiers import Outcome
|
||
from app.domain.events import (
|
||
CallEnded,
|
||
CallEndReason,
|
||
CallIncoming,
|
||
CardBriefing,
|
||
ErrorEvent,
|
||
ErrorKind,
|
||
Exercise,
|
||
InstructorNoteShown,
|
||
InstructorToServer,
|
||
ModeSet,
|
||
ReferenceStarted,
|
||
SessionEnded,
|
||
StationState,
|
||
)
|
||
from app.domain.roles import Role
|
||
from app.domain.timers import TimerCode
|
||
from app.scenarios import store
|
||
from app.session.dds import prepare_queue
|
||
from app.session.access import can_access
|
||
from app.session.hub import hub
|
||
from app.session.finish import end_session, override_score
|
||
from app.session.state import SessionState, now_utc
|
||
from app.session.store import LessonIdentity, LessonRequest, NoteAdded, ScoreOverridden
|
||
from app.voice.models import get_voice_models
|
||
from app.voice.pipeline import FILLERS, prefetch
|
||
|
||
log = logging.getLogger(__name__)
|
||
router = APIRouter()
|
||
|
||
_adapter = TypeAdapter(InstructorToServer)
|
||
|
||
# Пульт может молчать; без входящих кадров потеря lease замечается этим опросом.
|
||
_FENCE_POLL_SECONDS = 1.0
|
||
|
||
|
||
def _dds_ineligible_scenarios(scenarios):
|
||
"""Консультация и передача региона не являются готовыми карточками ДДС."""
|
||
return [scenario for scenario in scenarios if scenario.outcome is not Outcome.CARD]
|
||
|
||
|
||
def card_briefing(state: SessionState) -> CardBriefing:
|
||
"""Первую реплику показывает курсант; факты раскрываются только в ответах."""
|
||
scenario = state.scenario
|
||
return CardBriefing(
|
||
scenario_id=scenario.id, mode=state.mode, text=scenario.first_line,
|
||
required_fields=([field for field in state.required_fields if field != "dds"]
|
||
if scenario.ground_truth.incident_code else list(state.required_fields)),
|
||
card=state.kio,
|
||
handoff_to_dds=state.handoff_to_dds,
|
||
)
|
||
|
||
|
||
def _build_state(session_id: UUID, event, who, scenario, scenarios,
|
||
identity: LessonIdentity) -> SessionState:
|
||
"""Занятие собирается целиком и только потом регистрируется: иначе
|
||
наблюдатель, подключившийся в эту щель, увидит полусобранное состояние
|
||
без слот-автомата и звонящего. Снимок пишется в той же транзакции, что
|
||
и строка занятия."""
|
||
state = SessionState(
|
||
session_id=session_id,
|
||
scenario_id=scenario.id,
|
||
scenario_title=scenario.title,
|
||
level=scenario.level.value,
|
||
mode=event.mode,
|
||
owner_login=who.login if who is not None else None,
|
||
backend_fencing_epoch=identity.fencing_epoch,
|
||
exercise=event.exercise,
|
||
handoff_to_dds=event.handoff_to_dds,
|
||
scenario=scenario.model_copy(deep=True),
|
||
required_fields=scenario.required_fields,
|
||
trainee_name=event.trainee,
|
||
trainee_id=identity.trainee_id,
|
||
dds_service=identity.service or event.dds_service,
|
||
attempt=identity.attempt,
|
||
criteria=event.criteria,
|
||
)
|
||
state.timers.limits[TimerCode.DDS_ACK] = event.criteria.decision_time_limit_seconds * 1000
|
||
state.timers.limits[TimerCode.CARD_FILL] = event.criteria.card_fill_time_limit_seconds * 1000
|
||
state.timers.limits[TimerCode.DDS_WORK] = event.criteria.dds_card_work_time_limit_seconds * 1000
|
||
if event.exercise is Exercise.CALL:
|
||
embedder = get_embedder()
|
||
if embedder is not None:
|
||
state.slots = SlotMachine(state.scenario, embedder)
|
||
state.persona = PersonaState(state.scenario.persona)
|
||
state.caller = build_caller(
|
||
scenario.id,
|
||
use_pregenerated=scenario.tree.pregenerated,
|
||
)
|
||
elif event.exercise is Exercise.DDS:
|
||
prepare_queue(
|
||
state,
|
||
[item.model_copy(deep=True) for item in scenarios],
|
||
arrival_interval_seconds=event.dds_arrival_interval_seconds,
|
||
max_waiting=event.dds_max_waiting,
|
||
)
|
||
state.started_at = state.desk.active.dispatched_at if state.desk.active else None
|
||
else:
|
||
state.started_at = now_utc()
|
||
if event.exercise is Exercise.CARD:
|
||
embedder = get_embedder()
|
||
if embedder is not None:
|
||
state.slots = SlotMachine(state.scenario, embedder)
|
||
state.persona = PersonaState(state.scenario.persona)
|
||
state.caller = build_caller(
|
||
scenario.id, use_pregenerated=scenario.tree.pregenerated,
|
||
)
|
||
state.on_event("card.start")
|
||
if event.handoff_to_dds:
|
||
state.pending_dds_scenarios = [item.model_copy(deep=True) for item in scenarios[1:]]
|
||
state.desk.arrival_interval_seconds = event.dds_arrival_interval_seconds
|
||
state.desk.max_waiting = event.dds_max_waiting
|
||
if event.exercise is Exercise.CALL:
|
||
state.on_event("call.incoming")
|
||
return state
|
||
|
||
|
||
async def _start(session_id: UUID, event, who=None) -> None:
|
||
if event.handoff_to_dds and event.exercise is not Exercise.CARD:
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.SCENARIO_INVALID,
|
||
message="Передача в ДДС доступна только для текстовой карточки 112",
|
||
))
|
||
return
|
||
scenario_ids = list(dict.fromkeys([
|
||
event.scenario_id, *(event.scenario_ids or []), *(event.random_scenario_ids or []),
|
||
]))
|
||
if get_settings().demo_no_db or not hub.store.persistent:
|
||
catalog = {
|
||
scenario_id: scenario for scenario_id in scenario_ids
|
||
if (scenario := store.get(scenario_id)) is not None
|
||
}
|
||
hidden_scenario_ids = (
|
||
await store.scenario_ids_owned_by_other(None, who.login)
|
||
if who is not None else set()
|
||
)
|
||
else:
|
||
try:
|
||
async with get_sessionmaker()() as db:
|
||
catalog, hidden_scenario_ids = await store.published_catalog(
|
||
db, scenario_ids, who.login if who is not None else None,
|
||
)
|
||
except Exception as exc:
|
||
# Avoid serializing scenario facts or SQL bind values into application logs.
|
||
log.error("не удалось разрешить сценарий из общей библиотеки (%s)", type(exc).__name__)
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.INTERNAL,
|
||
message="Не удалось проверить сценарий в общей библиотеке; запуск отменён.",
|
||
))
|
||
return
|
||
scenario = catalog.get(event.scenario_id)
|
||
if scenario is None:
|
||
hub.to_observers(
|
||
session_id,
|
||
ErrorEvent(code=ErrorKind.SCENARIO_INVALID, message=f"Нет сценария {event.scenario_id}"),
|
||
)
|
||
return
|
||
if scenario.id in hidden_scenario_ids:
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.SCENARIO_INVALID,
|
||
message="Сценарий не найден или недоступен этому преподавателю",
|
||
))
|
||
return
|
||
|
||
if event.random_scenario_ids:
|
||
if len(event.random_scenario_ids) > 96:
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.SCENARIO_INVALID,
|
||
message="Случайный отбор ограничен 96 карточками",
|
||
))
|
||
return
|
||
pool_ids = list(dict.fromkeys(event.random_scenario_ids))
|
||
pool = [catalog.get(scenario_id) for scenario_id in pool_ids]
|
||
if any(item is None for item in pool):
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.SCENARIO_INVALID,
|
||
message="В случайном отборе есть неизвестный сценарий",
|
||
))
|
||
return
|
||
if hidden_scenario_ids.intersection(pool_ids):
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.SCENARIO_INVALID,
|
||
message="Сценарий не найден или недоступен этому преподавателю",
|
||
))
|
||
return
|
||
ineligible_pool = _dds_ineligible_scenarios(pool)
|
||
if ineligible_pool:
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.SCENARIO_INVALID,
|
||
message="Случайный отбор должен содержать только готовые карточки ДДС",
|
||
))
|
||
return
|
||
scenario = secrets.choice(pool)
|
||
|
||
scenario_ids = event.scenario_ids or [event.scenario_id]
|
||
if event.exercise is Exercise.DDS or event.handoff_to_dds:
|
||
if (not scenario_ids or scenario_ids[0] != event.scenario_id
|
||
or len(scenario_ids) > 96):
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.SCENARIO_INVALID,
|
||
message="Очередь ДДС должна начинаться с scenario_id и содержать не более 96 карточек",
|
||
))
|
||
return
|
||
if event.random_scenario_ids:
|
||
extra_ids = list(dict.fromkeys(
|
||
item for item in scenario_ids[1:] if item != scenario.id
|
||
))
|
||
extras = [catalog.get(item) for item in extra_ids]
|
||
remaining_random = [item for item in pool if item.id != scenario.id
|
||
and item.id not in extra_ids]
|
||
randomized_tail = secrets.SystemRandom().sample(
|
||
remaining_random, k=len(remaining_random)
|
||
)
|
||
scenarios = [scenario, *extras, *randomized_tail]
|
||
else:
|
||
scenarios = [catalog.get(scenario_id) for scenario_id in scenario_ids]
|
||
if len(scenarios) > 96:
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.SCENARIO_INVALID,
|
||
message="Очередь ДДС не может содержать более 96 карточек",
|
||
))
|
||
return
|
||
if any(item is None for item in scenarios):
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.SCENARIO_INVALID, message="В очереди ДДС есть неизвестный сценарий",
|
||
))
|
||
return
|
||
if any(item.id in hidden_scenario_ids for item in scenarios):
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.SCENARIO_INVALID,
|
||
message="Сценарий не найден или недоступен этому преподавателю",
|
||
))
|
||
return
|
||
ineligible = _dds_ineligible_scenarios(scenarios)
|
||
if ineligible:
|
||
titles = ", ".join(item.title for item in ineligible)
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.SCENARIO_INVALID,
|
||
message=("В очередь ДДС можно добавить только готовые карточки с исходом "
|
||
f"«карточка и передача в ДДС». Исключите: {titles}"),
|
||
))
|
||
return
|
||
else:
|
||
scenarios = []
|
||
|
||
try:
|
||
state = await hub.store.open(LessonRequest(
|
||
session_id=session_id, scenario_id=scenario.id, mode=event.mode.value,
|
||
trainee_name=event.trainee, trainee_id=event.trainee_id,
|
||
owner_login=who.login if who is not None else None,
|
||
backend_node_id=get_settings().backend_node_id,
|
||
), lambda identity: _build_state(session_id, event, who, scenario, scenarios, identity))
|
||
except SessionNodeConflict:
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.FORBIDDEN,
|
||
message="Сессия закреплена за другим backend-узлом; проверьте маршрутизацию proxy",
|
||
))
|
||
return
|
||
except PermissionError:
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.FORBIDDEN,
|
||
message="Занятие создано другим преподавателем",
|
||
))
|
||
return
|
||
except Exception as exc: # noqa: BLE001 — без долговечной истории занятие не стартует
|
||
log.error("занятие %s не записано (%s)", session_id, type(exc).__name__)
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.INTERNAL,
|
||
message="Не удалось записать занятие и аудит; запуск отменён.",
|
||
))
|
||
return
|
||
hub.register(state)
|
||
|
||
# Первая реплика и филлеры синтезируются, пока курсант не снял трубку:
|
||
# «Алло! Помогите!» должно прозвучать мгновенно (docs/arch/BACKEND.md).
|
||
if event.exercise is Exercise.CALL:
|
||
models = get_voice_models()
|
||
if models is not None:
|
||
asyncio.create_task(prefetch(models, [scenario.first_line, *FILLERS.values()]))
|
||
hub.start_ticker(session_id)
|
||
if event.exercise is Exercise.CALL:
|
||
hub.to_trainee(
|
||
session_id,
|
||
CallIncoming(
|
||
scenario_id=scenario.id,
|
||
caller_number="+7 (495) 000-00-00",
|
||
level=scenario.level,
|
||
mode=event.mode,
|
||
required_fields=[field for field in scenario.required_fields
|
||
if field != "dds" or not scenario.ground_truth.incident_code],
|
||
),
|
||
)
|
||
elif event.exercise is Exercise.DDS:
|
||
hub.to_station(session_id, state.card_received_event())
|
||
hub.to_station(session_id, StationState(snapshot=state.station_snapshot()))
|
||
else:
|
||
hub.to_trainee(session_id, card_briefing(state))
|
||
hub.to_observers(session_id, ModeSet(mode=event.mode))
|
||
hub.to_observers(session_id, state.snapshot())
|
||
|
||
|
||
async def _stop(session_id: UUID) -> None:
|
||
state = hub.get(session_id)
|
||
if state is None:
|
||
return
|
||
await end_session(session_id, state, CallEndReason.INSTRUCTOR)
|
||
|
||
|
||
async def _command(session_id: UUID, event, who) -> None:
|
||
"""Команда пульта. Всё, кроме запуска, идёт внутри `hub.operation`."""
|
||
state = hub.get(session_id)
|
||
match event.type:
|
||
case "scenario.start":
|
||
await _start(session_id, event, who)
|
||
case "session.stop":
|
||
await _stop(session_id)
|
||
case "instructor_note.add":
|
||
if state is not None:
|
||
state.notes.append({
|
||
"type": "instructor_note.shown",
|
||
"transcript_ref": event.transcript_ref,
|
||
"text": event.text,
|
||
"author": "преподаватель",
|
||
})
|
||
hub.record(session_id, NoteAdded(event.transcript_ref, event.text, "преподаватель"))
|
||
hub.to_observers(
|
||
session_id,
|
||
InstructorNoteShown(
|
||
transcript_ref=event.transcript_ref,
|
||
text=event.text,
|
||
author="преподаватель",
|
||
),
|
||
)
|
||
case "reference.play":
|
||
if state is not None:
|
||
hub.to_observers(session_id, ReferenceStarted(scenario_id=state.scenario_id))
|
||
case "score.override":
|
||
if event.session_id != session_id:
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.FORBIDDEN,
|
||
message="Оценка относится к другому занятию",
|
||
))
|
||
return
|
||
if state is None or state.score is None:
|
||
return
|
||
try:
|
||
verdict = float(event.verdict)
|
||
except ValueError:
|
||
verdict = float("nan")
|
||
if not math.isfinite(verdict) or not 0 <= verdict <= 100:
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.UNSUPPORTED_EVENT,
|
||
message="Оценка должна быть числом от 0 до 100",
|
||
))
|
||
return
|
||
override_score(state, ScoreOverridden(
|
||
score_final=verdict, author=who.login, role=who.role.value,
|
||
comment=event.comment,
|
||
))
|
||
case "director.inject":
|
||
if state is None:
|
||
return
|
||
if state.exercise is not Exercise.CALL:
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.UNSUPPORTED_EVENT,
|
||
message="Директивы звонящему доступны только в голосовом упражнении",
|
||
))
|
||
return
|
||
result = apply_directive(state, event.directive)
|
||
if result.needs_network:
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.DIRECTIVE_NEEDS_NETWORK,
|
||
message="Свободный текст требует локально запущенной модели; без неё доступны кнопки",
|
||
))
|
||
return
|
||
state.directives.append(event.directive)
|
||
voice = state.voice
|
||
if result.drop_line and voice is not None:
|
||
# Обрыв рвёт звук на полуслове тем же механизмом, что
|
||
# перебивание, и запускает норматив обратного дозвона.
|
||
voice.barge_in()
|
||
if result.drop_line:
|
||
state.on_event("call.dropped")
|
||
hub.to_trainee(session_id, CallEnded(reason=CallEndReason.DROPPED))
|
||
hub.to_observers(session_id, SessionEnded(reason=CallEndReason.DROPPED))
|
||
elif result.say and voice is not None:
|
||
voice.speak(result.say, mood_of(state))
|
||
case _:
|
||
hub.to_observers(
|
||
session_id,
|
||
ErrorEvent(
|
||
code=ErrorKind.UNSUPPORTED_EVENT,
|
||
message=f"{event.type} ещё не реализовано",
|
||
),
|
||
)
|
||
|
||
|
||
@router.websocket("/ws/control/{session_id}")
|
||
async def control(ws: WebSocket, session_id: UUID) -> None:
|
||
# Пульт преподавателя: управление занятием доступно только ему. Занятия
|
||
# до `scenario.start` ещё нет — доступ проверяется на каждой команде.
|
||
entered = await session_socket(ws, session_id, (Role.INSTRUCTOR,), require_state=False)
|
||
if entered is None:
|
||
return
|
||
who, _state = entered
|
||
# Чтение живёт дольше тика опроса: wait_for отменял бы его, а отмена после
|
||
# того, как receive уже забрал кадр, теряет команду или websocket.disconnect.
|
||
# Команда обрабатывается здесь, одной операцией хранилища.
|
||
read: asyncio.Future | None = None
|
||
try:
|
||
while True:
|
||
if read is None:
|
||
read = asyncio.ensure_future(ws.receive_json())
|
||
done, _ = await asyncio.wait({read}, timeout=_FENCE_POLL_SECONDS)
|
||
if not done:
|
||
if hub.is_lease_fenced(session_id):
|
||
await close_fenced(ws)
|
||
return
|
||
continue
|
||
payload, read = read.result(), None
|
||
try:
|
||
event = _adapter.validate_python(payload)
|
||
except ValidationError:
|
||
hub.to_observers(
|
||
session_id,
|
||
ErrorEvent(code=ErrorKind.UNSUPPORTED_EVENT, message=str(payload)[:200]),
|
||
)
|
||
continue
|
||
|
||
existing_state = hub.get(session_id)
|
||
if event.type == "scenario.start":
|
||
if existing_state is not None:
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.FORBIDDEN,
|
||
message="Для нового занятия создайте новый номер сессии",
|
||
))
|
||
continue
|
||
# Запуск пишется своей транзакцией в `store.open`: до неё занятия нет.
|
||
await _command(session_id, event, who)
|
||
continue
|
||
if existing_state is None or not can_access(who, existing_state):
|
||
hub.to_observers(session_id, ErrorEvent(
|
||
code=ErrorKind.FORBIDDEN,
|
||
message="Это занятие принадлежит другому преподавателю",
|
||
))
|
||
continue
|
||
if not await run_command(ws, session_id, lambda: _command(session_id, event, who)):
|
||
return
|
||
except WebSocketDisconnect:
|
||
return
|
||
finally:
|
||
if read is not None:
|
||
read.cancel()
|