POSA_LEAKSMS/raspberry_pi4_sht30/sht30_sensor.py

#!/usr/bin/env python3
# =============================================================================
# sht30_sensor.py - SHT30 온습도 모니터링 메인 (2번 모듈, Raspberry Pi 4 전용)
#
# [대상 하드웨어] Raspberry Pi 4 Model B (BCM2711, 40핀 GPIO 헤더)
#   - SHT30(Sensirion, 한진데이터 P4422-3) I2C 온습도 센서를 I2C 버스 1에 연결.
#   - SDA=GPIO2(물리 3), SCL=GPIO3(물리 5), ADDR=GND → 0x44, 3.3V/GND, 4.7kΩ 풀업.
#
# [이식 출처] STM32 펌웨어 sht30_fw 의 동작을 Raspberry Pi 4 + Python 으로 이식.
#   firmware/board_sht30/app_sht30.c  (측정 루프 / startup→periodic / 범위 분류)
#   firmware/common/reporter.c        (본문 생성 → 서명 → POST → 재시도)
#   firmware/common/sht30*.c          (I2C 측정 / CRC / 변환)  → sht30_driver.py
#   firmware/test/host/reference.py   (wire 포맷)             → wire.py
#
# [동작 흐름]
#   1. SHT30 드라이버 open (실패/미가용 시 시뮬레이션 폴백)
#   2. 주기(REPORT_INTERVAL)마다 측정 → 서버에 POST
#   3. "최초 성공" 측정만 event_type="startup", 이후 "periodic"
#   4. 측정 실패(통신/CRC)는 크래시하지 않고 로그 후 짧게 쉬고 재시도
#   5. 물리 범위를 벗어나면 metric_status="out_of_range" (서버가 운영 임계 재판정)
#
# [인증] 서버는 raw-body HMAC 서명을 요구한다(레거시 body-sig 비활성).
#   X-Signature = lowercase_hex(HMAC-SHA256(key=API_KEY, msg=전송 본문 바이트))
#   → 서명한 바이트와 전송 바이트가 동일하도록 본문을 직접 만들어 raw POST 한다.
#
# [설정·비밀값] config.py 는 비밀이 아닌 기본값만, 서버 URL/API_KEY/기기 식별자는
#   config_local.py(미커밋)에서 로드한다. 없으면 기동 시 fail-closed 로 즉시 중단.
# =============================================================================

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


# ─────────────────────────────────────────────
# 로거 설정
# ─────────────────────────────────────────────
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  # 권한 없으면 콘솔만 사용

    ch = logging.StreamHandler(sys.stdout)
    ch.setFormatter(fmt)
    logger.addHandler(ch)

    return logger

log = setup_logger()


# ─────────────────────────────────────────────
# 종료 시그널 처리 (systemd stop = SIGTERM 깔끔 종료)
# ─────────────────────────────────────────────
class StopFlag:
    def __init__(self):
        self.stop = False

    def request(self, signum, frame):
        self.stop = True
        log.info(f"[SIGNAL] {signal.Signals(signum).name} 수신 → 종료 준비")

stop_flag = StopFlag()


# ─────────────────────────────────────────────
# 온습도 물리 범위 분류 (firmware classify_metric_status 와 동일)
#   서버는 이 값을 신뢰하지 않고 운영 임계로 재판정하므로, 여기서는 센서
#   물리 plausibility(범위/ 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"


# ─────────────────────────────────────────────
# 서버 보고 (raw-body + X-Signature HMAC)
# ─────────────────────────────────────────────
def post_reading(event_type: str, temp_c: float, rh: float, metric_status: str) -> bool:
    """SHT30 측정값을 서버에 POST. 성공(HTTP 200) 시 True.

    본문을 직접 만들어 그 바이트에 서명하고 그 바이트를 그대로 전송한다
    (서명한 바이트 == 전송 바이트 보장 → 서버 raw-body 검증과 일치).
    """
    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:
            # data=body_bytes 로 raw 본문을 그대로 전송(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}/{HTTP_RETRY_COUNT}: "
                        f"{resp.text[:160]}")
        except requests.exceptions.Timeout:
            log.warning(f"[POST TIMEOUT] 시도 {attempt}/{HTTP_RETRY_COUNT}")
        except requests.exceptions.ConnectionError as e:
            log.warning(f"[POST CONN ERR] 시도 {attempt}/{HTTP_RETRY_COUNT}: {e}")
        except requests.exceptions.RequestException as e:
            log.warning(f"[POST ERR] 시도 {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] {HTTP_RETRY_COUNT}회 모두 실패 (event={event_type})")
    return False


# ─────────────────────────────────────────────
# 측정 루프 (firmware sht30_measure_task 대응)
# ─────────────────────────────────────────────
def measure_loop(sensor: SHT30):
    # 최초 성공 측정만 startup, 이후 periodic (서버 sensor_log 1회 startup 기록)
    first_success = True

    log.info(f"[LOOP] 측정 시작 (간격={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:
            # 측정 실패(통신/CRC): 크래시 금지. 로그 후 짧게 쉬고 재시도.
            log.error(f"[MEASURE FAIL] {e} → {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:
            # 최초 성공 보고 후에만 startup→periodic 전환(최초 보고가 반드시 startup)
            first_success = False

        # 정상 보고 주기 대기 (종료 시그널이 오면 즉시 깸)
        _interruptible_sleep(REPORT_INTERVAL)


def _interruptible_sleep(seconds: float):
    """1초 단위로 나눠 자며 종료 시그널에 빠르게 반응한다."""
    end = time.time() + seconds
    while not stop_flag.stop and time.time() < end:
        time.sleep(min(1.0, end - time.time()))


# ─────────────────────────────────────────────
# 메인
# ─────────────────────────────────────────────
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"온습도(SHT30) 모니터링 v2 (Pi 4, 2번 모듈) 시작")
    log.info(f"기기={DEVICE_ID} | 위치={DEVICE_LOCATION} | 센서={SENSOR_NAME}(id={SENSOR_ID})")
    log.info(f"API={API_URL}")
    log.info(f"모드={'시뮬레이션' if simulate else f'실제 I2C (bus={I2C_BUS}, addr=0x{I2C_ADDR:02X})'}")
    if simulate and not FORCE_SIMULATION:
        log.warning("smbus2 미설치/ I2C 미가용 → 시뮬레이션 모드. 실측은 install.sh 후 재실행.")
    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 — 최상위 안전망(로그 후 종료)
        log.exception(f"[MAIN] 예외 발생: {e}")
        raise
    finally:
        sensor.close()
        log.info("[MAIN] 종료 완료.")


if __name__ == "__main__":
    main()