feat: классификатор ЕКП — признаки вместо выбора службы (lct-32)

Оператор системы-112 службу не выбирает: он проставляет формализованные
признаки происшествия, комбинация признаков даёт код ЕКП, а коду соответствует
список оповещения, который система собирает сама (docs/spec/DATASET.md).
Прежняя модель с полем dds из пяти значений оценивала действие, которого
в боевой работе нет.

- scripts/import_ekp.py (make ekp): книга заказчика → app/domain/ekp.json,
  1283 кода, 61 служба, 23 группы. Разовый импорт, результат под гитом:
  читать xlsx в рантайме — лишняя зависимость и полсекунды на старте.
- domain/ekp.py: справочник с ленивой загрузкой, каскад значений признаков,
  список оповещения с модификаторами (нет доступа, угроза людям, пострадавшие,
  газификация и ещё десяток).
- КИО: поля signs, incident_code, notify. Последние два только на чтение
  и пересчитываются при каждой правке признаков; добавленная вручную служба
  не теряется, удалить службу нельзя — как в боевом АРМ.
- GET /api/ekp/signs отдаёт один уровень признаков, а не дерево на полмегабайта.
- Карточка на фронте: три каскадных селектора вместо выбора службы.
- Оценка: метрика incident_signs (E2, маршрутизация). Для размеченных сценариев
  dds_choice больше не считается — список оповещения производен от признаков,
  и штрафовать за него отдельно значит наказать дважды за одну ошибку.

Разбор книги оказался основной работой: имя службы лежит то в первой строке
заголовка, то во второй, то склеено с модификатором; «Классификатор МЧС» —
заголовок группы колонок, а не служба. Правила разбора в докстринге импорта.

113 тестов зелёных (14 новых в tests/test_ekp.py), make typecheck чистый.
This commit is contained in:
Ivan Gerasimov 2026-09-19 20:11:50 +03:00
commit 7212251112
16 changed files with 522 additions and 19 deletions

View file

@ -0,0 +1,50 @@
"""Справочник ЕКП по HTTP: группы происшествий и значения признаков.
Отдавать его курсанту безопасно, в отличие от сценария: это классификатор,
а не содержимое вызова. Оператор в боевом АРМ видит тот же список и выбирает
из него — скрывать нечего.
Значения признака зависят от уже выбранных: «жилой дом» → «балкон» → «открытое
пламя». Поэтому уровень запрашивается отдельно, а не отдаётся деревом целиком:
дерево на 1283 листа весит полмегабайта, а нужно из него три десятка строк.
"""
from fastapi import APIRouter, Query
from app.domain import ekp
router = APIRouter(prefix="/api/ekp", tags=["ekp"])
@router.get("/groups")
async def groups() -> dict:
"""Группы происшествий — «категории событий» из ТЗ, тема занятия."""
return {
"version": ekp.reference().version,
"groups": [group.model_dump() for group in ekp.groups()],
}
@router.get("/signs")
async def signs(prefix: list[str] = Query(default=[])) -> dict:
"""Значения следующего признака при уже выбранных.
Пустой `prefix` — значения первого уровня. Возвращается и найденное
происшествие, если признаков хватило: по нему курсант видит, что именно
он классифицировал, до того, как отправит карточку.
"""
level = len(prefix) + 1
found = ekp.by_signs(prefix) if prefix else None
return {
"level": level,
"values": ekp.signs_at_level(level, prefix) if level <= 3 else [],
"incident": (
{
"code": found.code,
"type": found.type,
"notify": ekp.notify_list(found.code),
}
if found
else None
),
}

File diff suppressed because one or more lines are too long

165
backend/app/domain/ekp.py Normal file
View file

