7212251112
Оператор системы-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 чистый.
204 lines
10 KiB
Python
204 lines
10 KiB
Python
"""Детерминированная оценка по ГОСТ.
|
||
|
||
Проверяются пункты приёмки lct-12: воспроизводимость, ни одной отметки без
|
||
обоснования, время опроса, полнота опроса, неверная ДДС.
|
||
"""
|
||
|
||
import json
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from app.domain.events import CallEndReason
|
||
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
|
||
from app.scoring.gost import evaluate
|
||
from app.session.timers import SessionTimers
|
||
|
||
LIBRARY = Path(__file__).resolve().parents[2] / "scenarios"
|
||
ALL_FACTS = ["f_address", "f_what_burns", "f_people", "f_smoke", "f_gas"]
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def scenario():
|
||
return load_file(LIBRARY / "fire-apartment-l2.yaml", LIBRARY)
|
||
|
||
|
||
def timeline(answer_s: float = 5, interview_s: float = 60) -> SessionTimers:
|
||
"""Таймеры с явными моментами событий — без часов, чтобы результат не плыл."""
|
||
timers = SessionTimers()
|
||
timers.on_event("call.incoming", now=0.0)
|
||
timers.on_event("call.answer", now=answer_s)
|
||
timers.on_event("dds.dispatch", now=answer_s + interview_s)
|
||
return timers
|
||
|
||
|
||
def good_card() -> 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):
|
||
result = evaluate(scenario=scenario, kio=good_card(), timers=timeline(), revealed_facts=ALL_FACTS)
|
||
failed = [metric.key for metric in result.metrics if not metric.passed]
|
||
assert failed == [], f"провалено: {failed}"
|
||
assert result.findings == []
|
||
assert result.score == 100.0
|
||
|
||
|
||
def test_same_call_gives_identical_json(scenario):
|
||
"""Оценку можно предъявить и проверить только если она повторяется."""
|
||
def run():
|
||
result = evaluate(
|
||
scenario=scenario, kio=KIO(dds="03"), timers=timeline(interview_s=94),
|
||
revealed_facts=["f_address"],
|
||
)
|
||
return json.dumps(
|
||
{
|
||
"metrics": [m.model_dump() for m in result.metrics],
|
||
"findings": [f.model_dump(mode="json") for f in result.findings],
|
||
"score": result.score,
|
||
},
|
||
ensure_ascii=False,
|
||
sort_keys=True,
|
||
)
|
||
|
||
assert run() == run()
|
||
|
||
|
||
def test_interview_over_75_seconds_is_e3_with_the_numbers(scenario):
|
||
result = evaluate(scenario=scenario, kio=good_card(), timers=timeline(interview_s=94), revealed_facts=ALL_FACTS)
|
||
metric = next(m for m in result.metrics if m.key == "interview_time")
|
||
assert not metric.passed
|
||
assert (metric.fact, metric.norm) == ("94 с", "≤ 75 с")
|
||
assert metric.ref == "ГОСТ Р 22.7.03-2021"
|
||
|
||
finding = next(f for f in result.findings if f.code is ErrorCode.E3)
|
||
assert "94 с" in finding.summary and "75 с" in finding.summary
|
||
|
||
|
||
def test_interview_never_stopped_is_a_failure_not_a_pass(scenario):
|
||
"""Таймер не остановлен событием — время недоказуемо. Молча засчитать
|
||
такую метрику значит поставить зачёт за непереданную карточку."""
|
||
timers = SessionTimers()
|
||
timers.on_event("call.incoming", now=0.0)
|
||
timers.on_event("call.answer", now=3.0)
|
||
result = evaluate(scenario=scenario, kio=good_card(), timers=timers, revealed_facts=ALL_FACTS)
|
||
metric = next(m for m in result.metrics if m.key == "interview_time")
|
||
assert not metric.passed and "не зафиксировано" in metric.fact
|
||
|
||
|
||
def test_each_missing_fact_is_its_own_e1_with_the_reference_question(scenario):
|
||
"""В разборе нужен конкретный пропущенный вопрос, а не процент."""
|
||
result = evaluate(
|
||
scenario=scenario, kio=good_card(), timers=timeline(),
|
||
revealed_facts=["f_address", "f_what_burns", "f_people"],
|
||
)
|
||
e1 = [f for f in result.findings if f.code is ErrorCode.E1]
|
||
assert len(e1) == 2
|
||
assert all("эталонный вопрос" in f.norm for f in e1)
|
||
|
||
completeness = next(m for m in result.metrics if m.key == "checklist_completeness")
|
||
assert completeness.fact == "добыто 3 из 5 обязательных фактов"
|
||
|
||
|
||
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 "квартира" 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):
|
||
for written in ("улица Ленина, 14", "ул. Ленина д. 14 кв. 47", "Ленина 14"):
|
||
card = good_card()
|
||
card.address = written
|
||
result = evaluate(scenario=scenario, kio=card, timers=timeline(), revealed_facts=ALL_FACTS)
|
||
assert next(m for m in result.metrics if m.key == "address").passed, written
|
||
|
||
|
||
def test_wrong_building_is_caught(scenario):
|
||
card = good_card()
|
||
card.address = "улица Ленина, 41"
|
||
result = evaluate(scenario=scenario, kio=card, timers=timeline(), revealed_facts=ALL_FACTS)
|
||
assert not next(m for m in result.metrics if m.key == "address").passed
|
||
|
||
|
||
def test_no_finding_without_justification(scenario):
|
||
"""Ни одной отметки без кода и обоснования — ни в одном интерфейсе."""
|
||
result = evaluate(
|
||
scenario=scenario, kio=KIO(), timers=SessionTimers(), revealed_facts=[],
|
||
end_reason=CallEndReason.DROPPED,
|
||
)
|
||
assert result.findings, "пустая карточка должна дать отметки"
|
||
for finding in result.findings:
|
||
assert finding.code and finding.summary and finding.fact and finding.norm, finding
|
||
|
||
|
||
def test_without_slot_machine_completeness_is_not_silently_passed(scenario):
|
||
result = evaluate(scenario=scenario, kio=good_card(), timers=timeline(), revealed_facts=None)
|
||
assert "checklist_completeness" not in [m.key for m in result.metrics]
|
||
assert any("checklist_completeness" in note for note in result.unavailable)
|
||
|
||
|
||
def test_dropped_call_without_callback_fails(scenario):
|
||
result = evaluate(
|
||
scenario=scenario, kio=good_card(), timers=timeline(), revealed_facts=ALL_FACTS,
|
||
end_reason=CallEndReason.DROPPED,
|
||
)
|
||
callback = next(m for m in result.metrics if m.key == "callback")
|
||
assert not callback.passed and callback.fact == "обратного дозвона не было"
|
||
|
||
|
||
def wrong_routing_card() -> KIO:
|
||
"""Тот же вызов, но признак второго уровня выбран неверно: «квартира»
|
||
вместо «балкона» — другой код ЕКП и другой список оповещения."""
|
||
card = good_card()
|
||
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)}
|
||
|
||
# Маршрутизацию меряет одна метрика — признаки. Выбор службы из неё убран:
|
||
# список оповещения производен, и штрафовать за него отдельно значит
|
||
# наказать дважды за одну ошибку (docs/spec/DATASET.md).
|
||
assert values["routing"] == 0.0, "единственная метрика маршрутизации провалена"
|
||
assert values["card"] == 1.0
|
||
assert Competency.COMMUNICATION.value not in values, "без судьи коммуникация не оценивалась, а не провалена"
|
||
|
||
|
||
def test_legal_interview_is_not_an_e3(scenario):
|
||
"""70 с опроса укладываются в норматив 75 с. Раньше таймер оповещения ДДС
|
||
стартовал на ответе, мерил тот же отрезок с лимитом 60 с и выставлял E3
|
||
за законный опрос. Пока нет события конца опроса, он не считается."""
|
||
result = evaluate(scenario=scenario, kio=good_card(), timers=timeline(interview_s=70), revealed_facts=ALL_FACTS)
|
||
assert [f for f in result.findings if f.code is ErrorCode.E3] == []
|
||
assert any(note.startswith("dds_notify_time") for note in result.unavailable)
|