POSA_LEAKSMS/raspberry_pi4_sht30/sht30_sensor.py

#!/usr/bin/env python3
# =============================================================================
# sht30_sensor.py - SHT30 temperature/humidity monitoring main (Module 2, Raspberry Pi 4 only)
#
# [Target hardware] Raspberry Pi 4 Model B (BCM2711, 40-pin GPIO header)
#   - SHT30 (Sensirion, Hanjin Data P4422-3) I2C temp/humidity sensor connected to I2C bus 1.
#   - SDA=GPIO2 (physical 3), SCL=GPIO3 (physical 5), ADDR=GND → 0x44, 3.3V/GND, 4.7kΩ pull-up.
#
# [Port source] Ported the behavior of STM32 firmware sht30_fw to Raspberry Pi 4 + Python.
#   firmware/board_sht30/app_sht30.c  (measurement loop / startup→periodic / range classification)
#   firmware/common/reporter.c        (build body → sign → POST → retry)
#   firmware/common/sht30*.c          (I2C measurement / CRC / conversion)  → sht30_driver.py
#   firmware/test/host/reference.py   (wire format)             → wire.py
#
# [Operation flow]
#   1. Open SHT30 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/CRC) 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,
    I2C_BUS, I2C_ADDR, FORCE_SIMULATION,
    API_URL, API_KEY,
    DEVICE_ID, DEVICE_LOCATION,
    REPORT_INTERVAL, MEASURE_RETRY_DELAY,
    TEMP_LOW_C, TEMP_HIGH_C, RH_LOW, RH_HIGH,
    APP_VERSION,
    LOG_FILE, LOG_MAX_BYTES, LOG_BACKUP_COUNT,
    HTTP_TIMEOUT, HTTP_RETRY_COUNT, HTTP_RETRY_DELAY,
)
import wire
from sht30_driver import SHT30, SMBUS_AVAILABLE


# ─────────────────────────────────────────────
# Logger setup
# ─────────────────────────────────────────────
def setup_logger():
    logger = logging.getLogger("sht30_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()


# ─────────────────────────────────────────────
# Temp/humidity physical range classification (same as firmware 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(temp_c: float, rh: float) -> str:
    if not (math.isfinite(temp_c) and math.isfinite(rh)):
        return "out_of_range"
    if temp_c < TEMP_LOW_C or temp_c > TEMP_HIGH_C:
        return "out_of_range"
    if rh < RH_LOW or rh > RH_HIGH:
        return "out_of_range"
    return "normal"


# ─────────────────────────────────────────────
# Server reporting (raw-body + X-Signature HMAC)
# ─────────────────────────────────────────────
def post_reading(event_type: str, temp_c: float, rh: float, metric_status: str) -> bool:
    """POST the SHT30 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.sht30_event_body(
        device_id=DEVICE_ID,
        device_location=DEVICE_LOCATION,
        sensor_id=SENSOR_ID,
        sensor_name=SENSOR_NAME,
        event_type=event_type,
        timestamp=timestamp,
        temperature_c=temp_c,
        humidity_percent=rh,
        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": "SHT30Sensor/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} T={temp_c:.2f}C RH={rh:.2f}% "
                         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 (corresponds to firmware sht30_measure_task)
# ─────────────────────────────────────────────
def measure_loop(sensor: SHT30):
    # 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, addr=0x{I2C_ADDR:02X}, bus={I2C_BUS})")

    while not stop_flag.stop:
        try:
            temp_c, rh = sensor.measure()
        except (OSError, ValueError, RuntimeError) as e:
            # Measurement failure (comm/CRC): 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(temp_c, rh)
        event_type = "startup" if first_success else "periodic"

        log.info(f"[MEASURE] {event_type}: T={temp_c:.2f}C RH={rh:.2f}% status={metric_status}")

        ok = post_reading(event_type, temp_c, rh, 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 SMBUS_AVAILABLE
    log.info("=" * 60)
    log.info(f"Temp/humidity (SHT30) monitoring v2 (Pi 4, Module 2) 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 I2C (bus={I2C_BUS}, addr=0x{I2C_ADDR:02X})'}")
    if simulate and not FORCE_SIMULATION:
        log.warning("smbus2 not installed / I2C unavailable → simulation mode. For real readings, run install.sh then restart.")
    log.info("=" * 60)

    sensor = SHT30(bus_num=I2C_BUS, addr=I2C_ADDR, simulate=simulate)
    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()