dev: Zigbee koguja häirete logi (uus67.vead)

- collector: FaultTracker — low_battery, weak_signal, low_voltage, offline, mqtt_disconnected, bridge_offline
- ainult muutused (aktiveeritud/taastatud), hüstereesiga, aktiivsed häired seemendatakse DB-st
- LOEMIND: vead skeem, lävendid ja käsud

dev/LOEMIND.md

@@ -63,6 +63,7 @@ Valvurile ei lähe andmebaasi kredentiaale.
 | MQTT allikas | `mqtt://100.80.254.54:1883` |
 | Temperatuuriandur | SONOFF SNZB-02P (`0x70d07efffea3e4f6`, `Temperatuur`) → `uus67.temperatuur` |
 | Kontaktandurid | SONOFF SNZB-04PR2 ×4 (Välisuks, Rõdu, Kontori uks, Kontori aken) → `uus67.ukseandur` |
+| Häired | kõik andurid + süsteem → `uus67.vead` |
 
 **Loogika:**
 - **Temperatuur** (`MQTT_TOPIC`): mõõtmised kogutakse mälus täistunni kaupa ja
@@ -72,6 +73,9 @@ Valvurile ei lähe andmebaasi kredentiaale.
 - **Uksed** (`DOOR_TOPICS`): salvestatakse **ainult olekumuutused**. Perioodilised
   ~2 h südamelöögid (sama `contact`) jäetakse vahele. Käivitusel seemendatakse
   viimane teadaolev olek andmebaasist; esimene nähtud olek salvestatakse baasreana.
+- **Häired** (`CH_FAULT_TABLE`): salvestatakse **ainult muutused** — iga häire
+  kohta kaks sündmust (`aktiveeritud` / `taastatud`). Aktiivsed häired
+  seemendatakse käivitusel andmebaasist, et restart ei dubleeriks.
 
 ### Skeem
 
@@ -116,6 +120,36 @@ ORDER BY (device, ts);
 > `contact: false` = **avatud**. Kinnitatud Zigbee2MQTT HA-discovery'st
 > (`device_class: door`, `payload_on: false`, `payload_off: true`).
 
+Häirete tabel (ainult muutused):
+
+```sql
+CREATE TABLE uus67.vead (
+    ts      DateTime('Europe/Tallinn'),
+    device  LowCardinality(String),
+    tyyp    LowCardinality(String),   -- low_battery | weak_signal | low_voltage | offline | mqtt_disconnected | bridge_offline
+    olek    LowCardinality(String),   -- 'aktiveeritud' / 'taastatud'
+    detail  String,                   -- nt "battery 15%", "linkquality 12"
+    vaartus Nullable(Float64)
+) ENGINE = MergeTree
+PARTITION BY toYYYYMM(ts)
+ORDER BY (device, ts);
+```
+
+| Häire (`tyyp`) | Tingimus | Lävend (aktiveerub / taastub) |
+|----------------|----------|-------------------------------|
+| `low_battery` | `battery_low == true` või `battery` madal | ≤ 20% / ≥ 25% |
+| `weak_signal` | `linkquality` madal | ≤ 20 / ≥ 30 |
+| `low_voltage` | `voltage` madal (kontaktandurid) | ≤ 2600 mV / ≥ 2700 mV |
+| `offline` | andurilt pole teadet | temp 120 min / kontakt 240 min |
+| `mqtt_disconnected` | koguja MQTT-ühendus katkes (`device=koguja`) | – |
+| `bridge_offline` | `zigbee2mqtt/bridge/state = offline` (`device=zigbee2mqtt`) | – |
+
+> Lävendid on üle kirjutatavad `.env`-is / keskkonnamuutujatega
+> (`BATTERY_LOW_THRESHOLD`, `LINKQUALITY_LOW_THRESHOLD`, `VOLTAGE_LOW_THRESHOLD`,
+> `OFFLINE_TIMEOUT_TEMP_MIN`, `OFFLINE_TIMEOUT_CONTACT_MIN`).
+> **`tamper`** on teadlikult välja jäetud — SNZB-04PR2 raporteerib selle
+> pidevalt `true` (tõenäoliselt pööratud), mis annaks valehäireid.
+
 ### Käsud
 
 ```bash
@@ -130,6 +164,10 @@ docker exec clickhouse clickhouse-client --user default --password dev_clickhous
 docker exec clickhouse clickhouse-client --user default --password dev_clickhouse_2026 \
   --query "SELECT * FROM uus67.ukseandur ORDER BY ts DESC LIMIT 24"
 
+# Viimased häired (ainult muutused)
+docker exec clickhouse clickhouse-client --user default --password dev_clickhouse_2026 \
+  --query "SELECT * FROM uus67.vead ORDER BY ts DESC LIMIT 50"
+
 # Koguja taaskäivitamine (nt .env muudatuse järel)
 docker compose -f /opt/docker-compose/dev/docker-compose.yml up -d --build zigbee-collector
 ```

