YANG JIANKUAN
f96ae29590
- 新增 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>
53 lines
2.5 KiB
Python
53 lines
2.5 KiB
Python
#!/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.interval（300s）上报一次，故 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())
|