каркас: домен, генерация типов фронта, стенд и карточки задач

Контракт в коде: КИО по нормативу, 43 события WS по шести союзам направлений,
таймеры ГОСТ, таксономия E1–E6, шесть компетенций радара. make types гоняет
их через JSON Schema в generated.ts, тест ловит забытую регенерацию.

Стенд: docker compose postgres + backend, health отвечает, база досягаема
из контейнера бэкенда. Makefile не врёт — нереализованные цели падают
и называют карточку, которая их закроет.

frontend/.npmrc: под WSL2 npm install зависает на дефолтных 15 сокетах.
This commit is contained in:
Ivan Gerasimov 2026-09-15 19:35:03 +03:00
commit 7b7dd38e9f
36 changed files with 4997 additions and 0 deletions

0
backend/app/__init__.py Normal file
View file

40
backend/app/config.py Normal file
View file

@ -0,0 +1,40 @@
"""Настройки. Нормативы ГОСТ — в миллисекундах и из конфига, не из кода."""
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict
from app.domain.timers import NORMATIVES, TimerCode
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
# Данные
database_url: str = "postgresql+asyncpg://lct:lct@localhost:5432/lct"
# Голосовой контур
models_dir: str = "models"
stt_model: str = "gigaam-v3-rnnt"
endpointing_ms: int = 600
mic_sample_rate: int = 16_000
tts_sample_rate: int = 24_000
offline: bool = False
# LLM. Провайдер меняется значением, не кодом (docs/arch/STACK.md).
llm_provider: str = "gigachat"
llm_api_key: str = ""
llm_model_caller: str = ""
llm_model_judge: str = ""
judge_temperature: float = 0.0
# Нормативы: переопределяют значения по умолчанию из domain/timers.py
timer_limits_ms: dict[TimerCode, int] = {}
def limit_ms(self, code: TimerCode) -> int:
return self.timer_limits_ms.get(code, NORMATIVES[code].limit_ms)
@lru_cache
def get_settings() -> Settings:
return Settings()

View file

@ -0,0 +1,5 @@
"""Общий домен: контракт между бэкендом и фронтендом.
Правится только по уговору двоих — см. docs/arch/CONTRACT.md.
Источник истины для frontend/src/shared/types/generated.ts (make types).
"""

View file

@ -0,0 +1,64 @@
"""Классификаторы ДДС, типы происшествий, уровни сложности.
Норматив: docs/spec/NORMATIVES.md#классификаторы-ддс
"""
from enum import StrEnum
class DDSCode(StrEnum):
"""Дежурно-диспетчерская служба, в которую уходит карточка."""
FIRE = "01"
POLICE = "02"
MEDICAL = "03"
OTHER = "04"
GKH = "gkh"
class IncidentType(StrEnum):
"""Тип происшествия. Совпадает с полем `type` сценария."""
FIRE = "fire"
MEDICAL = "medical"
POLICE = "police"
OTHER = "other"
GKH = "gkh"
ERA_GLONASS = "era_glonass"
class Level(StrEnum):
"""Уровень сложности сценария."""
L1 = "L1"
L2 = "L2"
L3 = "L3"
DDS_LABELS: dict[DDSCode, str] = {
DDSCode.FIRE: "01 — МЧС",
DDSCode.POLICE: "02 — полиция",
DDSCode.MEDICAL: "03 — скорая помощь",
DDSCode.OTHER: "04 — прочие службы",
DDSCode.GKH: "ЖКХ",
}
INCIDENT_LABELS: dict[IncidentType, str] = {
IncidentType.FIRE: "Пожар",
IncidentType.MEDICAL: "Медицинский вызов",
IncidentType.POLICE: "Правонарушение",
IncidentType.OTHER: "Прочее",
IncidentType.GKH: "Авария ЖКХ",
IncidentType.ERA_GLONASS: "ДТП, автовызов ЭРА-ГЛОНАСС",
}
#: Штатное соответствие типа происшествия и службы. Используется как значение
#: по умолчанию при выводе ground_truth сценария; сценарий вправе переопределить.
DDS_BY_INCIDENT: dict[IncidentType, DDSCode] = {
IncidentType.FIRE: DDSCode.FIRE,
IncidentType.MEDICAL: DDSCode.MEDICAL,
IncidentType.POLICE: DDSCode.POLICE,
IncidentType.OTHER: DDSCode.OTHER,
IncidentType.GKH: DDSCode.GKH,
IncidentType.ERA_GLONASS: DDSCode.MEDICAL,
}