@ -0,0 +1,165 @@
"""Справочник ЕКП: признаки → итоговый тип → список оповещения.
Оператор системы-112 службу не выбирает. Он проставляет формализованные признаки
происшествия, комбинация признаков даёт итоговый тип, а итоговому типу соответствует
набор служб, который система формирует сама (docs/spec/DATASET.md).
Данные — `ekp.json`, собранный из книги заказчика.
Грузится один раз и лениво: `make types` импортирует домен, и читать полмегабайта
ради генерации TypeScript незачем.
"""
from functools import lru_cache
from pathlib import Path
from pydantic import BaseModel, Field
DATA = Path(__file__).parent / "ekp.json"
#: Часть строк книги существует ради интеграции с чужими системами, а оператору
#: не показывается — книга помечает их прямо в первом признаке.
HIDDEN_SIGN = "не отображается оператору 112"
#: Условия, меняющие состав списка оповещения. Ключи — из `ekp.json`,
#: значения — то, как это называет оператор.
MODIFIER_LABELS: dict[str, str] = {
"no_access": "нет доступа",
"life_threat": "угроза людям",
"casualties": "есть пострадавшие или погибшие",
"gas": "газифицированный объект",
"evacuation": "требуется эвакуация",
"medical": "нужна медицинская помощь",
"offence": "признаки правонарушения",
"road_block": "перекрыто движение",
"construction": "стройка",
"tunnel": "тоннель",
"pedestrian": "пешеходный объект",
"vehicle": "автомобильный объект",
"mass": "более пяти человек",
"comm_object": "объект связи",
"listed_object": "объект из перечня",
}
class Service(BaseModel):
"""Служба — участник информационного взаимодействия."""
id: str
name: str
class IncidentGroup(BaseModel):
"""Группа происшествий верхнего уровня: их 23, и это то, что преподаватель
выбирает как тему занятия («категория событий» в ТЗ)."""
id: int
title: str
class Incident(BaseModel):
"""Одна строка классификатора: комбинация признаков и её последствия."""
code: str = Field(description="Семизначный код ЕКП")
group: int
signs: list[str] = Field(description="Признаки 1–3, как их выбирает оператор")
extra: str | None = Field(default=None, description="Дополнительные признаки")
type: str = Field(description="Итоговый тип происшествия")
ekp: str = Field(description="Тип по ЕКП-35")
main: str | None = Field(default=None, description="Главная служба")
notify: list[int] = Field(description="Индексы служб базового оповещения")
notify_if: dict[str, list[int]] = Field(
default_factory=dict, description="Службы, добавляемые модификатором"
)
class Reference(BaseModel):
"""Справочник целиком. Версия — часть данных: классификатор обновляется,
и в отчёте должно быть видно, по какой редакции считали."""
version: str
source: str
imported_at: str
groups: list[IncidentGroup]
services: list[Service]
incidents: list[Incident]
@lru_cache(maxsize=1)
def reference() -> Reference:
return Reference.model_validate_json(DATA.read_text(encoding="utf-8"))
@lru_cache(maxsize=1)
def _by_code() -> dict[str, Incident]:
return {incident.code: incident for incident in reference().incidents}
@lru_cache(maxsize=1)
def _by_signs() -> dict[tuple[str, ...], Incident]:
"""Признаки → строка классификатора. Нормализуем регистр: методист пишет
«Жилой дом», книга — «жилой дом», и это одно и то же."""
table: dict[tuple[str, ...], Incident] = {}
for incident in reference().incidents:
table.setdefault(tuple(sign.lower() for sign in incident.signs), incident)
return table
def incident(code: str) -> Incident | None:
return _by_code().get(code)
def by_signs(signs: list[str]) -> Incident | None:
"""Комбинация признаков → происшествие. Порядок признаков значим:
в книге это три уровня, а не множество."""
return _by_signs().get(tuple(sign.strip().lower() for sign in signs if sign.strip()))
def groups() -> list[IncidentGroup]:
return reference().groups
def services() -> list[Service]:
return reference().services
def service_names(indexes: list[int]) -> list[str]:
known = reference().services
return [known[i].name for i in indexes if 0 <= i < len(known)]
def notify_list(code: str, modifiers: list[str] | None = None) -> list[str]:
"""Список оповещения: базовый плюс то, что добавили модификаторы.
Порядок — как в книге заказчика, и он не случаен: экстренные службы идут
первыми, собирающие информацию — последними. Интерфейс показывает начало
списка, поэтому сортировать его по-своему нельзя.
"""
found = incident(code)
if found is None:
return []
indexes = list(found.notify)
for modifier in modifiers or []:
for index in found.notify_if.get(modifier, []):
if index not in indexes:
indexes.append(index)
indexes.sort()
return service_names(indexes)
def signs_at_level(level: int, prefix: list[str] | None = None) -> list[str]:
"""Значения признака уровня 1–3 при уже выбранных предыдущих.
Это источник выпадающих списков в карточке: выбрал «жилой дом» — во втором
признаке остаются только те значения, что встречаются вместе с ним.
"""
chosen = [sign.strip().lower() for sign in (prefix or [])]
values: list[str] = []
for item in reference().incidents:
if len(item.signs) < level or item.signs[0].lower() == HIDDEN_SIGN:
continue
if [sign.lower() for sign in item.signs[: len(chosen)]] != chosen:
continue
value = item.signs[level - 1]
if value not in values:
values.append(value)
return values

