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 커밋 차단
194 lines
8.2 KiB
Python
194 lines
8.2 KiB
Python
# =============================================================================
|
|
# pms7003_driver.py - PMS7003 UART measurement driver (Raspberry Pi 4, /dev/serial0)
|
|
#
|
|
# [Sensor] Plantower PMS7003 laser particulate-matter sensor.
|
|
# Streams a fixed 32-byte frame over UART (3.3V logic, 9600 8N1). By default the
|
|
# sensor powers up in ACTIVE mode and pushes frames continuously (~1 frame/sec in
|
|
# stable mode). We do not switch it to passive mode — we just sync on the header and
|
|
# read one complete valid frame per measurement, discarding partials/stale bytes.
|
|
#
|
|
# [Frame layout] (big-endian words; total 32 bytes)
|
|
# byte 0 : 0x42 start char 1
|
|
# byte 1 : 0x4D start char 2
|
|
# byte 2-3 : frame length = 2*13 + 2 = 28 (0x001C)
|
|
# byte 4-29 : 13 uint16 data words (big-endian)
|
|
# data[0..2] = PM1.0 / PM2.5 / PM10, CF=1 (standard particle)
|
|
# data[3] = PM1.0 atmospheric (bytes 10-11) <- reported
|
|
# data[4] = PM2.5 atmospheric (bytes 12-13) <- reported
|
|
# data[5] = PM10 atmospheric (bytes 14-15) <- reported
|
|
# data[6..11]= particle counts (>0.3/0.5/1.0/2.5/5.0/10 µm)
|
|
# data[12] = reserved (version / error code)
|
|
# byte 30-31 : checksum = uint16(sum of bytes 0..29)
|
|
#
|
|
# We report the ATMOSPHERIC concentration words (µg/m³) — the value meant for ambient
|
|
# air-quality reporting (the CF=1 words are for controlled/lab conditions).
|
|
#
|
|
# [Simulation] If pyserial is not installed or the UART device is absent (dev PC), it
|
|
# switches to SIMULATION=True and returns plausible dummy PM readings (for server
|
|
# integration/format checks) — same pattern as sht30_driver's SMBUS_AVAILABLE fallback.
|
|
# =============================================================================
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
# ── Constants ────────────────────────────────────────────────────────────────
|
|
START1 = 0x42 # frame start char 1 ('B')
|
|
START2 = 0x4D # frame start char 2 ('M')
|
|
FRAME_LEN = 32 # total frame length (bytes)
|
|
BODY_LEN_FIELD = 28 # value of the length field (bytes 2-3) = 2*13 data + 2 checksum
|
|
DATA_WORD_COUNT = 13 # number of uint16 data words (bytes 4..29)
|
|
|
|
# Index of the reported ATMOSPHERIC concentration words within the 13-word data array.
|
|
IDX_PM1_0 = 3 # bytes 10-11
|
|
IDX_PM2_5 = 4 # bytes 12-13
|
|
IDX_PM10 = 5 # bytes 14-15
|
|
|
|
try:
|
|
import serial # pyserial (pure Python, arch-independent py3-none-any wheel)
|
|
SERIAL_AVAILABLE = True
|
|
except ImportError:
|
|
SERIAL_AVAILABLE = False
|
|
|
|
|
|
# ── Frame parsing (pure; importable without serial hardware) ─────────────────
|
|
def parse_frame(frame: bytes):
|
|
"""Verify + decode a 32-byte PMS7003 frame -> (pm1_0, pm2_5, pm10) ints (µg/m³).
|
|
|
|
Raises ValueError on wrong length / bad header / length-field mismatch / checksum
|
|
mismatch (caller retries/skips). Reports the atmospheric concentration words.
|
|
"""
|
|
if len(frame) != FRAME_LEN:
|
|
raise ValueError(f"frame length error: {len(frame)} (expected {FRAME_LEN})")
|
|
if frame[0] != START1 or frame[1] != START2:
|
|
raise ValueError(f"bad header: {frame[0]:#04x} {frame[1]:#04x} "
|
|
f"(expected {START1:#04x} {START2:#04x})")
|
|
|
|
length_field = (frame[2] << 8) | frame[3]
|
|
if length_field != BODY_LEN_FIELD:
|
|
raise ValueError(f"frame length field mismatch: {length_field} (expected {BODY_LEN_FIELD})")
|
|
|
|
# Checksum = sum of bytes 0..29, compared to the uint16 at bytes 30-31.
|
|
checksum = sum(frame[0:30]) & 0xFFFF
|
|
expected = (frame[30] << 8) | frame[31]
|
|
if checksum != expected:
|
|
raise ValueError(f"checksum mismatch: {checksum:#06x} (expected {expected:#06x})")
|
|
|
|
# Decode the 13 big-endian data words (bytes 4..29).
|
|
data = [
|
|
(frame[4 + 2 * i] << 8) | frame[5 + 2 * i]
|
|
for i in range(DATA_WORD_COUNT)
|
|
]
|
|
pm1_0 = data[IDX_PM1_0]
|
|
pm2_5 = data[IDX_PM2_5]
|
|
pm10 = data[IDX_PM10]
|
|
return pm1_0, pm2_5, pm10
|
|
|
|
|
|
class PMS7003:
|
|
"""PMS7003 UART measurement driver (active mode, frame-synced read).
|
|
|
|
port: serial device (Pi 4: /dev/serial0 -> primary UART; also /dev/ttyS0 / /dev/ttyAMA0)
|
|
baud: 9600 (PMS7003 fixed)
|
|
simulate: if True, returns dummy values without hardware (forced simulation)
|
|
read_timeout: pyserial per-read timeout (seconds)
|
|
frame_wait: max time to wait for a valid frame in one measure() call (seconds)
|
|
"""
|
|
|
|
def __init__(self, port: str = "/dev/serial0", baud: int = 9600,
|
|
simulate: bool = False, read_timeout: float = 2.0,
|
|
frame_wait: float = 5.0):
|
|
self.port = port
|
|
self.baud = baud
|
|
self.simulate = simulate or not SERIAL_AVAILABLE
|
|
self.read_timeout = read_timeout
|
|
self.frame_wait = frame_wait
|
|
self._ser = None
|
|
self._sim_tick = 0
|
|
|
|
def open(self):
|
|
if self.simulate:
|
|
return
|
|
# If the UART device is missing (serial console still enabled / UART not enabled),
|
|
# report a clear, actionable error.
|
|
try:
|
|
self._ser = serial.Serial(
|
|
port=self.port,
|
|
baudrate=self.baud,
|
|
bytesize=serial.EIGHTBITS,
|
|
parity=serial.PARITY_NONE,
|
|
stopbits=serial.STOPBITS_ONE,
|
|
timeout=self.read_timeout,
|
|
)
|
|
except (FileNotFoundError, PermissionError, OSError) as e:
|
|
raise RuntimeError(
|
|
f"Failed to open UART {self.port}: {e}. "
|
|
f"Enable UART + disable the serial console via `sudo raspi-config` "
|
|
f"(Interface Options -> Serial Port), and verify the user's dialout group."
|
|
) from e
|
|
# Discard any stale/partial bytes buffered before we started reading.
|
|
try:
|
|
self._ser.reset_input_buffer()
|
|
except OSError:
|
|
pass
|
|
|
|
def close(self):
|
|
if self._ser is not None:
|
|
try:
|
|
self._ser.close()
|
|
finally:
|
|
self._ser = None
|
|
|
|
def measure(self):
|
|
"""Read one valid frame -> (pm1_0, pm2_5, pm10) ints. Raises on failure/timeout.
|
|
|
|
In simulation mode, returns dummy values in a plausible indoor range.
|
|
Syncs on the 0x42 0x4D header, reads the remaining 30 bytes, and validates via
|
|
parse_frame(). Bad/partial frames are discarded and retried until frame_wait elapses.
|
|
"""
|
|
if self.simulate:
|
|
self._sim_tick += 1
|
|
pm1_0 = 5 + (self._sim_tick % 6) # 5..10
|
|
pm2_5 = 10 + (self._sim_tick % 11) # 10..20
|
|
pm10 = 18 + (self._sim_tick % 13) # 18..30
|
|
return pm1_0, pm2_5, pm10
|
|
|
|
if self._ser is None:
|
|
self.open()
|
|
|
|
deadline = time.time() + self.frame_wait
|
|
last_err = None
|
|
while time.time() < deadline:
|
|
# 1) Sync on start char 1.
|
|
b = self._ser.read(1)
|
|
if not b:
|
|
last_err = "read timeout (no data on UART)"
|
|
continue
|
|
if b[0] != START1:
|
|
continue
|
|
# 2) Expect start char 2 immediately after.
|
|
b2 = self._ser.read(1)
|
|
if not b2 or b2[0] != START2:
|
|
continue
|
|
# 3) Read the remaining 30 bytes of the frame.
|
|
rest = self._ser.read(FRAME_LEN - 2)
|
|
if len(rest) != FRAME_LEN - 2:
|
|
last_err = "partial frame (short read)"
|
|
continue
|
|
frame = bytes([START1, START2]) + rest
|
|
# 4) Validate + decode (bad checksum -> keep trying within frame_wait).
|
|
try:
|
|
return parse_frame(frame)
|
|
except ValueError as e:
|
|
last_err = str(e)
|
|
continue
|
|
|
|
raise ValueError(f"no valid PMS7003 frame within {self.frame_wait}s "
|
|
f"(last: {last_err or 'unknown'})")
|
|
|
|
def __enter__(self):
|
|
self.open()
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
self.close()
|
|
return False
|