ec78be0bda
센서1·3(누수 노드)에 PMS7003 를 온습도와 동일 sensor_id 로 통합하고 metric_type='pms7003'
로 구분해 수집한다. LED 는 스위칭 없이 상시 점등 방식으로 확정.
- raspberry_pi4_pms7003/: PMS7003 노드 풀세트(드라이버·센서·install·systemd·vendor·배선·연결테스트·프레임 단위테스트)
- 서버: evaluate_metric_thresholds() 에 pm25/pm10 선택 인자 추가(온습도 전용 호출 후방호환),
sensor_data.php 수집 경로에 pms7003 metric 반영, config.php/예시에 PM 임계 추가
- SQL: migration_pms7003.sql · schema_sht30.sql 정합, tools/migration_leak_add_{pms7003,sht30} 마이그레이터
- tests/pm_threshold_test.php + run_tests.sh 러너 등록
- .gitignore: 노드별 config_local.py 커밋 차단
102 lines
4.1 KiB
Python
102 lines
4.1 KiB
Python
# =============================================================================
|
|
# wire.py - wire format for server transmission (body JSON + raw-body HMAC signature)
|
|
#
|
|
# [Why a separate module]
|
|
# The server (php/api/sensor_data.php) verifies, against the received "body bytes as-is",
|
|
# X-Signature = lowercase_hex(HMAC-SHA256(key=API_KEY, msg=raw_body)).
|
|
# Therefore the "signed bytes" and the "actually transmitted bytes" must be 100% identical.
|
|
# We build the body directly as a string, sign those bytes, then POST those same bytes as-is.
|
|
# (Leaving serialization to requests' json= can change float/whitespace/order and break the signature.)
|
|
#
|
|
# [Byte-level compatibility targets]
|
|
# PMS7003 is a Pi-only metric (no STM32 firmware), so pms_event_body() has no firmware
|
|
# counterpart. But the three helpers below are copied VERBATIM from the SHT30 module's
|
|
# wire.py so raw-body signing stays byte-identical to the shared reference:
|
|
# - firmware/common/sig.c::sig_raw_body (STM32 HMAC signature)
|
|
# - firmware/test/host/reference.py (verified Python reference)
|
|
# - php/config.php::verify_signature_raw (server verification part)
|
|
#
|
|
# fixed2() <-> jb_fixed2() / reference.fixed2() (kept for parity; PM body is integer-only)
|
|
# sign_raw() <-> sig_raw_body() / reference.sign_raw()
|
|
# pms_event_body() <- Pi-only PMS7003 body (mirrors sht30_event_body field order/escaping)
|
|
# =============================================================================
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
|
|
|
|
def fixed2(v: float) -> str:
|
|
"""Fixed 2-decimal format (round-half-up, locale-independent).
|
|
|
|
Byte-identical to jb_fixed2() / reference.fixed2().
|
|
24.0 -> "24.00"
|
|
48.5 -> "48.50"
|
|
-3.245 -> "-3.25"
|
|
"""
|
|
neg = v < 0
|
|
if neg:
|
|
v = -v
|
|
scaled = int(v * 100.0 + 0.5) # round half up (C: (long long)(v*100+0.5))
|
|
ip = scaled // 100
|
|
fp = scaled % 100
|
|
sign = "-" if (neg and scaled != 0) else ""
|
|
return f"{sign}{ip}.{fp:02d}"
|
|
|
|
|
|
def _esc(s: str) -> str:
|
|
r"""Escape JSON string value (identical to jb_estr / reference._esc).
|
|
|
|
Escapes only '"', '\\', and control characters (<0x20). '/' is left as-is.
|
|
Multibyte UTF-8 (e.g. Korean) is left as-is (server verifies raw bytes).
|
|
"""
|
|
out = []
|
|
for ch in s:
|
|
o = ord(ch)
|
|
if ch == '"':
|
|
out.append('\\"')
|
|
elif ch == '\\':
|
|
out.append('\\\\')
|
|
elif o < 0x20:
|
|
out.append('\\u%04x' % o)
|
|
else:
|
|
out.append(ch)
|
|
return "".join(out)
|
|
|
|
|
|
def pms_event_body(device_id: str, device_location: str, sensor_id: int,
|
|
sensor_name: str, event_type: str, timestamp: int,
|
|
pm1_0: int, pm2_5: int, pm10: int,
|
|
metric_status: str, app_version: str) -> str:
|
|
"""Generate PMS7003 event body JSON string.
|
|
|
|
Field order is fixed (shared wire contract) and mirrors sht30_event_body() so the
|
|
server can verify the raw body generically. metric_type is always "pms7003".
|
|
PM values are serialized as BARE INTEGERS (µg/m³, atmospheric concentration).
|
|
"""
|
|
return (
|
|
'{'
|
|
f'"device_id":"{_esc(device_id)}",'
|
|
f'"device_location":"{_esc(device_location)}",'
|
|
f'"sensor_id":{int(sensor_id)},'
|
|
f'"sensor_name":"{_esc(sensor_name)}",'
|
|
f'"event_type":"{_esc(event_type)}",'
|
|
f'"timestamp":{int(timestamp)},'
|
|
f'"metric_type":"pms7003",'
|
|
f'"pm1_0":{int(pm1_0)},'
|
|
f'"pm2_5":{int(pm2_5)},'
|
|
f'"pm10":{int(pm10)},'
|
|
f'"metric_status":"{_esc(metric_status)}",'
|
|
f'"app_version":"{_esc(app_version)}"'
|
|
'}'
|
|
)
|
|
|
|
|
|
def sign_raw(api_key: str, body: str) -> str:
|
|
"""raw-body signature: lowercase_hex(HMAC-SHA256(key=api_key, msg=body)).
|
|
|
|
Identical to sig_raw_body() / reference.sign_raw() / php verify_signature_raw.
|
|
The body is encoded as UTF-8 then signed (same as transmitted bytes).
|
|
"""
|
|
return hmac.new(api_key.encode("utf-8"), body.encode("utf-8"),
|
|
hashlib.sha256).hexdigest()
|