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 커밋 차단
89 lines
3.5 KiB
Python
89 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Server connection test — run on the Pi 4 PMS7003 (미세먼지) module.
|
|
|
|
Builds the body with wire.py, attaches a raw-body HMAC (X-Signature), and
|
|
actually sends one signed POST. Secrets (API_URL/API_KEY) are loaded from config_local.py.
|
|
The combined-node setup.sh gates hardening on this script's exit code (0 = signed 200).
|
|
"""
|
|
import sys
|
|
import time
|
|
import subprocess
|
|
|
|
import requests
|
|
|
|
signed_ok = False # whether the signed POST returned 200 — setup.sh gates hardening on this exit code
|
|
|
|
from config import (
|
|
API_URL, API_KEY, DEVICE_ID, DEVICE_LOCATION,
|
|
SENSOR_ID, SENSOR_NAME, APP_VERSION,
|
|
)
|
|
import wire
|
|
|
|
print("=" * 50)
|
|
print(" Server connection test (Pi 4 PMS7003 / 미세먼지 module)")
|
|
print("=" * 50)
|
|
print(f" API_URL = {API_URL}")
|
|
|
|
# ── Step 1: check endpoint reachability with curl (no signature → 403 auth failure is expected) ──
|
|
print("\n[1/2] Checking sensor_data.php reachability with curl (no signature → 403 auth failure is expected)...")
|
|
try:
|
|
result = subprocess.run(
|
|
["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}",
|
|
"-A", "PMS7003Sensor/2.0", "-X", "POST", API_URL,
|
|
"-H", "Content-Type: application/json", "-d", "{}"],
|
|
capture_output=True, text=True, timeout=10
|
|
)
|
|
code = result.stdout.strip()
|
|
print(f" HTTP status code: {code}")
|
|
if code in ("400", "403"):
|
|
print(" → Endpoint reachable OK (rejected without signature/fields — expected)")
|
|
elif code == "200":
|
|
print(" → 200? Server accepted without a signature — check server configuration")
|
|
elif code == "404":
|
|
print(" → 404: check the API_URL path")
|
|
else:
|
|
print(f" → Response: {code} (check hosting/firewall/.htaccess)")
|
|
except Exception as e:
|
|
print(f" → curl failed: {e}")
|
|
|
|
# ── Step 2: real measurement POST signed with wire.py (startup, dummy PM values) ──
|
|
print("\n[2/2] Signed measurement POST (startup, dummy PM1.0=8 PM2.5=12 PM10=18)...")
|
|
try:
|
|
ts = int(time.time())
|
|
body = wire.pms_event_body(
|
|
DEVICE_ID, DEVICE_LOCATION, SENSOR_ID, SENSOR_NAME,
|
|
"startup", ts, 8, 12, 18, "normal", APP_VERSION,
|
|
)
|
|
sig = wire.sign_raw(API_KEY, body)
|
|
print(f" Body: {body}")
|
|
print(f" X-Signature: {sig[:16]}...")
|
|
resp = requests.post(
|
|
API_URL, data=body.encode("utf-8"),
|
|
headers={
|
|
"Content-Type": "application/json",
|
|
"X-Signature": sig,
|
|
"X-Device-Id": DEVICE_ID,
|
|
"User-Agent": "PMS7003Sensor/2.0",
|
|
},
|
|
timeout=10,
|
|
)
|
|
print(f" HTTP status code: {resp.status_code}")
|
|
print(f" Response body: {resp.text[:200]}")
|
|
if resp.status_code == 200:
|
|
signed_ok = True
|
|
print(f" → Success! Signature verified, measurement stored (check dashboard sensor_id={SENSOR_ID})")
|
|
elif resp.status_code == 403:
|
|
print(" → 403 auth failure: API_KEY differs from the server's config.local.php, or signature mismatch")
|
|
elif resp.status_code == 400:
|
|
print(" → 400: check timestamp freshness (±600s) — verify Pi clock (NTP) sync")
|
|
else:
|
|
print(f" → Status code: {resp.status_code}")
|
|
except ImportError:
|
|
print(" → requests not installed. Run install.sh or sudo apt install python3-requests")
|
|
except Exception as e:
|
|
print(f" → Failed: {e}")
|
|
|
|
print("\n" + "=" * 50)
|
|
print(" Test complete" + ("" if signed_ok else " — signed POST failed (see log above)"))
|
|
print("=" * 50)
|
|
sys.exit(0 if signed_ok else 1)
|