iAOP/core/edge-gateway/collector/engine.py

# -*- coding: utf-8 -*-
"""周期采集调度引擎 —— 只读采集 + 背压保护 + 健康度上报。

模板化封装要点：
- 采集配置全部外置（gateway.yaml + 点位字典 CSV），引擎不关心具体协议；
- 采样率外置（issue #23）：每个点位按点位字典 CSV 的 sampleRate 列调度，
  引擎以 gateway.yaml 的 interval_ms 为基准 tick，sampleRate > interval_ms 的
  点位按比例降频读取（去硬编码：点位采样率完全由模板配置驱动）；
- 点位按驱动实例的 device 匹配规则分组，一次 tick 内按驱动批量读取；
- 严格只读：引擎只调用 Driver.read_points()，不存在任何控制指令路径；
- 背压保护：未确认(spool 待上行)记录超过阈值时丢弃新样本并计入丢失，
  防止 Kafka 故障时内存/磁盘无限增长；
- 每个 tick 结束记录轮次耗时与样本成败 → HealthMetrics（P99/丢失率/可用性）。
"""
from __future__ import annotations

import logging
import threading
import time
from typing import Dict, List, Optional, Tuple

from drivers.base import Driver, SampleValue
from point_dict.loader import Point, PointDict
from .metrics import HealthMetrics
from .spool import SpoolStore

logger = logging.getLogger("edge_gateway.engine")


