video_monitor/app/analysis/pipeline.py

"""单摄像头分析流水线（实时解码 + 实时分析）

架构：
- 解码线程 _decode_loop：持续从 FrameSource 抓帧，写入有界 deque(maxlen=2)。
  队列满时自动丢弃最旧帧（deque 特性），绝不阻塞解码、也不阻塞分析。
- 分析循环 run()：从 deque 取最新帧，按 analyze_fps 节流跳帧，运行
  motion 门控 → 多检测器 → IoUTracker → 布控检测；仅业务算法规则命中时截图并报警。
- FPS 计数：每秒滚动刷新 self._analysis_fps，供 openStatus 实时展示。

事件通过 callback 上报给上层服务层（review/timeline/tracking），本模块不直接访问 ORM。
"""
import collections
import logging
import os
import threading
import time

from app.analysis.frames import FrameSource
from app.analysis.motion import MotionDetector
from app.analysis.tracker import IoUTracker

logger = logging.getLogger("analysis.pipeline")

try:
    import cv2
    _CV2_AVAILABLE = True
except Exception:
    cv2 = None
    _CV2_AVAILABLE = False

# 快照保存目录（相对项目根/static/storage/snapshots/）
_SNAPSHOT_DIRNAME = "snapshots"
_SNAPSHOT_MAX_PER_MIN = 60  # 每路每分钟最多快照数，避免磁盘暴涨


