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 커밋 차단
247 lines
9.6 KiB
Python
247 lines
9.6 KiB
Python
#!/usr/bin/env python3
|
|
# =============================================================================
|
|
# pms7003_sensor.py - PMS7003 particulate-matter monitoring main (Raspberry Pi 4 only)
|
|
#
|
|
# [Target hardware] Raspberry Pi 4 Model B (BCM2711, 40-pin GPIO header)
|
|
# - Plantower PMS7003 laser PM sensor connected to the primary UART.
|
|
# - PMS7003 TXD -> Pi RXD = GPIO15 (physical 10). VCC=5V (pin 2), GND (pin 6). 3.3V logic, 9600 8N1.
|
|
#
|
|
# [Design] PM is a NEW metric on the EXISTING metric pipeline; this module mirrors the SHT30
|
|
# module (sht30_sensor.py) exactly - measurement loop, startup->periodic, raw-body signing,
|
|
# POST + retry. There is no STM32 firmware for PMS (Pi-only).
|
|
# wire.py (build body -> sign) -> wire.pms_event_body / wire.sign_raw
|
|
# pms7003_driver.py (UART frame read / checksum / decode) -> PMS7003.measure()
|
|
#
|
|
# [Operation flow]
|
|
# 1. Open PMS7003 driver (fall back to simulation on failure/unavailability)
|
|
# 2. Measure every interval (REPORT_INTERVAL) -> POST to server
|
|
# 3. Only the "first success" measurement uses event_type="startup", afterwards "periodic"
|
|
# 4. Measurement failure (comm/checksum) does not crash; log, sleep briefly, then retry
|
|
# 5. Out of physical range -> metric_status="out_of_range" (server re-evaluates operational thresholds)
|
|
#
|
|
# [Auth] The server requires a raw-body HMAC signature (legacy body-sig disabled).
|
|
# X-Signature = lowercase_hex(HMAC-SHA256(key=API_KEY, msg=sent body bytes))
|
|
# -> Build the body directly and raw-POST it so the signed bytes match the sent bytes.
|
|
#
|
|
# [Config/secrets] config.py holds only non-secret defaults; the server URL/API_KEY/device
|
|
# identifiers are loaded from config_local.py (uncommitted). If missing, fail-closed and abort at startup.
|
|
# =============================================================================
|
|
|
|
import time
|
|
import math
|
|
import logging
|
|
import logging.handlers
|
|
import signal
|
|
import sys
|
|
|
|
import requests
|
|
|
|
from config import (
|
|
SENSOR_ID, SENSOR_NAME,
|
|
UART_PORT, UART_BAUD, UART_TIMEOUT, FRAME_WAIT_TIMEOUT, FORCE_SIMULATION,
|
|
API_URL, API_KEY,
|
|
DEVICE_ID, DEVICE_LOCATION,
|
|
REPORT_INTERVAL, MEASURE_RETRY_DELAY,
|
|
PM_VALUE_MIN, PM_VALUE_MAX,
|
|
APP_VERSION,
|
|
LOG_FILE, LOG_MAX_BYTES, LOG_BACKUP_COUNT,
|
|
HTTP_TIMEOUT, HTTP_RETRY_COUNT, HTTP_RETRY_DELAY,
|
|
)
|
|
import wire
|
|
from pms7003_driver import PMS7003, SERIAL_AVAILABLE
|
|
|
|
|
|
# -----------------------------------------------
|
|
# Logger setup
|
|
# -----------------------------------------------
|
|
def setup_logger():
|
|
logger = logging.getLogger("pms7003_sensor")
|
|
logger.setLevel(logging.DEBUG)
|
|
|
|
fmt = logging.Formatter(
|
|
"%(asctime)s [%(levelname)s] %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S"
|
|
)
|
|
|
|
try:
|
|
fh = logging.handlers.RotatingFileHandler(
|
|
LOG_FILE, maxBytes=LOG_MAX_BYTES, backupCount=LOG_BACKUP_COUNT,
|
|
encoding="utf-8"
|
|
)
|
|
fh.setFormatter(fmt)
|
|
logger.addHandler(fh)
|
|
except PermissionError:
|
|
pass # No permission -> console only
|
|
|
|
ch = logging.StreamHandler(sys.stdout)
|
|
ch.setFormatter(fmt)
|
|
logger.addHandler(ch)
|
|
|
|
return logger
|
|
|
|
log = setup_logger()
|
|
|
|
|
|
# -----------------------------------------------
|
|
# Shutdown signal handling (systemd stop = SIGTERM clean shutdown)
|
|
# -----------------------------------------------
|
|
class StopFlag:
|
|
def __init__(self):
|
|
self.stop = False
|
|
|
|
def request(self, signum, frame):
|
|
self.stop = True
|
|
log.info(f"[SIGNAL] {signal.Signals(signum).name} received -> preparing to shut down")
|
|
|
|
stop_flag = StopFlag()
|
|
|
|
|
|
# -----------------------------------------------
|
|
# PM physical range classification (mirrors the SHT30 classify_metric_status)
|
|
# The server does not trust this value and re-evaluates with operational thresholds, so
|
|
# here we only check sensor physical plausibility (range / NaN / Inf).
|
|
# -----------------------------------------------
|
|
def classify_metric_status(pm1_0, pm2_5, pm10) -> str:
|
|
for v in (pm1_0, pm2_5, pm10):
|
|
if not math.isfinite(v):
|
|
return "out_of_range"
|
|
if v < PM_VALUE_MIN or v > PM_VALUE_MAX:
|
|
return "out_of_range"
|
|
return "normal"
|
|
|
|
|
|
# -----------------------------------------------
|
|
# Server reporting (raw-body + X-Signature HMAC)
|
|
# -----------------------------------------------
|
|
def post_reading(event_type: str, pm1_0: int, pm2_5: int, pm10: int, metric_status: str) -> bool:
|
|
"""POST the PMS7003 measurement to the server. Returns True on success (HTTP 200).
|
|
|
|
Build the body directly, sign those bytes, and send those exact bytes
|
|
(guarantees signed bytes == sent bytes -> matches the server's raw-body verification).
|
|
"""
|
|
timestamp = int(time.time())
|
|
body = wire.pms_event_body(
|
|
device_id=DEVICE_ID,
|
|
device_location=DEVICE_LOCATION,
|
|
sensor_id=SENSOR_ID,
|
|
sensor_name=SENSOR_NAME,
|
|
event_type=event_type,
|
|
timestamp=timestamp,
|
|
pm1_0=pm1_0,
|
|
pm2_5=pm2_5,
|
|
pm10=pm10,
|
|
metric_status=metric_status,
|
|
app_version=APP_VERSION,
|
|
)
|
|
body_bytes = body.encode("utf-8")
|
|
signature = wire.sign_raw(API_KEY, body)
|
|
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"X-Signature": signature,
|
|
"X-Device-Id": DEVICE_ID,
|
|
"User-Agent": "PMS7003Sensor/2.0",
|
|
}
|
|
|
|
for attempt in range(1, HTTP_RETRY_COUNT + 1):
|
|
try:
|
|
# Send the raw body as-is via data=body_bytes (do not re-serialize with json=).
|
|
resp = requests.post(
|
|
API_URL, data=body_bytes, headers=headers, timeout=HTTP_TIMEOUT
|
|
)
|
|
if resp.status_code == 200:
|
|
log.info(f"[POST OK] {event_type} PM1.0={pm1_0} PM2.5={pm2_5} PM10={pm10} ug/m3 "
|
|
f"status={metric_status} -> {resp.text[:100]}")
|
|
return True
|
|
log.warning(f"[POST {resp.status_code}] attempt {attempt}/{HTTP_RETRY_COUNT}: "
|
|
f"{resp.text[:160]}")
|
|
except requests.exceptions.Timeout:
|
|
log.warning(f"[POST TIMEOUT] attempt {attempt}/{HTTP_RETRY_COUNT}")
|
|
except requests.exceptions.ConnectionError as e:
|
|
log.warning(f"[POST CONN ERR] attempt {attempt}/{HTTP_RETRY_COUNT}: {e}")
|
|
except requests.exceptions.RequestException as e:
|
|
log.warning(f"[POST ERR] attempt {attempt}/{HTTP_RETRY_COUNT}: {e}")
|
|
|
|
if attempt < HTTP_RETRY_COUNT and not stop_flag.stop:
|
|
time.sleep(HTTP_RETRY_DELAY)
|
|
|
|
log.error(f"[POST FAIL] all {HTTP_RETRY_COUNT} attempts failed (event={event_type})")
|
|
return False
|
|
|
|
|
|
# -----------------------------------------------
|
|
# Measurement loop (mirrors the SHT30 measure loop)
|
|
# -----------------------------------------------
|
|
def measure_loop(sensor: PMS7003):
|
|
# Only the first successful measurement is startup, afterwards periodic (server records startup once in leak_sensor_log)
|
|
first_success = True
|
|
|
|
log.info(f"[LOOP] measurement started (interval={REPORT_INTERVAL}s, port={UART_PORT}, baud={UART_BAUD})")
|
|
|
|
while not stop_flag.stop:
|
|
try:
|
|
pm1_0, pm2_5, pm10 = sensor.measure()
|
|
except (OSError, ValueError, RuntimeError) as e:
|
|
# Measurement failure (comm/checksum): no crash. Log, sleep briefly, then retry.
|
|
log.error(f"[MEASURE FAIL] {e} -> retry after {MEASURE_RETRY_DELAY}s")
|
|
_interruptible_sleep(MEASURE_RETRY_DELAY)
|
|
continue
|
|
|
|
metric_status = classify_metric_status(pm1_0, pm2_5, pm10)
|
|
event_type = "startup" if first_success else "periodic"
|
|
|
|
log.info(f"[MEASURE] {event_type}: PM1.0={pm1_0} PM2.5={pm2_5} PM10={pm10} ug/m3 status={metric_status}")
|
|
|
|
ok = post_reading(event_type, pm1_0, pm2_5, pm10, metric_status)
|
|
if ok and first_success:
|
|
# Switch startup->periodic only after the first successful report (first report is always startup)
|
|
first_success = False
|
|
|
|
# Wait for the normal report interval (wakes immediately on a shutdown signal)
|
|
_interruptible_sleep(REPORT_INTERVAL)
|
|
|
|
|
|
def _interruptible_sleep(seconds: float):
|
|
"""Sleep in 1-second chunks to react quickly to shutdown signals."""
|
|
end = time.time() + seconds
|
|
while not stop_flag.stop and time.time() < end:
|
|
time.sleep(min(1.0, end - time.time()))
|
|
|
|
|
|
# -----------------------------------------------
|
|
# Main
|
|
# -----------------------------------------------
|
|
def main():
|
|
signal.signal(signal.SIGTERM, stop_flag.request)
|
|
signal.signal(signal.SIGINT, stop_flag.request)
|
|
|
|
simulate = FORCE_SIMULATION or not SERIAL_AVAILABLE
|
|
log.info("=" * 60)
|
|
log.info(f"Particulate-matter (PMS7003) monitoring v2 (Pi 4) starting")
|
|
log.info(f"device={DEVICE_ID} | location={DEVICE_LOCATION} | sensor={SENSOR_NAME}(id={SENSOR_ID})")
|
|
log.info(f"API={API_URL}")
|
|
log.info(f"mode={'simulation' if simulate else f'real UART (port={UART_PORT}, baud={UART_BAUD})'}")
|
|
if simulate and not FORCE_SIMULATION:
|
|
log.warning("pyserial not installed / UART unavailable -> simulation mode. For real readings, run install.sh then restart.")
|
|
log.info("=" * 60)
|
|
|
|
sensor = PMS7003(port=UART_PORT, baud=UART_BAUD, simulate=simulate,
|
|
read_timeout=UART_TIMEOUT, frame_wait=FRAME_WAIT_TIMEOUT)
|
|
try:
|
|
sensor.open()
|
|
except RuntimeError as e:
|
|
log.error(f"[INIT] {e}")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
measure_loop(sensor)
|
|
except Exception as e: # noqa: BLE001 - top-level safety net (log then exit)
|
|
log.exception(f"[MAIN] exception occurred: {e}")
|
|
raise
|
|
finally:
|
|
sensor.close()
|
|
log.info("[MAIN] shutdown complete.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|