1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159
| import numpy as np from typing import List, Tuple
class LatentDynamicCRF: """ Latent-Dynamic CRF 论文方法: 捕捉疲劳状态的时序动态 优势: - 考虑状态转移概率 - 建模潜在疲劳趋势 - 预测未来疲劳等级 """ def __init__(self, num_states: int = 5, num_features: int = 20): self.num_states = num_states self.num_features = num_features self.transition_matrix = self._init_transitions() self.emission_params = np.random.randn(num_states, num_features) def _init_transitions(self) -> np.ndarray: """ 初始化状态转移矩阵 疲劳状态转移特点: - 疲劳是渐进过程(很少跳跃) - 状态保持概率高 """ transitions = np.eye(self.num_states) * 0.6 for i in range(self.num_states - 1): transitions[i, i+1] = 0.3 transitions[i+1, i] = 0.1 transitions = transitions / transitions.sum(axis=1, keepdims=True) return transitions def predict_sequence(self, observations: np.ndarray) -> Tuple[np.ndarray, np.ndarray]: """ 预测疲劳序列 使用Viterbi算法解码最优路径 Args: observations: 观测序列, shape=(T, D) Returns: states: 状态序列, shape=(T,) probs: 状态概率, shape=(T,) """ T = len(observations) viterbi_table = np.zeros((T, self.num_states)) backpointer = np.zeros((T, self.num_states), dtype=int) for s in range(self.num_states): viterbi_table[0, s] = self._emission_prob(observations[0], s) for t in range(1, T): for s in range(self.num_states): max_prob = 0 best_prev = 0 for prev_s in range(self.num_states): prob = viterbi_table[t-1, prev_s] * self.transition_matrix[prev_s, s] if prob > max_prob: max_prob = prob best_prev = prev_s viterbi_table[t, s] = max_prob * self._emission_prob(observations[t], s) backpointer[t, s] = best_prev states = np.zeros(T, dtype=int) states[T-1] = np.argmax(viterbi_table[T-1]) for t in range(T-2, -1, -1): states[t] = backpointer[t+1, states[t+1]] probs = np.max(viterbi_table, axis=1) probs = probs / probs.sum() return states, probs def predict_future(self, current_state: int, horizon: int = 10) -> np.ndarray: """ 预测未来疲劳趋势 Args: current_state: 当前疲劳等级 horizon: 预测帧数 Returns: future_probs: 未来状态概率, shape=(horizon, num_states) """ prob_dist = np.zeros(self.num_states) prob_dist[current_state] = 1.0 future_probs = [prob_dist] for _ in range(horizon - 1): prob_dist = self.transition_matrix.T @ prob_dist future_probs.append(prob_dist) return np.array(future_probs) def _emission_prob(self, observation: np.ndarray, state: int) -> float: """计算发射概率""" distance = np.linalg.norm(observation - self.emission_params[state]) prob = np.exp(-distance ** 2 / 2) return prob
if __name__ == "__main__": crf = LatentDynamicCRF(num_states=5, num_features=20) T = 30 observations = np.random.randn(T, 20) states, probs = crf.predict_sequence(observations) print(f"疲劳等级序列: {states}") print(f"初始状态: {states[0]}, 最终状态: {states[-1]}") current_state = states[-1] future_probs = crf.predict_future(current_state, horizon=10) print(f"\n未来10帧疲劳趋势:") for i, prob in enumerate(future_probs[:5]): most_likely = np.argmax(prob) print(f"帧{i+1}: 疲劳等级 {most_likely} (概率 {prob[most_likely]:.2f})")
|