#!/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()