# ============================================================================= # 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