42a6ee4bb0
Pi Zero 2 W + ST7789 패널에 노드 상태(누수/온습도/미세먼지·오프라인·정체)를 상시 표시하는 디스플레이 노드 신설. 배포 전 렌더 검증 원칙에 따라 합성 최악 프레임 11종을 패널 실해상도로 render_preview/ 에 PNG 로 선렌더. - st7789_driver.py · display_dashboard.py · render_preview.py(합성 프레임 생성) - install.sh · vendor.sh · systemd/display-dashboard.service · 배선(WIRING)·연결테스트 - render_preview/01~11: 요약/통합/메트릭/오프라인/stale/legacy-무PM 최악 케이스
409 lines
16 KiB
Python
409 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
# =============================================================================
|
|
# display_dashboard.py — 미니 상태 표시기 메인 루프 (Pi Zero 2 W + ST7789V 2.0")
|
|
#
|
|
# 서버 api/status_summary.php 를 30초 폴링 → 240x320 화면 렌더(Pillow) → SPI 블릿.
|
|
# 페이지 자동 순환: [0] 시스템 요약 → [1..n] 센서별 상세.
|
|
# 배경색 = 시스템 상태(정상 녹색 / 온습도 주황 / 미세먼지 마젠타 / 누수 청색 / 오프라인 회색).
|
|
# 서버 미응답 시 마지막 데이터 + STALE 배너(경과 시간) 표시.
|
|
#
|
|
# 글꼴: 기본 DejaVu(영문/숫자) — 2인치 글랜스 화면이라 라벨은 영문 축약.
|
|
# 한글 라벨 원하면 config FONT_PATH 에 한글 TTF 지정.
|
|
# =============================================================================
|
|
import sys
|
|
import time
|
|
|
|
import requests
|
|
from PIL import Image, ImageDraw, ImageFont
|
|
|
|
import config
|
|
from st7789_driver import open_display
|
|
|
|
W, H = config.WIDTH, config.HEIGHT
|
|
|
|
# 상태색 (배경, 강조) — 대시보드 팔레트와 동일 계열
|
|
STATE_COLORS = {
|
|
"ok": ((10, 40, 28), (53, 214, 160)),
|
|
"alert": ((52, 30, 6), (255, 178, 62)),
|
|
"pm": ((42, 12, 48), (208, 120, 236)), # 미세먼지 — 마젠타/바이올렛(온습도 주황과 명확히 구분)
|
|
"leak": ((8, 30, 52), (63, 199, 230)),
|
|
"offline": ((30, 32, 44), (120, 133, 184)),
|
|
"stale": ((40, 12, 20), (255, 84, 112)),
|
|
}
|
|
INK = (234, 238, 255) # 1차: 값/제목
|
|
INK2 = (150, 160, 200) # 2차: 보조 텍스트
|
|
INK3 = (96, 104, 140) # 3차: eyebrow 라벨·눈금
|
|
HAIR = (42, 48, 72) # 헤어라인 구분선
|
|
TEMP_C = (255, 149, 100)
|
|
HUM_C = (74, 194, 224)
|
|
PM_C = (208, 120, 236) # 미세먼지 수치 — 마젠타(온습도경보로 오독 방지)
|
|
|
|
PAD = 16 # 좌우 여백(8px 그리드)
|
|
|
|
|
|
def font(size, bold=True):
|
|
path = config.FONT_PATH if bold else config.FONT_PATH_LIGHT
|
|
try:
|
|
return ImageFont.truetype(path, size)
|
|
except OSError:
|
|
return ImageFont.load_default()
|
|
|
|
|
|
# 타이포 위계: huge(위젯 히어로) > hero(상태) > value(수치) > label > eyebrow
|
|
F_HUGE = font(66)
|
|
F_HERO = font(44)
|
|
F_BIG = font(44) # draw_error 호환 별칭
|
|
F_VALUE = font(30)
|
|
F_MED = font(30) # 호환 별칭
|
|
F_LABEL = font(16)
|
|
F_SM = font(16) # 호환 별칭
|
|
F_XS = font(13, bold=False)
|
|
F_EYE = font(12, bold=False)
|
|
|
|
|
|
def _dim(color, f=0.30):
|
|
"""색을 배경쪽으로 눌러 채움용 저채도 버전 생성."""
|
|
return tuple(int(c * f) for c in color)
|
|
|
|
|
|
def _eyebrow(d, xy, text, fill=INK3, ls=2):
|
|
"""트래킹된 대문자 소형 라벨(Apple eyebrow). 그린 뒤 끝 x 반환."""
|
|
x, y = xy
|
|
for ch in text.upper():
|
|
d.text((x, y), ch, font=F_EYE, fill=fill)
|
|
x += d.textlength(ch, font=F_EYE) + ls
|
|
return x
|
|
|
|
|
|
def fetch_status():
|
|
# 최소 권한: config_local 에는 파생 STATUS_KEY 만 있다(API_KEY 비보유).
|
|
r = requests.get(
|
|
config.STATUS_URL,
|
|
headers={
|
|
"X-Status-Key": config.STATUS_KEY,
|
|
# 커스텀 UA — 공유호스팅 WAF가 기본 python-requests UA를 403 차단하므로
|
|
"User-Agent": "LeakDisplay/2.0",
|
|
},
|
|
timeout=config.FETCH_TIMEOUT,
|
|
)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def state_label(state):
|
|
return {"ok": "ALL OK", "alert": "ALERT", "pm": "PM ALERT",
|
|
"leak": "LEAK!", "offline": "OFFLINE"}.get(state, state.upper())
|
|
|
|
|
|
def sensor_state(s):
|
|
"""센서 1개의 표시 상태 — 우선순위 누수 > 미세먼지 > 온습도 > offline > ok.
|
|
|
|
(기존 버그: offline 을 leak/미세먼지보다 먼저 판정 → 오프라인이 누수를 가림.
|
|
또한 미세먼지 티어 부재. 이 순서로 재정렬해 실제 경보가 오프라인/온습도에 묻히지 않게.)
|
|
누수·미세먼지는 값이 있으면 오프라인이라도 우선 노출한다.
|
|
"""
|
|
if s.get("leak") == 1:
|
|
return "leak", "LEAK!"
|
|
pm_status = s.get("pm_status", "normal") or "normal"
|
|
if pm_status != "normal":
|
|
return "pm", {"high_pm25": "PM2.5", "high_pm10": "PM10"}.get(pm_status, "DUST")
|
|
if s.get("pending"):
|
|
return "alert", "CHECK.."
|
|
# th_status 신규 키 우선, 없으면 metric_status 별칭(구 API 호환)
|
|
th_status = s.get("th_status", s.get("metric_status", "normal")) or "normal"
|
|
if th_status != "normal":
|
|
return "alert", th_status.replace("_", " ").upper()
|
|
if s.get("offline"):
|
|
return "offline", "OFFLINE"
|
|
return "ok", "OK"
|
|
|
|
|
|
def new_frame(state):
|
|
bg, accent = STATE_COLORS.get(state, STATE_COLORS["offline"])
|
|
img = Image.new("RGB", (W, H), bg)
|
|
d = ImageDraw.Draw(img)
|
|
d.rectangle([0, 0, W, 4], fill=accent) # 상단 상태 스트립(얇게)
|
|
return img, d, accent
|
|
|
|
|
|
def _hairline(d, y):
|
|
d.line([(PAD, y), (W - PAD, y)], fill=HAIR, width=1)
|
|
|
|
|
|
def draw_summary(data, stale_sec):
|
|
sys_ = data["system"]
|
|
state = "stale" if stale_sec is not None else sys_["state"]
|
|
img, d, accent = new_frame(state)
|
|
|
|
# ── 헤더: eyebrow → hero 상태 → 시각 ──────────────────────────────
|
|
_eyebrow(d, (PAD, 22), "SYSTEM STATUS")
|
|
d.text((PAD, 38), state_label(sys_["state"]), font=F_HERO, fill=accent)
|
|
if stale_sec is not None:
|
|
d.text((PAD, 90), f"STALE {int(stale_sec // 60)}m · no server",
|
|
font=F_XS, fill=(255, 150, 170))
|
|
else:
|
|
d.text((PAD, 90), data["ts"][5:16].replace("-", "."), font=F_XS, fill=INK2)
|
|
|
|
_hairline(d, 116)
|
|
|
|
# ── KPI 목록: 라벨(eyebrow) 좌 / 값 우 ───────────────────────────
|
|
rows = [
|
|
("ONLINE", f"{sys_['online']}/{sys_['total']}"),
|
|
("LEAK OPEN", str(sys_["unrecovered"])),
|
|
("TODAY ALERTS", str(sys_["today_alerts"])),
|
|
("SMS LEFT", str(sys_["sms_balance"]) if sys_["sms_balance"] is not None else "-"),
|
|
]
|
|
y = 130
|
|
for label, val in rows:
|
|
_eyebrow(d, (PAD, y + 4), label)
|
|
d.text((W - PAD, y), val, font=F_LABEL, fill=INK, anchor="ra")
|
|
y += 34
|
|
|
|
# ── 푸터: 센서 상태 점 ───────────────────────────────────────────
|
|
_hairline(d, H - 44)
|
|
x = PAD
|
|
for s in data["sensors"]:
|
|
st, _ = sensor_state(s)
|
|
_, dot = STATE_COLORS.get(st, STATE_COLORS["offline"])
|
|
cy = H - 24
|
|
d.ellipse([x, cy - 5, x + 10, cy + 5], fill=dot)
|
|
d.text((x + 16, cy - 9), f"S{s['id']}", font=F_LABEL, fill=INK)
|
|
x += 68
|
|
return img
|
|
|
|
|
|
def _sparkline(d, x, y, w, h, values, color):
|
|
"""면적 채움 + 추세선 + 현재값 도트. 반환: (lo, hi) 또는 None(데이터부족)."""
|
|
pts = [(i, v) for i, v in enumerate(values) if v is not None]
|
|
if len(pts) < 2:
|
|
d.text((x, y + h // 2 - 7), "no data", font=F_XS, fill=INK3)
|
|
return None
|
|
vals = [v for _, v in pts]
|
|
lo, hi = min(vals), max(vals)
|
|
span = (hi - lo) or 1.0
|
|
n = len(values)
|
|
coords = [
|
|
(x + w * (i / (n - 1) if n > 1 else 0), y + h - h * ((v - lo) / span))
|
|
for i, v in pts
|
|
]
|
|
# 저채도 면적 채움(라인 아래) → 라인 → 현재값 도트
|
|
poly = coords + [(coords[-1][0], y + h), (coords[0][0], y + h)]
|
|
d.polygon(poly, fill=_dim(color))
|
|
d.line(coords, fill=color, width=2)
|
|
cx, cyy = coords[-1]
|
|
d.ellipse([cx - 3, cyy - 3, cx + 3, cyy + 3], fill=color, outline=INK)
|
|
return lo, hi
|
|
|
|
|
|
def _leak_timeline(d, x, y, w, events, accent):
|
|
"""24h(0~1440분) 축에 누수 이벤트 시각(HH:MM)을 틱+도트로 표시."""
|
|
base = y + 12
|
|
d.line([(x, base), (x + w, base)], fill=HAIR, width=1)
|
|
d.text((x, base + 6), "0h", font=F_XS, fill=INK3)
|
|
d.text((x + w, base + 6), "24h", font=F_XS, fill=INK3, anchor="ra")
|
|
if not events:
|
|
d.text((x + w / 2, base - 18), "no events", font=F_XS, fill=INK3, anchor="ma")
|
|
return
|
|
for t in events:
|
|
try:
|
|
hh, mm = t.split(":")
|
|
frac = (int(hh) * 60 + int(mm)) / 1440.0
|
|
except (ValueError, AttributeError):
|
|
continue
|
|
px = x + w * frac
|
|
d.line([(px, base - 9), (px, base + 9)], fill=accent, width=2)
|
|
d.ellipse([px - 3, base - 13, px + 3, base - 7], fill=accent)
|
|
|
|
|
|
def _wide_spark(d, x, y, w, h, values, color):
|
|
"""넓은 히어로 추세: 면적+라인+도트, min/max 를 우측 상·하단 코너에."""
|
|
rng = _sparkline(d, x, y, w, h, values, color)
|
|
if rng is not None:
|
|
lo, hi = rng
|
|
d.text((x + w, y - 14), f"{hi:.0f}", font=F_XS, fill=INK3, anchor="ra")
|
|
d.text((x + w, y + h + 2), f"{lo:.0f}", font=F_XS, fill=INK3, anchor="ra")
|
|
|
|
|
|
def _leak_sub(s):
|
|
"""누수 서브상태 텍스트/색 — 결합 노드 ZONE A용(전체 상태와 독립 표기)."""
|
|
if s.get("leak") == 1:
|
|
return "LEAK!", STATE_COLORS["leak"][1]
|
|
return "DRY", STATE_COLORS["ok"][1]
|
|
|
|
|
|
def _pm_fmt(v):
|
|
"""PM 정수 표기 — 값 없으면 em dash."""
|
|
return "—" if v is None else str(int(round(v)))
|
|
|
|
|
|
def _zone_leak(d, s, y0):
|
|
"""ZONE A(y46~120) — 누수: 큰 DRY/LEAK! + 24h 이벤트 수."""
|
|
_eyebrow(d, (PAD, y0), "LEAK STATUS")
|
|
txt, col = _leak_sub(s)
|
|
d.text((PAD, y0 + 14), txt, font=F_HERO, fill=col)
|
|
d.text((W - PAD, y0 + 2), "24H EVENTS", font=F_EYE, fill=INK3, anchor="ra")
|
|
d.text((W - PAD, y0 + 18), str(len(s.get("leak_events") or [])),
|
|
font=F_VALUE, fill=INK, anchor="ra")
|
|
|
|
|
|
def _zone_pm(d, s, y0):
|
|
"""ZONE B(y128~198) — 미세먼지: PM2.5 큰 값(마젠타) + PM10 보조. 값 없으면 '—'."""
|
|
_eyebrow(d, (PAD, y0), "FINE DUST · PM2.5")
|
|
pm25 = _pm_fmt(s.get("pm25"))
|
|
d.text((PAD, y0 + 14), pm25, font=F_HERO, fill=PM_C)
|
|
w = d.textlength(pm25, font=F_HERO)
|
|
d.text((PAD + w + 8, y0 + 40), "µg/m³", font=F_XS, fill=INK3)
|
|
d.text((W - PAD, y0 + 2), "PM10", font=F_EYE, fill=INK3, anchor="ra")
|
|
d.text((W - PAD, y0 + 18), _pm_fmt(s.get("pm10")), font=F_VALUE, fill=PM_C, anchor="ra")
|
|
|
|
|
|
def _zone_th(d, s, y0):
|
|
"""ZONE C(y208~300) — 온습도: TEMP / HUM 2단(오직 sht30 행 값)."""
|
|
temp_txt = f"{s['temp']:.1f}°" if s.get("temp") is not None else "—"
|
|
hum_txt = f"{s['hum']:.0f}%" if s.get("hum") is not None else "—"
|
|
_eyebrow(d, (PAD, y0), "TEMP °C")
|
|
d.text((PAD, y0 + 16), temp_txt, font=F_HERO, fill=TEMP_C)
|
|
xr = W // 2 + 8
|
|
_eyebrow(d, (xr, y0), "HUM %")
|
|
d.text((xr, y0 + 16), hum_txt, font=F_HERO, fill=HUM_C)
|
|
|
|
|
|
def _draw_combined(d, s):
|
|
"""결합 노드(sensor 1·3): 240x320 3분할 — LEAK / 미세먼지 / 온습도."""
|
|
_zone_leak(d, s, 48)
|
|
_hairline(d, 124)
|
|
_zone_pm(d, s, 130)
|
|
_hairline(d, 202)
|
|
_zone_th(d, s, 210)
|
|
|
|
|
|
def _draw_metric(d, s):
|
|
"""온습도 전용 노드(sensor 2): 온도 히어로 + 습도 보조 (기존 레이아웃)."""
|
|
hist = s.get("history") or []
|
|
temps = [p.get("temp") for p in hist]
|
|
hums = [p.get("hum") for p in hist]
|
|
_eyebrow(d, (PAD, 46), "TEMPERATURE °C")
|
|
temp_txt = f"{s['temp']:.1f}°" if s.get("temp") is not None else "—"
|
|
# 이상치(예: -999.0)로 240px 를 넘으면 폭에 맞는 작은 폰트로 폴백(무단 클립 방지)
|
|
hero_font = F_HUGE if d.textlength(temp_txt, font=F_HUGE) <= W - 2 * PAD else F_HERO
|
|
d.text((PAD, 56), temp_txt, font=hero_font, fill=TEMP_C)
|
|
_wide_spark(d, PAD, 146, W - 2 * PAD, 58, temps, TEMP_C)
|
|
_hairline(d, 226)
|
|
_eyebrow(d, (PAD, 238), "HUMIDITY %")
|
|
hum_txt = f"{s['hum']:.0f}%" if s.get("hum") is not None else "—"
|
|
d.text((PAD, 252), hum_txt, font=F_VALUE, fill=HUM_C)
|
|
_sparkline(d, W - PAD - 96, 254, 96, 30, hums, HUM_C)
|
|
|
|
|
|
def _draw_leak_only(d, s, st_text, accent):
|
|
"""누수 전용 노드: 누수 히어로 + 24h 타임라인 (기존 레이아웃)."""
|
|
_eyebrow(d, (PAD, 46), "LEAK STATUS")
|
|
d.text((PAD, 64), st_text, font=F_HUGE, fill=accent)
|
|
_hairline(d, 190)
|
|
_eyebrow(d, (PAD, 202), "LEAK · 24H")
|
|
_leak_timeline(d, PAD, 224, W - 2 * PAD, s.get("leak_events") or [], accent)
|
|
|
|
|
|
def draw_sensor(data, idx, stale_sec):
|
|
s = data["sensors"][idx]
|
|
st, st_text = sensor_state(s)
|
|
frame_state = "stale" if stale_sec is not None else st
|
|
img, d, accent = new_frame(frame_state)
|
|
|
|
leak_node = bool(s.get("leak_node"))
|
|
metric_node = bool(s.get("metric_node"))
|
|
pm_node = bool(s.get("pm_node")) # 구 API 는 키 부재 → False
|
|
|
|
# ── 헤더 2행: (1) SENSOR id 좌 · 상태 우 (2) 능력 칩 ─────────
|
|
# 칩+상태를 한 줄에 몰면 결합 노드(칩 3개)에서 우측 상태와 겹친다 →
|
|
# 식별/상태는 위 행, 능력 칩은 아래 행으로 분리(충돌 없음).
|
|
chips = []
|
|
if leak_node:
|
|
chips.append("LEAK")
|
|
if metric_node:
|
|
chips.append("T/H")
|
|
if pm_node:
|
|
chips.append("PM")
|
|
_eyebrow(d, (PAD, 12), f"SENSOR {s['id']}")
|
|
d.text((W - PAD, 10), st_text, font=F_LABEL, fill=accent, anchor="ra")
|
|
_eyebrow(d, (PAD, 30), " · ".join(chips), fill=INK3)
|
|
|
|
# ── 본문: 능력 조합으로 분기(우발적 순서 아님) ────────────────
|
|
if leak_node and (metric_node or pm_node):
|
|
_draw_combined(d, s) # 결합 노드 3분할
|
|
elif metric_node:
|
|
_draw_metric(d, s) # 온습도 전용
|
|
elif leak_node:
|
|
_draw_leak_only(d, s, st_text, accent) # 누수 전용
|
|
else:
|
|
d.text((PAD, 120), st_text, font=F_HUGE, fill=accent)
|
|
|
|
# ── 푸터: 갱신 시각 ──────────────────────────────────────────
|
|
if s.get("last_seen"):
|
|
d.text((PAD, H - 22), f"upd {s['last_seen'][5:16]}", font=F_XS, fill=INK3)
|
|
|
|
return img
|
|
|
|
|
|
def draw_error(err_text):
|
|
img, d, accent = new_frame("stale")
|
|
_eyebrow(d, (PAD, 22), "CONNECTION")
|
|
d.text((PAD, 38), "NO DATA", font=F_HERO, fill=accent)
|
|
_hairline(d, 116)
|
|
d.text((PAD, 130), "Server unreachable", font=F_LABEL, fill=INK)
|
|
d.text((PAD, 158), str(err_text)[:64], font=F_XS, fill=INK2)
|
|
d.text((PAD, H - 30), time.strftime("%m-%d %H:%M"), font=F_XS, fill=INK3)
|
|
return img
|
|
|
|
|
|
def main():
|
|
print(f"[INFO] display_dashboard start — {getattr(config, 'DEVICE_NAME', 'display')}")
|
|
disp, simulated = open_display(config)
|
|
if simulated:
|
|
print("[INFO] simulation mode — frames saved to /tmp/display_preview.png")
|
|
|
|
data = None
|
|
last_ok = 0.0
|
|
last_fetch = 0.0
|
|
last_err = "not fetched yet"
|
|
page = 0
|
|
last_page_flip = 0.0
|
|
|
|
while True:
|
|
now = time.time()
|
|
|
|
if now - last_fetch >= config.POLL_INTERVAL or (data is None and now - last_fetch >= 5):
|
|
last_fetch = now
|
|
try:
|
|
data = fetch_status()
|
|
last_ok = now
|
|
except Exception as e:
|
|
last_err = e
|
|
print(f"[WARN] fetch failed: {e}")
|
|
|
|
stale_sec = None
|
|
if data is not None and (now - last_ok) > config.STALE_AFTER:
|
|
stale_sec = now - last_ok
|
|
|
|
try:
|
|
if data is None:
|
|
img = draw_error(last_err)
|
|
else:
|
|
pages = 1 + len(data["sensors"])
|
|
page = min(page, pages - 1) # 센서 수 감소 시 즉시 클램프(IndexError 방지)
|
|
if now - last_page_flip >= config.PAGE_INTERVAL:
|
|
page = (page + 1) % pages
|
|
last_page_flip = now
|
|
img = draw_summary(data, stale_sec) if page == 0 else draw_sensor(data, page - 1, stale_sec)
|
|
disp.display(img)
|
|
except Exception as e:
|
|
print(f"[ERROR] render/display failed: {e}")
|
|
|
|
time.sleep(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except KeyboardInterrupt:
|
|
sys.exit(0)
|