View file

@ -0,0 +1,529 @@
"""Схемы событий WebSocket — источник истины для make types.
Контракт целиком: docs/arch/CONTRACT.md
Каналы: /ws/call (курсант), /ws/station (ДДС), /ws/observe (монитор и пульт,
только приём), /ws/control (только преподаватель, только передача).
Правила, зашитые в схемы:
* курсанту уходят дельты, наблюдателям — состояние целиком;
* все `at` — серверное UTC, фронт время не считает;
* ошибки идут одним каналом `error`, без HTTP-кодов внутри WS.
"""
from datetime import datetime
from enum import StrEnum
from typing import Annotated, Any, Literal
from uuid import UUID
from pydantic import BaseModel, Field
from app.domain.classifiers import DDSCode, IncidentType, Level
from app.domain.kio import KIO
from app.domain.taxonomy import Finding
from app.domain.timers import TimerSnapshot
class SessionMode(StrEnum):
"""Режим сессии. Меняет доступность подсказок и протоколирование,
но не поведение звонящего (docs/product/MODES.md)."""
TRAINING = "training"
EXAM = "exam"
SELF = "self"
class Speaker(StrEnum):
CALLER = "caller"
OPERATOR = "operator"
class Mood(StrEnum):
"""Состояние звонящего. Фон подачи, не факты."""
CALM = "calm"
WORRIED = "worried"
PANIC = "panic"
AGGRESSIVE = "aggressive"
CONFUSED = "confused"
class CallEndReason(StrEnum):
HANGUP = "hangup"
DROPPED = "dropped"
INSTRUCTOR = "instructor"
class PatchSource(StrEnum):
"""`auto` — сервер распознал факт из речи, `operator` — эхо правки курсанта."""
AUTO = "auto"
OPERATOR = "operator"
class DirectiveMode(StrEnum):
"""`immediate` — только для обрыва связи: рвёт TTS на полуслове."""
NEXT_TURN = "next_turn"
IMMEDIATE = "immediate"
class TranscriptEntry(BaseModel):
"""Реплика в ленте. `ref` — якорь для пометок преподавателя и отметок разбора."""
ref: str
speaker: Speaker
text: str
at: datetime
mood: Mood | None = None
class Coords(BaseModel):
lat: float
lon: float
# ─────────────────────────── сервер → курсант ───────────────────────────
class CallIncoming(BaseModel):
type: Literal["call.incoming"] = "call.incoming"
scenario_id: str
caller_number: str
level: Level
mode: SessionMode
class CallStarted(BaseModel):
type: Literal["call.started"] = "call.started"
started_at: datetime
class SttPartial(BaseModel):
type: Literal["stt.partial"] = "stt.partial"
text: str
class SttFinal(BaseModel):
type: Literal["stt.final"] = "stt.final"
text: str
at: datetime
class CallerUtterance(BaseModel):
type: Literal["caller.utterance"] = "caller.utterance"
utterance_id: UUID
text: str
at: datetime
mood: Mood
class TtsBegin(BaseModel):
type: Literal["tts.begin"] = "tts.begin"
utterance_id: UUID
class TtsEnd(BaseModel):
type: Literal["tts.end"] = "tts.end"
utterance_id: UUID
class TtsCancel(BaseModel):
"""Оператор перебил. Фронт мгновенно чистит очередь воспроизведения."""
type: Literal["tts.cancel"] = "tts.cancel"
utterance_id: UUID
reason: Literal["barge_in", "director"] = "barge_in"
class BgStart(BaseModel):
"""Аудио-фон происшествия. Файл лежит в сборке фронта, трафика фон не создаёт."""
type: Literal["bg.start"] = "bg.start"
loop: str
gain_db: float
class BgStop(BaseModel):
type: Literal["bg.stop"] = "bg.stop"
class EraData(BaseModel):
"""Автоданные ЭРА-ГЛОНАСС до соединения с водителем."""
type: Literal["era.data"] = "era.data"
vin: str
coords: Coords
passengers: int
impact_force: str
class KioPatchOut(BaseModel):
"""Сервер распознал факт из речи либо шлёт эхо чужой правки."""
type: Literal["kio.patch"] = "kio.patch"
fields: dict[str, Any]
source: PatchSource
class HintShown(BaseModel):
type: Literal["hint.shown"] = "hint.shown"
checklist_id: str
question: str
class TimerTick(BaseModel):
"""Раз в секунду, не на каждое изменение: таймеров дюжина, UI рисует секунды."""
type: Literal["timer.tick"] = "timer.tick"
timers: list[TimerSnapshot]
class CallEnded(BaseModel):
type: Literal["call.ended"] = "call.ended"
reason: CallEndReason
class ScoreReady(BaseModel):
type: Literal["score.ready"] = "score.ready"
session_id: UUID
class ErrorEvent(BaseModel):
type: Literal["error"] = "error"
code: str
message: str
ServerToTrainee = Annotated[
CallIncoming
| CallStarted
| SttPartial
| SttFinal
| CallerUtterance
| TtsBegin
| TtsEnd
| TtsCancel
| BgStart
| BgStop
| EraData
| KioPatchOut
| HintShown
| TimerTick
| CallEnded
| ScoreReady
| ErrorEvent,
Field(discriminator="type"),
]
# ─────────────────────────── курсант → сервер ───────────────────────────
class CallAnswer(BaseModel):
"""Снял гарнитуру. Останавливает норматив `answer`."""
type: Literal["call.answer"] = "call.answer"
class KioPatchIn(BaseModel):
"""Правка карточки. Дебаунс 300 мс, шлётся только дельта."""
type: Literal["kio.patch"] = "kio.patch"
fields: dict[str, Any]
class HintRequest(BaseModel):
"""В режиме `exam` сервер отвечает событием `error`."""
type: Literal["hint.request"] = "hint.request"
class SelfAssessmentSubmit(BaseModel):
"""Самооценка до показа автооценки."""
type: Literal["self_assessment.submit"] = "self_assessment.submit"
missed: list[str]
comment: str = ""
class DdsDispatch(BaseModel):
"""Передача в ДДС. Замораживает карточку снимком и останавливает `dds_notify`."""
type: Literal["dds.dispatch"] = "dds.dispatch"
service: DDSCode
class CallHangup(BaseModel):
type: Literal["call.hangup"] = "call.hangup"
class CallbackDial(BaseModel):
"""Обратный дозвон после обрыва: 3 попытки по 10 с."""
type: Literal["callback.dial"] = "callback.dial"
TraineeToServer = Annotated[
CallAnswer
| KioPatchIn
| HintRequest
| SelfAssessmentSubmit
| DdsDispatch
| CallHangup
| CallbackDial,
Field(discriminator="type"),
]
# ───────────────────────── сервер → наблюдателям ─────────────────────────
class SessionSnapshot(BaseModel):
"""Обязательно при подключении: монитор в классе включают посреди занятия."""
type: Literal["session.snapshot"] = "session.snapshot"
session_id: UUID
scenario_id: str
scenario_title: str
level: Level
mode: SessionMode
trainee_name: str | None = None
started_at: datetime | None = None
kio: KIO
transcript: list[TranscriptEntry]
timers: list[TimerSnapshot]
hints_used: int = 0
ended: bool = False
class TranscriptAppend(BaseModel):
type: Literal["transcript.append"] = "transcript.append"
entry: TranscriptEntry
class KioState(BaseModel):
"""Полная карточка, не дельта: наблюдателю проще, рассинхрон дороже килобайт."""
type: Literal["kio.state"] = "kio.state"
kio: KIO
class ModeSet(BaseModel):
type: Literal["mode.set"] = "mode.set"
mode: SessionMode
class SessionEnded(BaseModel):
type: Literal["session.ended"] = "session.ended"
reason: CallEndReason
class InstructorNoteShown(BaseModel):
"""Пометка преподавателя видна всем, кому виден транскрипт."""
type: Literal["instructor_note.shown"] = "instructor_note.shown"
transcript_ref: str
text: str
author: str
class ReferenceStarted(BaseModel):
"""Автопроигрывание эталонного звонка на внешнем мониторе."""
type: Literal["reference.started"] = "reference.started"
scenario_id: str
ServerToObserver = Annotated[
SessionSnapshot
| TranscriptAppend
| KioState
| TimerTick
| ModeSet
| HintShown
| BgStart
| BgStop
| CallerUtterance
| SessionEnded
| ScoreReady
| InstructorNoteShown
| ReferenceStarted
| ErrorEvent,
Field(discriminator="type"),
]
# ──────────────────── преподаватель → сервер (control) ────────────────────
class ScenarioStart(BaseModel):
type: Literal["scenario.start"] = "scenario.start"
scenario_id: str
trainee: str
group_id: str | None = None
mode: SessionMode
class DirectorInject(BaseModel):
"""Директива звонящему. Карточку курсанта не трогает ни одна команда."""
type: Literal["director.inject"] = "director.inject"
directive: str
mode: DirectiveMode = DirectiveMode.NEXT_TURN
class ReferencePlay(BaseModel):
type: Literal["reference.play"] = "reference.play"
class InstructorNoteAdd(BaseModel):
type: Literal["instructor_note.add"] = "instructor_note.add"
transcript_ref: str
text: str
class ScoreOverride(BaseModel):
"""Коррекция оценки. Автооценка сохраняется рядом."""
type: Literal["score.override"] = "score.override"
session_id: UUID
verdict: str
comment: str
class ScenarioPublish(BaseModel):
type: Literal["scenario.publish"] = "scenario.publish"
scenario_id: str
class SessionStop(BaseModel):
type: Literal["session.stop"] = "session.stop"
InstructorToServer = Annotated[
ScenarioStart
| DirectorInject
| ReferencePlay
| InstructorNoteAdd
| ScoreOverride
| ScenarioPublish
| SessionStop,
Field(discriminator="type"),
]
# ───────────────────────────── станция ДДС ─────────────────────────────
class CardReceived(BaseModel):
"""Снимок КИО. После передачи не меняется — оператор не дописывает задним числом."""
type: Literal["card.received"] = "card.received"
card: KIO
from_operator: str
at: datetime
class CardAck(BaseModel):
"""Останавливает норматив `dds_ack` (≤ 4 с)."""
type: Literal["card.ack"] = "card.ack"
class CardBounce(BaseModel):
"""Возврат на уточнение: неполнота КИО становится сорванным выездом."""
type: Literal["card.bounce"] = "card.bounce"
missing_fields: list[str]
comment: str = ""
class ZoneDecision(BaseModel):
type: Literal["zone.decision"] = "zone.decision"
in_zone: bool
class CrewDispatched(BaseModel):
type: Literal["crew.dispatched"] = "crew.dispatched"
at: datetime
class CrewArrived(BaseModel):
type: Literal["crew.arrived"] = "crew.arrived"
at: datetime
ServerToStation = Annotated[
CardReceived | TimerTick | SessionEnded | ErrorEvent,
Field(discriminator="type"),
]
StationToServer = Annotated[
CardAck | CardBounce | ZoneDecision | CrewDispatched | CrewArrived,
Field(discriminator="type"),
]
# ─────────────────────────── отчёт по HTTP ───────────────────────────
class Metric(BaseModel):
"""Метрика оценки: факт против норматива со ссылкой. Не балл, а обоснование."""
key: str
title: str
fact: str
norm: str
ref: str | None = None
passed: bool
weight: float = 1.0
class CompetencyScore(BaseModel):
competency: str
value: float
class HintUsage(BaseModel):
checklist_id: str
question: str
at: datetime
class SelfAssessment(BaseModel):
missed: list[str]
comment: str = ""
submitted_at: datetime
class SessionReport(BaseModel):
"""Единица истории: из отчётов складываются профиль, дельта попыток, аналитика."""
session_id: UUID
scenario_id: str
mode: SessionMode
attempt: int = 1
transcript: list[TranscriptEntry]
findings: list[Finding]
metrics: list[Metric]
competencies: list[CompetencyScore]
reference_questions: list[HintShown]
missed_checklist: list[str]
hints_used: list[HintUsage]
self_assessment: SelfAssessment | None = None
score_auto: float
score_final: float
overridden_by: str | None = None
override_comment: str | None = None
class EventCatalog(BaseModel):
"""Единственное назначение — собрать все союзы в одну JSON Schema для make types."""
server_to_trainee: ServerToTrainee
trainee_to_server: TraineeToServer
server_to_observer: ServerToObserver
instructor_to_server: InstructorToServer
server_to_station: ServerToStation
station_to_server: StationToServer
session_report: SessionReport

