延續上一篇 ESP32-S3 WiFi CSI 感測:第一次真板實測、Serial 瓶頸與雙機 ESP-NOW。實驗證明「動作指標」可以分辨有人和沒人:靜止約 0.03,走動約 0.2。但每次都要收錄、跑分析、看圖,太麻煩了。這篇記錄怎麼把它做成一個即時監看網頁:Pi 4 一邊收 ESP32 的 CSI 一邊判斷,手機或電腦開瀏覽器就能看到現在有沒有人。整支程式只有一個 Python 檔加一個 HTML 檔,不裝任何新套件。
圖說:完成的監看頁(桌面版):有人剛走進兩片 ESP32-S3 之間,狀態轉紅、指標衝過門檻、瀑布圖右邊出現色塊
一、整體架構
ESP32-S3 (TX) ── ESP-NOW 每 10 ms 廣播 ──▶ ESP32-S3 (RX)
│ USB Serial 2 Mbaud,一行一筆 CSI
▼
Pi 4 csi_monitor.py
├─ Serial 執行緒:讀行 → 解析 → 算指標 → 判斷有人/沒人
│ └─ 事件寫 data/events_*.csv
└─ HTTP 伺服器(port 8090)
├─ GET / → monitor.html
├─ GET /stream → SSE,每 0.1 秒推一格資料
└─ POST /threshold → 改門檻
│
▼
手機/電腦瀏覽器:EventSource 接資料 → Canvas 畫圖
| 檔案 | 做什麼 |
|---|---|
host/csi_parse.py |
解析 CSI_DATA,... 行(上一篇就有,直接沿用) |
host/csi_monitor.py |
收資料、算指標、判斷、HTTP/SSE 伺服器,約 230 行 |
host/monitor.html |
網頁本身:版面、Canvas 畫圖、互動,約 300 行,不引用外部函式庫 |
Pi 4 上原本只有 numpy 和 pyserial,所以刻意只用 Python 標準函式庫:http.server 的 ThreadingHTTPServer 當伺服器,前端也不用圖表函式庫,全部用 Canvas 自己畫。好處是不必 pip install,家裡網路斷了也能用。
二、GUI 設計:畫面由上而下
設計原則是「最常看的放最大」。大部分時候只想知道一件事:現在有沒有人?其他區塊用來回答第二層的問題:判斷可不可靠、為什麼這樣判。
| 區塊 | 內容 | 設計考量 |
|---|---|---|
| 狀態 | 大字「有人」(紅框)/「沒人」(綠字),有人時顯示已持續幾秒 | 放最大、最左上。除了紅綠色,還有 ● ○ 圖示和文字,色盲也分得出來 |
| 數值方塊 | 動作指標、RSSI、封包率、漏收、子載波數 | 用來確認資料源正常。如果封包率掉、漏收升,代表發射端沒電或太遠,不是「沒人」 |
| 動作指標曲線 | 最近 60 秒;紅虛線是門檻,判定有人的時段塗淡紅底;下方有門檻滑桿 | 看指標離門檻多遠。換擺位時直接拖滑桿調門檻,不用改程式、不用重啟 |
| 子載波瀑布圖 | 最近 30 秒 × 128 個子載波,顏色是每格和 3 秒平均的差 | 只畫「變化」,不畫原始振幅:靜止時一片灰,有人時色塊非常明顯。用藍-灰-紅,不用彩虹色階 |
| RSSI 曲線 | 最近 60 秒 | 輔助指標,有人擋住直線路徑會往下掉。和動作指標分開畫,不做雙 Y 軸 |
| 事件紀錄 | 每次「有人」的開始、結束、秒數 | 離開一陣子回來也知道剛才發生什麼;同時寫進 CSV,之後拿來算誤報率 |
手機上是窄螢幕,自動改成單欄排列;系統設成深色模式時,整頁也跟著換成深色。所有顏色都定義成 CSS 變數,淺色和深色各一組。下面是手機實拍,同一分鐘內的「沒人」和「有人」:
圖說:手機實拍:左邊沒人(指標 0.025),右邊有人且已持續 5 秒。右邊的指標 0.055 已經低於門檻,但還在 2 秒的保持時間內,所以仍顯示有人
電腦版實際跑起來的樣子:人已經離開一陣子,狀態是「沒人」,右下角事件紀錄留下剛才的 4 次走動。
圖說:電腦版實拍(深色模式):沒人時動作指標穩定在 0.03~0.05,離 0.06 門檻還有一段距離;封包率 93/秒、漏收 7.2%,資料源正常。事件紀錄最近一次是 11:12:21~11:12:26,持續 5.2 秒
三、後端:csi_monitor.py
1. 讀 Serial:整塊讀,自己切行
上一篇踩過的坑:pyserial 的 readline() 一次讀 1 byte,在 Pi 4 上跟不上每秒 135 KB 的 CSI,會掉字元。所以這裡沿用整塊讀的寫法,並且包一層 try/except。板子拔掉或重燒時不會整支程式掛掉,而是 3 秒後自動重開:
while True:
try:
ser = serial.Serial(args.port, args.baud, timeout=0.2)
buf = b""
while True:
buf += ser.read(max(1, ser.in_waiting))
*lines, buf = buf.split(b"\n")
for line in lines:
r = parse_line(line.decode(errors="replace"))
if r:
mon.feed(*r, time.time())
except Exception as e: # 板子拔掉、重燒時等它回來
print(f"Serial 錯誤:{e},3 秒後重試")
time.sleep(3)
Serial 埠用 /dev/serial/by-id/usb-1a86_..._5D01087713-if00,不用 /dev/ttyACM0。兩片板子都插在同一台 Pi 4 上,ttyACM0 和 ttyACM1 的順序會隨插拔改變,by-id 則固定對應到同一片板子。
2. 算指標:和實驗完全一樣
每收到一包 CSI 就呼叫 feed(),步驟和上一篇的對照實驗相同:
- 振幅除以這一包自己的平均值,去掉自動增益(AGC)造成的整體忽大忽小
- 累積 10 包平均成一「格」(約 0.1 秒),壓掉單包雜訊
- 最近 30 格(3 秒)每個子載波算標準差,再取平均,就是動作指標
def feed(self, meta, csi, now):
amp = np.abs(csi)
if self.n_sub != len(amp): # 換封包長度就重來
self.n_sub, self.acc, self.frames = len(amp), [], deque(maxlen=WIN_FRAMES)
self.acc.append(amp / amp.mean()) # ① 去 AGC
if len(self.acc) < AVG_N:
return
frame = np.mean(self.acc, axis=0) # ② 10 包平均成一格
self.acc = []
self.frames.append(frame)
win = np.array(self.frames)
motion = float(win.std(axis=0).mean()) # ③ 3 秒窗標準差取平均
dev = frame - win.mean(axis=0) # 瀑布圖用:這格偏離平均多少
self.judge(motion, now)
...
瀑布圖畫的是 dev,也就是「這一格和最近 3 秒平均的差」,所以靜止時接近 0(灰色),有人時才出現紅藍。
為了確認即時版的算法沒寫錯,我把早上的實驗收錄檔丟進同一個 feed() 重播。結果靜止 0.030、走動 0.196、靜止 0.026,和實驗時另外寫的分析程式一致。
3. 判斷:門檻+2 秒保持
def judge(self, motion, now):
if motion > self.threshold:
self.last_hit = now
if not self.present:
self.present, self.event_start = True, now # 事件開始
elif self.present and now - self.last_hit > HOLD_S:
self.present = False # 事件結束,寫 CSV
...
進入「有人」很快,只要一格超過門檻就算。退出則要指標持續低於門檻 2 秒。人走路時兩步之間指標會短暫下降,如果沒有這段保持時間,狀態會在門檻附近一直閃,事件紀錄也會被切成一堆很短的片段。
判斷放在伺服器端,不放在網頁裡。所以不管開了幾個網頁、甚至都沒人開,判斷結果都一致,事件也都會寫進 CSV。
4. 推資料:Server-Sent Events
瀏覽器要每 0.1 秒拿到新資料。我選 SSE(Server-Sent Events)而不是 WebSocket,原因是:
- 資料只有伺服器推往瀏覽器一個方向,SSE 剛好夠用;改門檻這種少量操作,用一般的 POST 就好
- SSE 就是一個不結束的 HTTP 回應,用內建
http.server就能做,不用另外裝 websocket 套件 - 瀏覽器的
EventSource斷線會自己重連,不用另外寫重連邏輯
每個連上來的網頁有自己的一個 queue。Serial 執行緒算完一格就丟進所有 queue,各自的 HTTP 執行緒再寫出去:
def broadcast(self, kind, data):
payload = f"event: {kind}\ndata: {json.dumps(data)}\n\n".encode()
with self.lock:
for q in list(self.clients):
try:
q.put_nowait(payload)
except queue.Full:
pass # 網頁跟不上就丟格,不能卡住收資料的執行緒
這裡有兩個細節:
- queue 有上限(50 格),滿了就丟:手機切到背景時瀏覽器會暫停讀資料。如果這時候用會等待的
put(),連 Serial 執行緒都會被卡住,CSI 就漏收了 - 新連上的網頁先補資料:伺服器保留最近 600 格(60 秒)。網頁一連上,先送一個
init事件,內含歷史資料、事件紀錄和目前門檻,畫面一打開就是滿的
每一格的資料長這樣,128 個子載波的差值取到小數 3 位,一格約 1 KB,每秒約 10 KB,手機用行動網路看也沒問題:
{"t": 1793414664.94, "motion": 0.0254, "rssi": -36, "pps": 95.0, "loss": 0.0494,
"len": 256, "bad": 0, "present": false, "th": 0.06, "dev": [0.0, -0.024, -0.031, ...]}
5. 重播模式:沒有板子也能開發
--replay 收錄檔 會照封包原本的時間戳記間隔重播,網頁看起來就跟現場一樣。上面「指標和實驗一致」的驗證就是在 gx10 上用重播模式做的,調畫面時也不必每次都到 Pi 4 旁邊走來走去。
python3 csi_monitor.py --replay ../data/csi_20261031_092328.log
四、前端:monitor.html
1. 接資料
const es = new EventSource('stream');
es.addEventListener('init', e => { /* 歷史、事件、門檻 */ });
es.addEventListener('frame', e => {
frames.push(JSON.parse(e.data));
setStatus(...); // 數值方塊與狀態立刻更新
if (!pending) { pending = true; requestAnimationFrame(() => { pending = false; draw(); }); }
});
es.addEventListener('event', e => { /* 事件紀錄表 */ });
重畫用 requestAnimationFrame 合併:一次畫面更新裡收到再多格,也只重畫一次。手機切到背景時,瀏覽器也會自動暫停重畫。
2. 折線圖:Canvas 自己畫
動作指標和 RSSI 用同一個 lineChart(),依序畫這幾層:
- 「有人」時段的淡紅底
- 格線與刻度
- 門檻虛線
- 2 px 的資料線
- 滑鼠或手指停留時的十字線和提示框
動作指標的 Y 軸上限取「門檻 2 倍」和「資料最大值」較大的那個,所以沒人時門檻線也不會跑出畫面。
3. 瀑布圖:ImageData 一格一像素
30 秒 × 128 個子載波,大約 300 × 128 個點。不需要一個一個畫方塊,而是先建一張 300×128 像素的小圖,每個像素填顏色,再用 drawImage 放大貼上去,並關掉平滑處理(程式碼略為簡化):
const img = g.createImageData(data.length, n);
data.forEach((f, x) => f.dev.forEach((v, y) => {
const [r, gg, b] = divColor(v), i = ((n - 1 - y) * data.length + x) * 4;
img.data.set([r, gg, b, 255], i);
}));
off.getContext('2d').putImageData(img, 0, 0);
g.imageSmoothingEnabled = false;
g.drawImage(off, x0, top, width, height);
色階 divColor() 是三段漸層:-0.25 藍、0 灰、+0.25 紅。三個端點的顏色也從 CSS 變數讀,深色模式的灰會換成接近背景的深灰,靜止時整片就融進背景裡。
4. 門檻滑桿
拖滑桿時畫面上的紅虛線會立刻移動,但要停手 0.3 秒後才送一次 POST /threshold 給伺服器,避免拖一下就送幾十個請求。伺服器改了門檻後,下一格資料裡的 th 會帶新值。其他開著的網頁收到後,也會把自己的滑桿同步過去。
五、踩到的坑:手機上曲線越拉越高
電腦上一切正常,手機一打開,動作指標那張圖卻只剩上面 0.099~0.132 一小段,曲線大半跑到畫面外。
原因是高解析度螢幕的處理。為了讓 Canvas 在 3 倍螢幕上不糊,程式會把 Canvas 的像素尺寸設成「顯示尺寸 × devicePixelRatio」。但我把顯示高度存在 HTML 的 height 屬性裡,而設定 canvas.height 正好會改寫這個屬性:
第 1 次:讀 height=180 → 設 canvas.height = 540(× 3) 第 2 次:讀 height=540 → 設 canvas.height = 1620 ……
每重畫一次就變高 3 倍。電腦螢幕的 devicePixelRatio 是 1,乘完不變,所以在電腦上完全看不出來。修法是第一次就把原始高度另外存起來:
if (!c.dataset.h) c.dataset.h = c.getAttribute('height');
const h = +c.dataset.h;
之後我截圖驗證時,都把無頭瀏覽器的 deviceScaleFactor 設成 3,模擬手機螢幕。
六、怎麼跑、資料量多大
cd ~/esp32-csi/host python3 csi_monitor.py --port /dev/serial/by-id/usb-1a86_USB_Single_Serial_5D01087713-if00 # 瀏覽器開 http://<Pi 4 的 IP>:8090
| 參數 | 用途 |
|---|---|
--threshold 0.06 |
初始門檻(網頁上也能調) |
--replay 檔案 |
重播收錄檔,不讀 Serial |
--record |
同時存原始 CSI 到 data/ |
--http 8090 |
網頁 port |
平常不開 --record,這時會一直長大的只有事件 CSV,每筆約 25 bytes,一天下來也才幾十 KB。曲線和瀑布圖都只存在記憶體裡,最多 60 秒。如果開 --record 存原始 CSI,每秒約 75 KB、每天約 6.5 GB,只適合做實驗時錄幾分鐘。
從外面看的話,Pi 4 有裝 Tailscale,手機開 Tailscale App 後用 Pi 4 的 Tailscale IP 連就可以。要用 IP,不要用主機簡稱,iPhone 上簡稱常常解析不到。
七、小結與下一步
- ✅ 一個 Python 檔加一個 HTML 檔,不裝新套件,Pi 4 即時判斷有人/沒人,手機、電腦都能看
- ✅ 指標和實驗分析程式一致;判斷放在伺服器端,用門檻加 2 秒保持,事件寫成 CSV
- ✅ SSE 推資料、新網頁先補 60 秒歷史、queue 滿了就丟格,不拖慢收資料
- ✅ 有重播模式,沒有板子也能開發和測試
- ⏳ 做成開機自動啟動的 systemd 服務
- ⏳ 放著跑一整天,用事件 CSV 算誤報率,再決定門檻要不要調