class CollectorEngine:
    """只读采集调度引擎（单线程 tick，可替换为 asyncio 版本保持接口不变）。"""

    def __init__(
        self,
        point_dict: PointDict,
        driver_slots: List[Tuple[str, Driver, List[str]]],
        spool: SpoolStore,
        metrics: HealthMetrics,
        interval_ms: int = 1000,
        max_pending: int = 100_000,
    ):
        """
        Args:
            point_dict: 点位字典（已通过校验）；
            driver_slots: [(protocol, driver, device_prefixes)]，
                device_prefixes 为空列表 = 兜底驱动（接收未分配点位）；
            spool: 本地缓存（断点续传）；
            metrics: 健康度统计；
            interval_ms: 采集周期（模板配置，点位字典未覆盖时的默认值）；
            max_pending: 背压阈值（未确认 spool 记录数上限）。
        """
        self.point_dict = point_dict
        self.driver_slots = driver_slots
        self.spool = spool
        self.metrics = metrics
        self.interval_ms = max(50, int(interval_ms))
        self.max_pending = max(1, int(max_pending))
        self._stop = threading.Event()
        self._thread: Optional[threading.Thread] = None
        self._point_to_driver = self._build_routing()
        # 每点位采样调度（issue #23）：tick 计数 + 各点位下次应采集的 tick
        self._tick = 0
        self._next_due_tick: Dict[str, int] = {p.point_id: 0 for p in point_dict.points}

    # ------------------------------------------------------------------
    def _ticks_per_sample(self, p: Point) -> int:
        """点位采样间隔（以基准 tick 计）。

        sampleRate ≤ interval_ms 时每个 tick 都采；
        sampleRate > interval_ms 时按比例降频（四舍五入，至少 1 tick）。
        """
        if p.sample_rate <= 0:
            return 1
        return max(1, int(round(p.sample_rate / self.interval_ms)))

    def _due_points(self) -> List[Point]:
        """本轮 tick 到期待采集的点位（按点位 sampleRate 调度）。"""
        due: List[Point] = []
        for p in self.point_dict.points:
            if self._tick >= self._next_due_tick.get(p.point_id, 0):
                due.append(p)
        return due

    # ------------------------------------------------------------------
    def _build_routing(self) -> Dict[str, Driver]:
        """点位 → 驱动实例路由。

        匹配优先级（issue #21「协议」维度）：
        1. 点位级：点位字典 CSV protocol 列精确匹配驱动槽位协议；
        2. 模板级（缺省）：按设备前缀 device_prefixes 匹配；
        3. 兜底：无前缀的空槽位接收未匹配点位。
        """
        routing: Dict[str, Driver] = {}
        slot_by_protocol: Dict[str, Driver] = {}
        for protocol, driver, prefixes in self.driver_slots:
            if protocol:
                slot_by_protocol.setdefault(protocol, driver)
            for p in self.point_dict.points:
                if p.point_id in routing:
                    continue
                if not prefixes or any(p.device_id.startswith(pre) for pre in prefixes):
                    routing[p.point_id] = driver
        # 点位级协议覆盖：优先于前缀路由
        for p in self.point_dict.points:
            if p.protocol and p.protocol in slot_by_protocol:
                routing[p.point_id] = slot_by_protocol[p.protocol]
        return routing

    # ------------------------------------------------------------------
    def collect_once(self, sink=None) -> int:
        """执行一轮采集：读点位 → 写 spool → 上行。

        Args:
            sink: 可选 KafkaSink；为 None 时仅写入 spool（离线采集模式）。

        Returns:
            本轮成功读到的样本数。
        """
        started = time.monotonic()
        # 本轮 tick 递增 + 取到期待集（按点位 sampleRate 调度，issue #23）
        self._tick += 1
        due = self._due_points()
        expected = len(due)
        got = 0
        samples: List[dict] = []

        # 1) 按驱动分组读取（只读）
        by_driver: Dict[Driver, List[Point]] = {}
        for p in due:
            drv = self._point_to_driver.get(p.point_id)
            if drv is None:
                continue
            by_driver.setdefault(drv, []).append(p)

        for drv, points in by_driver.items():
            try:
                values = drv.read_points(points)
            except Exception:  # 驱动级异常：本轮整体记失败，不中断网关
                logger.exception("驱动读取异常: %s", drv.protocol)
                self.metrics.record_round(time.monotonic() - started, expected, got, failed=True)
                return got
            for p in points:
                # 无论本轮是否读到，都推进该点位采样调度（降频点位不连续空读）
                self._next_due_tick[p.point_id] = self._tick + self._ticks_per_sample(p)
                value = values.get(p.point_id)
                if value is None:
                    continue  # 未读到 → 计入丢失
                got += 1
                ts = time.time()
                samples.append(
                    {"device_id": p.device_id, "point_id": p.point_id,
                     "value": value, "ts": ts, "unit": p.unit}
                )

        # 2) 背压保护：待上行记录超阈值时丢弃新样本
        pending = self.spool.total_pending()
        if pending >= self.max_pending:
            dropped = len(samples)
            samples = []
            # 丢弃样本计入丢失率
            self.metrics.record_round(time.monotonic() - started, expected, got, failed=False)
            logger.warning("背压保护触发：spool 待上行 %d 条 ≥ 阈值 %d，丢弃本轮 %d 条样本",
                           pending, self.max_pending, dropped)
            return got

        # 3) 写 spool（断点续传落盘）
        for s in samples:
            self.spool.append(s["device_id"], s["point_id"], s["value"], s["ts"])

        # 4) 上行（Kafka）；失败由 sink 内部重试/保留 spool
        if sink is not None:
            try:
                sink.publish(samples)
            except Exception:
                logger.exception("上行异常，样本保留在 spool 等待重发")

        latency = time.monotonic() - started
        self.metrics.record_round(latency, expected, got)
        return got

    # ------------------------------------------------------------------
    def run_forever(self, sink=None) -> None:
        """tick 循环入口（供线程调用）。"""
        while not self._stop.is_set():
            try:
                self.collect_once(sink=sink)
            except Exception:
                logger.exception("采集轮次异常")
            # 下一轮 tick 对齐 interval_ms
            self._stop.wait(self.interval_ms / 1000.0)

    def start(self, sink=None) -> None:
        """后台启动采集线程。"""
        if self._thread is not None and self._thread.is_alive():
            return
        self._stop.clear()
        self._thread = threading.Thread(
            target=self.run_forever, args=(sink,), name="edge-gateway-collector", daemon=True
        )
        self._thread.start()

    def stop(self) -> None:
        """停止采集线程。"""
        self._stop.set()
        if self._thread is not None:
            self._thread.join(timeout=5)
            self._thread = None