151
backend/app/domain/kio.py Normal file
View file

@ -0,0 +1,151 @@
"""Карточка информационного обмена (КИО).
Это не дизайн формы, это норматив: docs/spec/NORMATIVES.md#карточка-информационного-обмена-кио
Обязательность полей задаётся сценарием (`required_fields`), а не моделью:
в ложном вызове число пострадавших не требуется, в ЧС требуется.
"""
from datetime import datetime
from enum import StrEnum
from typing import Any
from uuid import UUID, uuid4
from pydantic import BaseModel, Field
from app.domain.classifiers import DDSCode, IncidentType
class ResponseStatus(StrEnum):
REGISTERED = "registered"
TRANSFERRED = "transferred"
IN_PROGRESS = "in_progress"
CLOSED = "closed"
class FireDetails(BaseModel):
"""Уточняющие поля 01 — МЧС."""
fire_nature: str | None = None
object_kind: str | None = None
floors: int | None = None
gasified: bool | None = None
people_inside: bool | None = None
class PoliceDetails(BaseModel):
"""Уточняющие поля 02 — полиция."""
offence_kind: str | None = None
suspects: str | None = None
vehicle: str | None = None
class MedicalDetails(BaseModel):
"""Уточняющие поля 03 — медицина."""
reason: str | None = None
conscious: bool | None = None
breathing: bool | None = None
can_move: bool | None = None
age: int | None = None
class UtilityDetails(BaseModel):
"""Уточняющие поля 04 / ЖКХ."""
failure_kind: str | None = None
scale: str | None = None
threat_to_residents: bool | None = None
class KIO(BaseModel):
"""Полная карточка. Наблюдателям уходит целиком (`kio.state`),
курсанту — дельтой (`kio.patch`)."""
# Служебное — заполняется системой
card_id: UUID = Field(default_factory=uuid4)
registered_at: datetime | None = None
response_status: ResponseStatus = ResponseStatus.REGISTERED
# Звонящий
caller_number: str | None = None
caller_name: str | None = None
caller_contact: str | None = None
language: str = "ru"
# Место
okato: str | None = None
address: str | None = None
street: str | None = None
building: str | None = None
entrance: str | None = None
floor: str | None = None
coords: tuple[float, float] | None = None
# Происшествие
incident_type: IncidentType | None = None
description: str | None = None
victims_count: int | None = None
is_emergency: bool = False
life_threat: bool = False
evacuation_needed: bool = False
# ДДС
dds: DDSCode | None = None
dispatch_order_at: datetime | None = None
arrival_at: datetime | None = None
# Уточняющие по службе
fire: FireDetails | None = None
police: PoliceDetails | None = None
medical: MedicalDetails | None = None
utility: UtilityDetails | None = None
#: Поля, которые курсант не редактирует: их проставляет система.
READ_ONLY_FIELDS: frozenset[str] = frozenset(
{"card_id", "registered_at", "caller_number", "response_status"}
)
def get_field(card: KIO, path: str) -> Any:
"""Значение поля по пути вида `floor` или `fire.floors`."""
value: Any = card
for part in path.split("."):
if value is None:
return None
value = getattr(value, part, None)
return value
def missing_fields(card: KIO, required: list[str]) -> list[str]:
"""Незаполненные обязательные поля — основание отметки E5.
Обязательность приходит из сценария, пустой список означает,
что сценарий требований к карточке не предъявляет.
"""
empty: list[str] = []
for path in required:
value = get_field(card, path)
if value is None or (isinstance(value, str) and not value.strip()):
empty.append(path)
return empty
def apply_patch(card: KIO, fields: dict[str, Any]) -> KIO:
"""Применить дельту `kio.patch`. Служебные поля игнорируются.
Вложенные поля приходят плоским путём: {"fire.floors": 5}.
"""
data = card.model_dump()
for path, value in fields.items():
if path in READ_ONLY_FIELDS:
continue
head, _, tail = path.partition(".")
if not tail:
data[head] = value
continue
nested = data.get(head) or {}
nested[tail] = value
data[head] = nested
return KIO.model_validate(data)

