"""单摄像头分析流水线（实时解码 + 实时分析） 架构： - 解码线程 _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