View file

@ -102,7 +102,10 @@ class KIO(BaseModel):
intercom_code: str | None = None # спрашивается чек-листом 03, без него скорая стоит у двери
coords: Coords | None = None
# Происшествие
# Происшествие. Оператор проставляет признаки, остальное считает система
# по ЕКП (domain/ekp.py) — службу он не выбирает.
signs: list[str] = Field(default_factory=list, max_length=3)
incident_code: str | None = Field(default=None, description="Код ЕКП по признакам")
incident_type: IncidentType | None = None
description: str | None = None
victims_count: int | None = None
@ -110,8 +113,12 @@ class KIO(BaseModel):
life_threat: bool = False
evacuation_needed: bool = False
# ДДС
dds: DDSCode | None = None
# ДДС. `notify` считается из признаков и правится только системой:
# в боевом АРМ оператор может добавить службу вручную, но не удалить.
notify: list[str] = Field(default_factory=list, description="Список оповещения")
dds: DDSCode | None = Field(
default=None, description="Устаревшее: одна служба. Заменено списком оповещения"
)
dispatch_order_at: datetime | None = None
arrival_at: datetime | None = None
@ -124,7 +131,7 @@ class KIO(BaseModel):
#: Поля, которые курсант не редактирует: их проставляет система.
READ_ONLY_FIELDS: frozenset[str] = frozenset(
{"card_id", "registered_at", "caller_number", "response_status"}
{"card_id", "registered_at", "caller_number", "response_status", "incident_code", "notify"}
)
@ -152,6 +159,23 @@ def missing_fields(card: KIO, required: list[str]) -> list[str]:
return empty
def derive_incident(card: KIO) -> KIO:
"""Пересчитать код происшествия и список оповещения по признакам.
Вызывается после каждой правки признаков: в боевом АРМ список оповещения
пересобирается сам, и это главное, чему учит карточка.
"""
from app.domain import ekp
found = ekp.by_signs(card.signs)
data = card.model_dump()
data["incident_code"] = found.code if found else None
# Добавленную вручную службу не теряем: удалить её оператор не может.
manual = [name for name in card.notify if name not in ekp.notify_list(data["incident_code"] or "")]
data["notify"] = (ekp.notify_list(found.code) if found else []) + manual
return KIO.model_validate(data)
def apply_patch(card: KIO, fields: dict[str, Any]) -> KIO:
"""Применить дельту `kio.patch`. Служебные поля игнорируются.
@ -168,4 +192,5 @@ def apply_patch(card: KIO, fields: dict[str, Any]) -> KIO:
nested = data.get(head) or {}
nested[tail] = value
data[head] = nested
return KIO.model_validate(data)
updated = KIO.model_validate(data)
return derive_incident(updated) if "signs" in fields else updated

View file

@ -7,6 +7,7 @@ from fastapi import FastAPI
from pathlib import Path
from app.api.http import ekp as ekp_api
from app.api.http import scenarios as scenarios_api
from app.api.http import sessions
from app.api.http import trainees
@ -62,6 +63,7 @@ async def lifespan(app: FastAPI):
app = FastAPI(title="Учебный симулятор занятия для системы 112", lifespan=lifespan)
app.include_router(sessions.router)
app.include_router(scenarios_api.router)
app.include_router(ekp_api.router)
app.include_router(trainees.router)
app.include_router(call_ws.router)
app.include_router(observe_ws.router)

View file

@ -10,6 +10,7 @@ from pathlib import Path
import yaml
from pydantic import ValidationError
from app.domain import ekp
from app.domain.classifiers import DDS_BY_INCIDENT
from app.scenarios.schema import ChecklistItem, Scenario
@ -53,6 +54,18 @@ def _derive_ground_truth(scenario: Scenario) -> Scenario:
scenario.ground_truth.incident_type = scenario.type
scenario.ground_truth.dds = DDS_BY_INCIDENT[scenario.type]
scenario.ground_truth.required_facts = required
# Признаки → код ЕКП → список оповещения. Сценарий без признаков остаётся
# рабочим: такие оцениваются по старым полям, пока их не разметят (lct-34).
if scenario.signs:
found = ekp.by_signs(scenario.signs)
if found is None:
raise ScenarioError(
f"{scenario.id}: признаки {scenario.signs} не найдены в классификаторе ЕКП "
f"версии {ekp.reference().version}"
)
scenario.ground_truth.incident_code = found.code
scenario.ground_truth.notify = ekp.notify_list(found.code)
return scenario
@ -71,7 +84,7 @@ def load_file(path: Path, root: Path) -> Scenario:
raw["not_questions"] = _common_not_questions(root) + list(raw.get("not_questions", []))
declared = raw.get("ground_truth") or {}
forbidden = {"incident_type", "dds", "required_facts"} & set(declared)
forbidden = {"incident_type", "dds", "required_facts", "incident_code", "notify"} & set(declared)
if forbidden:
raise ScenarioError(
f"{path.name}: {', '.join(sorted(forbidden))} в ground_truth выводится кодом "

View file

@ -96,6 +96,8 @@ class GroundTruth(Strict):
incident_type: IncidentType | None = None
dds: DDSCode | None = None
incident_code: str | None = None
notify: list[str] = []
required_facts: list[str] = []
address: str | None = None
victims: int | None = None
@ -114,6 +116,10 @@ class Scenario(Strict):
background: Background | None = None
first_line: str
# Признаки происшествия по ЕКП — то, что курсант обязан проставить в карточке.
# Из них загрузчик выводит код и список оповещения (domain/ekp.py).
signs: list[str] = Field(default_factory=list, max_length=3)
facts: list[Fact] = []
checklist: list[ChecklistItem] = []
required_fields: list[str] = Field(default_factory=list)

View file

@ -9,6 +9,7 @@
import re
from dataclasses import dataclass, field
from app.domain import ekp
from app.domain.events import CallEndReason, Metric
from app.domain.kio import KIO, missing_fields
from app.domain.taxonomy import ERRORS, Competency, Finding, FindingSource
@ -182,7 +183,27 @@ def evaluate(
)
# ── классификация и маршрутизация, E2 ──
if truth.incident_type is not None:
#
# Признаки — то, что оператор действительно делает в боевом АРМ-112: службу
# он не выбирает, её считает система по ЕКП (docs/spec/DATASET.md). Поэтому
# размеченный признаками сценарий оценивается по ним, а не по выбору ДДС.
if truth.incident_code is not None:
found = ekp.incident(truth.incident_code)
expected = " · ".join(found.signs) if found else truth.incident_code
actual = " · ".join(kio.signs) if kio.signs else "не проставлены"
build.add(
"incident_signs", "Признаки происшествия",
actual, expected,
passed=kio.incident_code == truth.incident_code,
ref=f"классификатор ЕКП {ekp.reference().version}",
finding=(
f"Признаки «{actual}» дают {kio.incident_code or 'пустой код'}, "
f"верные — «{expected}» ({truth.incident_code})"
),
)
# Список оповещения производен: при верных признаках он верен всегда,
# и штрафовать за него отдельно — штрафовать дважды за одно.
elif truth.incident_type is not None:
actual = kio.incident_type.value if kio.incident_type else "не указан"
build.add(
"incident_type", "Тип происшествия",
@ -191,7 +212,7 @@ def evaluate(
ref="классификатор происшествий",
finding=f"Тип происшествия {actual}, верный — {truth.incident_type.value}",
)
if truth.dds is not None:
if truth.dds is not None and truth.incident_code is None:
actual = kio.dds.value if kio.dds else "не выбрана"
build.add(
"dds_choice", "Выбор ДДС",

View file

@ -13,6 +13,7 @@ METRIC_MAP: dict[str, tuple[ErrorCode, Competency]] = {
"callback": (ErrorCode.E3, Competency.INTAKE),
"checklist_completeness": (ErrorCode.E1, Competency.INTERVIEW),
"interview_time": (ErrorCode.E3, Competency.NORMS),
"incident_signs": (ErrorCode.E2, Competency.ROUTING),
"incident_type": (ErrorCode.E2, Competency.ROUTING),
"dds_choice": (ErrorCode.E2, Competency.ROUTING),
"address": (ErrorCode.E5, Competency.CARD),
@ -29,6 +30,7 @@ METRIC_MAP: dict[str, tuple[ErrorCode, Competency]] = {
#: тренажёр, поэтому он весит больше всего.
METRIC_WEIGHTS: dict[str, float] = {
"checklist_completeness": 4.0,
"incident_signs": 2.0,
"incident_type": 2.0,
"dds_choice": 2.0,
"address": 2.0,

99
backend/tests/test_ekp.py Normal file
View file

@ -0,0 +1,99 @@
"""Справочник ЕКП и карточка, которая из него считается.
Проверяются пункты приёмки lct-32: признаки дают код, список оповещения
производен и не выбирается руками, сценарий размечается признаками.
"""
import pytest
from app.domain import ekp
from app.domain.kio import KIO, derive_incident
def test_reference_loads_whole_book():
"""Импорт не должен незаметно потерять часть строк: их 1283, и это число
сверяется с книгой заказчика."""
reference = ekp.reference()
assert reference.version == "046.24"
assert len(reference.incidents) == 1283
assert len(reference.groups) == 23
assert all(incident.type for incident in reference.incidents), "код без итогового типа"
def test_signs_give_the_code_from_the_book():
found = ekp.by_signs(["жилой дом", "подъезд", "открытое пламя"])
assert found is not None
assert found.code == "1050701"
assert found.type == "пожар: подъезд"
def test_signs_are_case_insensitive():
"""Методист пишет «Жилой дом», книга — «жилой дом»."""
assert ekp.by_signs(["Жилой дом", "Подъезд", "Открытое пламя"]) is not None
def test_next_level_narrows_by_previous():
first = ekp.signs_at_level(1)
assert "жилой дом" in first
second = ekp.signs_at_level(2, ["жилой дом"])
assert "балкон" in second
assert "мусор" not in second, "«мусор» относится к улице, а не к жилому дому"
def test_service_rows_hidden_from_operator_are_not_offered():
"""Часть строк книги существует ради интеграции и оператору не показывается."""
assert all(ekp.HIDDEN_SIGN not in sign.lower() for sign in ekp.signs_at_level(1))
def test_notify_list_is_not_a_single_service():
"""Главное расхождение с прежней моделью: происшествие почти всегда
поднимает больше одной службы (docs/spec/DATASET.md)."""
services = ekp.notify_list("1050701")
assert "Служба 101" in services
assert len(services) > 1
def test_modifier_adds_services_and_never_removes():
base = ekp.notify_list("1050701")
with_gas = ekp.notify_list("1050701", ["gas"])
assert set(base) < set(with_gas), "модификатор только добавляет службы"
assert "МОСГАЗ" in with_gas
def test_unknown_code_is_empty_not_an_error():
"""Сценарий без признаков остаётся рабочим: занятие не должно падать
из-за неразмеченной карточки."""
assert ekp.notify_list("0000000") == []
assert ekp.incident("0000000") is None
def test_card_derives_code_and_notify_from_signs():
card = derive_incident(KIO(signs=["жилой дом", "балкон", "открытое пламя"]))
assert card.incident_code == "1050201"
assert "Служба 101" in card.notify
def test_operator_cannot_edit_the_notify_list():
"""В боевом АРМ службу можно добавить, но не удалить, и список целиком
оператор не переписывает — он производен от признаков."""
from app.domain.kio import apply_patch
card = derive_incident(KIO(signs=["жилой дом", "балкон", "открытое пламя"]))
patched = apply_patch(card, {"notify": [], "incident_code": "9999999"})
assert patched.notify == card.notify
assert patched.incident_code == card.incident_code
def test_changing_signs_recalculates_the_card():
from app.domain.kio import apply_patch
card = derive_incident(KIO(signs=["жилой дом", "балкон", "открытое пламя"]))
moved = apply_patch(card, {"signs": ["жилой дом", "квартира", "открытое пламя"]})
assert moved.incident_code != card.incident_code
assert moved.notify, "после смены признаков список оповещения пересобирается"
@pytest.mark.parametrize("level", [1, 2, 3])
def test_every_level_offers_something(level):
prefix = ["жилой дом", "подъезд"][: level - 1]
assert ekp.signs_at_level(level, prefix), f"уровень {level} пуст"

View file

@ -10,7 +10,7 @@ from pathlib import Path
import pytest
from app.domain.events import CallEndReason
from app.domain.kio import KIO
from app.domain.kio import KIO, derive_incident
from app.domain.taxonomy import Competency, ErrorCode
from app.scenarios.loader import load_file
from app.scoring.competency import radar
@ -36,14 +36,18 @@ def timeline(answer_s: float = 5, interview_s: float = 60) -> SessionTimers:
def good_card() -> KIO:
return KIO(
"""Карточка, заполненная верно. Признаки — то, что курсант проставляет сам;
код ЕКП и список оповещения из них выводит домен, а не тест."""
card = KIO(
address="ул. Ленина, д. 14, кв. 47",
floor="5",
incident_type="fire",
dds="01",
victims_count=2,
description="горит балкон",
signs=["жилой дом", "балкон", "открытое пламя"],
)
return derive_incident(card)
def test_perfect_call_passes_everything(scenario):
@ -110,12 +114,23 @@ def test_each_missing_fact_is_its_own_e1_with_the_reference_question(scenario):
assert completeness.fact == "добыто 3 из 5 обязательных фактов"
def test_wrong_dds_is_e2(scenario):
card = good_card()
card.dds = "03"
def test_wrong_signs_are_e2(scenario):
"""Оператор ошибся признаком — и происшествие уехало не в ту службу.
Службу он не выбирает, поэтому и ошибка именно в признаках."""
card = wrong_routing_card()
result = evaluate(scenario=scenario, kio=card, timers=timeline(), revealed_facts=ALL_FACTS)
finding = next(f for f in result.findings if f.code is ErrorCode.E2)
assert "03" in finding.summary and "01" in finding.summary
assert "квартира" in finding.summary, finding.summary
assert "балкон" in finding.summary, "в отметке не видно верного признака"
def test_notify_list_is_derived_not_chosen(scenario):
"""Список оповещения считается по ЕКП: 01 в нём есть, а выбирать его
оператор не может."""
card = good_card()
assert card.incident_code == scenario.ground_truth.incident_code
assert "Служба 101" in card.notify
assert len(card.notify) > 1, "происшествие почти всегда поднимает больше одной службы"
def test_address_abbreviations_are_not_punished(scenario):
@ -159,13 +174,23 @@ def test_dropped_call_without_callback_fails(scenario):
assert not callback.passed and callback.fact == "обратного дозвона не было"
def test_radar_is_a_projection_of_the_same_metrics(scenario):
def wrong_routing_card() -> KIO:
"""Тот же вызов, но признак второго уровня выбран неверно: «квартира»
вместо «балкона» — другой код ЕКП и другой список оповещения."""
card = good_card()
card.dds = "03"
card.signs = ["жилой дом", "квартира", "открытое пламя"]
return derive_incident(card)
def test_radar_is_a_projection_of_the_same_metrics(scenario):
card = wrong_routing_card()
result = evaluate(scenario=scenario, kio=card, timers=timeline(interview_s=94), revealed_facts=ALL_FACTS)
values = {score.competency: score.value for score in radar(result.metrics)}
assert values["routing"] == 0.5, "из двух метрик маршрутизации провалена одна"
# Маршрутизацию меряет одна метрика — признаки. Выбор службы из неё убран:
# список оповещения производен, и штрафовать за него отдельно значит
# наказать дважды за одну ошибку (docs/spec/DATASET.md).
assert values["routing"] == 0.0, "единственная метрика маршрутизации провалена"
assert values["card"] == 1.0
assert Competency.COMMUNICATION.value not in values, "без судьи коммуникация не оценивалась, а не провалена"