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
160
161
162
163
164
165
166
167
168
169
170
| """
Wi-Fi CSI/BFI 活动检测 - 海事值班疲劳监测
基于 JMSE 2024 论文方法
依赖: pip install numpy scipy
原理:
- Wi-Fi beamforming反馈信息(BFI)包含信道状态
- 人体运动引起多径变化→CSI波动
- 静止(疲劳/睡着)→CSI方差降低
"""
import numpy as np
from scipy.signal import butter, filtfilt
from scipy.fft import fft, fftfreq
from typing import Dict, Tuple
class WiFiActivityDetector:
"""
Wi-Fi感知活动状态检测
应用场景:
- 海事桥楼值班监控
- 汽车座舱乘员活动检测
- 低成本无隐私侵入方案
优势:
- 无需摄像头，隐私友好
- 全天候工作（不受光照影响）
- 已有Wi-Fi基础设施即可
"""
def __init__(self, fs: int = 100, n_antennas: int = 3):
self.fs = fs
self.n_antennas = n_antennas
def extract_csi_features(self, csi_data: np.ndarray) -> Dict:
"""
从CSI/BFI数据提取活动特征
Args:
csi_data: shape=(n_frames, n_subcarriers, n_antennas)
Returns:
features: 活动特征字典
"""
n_frames = csi_data.shape[0]
amplitude = np.abs(csi_data)
spatial_var = np.var(amplitude, axis=2)
temporal_var = np.var(amplitude, axis=0)
activity_level = np.mean(np.diff(amplitude, axis=0)**2)
freq_features = []
for ant in range(self.n_antennas):
for sc in range(0, csi_data.shape[1], 10):
signal = amplitude[:, sc, ant]
freqs = fftfreq(len(signal), 1/self.fs)
spectrum = np.abs(fft(signal))
mask = (freqs >= 0.5) & (freqs <= 3.0)
motion_power = np.sum(spectrum[mask]**2) if np.any(mask) else 0
freq_features.append(motion_power)
motion_energy = np.mean(freq_features)
breathing_power = 0
if activity_level < 0.01:
for ant in range(self.n_antennas):
signal = amplitude[:, 0, ant]
freqs = fftfreq(len(signal), 1/self.fs)
spectrum = np.abs(fft(signal))
mask = (freqs >= 0.1) & (freqs <= 0.5)
breathing_power += np.sum(spectrum[mask]**2)
return {
'activity_level': float(activity_level),
'spatial_variance': float(np.mean(spatial_var)),
'temporal_variance': float(np.mean(temporal_var)),
'motion_energy': float(motion_energy),
'breathing_power': float(breathing_power),
'is_active': activity_level > 0.05,
'is_static': activity_level < 0.01,
}
def classify_activity(self, features: Dict) -> Dict:
"""
分类活动状态
类别:
- walking: 走动值班（正常）
- standing_moving: 站立微动（正常）
- sitting_quiet: 坐着少动（关注）
- sleeping: 睡着（报警）
"""
activity = features['activity_level']
motion = features['motion_energy']
if activity > 0.1:
state = 'walking'
risk = 'normal'
elif activity > 0.03:
state = 'standing_moving'
risk = 'normal'
elif activity > 0.005:
state = 'sitting_quiet'
risk = 'attention'
else:
state = 'sleeping'
risk = 'alarm'
return {
'state': state,
'risk_level': risk,
'activity_score': round(activity, 4),
'motion_score': round(motion, 4),
'breathing_detected': features['breathing_power'] > 0.1,
}
if __name__ == "__main__":
np.random.seed(42)
detector = WiFiActivityDetector(fs=100, n_antennas=3)
n_frames = 1000
csi_walking = np.random.randn(n_frames, 56, 3) * 0.1
t = np.arange(n_frames) / 100
for i in range(n_frames):
csi_walking[i] += 0.5 * np.sin(2*np.pi*1.5*t[i])
csi_sleeping = np.random.randn(n_frames, 56, 3) * 0.01
for i in range(n_frames):
csi_sleeping[i, 0, :] += 0.02 * np.sin(2*np.pi*0.25*t[i])
print("=== 走动值班 ===")
feat_w = detector.extract_csi_features(csi_walking)
result_w = detector.classify_activity(feat_w)
for k, v in result_w.items():
print(f" {k}: {v}")
print(f"\n=== 静止值班（疑似疲劳）===")
feat_s = detector.extract_csi_features(csi_sleeping)
result_s = detector.classify_activity(feat_s)
for k, v in result_s.items():
print(f" {k}: {v}")
print(f"\n=== 关键指标对比 ===")
print(f"活动水平: 走动={feat_w['activity_level']:.4f}, "
f"静止={feat_s['activity_level']:.4f}")
print(f"运动能量: 走动={feat_w['motion_energy']:.4f}, "
f"静止={feat_s['motion_energy']:.4f}")
print(f"呼吸检测: 走动={'Yes' if feat_w['breathing_power']>0.1 else 'No'}, "
f"静止={'Yes' if feat_s['breathing_power']>0.1 else 'No'}")
|