class CameraPipeline(object):
    """单摄像头分析流水线（解码线程 + 分析线程分离）"""

    def __init__(self, stream_id, stream_code, rtsp_url,
                 detector=None,
                 detectors=None,
                 motion=None,
                 target_fps=5,
                 analyze_fps=None,
                 on_event=None,
                 on_track_snapshot=None,
                 on_preview=None,
                 alarm_enabled=True,
                 zone_polygons=None,
                 storage_alarm_dir=None,
                 static_dir=None,
                 queue_size=2):
        self.stream_id = stream_id
        self.stream_code = stream_code
        self.rtsp_url = rtsp_url
        # 子进程内不可 import GlobalUtils/Django，路径由 AnalysisManager 注入
        self._storage_alarm_dir = storage_alarm_dir or ""
        self._static_dir = static_dir or ""
        self.target_fps = target_fps
        # analyze_fps：算法分析频率（帧/秒），支持小于 1（如 2 秒 1 帧 = 0.5）
        fps = float(analyze_fps if analyze_fps is not None else target_fps)
        self.analyze_fps = fps if fps > 0 else 1.0
        self._analyze_interval = 1.0 / self.analyze_fps

        # 多检测器：list[dict(algorithm_id, algorithm_name, engine)]
        self._detectors = list(detectors or [])
        # 兼容旧调用：单 detector 入参也并入 _detectors
        if detector is not None and not self._detectors:
            self._detectors = [{"algorithm_id": 0, "algorithm_name": "legacy", "engine": detector}]
        # _detector 指向首个引擎，便于 info() / 兼容外部访问
        self._detector = self._detectors[0]["engine"] if self._detectors else None
        self._algorithm_name = (", ".join(d["algorithm_name"] for d in self._detectors)
                                if self._detectors else "motion-only")

        self._motion = motion or MotionDetector()
        self._tracker = IoUTracker()
        self._frame_source = FrameSource(rtsp_url, target_fps=target_fps)
        self._on_event = on_event or (lambda *a, **kw: None)
        self._on_track_snapshot = on_track_snapshot or (lambda *a, **kw: None)
        self._on_preview = on_preview or (lambda *a, **kw: None)
        self._alarm_enabled = bool(alarm_enabled)
        self.zone_polygons = zone_polygons or []

        # 解码/分析分离
        self._frame_queue = collections.deque(maxlen=max(1, int(queue_size)))
        self._queue_lock = threading.Lock()
        self._decode_thread = None
        self._decode_running = False

        # 帧计数与 FPS
        self._frame_index = 0
        self._decoded_count = 0
        self._analyzed_count = 0
        self._dropped_count = 0
        self._analysis_fps = 0.0
        self._decode_fps = 0.0
        self._fps_window_ts = time.time()
        self._fps_window_analyzed = 0
        self._fps_window_decoded = 0
        self._last_analyze_ts = 0.0

        self._running = False
        self._started_event = threading.Event()
        self._analysis_error = ""
        self._last_analysis_ts = 0.0
        self._last_zone_state = {}  # track_id -> set(zone_ids)
        self._outside_zone_hits = {}  # (track_id, zone_id) -> 连续明确在区域外次数
        self._track_enter_ts = {}   # (track_id, zone_id) -> ts（滞留计时）
        self._last_w = 0
        self._last_h = 0
        self._last_frame = None     # 最近一帧（供事件触发时绘制快照）
        # 快照节流：每路每分钟上限
        self._snapshot_times = collections.deque(maxlen=_SNAPSHOT_MAX_PER_MIN)
        self._llm_zone_last_ts = {}     # (zone_id, biz_id) -> ts
        self._llm_track_last_ts = {}    # (track_id, biz_id) -> ts
        self._loiter_fired = set()      # (track_id, zone_id, biz_id) 本次停留已报滞留（按算法独立）
        # 扩展后处理运行时状态
        self._track_centers = {}        # track_id -> (cx, cy) 上一帧中心，用于越线/方向判断
        self._line_cross_fired = set()  # (track_id, zone_id, biz_id) 本次已报越线（避免重复）
        self._line_count_state = {}     # (zone_id, biz_id) -> {forward, reverse}
        self._line_count_alert_ts = {}  # (zone_id, biz_id, direction) -> last alert ts
        self._density_fired = set()     # (zone_id, biz_id) 本次密度已报警（按算法独立，目标数降回前不重复）
        self._direction_fired = set()   # (track_id, zone_id, biz_id) 方向报警去重
        self._area_alarm_last_ts = {}   # (track_id, zone_id, biz_id) -> ts 区域入侵持续报警节流
        self._force_detect = False        # 有区域类布控时持续跑检测，避免 MOG2 门控漏报
        self._zone_config_warned = set()
        self._start_ts = 0.0
        self._last_frame_ts = 0.0
        self._current_capture_ts = 0.0
        self._update_detector_policy()

    def _update_detector_policy(self):
        """有区域/越线等小模型后处理时，不应仅依赖运动门控触发检测。"""
        self._force_detect = False
        for z in self.zone_polygons or []:
            coords = z.get("coords") or []
            zid = z.get("id")
            for r in (z.get("biz_algorithms") or []):
                flow = int(r.get("flow_type") or 0)
                post = r.get("post_process") or ""
                if flow not in (1, 3, 4):
                    continue
                if post in ("AREA", "DWELL", "DENSITY", "LINE_CROSS", "LINE_COUNT", "DIRECTION"):
                    self._force_detect = True
                    if len(coords) < 3 and post in ("AREA", "DWELL", "DENSITY"):
                        key = (zid, post)
                        if key not in self._zone_config_warned:
                            self._zone_config_warned.add(key)
                            logger.warning(
                                "pipeline[%s] 布控 #%s「%s」缺少有效区域(≥3点)，%s 不会触发",
                                self.stream_code, zid, z.get("name") or "—", post,
                            )
        if self._force_detect:
            logger.info(
                "pipeline[%s] 已启用持续目标检测（布控含区域/越线类后处理，不受运动门控限制）",
                self.stream_code,
            )

    def set_zone_polygons(self, zones):
        self.zone_polygons = zones or []
        self._update_detector_policy()
        self.reset_zone_runtime_state()

    def reset_zone_runtime_state(self):
        """热更新/重启布控后清空运行时状态，使进入区域报警可重新触发。"""
        self._last_zone_state = {}
        self._outside_zone_hits = {}
        self._track_enter_ts = {}
        self._llm_zone_last_ts = {}
        self._llm_track_last_ts = {}
        self._loiter_fired = set()
        self._track_centers = {}
        self._line_cross_fired = set()
        self._line_count_state = {}
        self._line_count_alert_ts = {}
        self._density_fired = set()
        self._direction_fired = set()
        self._area_alarm_last_ts = {}
        try:
            self._tracker.reset()
        except Exception:
            pass

    def set_analyze_fps(self, analyze_fps):
        fps = float(analyze_fps)
        if fps <= 0:
            fps = 1.0
        self.analyze_fps = fps
        self._analyze_interval = 1.0 / fps
        logger.info("pipeline[%s] 分析频率调整为 %.3f fps (间隔 %.2fs)",
                    self.stream_code, fps, self._analyze_interval)

    def _frame_size(self, frame_index):
        return self._last_h, self._last_w

    @staticmethod
    def _scale_zone(coords, w, h):
        if not coords or w <= 0 or h <= 0:
            return coords
        out = []
        for p in coords:
            try:
                nx, ny = float(p[0]), float(p[1])
            except Exception:
                continue
            if nx > 1.0 or ny > 1.0:
                return coords
            out.append((nx * w, ny * h))
        return out

    # ============ 解码线程 ============
    def _decode_loop(self):
        logger.info("pipeline[%s] 解码线程启动 url=%s" % (self.stream_code, self.rtsp_url))
        while self._decode_running and self._running:
            try:
                if self._frame_source._cap is None or not self._frame_source._cap.isOpened():
                    try:
                        self._frame_source.open()
                    except Exception as e:
                        logger.warning(
                            "pipeline[%s] 打开流失败，%ss 后重试: %s",
                            self.stream_code, self._frame_source.reconnect_interval, str(e),
                        )
                        time.sleep(self._frame_source.reconnect_interval)
                        continue
                ok, frame = self._frame_source.read()
                if not ok or frame is None:
                    time.sleep(0.2)
                    continue
                self._last_frame_ts = time.time()
                with self._queue_lock:
                    if len(self._frame_queue) >= self._frame_queue.maxlen:
                        self._dropped_count += 1
                    self._frame_queue.append((frame, self._last_frame_ts))
                self._decoded_count += 1
                self._fps_window_decoded += 1
            except Exception as e:
                logger.warning("pipeline[%s] 解码异常: %s" % (self.stream_code, str(e)))
                time.sleep(0.5)
        try:
            self._frame_source.close()
        except Exception:
            pass
        logger.info("pipeline[%s] 解码线程退出 (decoded=%d dropped=%d)"
                    % (self.stream_code, self._decoded_count, self._dropped_count))

    def _pop_latest_frame(self):
        """取队列里最新一帧，丢弃中间帧（保证分析总是处理最近画面）"""
        with self._queue_lock:
            if not self._frame_queue:
                return None
            # 取最右（最新），清空其余
            frame = self._frame_queue[-1]
            self._frame_queue.clear()
            return frame

    # ============ 分析循环 ============
    def run(self):
        self._running = True
        self._start_ts = time.time()
        self._decode_running = True
        self._started_event.set()
        self._decode_thread = threading.Thread(
            target=self._decode_loop, name="decode-%s" % self.stream_id, daemon=True)
        self._decode_thread.start()
        logger.info("pipeline[%s] 分析循环启动 analyze_fps=%.3f interval=%.2fs detectors=%d"
                    % (self.stream_code, self.analyze_fps, self._analyze_interval, len(self._detectors)))
        try:
            while self._running:
                packet = self._pop_latest_frame()
                if packet is None:
                    time.sleep(0.02)
                    continue
                if isinstance(packet, tuple) and len(packet) == 2:
                    frame, capture_ts = packet
                else:  # 兼容测试和旧调用
                    frame, capture_ts = packet, time.time()
                # 跳帧节流：未到分析间隔则丢弃此帧
                now = time.time()
                if now - self._last_analyze_ts < self._analyze_interval:
                    continue
                self._last_analyze_ts = now
                self._frame_index += 1
                try:
                    self._last_h, self._last_w = frame.shape[:2]
                except Exception:
                    pass
                self._last_frame = frame
                self._current_capture_ts = capture_ts
                try:
                    self._process_frame(frame)
                    self._analysis_error = ""
                    self._last_analysis_ts = time.time()
                    self._analyzed_count += 1
                    self._fps_window_analyzed += 1
                except Exception as e:
                    self._analysis_error = str(e)
                    logger.warning("pipeline[%s] 处理帧异常: %s" % (self.stream_code, str(e)))
                self._refresh_fps()
        except Exception as e:
            logger.exception("pipeline[%s] 异常: %s" % (self.stream_code, str(e)))
        finally:
            self._running = False
            self._decode_running = False
            try:
                if self._decode_thread and self._decode_thread.is_alive():
                    self._decode_thread.join(timeout=2)
            except Exception:
                pass
            logger.info("pipeline[%s] 已停止 (analyzed=%d)" % (self.stream_code, self._analyzed_count))

    def _refresh_fps(self):
        """每 1 秒滚动刷新一次解码/分析 FPS"""
        now = time.time()
        elapsed = now - self._fps_window_ts
        if elapsed >= 1.0:
            self._analysis_fps = self._fps_window_analyzed / elapsed
            self._decode_fps = self._fps_window_decoded / elapsed
            self._fps_window_analyzed = 0
            self._fps_window_decoded = 0
            self._fps_window_ts = now

    def _process_frame(self, frame, capture_ts=None):
        motion_boxes = self._motion.detect(frame)
        has_motion = len(motion_boxes) > 0
        run_detect = has_motion or self._force_detect

        detections = []
        if run_detect:
            if self._detectors:
                # 多检测器：依次推理，合并结果，每条标注来源算法
                for d in self._detectors:
                    eng = d.get("engine")
                    if not eng or not eng.ready():
                        raise RuntimeError("检测器 %s 未就绪" % d.get("algorithm_name"))
                    try:
                        res = eng.detect(frame)
                    except Exception as e:
                        logger.warning("pipeline[%s] 检测器 %s 异常: %s"
                                       % (self.stream_code, d.get("algorithm_name"), str(e)))
                        raise
                    algo_id = d.get("algorithm_id")
                    algo_name = d.get("algorithm_name")
                    allowed = set(d.get("target_labels") or [])
                    for r in res:
                        if allowed and r.get("label") not in allowed:
                            continue
                        r["algorithm_id"] = algo_id
                        r["algorithm_name"] = algo_name
                        detections.append(r)
            # 无小模型检测器时不伪造 motion 目标；流程2 仍由 _check_llm_zones 按业务规则报警

        processed_ts = time.time()
        capture_ts = float(capture_ts or self._current_capture_ts or processed_ts)
        active, ended, new_tracks, frame_index = self._tracker.update(
            detections, self._frame_index, timestamp=capture_ts,
        )

        for tid in new_tracks:
            tr = next((t for t in active if t["track_id"] == tid), None)
            if tr:
                self._on_event({
                    "stream_id": self.stream_id,
                    "stream_code": self.stream_code,
                    "type": "object_start",
                    "track_id": tid,
                    "label": tr.get("label", "unknown"),
                    "box": tr.get("box"),
                    "score": tr.get("score", 0),
                    "timestamp": time.time(),
                })

        # 已停用每帧 on_track_snapshot 回调：
        # 原实现会每帧调 touch_tracks 写 DB（last_seen/duration/confidence），
        # 非报警却高频写库，是布控期间 SQLite 写锁与页面卡顿的主要来源。
        # 现仅保留 object_start/object_end 事件 + 业务报警事件写库，
        # 满足"仅业务算法报警才写入"的要求。
        # self._on_track_snapshot(self.stream_id, frame_index, active, has_motion)

        for tid in ended:
            self._on_event({
                "stream_id": self.stream_id,
                "stream_code": self.stream_code,
                "type": "object_end",
                "track_id": tid,
                "timestamp": time.time(),
            })

        # 未确认的中等置信度轨迹不会产生区域报警；第二次连续命中后才参与。
        alarm_tracks = [tr for tr in active if tr.get("observed") and tr.get("confirmed")]
        self._check_zones(alarm_tracks, frame_index, frame)
        self._publish_preview(capture_ts, processed_ts)

        if has_motion:
            self._check_llm_zones(frame, motion_boxes)

    def _emit_biz_alarm(self, event_type, frame, tr, zone_cfg, biz_rule, track_id, zone_id, box, now, **extra):
        """仅业务算法规则命中时：截图、生成描述、上报报警事件。"""
        if not biz_rule:
            return False
        if not self._alarm_enabled:
            return False
        snap_path = self._save_alarm_snapshot(frame, tr, zone_cfg, event_type)
        evt = {
            "stream_id": self.stream_id,
            "stream_code": self.stream_code,
            "type": event_type,
            "track_id": track_id,
            "zone_id": zone_id,
            "label": (tr or {}).get("label", ""),
            "timestamp": now,
            "box": box,
            "score": float((tr or {}).get("score", 0) or 0),
            "frame_width": int(self._last_w or 0),
            "frame_height": int(self._last_h or 0),
            "snapshot_path": snap_path,
        }
        evt.update(extra)
        self._attach_alarm_context(evt, event_type, tr or {}, zone_cfg, biz_rule)
        self._on_event(evt)
        return True

    def _matched_area_rules(self, tr, zone_cfg):
        """旧名兼容：仅返回 AREA 后处理的匹配规则"""
        from app.analysis.biz_rules import matched_area_rules
        rules = (zone_cfg or {}).get("biz_algorithms") or []
        if not rules:
            return []
        return matched_area_rules(tr, zone_cfg)

    def _matched_rules(self, tr, zone_cfg):
        """统一调度：返回与当前目标匹配的所有后处理规则（AREA/LINE_CROSS/DIRECTION/DENSITY/DWELL）"""
        from app.analysis.biz_rules import matched_rules_for_track
        rules = (zone_cfg or {}).get("biz_algorithms") or []
        if not rules:
            return []
        return matched_rules_for_track(tr, zone_cfg)

    def _should_alarm_track_in_zone(self, tr, zone_cfg):
        """该目标在区域内是否需要被关注（命中任意后处理规则）"""
        rules = (zone_cfg or {}).get("biz_algorithms") or []
        if not rules:
            return False
        return len(self._matched_rules(tr, zone_cfg)) > 0

    def _attach_alarm_context(self, event, event_type, tr, zone_cfg, biz_rule=None):
        from app.analysis.biz_rules import build_alarm_context
        ctx = build_alarm_context(event_type, tr, zone_cfg, biz_rule)
        event.update(ctx)
        if ctx.get("alarm_reason"):
            event["description"] = ctx["alarm_reason"]
        return event

    def _llm_verify_track(self, frame, tr, biz_rule):
        """流程3：对大模型做二次校验"""
        llm = biz_rule.get("llm") or {}
        flow = int(biz_rule.get("flow_type") or 0)
        if flow == 3 and (not llm or not biz_rule.get("llm_prompt")):
            logger.warning("pipeline[%s] 流程3 缺少 LLM 配置，跳过报警" % self.stream_code)
            return False
        if not llm or not biz_rule.get("llm_prompt"):
            return True
        tid = tr.get("track_id")
        key = (tid, biz_rule.get("id"))
        now = time.time()
        if now - self._llm_track_last_ts.get(key, 0) < 6.0:
            return False
        if not _CV2_AVAILABLE or frame is None:
            return False
        box = tr.get("box") or []
        if len(box) < 4:
            return False
        try:
            h, w = frame.shape[:2]
            x1, y1, x2, y2 = [int(v) for v in box]
            x1, y1 = max(0, x1), max(0, y1)
            x2, y2 = min(w, x2), min(h, y2)
            if x2 <= x1 or y2 <= y1:
                return False
            crop = frame[y1:y2, x1:x2]
            ok, buf = cv2.imencode('.jpg', crop)
            if not ok:
                return False
            from app.utils.LLMUtils import LLMUtils
            utils = LLMUtils(
                llm.get("api_url"), llm.get("api_key"), llm.get("timeout", 30),
                llm.get("inference_tool", "OpenAI"), llm.get("model_name"),
            )
            result = utils.infer(biz_rule.get("llm_prompt", ""), buf.tobytes())
            passed = LLMUtils.check_happen(result, biz_rule.get("llm_validate", ""))
            # 无论 LLM 判定是否通过，都更新冷却时间，避免每帧反复调用 LLM API
            self._llm_track_last_ts[key] = now
            return passed
        except Exception as e:
            logger.warning("pipeline[%s] LLM 校验失败: %s" % (self.stream_code, str(e)))
            return False

    def _flow3_needs_llm(self, zone_cfg, tr, post_process=None):
        """查找匹配当前目标的 flow3 业务算法（小模型+大模型）。
        post_process 指定时只匹配该后处理类型，否则匹配所有类型。
        """
        for r in (zone_cfg or {}).get("biz_algorithms") or []:
            if int(r.get("flow_type") or 0) != 3:
                continue
            if post_process and r.get("post_process") != post_process:
                continue
            from app.analysis.biz_rules import matched_rules_for_track
            # 复用统一匹配逻辑判断该规则是否命中当前目标
            matched = matched_rules_for_track(tr, {"biz_algorithms": [r]})
            if matched:
                return r
        return None

    def _check_llm_zones(self, frame, motion_boxes):
        """流程2：纯大模型 + 区域运动触发"""
        if not _CV2_AVAILABLE or frame is None or not motion_boxes:
            return
        from app.analysis.biz_rules import llm_rules_for_zone
        from app.utils.LLMUtils import LLMUtils
        h, w = frame.shape[:2]
        now = time.time()
        for z in self.zone_polygons:
            for rule in llm_rules_for_zone(z):
                if int(rule.get("flow_type") or 0) != 2:
                    continue
                key = (z.get("id"), rule.get("id"))
                if now - self._llm_zone_last_ts.get(key, 0) < 8.0:
                    continue
                coords = self._scale_zone(z.get("coords", []), w, h)
                motion_in = False
                hit_box = None
                for mb in motion_boxes:
                    box = mb.get("box") or []
                    if len(box) >= 4:
                        cx = (box[0] + box[2]) / 2
                        cy = (box[1] + box[3]) / 2
                        if self._point_in_polygon((cx, cy), coords):
                            motion_in = True
                            hit_box = box
                            break
                if not motion_in:
                    continue
                llm = rule.get("llm") or {}
                try:
                    ok, buf = cv2.imencode('.jpg', frame)
                    if not ok:
                        continue
                    utils = LLMUtils(
                        llm.get("api_url"), llm.get("api_key"), llm.get("timeout", 30),
                        llm.get("inference_tool", "OpenAI"), llm.get("model_name"),
                    )
                    result = utils.infer(rule.get("llm_prompt", ""), buf.tobytes())
                    if not LLMUtils.check_happen(result, rule.get("llm_validate", "")):
                        # LLM 判定无事件，仍更新冷却时间避免每帧反复调用 API
                        self._llm_zone_last_ts[key] = now
                        continue
                    self._llm_zone_last_ts[key] = now
                    tr_stub = {"box": hit_box, "label": "llm", "track_id": 0}
                    self._emit_biz_alarm(
                        "entered_zone", frame, tr_stub, z, rule,
                        track_id=0, zone_id=z.get("id"), box=hit_box, now=now,
                    )
                except Exception as e:
                    logger.warning("pipeline[%s] LLM 区域分析失败: %s" % (self.stream_code, str(e)))

    def _process_line_rules(self, tr, zone_cfg, zid, tid, box, prev_center, cur_center, w, h, frame, now):
        """LINE_CROSS / LINE_COUNT 后处理（需 zone 已配置 line_a/line_b）。"""
        if not prev_center:
            return
        matched_rules = self._matched_rules(tr, zone_cfg)
        if not matched_rules:
            return
        line_a = (zone_cfg or {}).get("line_a")
        line_b = (zone_cfg or {}).get("line_b")
        if not line_a or not line_b:
            return
        ax = float(line_a[0]) * w
        ay = float(line_a[1]) * h
        bx = float(line_b[0]) * w
        by = float(line_b[1]) * h

        line_rules = [r for r in matched_rules if r.get("post_process") == "LINE_CROSS"]
        for lr in line_rules:
            key = (tid, zid, lr.get("id"))
            if key in self._line_cross_fired:
                continue
            if self._cross_line(prev_center, cur_center, (ax, ay), (bx, by)):
                flow3_rule = self._flow3_needs_llm(zone_cfg, tr, post_process="LINE_CROSS") if lr.get("flow_type") == 3 else None
                if flow3_rule and not self._llm_verify_track(frame, tr, flow3_rule):
                    continue
                if self._emit_biz_alarm(
                    "line_cross", frame, tr, zone_cfg, lr,
                    track_id=tid, zone_id=zid, box=box, now=now,
                ):
                    self._line_cross_fired.add(key)

        count_rules = [r for r in matched_rules if r.get("post_process") == "LINE_COUNT"]
        for cr in count_rules:
            direction = self._cross_line_direction(prev_center, cur_center, (ax, ay), (bx, by))
            if not direction:
                continue
            flow3_rule = self._flow3_needs_llm(zone_cfg, tr, post_process="LINE_COUNT") if cr.get("flow_type") == 3 else None
            if flow3_rule and not self._llm_verify_track(frame, tr, flow3_rule):
                continue
            counts = self._get_line_counts(zid, cr.get("id"))
            counts[direction] = int(counts.get(direction) or 0) + 1
            self._maybe_alert_line_count(frame, tr, zone_cfg, cr, direction, counts, tid, zid, box, now)

    def _fire_area_alarms(self, tr, zid, zone_cfg, prev_zones, frame, box, now, tid):
        """区域入侵：首次进入立即报警，持续停留按独立重复间隔重报。"""
        matched_rules = self._matched_rules(tr, zone_cfg)
        if not matched_rules:
            return False
        repeat_sec = max(0.0, float((zone_cfg or {}).get("alarm_repeat_sec", 30) or 0))
        is_new = zid not in prev_zones
        fired_any = False
        flow3_rule = self._flow3_needs_llm(zone_cfg, tr)
        area_rules = [r for r in matched_rules if r.get("post_process") == "AREA"]
        dwell_rules = [r for r in matched_rules if r.get("post_process") == "DWELL"]

        def _should_fire(biz_id):
            key = (tid, zid, biz_id)
            last = self._area_alarm_last_ts.get(key, 0)
            return is_new or (repeat_sec > 0 and now - last >= repeat_sec)

        def _mark_fired(biz_id):
            self._area_alarm_last_ts[(tid, zid, biz_id)] = now

        if flow3_rule:
            biz_id = flow3_rule.get("id")
            if _should_fire(biz_id):
                if self._llm_verify_track(frame, tr, flow3_rule):
                    if self._emit_biz_alarm(
                        "entered_zone", frame, tr, zone_cfg, flow3_rule,
                        track_id=tid, zone_id=zid, box=box, now=now,
                    ):
                        _mark_fired(biz_id)
                        fired_any = True
        else:
            for r in area_rules:
                biz_id = r.get("id")
                if not _should_fire(biz_id):
                    continue
                if self._emit_biz_alarm(
                    "entered_zone", frame, tr, zone_cfg, r,
                    track_id=tid, zone_id=zid, box=box, now=now,
                ):
                    _mark_fired(biz_id)
                    fired_any = True
            if is_new:
                for r in dwell_rules:
                    if self._emit_biz_alarm(
                        "entered_zone", frame, tr, zone_cfg, r,
                        track_id=tid, zone_id=zid, box=box, now=now,
                    ):
                        _mark_fired(r.get("id"))
                        fired_any = True

        if fired_any or matched_rules:
            if (tid, zid) not in self._track_enter_ts:
                self._track_enter_ts[(tid, zid)] = now
        return fired_any or bool(matched_rules)

    def _check_zones(self, active, frame_index, frame):
        now = time.time()
        h, w = self._frame_size(frame_index)
        # 短暂缺检的轨迹保留原区域状态，但本帧不触发任何报警；只有重新观测到
        # 明确在区域外两次，或轨迹超时结束，才判定离开。
        retained_ids = {t.track_id for t in self._tracker.all_active()}
        cur_state = {tid: set(zones) for tid, zones in self._last_zone_state.items()
                     if tid in retained_ids}
        # 统计每个区域内的目标数（用于 DENSITY 后处理）
        zone_density_count = {}  # zone_id -> count

        for tr in active:
            tid = tr["track_id"]
            box = tr["box"]
            cx = (box[0] + box[2]) / 2
            # 人员用人体框底部向上内缩 5% 的落脚点判断入区，避免框中心落在区域外。
            cy = box[3] - (box[3] - box[1]) * 0.05 if tr.get("label") == "person" else (box[1] + box[3]) / 2
            physical_in = set()
            for z in self.zone_polygons:
                coords = self._scale_zone(z.get("coords", []), w, h)
                if self._point_in_polygon((cx, cy), coords):
                    physical_in.add(z.get("id"))
                    zone_density_count[z.get("id")] = zone_density_count.get(z.get("id"), 0) + 1
            prev = self._last_zone_state.get(tid, set())
            confirmed = set(prev & physical_in)

            for zid in physical_in:
                self._outside_zone_hits.pop((tid, zid), None)
                zone_cfg = next((z for z in self.zone_polygons if z.get("id") == zid), None)
                if not self._should_alarm_track_in_zone(tr, zone_cfg):
                    logger.debug(
                        "pipeline[%s] track %s in zone %s but no biz rule matched (label=%s algo=%s)",
                        self.stream_code, tid, zid, tr.get("label"), tr.get("algorithm_id"),
                    )
                    confirmed.add(zid)
                    continue
                self._fire_area_alarms(tr, zid, zone_cfg, prev, frame, box, now, tid)
                confirmed.add(zid)

            for zid in prev - physical_in:
                outside_key = (tid, zid)
                outside_hits = self._outside_zone_hits.get(outside_key, 0) + 1
                self._outside_zone_hits[outside_key] = outside_hits
                if outside_hits < 2:
                    confirmed.add(zid)
                    continue
                self._outside_zone_hits.pop(outside_key, None)
                self._on_event({
                    "stream_id": self.stream_id, "stream_code": self.stream_code,
                    "type": "left_zone", "track_id": tid, "zone_id": zid,
                    "label": tr["label"], "timestamp": now,
                })
                self._track_enter_ts.pop((tid, zid), None)
                # 清理该目标在该区域所有算法的滞留标记
                for lk in list(self._loiter_fired):
                    if lk[0] == tid and lk[1] == zid:
                        self._loiter_fired.discard(lk)
                # 清理该目标在该区域的所有越线标记
                for key in list(self._line_cross_fired):
                    if key[0] == tid and key[1] == zid:
                        self._line_cross_fired.discard(key)
                # 清理该目标在该区域的所有方向报警标记
                for key in list(self._direction_fired):
                    if key[0] == tid and key[1] == zid:
                        self._direction_fired.discard(key)
                for key in list(self._area_alarm_last_ts.keys()):
                    if key[0] == tid and key[1] == zid:
                        self._area_alarm_last_ts.pop(key, None)

            for zid in confirmed:
                ts = self._track_enter_ts.get((tid, zid))
                if not ts:
                    continue
                zone_cfg = next((z for z in self.zone_polygons if z.get("id") == zid), None)
                matched_rules = self._matched_rules(tr, zone_cfg)
                if not matched_rules:
                    continue

                # —— DWELL / AREA 滞留报警：每个匹配的 AREA/DWELL 算法独立触发 ——
                threshold = (zone_cfg or {}).get("loiter_threshold", 0)
                if threshold and (now - ts) >= threshold:
                    dwell_rules = [r for r in matched_rules if r.get("post_process") in ("AREA", "DWELL")]
                    flow3_rule = self._flow3_needs_llm(zone_cfg, tr)
                    if flow3_rule and not self._llm_verify_track(frame, tr, flow3_rule):
                        rules_to_fire = []  # flow3 校验失败，跳过滞留
                    else:
                        rules_to_fire = [flow3_rule] if flow3_rule else dwell_rules
                    for matched in rules_to_fire:
                        biz_id = matched.get("id")
                        if (tid, zid, biz_id) in self._loiter_fired:
                            continue  # 该算法本次停留已报过，不重复
                        tr_loiter = dict(tr)
                        tr_loiter["duration"] = now - ts
                        evt_type = "dwell" if matched.get("post_process") == "DWELL" else "loiter"
                        if self._emit_biz_alarm(
                            evt_type, frame, tr_loiter, zone_cfg, matched,
                            track_id=tid, zone_id=zid, box=box, now=now,
                            duration=now - ts,
                        ):
                            self._loiter_fired.add((tid, zid, biz_id))

                # —— LINE_CROSS / LINE_COUNT（必需区域内）——
                prev_center = self._track_centers.get(tid)
                cur_center = (cx, cy)
                self._process_line_rules(
                    tr, zone_cfg, zid, tid, box, prev_center, cur_center, w, h, frame, now)

                # —— DIRECTION 方向入侵 ——
                dir_rules = [r for r in matched_rules if r.get("post_process") == "DIRECTION"]
                for dr in dir_rules:
                    if prev_center:
                        dx = cur_center[0] - prev_center[0]
                        dy = cur_center[1] - prev_center[1]
                        ref_angle = float(dr.get("ref_angle", 90.0))
                        tol = float(dr.get("angle_tolerance", 45.0))
                        if self._direction_match(dx, dy, ref_angle, tol):
                            flow3_rule = self._flow3_needs_llm(zone_cfg, tr, post_process="DIRECTION") if dr.get("flow_type") == 3 else None
                            if flow3_rule and not self._llm_verify_track(frame, tr, flow3_rule):
                                continue
                            # 方向匹配：去重节流（同目标同区域同算法 60 秒内只报一次）
                            dir_key = (tid, zid, dr.get("id"))
                            if dir_key in self._direction_fired:
                                continue
                            if self._emit_biz_alarm(
                                "direction", frame, tr, zone_cfg, dr,
                                track_id=tid, zone_id=zid, box=box, now=now,
                            ):
                                self._direction_fired.add(dir_key)

                # 记录中心点供下一帧使用（移到循环外，避免多区域时覆盖 prev_center）
                # 见下方统一更新

            # —— 非必需区域：仅按方向线做越线检测/计数（不要求目标在多边形内）——
            prev_center = self._track_centers.get(tid)
            cur_center = (cx, cy)
            for z in self.zone_polygons:
                if z.get("is_required"):
                    continue
                zid = z.get("id")
                self._process_line_rules(
                    tr, z, zid, tid, box, prev_center, cur_center, w, h, frame, now)

            cur_state[tid] = confirmed
            # 统一更新 track_centers：每帧每目标只更新一次，避免多区域循环内覆盖
            self._track_centers[tid] = (cx, cy)

        # —— DENSITY 密度报警（区域级，每帧检查）：每个匹配的密度算法独立触发 ——
        for z in self.zone_polygons:
            zid = z.get("id")
            density_rules = [r for r in (z.get("biz_algorithms") or [])
                             if r.get("post_process") == "DENSITY" and int(r.get("flow_type") or 0) in (1, 3, 4)]
            if not density_rules:
                continue
            count = zone_density_count.get(zid, 0)
            threshold = int(z.get("density_threshold", 0))
            if not threshold or count < threshold:
                # 密度降回，清理该区域所有算法的已报警标记
                for dk in list(self._density_fired):
                    if dk[0] == zid:
                        self._density_fired.discard(dk)
                continue
            # 每个匹配的密度算法独立触发报警
            for dr in density_rules:
                biz_id = dr.get("id")
                if (zid, biz_id) in self._density_fired:
                    continue  # 该算法本次密度已报过，不重复
                tr_stub = {"density_count": count, "label": "—", "track_id": 0}
                if self._emit_biz_alarm(
                    "density", frame, tr_stub, z, dr,
                    track_id=0, zone_id=zid, box=None, now=now,
                    density_count=count,
                ):
                    self._density_fired.add((zid, biz_id))

        for tid in list(self._last_zone_state.keys()):
            if tid not in cur_state:
                self._last_zone_state.pop(tid, None)
                self._track_centers.pop(tid, None)
                for key in list(self._track_enter_ts.keys()):
                    if key[0] == tid:
                        self._track_enter_ts.pop(key, None)
                # 清理已结束目标的报警去重标记，避免内存泄漏
                for fk in list(self._loiter_fired):
                    if fk[0] == tid:
                        self._loiter_fired.discard(fk)
                for fk in list(self._line_cross_fired):
                    if fk[0] == tid:
                        self._line_cross_fired.discard(fk)
                for fk in list(self._direction_fired):
                    if fk[0] == tid:
                        self._direction_fired.discard(fk)
        self._last_zone_state = cur_state

    def _preview_inside_zones(self, tr):
        h, w = self._last_h, self._last_w
        box = tr.get("box") or []
        if len(box) != 4 or not w or not h:
            return []
        x = (box[0] + box[2]) * 0.5
        y = box[3] - (box[3] - box[1]) * 0.05 if tr.get("label") == "person" else (box[1] + box[3]) * 0.5
        return [z.get("id") for z in self.zone_polygons
                if self._point_in_polygon((x, y), self._scale_zone(z.get("coords", []), w, h))]

    def _publish_preview(self, capture_ts, processed_ts=None):
        """只发布最新检测状态；事件桥覆盖旧快照，不落库、不累积帧。"""
        if not self._last_w or not self._last_h:
            return
        tracks = []
        processed_ts = float(processed_ts or time.time())
        for tr in self._tracker.display_tracks(capture_ts):
            box = tr.get("box") or []
            if len(box) != 4:
                continue
            item = dict(tr)
            item["box"] = [max(0.0, min(1.0, box[0] / self._last_w)),
                           max(0.0, min(1.0, box[1] / self._last_h)),
                           max(0.0, min(1.0, box[2] / self._last_w)),
                           max(0.0, min(1.0, box[3] / self._last_h))]
            velocity = item.get("velocity") or [0, 0, 0, 0]
            item["velocity"] = [float(velocity[0]) / self._last_w,
                                float(velocity[1]) / self._last_h,
                                float(velocity[2]) / self._last_w,
                                float(velocity[3]) / self._last_h]
            item["inside_zone_ids"] = self._preview_inside_zones(tr)
            tracks.append(item)
        self._on_preview({
            "stream_id": self.stream_id,
            "sequence": self._frame_index,
            "timestamp": capture_ts,
            "processed_timestamp": processed_ts,
            "inference_latency": max(0.0, processed_ts - capture_ts),
            "frame_width": self._last_w,
            "frame_height": self._last_h,
            "analysis_fps": self._analysis_fps,
            "model_names": [d.get("algorithm_name", "") for d in self._detectors],
            "model_devices": [d.get("device", "cpu") for d in self._detectors],
            "tracks": tracks,
            "zones": [{"id": z.get("id"), "name": z.get("name", ""),
                       "color": z.get("color", "#16a085"), "coords": z.get("coords") or []}
                      for z in self.zone_polygons],
        })

    def _cross_line(self, prev_pt, cur_pt, line_a, line_b):
        from app.analysis.biz_rules import cross_line_segment
        return cross_line_segment(prev_pt, cur_pt, line_a, line_b)

    def _cross_line_direction(self, prev_pt, cur_pt, line_a, line_b):
        from app.analysis.biz_rules import cross_line_direction
        return cross_line_direction(prev_pt, cur_pt, line_a, line_b)

    def _line_count_key(self, zone_id, biz_id):
        return (int(zone_id or 0), int(biz_id or 0))

    def _get_line_counts(self, zone_id, biz_id):
        key = self._line_count_key(zone_id, biz_id)
        st = self._line_count_state.get(key)
        if not st:
            st = {"forward": 0, "reverse": 0}
            self._line_count_state[key] = st
        return st

    def _maybe_alert_line_count(self, frame, tr, zone_cfg, lr, direction, counts, track_id, zone_id, box, now):
        biz_id = lr.get("id")
        threshold = int(lr.get("forward_count_threshold" if direction == "forward" else "reverse_count_threshold") or 0)
        if threshold <= 0:
            return
        current = int(counts.get(direction) or 0)
        if current < threshold:
            return
        alert_key = (zone_id, biz_id, direction)
        last_ts = self._line_count_alert_ts.get(alert_key, 0)
        if now - last_ts < 30.0:
            return
        tr_count = dict(tr)
        tr_count["line_count_direction"] = direction
        tr_count["forward_count"] = counts.get("forward", 0)
        tr_count["reverse_count"] = counts.get("reverse", 0)
        if self._emit_biz_alarm(
            "line_count", frame, tr_count, zone_cfg, lr,
            track_id=track_id, zone_id=zone_id, box=box, now=now,
            forward_count=counts.get("forward", 0),
            reverse_count=counts.get("reverse", 0),
            line_count_direction=direction,
        ):
            self._line_count_alert_ts[alert_key] = now

    def _direction_match(self, dx, dy, ref_angle_deg, tolerance_deg):
        from app.analysis.biz_rules import direction_match
        return direction_match(dx, dy, ref_angle_deg, tolerance_deg)

    def _alarm_snapshot_dir(self):
        """返回当日报警快照目录绝对路径：{storageAlarmDir}/{stream_code}/{YYYYMMDD}/"""
        from datetime import datetime as _dt
        base = self._storage_alarm_dir or os.path.join(os.getcwd(), "static", "storage", "alarm")
        day = _dt.now().strftime("%Y%m%d")
        code = "".join(c for c in str(self.stream_code or "unknown") if c.isalnum() or c in "_-") or "unknown"
        d = os.path.join(base, code, day)
        try:
            os.makedirs(d, exist_ok=True)
        except Exception as e:
            logger.warning("pipeline[%s] 创建快照目录失败 %s: %s" % (self.stream_code, d, e))
        return d

    def _alarm_rel_path(self, fname):
        """根据绝对路径计算浏览器可访问的相对路径（相对 static/ 根目录）"""
        import datetime as _dt2
        code = "".join(c for c in str(self.stream_code or "unknown") if c.isalnum() or c in "_-") or "unknown"
        day = _dt2.datetime.now().strftime("%Y%m%d")
        static_root = self._static_dir or os.path.join(os.getcwd(), "static")
        abs_path = os.path.join(self._alarm_snapshot_dir(), fname)
        try:
            rel = os.path.relpath(abs_path, static_root).replace("\\", "/")
            if rel.startswith(".."):
                raise ValueError("snapshot outside static root")
            return rel
        except Exception:
            return "storage/alarm/%s/%s/%s" % (code, day, fname)

    def _save_alarm_snapshot(self, frame, track, zone_cfg, event_type):
        """绘制检测框 + 区域多边形 + 标签后保存 JPEG，返回相对 static/ 的路径；失败返回 ''"""
        if not _CV2_AVAILABLE or frame is None:
            if not _CV2_AVAILABLE:
                logger.warning("pipeline[%s] OpenCV 不可用，无法保存报警快照" % self.stream_code)
            return ""
        now = time.time()
        # 节流：60 秒窗口内超过上限则跳过
        while self._snapshot_times and now - self._snapshot_times[0] > 60.0:
            self._snapshot_times.popleft()
        if len(self._snapshot_times) >= _SNAPSHOT_MAX_PER_MIN:
            return ""
        try:
            import numpy as np
            img = frame.copy()
            h, w = img.shape[:2]
            # 画区域多边形（半透明填充 + 描边）
            if zone_cfg:
                coords = self._scale_zone(zone_cfg.get("coords", []), w, h)
                if coords and len(coords) >= 3:
                    pts = np.array([(int(p[0]), int(p[1])) for p in coords], dtype=np.int32)
                    overlay = img.copy()
                    cv2.fillPoly(overlay, [pts], (22, 159, 133))
                    cv2.addWeighted(overlay, 0.18, img, 0.82, 0, img)
                    cv2.polylines(img, [pts], True, (22, 159, 133), 2)
            # 画检测框 + 标签
            box = track.get("box") or [0, 0, 0, 0]
            x1, y1, x2, y2 = int(box[0]), int(box[1]), int(box[2]), int(box[3])
            cv2.rectangle(img, (x1, y1), (x2, y2), (220, 38, 38), 2)
            label = str(track.get("label") or "")
            score = track.get("score")
            txt = label + (" %.2f" % score if isinstance(score, (int, float)) else "")
            (tw, th), _ = cv2.getTextSize(txt, cv2.FONT_HERSHEY_SIMPLEX, 0.5, 1)
            ty = max(0, y1 - 6)
            cv2.rectangle(img, (x1, max(0, ty - th - 4)), (x1 + tw + 6, ty + 2), (220, 38, 38), -1)
            cv2.putText(img, txt, (x1 + 3, ty - 2), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 255), 1, cv2.LINE_AA)
            # 事件类型角标
            tag = event_type == "loiter" and "LOITER" or "ALARM"
            cv2.rectangle(img, (w - 90, 6), (w - 6, 26), (220, 38, 38), -1)
            cv2.putText(img, tag, (w - 84, 21), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 255), 1, cv2.LINE_AA)
            fname = "%s_%s_%d_%s.jpg" % (self.stream_id, event_type, int(now * 1000), track.get("track_id", "x"))
            fpath = os.path.join(self._alarm_snapshot_dir(), fname)
            cv2.imwrite(fpath, img)
            self._snapshot_times.append(now)
            return self._alarm_rel_path(fname)
        except Exception as e:
            logger.warning("pipeline[%s] 保存快照失败: %s" % (self.stream_code, str(e)))
            return ""

    @staticmethod
    def _point_in_polygon(pt, poly):
        if not poly or len(poly) < 3:
            return False
        x, y = pt
        inside = False
        n = len(poly)
        j = n - 1
        for i in range(n):
            xi, yi = poly[i]
            xj, yj = poly[j]
            if ((yi > y) != (yj > y)) and (x < (xj - xi) * (y - yi) / (yj - yi + 1e-9) + xi):
                inside = not inside
            j = i
        return inside

    def _compute_stream_health(self):
        now = time.time()
        if not self._running:
            return "stopped", 0.0
        src = self._frame_source.health_snapshot() if self._frame_source else {}
        src_health = src.get("stream_health") or "connecting"
        stalled_sec = src.get("stalled_sec") or 0.0
        if self._last_frame_ts > 0:
            stalled_sec = max(stalled_sec, now - self._last_frame_ts)
        if src_health in ("reconnecting", "disconnected", "connecting"):
            if stalled_sec >= 12 or (self._decoded_count == 0 and now - self._start_ts >= 15):
                return src_health if src_health != "connecting" else "reconnecting", stalled_sec
            return src_health, stalled_sec
        if stalled_sec >= 15 and self._analysis_fps < 0.05:
            return "stalled", stalled_sec
        if self._decoded_count == 0 and now - self._start_ts >= 20:
            return "stalled", stalled_sec
        return "ok", stalled_sec

    # ============ 运行状态 ============
    def status(self):
        health, stalled_sec = self._compute_stream_health()
        src = self._frame_source.health_snapshot() if self._frame_source else {}
        effective_running = self._running and health in ("ok", "reconnecting", "connecting")
        if self._analysis_error:
            analysis_health = "error"
        elif not self._last_analysis_ts:
            analysis_health = "starting"
        elif time.time() - self._last_analysis_ts > max(30.0, self._analyze_interval * 2):
            analysis_health = "stalled"
        else:
            analysis_health = "running"
        return {
            "stream_id": self.stream_id,
            "running": effective_running,
            "stream_health": health,
            "analysis_health": analysis_health,
            "analysis_error": self._analysis_error,
            "stalled_sec": round(stalled_sec, 1),
            "reconnect_fail_count": src.get("reconnect_fail_count", 0),
            "total_reconnects": src.get("total_reconnects", 0),
            "algorithm_name": self._algorithm_name,
            "engine": self._detector.ENGINE_NAME if self._detector else "",
            "detectors": [{"algorithm_id": d.get("algorithm_id"),
                           "algorithm_name": d.get("algorithm_name"),
                           "engine": d["engine"].ENGINE_NAME if d.get("engine") else ""}
                          for d in self._detectors],
            "frame_index": self._frame_index,
            "decoded_count": self._decoded_count,
            "analyzed_count": self._analyzed_count,
            "dropped_count": self._dropped_count,
            "analysis_fps": round(self._analysis_fps, 3),
            "decode_fps": round(self._decode_fps, 1),
            "analyze_fps_target": self.analyze_fps,
            "active_zone_ids": sorted([
                int(z.get("id")) for z in (self.zone_polygons or [])
                if z.get("id") is not None
            ]),
        }

    def stop(self):
        self._running = False
        self._decode_running = False