73e49fd528
2번 모듈 신규(raspberry_pi4_sht30/): SHT30 온습도 → cafe24, raw-body X-Signature HMAC, 폐쇄망 오프라인 설치(vendor/install), systemd, CRC/포맷 자가점검. firmware sht30_fw wire 포맷과 byte-parity 검증. 1번 모듈 통합(raspberry_pi4/): 누수+카메라를 동일 서버로. 레거시 sha256(key+json) body-sig → X-Signature HMAC 전환, 사진 canonical=device_id|sensor_id|log_id|timestamp. 서버 통합(php/): api/sensor_data.php 누수 이벤트 수신, api/photo_upload.php 신규(HMAC·finfo·서버생성 파일명·nonce 멱등·sensor 결속), photo.php 인증 사진서빙, dashboard.php 누수 패널+사진, sms_send.php 누수 SMS, config.php 태그/요약. DB: sql/migration_add_leak.sql(가산·멱등), schema_sht30.sql 통합 스키마. 보안 하드닝(외부 펜테스트 대비, 확정 CRITICAL/HIGH 0): cron_heartbeat api_key fail-closed, setup_mfa 부트스트랩 토큰 post-enrollment 잠금, var/ PHP 실행차단, HSTS. 문서/테스트: SYSTEM_OVERVIEW.md, security_hardening_test.php 갱신. 참고: 사용자 요청으로 세션 외 기존 미커밋 변경도 함께 포함. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
531 lines
20 KiB
Python
531 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
# =============================================================================
|
|
# leak_sensor.py - e-con 5C-08 누수감지센서 모니터링 메인 스크립트 (v2, Pi 4 전용)
|
|
#
|
|
# [대상 하드웨어] Raspberry Pi 4 Model B (BCM2711, 40핀 GPIO 헤더)
|
|
# - 원래 Raspberry Pi Zero 2 W 용이던 모듈을 Pi 4 Model B 전용으로 이식.
|
|
# - GPIO BCM 번호 체계·rpicam/libcamera 카메라 스택은 두 보드가 동일하므로
|
|
# 센서/촬영 로직은 그대로 유지. 차이는 배선/전원/이더넷/카메라 커넥터 문서뿐.
|
|
# (Pi 4 = 15핀 1.0mm CSI / Zero 2 W = 22핀 0.5mm CSI → 리본 비호환,
|
|
# Gigabit Ethernet, USB-C 5V/3A) → README.md 참고.
|
|
#
|
|
# [동작 흐름]
|
|
# 1. GPIO 초기화 → 센서 핀 입력 설정
|
|
# 2. 주기적(REPORT_INTERVAL)으로 현재 상태를 PHP API에 POST
|
|
# 3. 누수 감지 인터럽트 발생 시 즉시 알림 POST
|
|
# 4. 누수 지속 중엔 LEAK_REPORT_INTERVAL 마다 재보고
|
|
# 5. 복구(정상) 감지 시 복구 알림 POST
|
|
# 6. 누수 감지 시 Pi Camera로 촬영 → photo_upload.php에 업로드
|
|
# (촬영/업로드 실패해도 알림+SMS는 이미 전송된 상태)
|
|
#
|
|
# [설정·비밀값] config.py 는 비밀이 아닌 기본값만 보유하고, 서버 URL/API_KEY/
|
|
# 기기 식별자는 config_local.py(미커밋)에서 불러온다. config_local.py 가 없으면
|
|
# 기동 시 명확한 오류로 즉시 중단(fail closed)된다.
|
|
# =============================================================================
|
|
|
|
import time
|
|
import threading
|
|
import logging
|
|
import logging.handlers
|
|
import hashlib
|
|
import hmac
|
|
import platform
|
|
import sys
|
|
import subprocess
|
|
import os
|
|
import tempfile
|
|
|
|
import requests
|
|
try:
|
|
import RPi.GPIO as GPIO
|
|
SIMULATION = False
|
|
except (ImportError, RuntimeError):
|
|
# 개발/테스트 환경에서 RPi.GPIO 없을 때 시뮬레이션 모드
|
|
SIMULATION = True
|
|
print("[WARNING] RPi.GPIO 없음 → 시뮬레이션 모드로 실행됩니다.")
|
|
|
|
from config import (
|
|
SENSORS, ACTIVE_LOW, USE_INTERNAL_PULL,
|
|
API_URL, API_KEY,
|
|
REPORT_INTERVAL, LEAK_REPORT_INTERVAL, DEBOUNCE_MS,
|
|
DEVICE_ID, DEVICE_LOCATION,
|
|
LOG_FILE, LOG_MAX_BYTES, LOG_BACKUP_COUNT,
|
|
HTTP_TIMEOUT, HTTP_RETRY_COUNT, HTTP_RETRY_DELAY,
|
|
CAMERA_ENABLED, PHOTO_UPLOAD_URL,
|
|
PHOTO_WIDTH, PHOTO_HEIGHT, PHOTO_QUALITY,
|
|
)
|
|
|
|
# ─────────────────────────────────────────────
|
|
# 로거 설정
|
|
# ─────────────────────────────────────────────
|
|
def setup_logger():
|
|
logger = logging.getLogger("leak_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()
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
# 시뮬레이션 GPIO (테스트용)
|
|
# ─────────────────────────────────────────────
|
|
class SimGPIO:
|
|
BCM = "BCM"
|
|
IN = "IN"
|
|
PUD_UP = "PUD_UP"
|
|
PUD_DOWN = "PUD_DOWN"
|
|
PUD_OFF = "PUD_OFF"
|
|
FALLING = "FALLING"
|
|
RISING = "RISING"
|
|
BOTH = "BOTH"
|
|
_state = {}
|
|
|
|
@staticmethod
|
|
def setmode(mode): pass
|
|
@staticmethod
|
|
def setup(pin, mode, pull_up_down=None): SimGPIO._state[pin] = 1
|
|
@staticmethod
|
|
def input(pin): return SimGPIO._state.get(pin, 1)
|
|
@staticmethod
|
|
def add_event_detect(pin, edge, callback=None, bouncetime=0): pass
|
|
@staticmethod
|
|
def cleanup(): pass
|
|
|
|
if SIMULATION:
|
|
GPIO = SimGPIO()
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
# 누수 상태 관리
|
|
# ─────────────────────────────────────────────
|
|
class SensorState:
|
|
def __init__(self):
|
|
self.lock = threading.Lock()
|
|
# 센서별 상태 딕셔너리 {sensor_id: bool}
|
|
self.leak_detected = {s["id"]: False for s in SENSORS}
|
|
self.last_report_time = {s["id"]: 0 for s in SENSORS}
|
|
|
|
def is_leaking(self, sensor_id):
|
|
with self.lock:
|
|
return self.leak_detected.get(sensor_id, False)
|
|
|
|
def any_leaking(self):
|
|
with self.lock:
|
|
return any(self.leak_detected.values())
|
|
|
|
def set_leak(self, sensor_id, value):
|
|
with self.lock:
|
|
changed = self.leak_detected.get(sensor_id) != value
|
|
self.leak_detected[sensor_id] = value
|
|
return changed
|
|
|
|
|
|
state = SensorState()
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
# HTTP POST 유틸
|
|
# ─────────────────────────────────────────────
|
|
# [보안 메모 — raw-body HMAC 서명 (2번 모듈과 동일 스킴)]
|
|
# 서버(php/api/sensor_data.php + config.php::verify_signature_raw)는 "받은 본문
|
|
# 바이트 그대로"에 대해 X-Signature = lowercase_hex(HMAC-SHA256(key=API_KEY,
|
|
# msg=raw_body)) 를 검증한다. 따라서 "서명한 바이트"와 "실제 전송한 바이트"가
|
|
# 100% 동일해야 한다. requests 의 json= 직렬화에 맡기면 공백/순서/부동소수 포맷이
|
|
# 달라져 서명이 깨질 수 있으므로, 본문을 직접 문자열로 만들고 그 바이트에 서명한 뒤
|
|
# 그 바이트를 그대로 POST(data=body_bytes) 한다.
|
|
# (구버전의 sha256(API_KEY + 정렬 JSON) body-field 서명은 폐기 — 서버가 거부한다.)
|
|
def _esc(s: str) -> str:
|
|
r"""JSON 문자열 값 이스케이프 ('"' / '\\' / 제어문자<0x20 만). 멀티바이트는 그대로."""
|
|
out = []
|
|
for ch in s:
|
|
o = ord(ch)
|
|
if ch == '"':
|
|
out.append('\\"')
|
|
elif ch == '\\':
|
|
out.append('\\\\')
|
|
elif o < 0x20:
|
|
out.append('\\u%04x' % o)
|
|
else:
|
|
out.append(ch)
|
|
return "".join(out)
|
|
|
|
|
|
def leak_event_body(event_type: str, sensor: dict, is_leak: bool, timestamp: int) -> str:
|
|
"""누수 이벤트 본문 JSON 문자열을 고정 필드 순서로 생성(서명 호환).
|
|
|
|
필드 순서(서버 파싱은 순서 무관이나, 서명은 바이트 동일성을 요구하므로 고정):
|
|
device_id, device_location, sensor_id, sensor_name, event_type, is_leak, timestamp
|
|
"""
|
|
return (
|
|
'{'
|
|
f'"device_id":"{_esc(DEVICE_ID)}",'
|
|
f'"device_location":"{_esc(DEVICE_LOCATION)}",'
|
|
f'"sensor_id":{int(sensor["id"])},'
|
|
f'"sensor_name":"{_esc(sensor["name"])}",'
|
|
f'"event_type":"{_esc(event_type)}",'
|
|
f'"is_leak":{1 if is_leak else 0},'
|
|
f'"timestamp":{int(timestamp)}'
|
|
'}'
|
|
)
|
|
|
|
|
|
def sign_raw(body: str) -> str:
|
|
"""raw-body 서명: lowercase_hex(HMAC-SHA256(key=API_KEY, msg=body 바이트))."""
|
|
return hmac.new(API_KEY.encode("utf-8"), body.encode("utf-8"),
|
|
hashlib.sha256).hexdigest()
|
|
|
|
|
|
def _post_event(event_type: str, sensor: dict, is_leak: bool, want_response: bool):
|
|
"""누수 이벤트를 raw-body HMAC 서명으로 POST.
|
|
|
|
body 문자열을 만들고 그 바이트에 서명한 뒤 data=body_bytes 로 그대로 전송한다.
|
|
want_response=False 면 성공 여부(bool), True 면 응답 JSON(dict) 또는 None 을 반환.
|
|
"""
|
|
body = leak_event_body(event_type, sensor, is_leak, int(time.time()))
|
|
body_bytes = body.encode("utf-8")
|
|
signature = sign_raw(body)
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"X-Device-Id": DEVICE_ID,
|
|
"X-Signature": signature,
|
|
"User-Agent": "LeakSensor/2.0",
|
|
}
|
|
|
|
for attempt in range(1, HTTP_RETRY_COUNT + 1):
|
|
try:
|
|
resp = requests.post(
|
|
API_URL,
|
|
data=body_bytes,
|
|
timeout=HTTP_TIMEOUT,
|
|
headers=headers,
|
|
)
|
|
if resp.status_code == 200:
|
|
log.info(f"[POST OK] 센서{sensor['id']} event={event_type} leak={is_leak} → {resp.text[:80]}")
|
|
if not want_response:
|
|
return True
|
|
try:
|
|
return resp.json()
|
|
except ValueError:
|
|
log.warning("[POST] 응답 JSON 파싱 실패")
|
|
return None
|
|
else:
|
|
log.warning(f"[POST {resp.status_code}] 시도 {attempt}/{HTTP_RETRY_COUNT}: {resp.text[:120]}")
|
|
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}")
|
|
|
|
if attempt < HTTP_RETRY_COUNT:
|
|
time.sleep(HTTP_RETRY_DELAY)
|
|
|
|
log.error(f"[POST FAIL] {HTTP_RETRY_COUNT}회 모두 실패. 센서{sensor['id']}")
|
|
return False if not want_response else None
|
|
|
|
|
|
def post_to_server(event_type: str, sensor: dict, is_leak: bool):
|
|
"""PHP API 서버에 센서 데이터 POST (성공 여부 bool 반환).
|
|
|
|
event_type: "periodic" | "startup" | "leak_alert" | "leak_recovery"
|
|
"""
|
|
return _post_event(event_type, sensor, is_leak, want_response=False)
|
|
|
|
|
|
def post_to_server_with_response(event_type: str, sensor: dict, is_leak: bool):
|
|
"""post_to_server 와 동일하나 응답 JSON(dict) 또는 None 을 반환."""
|
|
return _post_event(event_type, sensor, is_leak, want_response=True)
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
# 카메라 촬영 / 사진 업로드
|
|
# ─────────────────────────────────────────────
|
|
def capture_photo(output_path: str) -> bool:
|
|
"""
|
|
rpicam-still (Bookworm+) 우선, 없으면 libcamera-still (Bullseye)로 사진 촬영.
|
|
성공 시 True, 실패 시 False.
|
|
"""
|
|
for cmd in ["rpicam-still", "libcamera-still"]:
|
|
try:
|
|
result = subprocess.run(
|
|
[
|
|
cmd,
|
|
"--width", str(PHOTO_WIDTH),
|
|
"--height", str(PHOTO_HEIGHT),
|
|
"--quality", str(PHOTO_QUALITY),
|
|
"--nopreview",
|
|
"--timeout", "2000",
|
|
"-o", output_path,
|
|
],
|
|
capture_output=True, text=True, timeout=15
|
|
)
|
|
if result.returncode == 0 and os.path.exists(output_path):
|
|
file_size = os.path.getsize(output_path)
|
|
log.info(f"[CAMERA] 촬영 성공: {output_path} ({file_size} bytes, cmd={cmd})")
|
|
return True
|
|
else:
|
|
log.warning(f"[CAMERA] {cmd} 실패 (rc={result.returncode}): {result.stderr[:200]}")
|
|
except FileNotFoundError:
|
|
continue # 이 명령어가 없으면 다음 것 시도
|
|
except subprocess.TimeoutExpired:
|
|
log.warning(f"[CAMERA] {cmd} 타임아웃")
|
|
except Exception as e:
|
|
log.warning(f"[CAMERA] {cmd} 예외: {e}")
|
|
|
|
log.error("[CAMERA] 촬영 실패: rpicam-still, libcamera-still 모두 사용 불가")
|
|
return False
|
|
|
|
|
|
def upload_photo(photo_path: str, log_id: int, sensor: dict) -> bool:
|
|
"""
|
|
촬영된 사진을 photo_upload.php에 multipart POST.
|
|
성공 시 True, 실패 시 False.
|
|
|
|
[인증] X-Signature = lowercase_hex(HMAC-SHA256(key=API_KEY, msg=canonical))
|
|
canonical = f"{DEVICE_ID}|{sensor_id}|{log_id}|{timestamp}" (공백 없음)
|
|
서버 php/api/photo_upload.php 의 canonical 과 바이트 동일해야 한다.
|
|
"""
|
|
timestamp = int(time.time())
|
|
sensor_id = int(sensor["id"])
|
|
# 서버와 정확히 동일한 문자열: device_id|sensor_id|log_id|timestamp (공백 없음).
|
|
# 폼 필드도 동일한 문자열 표현으로 보내야 서버 재구성 canonical 과 일치한다.
|
|
sensor_id_str = str(sensor_id)
|
|
log_id_str = str(log_id)
|
|
timestamp_str = str(timestamp)
|
|
canonical = f"{DEVICE_ID}|{sensor_id_str}|{log_id_str}|{timestamp_str}"
|
|
signature = hmac.new(API_KEY.encode("utf-8"), canonical.encode("utf-8"),
|
|
hashlib.sha256).hexdigest()
|
|
|
|
try:
|
|
with open(photo_path, "rb") as f:
|
|
files = {"photo": (os.path.basename(photo_path), f, "image/jpeg")}
|
|
form_data = {
|
|
"device_id": DEVICE_ID,
|
|
"sensor_id": sensor_id_str,
|
|
"log_id": log_id_str,
|
|
"timestamp": timestamp_str,
|
|
}
|
|
|
|
resp = requests.post(
|
|
PHOTO_UPLOAD_URL,
|
|
files=files,
|
|
data=form_data,
|
|
timeout=30,
|
|
headers={
|
|
"X-Device-Id": DEVICE_ID,
|
|
"X-Signature": signature,
|
|
"User-Agent": "LeakSensor/2.0",
|
|
}
|
|
)
|
|
|
|
if resp.status_code == 200:
|
|
log.info(f"[PHOTO UPLOAD OK] log_id={log_id} → {resp.text[:80]}")
|
|
return True
|
|
else:
|
|
log.warning(f"[PHOTO UPLOAD {resp.status_code}] {resp.text[:120]}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
log.error(f"[PHOTO UPLOAD ERR] {e}")
|
|
return False
|
|
|
|
|
|
def _handle_leak_with_photo(sensor: dict):
|
|
"""
|
|
누수 감지 시 처리 순서:
|
|
1. sensor_data.php에 POST → log_id 수신 (+ SMS 발송)
|
|
2. 카메라 촬영 (CAMERA_ENABLED 시)
|
|
3. photo_upload.php에 업로드
|
|
4. 임시파일 삭제
|
|
촬영/업로드 실패해도 1단계(알림+SMS)는 이미 완료.
|
|
"""
|
|
# 1) leak_alert POST → log_id 수신
|
|
resp_data = post_to_server_with_response("leak_alert", sensor, True)
|
|
if resp_data is None:
|
|
log.error(f"[PHOTO] 센서{sensor['id']} leak_alert POST 실패 → 촬영 건너뜀")
|
|
return
|
|
|
|
log_id = resp_data.get("log_id")
|
|
if not log_id:
|
|
log.warning(f"[PHOTO] 센서{sensor['id']} 응답에 log_id 없음 → 촬영 건너뜀")
|
|
return
|
|
|
|
# 2) 카메라 촬영
|
|
if not CAMERA_ENABLED:
|
|
log.info(f"[PHOTO] 카메라 비활성화 → 촬영 건너뜀 (log_id={log_id})")
|
|
return
|
|
|
|
# /tmp에 임시 JPEG 파일 생성
|
|
fd, photo_path = tempfile.mkstemp(suffix=".jpg", prefix="leak_")
|
|
os.close(fd)
|
|
|
|
try:
|
|
if not capture_photo(photo_path):
|
|
log.warning(f"[PHOTO] 촬영 실패 → 업로드 건너뜀 (log_id={log_id})")
|
|
return
|
|
|
|
# 3) 업로드
|
|
upload_photo(photo_path, log_id, sensor)
|
|
|
|
finally:
|
|
# 4) 임시파일 정리
|
|
try:
|
|
if os.path.exists(photo_path):
|
|
os.unlink(photo_path)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
# GPIO 인터럽트 콜백
|
|
# ─────────────────────────────────────────────
|
|
def make_callback(sensor: dict):
|
|
def callback(channel):
|
|
gpio_val = GPIO.input(sensor["gpio"])
|
|
|
|
# ACTIVE_LOW: GPIO=0 이면 누수, GPIO=1 이면 정상
|
|
# ACTIVE_HIGH(트랜지스터 반전): GPIO=1 이면 누수
|
|
if ACTIVE_LOW:
|
|
is_leak = (gpio_val == 0)
|
|
else:
|
|
is_leak = (gpio_val == 1)
|
|
|
|
changed = state.set_leak(sensor["id"], is_leak)
|
|
|
|
if changed:
|
|
if is_leak:
|
|
log.warning(f"[LEAK DETECTED] 센서{sensor['id']} ({sensor['name']}) !!!")
|
|
# leak_alert + 사진 촬영/업로드를 한 스레드에서 순차 실행
|
|
threading.Thread(
|
|
target=_handle_leak_with_photo,
|
|
args=(sensor,),
|
|
daemon=True
|
|
).start()
|
|
else:
|
|
log.info(f"[LEAK RECOVERED] 센서{sensor['id']} ({sensor['name']}) 정상 복구")
|
|
threading.Thread(
|
|
target=post_to_server,
|
|
args=("leak_recovery", sensor, False),
|
|
daemon=True
|
|
).start()
|
|
|
|
return callback
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
# 주기 보고 스레드
|
|
# ─────────────────────────────────────────────
|
|
def periodic_reporter():
|
|
"""
|
|
정상 상태: REPORT_INTERVAL 주기로 보고
|
|
누수 상태: LEAK_REPORT_INTERVAL 주기로 반복 보고 (SMS 재발송 방지는 PHP에서 처리)
|
|
"""
|
|
log.info("[REPORTER] 주기 보고 스레드 시작")
|
|
last_sent = {s["id"]: 0 for s in SENSORS}
|
|
|
|
while True:
|
|
now = time.time()
|
|
for sensor in SENSORS:
|
|
sid = sensor["id"]
|
|
is_leak = state.is_leaking(sid)
|
|
interval = LEAK_REPORT_INTERVAL if is_leak else REPORT_INTERVAL
|
|
|
|
if now - last_sent[sid] >= interval:
|
|
post_to_server("periodic", sensor, is_leak)
|
|
last_sent[sid] = now
|
|
|
|
time.sleep(1)
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
# GPIO 초기화
|
|
# ─────────────────────────────────────────────
|
|
def setup_gpio():
|
|
GPIO.setmode(GPIO.BCM)
|
|
|
|
# 외부 4.7MΩ 풀다운만 사용 → 내부 풀저항은 PUD_OFF 필수
|
|
# (내부 50kΩ 풀다운을 켜면 1.5MΩ 센서가 GPIO를 HIGH로 올리지 못함)
|
|
pull = GPIO.PUD_DOWN if USE_INTERNAL_PULL else GPIO.PUD_OFF
|
|
|
|
for sensor in SENSORS:
|
|
gpio = sensor["gpio"]
|
|
GPIO.setup(gpio, GPIO.IN, pull_up_down=pull)
|
|
|
|
# 초기 상태 읽기
|
|
gpio_val = GPIO.input(gpio)
|
|
is_leak = (gpio_val == 0) if ACTIVE_LOW else (gpio_val == 1)
|
|
state.set_leak(sensor["id"], is_leak)
|
|
log.info(f"[INIT] 센서{sensor['id']} GPIO{gpio} → {'누수!' if is_leak else '정상'}")
|
|
|
|
# 엣지 감지 인터럽트 등록
|
|
GPIO.add_event_detect(
|
|
gpio,
|
|
GPIO.BOTH,
|
|
callback=make_callback(sensor),
|
|
bouncetime=DEBOUNCE_MS
|
|
)
|
|
|
|
log.info(f"[INIT] GPIO 설정 완료. 총 {len(SENSORS)}개 센서")
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
# 메인
|
|
# ─────────────────────────────────────────────
|
|
def main():
|
|
log.info("=" * 60)
|
|
log.info(f"누수감지 시스템 v2 (Pi 4) 시작 | 기기={DEVICE_ID} | 위치={DEVICE_LOCATION}")
|
|
log.info(f"플랫폼={platform.machine()} | API={API_URL}")
|
|
log.info(f"카메라={'활성' if CAMERA_ENABLED else '비활성'}")
|
|
log.info(f"센서={[s['name'] for s in SENSORS]}")
|
|
log.info(f"모드={'시뮬레이션' if SIMULATION else '실제 GPIO'}")
|
|
log.info("=" * 60)
|
|
|
|
try:
|
|
setup_gpio()
|
|
|
|
# 시작 시 현재 상태 즉시 보고
|
|
for sensor in SENSORS:
|
|
post_to_server("startup", sensor, state.is_leaking(sensor["id"]))
|
|
|
|
# 주기 보고 스레드 시작
|
|
reporter_thread = threading.Thread(target=periodic_reporter, daemon=True)
|
|
reporter_thread.start()
|
|
|
|
log.info("[MAIN] 모니터링 시작. Ctrl+C로 종료.")
|
|
|
|
# 메인 스레드는 살아있어야 인터럽트 작동
|
|
while True:
|
|
time.sleep(1)
|
|
|
|
except KeyboardInterrupt:
|
|
log.info("[MAIN] 사용자에 의해 종료됨.")
|
|
except Exception as e:
|
|
log.exception(f"[MAIN] 예외 발생: {e}")
|
|
raise
|
|
finally:
|
|
GPIO.cleanup()
|
|
log.info("[MAIN] GPIO 정리 완료.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|