Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -753,7 +753,7 @@ inference/

### Цикл `python -m inference.runner`

Старт: прогрев окон из `events` за 72 ч, pickle ядра не используется. Такт: новые строки `events` пачкой → один прогноз на лог (`channel`, `CHANNEL_EVENT`, `horizon_until = ts + 30 ч`) → запись пачки одной транзакцией. TTL удаляет истёкшие `catboost-aft-*` без решения диспетчера. Обрыв БД — reconnect с backoff.
Старт: прогрев окон из `events` за 72 ч, pickle ядра не используется. Такт: новые строки `events` пачкой → один прогноз на лог (`channel`, `EQUIPMENT_FAILURE` у канала без класса инцидента, иначе `CHANNEL_EVENT`; `horizon_until = ts + 30 ч`) → запись пачки одной транзакцией. TTL удаляет истёкшие `catboost-aft-*` без решения диспетчера. Обрыв БД — reconnect с backoff.

Пропуски и сбои считаются в строке `tick` раз в минуту: `skipped_stale`, `skipped_ooo`, `skipped_unknown`, `failed`, `failed_write_prep`.

Expand Down
6 changes: 4 additions & 2 deletions docs/incident-v4-500.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,10 @@

The second inference container uses `data/incident4/model_500.cbm` and
`PREDICTION_KIND=INCIDENT`. It writes FIRE, FLOOD, GAS or INTRUSION with a
30-hour horizon. The equipment container retains `CHANNEL_EVENT` and its
own cursor and model version.
30-hour horizon. The equipment container keeps its own cursor and model version. It writes
`EQUIPMENT_FAILURE` only for channels without an incident class: its target is
any alarm, which on a classed channel mixes faults with incidents, so there it
keeps `CHANNEL_EVENT` (hidden by Backend as LEGACY).

`hours_since_prev_target` is the time since the previous target alarm on the
same channel, excluding the current event. `target_seed.json` contains the
Expand Down
9 changes: 9 additions & 0 deletions inference/incident.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,15 @@
"""


def classed_channels(conn: psycopg.Connection) -> frozenset[int]:
"""Каналы с классом инцидента — по реестру, а не по фичам модели."""
rows = conn.execute(
"SELECT channel_id FROM dim_channels_current WHERE sensor_type = ANY(%s)",
(list(SENSOR_CLASSES),),
).fetchall()
return frozenset(int(row[0]) for row in rows)


def incident_class(sensor_type: str | None) -> str | None:
return SENSOR_CLASSES.get(sensor_type)

Expand Down
13 changes: 11 additions & 2 deletions inference/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,9 @@
from inference.config import connect as pg_connect
from inference.cursor import EventCursor
from inference.health import HEARTBEAT_FILE, READY_FILE
from inference.incident import TargetHistory, incident_class, is_target_alarm
from inference.incident import (
TargetHistory, classed_channels, incident_class, is_target_alarm,
)
from inference.predictions import (
PredictionRow,
delete_expired,
Expand Down Expand Up @@ -177,6 +179,7 @@ def __init__(
self.totals = TickCounters()
self.last_tick = TickCounters()
self.targets: TargetHistory | None = None
self.classed: frozenset[int] = frozenset()
self._tick_shap_ms = 0.0
self._logged_at = float("-inf")
self._traceback_at = float("-inf")
Expand Down Expand Up @@ -214,6 +217,7 @@ def start(self, conn: psycopg.Connection) -> None:
low, lag_seconds=self.options.reorder_lag_seconds
)
rows = self._warmup(conn, service, cursor)
self.classed = classed_channels(conn)
if self.settings.prediction_kind == "INCIDENT":
self.targets = TargetHistory.load(
self.path(TARGETS_FILE), cursor.low, cursor.seen
Expand Down Expand Up @@ -365,7 +369,11 @@ def _process(
sensor_type = service.metadata[event.channel_id].get("тип_датчика")
target_alarm = False
previous_hours = None
incident = "CHANNEL_EVENT"
# Модель поломок прогнозирует любую тревогу: на канале без класса это
# отказ оборудования, на канале с классом — смесь с инцидентом, её
# не выдаём за отказ. Тип датчика берём из реестра: в фичах его нет.
incident = ("CHANNEL_EVENT" if event.channel_id in self.classed
else "EQUIPMENT_FAILURE")
if self.targets is not None:
incident = incident_class(sensor_type)
if incident is None:
Expand Down Expand Up @@ -485,6 +493,7 @@ def _reload_metadata(self, conn: psycopg.Connection, reason: str) -> None:
service = self.service
assert service is not None
metadata = fetch_metadata(conn, service.compiled.static_features)
self.classed = classed_channels(conn)

# Rolling-state не трогаем: меняется только справочная часть.
with service.lock:
Expand Down
24 changes: 21 additions & 3 deletions tests/test_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,11 +30,12 @@ def __call__(self) -> float:
return self.now


def add_channel(conn: psycopg.Connection, channel_id: int) -> None:
def add_channel(conn: psycopg.Connection, channel_id: int,
sensor_type: str = "КД АВ") -> None:
conn.execute(
"INSERT INTO dim_channels VALUES "
"(%s, %s, 'Охранная подсистема', 'КД АВ', '15-11.1.131.2.', 'КД АВ', 20)",
(channel_id, SNAPSHOT),
"(%s, %s, 'Охранная подсистема', %s, '15-11.1.131.2.', 'КД АВ', 20)",
(channel_id, SNAPSHOT, sensor_type),
)


Expand Down Expand Up @@ -240,6 +241,23 @@ def test_unknown_channel_picked_up_after_reload(db, make_runner) -> None:
assert [row[1] for row in predictions(db)] == [str(NEW_CHANNEL)]


def test_classless_channel_writes_equipment_failure(db, make_runner) -> None:
runner, conn = started(make_runner, db, threshold=0.0)
now = datetime.now(timezone.utc)

with db.admin() as admin:
add_channel(admin, NEW_CHANNEL, "ИБП")
runner.monotonic.now += 700
runner._reload_metadata(conn, "test")
with db.admin() as admin:
add_event(admin, NEW_CHANNEL, now - timedelta(minutes=1), "Неисправен")
add_event(admin, CHANNEL, now - timedelta(minutes=1), "Неисправен")
runner.tick(conn)

types = {row[1]: row[2] for row in predictions(db)}
assert types == {str(NEW_CHANNEL): "EQUIPMENT_FAILURE", str(CHANNEL): "CHANNEL_EVENT"}


def test_ttl_keeps_decisions_and_foreign_rows(db, make_runner) -> None:
runner, conn = started(make_runner, db)
past = datetime.now(timezone.utc) - timedelta(minutes=1)
Expand Down
Loading