diff --git a/README.md b/README.md index b06c04e..3010e38 100644 --- a/README.md +++ b/README.md @@ -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`. diff --git a/docs/incident-v4-500.md b/docs/incident-v4-500.md index aa63a41..dca98be 100644 --- a/docs/incident-v4-500.md +++ b/docs/incident-v4-500.md @@ -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 diff --git a/inference/incident.py b/inference/incident.py index 0c9cea6..81456ac 100644 --- a/inference/incident.py +++ b/inference/incident.py @@ -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) diff --git a/inference/runner.py b/inference/runner.py index bf6ba50..3a641c8 100644 --- a/inference/runner.py +++ b/inference/runner.py @@ -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, @@ -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") @@ -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 @@ -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: @@ -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: diff --git a/tests/test_runner.py b/tests/test_runner.py index 5d74432..b093b1f 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -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), ) @@ -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)