View file

@ -0,0 +1,114 @@
"""Таксономия ошибок E1–E6 и компетенции.
Методика: docs/product/METHODOLOGY.md#таксономия-ошибок
Правило: каждая отметка несёт код и обоснование. Отметок без объяснения
не появляется ни в одном интерфейсе.
"""
from datetime import datetime
from enum import StrEnum
from pydantic import BaseModel
class ErrorCode(StrEnum):
E1 = "E1"
E2 = "E2"
E3 = "E3"
E4 = "E4"
E5 = "E5"
E6 = "E6"
class FindingSource(StrEnum):
"""Кто выставил отметку. `judge` — единственный недетерминированный источник."""
SLOTS = "slots"
GROUND_TRUTH = "ground_truth"
TIMERS = "timers"
KIO = "kio"
CHAIN = "chain"
JUDGE = "judge"
INSTRUCTOR = "instructor"
class ErrorSpec(BaseModel):
code: ErrorCode
title: str
detail: str
source: FindingSource
ERRORS: dict[ErrorCode, ErrorSpec] = {
ErrorCode.E1: ErrorSpec(
code=ErrorCode.E1,
title="Пропущенный факт",
detail="Обязательный вопрос не прозвучал, факт не добыт",
source=FindingSource.SLOTS,
),
ErrorCode.E2: ErrorSpec(
code=ErrorCode.E2,
title="Неверная классификация или маршрутизация",
detail="Тип происшествия или ДДС определены неверно",
source=FindingSource.GROUND_TRUTH,
),
ErrorCode.E3: ErrorSpec(
code=ErrorCode.E3,
title="Нарушение временного норматива",
detail="Опрос дольше 75 с, ДДС не оповещена за 60 с и т. д.",
source=FindingSource.TIMERS,
),
ErrorCode.E4: ErrorSpec(
code=ErrorCode.E4,
title="Коммуникативная ошибка",
detail="Тон, эмпатия, управление диалогом, лишние вопросы, игнорирование паники",
source=FindingSource.JUDGE,
),
ErrorCode.E5: ErrorSpec(
code=ErrorCode.E5,
title="Неполнота карточки",
detail="Обязательные поля КИО пусты или заполнены неверно",
source=FindingSource.KIO,
),
ErrorCode.E6: ErrorSpec(
code=ErrorCode.E6,
title="Неверное завершение",
detail="Карточка ушла в ДДС непригодной, снятие с контроля до подтверждения",
source=FindingSource.CHAIN,
),
}
class Competency(StrEnum):
"""Шесть осей радара. Проекция уже посчитанных метрик, без пересчёта весов."""
INTAKE = "intake"
INTERVIEW = "interview"
CARD = "card"
ROUTING = "routing"
NORMS = "norms"
COMMUNICATION = "communication"
COMPETENCY_LABELS: dict[Competency, str] = {
Competency.INTAKE: "Приём вызова",
Competency.INTERVIEW: "Опрос и добыча фактов",
Competency.CARD: "Заполнение КИО",
Competency.ROUTING: "Классификация и маршрутизация",
Competency.NORMS: "Соблюдение нормативов",
Competency.COMMUNICATION: "Коммуникация в стрессе",
}
class Finding(BaseModel):
"""Отметка в разборе. `fact` и `norm` — то самое обоснование."""
code: ErrorCode
source: FindingSource
summary: str
fact: str
norm: str | None = None
ref: str | None = None
competency: Competency | None = None
transcript_ref: str | None = None
at: datetime | None = None

