feat: E1002 顶部加设备电量/温湿度列+server 四路上游并行拉取与响应预算

- 新增 SenseCraft 设备遥测客户端(api-key 走 .env),读电量/充电/温度/湿度,DEVICE_TTL 与设备 300s 上报周期同步
- 大时钟 F84 右侧一列三行左缘对齐:电池(充电绿/≤20% 红)、温度计+蓝色水滴、星期日期(与时钟基线对齐)
- server 过期上游由串行改并行,整体最多等 FETCH_BUDGET 秒(默认 5);超预算的一路本次用旧值、后台拉完回填缓存,同一路不重复实拉
- 此前串行且上游 socket 超时 30s,任一路卡住即整图拖到平台抓取超时;「多路 TTL 同时到期」猜想经日志核查不成立,排查顺序写入 CLAUDE.md

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
2026-09-14 15:02:34 +08:00
parent 7378d6bb88
commit f96ae29590
8 changed files with 255 additions and 41 deletions

52
sensecraft_api.py Normal file
View File

@@ -0,0 +1,52 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
SenseCraft HMI 设备遥测客户端 —— 读 reTerminal E1002 自身上报的电量 / 温度 / 湿度。
接口GET {SENSECRAFT_API_BASE}/api/v1/user/device/iot_data/{SENSECRAFT_DEVICE_EUI},头 `api-key: <SENSECRAFT_API_KEY>`
(用户在 SenseCraft 后台生成的 API key放 .env不入库。响应2026-09-14 实测):
{"code":200,"result":{"battery":{"charging":true,"level":100},"sensor":{"humidity":42.9,"temp":30.3},
"dataaccess":{"interval":300},"devicestatus":{"status":1},...},"message":"success"}
取值result.battery.level%/ result.battery.charging / result.sensor.temp/ result.sensor.humidity%)。
设备本身每 dataaccess.interval300s上报一次故 DEVICE_TTL 取 300 与之同步即可。直连、不走代理,超时 10s。
密钥未配置直接抛错 → 上层回退占位 "--"
"""
import json
import urllib.error
import urllib.request
import config
TIMEOUT = 10
def get_iot_data():
"""返回 {battery(int|None), charging(bool), temp(float|None), humidity(float|None)};网络/鉴权失败抛出。"""
if not config.SENSECRAFT_API_KEY or not config.SENSECRAFT_DEVICE_EUI:
raise RuntimeError("未配置 SENSECRAFT_API_KEY / SENSECRAFT_DEVICE_EUI")
url = f"{config.SENSECRAFT_API_BASE.rstrip('/')}/api/v1/user/device/iot_data/{config.SENSECRAFT_DEVICE_EUI}"
req = urllib.request.Request(url, headers={"api-key": config.SENSECRAFT_API_KEY, "Accept": "application/json",
"User-Agent": "eink-push/sensecraft_api"})
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
try:
with opener.open(req, timeout=TIMEOUT) as resp:
body = json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as e:
raise RuntimeError(f"HTTP {e.code}{e.read().decode('utf-8', 'replace')[:200]}") from e
if body.get("code") != 200:
raise RuntimeError(f"接口返回 code={body.get('code')} message={body.get('message')}")
r = body.get("result") or {}
bat, sen = r.get("battery") or {}, r.get("sensor") or {}
return {"battery": _num(bat.get("level"), int), "charging": bool(bat.get("charging")),
"temp": _num(sen.get("temp"), float), "humidity": _num(sen.get("humidity"), float)}
def _num(v, cast):
try:
return cast(v) if v is not None else None
except (TypeError, ValueError):
return None
if __name__ == "__main__":
print(get_iot_data())