Spaces:
Running
Running
| """規則引擎:keypoint cache rows → 每 track 特徵管線 → 狀態機 → 跌倒事件。 | |
| 職責: | |
| 1. 逐幀分流:cache 的每列依 track_id 分派到該 track 的特徵管線 + 狀態機; | |
| ``track_id = -1``(未確認偵測)不餵狀態機。 | |
| 2. 特徵管線:平滑(滑動中位數)、以軀幹長 L̃ 正規化、固定「時間窗」差分 | |
| 算垂直速度與角速度(尺度與 fps 不變性都在這一層達成)。 | |
| 3. keypoint dropout 容忍:無效幀 hold-last(TTL = ``max_kpt_gap_s``), | |
| 逾時凍結;連續無有效觀測超過 ``track_lost_timeout_s`` 即終結該 track。 | |
| 4. track 縫合:ByteTrack 在跌倒瞬間常斷 id——新 track 出現時,若與剛消失的 | |
| 舊 track 末 bbox 的 IoU 夠高,直接繼承舊狀態機(不魔改 tracker)。 | |
| 舊 track 消失當下若已在 FALLING/FALLEN,縫合與同 id 重現都改用加長版 | |
| 時間窗(``track_stitch_window_falling_s``):此時已有獨立的速度觸發證據, | |
| 值得多等一下換回真實事件,而非被寫死的一般窗口攔截。 | |
| """ | |
| from __future__ import annotations | |
| import math | |
| import numpy as np | |
| import pandas as pd | |
| from ..config import Config | |
| from ..events.schema import FallEvent, postprocess_events | |
| from .features import compute_frame_geometry | |
| from .smoothing import RollingMedian, TimedBuffer | |
| from .state_machine import FallStateMachine, State, TickInput | |
| # L̃(軀幹長滑動中位數)的時窗:遠長於單次跌倒(~1s), | |
| # 朝鏡頭跌倒造成的軀幹投影縮短不會即刻拉低尺度基準。 | |
| TORSO_MEDIAN_WINDOW_S = 2.0 | |
| def _iou(a: tuple, b: tuple) -> float: | |
| ax1, ay1, ax2, ay2 = a | |
| bx1, by1, bx2, by2 = b | |
| ix1, iy1 = max(ax1, bx1), max(ay1, by1) | |
| ix2, iy2 = min(ax2, bx2), min(ay2, by2) | |
| iw, ih = max(0.0, ix2 - ix1), max(0.0, iy2 - iy1) | |
| inter = iw * ih | |
| if inter <= 0: | |
| return 0.0 | |
| area_a = max(0.0, ax2 - ax1) * max(0.0, ay2 - ay1) | |
| area_b = max(0.0, bx2 - bx1) * max(0.0, by2 - by1) | |
| union = area_a + area_b - inter | |
| return inter / union if union > 0 else 0.0 | |
| class _TrackRunner: | |
| """單一 track 的特徵管線(平滑/正規化/差分)+ 其狀態機。 | |
| 縫合時只繼承狀態機:平滑緩衝重新起算(幾幀內即填滿, | |
| 代價遠小於把兩條 track 的緩衝硬接在一起的複雜度)。 | |
| """ | |
| def __init__(self, cfg: Config, fps: float, fsm: FallStateMachine): | |
| r = cfg.rules | |
| smooth_n = max(1, round(fps * r.smooth_s)) | |
| self.theta_med = RollingMedian(smooth_n) | |
| self.aspect_med = RollingMedian(smooth_n) | |
| self.hhip_med = RollingMedian(smooth_n) | |
| self.hipy_med3 = RollingMedian(3) | |
| self.torso_med = RollingMedian(max(1, round(fps * TORSO_MEDIAN_WINDOW_S))) | |
| horizon = r.velocity_window_s * 1.5 + 0.2 | |
| self.hipy_buf = TimedBuffer(horizon) | |
| self.theta_buf = TimedBuffer(horizon) | |
| self.fsm = fsm | |
| self.fps = fps | |
| self.last_t: float | None = None | |
| self.last_valid_t: float | None = None | |
| self.last_bbox: tuple | None = None | |
| self.held: TickInput | None = None | |
| self.last_ticked_t: float | None = None | |
| def _tick(self, t_s: float, tick: TickInput) -> None: | |
| """呼叫 fsm.tick 前,先把「這次 tick 距上次 tick 的空窗」橋接掉。 | |
| 一般逐幀呼叫時 gap ≈ 一幀時長,橋接量 ≈ 0(無感);只有 tick 之間 | |
| 真的隔了一段沒有任何觀測的空窗(見 ``FallStateMachine.bridge_gap`` | |
| docstring)才會產生有意義的橋接量。 | |
| """ | |
| if self.last_ticked_t is not None: | |
| gap = (t_s - self.last_ticked_t) - (1.0 / self.fps) | |
| if gap > 1e-9: | |
| self.fsm.bridge_gap(gap) | |
| self.fsm.tick(tick) | |
| self.last_ticked_t = t_s | |
| def step( | |
| self, | |
| cfg: Config, | |
| track_id: int, | |
| frame_idx: int, | |
| t_s: float, | |
| bbox: tuple, | |
| kpts_xy: np.ndarray, | |
| kpts_conf: np.ndarray, | |
| debug: list | None, | |
| ) -> None: | |
| r = cfg.rules | |
| geo = compute_frame_geometry(kpts_xy, kpts_conf, np.asarray(bbox), cfg.model.kpt_conf_min) | |
| if geo.valid: | |
| l_tilde = self.torso_med.push(geo.torso_len) | |
| theta_s = self.theta_med.push(geo.theta_deg) | |
| aspect_s = ( | |
| self.aspect_med.push(geo.bbox_aspect) | |
| if not math.isnan(geo.bbox_aspect) | |
| else float("nan") | |
| ) | |
| h_hip = None | |
| if geo.ankle_valid and l_tilde > 0: | |
| h_hip = self.hhip_med.push(geo.hip_ankle_gap / l_tilde) | |
| hip_y_f = self.hipy_med3.push(geo.hip_y) | |
| # 差分先取歷史再 push 當前值(否則 Δ 內含自身) | |
| v_raw = self.hipy_buf.rate(t_s, hip_y_f, r.velocity_window_s) | |
| self.hipy_buf.push(t_s, hip_y_f) | |
| omega = self.theta_buf.rate(t_s, theta_s, r.velocity_window_s) | |
| self.theta_buf.push(t_s, theta_s) | |
| v_norm = (v_raw / l_tilde) if (v_raw is not None and l_tilde > 0) else None | |
| tick = TickInput( | |
| t_s=t_s, | |
| frame_idx=frame_idx, | |
| theta_deg=theta_s, | |
| bbox_aspect=aspect_s, | |
| h_hip=h_hip, | |
| v_norm=v_norm, | |
| omega=omega, | |
| ) | |
| self.held = tick | |
| self.last_valid_t = t_s | |
| self._tick(t_s, tick) | |
| elif ( | |
| self.held is not None | |
| and self.last_valid_t is not None | |
| and (t_s - self.last_valid_t) <= r.max_kpt_gap_s | |
| ): | |
| # hold-last:姿態沿用上次有效值,速度/角速度歸零(缺測不得觸發新事件, | |
| # 但已在 FALLEN/ALARM 的狀態得以維持,dropout 不會把一次跌倒切成兩段) | |
| tick = TickInput( | |
| t_s=t_s, | |
| frame_idx=frame_idx, | |
| theta_deg=self.held.theta_deg, | |
| bbox_aspect=self.held.bbox_aspect, | |
| h_hip=self.held.h_hip, | |
| v_norm=0.0, | |
| omega=0.0, | |
| ) | |
| self._tick(t_s, tick) | |
| # else:逾 TTL → 凍結(不 tick);終結與否由引擎依 track_lost_timeout_s 決定 | |
| self.last_t = t_s | |
| self.last_bbox = tuple(float(v) for v in bbox) | |
| if debug is not None: | |
| debug.append( | |
| { | |
| "frame_idx": int(frame_idx), | |
| "track_id": int(track_id), | |
| "t_s": round(t_s, 4), | |
| "valid": bool(geo.valid), | |
| "theta_deg": round(self.held.theta_deg, 2) if self.held else None, | |
| "bbox_aspect": ( | |
| round(self.held.bbox_aspect, 3) | |
| if self.held and not math.isnan(self.held.bbox_aspect) | |
| else None | |
| ), | |
| "h_hip": ( | |
| round(self.held.h_hip, 3) | |
| if self.held and self.held.h_hip is not None | |
| else None | |
| ), | |
| "v_norm": ( | |
| round(self.held.v_norm, 3) | |
| if geo.valid and self.held and self.held.v_norm is not None | |
| else None | |
| ), | |
| "state": self.fsm.state.value, | |
| } | |
| ) | |
| def _lost_window_for(ru: _TrackRunner, r, base: float) -> float: | |
| """該 runner 消失後可容忍的等待時間。 | |
| 一般情況用呼叫端各自的基準值(``base``);若消失當下已在 FALLING/FALLEN, | |
| 一律換成加長版縫合窗(``track_stitch_window_falling_s``)——見 engine 模組 | |
| docstring 第 4 點。 | |
| """ | |
| if ru.fsm.state in (State.FALLING, State.FALLEN): | |
| return r.track_stitch_window_falling_s | |
| return base | |
| def _pop_stitch_source( | |
| runners: dict[int, _TrackRunner], bbox: tuple, t_s: float, cfg: Config | |
| ) -> _TrackRunner | None: | |
| """在「最近消失」的 runner 中找縫合對象;找到即自 runners 移除並回傳。 | |
| 只考慮本幀沒被更新的 runner(仍活躍的 track 不是縫合對象)。 | |
| """ | |
| r = cfg.rules | |
| best_tid, best_iou = None, 0.0 | |
| for tid, ru in runners.items(): | |
| if ru.last_t is None or ru.last_bbox is None: | |
| continue | |
| if ru.last_t >= t_s - 1e-9: # 本幀已更新:仍活著 | |
| continue | |
| if t_s - ru.last_t > _lost_window_for(ru, r, r.track_stitch_window_s): | |
| continue | |
| iou = _iou(ru.last_bbox, bbox) | |
| if iou >= r.track_stitch_iou and iou > best_iou: | |
| best_tid, best_iou = tid, iou | |
| if best_tid is None: | |
| return None | |
| return runners.pop(best_tid) | |
| def run_engine( | |
| df: pd.DataFrame, fps: float, cfg: Config, collect_debug: bool = False | |
| ) -> tuple[list[FallEvent], list[dict]]: | |
| """對一支影片的 cache rows 執行規則引擎。 | |
| Args: | |
| df: keypoint cache rows(欄位見 ``io.cache.CACHE_COLUMNS``)。 | |
| fps: 影片幀率(取自 cache meta;所有時間計算的基準)。 | |
| cfg: 完整設定。 | |
| collect_debug: 是否回傳 per-frame per-track 特徵紀錄(失敗分析用)。 | |
| Returns: | |
| (後處理完成的事件列表, debug 紀錄列表) | |
| """ | |
| debug: list[dict] | None = [] if collect_debug else None | |
| if df.empty or fps <= 0: | |
| return [], (debug or []) | |
| r = cfg.rules | |
| runners: dict[int, _TrackRunner] = {} | |
| events_raw: list[FallEvent] = [] | |
| df = df.sort_values(["frame_idx", "track_id"], kind="stable") | |
| for _, grp in df.groupby("frame_idx", sort=True): | |
| rows = list(grp.itertuples(index=False)) | |
| t_frame = float(rows[0].t_ms) / 1000.0 | |
| # 先更新既有 track,再處理新 track:避免把「本幀仍活著的 track」誤當縫合對象 | |
| existing = [row for row in rows if int(row.track_id) in runners] | |
| newcomers = [ | |
| row for row in rows if int(row.track_id) >= 0 and int(row.track_id) not in runners | |
| ] | |
| for row in existing: | |
| tid = int(row.track_id) | |
| t_s = float(row.t_ms) / 1000.0 | |
| runner = runners[tid] | |
| if runner.last_valid_t is not None and ( | |
| t_s - runner.last_valid_t | |
| ) > _lost_window_for(runner, r, r.track_lost_timeout_s): | |
| # 同一 id 長時間無有效觀測後重現:舊片段終結,重新起算 | |
| # (FALLING/FALLEN 時用加長版時間窗,理由同縫合) | |
| events_raw.extend(runner.fsm.finalize()) | |
| runner = _TrackRunner(cfg, fps, FallStateMachine(r, tid)) | |
| runners[tid] = runner | |
| runner.step(cfg, tid, int(row.frame_idx), t_s, _bbox(row), row.kpts_xy, row.kpts_conf, debug) | |
| for row in newcomers: | |
| tid = int(row.track_id) | |
| t_s = float(row.t_ms) / 1000.0 | |
| bbox = _bbox(row) | |
| src = _pop_stitch_source(runners, bbox, t_s, cfg) | |
| if src is not None: | |
| src.fsm.adopt(tid) | |
| runner = _TrackRunner(cfg, fps, src.fsm) | |
| # 縫合時繼承尺度基準與 hold-last 狀態: | |
| # L̃ 若在跌倒瞬間重新定錨到「投影縮短的軀幹」,正規化就失真 | |
| runner.torso_med = src.torso_med | |
| runner.held = src.held | |
| runner.last_valid_t = src.last_valid_t | |
| # 也繼承上次 tick 時刻,讓縫合斷點的空窗被 _tick 正確橋接 | |
| # (否則消失期間的時間差會被誤算成「觀察了這麼久還沒確認」) | |
| runner.last_ticked_t = src.last_ticked_t | |
| else: | |
| runner = _TrackRunner(cfg, fps, FallStateMachine(r, tid)) | |
| runners[tid] = runner | |
| runner.step(cfg, tid, int(row.frame_idx), t_s, bbox, row.kpts_xy, row.kpts_conf, debug) | |
| # 清掃:超過各自可容忍等待時間的 runner 終結。FALLING/FALLEN 用加長版窗, | |
| # 否則會在縫合視窗生效前就被這裡提前終結(見 _lost_window_for)。 | |
| base = max(r.track_lost_timeout_s, r.track_stitch_window_s) | |
| stale = [ | |
| tid | |
| for tid, ru in runners.items() | |
| if ru.last_t is not None and ru.last_t < t_frame - _lost_window_for(ru, r, base) | |
| ] | |
| for tid in stale: | |
| events_raw.extend(runners.pop(tid).fsm.finalize()) | |
| for runner in runners.values(): | |
| events_raw.extend(runner.fsm.finalize()) | |
| return postprocess_events(events_raw, cfg.events), (debug or []) | |
| def _bbox(row) -> tuple: | |
| return ( | |
| float(row.bbox_x1), | |
| float(row.bbox_y1), | |
| float(row.bbox_x2), | |
| float(row.bbox_y2), | |
| ) | |