dev/docker-compose.yml

@@ -44,6 +44,7 @@ services:
       - CH_DB=${CH_DB:-uus67}
       - CH_TABLE=${CH_TABLE:-temperatuur}
       - CH_DOOR_TABLE=${CH_DOOR_TABLE:-ukseandur}
+      - CH_FAULT_TABLE=${CH_FAULT_TABLE:-vead}
     depends_on:
       - clickhouse
     networks:

dev/zigbee-collector/collector.py

@@ -1,16 +1,18 @@
 #!/usr/bin/env python3
 """Zigbee andurite koguja.
 
-Kaks voogu:
+Kolm voogu:
 
 1. Temperatuur/niiskus (``MQTT_TOPIC``): mõõtmised koondatakse mälus täistunni
    kaupa ja kirjutatakse iga täistunni järel üks agregeeritud rida
    (``CH_TABLE``, vaikimisi ``uus67.temperatuur``).
 2. Kontaktandurid (``DOOR_TOPICS``): salvestatakse **ainult olekumuutused**
-   (``CH_DOOR_TABLE``, vaikimisi ``uus67.ukseandur``). Perioodilised
-   südamelöögid (sama ``contact``) jäetakse vahele. Käivitusel seemendatakse
-   viimane teadaolev olek andmebaasist; kui rida pole, salvestatakse esimene
-   nähtud olek baasreana.
+   (``CH_DOOR_TABLE``, vaikimisi ``uus67.ukseandur``).
+3. Häired (``CH_FAULT_TABLE``, vaikimisi ``uus67.vead``): **ainult muutused**
+   (``aktiveeritud`` / ``taastatud``). Allikad: aku tühjenemine
+   (``battery_low`` või ``battery`` lävend), nõrk signaal (``linkquality``),
+   madal pinge (``voltage``), anduri kadumine (viimase teate timeout),
+   koguja MQTT-ühenduse kadu ja Zigbee2MQTT bridge'i offline-olek.
 
 Käivitub konteineris, mis on main_ai_network võrgus (ClickHouse nimega
 lahenduv). Välised seadistused tulevad keskkonnamuutujatest.
@@ -23,7 +25,7 @@ import logging
 import os
 import signal
 import threading
-from datetime import datetime
+from datetime import datetime, timedelta
 from typing import Any
 from zoneinfo import ZoneInfo
 
@@ -38,6 +40,14 @@ LOG = logging.getLogger("zigbee-collector")
 
 TZ = ZoneInfo(os.environ.get("TZ_NAME", "Europe/Tallinn"))
 
+
+def _env_float(name: str, default: float) -> float:
+    try:
+        return float(os.environ.get(name, default))
+    except (TypeError, ValueError):
+        return default
+
+
 MQTT_HOST = os.environ.get("MQTT_HOST", "127.0.0.1")
 MQTT_PORT = int(os.environ.get("MQTT_PORT", "1883"))
 MQTT_USER = os.environ.get("MQTT_USER", "")
@@ -48,6 +58,7 @@ DOOR_TOPICS = {
     for t in os.environ.get("DOOR_TOPICS", "").split(",")
     if t.strip()
 }
+BRIDGE_STATE_TOPIC = os.environ.get("BRIDGE_STATE_TOPIC", "zigbee2mqtt/bridge/state")
 
 CH_URL = os.environ.get("CH_URL", "http://clickhouse:8123").rstrip("/")
 CH_USER = os.environ.get("CH_USER", "default")
@@ -55,9 +66,27 @@ CH_PASS = os.environ.get("CH_PASS", "")
 CH_DB = os.environ.get("CH_DB", "uus67")
 CH_TABLE = os.environ.get("CH_TABLE", "temperatuur")
 CH_DOOR_TABLE = os.environ.get("CH_DOOR_TABLE", "ukseandur")
+CH_FAULT_TABLE = os.environ.get("CH_FAULT_TABLE", "vead")
+
+# Häirete lävendid (hüstereesiga: aktiveerub <= X, taastub >= Y).
+BATTERY_ACT = _env_float("BATTERY_LOW_THRESHOLD", 20)
+BATTERY_CLR = _env_float("BATTERY_CLEAR_THRESHOLD", 25)
+LQ_ACT = _env_float("LINKQUALITY_LOW_THRESHOLD", 20)
+LQ_CLR = _env_float("LINKQUALITY_CLEAR_THRESHOLD", 30)
+VOLT_ACT = _env_float("VOLTAGE_LOW_THRESHOLD", 2600)
+VOLT_CLR = _env_float("VOLTAGE_CLEAR_THRESHOLD", 2700)
+
+# Offline-tuvastuse timeoutid (minutites).
+OFFLINE_TEMP_MIN = _env_float("OFFLINE_TIMEOUT_TEMP_MIN", 120)
+OFFLINE_CONTACT_MIN = _env_float("OFFLINE_TIMEOUT_CONTACT_MIN", 240)
 
 # Anduri nimi tuletatakse teemast (zigbee2mqtt/Temperatuur -> Temperatuur).
 DEVICE = os.environ.get("DEVICE") or MQTT_TOPIC.rsplit("/", 1)[-1]
+DOOR_DEVICES = {t.rsplit("/", 1)[-1] for t in DOOR_TOPICS}
+MONITORED_DEVICES = {DEVICE, *DOOR_DEVICES}
+
+SYSTEM_DEVICE_BRIDGE = "zigbee2mqtt"
+SYSTEM_DEVICE_COLLECTOR = "koguja"
 
 CHECK_INTERVAL_SEC = 20
 
@@ -73,6 +102,30 @@ def _as_int(value: Any) -> int | None:
     return int(value)
 
 
+def _as_float(value: Any) -> float | None:
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        return None
+    return float(value)
+
+
+def _fault_row(
+    ts: datetime,
+    device: str,
+    tyyp: str,
+    olek: str,
+    detail: str,
+    value: float | None,
+) -> dict[str, Any]:
+    return {
+        "ts": ts.strftime("%Y-%m-%d %H:%M:%S"),
+        "device": device,
+        "tyyp": tyyp,
+        "olek": olek,
+        "detail": detail,
+        "vaartus": value,
+    }
+
+
 class HourBucket:
     """Ühe tunni jooksul kogutud temperatuuri/niiskuse mõõtmised."""
 
@@ -151,6 +204,91 @@ class DoorTracker:
         }
 
 
+class FaultTracker:
+    """Häirete olekumuutuste jälgija (ainult muutused).
+
+    ``current_keys`` hoiab iga seadme jaoks parajasti aktiivsete häirete
+    tüüpe. Iga hindamise järel emiteeritakse read ainult üleminekute kohta
+    (``aktiveeritud`` / ``taastatud``). Käivitusel seemendatakse aktiivsed
+    häired andmebaasist, et restart ei dubleeriks kirjeid.
+    """
+
+    def __init__(self, active: dict[str, set[str]] | None = None) -> None:
+        self.payload_faults: dict[str, dict[str, tuple[str, float | None]]] = {}
+        self.offline_faults: dict[str, tuple[str, float | None]] = {}
+        self.current_keys: dict[str, set[str]] = {
+            device: set(tyybid) for device, tyybid in (active or {}).items()
+        }
+
+    def _low(self, device: str, tyyp: str, value: float, activate_at: float, clear_at: float) -> bool:
+        if tyyp in self.current_keys.get(device, set()):
+            return value < clear_at
+        return value <= activate_at
+
+    def _derive(self, device: str, payload: dict[str, Any]) -> dict[str, tuple[str, float | None]]:
+        faults: dict[str, tuple[str, float | None]] = {}
+        battery_low = payload.get("battery_low")
+        battery = _as_float(payload.get("battery"))
+        if battery_low is True:
+            faults["low_battery"] = ("battery_low=true", battery)
+        elif battery is not None and self._low(device, "low_battery", battery, BATTERY_ACT, BATTERY_CLR):
+            faults["low_battery"] = (f"battery {battery:.0f}%", battery)
+        lq = _as_float(payload.get("linkquality"))
+        if lq is not None and self._low(device, "weak_signal", lq, LQ_ACT, LQ_CLR):
+            faults["weak_signal"] = (f"linkquality {lq:.0f}", lq)
+        voltage = _as_float(payload.get("voltage"))
+        if voltage is not None and self._low(device, "low_voltage", voltage, VOLT_ACT, VOLT_CLR):
+            faults["low_voltage"] = (f"voltage {voltage:.0f} mV", voltage)
+        return faults
+
+    def _reconcile(self, device: str, ts: datetime) -> list[dict[str, Any]]:
+        current: dict[str, tuple[str, float | None]] = {}
+        current.update(self.payload_faults.get(device, {}))
+        if device in self.offline_faults:
+            current["offline"] = self.offline_faults[device]
+        prev = self.current_keys.get(device, set())
+        rows: list[dict[str, Any]] = []
+        for tyyp, (detail, value) in current.items():
+            if tyyp not in prev:
+                rows.append(_fault_row(ts, device, tyyp, "aktiveeritud", detail, value))
+        for tyyp in prev - set(current):
+            rows.append(_fault_row(ts, device, tyyp, "taastatud", "", None))
+        self.current_keys[device] = set(current)
+        return rows
+
+    def on_device_message(
+        self, device: str, payload: dict[str, Any], ts: datetime
+    ) -> list[dict[str, Any]]:
+        self.payload_faults[device] = self._derive(device, payload)
+        self.offline_faults.pop(device, None)
+        return self._reconcile(device, ts)
+
+    def set_offline(
+        self, device: str, offline: bool, detail: str, ts: datetime
+    ) -> list[dict[str, Any]]:
+        if offline:
+            self.offline_faults[device] = (detail, None)
+        else:
+            self.offline_faults.pop(device, None)
+        return self._reconcile(device, ts)
+
+    def set_system_fault(
+        self,
+        device: str,
+        tyyp: str,
+        active: bool,
+        detail: str,
+        ts: datetime,
+    ) -> list[dict[str, Any]]:
+        faults = dict(self.payload_faults.get(device, {}))
+        if active:
+            faults[tyyp] = (detail, None)
+        else:
+            faults.pop(tyyp, None)
+        self.payload_faults[device] = faults
+        return self._reconcile(device, ts)
+
+
 class ClickHouseWriter:
     def __init__(self, table: str) -> None:
         self.table = table
@@ -159,6 +297,9 @@ class ClickHouseWriter:
     def enqueue(self, row: dict[str, Any]) -> None:
         self.pending.append(row)
 
+    def enqueue_many(self, rows: list[dict[str, Any]]) -> None:
+        self.pending.extend(rows)
+
     def flush(self) -> None:
         if not self.pending:
             return
@@ -185,6 +326,17 @@ class ClickHouseWriter:
             )
 
 
+def _ch_query(query: str) -> str:
+    resp = requests.post(
+        f"{CH_URL}/",
+        params={"database": CH_DB, "query": query},
+        auth=(CH_USER, CH_PASS),
+        timeout=15,
+    )
+    resp.raise_for_status()
+    return resp.text
+
+
 def load_last_states() -> dict[str, bool]:
     """Loeb andmebaasist iga anduri viimase teadaoleva ``contact`` väärtuse."""
     query = (
@@ -192,15 +344,8 @@ def load_last_states() -> dict[str, bool]:
         f"FROM {CH_DB}.{CH_DOOR_TABLE} GROUP BY device FORMAT JSONEachRow"
     )
     try:
-        resp = requests.post(
-            f"{CH_URL}/",
-            params={"database": CH_DB, "query": query},
-            auth=(CH_USER, CH_PASS),
-            timeout=15,
-        )
-        resp.raise_for_status()
         states: dict[str, bool] = {}
-        for line in resp.text.splitlines():
+        for line in _ch_query(query).splitlines():
             if not line.strip():
                 continue
             obj = json.loads(line)
@@ -213,25 +358,107 @@ def load_last_states() -> dict[str, bool]:
         return {}
 
 
+def load_active_faults() -> dict[str, set[str]]:
+    """Loeb andmebaasist iga seadme parajasti aktiivsed häirete tüübid."""
+    query = (
+        f"SELECT device, tyyp, argMax(olek, ts) AS olek "
+        f"FROM {CH_DB}.{CH_FAULT_TABLE} GROUP BY device, tyyp FORMAT JSONEachRow"
+    )
+    try:
+        active: dict[str, set[str]] = {}
+        for line in _ch_query(query).splitlines():
+            if not line.strip():
+                continue
+            obj = json.loads(line)
+            if obj.get("olek") == "aktiveeritud":
+                active.setdefault(obj["device"], set()).add(obj["tyyp"])
+        LOG.info("Seemendatud %d seadme aktiivsed häired", len(active))
+        return active
+    except Exception:
+        LOG.exception("Aktiivsete häirete laadimine ebaõnnestus, alustan tühjalt")
+        return {}
+
+
+def load_last_seen() -> dict[str, datetime]:
+    """Loeb andmebaasist iga anduri viimase kirje aja (offline-baas)."""
+    query = (
+        "SELECT device, max(ts) AS ts FROM ("
+        f"SELECT device, ts FROM {CH_DB}.{CH_DOOR_TABLE} "
+        "UNION ALL "
+        f"SELECT device, hour AS ts FROM {CH_DB}.{CH_TABLE}"
+        ") GROUP BY device FORMAT JSONEachRow"
+    )
+    try:
+        seen: dict[str, datetime] = {}
+        for line in _ch_query(query).splitlines():
+            if not line.strip():
+                continue
+            obj = json.loads(line)
+            seen[obj["device"]] = datetime.strptime(obj["ts"], "%Y-%m-%d %H:%M:%S").replace(
+                tzinfo=TZ
+            )
+        LOG.info("Seemendatud %d anduri viimane teate aeg", len(seen))
+        return seen
+    except Exception:
+        LOG.exception("Viimaste teate aegade laadimine ebaõnnestus, alustan tühjalt")
+        return {}
+
+
 def main() -> int:
     lock = threading.Lock()
     stop = threading.Event()
     bucket = HourBucket(current_hour())
     temp_writer = ClickHouseWriter(CH_TABLE)
     door_writer = ClickHouseWriter(CH_DOOR_TABLE)
+    fault_writer = ClickHouseWriter(CH_FAULT_TABLE)
     door_tracker = DoorTracker(load_last_states())
+    fault_tracker = FaultTracker(load_active_faults())
+
+    start = datetime.now(TZ)
+    last_seen: dict[str, datetime] = {}
+    for device, seen in load_last_seen().items():
+        last_seen[device] = seen
+    for device in MONITORED_DEVICES:
+        last_seen.setdefault(device, start)
+
+    def offline_timeout(device: str) -> float:
+        if device == DEVICE:
+            return OFFLINE_TEMP_MIN
+        return OFFLINE_CONTACT_MIN
 
     def on_connect(client, userdata, flags, reason_code, properties=None):
         if reason_code == 0:
-            topics = [MQTT_TOPIC, *sorted(DOOR_TOPICS)]
+            topics = [MQTT_TOPIC, BRIDGE_STATE_TOPIC, *sorted(DOOR_TOPICS)]
             LOG.info("MQTT ühendatud (%s:%s), tellin %s", MQTT_HOST, MQTT_PORT, ", ".join(topics))
             for topic in topics:
                 client.subscribe(topic, qos=1)
+            if not stop.is_set():
+                with lock:
+                    fault_writer.enqueue_many(
+                        fault_tracker.set_system_fault(
+                            SYSTEM_DEVICE_COLLECTOR,
+                            "mqtt_disconnected",
+                            False,
+                            "",
+                            datetime.now(TZ),
+                        )
+                    )
         else:
             LOG.error("MQTT ühendus ebaõnnestus: %s", reason_code)
 
     def on_disconnect(client, userdata, flags, reason_code, properties=None):
         LOG.warning("MQTT ühendus katkes: %s", reason_code)
+        if not stop.is_set():
+            with lock:
+                fault_writer.enqueue_many(
+                    fault_tracker.set_system_fault(
+                        SYSTEM_DEVICE_COLLECTOR,
+                        "mqtt_disconnected",
+                        True,
+                        f"MQTT ühendus katkes: {reason_code}",
+                        datetime.now(TZ),
+                    )
+                )
 
     def on_message(client, userdata, msg):
         try:
@@ -245,12 +472,27 @@ def main() -> int:
         with lock:
             if msg.topic == MQTT_TOPIC:
                 bucket.add(payload)
+                last_seen[DEVICE] = now
+                fault_writer.enqueue_many(fault_tracker.on_device_message(DEVICE, payload, now))
             elif msg.topic in DOOR_TOPICS:
                 device = msg.topic.rsplit("/", 1)[-1]
+                last_seen[device] = now
                 row = door_tracker.process(device, payload, now)
                 if row is not None:
                     LOG.info("Olekumuutus: %s -> %s", device, row["olek"])
                     door_writer.enqueue(row)
+                fault_writer.enqueue_many(fault_tracker.on_device_message(device, payload, now))
+            elif msg.topic == BRIDGE_STATE_TOPIC:
+                offline = payload.get("state") == "offline"
+                fault_writer.enqueue_many(
+                    fault_tracker.set_system_fault(
+                        SYSTEM_DEVICE_BRIDGE,
+                        "bridge_offline",
+                        offline,
+                        "zigbee2mqtt bridge offline",
+                        now,
+                    )
+                )
 
     client = mqtt.Client(
         mqtt.CallbackAPIVersion.VERSION2,
@@ -273,12 +515,14 @@ def main() -> int:
     signal.signal(signal.SIGINT, shutdown)
 
     LOG.info(
-        "Koguja käivitatud: temp=%s.%s, uksed=%s.%s (%d teemat)",
+        "Koguja käivitatud: temp=%s.%s, uksed=%s.%s (%d teemat), häired=%s.%s",
         CH_DB,
         CH_TABLE,
         CH_DB,
         CH_DOOR_TABLE,
         len(DOOR_TOPICS),
+        CH_DB,
+        CH_FAULT_TABLE,
     )
 
     while not stop.is_set():
@@ -290,13 +534,27 @@ def main() -> int:
                 bucket = HourBucket(current_hour(now))
             else:
                 finished = None
-        if finished is not None and not finished.empty:
-            temp_writer.enqueue(finished.row())
+            if finished is not None and not finished.empty:
+                temp_writer.enqueue(finished.row())
+            for device in sorted(MONITORED_DEVICES):
+                seen = last_seen.get(device, start)
+                timeout = timedelta(minutes=offline_timeout(device))
+                offline = (now - seen) > timeout
+                detail = (
+                    f"viimane teade {seen.strftime('%Y-%m-%d %H:%M:%S')}"
+                    if offline
+                    else ""
+                )
+                fault_writer.enqueue_many(
+                    fault_tracker.set_offline(device, offline, detail, now)
+                )
         temp_writer.flush()
         door_writer.flush()
+        fault_writer.flush()
 
     temp_writer.flush()
     door_writer.flush()
+    fault_writer.flush()
     client.loop_stop()
     client.disconnect()
     return 0