POSA_LEAKSMS/raspberry_pi4_sht30/sht30_driver.py

# =============================================================================
# sht30_driver.py - SHT30 I2C measurement driver (Raspberry Pi 4, I2C bus 1)
#
# [Ported from]
#   firmware/common/sht30.c          (STM32=0x2C06; Pi uses 0x2400 -> ~20ms wait -> 6-byte read)
#     * The Pi uses the clock-stretch-disabled command (0x2400) — Pi/Arduino I2C masters
#       handle clock stretching poorly (Pi BCM I2C stretch bug). The vendor example (P4422-2
#       ArtronShop_SHT3x) also avoids stretching for the same reason (periodic + fetch). STM32
#       (HAL supports stretching) keeps 0x2C06. Conversion/CRC are identical.
#   firmware/common/sht30_convert.c  (CRC-8 verification + physical-value conversion)
#   firmware/test/host/reference.py  (sht30_crc8 / sht30_parse Python reference)
#
# [Operation]
#   1) Send measurement command 0x24 0x00 (high-repeatability, clock stretching disabled)
#   2) Wait ~20ms (high-repeatability measurement max 15ms + margin)
#   3) 6-byte read: T_msb,T_lsb,T_crc, RH_msb,RH_lsb,RH_crc
#   4) CRC-8 (poly 0x31, init 0xFF) verification then conversion:
#        temp_c = -45 + 175 * raw_t / 65535
#        rh     = 100 * raw_rh / 65535
#
# [I2C access] Uses smbus2's (pure Python, arch-agnostic wheel) i2c_msg to send write/read
#   as explicitly separate transfers. read_i2c_block_data writes a register byte
#   first, so it breaks the SHT30 measurement frame -> use i2c_msg.read.
#
# [Simulation] If smbus2 is not installed or /dev/i2c-* is absent (dev PC), it switches to
#   SIMULATION=True and returns plausible dummy measurements (for server integration/format checks).
# =============================================================================
from __future__ import annotations

import time

# ── Constants ────────────────────────────────────────────────────────────────
# 0x2400 = single-shot, high-repeatability, clock stretching DISABLED (Sensirion SHT3x datasheet).
#   (0x2C06 = stretch ENABLED — for STM32 firmware. The Pi uses a write->wait->read pattern, so
#    the stretch-disabled command is accurate and safe. The vendor example ArtronShop_SHT3x also avoids stretching.)
SHT30_CMD = [0x24, 0x00]      # high-repeatability, clock stretching disabled (Pi-safe)
MEAS_DELAY_SEC = 0.02         # wait for measurement completion (15ms + margin)
FRAME_LEN = 6

try:
    from smbus2 import SMBus, i2c_msg
    SMBUS_AVAILABLE = True
except ImportError:
    SMBUS_AVAILABLE = False


# ── CRC-8 (identical to sht30_convert.c::sht30_crc8) ─────────────────────────
def crc8(data: bytes) -> int:
    """SHT3x CRC-8: poly 0x31, init 0xFF, no reflection, no final xor."""
    crc = 0xFF
    for b in data:
        crc ^= b
        for _ in range(8):
            crc = ((crc << 1) ^ 0x31) & 0xFF if (crc & 0x80) else (crc << 1) & 0xFF
    return crc


# ── Frame parsing (identical to sht30_convert.c::sht30_parse) ────────────────
def parse_frame(frame: bytes):
    """Verify + convert the 6-byte measurement frame -> (temp_c, rh).

    Raises ValueError on CRC mismatch (caller retries/skips).
    """
    if len(frame) != FRAME_LEN:
        raise ValueError(f"frame length error: {len(frame)} (expected {FRAME_LEN})")
    if crc8(frame[0:2]) != frame[2]:
        raise ValueError("temperature CRC mismatch")
    if crc8(frame[3:5]) != frame[5]:
        raise ValueError("humidity CRC mismatch")
    raw_t = (frame[0] << 8) | frame[1]
    raw_rh = (frame[3] << 8) | frame[4]
    temp_c = -45.0 + 175.0 * (raw_t / 65535.0)
    rh = 100.0 * (raw_rh / 65535.0)
    return temp_c, rh


class SHT30:
    """SHT30 single-shot measurement driver.

    bus_num: I2C bus number (Pi 4 40-pin header = 1)
    addr:    7-bit I2C address (0x44 if ADDR=GND, 0x45 if ADDR=VDD)
    simulate: if True, returns dummy values without hardware (forced simulation)
    """

    def __init__(self, bus_num: int = 1, addr: int = 0x44, simulate: bool = False):
        self.bus_num = bus_num
        self.addr = addr
        self.simulate = simulate or not SMBUS_AVAILABLE
        self._bus = None
        self._sim_tick = 0

    def open(self):
        if self.simulate:
            return
        # If /dev/i2c-<bus_num> is missing (e.g. I2C disabled), report a clear error.
        try:
            self._bus = SMBus(self.bus_num)
        except (FileNotFoundError, PermissionError, OSError) as e:
            raise RuntimeError(
                f"Failed to open I2C bus {self.bus_num}: {e}. "
                f"Enable I2C via `sudo raspi-config` and verify the user's i2c group."
            ) from e

    def close(self):
        if self._bus is not None:
            try:
                self._bus.close()
            finally:
                self._bus = None

    def measure(self):
        """Single-shot measurement -> (temp_c, rh). Raises on failure (comms/CRC).

        In simulation mode, returns dummy values around 24-26 C and 48-54%.
        """
        if self.simulate:
            self._sim_tick += 1
            temp = 24.0 + (self._sim_tick % 5) * 0.5      # 24.0~26.0
            rh = 48.0 + (self._sim_tick % 7) * 1.0        # 48.0~54.0
            return temp, rh

        if self._bus is None:
            self.open()

        # 1) Send measurement command (write)
        write = i2c_msg.write(self.addr, SHT30_CMD)
        self._bus.i2c_rdwr(write)

        # 2) Wait for measurement completion (clock stretching disabled -> host waits explicitly)
        time.sleep(MEAS_DELAY_SEC)

        # 3) 6-byte read
        read = i2c_msg.read(self.addr, FRAME_LEN)
        self._bus.i2c_rdwr(read)
        frame = bytes(read)

        # 4) CRC verification + conversion
        return parse_frame(frame)

    def __enter__(self):
        self.open()
        return self

    def __exit__(self, *exc):
        self.close()
        return False