View file

@ -0,0 +1,95 @@
"""Таймеры ГОСТ Р 22.7.03-2021: коды, нормативы, состояние.
Норматив: docs/spec/NORMATIVES.md#временные-нормативы-приёма-вызова
Правило: таймеры останавливаются событиями, а не таймаутами
(docs/arch/CONTRACT.md) — иначе метрика недоказуема.
"""
from enum import StrEnum
from pydantic import BaseModel, Field
GOST_REF = "ГОСТ Р 22.7.03-2021"
class TimerCode(StrEnum):
"""Коды одинаковы на бэке, фронте и в отчёте."""
ANSWER = "answer"
INTERVIEW = "interview"
DDS_NOTIFY = "dds_notify"
DDS_ACK = "dds_ack"
ZONE_CHECK = "zone_check"
CALLBACK = "callback"
CLOSE = "close"
class TimerState(StrEnum):
"""Единственное место, где цвет несёт смысл (docs/arch/FRONTEND.md)."""
OK = "ok"
WARN = "warn"
VIOLATED = "violated"
class Normative(BaseModel):
"""Норматив одной операции приёма вызова."""
code: TimerCode
title: str
limit_ms: int
attempts: int = 1
ref: str = GOST_REF
#: Значения по умолчанию. Рабочие лимиты берутся из конфига (app.config),
#: не из кода — норматив может отличаться у заказчика.
NORMATIVES: dict[TimerCode, Normative] = {
TimerCode.ANSWER: Normative(
code=TimerCode.ANSWER, title="Ответ на звонок", limit_ms=8_000
),
TimerCode.INTERVIEW: Normative(
code=TimerCode.INTERVIEW, title="Опрос / идентификация ситуации", limit_ms=75_000
),
TimerCode.DDS_NOTIFY: Normative(
code=TimerCode.DDS_NOTIFY, title="Оповещение ДДС", limit_ms=60_000
),
TimerCode.DDS_ACK: Normative(
code=TimerCode.DDS_ACK, title="Подтверждение получения карточки", limit_ms=4_000
),
TimerCode.ZONE_CHECK: Normative(
code=TimerCode.ZONE_CHECK, title="Проверка зоны ответственности", limit_ms=30_000
),
TimerCode.CALLBACK: Normative(
code=TimerCode.CALLBACK,
title="Обратный дозвон при обрыве",
limit_ms=10_000,
attempts=3,
),
TimerCode.CLOSE: Normative(
code=TimerCode.CLOSE, title="Снятие с контроля", limit_ms=300_000
),
}
#: Доля норматива, после которой таймер жёлтый.
WARN_RATIO = 0.8
def state_for(elapsed_ms: int, limit_ms: int) -> TimerState:
"""Состояние таймера по факту и нормативу."""
if elapsed_ms > limit_ms:
return TimerState.VIOLATED
if elapsed_ms >= limit_ms * WARN_RATIO:
return TimerState.WARN
return TimerState.OK
class TimerSnapshot(BaseModel):
"""Один таймер в событии `timer.tick`."""
code: TimerCode
elapsed_ms: int
limit_ms: int
state: TimerState
attempt: int = Field(default=1, description="Номер попытки для callback")
stopped: bool = Field(default=False, description="Остановлен событием, не тикает")

28
backend/app/main.py Normal file
View file

@ -0,0 +1,28 @@
"""Сборка приложения. Роутеры подключаются по мере готовности — см. tasks/."""
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.config import get_settings
@asynccontextmanager
async def lifespan(app: FastAPI):
# Прогрев моделей и валидация сценариев — карточки lct-04 и lct-06.
app.state.models_ready = False
yield
app = FastAPI(title="Учебный симулятор занятия для системы 112", lifespan=lifespan)
@app.get("/api/health")
async def health() -> dict:
"""Готовность стенда. Фронт показывает экран «модели прогреваются»."""
settings = get_settings()
return {
"status": "ok",
"models_ready": getattr(app.state, "models_ready", False),
"offline": settings.offline,
}